tutti/packages/connector/runtime/implementationhost/host.go

// Package implementationhost owns the host-neutral managed Connector runtime.
// Products provide artifact, runtime, process, and credential ports and expose
// immutable route projections through their native CLI and Agent runtimes.
package implementationhost

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"sort"
	"strings"
	"sync"
	"time"

	agentruntime "github.com/tutti-os/tutti/packages/agent/daemon/runtime"
	market "github.com/tutti-os/tutti/packages/connector/daemon/core"
	connectorruntime "github.com/tutti-os/tutti/packages/connector/runtime"
	connectorartifact "github.com/tutti-os/tutti/packages/connector/runtime/artifact"
	"github.com/tutti-os/tutti/packages/connector/runtime/command"
	"github.com/tutti-os/tutti/packages/connector/runtime/mcp"
)

type PreparedArtifactResolver interface {
	ResolvePrepared(context.Context, market.Release) (market.PreparedArtifactReceipt, error)
}

type ReconcileRequest struct {
	Runtime market.RuntimeReconcileRequest
}

// RemoteMCPClient is the protocol client required by one remote Connector
// route. Products decide whether the client connects directly to their MCP
// Gateway or through a product-owned relay.
type RemoteMCPClient interface {
	Call(context.Context, string, any) (json.RawMessage, error)
	RegisterTool(string, map[string]any) error
	ReplaceTools(map[string]map[string]any) error
	Close(context.Context) error
}

// RemoteMCPClientRequest carries the complete non-credential identity of one
// remote Connector route. A product adapter may use it to bind a relay request
// to the lifecycle operation without exposing account credentials to the
// runtime machine.
type RemoteMCPClientRequest struct {
	OperationID    string
	ConnectionID   string
	ConnectorKey   string
	AccountID      string
	ReleaseDigest  string
	Version        string
	Generation     market.HostGeneration
	Implementation market.RemoteStreamableHTTPImplementation
}

// RemoteMCPClientFactory is the product port for remote Connector MCP
// connectivity. ImplementationHost owns protocol bootstrap and route
// lifecycle; the factory owns the physical connection path and request
// authorization.
type RemoteMCPClientFactory interface {
	NewRemoteMCPClient(context.Context, RemoteMCPClientRequest) (RemoteMCPClient, error)
}

type Config struct {
	Artifacts              PreparedArtifactResolver
	CLIInstallations       market.CLIInstallationManager
	Runtimes               connectorruntime.ConnectorRuntimeResolver
	Processes              agentruntime.ProcessTransport
	Routes                 RouteObserver
	Authorization          AuthorizationObserver
	Registry               *RouteRegistry
	MCP                    *MCPRegistry
	StateRoot              string
	BinDir                 string
	UserHome               string
	MCPStartupTimeout      time.Duration
	RemoteMCPClientFactory RemoteMCPClientFactory
}

type Host struct {
	admission              sync.RWMutex
	laneMu                 sync.Mutex
	connectorLanes         map[string]*sync.Mutex
	artifacts              PreparedArtifactResolver
	planner                *connectorruntime.ManagedRoutePlanner
	processes              agentruntime.ProcessTransport
	routeObserver          RouteObserver
	authorizationObserver  AuthorizationObserver
	mcpStartupTimeout      time.Duration
	routes                 *connectorruntime.RouteTable
	snapshots              *connectorruntime.ExecutionSnapshotter
	authorizationProvider  *managedCredentialAuthorizationProvider
	authorizationMu        sync.Mutex
	authorizationRoutes    map[string]*connectorRoute
	remoteMCPClientFactory RemoteMCPClientFactory
	mcpRegistry            *MCPRegistry
	registry               *RouteRegistry
	binDir                 string
	remoteMCPTools         *remoteMCPToolCache
}

type connectorRoute struct {
	id                     string
	connectionID           string
	connectorKey           string
	connectorVersion       string
	releaseDigest          string
	generation             market.HostGeneration
	mcpTools               map[string]registeredMCPTool
	closeMu                sync.Mutex
	mcpClient              *mcp.StdioClient
	remoteMCP              RemoteMCPClient
	executionRoot          string
	installedRoot          string
	displayName            string
	description            string
	routingAliases         []string
	skillRoot              string
	skills                 []connectorartifact.SkillSummary
	processes              *connectorruntime.ProcessGroup
	snapshots              *connectorruntime.ExecutionSnapshotter
	userHome               string
	cliLaunch              *managedCLILaunch
	cliCommand             string
	cliInvocationCommand   string
	cliContractHash        string
	cliCommands            []market.CLICommand
	cliShimPath            string
	cliShimContent         []byte
	credentialBrokerLaunch *managedCredentialBrokerLaunch
	readiness              market.RuntimeReadiness
}

func New(config Config) (*Host, error) {
	if config.Artifacts == nil || config.Runtimes == nil || config.Processes == nil || config.Registry == nil || config.MCP == nil {
		return nil, errors.New("connector implementation host dependencies are required")
	}
	if !filepath.IsAbs(strings.TrimSpace(config.StateRoot)) {
		return nil, errors.New("connector implementation state root must be absolute")
	}
	if !filepath.IsAbs(strings.TrimSpace(config.UserHome)) {
		return nil, errors.New("connector implementation user home must be absolute")
	}
	if strings.TrimSpace(config.BinDir) == "" {
		config.BinDir = filepath.Join(config.StateRoot, "bin")
	}
	if !filepath.IsAbs(config.BinDir) {
		return nil, errors.New("connector CLI bin directory must be absolute")
	}
	if config.MCPStartupTimeout <= 0 {
		config.MCPStartupTimeout = 15 * time.Second
	}
	snapshots, err := connectorruntime.NewExecutionSnapshotter(config.StateRoot)
	if err != nil {
		return nil, err
	}
	if err := snapshots.CleanupOrphans(); err != nil {
		return nil, fmt.Errorf("clean orphaned connector execution snapshots: %w", err)
	}
	routes := connectorruntime.NewRouteTable()
	planner, err := connectorruntime.NewManagedRoutePlanner(connectorruntime.ManagedRoutePlannerConfig{
		StateRoot: config.StateRoot, UserHome: config.UserHome, Runtimes: config.Runtimes, CLIInstallations: config.CLIInstallations,
	})
	if err != nil {
		return nil, err
	}
	config.Registry.attach(routes)
	config.MCP.attach(routes)
	host := &Host{
		artifacts:              config.Artifacts,
		planner:                planner,
		processes:              config.Processes,
		routeObserver:          config.Routes,
		authorizationObserver:  config.Authorization,
		mcpStartupTimeout:      config.MCPStartupTimeout,
		routes:                 routes,
		snapshots:              snapshots,
		authorizationRoutes:    make(map[string]*connectorRoute),
		connectorLanes:         make(map[string]*sync.Mutex),
		remoteMCPClientFactory: config.RemoteMCPClientFactory,
		mcpRegistry:            config.MCP,
		registry:               config.Registry,
		binDir:                 config.BinDir,
		remoteMCPTools:         newRemoteMCPToolCache(),
	}
	host.authorizationProvider = newManagedCredentialAuthorizationProvider(host)
	return host, nil
}

func (host *Host) Reconcile(ctx context.Context, request ReconcileRequest) (market.RuntimeReceipt, error) {
	runtimeRequest := request.Runtime
	if host == nil || !hostIdentityPattern.MatchString(runtimeRequest.ConnectionID) ||
		!hostIdentityPattern.MatchString(runtimeRequest.Connector.Key) || runtimeRequest.Generation.BootEpoch == "" || runtimeRequest.Generation.Generation == 0 {
		return market.RuntimeReceipt{}, errors.New("connector runtime reconcile identity is invalid")
	}
	releaseLane := host.enterConnectorLane(runtimeRequest.Connector.Key)
	defer releaseLane()
	if !runtimeRequest.Enabled {
		// A connection id can rotate while an older account route is still
		// present. Disabled is connector-level desired state, so convergence must
		// fence every route and authorization session for the connector instead
		// of removing only the latest route key.
		if err := host.deactivateConnector(market.RuntimeDeactivationRequest{
			ConnectionID:   runtimeRequest.ConnectionID,
			ConnectorKey:   runtimeRequest.Connector.Key,
			AllConnections: true,
			Generation:     runtimeRequest.Generation,
			Deadline:       time.Now().Add(3 * time.Second),
		}); err != nil {
			return market.RuntimeReceipt{}, err
		}
		return market.RuntimeReceipt{OperationID: runtimeRequest.OperationID, ConnectionID: runtimeRequest.ConnectionID,
			ConnectorKey: runtimeRequest.Connector.Key, ReleaseDigest: runtimeRequest.Connector.Release.ReleaseDigest,
			Generation: runtimeRequest.Generation,
			Readiness: market.RuntimeReadiness{State: market.RuntimeReadinessBlocked,
				ReasonCode: market.RuntimeReadinessReasonRuntimeDisabled}}, nil
	}
	if err := market.ValidateRuntimeReleaseShape(runtimeRequest.Connector.Release); err != nil {
		return market.RuntimeReceipt{}, err
	}
	key := connectorRouteKey(runtimeRequest.ConnectionID, runtimeRequest.Connector.Key)
	if !installationTargetsRelease(runtimeRequest.Connector.Installation, runtimeRequest.Connector.Release.ReleaseDigest) {
		return market.RuntimeReceipt{}, errors.New("connector installed release is not active")
	}
	if err := host.validateAuthorization(runtimeRequest); err != nil {
		return market.RuntimeReceipt{}, err
	}
	var route *connectorRoute
	var err error
	if runtimeRequest.Connector.Release.Manifest.Implementation.Kind == market.ImplementationKindRemoteStreamableHTTP {
		route, err = host.buildRemoteRoute(ctx, runtimeRequest)
		if err == nil {
			prepared, resolveErr := host.artifacts.ResolvePrepared(ctx, runtimeRequest.Connector.Release)
			if resolveErr != nil {
				_ = route.Close(time.Now().Add(3 * time.Second))
				return market.RuntimeReceipt{}, fmt.Errorf("resolve remote connector artifact: %w", resolveErr)
			}
			route.installedRoot = prepared.PreparedPath
		}
	} else {
		prepared, resolveErr := host.artifacts.ResolvePrepared(ctx, runtimeRequest.Connector.Release)
		if resolveErr != nil {
			return market.RuntimeReceipt{}, fmt.Errorf("resolve prepared connector artifact: %w", resolveErr)
		}
		installedRoot := prepared.PreparedPath
		executionRoot, snapshotErr := host.snapshots.Create(prepared, artifactNativeEntrypoints(runtimeRequest.Connector.Release)...)
		if snapshotErr != nil {
			return market.RuntimeReceipt{}, fmt.Errorf("create connector execution snapshot: %w", snapshotErr)
		}
		prepared.PreparedPath = executionRoot
		route, err = host.buildManagedRoute(ctx, runtimeRequest, prepared)
		if err != nil {
			_ = host.snapshots.Remove(executionRoot)
			return market.RuntimeReceipt{}, err
		}
		route.executionRoot, route.installedRoot = executionRoot, installedRoot
	}
	if err != nil {
		return market.RuntimeReceipt{}, wrapRemoteMCPAuthorizationError(err)
	}
	route.displayName = runtimeRequest.Connector.Release.Manifest.DisplayName
	route.description = runtimeRequest.Connector.Release.Manifest.Description
	if routing := runtimeRequest.Connector.Release.Manifest.AgentRouting; routing != nil {
		route.routingAliases = append([]string(nil), routing.Aliases...)
	}
	route.snapshots = host.snapshots
	skillProjection, err := connectorartifact.InspectSkills(route.installedRoot)
	if err != nil {
		_ = route.Close(time.Now().Add(3 * time.Second))
		return market.RuntimeReceipt{}, fmt.Errorf("inspect connector Skills: %w", err)
	}
	route.skillRoot = skillProjection.Root
	route.skills = append([]connectorartifact.SkillSummary(nil), skillProjection.Skills...)
	if route.readiness.State == "" {
		route.readiness = readyRuntimeReadiness(route)
	}
	previous, _ := host.routes.Route(key).(*connectorRoute)
	if err := route.activateCLIShim(); err != nil {
		_ = route.Close(time.Now().Add(3 * time.Second))
		return market.RuntimeReceipt{}, err
	}
	if err := host.routes.Commit(route); err != nil {
		if previous != nil {
			_ = previous.activateCLIShim()
		} else {
			route.removeCLIShimIfCurrent()
		}
		_ = route.Close(time.Now().Add(3 * time.Second))
		return market.RuntimeReceipt{}, err
	}
	host.releaseAuthorizationRouteByKey(key)
	host.notifyRouteChanged()
	if route.mcpClient != nil {
		go host.monitorMCPRoute(route, route.mcpClient)
	}
	summary := connectorSummaryFromDescriptor(routeDescriptor(route))
	return market.RuntimeReceipt{OperationID: runtimeRequest.OperationID, ConnectionID: runtimeRequest.ConnectionID,
		ConnectorKey: runtimeRequest.Connector.Key, ReleaseDigest: route.releaseDigest,
		Generation: runtimeRequest.Generation, Readiness: cloneRuntimeReadiness(route.readiness), Summary: &summary}, nil
}

func installationTargetsRelease(installation market.Installation, releaseDigest string) bool {
	switch installation.State {
	case market.InstallationStateInstalled:
		return installation.InstalledReleaseDigest == releaseDigest
	case market.InstallationStateInstalling, market.InstallationStateUpdating:
		return installation.CandidateReleaseDigest == releaseDigest
	default:
		return false
	}
}

func (*Host) validateAuthorization(request market.RuntimeReconcileRequest) error {
	authKind := request.Connector.Release.Manifest.AuthorizationKind
	if request.Connector.Release.Manifest.Implementation.Kind == market.ImplementationKindRemoteStreamableHTTP {
		if authKind == "none" {
			if request.Connector.Authorization.State != market.AuthorizationStateNotRequired {
				return errors.New("authorization-free connector has an invalid credential binding")
			}
			return nil
		}
		if request.Connector.Authorization.State != market.AuthorizationStateConnected {
			return market.NewDomainError(
				market.ErrorCodeAuthorizationFailed,
				"authorized remote connector is not connected",
				false,
				nil,
			)
		}
		return nil
	}
	managed := request.Connector.Release.Manifest.Implementation.ManagedStdio
	if authKind == "none" {
		if request.Connector.Authorization.State != market.AuthorizationStateNotRequired {
			return errors.New("authorization-free connector has an invalid credential binding")
		}
		return nil
	}
	if managed == nil || managed.CredentialBroker == nil {
		return errors.New("authorized connector credential broker binding is unavailable")
	}
	if request.Connector.Authorization.State != market.AuthorizationStateConnected {
		return market.NewDomainError(
			market.ErrorCodeAuthorizationFailed,
			"authorized managed connector is not connected",
			false,
			nil,
		)
	}
	return nil
}

func newConnectorRoute(request market.RuntimeReconcileRequest) *connectorRoute {
	return &connectorRoute{id: connectorRouteKey(request.ConnectionID, request.Connector.Key), connectionID: request.ConnectionID,
		connectorKey: request.Connector.Key, connectorVersion: request.Connector.Release.Version,
		releaseDigest: request.Connector.Release.ReleaseDigest,
		generation:    request.Generation, mcpTools: make(map[string]registeredMCPTool),
		processes: connectorruntime.NewProcessGroup()}
}

func (host *Host) Close() error {
	if host == nil {
		return nil
	}
	deadline := time.Now().Add(3 * time.Second)
	host.admission.Lock()
	defer host.admission.Unlock()
	host.authorizationMu.Lock()
	authorizationRoutes := make([]*connectorRoute, 0, len(host.authorizationRoutes))
	for key, route := range host.authorizationRoutes {
		delete(host.authorizationRoutes, key)
		authorizationRoutes = append(authorizationRoutes, route)
	}
	host.authorizationMu.Unlock()
	errs := []error{host.routes.Close(deadline)}
	for _, route := range authorizationRoutes {
		route.Fence()
		errs = append(errs, route.Close(deadline))
	}
	host.remoteMCPTools.clear()
	return errors.Join(errs...)
}

func (host *Host) SetCapabilityPublication(enabled bool) {
	if host != nil {
		host.routes.SetPublished(enabled)
		host.notifyRouteChanged()
	}
}

func (host *Host) FenceAll(_ context.Context, deadline time.Time) error {
	if host == nil {
		return nil
	}
	host.admission.Lock()
	defer host.admission.Unlock()
	err := host.routes.FenceAll(deadline)
	host.notifyRouteChanged()
	return err
}

func (host *Host) FailClosed(ctx context.Context, deadline time.Time) error {
	if host == nil {
		return nil
	}
	host.SetCapabilityPublication(false)
	return host.FenceAll(ctx, deadline)
}

func (host *Host) DeactivateRuntime(ctx context.Context, request market.RuntimeDeactivationRequest) error {
	if host == nil {
		return errors.New("connector implementation host is unavailable")
	}
	if !request.Deadline.IsZero() && time.Now().After(request.Deadline) {
		return context.DeadlineExceeded
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	releaseLane := host.enterConnectorLane(request.ConnectorKey)
	defer releaseLane()
	if request.AllConnections {
		return host.deactivateConnector(request)
	}
	err := host.routes.Remove(connectorRouteKey(request.ConnectionID, request.ConnectorKey), request.Generation, request.ReleaseDigest, request.Deadline)
	host.notifyRouteChanged()
	return err
}

func (host *Host) enterConnectorLane(connectorKey string) func() {
	host.admission.RLock()
	host.laneMu.Lock()
	if host.connectorLanes == nil {
		host.connectorLanes = make(map[string]*sync.Mutex)
	}
	lane := host.connectorLanes[connectorKey]
	if lane == nil {
		lane = &sync.Mutex{}
		host.connectorLanes[connectorKey] = lane
	}
	host.laneMu.Unlock()
	lane.Lock()
	return func() {
		lane.Unlock()
		host.admission.RUnlock()
	}
}

func (host *Host) buildManagedRoute(ctx context.Context, request market.RuntimeReconcileRequest,
	prepared market.PreparedArtifactReceipt) (*connectorRoute, error) {
	plan, err := host.planner.Build(ctx, request, prepared)
	if err != nil {
		return nil, err
	}
	route := newConnectorRoute(request)
	route.userHome = plan.UserHome
	if plan.Managed.MCP != nil {
		if err := host.attachMCP(ctx, route, plan.Managed, prepared, plan.Executable, plan.StateDir, plan.UserHome, plan.ArtifactTrees); err != nil {
			_ = route.close(time.Now().Add(3 * time.Second))
			return nil, err
		}
	}
	if plan.Managed.CLI != nil {
		if err := host.attachCLI(route, plan.Managed, prepared, plan.InstalledCLI, plan.Executable, plan.StateDir, plan.ArtifactTrees); err != nil {
			_ = route.close(time.Now().Add(3 * time.Second))
			return nil, err
		}
		if err := route.prepareCLIShim(host.binDir); err != nil {
			_ = route.close(time.Now().Add(3 * time.Second))
			return nil, err
		}
		if err := host.checkCLIReadiness(ctx, route, plan.Managed.CLI.ReadinessProbe); err != nil {
			_ = route.close(time.Now().Add(3 * time.Second))
			return nil, fmt.Errorf("check connector CLI readiness: %w", err)
		}
	}
	if plan.Managed.CredentialBroker != nil {
		if err := host.attachCredentialBroker(route, plan.Managed.CredentialBroker, prepared, plan.Executable, plan.StateDir, plan.ArtifactTrees); err != nil {
			_ = route.close(time.Now().Add(3 * time.Second))
			return nil, err
		}
	}
	if route.cliLaunch == nil && len(route.mcpTools) == 0 {
		_ = route.close(time.Now().Add(3 * time.Second))
		return nil, errors.New("connector implementation exposed no MCP tools or CLI commands")
	}
	route.readiness = readyRuntimeReadiness(route)
	return route, nil
}

var mcpLocalToolNamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:-]*$`)
var hostIdentityPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,190}$`)

func (*Host) attachCredentialBroker(route *connectorRoute, broker *market.ManagedCredentialBroker,
	prepared market.PreparedArtifactReceipt, executable connectorruntime.ConnectorExecutable,
	stateDir string, artifactTrees []agentruntime.ArtifactTreeIdentity) error {
	if route.cliLaunch == nil {
		return errors.New("connector credential broker requires a managed CLI")
	}
	entrypoint, err := connectorruntime.PreparedEntrypoint(prepared.PreparedPath, broker.Entrypoint)
	if err != nil {
		return fmt.Errorf("resolve connector credential broker entrypoint: %w", err)
	}
	allowedHosts := make(map[string]struct{}, len(broker.AllowedHosts))
	for _, allowedHost := range broker.AllowedHosts {
		allowedHosts[strings.ToLower(strings.TrimSpace(allowedHost))] = struct{}{}
	}
	route.credentialBrokerLaunch = &managedCredentialBrokerLaunch{
		entrypoint: entrypoint, timeout: time.Duration(broker.TimeoutMS) * time.Millisecond, allowedHosts: allowedHosts,
		cliLaunch: credentialBrokerCLILaunch{Executable: route.cliLaunch.executable.Path,
			// Native CLIs legitimately have no argv; preserve [] instead of encoding null for the broker protocol.
			Arguments: append([]string{}, route.cliLaunch.arguments...), CWD: route.cliLaunch.cwd},
		executable: executable, language: route.cliLaunch.language, cwd: prepared.PreparedPath, stateDir: stateDir,
		artifactTrees: append([]agentruntime.ArtifactTreeIdentity(nil), artifactTrees...),
	}
	return nil
}

func (host *Host) attachMCP(ctx context.Context, route *connectorRoute, managed *market.ManagedStdioImplementation,
	prepared market.PreparedArtifactReceipt, executable connectorruntime.ConnectorExecutable,
	stateDir, userHome string, artifactTrees []agentruntime.ArtifactTreeIdentity) error {
	entrypoint, err := connectorruntime.PreparedEntrypoint(prepared.PreparedPath, managed.MCP.Entrypoint)
	if err != nil {
		return err
	}
	startupContext, cancelStartup := context.WithTimeout(ctx, host.mcpStartupTimeout)
	defer cancelStartup()
	spec := connectorruntime.ConnectorProcessSpec(route.connectionID, route.connectorKey, managed.Runtime.Language,
		executable, prepared.PreparedPath, append([]string{entrypoint}, managed.MCP.Arguments...), stateDir, userHome, artifactTrees)
	connection, processID, err := host.startProcess(startupContext, route, spec, false)
	if err != nil {
		return fmt.Errorf("start connector MCP process: %w", err)
	}
	release := func() { _ = route.releaseProcess(processID, connection) }
	client, err := mcp.NewStdioClient(mcp.StdioClientConfig{Connection: connection, ProcessName: route.connectorKey + " MCP"})
	if err != nil {
		release()
		return err
	}
	if _, err := client.Call(startupContext, "initialize", map[string]any{"protocolVersion": "2025-06-18", "capabilities": map[string]any{},
		"clientInfo": map[string]any{"name": "tutti-connector-host", "version": "1"}}); err != nil {
		release()
		return fmt.Errorf("initialize connector MCP process: %w", err)
	}
	if err := client.Notify("notifications/initialized", map[string]any{}); err != nil {
		release()
		return err
	}
	tools, err := listMCPTools(startupContext, client)
	if err != nil {
		release()
		return err
	}
	if len(tools) == 0 {
		release()
		return errors.New("connector MCP tools/list response is invalid")
	}
	if err := host.registerMCPTools(route, client, tools); err != nil {
		release()
		return err
	}
	route.mcpClient = client
	return nil
}

func readyRuntimeReadiness(route *connectorRoute) market.RuntimeReadiness {
	interfaces := make([]market.InterfaceReadiness, 0, 2)
	if len(route.mcpTools) > 0 {
		routeIDs := make([]string, 0, len(route.mcpTools))
		for _, tool := range route.mcpTools {
			routeIDs = append(routeIDs, tool.routeID)
		}
		sort.Strings(routeIDs)
		interfaces = append(interfaces, market.InterfaceReadiness{Kind: "mcp", State: market.RuntimeReadinessReady, RouteIDs: routeIDs})
	}
	if route.cliLaunch != nil {
		interfaces = append(interfaces, market.InterfaceReadiness{Kind: "cli", State: market.RuntimeReadinessReady})
	}
	return market.RuntimeReadiness{State: market.RuntimeReadinessReady, Interfaces: interfaces}
}

func cloneRuntimeReadiness(readiness market.RuntimeReadiness) market.RuntimeReadiness {
	cloned := readiness
	cloned.Interfaces = append([]market.InterfaceReadiness(nil), readiness.Interfaces...)
	for index := range cloned.Interfaces {
		cloned.Interfaces[index].RouteIDs = append([]string(nil), readiness.Interfaces[index].RouteIDs...)
	}
	return cloned
}

func (host *Host) notifyRouteChanged() {
	if host == nil {
		return
	}
	host.registry.notifyChanged()
	host.mcpRegistry.notifyChanged()
}

func (host *Host) monitorMCPRoute(route *connectorRoute, client *mcp.StdioClient) {
	<-client.Done()
	unexpected := host.routes.IsCurrent(route)
	_ = host.routes.RetireExact(route, time.Now().Add(3*time.Second))
	host.notifyRouteChanged()
	if unexpected && host.routeObserver != nil {
		host.routeObserver.ObserveRoute(context.Background(), RouteObservation{
			ConnectorKey: route.connectorKey, ConnectionID: route.connectionID,
			ReleaseDigest: route.releaseDigest, Generation: route.generation, ObservedAt: time.Now().UTC(),
		})
	}
}

func (*Host) attachCLI(route *connectorRoute, managed *market.ManagedStdioImplementation,
	prepared market.PreparedArtifactReceipt, installed *market.CLIInstallationReceipt,
	executable connectorruntime.ConnectorExecutable, stateDir string, artifactTrees []agentruntime.ArtifactTreeIdentity) error {
	entrypointRoot, entrypointRelative := prepared.PreparedPath, managed.CLI.Entrypoint
	if installed != nil {
		entrypointRoot, entrypointRelative = installed.InstallRoot, installed.Entrypoint
	}
	entrypoint, err := connectorruntime.PreparedEntrypoint(entrypointRoot, entrypointRelative)
	if err != nil {
		return err
	}
	contractHash, err := market.ManagedCLIContractHash(*managed.CLI)
	if err != nil {
		return fmt.Errorf("hash connector CLI contract: %w", err)
	}
	launchArguments := []string{entrypoint}
	launchExecutable := executable
	if managed.CLI.Launch != nil && managed.CLI.Launch.Kind == market.CLIArtifactLaunchKindNative {
		launchArguments = nil
		launchExecutable = connectorruntime.ConnectorExecutable{Path: entrypoint, SHA256: managed.CLI.Launch.SHA256,
			SizeBytes: managed.CLI.Launch.SizeBytes}
	} else if installed != nil && installed.LaunchKind == "native" {
		launchArguments = nil
		launchExecutable = connectorruntime.ConnectorExecutable{Path: entrypoint, SHA256: installed.EntrypointSHA256,
			SizeBytes: installed.EntrypointSize}
	}
	route.cliLaunch = &managedCLILaunch{arguments: append(append([]string{}, launchArguments...), managed.CLI.Arguments...),
		artifactTrees: append([]agentruntime.ArtifactTreeIdentity(nil), artifactTrees...), cwd: prepared.PreparedPath,
		executable: launchExecutable, language: managed.Runtime.Language, stateDir: stateDir,
		timeout: time.Duration(managed.CLI.TimeoutMS) * time.Millisecond}
	route.cliContractHash = contractHash
	route.cliInvocationCommand = market.ManagedCLICommandName(*managed.CLI)
	route.cliCommands = cloneCLICommands(managed.CLI.Commands)
	return nil
}

func artifactNativeEntrypoints(release market.Release) []string {
	managed := release.Manifest.Implementation.ManagedStdio
	if managed == nil || managed.CLI == nil || managed.CLI.Launch == nil ||
		managed.CLI.Launch.Kind != market.CLIArtifactLaunchKindNative {
		return nil
	}
	return []string{managed.CLI.Entrypoint}
}

func (host *Host) startProcess(ctx context.Context, route *connectorRoute, spec agentruntime.ProcessSpec,
	requireCurrent bool) (agentruntime.ProcessConnection, uint64, error) {
	if requireCurrent && !host.routes.IsCurrent(route) {
		return nil, 0, command.ErrServiceUnavailable
	}
	startContext, processID, accepted := route.processes.Begin(context.Background())
	if !accepted {
		return nil, 0, command.ErrServiceUnavailable
	}
	type startResult struct {
		connection agentruntime.ProcessConnection
		err        error
	}
	result := make(chan startResult, 1)
	go func() {
		connection, startErr := host.processes.Start(startContext, spec)
		result <- startResult{connection: connection, err: startErr}
	}()
	select {
	case started := <-result:
		if started.err != nil {
			route.processes.FailStart(processID)
			return nil, 0, started.err
		}
		if !route.processes.CommitStart(processID, started.connection) {
			_ = started.connection.Close()
			return nil, 0, command.ErrServiceUnavailable
		}
		return started.connection, processID, nil
	case <-ctx.Done():
		route.processes.FailStart(processID)
		go func() {
			started := <-result
			if started.connection != nil {
				_ = started.connection.Close()
			}
		}()
		return nil, 0, ctx.Err()
	}
}

func connectorRouteKey(connectionID, connectorKey string) string {
	return connectionID + "\x00" + connectorKey
}

func (host *Host) routeCurrent(route *connectorRoute) bool {
	return host.routes.IsCurrent(route) && !route.processes.IsFenced()
}

func (route *connectorRoute) RouteID() string                        { return route.id }
func (route *connectorRoute) RouteGeneration() market.HostGeneration { return route.generation }
func (route *connectorRoute) RouteReleaseDigest() string             { return route.releaseDigest }
func (route *connectorRoute) Fence()                                 { route.processes.Fence() }
func (route *connectorRoute) close(deadline time.Time) error         { return route.Close(deadline) }
func (route *connectorRoute) releaseProcess(id uint64, connection agentruntime.ProcessConnection) error {
	if route != nil && route.processes != nil {
		return route.processes.ReleaseWithError(id, connection)
	}
	return nil
}

func (route *connectorRoute) Close(deadline time.Time) error {
	if route == nil {
		return nil
	}
	route.closeMu.Lock()
	defer route.closeMu.Unlock()
	route.removeCLIShimIfCurrent()
	var remoteErr error
	if route.remoteMCP != nil {
		closeCtx, cancel := context.WithDeadline(context.Background(), deadline)
		remoteErr = route.remoteMCP.Close(closeCtx)
		cancel()
	}
	closeErr := route.processes.Close(deadline)
	if closeErr == nil {
		closeErr = remoteErr
	}
	if closeErr == nil && route.executionRoot != "" {
		if err := route.snapshots.Remove(route.executionRoot); err != nil {
			closeErr = err
		} else {
			route.executionRoot = ""
		}
	}
	return closeErr
}

func (route *connectorRoute) prepareCLIShim(binDir string) error {
	if route == nil || route.cliLaunch == nil {
		return nil
	}
	command := "tutti-connector-" + route.connectorKey
	if runtime.GOOS == "windows" {
		command += ".cmd"
	}
	path := filepath.Join(binDir, command)
	content, err := connectorCLIShimContent(route)
	if err != nil {
		return err
	}
	route.cliCommand, route.cliShimPath, route.cliShimContent = command, path, content
	return nil
}

func (route *connectorRoute) activateCLIShim() error {
	if route == nil || route.cliShimPath == "" || len(route.cliShimContent) == 0 {
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(route.cliShimPath), 0o700); err != nil {
		return fmt.Errorf("create connector CLI bin directory: %w", err)
	}
	temporary := route.cliShimPath + ".tmp-" + fmt.Sprintf("%d", route.generation.Generation)
	if err := os.WriteFile(temporary, route.cliShimContent, 0o700); err != nil {
		return fmt.Errorf("write connector CLI shim: %w", err)
	}
	if err := os.Rename(temporary, route.cliShimPath); err != nil {
		_ = os.Remove(temporary)
		return fmt.Errorf("activate connector CLI shim: %w", err)
	}
	return nil
}

func (route *connectorRoute) removeCLIShimIfCurrent() {
	if route == nil || route.cliShimPath == "" {
		return
	}
	current, err := os.ReadFile(route.cliShimPath)
	if err == nil && string(current) == string(route.cliShimContent) {
		_ = os.Remove(route.cliShimPath)
	}
}

func connectorCLIShimContent(route *connectorRoute) ([]byte, error) {
	launch := route.cliLaunch
	if launch == nil || strings.TrimSpace(launch.executable.Path) == "" {
		return nil, errors.New("connector CLI launch is unavailable")
	}
	arguments := append([]string(nil), launch.arguments...)
	if runtime.GOOS == "windows" {
		values := append([]string{launch.executable.Path}, arguments...)
		for _, value := range append(values, launch.cwd, launch.stateDir, route.userHome) {
			if strings.ContainsAny(value, "\r\n\"") {
				return nil, errors.New("connector CLI path cannot be represented by Windows shim")
			}
		}
		quoted := make([]string, 0, len(values))
		for _, value := range values {
			quoted = append(quoted, `"`+value+`"`)
		}
		content := "@echo off\r\n" +
			"set \"TUTTI_CONNECTOR_CONNECTION_ID=" + route.connectionID + "\"\r\n" +
			"set \"TUTTI_CONNECTOR_KEY=" + route.connectorKey + "\"\r\n" +
			"set \"TUTTI_CONNECTOR_LANGUAGE=" + launch.language + "\"\r\n" +
			"set \"TUTTI_CONNECTOR_STATE_DIR=" + launch.stateDir + "\"\r\n" +
			"set \"HOME=" + route.userHome + "\"\r\n" +
			"set \"USERPROFILE=" + route.userHome + "\"\r\n" +
			"cd /d \"" + launch.cwd + "\"\r\n" + strings.Join(quoted, " ") + " %*\r\n"
		return []byte(content), nil
	}
	values := append([]string{launch.executable.Path}, arguments...)
	quoted := make([]string, 0, len(values))
	for _, value := range values {
		quoted = append(quoted, shellQuote(value))
	}
	content := "#!/bin/sh\n" +
		"export TUTTI_CONNECTOR_CONNECTION_ID=" + shellQuote(route.connectionID) + "\n" +
		"export TUTTI_CONNECTOR_KEY=" + shellQuote(route.connectorKey) + "\n" +
		"export TUTTI_CONNECTOR_LANGUAGE=" + shellQuote(launch.language) + "\n" +
		"export TUTTI_CONNECTOR_STATE_DIR=" + shellQuote(launch.stateDir) + "\n" +
		"export HOME=" + shellQuote(route.userHome) + "\n" +
		"export USERPROFILE=" + shellQuote(route.userHome) + "\n" +
		"cd " + shellQuote(launch.cwd) + " || exit 1\n" +
		"exec " + strings.Join(quoted, " ") + " \"$@\"\n"
	return []byte(content), nil
}

func shellQuote(value string) string {
	return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
}

var _ connectorruntime.ManagedRoute = (*connectorRoute)(nil)