package daemon import ( "bytes" "context" "encoding/json" "errors" "fmt" "log/slog" "net" "sort" "strings" "sync" "time" "github.com/gorilla/websocket" "github.com/tidbcloud/AgentTeam/internal/api/dto" "github.com/tidbcloud/AgentTeam/internal/api/dto/converter" "github.com/tidbcloud/AgentTeam/internal/api/im" "github.com/tidbcloud/AgentTeam/internal/core/catalog" "github.com/tidbcloud/AgentTeam/internal/core/secrets" "github.com/tidbcloud/AgentTeam/internal/observability/logging" "github.com/tidbcloud/AgentTeam/internal/runtimecontrol" "github.com/tidbcloud/AgentTeam/internal/store/agent" "github.com/tidbcloud/AgentTeam/internal/store/app/runtime" "github.com/tidbcloud/AgentTeam/internal/store/message" "github.com/tidbcloud/AgentTeam/internal/store/rewind" "github.com/tidbcloud/AgentTeam/internal/store/workflow/run" ) type envelope struct { Type string `json:"type"` } type realtimeEmitter interface { EmitMachineStatus(workspaceID string, machineID string, status string) EmitMachineCapabilities(workspaceID string, machineID string, runtimes []string, runtimeDefaults map[string]runtime.RuntimeDefaults, hostname string, os string, daemonVersion string) EmitDaemonStatus(workspaceID string, daemonID string, status string) EmitAgentStatusWithInstance(workspaceID string, ownerUserID string, agentID string, status string, currentInstance *dto.AppAgentInstance) EmitAgentActivityWithInstance(workspaceID string, ownerUserID string, agentID string, activity string, detail string, currentInstance *dto.AppAgentInstance) EmitAgentSessionWithInstance(workspaceID string, ownerUserID string, agentID string, sessionID string, currentInstance *dto.AppAgentInstance) EmitAgentWakeStatus(workspaceID string, agentID string, agentName string, agentDisplayName string, messageID string, wakeToken string, target string, status string, replyMessageID string, detail string, timestamp string) EmitAgentTrajectory(workspaceID string, ownerUserID string, agentID string, sessionID string, timestamp int64, entries json.RawMessage) } // Store is the daemon protocol storage port. It keeps websocket handling // decoupled from the root store facade while preserving the existing runtime // semantics. type Store interface { MarkMachineDisconnected(ctx context.Context, machineID string) error PrepareAgentStart(ctx context.Context, agentID string, publicBaseURL string, wakeMessage *message.ReceiveMessage, allowWorkspaceWrite bool, includeLegacySession bool, useDirectMaaS bool) (agent.StartCommand, error) MarkMachineReady(ctx context.Context, identity agent.MachineIdentity, payload agent.ReadyPayload) error UpdateAgentStatusFromMachine(ctx context.Context, machineID string, agentID string, status string) (bool, error) UpdateAgentActivityFromMachine(ctx context.Context, machineID string, agentID string, activity string, detail string) (bool, error) UpdateAgentSessionFromMachine(ctx context.Context, machineID string, agentID string, sessionID string, persistLegacySession bool) (bool, error) AgentBelongsToMachine(ctx context.Context, machineID string, agentID string) (bool, error) AgentNamesForMachine(ctx context.Context, machineID string, agentID string) (string, string, bool, error) AgentPermissionModeForMachine(ctx context.Context, machineID string, agentID string) (string, bool, error) MachineIDForAgent(ctx context.Context, workspaceID string, agentID string) (string, error) ClearWorkspaceWriteAgentAuthTokensForMachine(ctx context.Context, machineID string) error RecordMessageRecoveryPointAndPruneOtherThreads(ctx context.Context, workspaceID string, messageID string, agentID string, threadID string, strategy string, recoveryRef string) ([]rewind.MessageRecoveryPoint, error) ListAgentsForMachine(ctx context.Context, machineID string) ([]agent.MachineAgent, error) GetTaskActiveTarget(ctx context.Context, workspaceID string, scope string, targetKey string) (message.ReceiveActiveTaskTarget, bool, error) } // RuntimeEventStore is the daemon protocol storage port for runtime event // persistence. It is separated from Store so the root store facade does not need // to retain agent-ops methods. type RuntimeEventStore interface { RecordAgentRuntimeEvent(ctx context.Context, event agent.AgentRuntimeEvent) error } type readyMessage struct { Type string `json:"type"` Capabilities []string `json:"capabilities"` TokenRefreshProof string `json:"tokenRefreshProof"` Runtimes []string `json:"runtimes"` RuntimeDefaults map[string]runtime.RuntimeDefaults `json:"runtimeDefaults"` RunningAgents []string `json:"runningAgents"` RunningAgentSessions map[string]string `json:"runningAgentSessions,omitempty"` RunningAgentTurnStates map[string]string `json:"runningAgentTurnStates,omitempty"` RunningAgentTurnStatesWatermark *int64 `json:"runningAgentTurnStatesWatermark,omitempty"` Hostname string `json:"hostname"` OS string `json:"os"` DaemonVersion string `json:"daemonVersion"` DaemonTransport daemonTransportReady `json:"daemonTransport,omitempty"` } type daemonTransportReady struct { SchemaVersion int `json:"schemaVersion"` SessionGeneration string `json:"sessionGeneration"` MaxElasticRequestLinks int `json:"maxElasticRequestLinks"` HardMaxElasticRequestLinks int `json:"hardMaxElasticRequestLinks"` SupportedLinkModes []string `json:"supportedLinkModes"` EndpointRegistryVersion string `json:"endpointRegistryVersion"` SupportedEndpoints []daemonTransportEndpointRef `json:"supportedEndpoints"` } type daemonTransportEndpointRef struct { EndpointID string `json:"endpointId"` Version int `json:"version"` } func runtimeDefaultsToCore(defaults map[string]runtime.RuntimeDefaults) map[string]agent.RuntimeDefaults { if len(defaults) == 0 { return map[string]agent.RuntimeDefaults{} } converted := make(map[string]agent.RuntimeDefaults, len(defaults)) for runtime, item := range defaults { converted[strings.TrimSpace(runtime)] = agent.RuntimeDefaults{ Model: strings.TrimSpace(item.Model), ReasoningEffort: strings.TrimSpace(item.ReasoningEffort), } } return converted } type agentStatusMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` Status string `json:"status"` // ExpectedSessionID fences a terminal lifecycle frame (Status "inactive"): // the frame is applied only while the Hub's current agent session for this // agent still matches this expected/prior session identity. A delayed stop // from a replaced generation is dropped entirely. A frame without this // optional field (legacy daemon) keeps the unconditional behavior. ExpectedSessionID string `json:"expectedSessionId,omitempty"` } type agentActivityMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` Activity string `json:"activity"` Detail string `json:"detail"` // ExpectedSessionID fences a terminal lifecycle frame (Activity "offline"), // with the same generation-fence semantics as agentStatusMessage. ExpectedSessionID string `json:"expectedSessionId,omitempty"` } type agentActivityState struct { Activity string Detail string } type agentSessionMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` SessionID string `json:"sessionId"` // ExpectedSessionID fences a terminal clear (SessionID empty): the clear is // applied only while the Hub's current agent session for this agent still // matches this expected/prior session identity. A delayed stop from a replaced // generation is dropped entirely, so it can never delete the replacement's // session. A clear without this optional fence (legacy daemon) keeps the // unconditional behavior. ExpectedSessionID string `json:"expectedSessionId,omitempty"` } // agentTurnStateMessage carries the daemon-derived real Agent turn execution // state (active/queued/idle) used by Workflow callback recovery. It is the // authoritative recovery working signal, distinct from process/status/activity // presence. SessionID is the RuntimeSessionID of the snapshot that produced the // state; the server drops a turn_state whose sessionId does not match the // current agent session (stale fence). Epoch is the daemon transport session // generation of the connection that produced the event; the server drops an // event whose epoch does not match the current connection's session generation // so an old-connection event can never overwrite the current snapshot. Revision // is the same-connection monotonic ordering revision (strictly newer than the // ready snapshot watermark and every earlier increment); the server drops an // event whose revision is older than or equal to the applied high-water so an // old/equal increment can never overwrite a newer snapshot. type agentTurnStateMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` TurnState string `json:"turnState"` SessionID string `json:"sessionId"` Epoch string `json:"epoch,omitempty"` Revision int64 `json:"revision,omitempty"` } type agentWakeStatusMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` MessageID string `json:"messageId"` WakeToken string `json:"wakeToken"` Target string `json:"target"` Status string `json:"status"` ReplyMessageID string `json:"replyMessageId"` Detail string `json:"detail"` Timestamp string `json:"timestamp"` Transient bool `json:"transient,omitempty"` } type agentRuntimeEventMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` MessageID string `json:"messageId"` WakeToken string `json:"wakeToken"` EventClass string `json:"eventClass"` Target string `json:"target"` ReplyMessageID string `json:"replyMessageId"` Detail string `json:"detail"` Timestamp string `json:"timestamp"` } type agentRecoveryPointPreparedMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` MessageID string `json:"messageId"` ThreadID string `json:"threadId"` Strategy string `json:"strategy"` RecoveryRef string `json:"recoveryRef"` } type agentTrajectoryMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` SessionID string `json:"sessionId"` Timestamp int64 `json:"timestamp"` Entries json.RawMessage `json:"entries"` } type agentTrajectoryHistoryResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Items []runtimecontrol.AgentTrajectoryItem `json:"items"` Error string `json:"error,omitempty"` } type deliverAckMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` MessageID string `json:"messageId,omitempty"` Seq int `json:"seq"` InterruptPolicy string `json:"interruptPolicy,omitempty"` Accepted bool `json:"accepted"` Detail string `json:"detail"` } type agentRestoreRecoveryPointResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Success bool `json:"success"` Error string `json:"error"` } type fileTreeMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` DirPath string `json:"dirPath,omitempty"` RootPath string `json:"rootPath,omitempty"` Files []dto.DaemonWorkspaceFile `json:"files"` Limit int `json:"limit,omitempty"` Cursor string `json:"cursor,omitempty"` HasMore bool `json:"hasMore,omitempty"` NextCursor string `json:"nextCursor,omitempty"` } type workspaceListRemoteRequestPayload struct { AgentID string `json:"agentId,omitempty"` TargetPath string `json:"targetPath,omitempty"` DirPath string `json:"dirPath,omitempty"` Limit int `json:"limit,omitempty"` Cursor string `json:"cursor,omitempty"` } type workspaceListRemoteRequestResponse struct { AgentID string `json:"agentId"` DirPath string `json:"dirPath,omitempty"` RootPath string `json:"rootPath,omitempty"` Files []dto.DaemonWorkspaceFile `json:"files"` Limit int `json:"limit,omitempty"` Cursor string `json:"cursor,omitempty"` HasMore bool `json:"hasMore,omitempty"` NextCursor string `json:"nextCursor,omitempty"` } type workspaceReadRemoteRequestPayload struct { AgentID string `json:"agentId,omitempty"` TargetPath string `json:"targetPath,omitempty"` Path string `json:"path"` } type workspaceReadRemoteRequestResponse struct { AgentID string `json:"agentId"` Content *string `json:"content"` Binary bool `json:"binary"` } type workspaceApplySeedFilesRemoteRequestPayload struct { AgentID string `json:"agentId"` SeedFiles []agentWorkspaceSeedFile `json:"seedFiles"` } type workspaceApplySeedFilesRemoteRequestResponse struct { AgentID string `json:"agentId"` Success bool `json:"success"` } type workspaceDeleteRemoteRequestPayload struct { AgentID string `json:"agentId"` Path string `json:"path"` } type workspaceDeleteRemoteRequestResponse struct { AgentID string `json:"agentId"` Success bool `json:"success"` } type fileContentMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Content *string `json:"content"` Binary bool `json:"binary"` } type applySeedFilesResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Success bool `json:"success"` Error string `json:"error,omitempty"` } type deleteWorkspacePathResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Success bool `json:"success"` Error string `json:"error,omitempty"` } type reviewSummaryResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Summary *agent.AgentReviewSummary `json:"summary,omitempty"` Error string `json:"error,omitempty"` // ErrorCode carries a stable wire signal for known product-limitation // conditions (currently: "review_target_unsupported" when the repo // tree contains a control-character path). Empty means "generic error; // caller maps to 500". Hub-side handler reconstructs a typed sentinel // from this code so the HTTP layer can return a controlled response // (e.g. 422 for unsupported) instead of a uniform 500. ErrorCode string `json:"errorCode,omitempty"` } type reviewFilesResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Files []agent.AgentReviewFile `json:"files"` Error string `json:"error,omitempty"` // ErrorCode — see reviewSummaryResultMessage.ErrorCode. ErrorCode string `json:"errorCode,omitempty"` } type reviewDiffResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Diff *agent.AgentReviewDiff `json:"diff,omitempty"` Error string `json:"error,omitempty"` // ErrorCode — see reviewSummaryResultMessage.ErrorCode. ErrorCode string `json:"errorCode,omitempty"` } type outputTargetValidateResultMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` RequestID string `json:"requestId"` Valid bool `json:"valid"` Error string `json:"error,omitempty"` } type scanResultMessage struct { Type string `json:"type"` Directories []agent.WorkspaceDirectory `json:"directories"` Limit int `json:"limit,omitempty"` Cursor string `json:"cursor,omitempty"` HasMore bool `json:"hasMore,omitempty"` NextCursor string `json:"nextCursor,omitempty"` } type machineWorkspacesRequestMessage struct { Type string `json:"type"` RequestID string `json:"requestId"` } type machineWorkspacesResponseMessage struct { Type string `json:"type"` RequestID string `json:"requestId"` Success bool `json:"success"` Workspaces []runtimecontrol.MachineWorkspace `json:"workspaces"` Error string `json:"error,omitempty"` } type deleteResultMessage struct { Type string `json:"type"` DirectoryName string `json:"directoryName"` Success bool `json:"success"` } // agentTurnStateRecord binds one Agent's real turn-execution state (active/ // queued/idle) to the exact transport generation and Agent session that produced // it. Recovery observation marks state known only when the record's generation // and session identity match the current guarded connection/session, so a stale // prior-session or prior-epoch state can never drive a park/clear decision. type agentTurnStateRecord struct { State string Revision int64 Generation string // transport session generation that produced this state SessionID string // Agent session that produced this state } type clientState struct { machineID string workspaceID string createdByUserID string conn *websocket.Conn logger *slog.Logger sendMu sync.Mutex mu sync.Mutex closed bool // guard is the per-machine observation read/write lease. It is shared by // every connection for the same machine (a reconnect reuses the guard) so a // recovery reader holding the read lease blocks state replacement from both // the current and any superseding connection. Set at ServeConnection. guard *machineObservationGuard capabilities map[string]struct{} tokenRefreshProofHash string sessionGeneration string transportProtocol int endpointRegistryVersion string supportedEndpointVersions map[string]map[int]struct{} runtimeCatalogKnown bool runtimes []string runtimeDefaults map[string]runtime.RuntimeDefaults runningAgents map[string]bool startingAgents map[string]bool stoppedAgents map[string]bool agentStatuses map[string]string agentActivities map[string]agentActivityState agentSessions map[string]string agentTurnStateRecords map[string]agentTurnStateRecord // turnStateSnapshot records the current-connection authoritative turn-state // snapshot facts used by Workflow callback recovery to bound the grace window // while the daemon has not yet reported per-agent turn states. connectedAt is // anchored once per connection and is never moved by ready re-application or // incremental events, so polling, duplicate ready, and old-epoch races cannot // extend the recovery grace forever. turnStateSnapshotReceived/At reflect the // last current-connection ready snapshot that actually carried turn states. connectedAt time.Time turnStateSnapshotReceived bool turnStateSnapshotAt time.Time // agentTurnStateRecords/turnStateHighWater implement the same-connection // monotonic ordering contract: every ready snapshot carries a watermark and // every incremental turn_state a strictly newer revision. A snapshot is // applied per-agent by merge (an older snapshot never overwrites a newer // per-agent increment, but its still-current entries for other agents remain // materializable); an older/equal increment never overwrites a newer snapshot // (revision <= high-water is dropped). Ordering is reset only on a genuinely // new authenticated connection (fresh clientState). // // agentTurnStateRecords binds each per-Agent state to the exact transport // generation and Agent session that produced it (see agentTurnStateRecord). // Recovery observation marks state known only when the record's generation/ // session identity matches the current guarded connection/session. turnStateHighWater int64 deliverAcks map[string]chan deliverAckResult readRequests map[string]chan fileContentResult reviewSummaryRequests map[string]chan reviewSummaryResult reviewFilesRequests map[string]chan reviewFilesResult reviewDiffRequests map[string]chan reviewDiffResult validateTargetRequests map[string]chan outputTargetValidateResult trajectoryRequests map[string]chan trajectoryHistoryResult restoreRequests map[string]chan restoreRecoveryPointResult applySeedRequests map[string]chan applySeedFilesResult deleteRequests map[string]chan deleteWorkspacePathResult remoteRequests map[string]chan RemoteRequestResult[json.RawMessage] workspaceRequestMu sync.Mutex workspaceStateMu sync.Mutex pendingTree chan fileTreeResult pendingScan chan scanResult pendingDelete chan deleteResult } type fileContentResult struct { content agent.WorkspaceFileContent err error } type fileTreeResult struct { page agent.WorkspaceFilePage err error } type reviewSummaryResult struct { summary agent.AgentReviewSummary err error } type reviewFilesResult struct { files []agent.AgentReviewFile err error } type reviewDiffResult struct { diff agent.AgentReviewDiff err error } type outputTargetValidateResult struct { valid bool err error } type trajectoryHistoryResult struct { items []runtimecontrol.AgentTrajectoryItem err error } type scanResult struct { page agent.WorkspaceDirectoryPage err error } type deleteResult struct { success bool err error } type deliverAckResult struct { accepted bool detail string } type restoreRecoveryPointResult struct { success bool err error } type applySeedFilesResult struct { success bool err error } type deleteWorkspacePathResult struct { success bool err error } type machineAgentForDaemon struct { ID string `json:"id"` Name string `json:"name"` DisplayName string `json:"displayName,omitempty"` Description string `json:"description,omitempty"` Runtime string `json:"runtime"` PermissionMode string `json:"permissionMode,omitempty"` RuntimePolicy *runtimePolicyForDaemon `json:"runtimePolicy,omitempty"` VisibleReplyContract *visibleReplyContractForDaemon `json:"visibleReplyContract,omitempty"` RuntimeDebug *runtimeDebugForDaemon `json:"runtimeDebug,omitempty"` Model string `json:"model"` ReasoningEffort string `json:"reasoningEffort"` LLMMode string `json:"llmMode,omitempty"` BaseURL string `json:"baseUrl,omitempty"` APIKey string `json:"apiKey,omitempty"` Headers map[string]string `json:"headers,omitempty"` WireAPI string `json:"wireApi,omitempty"` CredentialKind string `json:"credentialKind,omitempty"` Status string `json:"status"` Activity string `json:"activity"` ExtensionCatalog *ExtensionCatalogHint `json:"extensionCatalog,omitempty"` } // ExtensionCatalogHint is the bounded machine hint for one Agent's Extension // Catalog (spec §8.2). It carries ONLY the revision counter — no schemas, // Definitions, bindings, credentials, or endpoints. The daemon treats it purely // as a "did it change" signal: it never orders it or binds it into identities. type ExtensionCatalogHint struct { Version int64 `json:"version"` } type agentStartCommandMessage struct { Type string `json:"type"` AgentID string `json:"agentId"` Config agentStartConfig `json:"config"` WakeMessage any `json:"wakeMessage,omitempty"` UnreadSummary map[string]int `json:"unreadSummary,omitempty"` } type agentStartConfig struct { Name string `json:"name"` DisplayName string `json:"displayName,omitempty"` Description string `json:"description,omitempty"` Runtime string `json:"runtime,omitempty"` PermissionMode string `json:"permissionMode,omitempty"` RuntimePolicy *runtimePolicyForDaemon `json:"runtimePolicy,omitempty"` VisibleReplyContract *visibleReplyContractForDaemon `json:"visibleReplyContract,omitempty"` RuntimeDebug *runtimeDebugForDaemon `json:"runtimeDebug,omitempty"` Model string `json:"model,omitempty"` ReasoningEffort string `json:"reasoningEffort,omitempty"` LLMMode string `json:"llmMode,omitempty"` BaseURL string `json:"baseUrl,omitempty"` APIKey string `json:"apiKey,omitempty"` Headers map[string]string `json:"headers,omitempty"` WireAPI string `json:"wireApi,omitempty"` CredentialKind string `json:"credentialKind,omitempty"` ServerURL string `json:"serverUrl"` AuthToken string `json:"authToken,omitempty"` SessionID string `json:"sessionId,omitempty"` Materialization *agentWorkspaceMaterialization `json:"materialization,omitempty"` } type runtimePolicyForDaemon struct { SchemaVersion int `json:"schemaVersion"` Generation int64 `json:"generation"` Fingerprint string `json:"fingerprint"` Runtime string `json:"runtime"` RuntimeType string `json:"runtimeType"` PermissionMode string `json:"permissionMode"` VisibleReplyContractEnabled bool `json:"visibleReplyContractEnabled,omitempty"` } type visibleReplyContractForDaemon struct { Enabled bool `json:"enabled"` } type runtimeDebugForDaemon struct { Enabled bool `json:"enabled"` } type agentWorkspaceMaterialization struct { RoleID string `json:"roleId,omitempty"` SourceKind string `json:"sourceKind,omitempty"` SourceRef string `json:"sourceRef,omitempty"` ActiveDefinitionPath string `json:"activeDefinitionPath,omitempty"` SeedFiles []agentWorkspaceSeedFile `json:"seedFiles,omitempty"` } type agentWorkspaceSeedFile struct { Path string `json:"path"` Content string `json:"content"` } const ( workspaceTargetPathCapability = runtimecontrol.WorkspaceTargetPathCapability agentPermissionModeCapability = runtimecontrol.AgentPermissionModeCapability agentMaaSDirectCapability = runtimecontrol.AgentMaaSDirectCapability agentRuntimeSessionAuthorityCapability = runtimecontrol.AgentRuntimeSessionAuthorityCapability agentLocalSessionResumeCapability = runtimecontrol.AgentLocalSessionResumeCapability agentVisibleReplyContractCapability = runtimecontrol.AgentVisibleReplyContractCapability daemonReadyRunningAgentsLimit = 256 daemonPingInterval = 15 * time.Second daemonPongWait = 45 * time.Second daemonWriteWait = 10 * time.Second daemonDeliverWait = 2 * time.Second daemonStartWakeAckWait = 30 * time.Second ) // Hub manages all connected daemon machines. // machineObservationGuard is the per-machine read/write lease that fences a // recovery observation against authoritative turn-state replacement. // // Lock order contract (never machine-resolution DB reads while holding the // guard, never a DB->Hub inversion): // - Recovery observation readers take the guard read side, then read client // state under client.mu. They hold the read side across the row-locked // RecoveryTick transaction commit, so no turn-state replacement can // invalidate the observation while it is in use. // - Hub turn-state writers (setAgentTurnState, replaceReadyAgentTurnStates, // and the ready-snapshot/clear paths) take the guard write side, then mutate // under client.mu. A writer blocked by an active reader simply waits; it // never acquires hub.mu while holding the guard, so there is no lock // inversion. // - Machine disconnect (unregister) removes the client under hub.mu first, // releases hub.mu, then takes the guard write side to provide the exclusive // boundary — never hub.mu -> guard, avoiding a deadlock with a reader that // holds the guard read side and briefly re-checks connectivity under hub.mu. // // The guard is a condition-based read/write gate (not a bare sync.RWMutex) so // the writer's waiting and acquired stages are AUTHORITATIVE and observable: a // writer registers in the gate's waiting registry atomically with entering the // blocked state, then signals writeAcquired after it owns the write. The // stage hooks are inert in production (nil); the gate is a correct RW lock. type machineObservationGuard struct { gate machineObservationGate // test hooks, inert in production. writeWaiting fires when a writer is // REGISTERED as waiting and about to block (atomically with the blocked // state); writeAcquired fires after it owns the write side. The // scheduler-independent contention tests use both to prove a writer has // actually reached and registered on the lock, and separately acquired it, // without a fixed-sleep negative check. writeWaiting func() writeAcquired func() } // acquireWrite takes the write side of the machine observation guard, firing the // optional writeWaiting (after the writer registers as waiting, atomically with // entering the blocked state) and writeAcquired (after it owns the write). The // hooks are nil in production and inert; only non-nil hooks are passed so a nil // hook can never be invoked. func (guard *machineObservationGuard) acquireWrite() { guard.gate.acquireWrite(guard.nonNil(func() { guard.writeWaiting() }, guard.writeWaiting), guard.nonNil(func() { guard.writeAcquired() }, guard.writeAcquired)) } // nonNil returns hook when hook != nil, otherwise returns fallback (nil). func (guard *machineObservationGuard) nonNil(wrapped, hook func()) func() { if hook == nil { return nil } return wrapped } // releaseWrite releases the write side of the machine observation guard. func (guard *machineObservationGuard) releaseWrite() { guard.gate.releaseWrite() } // acquireRead takes the read side of the machine observation guard. func (guard *machineObservationGuard) acquireRead() { guard.gate.acquireRead() } // releaseRead releases the read side of the machine observation guard. func (guard *machineObservationGuard) releaseRead() { guard.gate.releaseRead() } // machineObservationGate is a condition-based read/write gate whose waiting and // acquired stages are authoritative and test-observable. A writer registers in // the waiting registry (waitingWriters) atomically with entering the actual // blocked state; the optional stage callbacks fire at that exact boundary. // // The gate is a correct RW lock: readers proceed concurrently unless a writer // owns the write; a writer waits while any reader holds the read side or another // writer owns the write; releasing the last reader or the writer broadcasts. The // stage callbacks are inert in production (nil) and never affect the gate's // correctness. type machineObservationGate struct { mu sync.Mutex cond *sync.Cond readers int writer bool waitingWriters int writeWaiting func() writeAcquired func() } // init lazily constructs the condition variable exactly once, safely, even for // concurrent first use of the zero value. The cond is published while holding // mu, and all readers of gate.cond (Wait/Broadcast) happen under mu, so no // first-use read/write can race with construction and no Wait can strand on an // unpublished cond. func (gate *machineObservationGate) init() { gate.mu.Lock() if gate.cond == nil { gate.cond = sync.NewCond(&gate.mu) } gate.mu.Unlock() } func (gate *machineObservationGate) acquireRead() { gate.init() gate.mu.Lock() // A registered writer must not be starved: a new reader waits behind any // writer that has registered as waiting, and behind an owning writer. for gate.writer || gate.waitingWriters > 0 { gate.cond.Wait() } gate.readers++ gate.mu.Unlock() } func (gate *machineObservationGate) releaseRead() { gate.init() gate.mu.Lock() if gate.readers > 0 { gate.readers-- } if gate.readers == 0 { gate.cond.Broadcast() } gate.mu.Unlock() } // acquireWrite takes the write side of the machine observation gate. // // The check-to-claim is ATOMIC under the first gate.mu hold: if the gate is // initially free (no reader and no owning writer), the writer claims ownership // (writer = true) before the first unlock, so no reader/writer can slip in // between the check and the claim (the writer never silently waits without a // waitingWriters registration). If the gate is initially blocked, the writer // increments waitingWriters and fires the nonblocking waiting stage before the // first unlock, then relocks and predicate-loops until the write is free, and // claims it before unlock. // // Stage callbacks are nonblocking/non-reentrant stage notifications and are // NEVER invoked while gate.mu is held: acquired runs after the writer has // logically claimed the write and must not try to reacquire the same gate // before releaseWrite. func (gate *machineObservationGate) acquireWrite(waiting func(), acquired func()) { gate.init() gate.mu.Lock() if !gate.writer && gate.readers == 0 { // Initially free: claim ownership atomically with the check, before any // other goroutine can observe the gap. gate.writer = true gate.mu.Unlock() if acquired != nil { acquired() } return } // Initially blocked: register as waiting, fire the nonblocking waiting stage, // then wait for the write to become free. gate.waitingWriters++ gate.mu.Unlock() if waiting != nil { waiting() } gate.mu.Lock() for gate.writer || gate.readers > 0 { gate.cond.Wait() } gate.writer = true gate.waitingWriters-- gate.mu.Unlock() if acquired != nil { acquired() } } func (gate *machineObservationGate) releaseWrite() { gate.init() gate.mu.Lock() gate.writer = false gate.cond.Broadcast() gate.mu.Unlock() } // withObservationGuardWrite runs fn while holding the machine observation guard // write side. It is the shared exclusive boundary for every mutation of an // observation-determining field (current client identity, connectedAt, // session generation, agentTurnStateRecords, turnStateHighWater, // and connection removal), so a recovery reader holding the read lease across a // row-locked transaction never observes a torn transition. A nil guard runs fn // directly (legacy test clients that were not built by ServeConnection). func (client *clientState) withObservationGuardWrite(fn func()) { if client.guard != nil { client.guard.acquireWrite() defer client.guard.releaseWrite() } fn() } type Hub struct { logger *slog.Logger store Store runtimeEvents RuntimeEventStore publicBaseURL string maasRuntimeBaseURL string realtime realtimeEmitter hintReader func(ctx context.Context, workspaceID, agentID string) (*ExtensionCatalogHint, error) mu sync.RWMutex clients map[string]*clientState remoteEndpoints *RemoteEndpointRegistry elasticRequestLinks *ElasticRequestLinkManager // testTurnStateApplyHook, when non-nil, is invoked inside // handleSessionLinkAgentTurnStateCommand AFTER the event is decoded and // machine-ownership validated but BEFORE the guarded setAgentTurnState apply. // It is nil in production. The atomic old-epoch barrier test uses it to pause // an event at that exact boundary, change the transport generation, resume, // and prove the event is dropped. testTurnStateApplyHook func(ctx context.Context) // observationGuards are the per-machine recovery observation read/write // leases. They are created lazily and live for the Hub's lifetime. Readers // hold mu.RLock across a recovery transaction; Hub turn-state writers take // mu.Lock. Keyed by machineID. observationGuards sync.Map // machineID string -> *machineObservationGuard } type HubOptions struct { MaaSRuntimeBaseURL string ElasticRequestLinkConfig ElasticRequestLinkManagerConfig RemoteEndpointRegistry *RemoteEndpointRegistry HintReader func(ctx context.Context, workspaceID, agentID string) (*ExtensionCatalogHint, error) // TestTurnStateApplyHook, when non-nil, is invoked at the turn-state apply // boundary (after decode/ownership validation, before the guarded mutation). // Nil in production; used by the atomic old-epoch barrier test. TestTurnStateApplyHook func(ctx context.Context) } // New constructs a Hub. func New(logger *slog.Logger, dataStore Store, publicBaseURL string, emitter realtimeEmitter, runtimeEvents ...RuntimeEventStore) *Hub { return NewWithOptions(logger, dataStore, publicBaseURL, emitter, HubOptions{}, runtimeEvents...) } func NewWithOptions(logger *slog.Logger, dataStore Store, publicBaseURL string, emitter realtimeEmitter, options HubOptions, runtimeEvents ...RuntimeEventStore) *Hub { var runtimeEventStore RuntimeEventStore if len(runtimeEvents) > 0 { runtimeEventStore = runtimeEvents[0] } else if compatible, ok := dataStore.(RuntimeEventStore); ok { runtimeEventStore = compatible } hubLogger := logger if hubLogger == nil { hubLogger = slog.Default() } remoteEndpoints := options.RemoteEndpointRegistry if remoteEndpoints == nil { remoteEndpoints = defaultRemoteEndpointRegistry() } return &Hub{ logger: hubLogger, store: dataStore, runtimeEvents: runtimeEventStore, publicBaseURL: publicBaseURL, maasRuntimeBaseURL: strings.TrimSpace(options.MaaSRuntimeBaseURL), realtime: emitter, hintReader: options.HintReader, clients: make(map[string]*clientState), remoteEndpoints: remoteEndpoints, elasticRequestLinks: NewElasticRequestLinkManager(options.ElasticRequestLinkConfig, remoteEndpoints, hubLogger), testTurnStateApplyHook: options.TestTurnStateApplyHook, } } // ServeConnection registers one daemon websocket until it closes. func (hub *Hub) ServeConnection(ctx context.Context, identity agent.MachineIdentity, conn *websocket.Conn) error { client := &clientState{ machineID: identity.MachineID, workspaceID: identity.WorkspaceID, createdByUserID: identity.CreatedByUserID, conn: conn, logger: hub.logger.With(slog.String("machine_id", identity.MachineID)), guard: hub.observationGuard(identity.MachineID), capabilities: make(map[string]struct{}), supportedEndpointVersions: make(map[string]map[int]struct{}), runningAgents: make(map[string]bool), startingAgents: make(map[string]bool), stoppedAgents: make(map[string]bool), agentStatuses: make(map[string]string), agentActivities: make(map[string]agentActivityState), agentSessions: make(map[string]string), agentTurnStateRecords: make(map[string]agentTurnStateRecord), connectedAt: time.Now().UTC(), deliverAcks: make(map[string]chan deliverAckResult), readRequests: make(map[string]chan fileContentResult), reviewSummaryRequests: make(map[string]chan reviewSummaryResult), reviewFilesRequests: make(map[string]chan reviewFilesResult), reviewDiffRequests: make(map[string]chan reviewDiffResult), validateTargetRequests: make(map[string]chan outputTargetValidateResult), trajectoryRequests: make(map[string]chan trajectoryHistoryResult), restoreRequests: make(map[string]chan restoreRecoveryPointResult), applySeedRequests: make(map[string]chan applySeedFilesResult), deleteRequests: make(map[string]chan deleteWorkspacePathResult), remoteRequests: make(map[string]chan RemoteRequestResult[json.RawMessage]), } _ = conn.SetReadDeadline(time.Now().Add(daemonPongWait)) conn.SetPongHandler(func(string) error { _ = conn.SetReadDeadline(time.Now().Add(daemonPongWait)) return nil }) if hub.elasticRequestLinks != nil { logging.Go(hub.logger, func() { hub.elasticRequestLinks.RunCleanupLoop(ctx) }) } if err := client.sessionLinkSender().Send(serverCapabilitiesCommand{ Type: sessionLinkCommandServerCapabilities, Capabilities: []string{ transientLegacyWakeStatusServerCapability, }, }); err != nil { _ = conn.Close() return err } hub.register(client) defer func() { removedCurrent := hub.unregister(client) if hub.elasticRequestLinks != nil { hub.elasticRequestLinks.CleanupSession(client.machineID, client.currentSessionGeneration(), "session_disconnected") } _ = conn.Close() if removedCurrent { if err := hub.store.MarkMachineDisconnected(context.Background(), identity.MachineID); err != nil { client.logger.Warn("mark machine disconnected", slog.Any("error", err)) } if hub.realtime != nil { hub.realtime.EmitMachineStatus(identity.WorkspaceID, identity.MachineID, "offline") hub.realtime.EmitDaemonStatus(identity.WorkspaceID, identity.MachineID, "offline") } } }() pingTicker := time.NewTicker(daemonPingInterval) defer pingTicker.Stop() logging.Go(hub.logger, func() { for { select { case <-ctx.Done(): return case <-pingTicker.C: if err := client.sendPing(daemonWriteWait); err != nil { return } } } }) for { select { case <-ctx.Done(): return ctx.Err() default: } _, payload, err := conn.ReadMessage() if err != nil { if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { return nil } var netErr net.Error if errors.As(err, &netErr) && netErr.Timeout() { return nil } return err } _ = conn.SetReadDeadline(time.Now().Add(daemonPongWait)) if err := hub.handleMessage(ctx, client, payload); err != nil { client.logger.Warn("handle daemon message", slog.Any("error", err)) } } } // DisconnectMachine closes the active websocket for one machine, if any. func (hub *Hub) DisconnectMachine(machineID string) { client := hub.client(machineID) if client == nil { return } client.close(websocket.CloseNormalClosure, "machine key rotated") } // MachineConnected reports whether the hub currently has an authenticated // daemon connection for the machine. func (hub *Hub) MachineConnected(machineID string) bool { return hub.client(machineID) != nil } // MachineDirectoryBrowseSupported reports whether the connected daemon can // service machine.directory.list@1. It returns true only when the daemon is // connected AND advertises the endpoint at the matching registry version. // The result is computed live from the current advertisement and is never // persisted: callers MUST treat it as a point-in-time observation. Any // failure mode that is not a clean capability miss (invalid path, unreadable // directory, transient busy/unavailable) is reported through the browse RPC // itself, not through this boolean, so the Web UI's manual path fallback // fires only for genuine capability gaps. func (hub *Hub) MachineDirectoryBrowseSupported(machineID string) bool { client := hub.client(machineID) if client == nil { return false } descriptor, ok := hub.remoteEndpointDescriptor(remoteEndpointMachineDirectoryList, 1) if !ok { return false } return client.supportsRemoteEndpoint(descriptor.EndpointID, descriptor.Version, descriptor.RegistryVersion) } // MachineDashboardSupported reports whether the connected daemon can service // machine.dashboard@2. It mirrors the directory browse capability flag: only // a connected daemon advertising the endpoint at the matching registry version // returns true. func (hub *Hub) MachineDashboardSupported(machineID string) bool { return hub.machineDashboardSupported(hub.client(machineID)) } // QueryMachineDashboard routes one machine dashboard metrics request to the // connected daemon. Offline and transport failures are normalized to stable // dashboard errors so the Web UI can show either upgrade-required or retryable // unavailable states without inspecting raw transport text. func (hub *Hub) QueryMachineDashboard(ctx context.Context, machineID string, request runtimecontrol.MachineDashboardRequest) (runtimecontrol.MachineDashboardResponse, error) { trimmedMachineID := strings.TrimSpace(machineID) client := hub.client(trimmedMachineID) if client == nil { return runtimecontrol.MachineDashboardResponse{}, runtimecontrol.ErrMachineDashboardUnavailable } request.MachineID = trimmedMachineID dashboard, err := hub.queryMachineDashboard(ctx, client, request) if err == nil { return dashboard, nil } return runtimecontrol.MachineDashboardResponse{}, classifyMachineDashboardError(err) } // ListMachineDirectoriesPage routes one machine directory browse request to // the connected daemon. It returns ErrMachineDirectoryUnavailable when no // client is connected, ErrMachineDirectoryUnsupported before any assignment // is built when capability is missing, and the post-classifier's typed // sentinel for every other route outcome. An unexpected failure additionally // records one path-free internal warning carrying a fixed cause_category. func (hub *Hub) ListMachineDirectoriesPage(ctx context.Context, machineID string, path string, opts agent.WorkspaceListOptions) (agent.MachineDirectoryPage, error) { client := hub.client(machineID) if client == nil { return agent.MachineDirectoryPage{}, runtimecontrol.ErrMachineDirectoryUnavailable } page, err := hub.listMachineDirectories(ctx, client, path, opts) if err == nil { return page, nil } classified, causeCategory := classifyMachineDirectoryError(err) if causeCategory != "" { hub.warnMachineDirectoryUnexpectedFailure(causeCategory) } return agent.MachineDirectoryPage{}, classified } // QueryMachineSettings routes one machine settings bundle request to the // connected daemon. The response payload is opaque JSON sourced from the // daemon (it owns the desktop-local settings schema); the server only // validates that it is syntactically valid JSON before echoing it. func (hub *Hub) QueryMachineSettings(ctx context.Context, workspaceID string, machineID string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineSettingsUnavailable } payload, err := hub.queryMachineSettings(ctx, client, workspaceID, machineID) if err == nil { return payload, nil } return nil, classifyMachineSettingsError(err) } // UpdateMachineSettings routes one machine settings update to the connected // daemon. Only the desktop sub-field is mutable; system and daemon are // read-only and the daemon enforces that contract server-side. func (hub *Hub) UpdateMachineSettings(ctx context.Context, workspaceID string, machineID string, desktop json.RawMessage) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineSettingsUnavailable } payload, err := hub.updateMachineSettings(ctx, client, workspaceID, machineID, desktop) if err == nil { return payload, nil } return nil, classifyMachineSettingsError(err) } // QueryMachineRuntimesBundle routes one runtimes-bundle request (summary + // per-runtime status + agents + freshly detected runtimes) to the daemon. func (hub *Hub) QueryMachineRuntimesBundle(ctx context.Context, workspaceID string, machineID string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineRuntimesUnavailable } payload, err := hub.queryMachineRuntimesBundle(ctx, client, workspaceID, machineID) if err == nil { return payload, nil } return nil, classifyMachineRuntimesError(err) } // QueryMachineRuntimeConfig routes one runtime config read to the daemon. func (hub *Hub) QueryMachineRuntimeConfig(ctx context.Context, workspaceID string, machineID string, name string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineRuntimesUnavailable } payload, err := hub.queryMachineRuntimeConfig(ctx, client, workspaceID, machineID, name) if err == nil { return payload, nil } return nil, classifyMachineRuntimesError(err) } // UpdateMachineRuntimeConfig routes one runtime config update to the daemon. func (hub *Hub) UpdateMachineRuntimeConfig(ctx context.Context, workspaceID string, machineID string, name string, config json.RawMessage) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineRuntimesUnavailable } payload, err := hub.updateMachineRuntimeConfig(ctx, client, workspaceID, machineID, name, config) if err == nil { return payload, nil } return nil, classifyMachineRuntimesError(err) } // ValidateMachineRuntimeConfig routes one runtime config validation // (side-effect free probe) to the daemon. func (hub *Hub) ValidateMachineRuntimeConfig(ctx context.Context, workspaceID string, machineID string, name string, config json.RawMessage) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineRuntimesUnavailable } payload, err := hub.validateMachineRuntimeConfig(ctx, client, workspaceID, machineID, name, config) if err == nil { return payload, nil } return nil, classifyMachineRuntimesError(err) } // DetectMachineRuntimes triggers one runtime detection pass on the daemon // and returns the refreshed runtimes bundle. func (hub *Hub) DetectMachineRuntimes(ctx context.Context, workspaceID string, machineID string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineRuntimesUnavailable } payload, err := hub.detectMachineRuntimes(ctx, client, workspaceID, machineID) if err == nil { return payload, nil } return nil, classifyMachineRuntimesError(err) } // QueryMachineMCPServers lists the daemon-managed MCP servers. func (hub *Hub) QueryMachineMCPServers(ctx context.Context, workspaceID string, machineID string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineMCPServersUnavailable } payload, err := hub.queryMachineMCPServers(ctx, client, workspaceID, machineID) if err == nil { return payload, nil } return nil, classifyMachineMCPServersError(err) } // UpsertMachineMCPServer creates or updates one MCP server config by name. func (hub *Hub) UpsertMachineMCPServer(ctx context.Context, workspaceID string, machineID string, server json.RawMessage) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineMCPServersUnavailable } payload, err := hub.upsertMachineMCPServer(ctx, client, workspaceID, machineID, server) if err == nil { return payload, nil } return nil, classifyMachineMCPServersError(err) } // DeleteMachineMCPServer removes one MCP server config by name. func (hub *Hub) DeleteMachineMCPServer(ctx context.Context, workspaceID string, machineID string, name string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineMCPServersUnavailable } payload, err := hub.deleteMachineMCPServer(ctx, client, workspaceID, machineID, name) if err == nil { return payload, nil } return nil, classifyMachineMCPServersError(err) } // TestMachineMCPServer probes connectivity for one MCP server config. func (hub *Hub) TestMachineMCPServer(ctx context.Context, workspaceID string, machineID string, name string) (json.RawMessage, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineMCPServersUnavailable } payload, err := hub.testMachineMCPServer(ctx, client, workspaceID, machineID, name) if err == nil { return payload, nil } return nil, classifyMachineMCPServersError(err) } // MachineRuntimeCatalog returns the current runtime catalog from the active // daemon ready message. func (hub *Hub) MachineRuntimeCatalog(machineID string) ([]string, map[string]runtime.RuntimeDefaults, bool) { client := hub.client(machineID) if client == nil { return nil, nil, false } return client.machineRuntimeCatalog() } // WorkspaceRuntimeSnapshot returns the current live runtime truth already held // in hub memory for one workspace. It intentionally replays only live evidence; // offline placeholders stay sourced from DB read models. func (hub *Hub) WorkspaceRuntimeSnapshot(workspaceID string) im.WorkspaceRuntimeSnapshot { trimmedWorkspaceID := strings.TrimSpace(workspaceID) if hub == nil || trimmedWorkspaceID == "" { return im.WorkspaceRuntimeSnapshot{} } hub.mu.RLock() clients := make([]*clientState, 0, len(hub.clients)) for _, client := range hub.clients { if client == nil || strings.TrimSpace(client.workspaceID) != trimmedWorkspaceID { continue } clients = append(clients, client) } hub.mu.RUnlock() sort.Slice(clients, func(i, j int) bool { return strings.TrimSpace(clients[i].machineID) < strings.TrimSpace(clients[j].machineID) }) machines := make([]im.WorkspaceRuntimeMachineSnapshot, 0, len(clients)) for _, client := range clients { machine, ok := client.workspaceRuntimeMachineSnapshot() if !ok { continue } machines = append(machines, machine) } return im.WorkspaceRuntimeSnapshot{Machines: machines} } // NotifyAgentMessage wakes or starts an agent when a new message arrives. func (hub *Hub) NotifyAgentMessage(ctx context.Context, recipient message.AgentRecipient, seq int64, msg message.ReceiveMessage) error { requiresLiveHandoff := receiveMessageRequiresLiveHandoff(msg) if !recipient.AutoWakeEnabled && !agentRecipientBypassesAutoWakePreference(recipient) && !receiveMessageBypassesAutoWakePreference(msg) { return nil } return hub.notifyAgentMessage(ctx, recipient, seq, msg, requiresLiveHandoff) } func receiveMessageBypassesAutoWakePreference(msg message.ReceiveMessage) bool { policy, ok := message.NormalizeAgentInterruptPolicy(msg.InterruptPolicy) return ok && policy == message.AgentInterruptPolicyCurrentRun } func receiveMessageRequiresLiveHandoff(msg message.ReceiveMessage) bool { return msg.RequireLiveHandoff || receiveMessageBypassesAutoWakePreference(msg) || strings.TrimSpace(msg.ChannelType) == message.WorkflowScopeType } func agentRecipientBypassesAutoWakePreference(recipient message.AgentRecipient) bool { switch strings.TrimSpace(recipient.DeliveryReason) { case message.AgentDeliveryReasonDirectMention, message.AgentDeliveryReasonTeamMention, message.AgentDeliveryReasonTaskCommandNotify, message.AgentDeliveryReasonTaskAssignee, message.AgentDeliveryReasonChannelBroadcastAll, message.AgentDeliveryReasonChannelBroadcastOnline: return true default: return false } } func (hub *Hub) notifyAgentMessage(ctx context.Context, recipient message.AgentRecipient, seq int64, message message.ReceiveMessage, requireClient bool) error { client := hub.client(recipient.MachineID) if client == nil { if requireClient { return runtimecontrol.ErrMachineOffline } return nil } if client.isAgentStopped(recipient.AgentID) { if requireClient { return runtimecontrol.ErrAgentManuallyStopped } return nil } message = receiveMessageWithFreshnessSeq(message, seq) message = hub.withAgentSessionActiveTaskTarget(ctx, client, recipient.AgentID, message) if client.isAgentRunning(recipient.AgentID) { ack, err := hub.deliverToRunningAgentOrMarkStopped(ctx, client, recipient.AgentID, seq, message, !requireClient) if err == nil && ack.accepted { return nil } if requireClient && err == nil && !ack.accepted { return fmt.Errorf("urgent delivery rejected: %s", ack.detail) } } if requireClient { currentClient := hub.client(recipient.MachineID) if currentClient == nil { return runtimecontrol.ErrMachineOffline } if currentClient != client && currentClient.isAgentRunning(recipient.AgentID) { ack, err := hub.deliverToRunningAgentOrMarkStopped(ctx, currentClient, recipient.AgentID, seq, message, false) if err == nil && ack.accepted { return nil } if err == nil && !ack.accepted { return fmt.Errorf("urgent delivery rejected: %s", ack.detail) } } client = currentClient } return hub.startAgentForMessage(ctx, client, recipient.AgentID, &message) } func (hub *Hub) deliverToRunningAgentOrMarkStopped(ctx context.Context, client *clientState, agentID string, seq int64, message message.ReceiveMessage, allowRejectedFallback bool) (deliverAckResult, error) { deliverCtx, cancel := context.WithTimeout(ctx, daemonDeliverWait) ack, err := client.deliverToRunningAgent(deliverCtx, agentID, seq, message) cancel() if err == nil && ack.accepted { return ack, nil } client.setAgentRunning(agentID, false) if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) { client.logger.Warn("deliver to running agent failed; falling back to start", slog.String("agent_id", agentID), slog.Int64("seq", seq), slog.Any("error", err)) } else if err != nil { client.logger.Warn("deliver to running agent timed out; falling back to start", slog.String("agent_id", agentID), slog.Int64("seq", seq), slog.Any("error", err)) } else if allowRejectedFallback { client.logger.Warn("daemon rejected running-agent delivery; falling back to start", slog.String("agent_id", agentID), slog.Int64("seq", seq), slog.String("detail", ack.detail)) } else { client.logger.Warn("daemon rejected urgent running-agent delivery", slog.String("agent_id", agentID), slog.Int64("seq", seq), slog.String("detail", ack.detail)) } return ack, err } func receiveMessageWithFreshnessSeq(msg message.ReceiveMessage, seq int64) message.ReceiveMessage { if !msg.CarriesScopeTargetSeq() { msg.TargetSeq = 0 } else if msg.TargetSeq <= 0 && seq > 0 { msg.TargetSeq = seq } return msg } func (hub *Hub) withAgentSessionActiveTaskTarget(ctx context.Context, client *clientState, agentID string, msg message.ReceiveMessage) message.ReceiveMessage { if msg.ActiveTaskTarget.Valid() { return msg } target, found, err := hub.store.GetTaskActiveTarget(ctx, client.workspaceID, message.TaskActiveTargetScopeAgentSession, agentID) if err != nil { client.logger.Warn("load agent-session active task target", slog.String("agent_id", agentID), slog.Any("error", err)) return msg } if found && agentSessionActiveTaskTargetAppliesToWake(target) { msg.ActiveTaskTarget = target } return msg } func agentSessionActiveTaskTargetAppliesToWake(target message.ReceiveActiveTaskTarget) bool { target = target.Normalize() if !target.Valid() { return false } return strings.TrimSpace(target.Source) == "human_selected" } func (hub *Hub) startAgentForMessage(ctx context.Context, client *clientState, agentID string, wakeMessage *message.ReceiveMessage) error { if client.isAgentRunning(agentID) { return runtimecontrol.ErrAgentStillRunning } if receiveMessageRequiresDurableStartAck(wakeMessage) && strings.TrimSpace(wakeMessage.MessageID) == "" { return errors.New("message wake id is required for durable start acknowledgement") } if err := hub.requireAgentRuntimeAllowedOnClient(ctx, client, agentID); err != nil { return err } startState := client.markAgentStarting(agentID, false) startCommand, err := hub.prepareAgentStartForClient(ctx, client, agentID, wakeMessage) if err != nil { client.restoreAgentStartState(agentID, startState) return startPermissionModeError(err) } var sendErr error if receiveMessageRequiresDurableStartAck(wakeMessage) { ackCtx, cancel := context.WithTimeout(ctx, daemonStartWakeAckWait) var ack deliverAckResult ack, sendErr = client.startAgentForMessageAndWaitAck(ackCtx, startCommand) cancel() if sendErr == nil && !ack.accepted { sendErr = fmt.Errorf("start wake rejected: %s", ack.detail) } } else { sendErr = client.sendStartCommandIfAllowed(startCommand) } if sendErr != nil { client.restoreAgentStartState(agentID, startState) return sendErr } return nil } func receiveMessageRequiresDurableStartAck(wakeMessage *message.ReceiveMessage) bool { if wakeMessage == nil { return false } switch strings.TrimSpace(wakeMessage.ChannelType) { case "task", message.WorkflowScopeType: return true default: return false } } // StartAgent explicitly starts one agent. func (hub *Hub) StartAgent(ctx context.Context, machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } if client.isAgentRunning(agentID) { return runtimecontrol.ErrAgentStillRunning } if err := hub.requireAgentRuntimeAllowedOnClient(ctx, client, agentID); err != nil { return err } startState := client.markAgentStarting(agentID, true) startCommand, err := hub.prepareAgentStartForClient(ctx, client, agentID, nil) if err != nil { client.restoreAgentStartState(agentID, startState) return startPermissionModeError(err) } if err := client.sendStartCommandIfAllowed(startCommand); err != nil { if errors.Is(err, runtimecontrol.ErrDaemonPermissionModeUpgradeRequired) || errors.Is(err, runtimecontrol.ErrDaemonMaaSDirectUpgradeRequired) { client.restoreAgentStartState(agentID, startState) return err } client.setAgentRunning(agentID, false) client.setAgentStopped(agentID, true) return err } return nil } func (hub *Hub) prepareAgentStartForClient(ctx context.Context, client *clientState, agentID string, wakeMessage *message.ReceiveMessage) (agent.StartCommand, error) { runtimeSessionAuthority := client.hasCapability(agentRuntimeSessionAuthorityCapability) // Runtime-authority telemetry carries a AgentTeam-owned ars_ identity, not a // provider resume handle. Keep it out of agent:start sessionId. includeLegacySession := !client.hasCapability(agentLocalSessionResumeCapability) command, err := hub.store.PrepareAgentStart( ctx, agentID, hub.publicBaseURL, wakeMessage, client.hasCapability(agentPermissionModeCapability), includeLegacySession, client.hasCapability(agentMaaSDirectCapability), ) if err != nil { return agent.StartCommand{}, err } applyVisibleReplyContractCapabilityToStartConfig(&command.Config, client.hasCapability(agentVisibleReplyContractCapability)) if !runtimeSessionAuthority { if sessionID := client.agentSession(agentID); sessionID != "" { command.Config.SessionID = sessionID } } return command, nil } // AgentRuntimeRunning reports whether the daemon currently has, or is starting, // a runtime process for the agent. func (hub *Hub) AgentRuntimeRunning(machineID string, agentID string) bool { client := hub.client(machineID) if client == nil { return false } return client.isAgentRunning(agentID) } // AgentRuntimeSessionID returns the current runtime session reported by the // connected daemon for one agent. func (hub *Hub) AgentRuntimeSessionID(machineID string, agentID string) (string, bool) { client := hub.client(machineID) if client == nil { return "", false } sessionID := client.agentSession(agentID) return sessionID, sessionID != "" } // AgentRuntimeStatus returns the current connection-scoped status/activity // reported by the daemon for one agent. func (hub *Hub) AgentRuntimeStatus(machineID string, agentID string) (string, string, string, bool) { client := hub.client(machineID) if client == nil { return "", "", "", false } return client.agentRuntimeStatus(agentID) } // AgentRuntimeTurnState returns the daemon-derived real Agent turn execution // state (active/queued/idle) for one agent and whether it is known. It is the // authoritative Workflow callback recovery working signal. known=false means the // machine is not connected, the daemon does not support the turn-state overlay // (old daemon), or the state has not been reported yet — recovery must not send // and must not start timing. func (hub *Hub) AgentRuntimeTurnState(machineID string, agentID string) (string, bool) { client := hub.client(machineID) if client == nil { return "", false } return client.agentTurnState(agentID) } // AgentPermissionModeSupported reports whether the connected daemon advertises // support for enforcing agent permission modes. func (hub *Hub) AgentPermissionModeSupported(machineID string) bool { client := hub.client(machineID) if client == nil { return false } return client.hasCapability(agentPermissionModeCapability) } // AgentMaaSDirectSupport reports whether the connected daemon advertises // support for direct MaaS runtime capabilities. func (hub *Hub) AgentMaaSDirectSupport(machineID string) (supported bool, known bool) { client := hub.client(machineID) if client == nil { return false, false } return client.hasCapability(agentMaaSDirectCapability), true } // RestartAgent starts an agent from the latest server-side config even when the // daemon still reports a live process. The existing agent:start command is // replace-capable on the daemon side, so config updates do not need to use the // manual-stop path that suppresses later auto-wake. func (hub *Hub) RestartAgent(ctx context.Context, machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } if err := hub.requireAgentRuntimeAllowedOnClient(ctx, client, agentID); err != nil { return err } startState := client.markAgentStarting(agentID, true) startCommand, err := hub.prepareAgentStartForClient(ctx, client, agentID, nil) if err != nil { client.restoreAgentStartState(agentID, startState) return startPermissionModeError(err) } if err := client.sendStartCommandIfAllowed(startCommand); err != nil { if errors.Is(err, runtimecontrol.ErrDaemonPermissionModeUpgradeRequired) || errors.Is(err, runtimecontrol.ErrDaemonMaaSDirectUpgradeRequired) { client.setAgentRunning(agentID, false) return err } client.setAgentRunning(agentID, false) return err } return nil } func startCommandForDaemon(command agent.StartCommand) agentStartCommandMessage { var wakeMessage any if command.WakeMessage != nil { wakeMessage = converter.ToOldDaemonReceiveMessage(*command.WakeMessage) } return agentStartCommandMessage{ Type: sessionLinkCommandAgentStart, AgentID: command.AgentID, Config: agentStartConfig{ Name: command.Config.Name, DisplayName: command.Config.DisplayName, Description: command.Config.Description, Runtime: command.Config.Runtime, PermissionMode: agent.DefaultAgentPermissionMode(command.Config.PermissionMode), RuntimePolicy: runtimePolicyForDaemonFromStore(command.Config.RuntimePolicy), VisibleReplyContract: visibleReplyContractForDaemonFromEnabled(command.Config.VisibleReplyContractEnabled), RuntimeDebug: runtimeDebugForDaemonFromEnabled(command.Config.RuntimeDebugEnabled), Model: command.Config.Model, ReasoningEffort: command.Config.ReasoningEffort, LLMMode: command.Config.LLMMode, BaseURL: command.Config.BaseURL, APIKey: command.Config.APIKey, Headers: command.Config.Headers, WireAPI: command.Config.WireAPI, CredentialKind: command.Config.CredentialKind, ServerURL: command.Config.ServerURL, AuthToken: command.Config.AuthToken, SessionID: command.Config.SessionID, Materialization: agentWorkspaceMaterializationForDaemon(command.Config.Materialization), }, WakeMessage: wakeMessage, UnreadSummary: command.UnreadSummary, } } func applyVisibleReplyContractCapabilityToStartConfig(config *agent.StartConfig, supported bool) { if config == nil { return } // "Agent reply enhancement" (VisibleReplyContract) is always-on. For clients // that still declare the agent:visible-reply-contract-v1 capability we keep // emitting the projection fixed at enabled=true; clients without the // capability never receive it. The per-agent DB value is no longer an // effective config source. effective := supported config.VisibleReplyContractEnabled = effective config.RuntimePolicy = runtimePolicySnapshotWithVisibleReplyContract(config.RuntimePolicy, effective) } func runtimePolicySnapshotWithVisibleReplyContract(snapshot agent.RuntimePolicySnapshot, enabled bool) agent.RuntimePolicySnapshot { if strings.TrimSpace(snapshot.Fingerprint) == "" && strings.TrimSpace(snapshot.Runtime) == "" && strings.TrimSpace(snapshot.RuntimeType) == "" && strings.TrimSpace(snapshot.PermissionMode) == "" && snapshot.Generation == 0 { snapshot.VisibleReplyContractEnabled = enabled return snapshot } if snapshot.SchemaVersion <= 0 { snapshot.SchemaVersion = agent.RuntimePolicySchemaVersion } if snapshot.Generation <= 0 { snapshot.Generation = 1 } if strings.TrimSpace(snapshot.RuntimeType) == "" { snapshot.RuntimeType = strings.TrimSpace(snapshot.Runtime) } snapshot.VisibleReplyContractEnabled = enabled snapshot.Fingerprint = agent.RuntimePolicyFingerprint( snapshot.SchemaVersion, snapshot.Generation, snapshot.Runtime, snapshot.RuntimeType, snapshot.PermissionMode, ) return snapshot } func (client *clientState) sendStartCommandIfAllowed(command agent.StartCommand) error { client.sendMu.Lock() defer client.sendMu.Unlock() client.mu.Lock() defer client.mu.Unlock() if err := client.requireAgentPermissionModeCapabilityLocked(command.Config.PermissionMode); err != nil { return err } if err := client.requireMaaSDirectCapabilityLocked(command); err != nil { return err } applyVisibleReplyContractCapabilityToStartConfig(&command.Config, client.hasCapabilityLocked(agentVisibleReplyContractCapability)) message := startCommandForDaemon(command) return client.sessionLinkSender().SendLocked(message, daemonWriteWait) } func runtimePolicyForDaemonFromStore(snapshot agent.RuntimePolicySnapshot) *runtimePolicyForDaemon { if strings.TrimSpace(snapshot.Fingerprint) == "" && strings.TrimSpace(snapshot.Runtime) == "" && strings.TrimSpace(snapshot.RuntimeType) == "" && strings.TrimSpace(snapshot.PermissionMode) == "" && snapshot.Generation == 0 { return nil } if snapshot.SchemaVersion <= 0 { snapshot.SchemaVersion = agent.RuntimePolicySchemaVersion } if snapshot.Generation <= 0 { snapshot.Generation = 1 } if strings.TrimSpace(snapshot.RuntimeType) == "" { snapshot.RuntimeType = strings.TrimSpace(snapshot.Runtime) } originalPermissionMode := strings.TrimSpace(snapshot.PermissionMode) snapshot.PermissionMode = agent.DefaultAgentPermissionMode(snapshot.PermissionMode) if strings.TrimSpace(snapshot.Fingerprint) == "" || snapshot.PermissionMode != originalPermissionMode { snapshot.Fingerprint = agent.RuntimePolicyFingerprint(snapshot.SchemaVersion, snapshot.Generation, snapshot.Runtime, snapshot.RuntimeType, snapshot.PermissionMode) } return &runtimePolicyForDaemon{ SchemaVersion: snapshot.SchemaVersion, Generation: snapshot.Generation, Fingerprint: strings.TrimSpace(snapshot.Fingerprint), Runtime: strings.TrimSpace(snapshot.Runtime), RuntimeType: strings.TrimSpace(snapshot.RuntimeType), PermissionMode: strings.TrimSpace(snapshot.PermissionMode), VisibleReplyContractEnabled: snapshot.VisibleReplyContractEnabled, } } func visibleReplyContractForDaemonFromEnabled(enabled bool) *visibleReplyContractForDaemon { if !enabled { return nil } return &visibleReplyContractForDaemon{Enabled: true} } func runtimeDebugForDaemonFromEnabled(enabled bool) *runtimeDebugForDaemon { if !enabled { return nil } return &runtimeDebugForDaemon{Enabled: true} } func agentWorkspaceMaterializationForDaemon(materialization *agent.WorkspaceMaterialization) *agentWorkspaceMaterialization { if materialization == nil { return nil } return &agentWorkspaceMaterialization{ RoleID: materialization.RoleID, SourceKind: materialization.SourceKind, SourceRef: materialization.SourceRef, ActiveDefinitionPath: materialization.ActiveDefinitionPath, SeedFiles: agentWorkspaceSeedFilesForDaemon(materialization.SeedFiles), } } func agentWorkspaceSeedFilesForDaemon(seedFiles []agent.WorkspaceSeedFile) []agentWorkspaceSeedFile { if len(seedFiles) == 0 { return nil } items := make([]agentWorkspaceSeedFile, 0, len(seedFiles)) for _, seedFile := range seedFiles { items = append(items, agentWorkspaceSeedFile{Path: seedFile.Path, Content: seedFile.Content}) } return items } func (hub *Hub) UpdateAgentIdentity(machineID string, agentID string, name string, displayName string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return client.sessionLinkSender().Send(newAgentUpdateIdentityCommand(agentID, strings.TrimSpace(name), strings.TrimSpace(displayName))) } // SyncMachineCatalog pushes the current server-side machine agent catalog to // an online daemon using the same payload shape as a machine:agent:list // response. This is used after server-side mutations that should be reflected // locally immediately instead of waiting for the daemon's periodic pull sync. func (hub *Hub) SyncMachineCatalog(ctx context.Context, machineID string) error { if hub == nil { return runtimecontrol.ErrMachineOffline } client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } agents, err := hub.store.ListAgentsForMachine(ctx, machineID) if err != nil { return err } projectedAgents, err := projectMachineAgentsForDaemon(agents, hub.publicBaseURL, client.workspaceID, false, client.hasCapability(agentMaaSDirectCapability), hub.maasRuntimeBaseURL, client.hasCapability(agentVisibleReplyContractCapability)) if err != nil { return err } projectedAgents = client.filterProjectedAgentsByPermissionModeCapability(projectedAgents) if hub.hintReader != nil && client.workspaceID != "" { for i := range projectedAgents { hint, hintErr := hub.hintReader(ctx, client.workspaceID, projectedAgents[i].ID) if hintErr != nil { return fmt.Errorf("sync machine catalog: read extension catalog hint for agent %s: %w", projectedAgents[i].ID, hintErr) } projectedAgents[i].ExtensionCatalog = hint } } return client.sessionLinkSender().Send(newMachineAgentListResponseCommand(projectedAgents)) } // StopAgent tells the daemon to stop one agent. func (hub *Hub) StopAgent(machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } client.mu.Lock() client.stoppedAgents[agentID] = true client.mu.Unlock() return client.sessionLinkSender().Send(newAgentStopCommand(agentID, false)) } // ResetAgentSession stops the daemon runtime for a session reset without // treating the agent as manually stopped. Auto-wake remains governed by the // persisted agent and scope notification state. func (hub *Hub) ResetAgentSession(machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } err := client.sessionLinkSender().Send(newAgentStopCommand(agentID, true)) client.setAgentSession(agentID, "") client.setAgentRunning(agentID, false) client.setAgentStopped(agentID, false) return err } // InterruptAgentRun asks the daemon to interrupt the current runtime turn/run // for an agent while preserving its daemon-local session metadata. This is used // when a server-side operation needs to discard a pending runtime input without // resetting the whole agent session. func (hub *Hub) InterruptAgentRun(machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } err := client.sendJSON(map[string]any{"type": "agent:interrupt", "agentId": agentID}) client.setAgentRunning(agentID, false) client.setAgentStopped(agentID, false) return err } // NotifyMemoryAccessChanged pushes a WebSocket notification to the daemon // that memory access for the given agents has been updated (granted or // revoked). The daemon re-evaluates memory watcher state on receipt. func (hub *Hub) NotifyMemoryAccessChanged(machineID string, agentIDs []string) error { if hub == nil { return runtimecontrol.ErrMachineOffline } client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return client.sessionLinkSender().Send(newAgentMemoryAccessChangedCommand(agentIDs)) } // AssetsAccessChangedEvent is the unified daemon payload for managed // agent asset access changes. Consumers use EventID or (agent, asset type, version) // for idempotency when the durable outbox replays an event. type AssetsAccessChangedEvent struct { EventID string `json:"event_id,omitempty"` AgentIDs []string `json:"agent_ids"` AssetTypes []string `json:"asset_types,omitempty"` Version int64 `json:"version,omitempty"` OccurredAt time.Time `json:"occurred_at,omitempty"` Source string `json:"source,omitempty"` } // NotifyAssetsAccessChanged pushes the unified Memory/Skill access-change // event to a daemon machine. func (hub *Hub) NotifyAssetsAccessChanged(machineID string, event AssetsAccessChangedEvent) error { if hub == nil { return runtimecontrol.ErrMachineOffline } client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return client.sessionLinkSender().Send(newAgentAssetsAccessChangedCommand(event)) } // NotifyAssetsAccessChangedByAgent resolves the machine ID for a single agent // and pushes the unified access-change event to that daemon. func (hub *Hub) NotifyAssetsAccessChangedByAgent(ctx context.Context, workspaceID string, agentID string, event AssetsAccessChangedEvent) error { if hub == nil { return runtimecontrol.ErrMachineOffline } machineID, err := hub.store.MachineIDForAgent(ctx, workspaceID, agentID) if err != nil { return err } if machineID == "" { return runtimecontrol.ErrMachineOffline } if len(event.AgentIDs) == 0 { event.AgentIDs = []string{agentID} } return hub.NotifyAssetsAccessChanged(machineID, event) } // NotifyMemoryAccessChangedByAgent resolves the machine ID for the given agent // and then pushes the WebSocket notification. Returns ErrMachineOffline when // the daemon is offline or the agent has no machine assignment. func (hub *Hub) NotifyMemoryAccessChangedByAgent(ctx context.Context, workspaceID string, agentID string) error { if hub == nil { return runtimecontrol.ErrMachineOffline } machineID, err := hub.store.MachineIDForAgent(ctx, workspaceID, agentID) if err != nil { return err } if machineID == "" { return runtimecontrol.ErrMachineOffline } return hub.NotifyMemoryAccessChanged(machineID, []string{agentID}) } // ResetAgentWorkspace sends the reset-workspace control command. func (hub *Hub) ResetAgentWorkspace(machineID string, agentID string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } if err := client.sessionLinkSender().Send(newAgentResetWorkspaceCommand(agentID)); err != nil { return err } client.setAgentSession(agentID, "") client.setAgentRunning(agentID, false) client.setAgentStopped(agentID, false) return nil } // ListAgentWorkspace returns the daemon file tree response. func (hub *Hub) ListAgentWorkspace(ctx context.Context, machineID string, agentID string, dirPath string) ([]agent.WorkspaceFile, error) { page, err := hub.ListAgentWorkspacePage(ctx, machineID, agentID, dirPath, agent.WorkspaceListOptions{}) return page.Files, err } // ListAgentWorkspacePage returns one bounded daemon file tree response page. func (hub *Hub) ListAgentWorkspacePage(ctx context.Context, machineID string, agentID string, dirPath string, opts agent.WorkspaceListOptions) (agent.WorkspaceFilePage, error) { client := hub.client(machineID) if client == nil { return agent.WorkspaceFilePage{}, runtimecontrol.ErrMachineOffline } return hub.listWorkspace(ctx, client, agentID, "", dirPath, opts) } // ReadAgentWorkspaceFile returns the daemon file-content response. func (hub *Hub) ReadAgentWorkspaceFile(ctx context.Context, machineID string, agentID string, path string) (agent.WorkspaceFileContent, error) { client := hub.client(machineID) if client == nil { return agent.WorkspaceFileContent{}, runtimecontrol.ErrMachineOffline } return hub.readWorkspaceFile(ctx, client, agentID, "", path) } // ListAgentOutputTargetFiles returns a file tree rooted at the configured output target. func (hub *Hub) ListAgentOutputTargetFiles(ctx context.Context, machineID string, agentID string, targetPath string, dirPath string) ([]agent.WorkspaceFile, error) { page, err := hub.ListAgentOutputTargetFilesPage(ctx, machineID, agentID, targetPath, dirPath, agent.WorkspaceListOptions{}) return page.Files, err } // ListAgentOutputTargetFilesPage returns one bounded output-target file tree page. func (hub *Hub) ListAgentOutputTargetFilesPage(ctx context.Context, machineID string, agentID string, targetPath string, dirPath string, opts agent.WorkspaceListOptions) (agent.WorkspaceFilePage, error) { client := hub.client(machineID) if client == nil { return agent.WorkspaceFilePage{}, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.WorkspaceFilePage{}, err } return hub.listWorkspace(ctx, client, agentID, targetPath, dirPath, opts) } // ReadAgentOutputTargetFile reads one file rooted at the configured output target. func (hub *Hub) ReadAgentOutputTargetFile(ctx context.Context, machineID string, agentID string, targetPath string, path string) (agent.WorkspaceFileContent, error) { client := hub.client(machineID) if client == nil { return agent.WorkspaceFileContent{}, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.WorkspaceFileContent{}, err } return hub.readWorkspaceFile(ctx, client, agentID, targetPath, path) } func (hub *Hub) GetAgentReviewSummary(ctx context.Context, machineID string, agentID string, targetPath string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewSummary, error) { client := hub.client(machineID) if client == nil { return agent.AgentReviewSummary{}, runtimecontrol.ErrMachineOffline } // Capability gate runs BEFORE the delegation so a target-path capability // miss surfaces as the upgrade-required error rather than a downstream // RPC failure. The gate is a no-op for empty targetPath (default workdir). if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.AgentReviewSummary{}, err } return hub.reviewSummary(ctx, client, agentID, targetPath, baseRef, headRef, comparisonMode) } func (hub *Hub) GetAgentReviewSnapshot(ctx context.Context, machineID string, agentID string, targetPath string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewSnapshot, error) { client := hub.client(machineID) if client == nil { return agent.AgentReviewSnapshot{}, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.AgentReviewSnapshot{}, err } return hub.reviewSnapshot(ctx, client, agentID, targetPath, baseRef, headRef, comparisonMode) } func (hub *Hub) GetAgentReviewFingerprint(ctx context.Context, machineID string, agentID string, targetPath string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewFingerprint, error) { client := hub.client(machineID) if client == nil { return agent.AgentReviewFingerprint{}, runtimecontrol.ErrMachineOffline } // Target-path capability gate runs first: when both a non-empty targetPath // and a fingerprint capability miss are present, the target-path upgrade // error wins over ErrDaemonReviewFingerprintUnavailable so the frontend // surfaces the broader daemon upgrade requirement. When targetPath is // empty the gate is a no-op and the fingerprint capability miss (if any) // surfaces from the delegation as ErrDaemonReviewFingerprintUnavailable, // with zero snapshot/legacy RPC traffic. if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.AgentReviewFingerprint{}, err } return hub.reviewFingerprint(ctx, client, agentID, targetPath, baseRef, headRef, comparisonMode) } func (hub *Hub) ListAgentReviewFiles(ctx context.Context, machineID string, agentID string, targetPath string, scope string, baseRef string, headRef string, comparisonMode string) ([]agent.AgentReviewFile, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return nil, err } return hub.reviewFiles(ctx, client, agentID, targetPath, scope, baseRef, headRef, comparisonMode) } func (hub *Hub) GetAgentReviewDiff(ctx context.Context, machineID string, agentID string, targetPath string, scope string, path string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewDiff, error) { client := hub.client(machineID) if client == nil { return agent.AgentReviewDiff{}, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return agent.AgentReviewDiff{}, err } return hub.reviewDiff(ctx, client, agentID, targetPath, scope, path, baseRef, headRef, comparisonMode) } // GetAgentReviewDiffBatch resolves diffs for multiple paths in a single call. // It requires the same workspace-target-path capability as GetAgentReviewDiff // and routes through Hub.reviewDiffBatch, which uses the daemon's batch // endpoint when advertised and falls back to per-path fan-out otherwise. func (hub *Hub) GetAgentReviewDiffBatch(ctx context.Context, machineID string, agentID string, targetPath string, scope string, paths []string, baseRef string, headRef string, comparisonMode string) (map[string]AgentReviewDiffBatchResult, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineOffline } if err := client.requireWorkspaceTargetPathCapability(targetPath); err != nil { return nil, err } return hub.reviewDiffBatch(ctx, client, agentID, targetPath, scope, paths, baseRef, headRef, comparisonMode) } func (hub *Hub) ValidateAgentOutputTarget(ctx context.Context, machineID string, agentID string, kind string, targetPath string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return hub.validateOutputTarget(ctx, client, agentID, kind, targetPath) } // GetAgentTrajectoryHistory asks the daemon for locally retained trajectory // history. Items are ordered newest-first. func (hub *Hub) GetAgentTrajectoryHistory(ctx context.Context, machineID string, agentID string, limit int, before int64) ([]runtimecontrol.AgentTrajectoryItem, error) { client := hub.client(machineID) if client == nil { return nil, runtimecontrol.ErrMachineOffline } return hub.trajectoryHistory(ctx, client, agentID, limit, before) } // ApplyAgentSeedFiles overwrites the provided seed files inside an agent workspace. func (hub *Hub) ApplyAgentSeedFiles(ctx context.Context, machineID string, agentID string, seedFiles []agent.WorkspaceSeedFile) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return hub.applySeedFiles(ctx, client, agentID, seedFiles) } // DeleteAgentWorkspacePath removes one file or directory from an agent workspace. func (hub *Hub) DeleteAgentWorkspacePath(ctx context.Context, machineID string, agentID string, path string) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return hub.deleteWorkspacePath(ctx, client, agentID, path) } // ScanMachineWorkspace returns the daemon workspace scan response. func (hub *Hub) ScanMachineWorkspace(ctx context.Context, machineID string) ([]agent.WorkspaceDirectory, error) { page, err := hub.ScanMachineWorkspacePage(ctx, machineID, agent.WorkspaceListOptions{}) return page.Directories, err } // ScanMachineWorkspacePage returns one bounded daemon workspace scan page. func (hub *Hub) ScanMachineWorkspacePage(ctx context.Context, machineID string, opts agent.WorkspaceListOptions) (agent.WorkspaceDirectoryPage, error) { client := hub.client(machineID) if client == nil { return agent.WorkspaceDirectoryPage{}, runtimecontrol.ErrMachineOffline } return hub.scanMachineWorkspace(ctx, client, opts) } // DeleteMachineWorkspaceDirectory asks the daemon to delete one workspace directory. func (hub *Hub) DeleteMachineWorkspaceDirectory(ctx context.Context, machineID string, directoryName string) (bool, error) { client := hub.client(machineID) if client == nil { return false, runtimecontrol.ErrMachineOffline } return hub.deleteMachineWorkspaceDirectory(ctx, client, directoryName) } func (hub *Hub) RestoreAgentRecoveryPoint(ctx context.Context, machineID string, request runtimecontrol.RestoreAgentRecoveryPointRequest) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } return hub.restoreAgentRecoveryPoint(ctx, client, request) } func (hub *Hub) GCRecoveryPoints(machineID string, agentID string, points []rewind.MessageRecoveryPoint) error { client := hub.client(machineID) if client == nil { return runtimecontrol.ErrMachineOffline } requests := recoveryPointGCRequests(points) if len(requests) == 0 { return nil } return client.sessionLinkSender().Send(newAgentGCRecoveryPointsCommand(strings.TrimSpace(agentID), requests)) } func (hub *Hub) handleMessage(ctx context.Context, client *clientState, payload []byte) error { var base envelope if err := json.Unmarshal(payload, &base); err != nil { return fmt.Errorf("decode envelope: %w", err) } descriptor, ok := defaultSessionLinkCommandRegistry().Lookup(base.Type) if !ok { return nil } return hub.handleRegisteredSessionLinkCommand(ctx, client, descriptor, payload) } type sessionLinkCommandHandler func(hub *Hub, ctx context.Context, client *clientState, payload []byte) error var defaultSessionLinkCommandHandlers = map[string]sessionLinkCommandHandler{ sessionLinkCommandReady: handleSessionLinkReadyCommand, sessionLinkCommandAgentStatus: handleSessionLinkAgentStatusCommand, sessionLinkCommandAgentActivity: handleSessionLinkAgentActivityCommand, sessionLinkCommandAgentSession: handleSessionLinkAgentSessionCommand, sessionLinkCommandAgentTurnState: handleSessionLinkAgentTurnStateCommand, sessionLinkCommandAgentWakeStatus: handleSessionLinkAgentWakeStatusCommand, sessionLinkCommandAgentRuntimeEvent: handleSessionLinkAgentRuntimeEventCommand, sessionLinkCommandAgentRecoveryPointPrepared: handleSessionLinkAgentRecoveryPointPreparedCommand, sessionLinkCommandAgentTrajectory: handleSessionLinkAgentTrajectoryCommand, sessionLinkCommandAgentTrajectoryHistoryResult: handleSessionLinkAgentTrajectoryHistoryResultCommand, sessionLinkCommandAgentDeliverAck: handleSessionLinkAgentDeliverAckCommand, sessionLinkCommandAgentRestoreRecoveryPointDone: handleSessionLinkAgentRestoreRecoveryPointResultCommand, sessionLinkCommandAgentWorkspaceFileTree: handleSessionLinkAgentWorkspaceFileTreeCommand, sessionLinkCommandAgentWorkspaceFileContent: handleSessionLinkAgentWorkspaceFileContentCommand, sessionLinkCommandAgentWorkspaceApplySeedDone: handleSessionLinkAgentWorkspaceApplySeedFilesResultCommand, sessionLinkCommandAgentWorkspaceDeleteResult: handleSessionLinkAgentWorkspaceDeleteResultCommand, sessionLinkCommandAgentReviewSummaryResult: handleSessionLinkAgentReviewSummaryResultCommand, sessionLinkCommandAgentReviewFilesResult: handleSessionLinkAgentReviewFilesResultCommand, sessionLinkCommandAgentReviewDiffResult: handleSessionLinkAgentReviewDiffResultCommand, sessionLinkCommandOutputTargetValidateResult: handleSessionLinkAgentOutputTargetValidateResultCommand, sessionLinkCommandMachineWorkspaceScanResult: handleSessionLinkMachineWorkspaceScanResultCommand, sessionLinkCommandMachineWorkspaceDeleteResult: handleSessionLinkMachineWorkspaceDeleteResultCommand, sessionLinkCommandMachineWorkspacesList: handleSessionLinkMachineWorkspacesListCommand, sessionLinkCommandMachineAgentList: handleSessionLinkMachineAgentListCommand, sessionLinkCommandRemoteRequestResult: handleSessionLinkRemoteRequestResultCommand, } func (hub *Hub) handleRegisteredSessionLinkCommand(ctx context.Context, client *clientState, descriptor CommandDescriptor, payload []byte) error { handler := defaultSessionLinkCommandHandlers[descriptor.WireType] if handler == nil { return nil } return handler(hub, ctx, client, payload) } func handleSessionLinkReadyCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message readyMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode ready message: %w", err) } oldGeneration := client.currentSessionGeneration() client.setCapabilities(message.Capabilities) client.setDaemonTransportReady(message.DaemonTransport) if hub.elasticRequestLinks != nil && oldGeneration != "" && oldGeneration != client.currentSessionGeneration() { hub.elasticRequestLinks.CleanupSession(client.machineID, oldGeneration, "session_generation_changed") } client.setTokenRefreshProof(message.TokenRefreshProof) filteredRuntimes, filteredDefaults := filterReadyRuntimePayload(message.Runtimes, message.RuntimeDefaults) client.setMachineRuntimeCatalog(filteredRuntimes, filteredDefaults) runningAgents, blockedAgents, err := hub.filterRunnableAgentIDs(ctx, client, message.RunningAgents) if err != nil { client.clearReadyState() return err } blockedStartingAgents, err := hub.blockedStartingAgentIDs(ctx, client) if err != nil { client.clearReadyState() return err } if err := hub.store.MarkMachineReady(ctx, agent.MachineIdentity{MachineID: client.machineID, WorkspaceID: client.workspaceID}, agent.ReadyPayload{ Runtimes: filteredRuntimes, RuntimeDefaults: runtimeDefaultsToCore(filteredDefaults), RunningAgents: runningAgents, Hostname: message.Hostname, OS: message.OS, DaemonVersion: message.DaemonVersion, }); err != nil { client.clearReadyState() return err } // The complete ready application (running-set replacement, sessions, and // turn-state snapshot with its watermark) is applied atomically under a single // machine observation guard write acquisition, so a recovery reader holding // the read lease across a transaction commit can never observe the // intermediate running-set clear before the corresponding watermark/state // snapshot. client.applyReadySnapshot(runningAgents, agentIDSet(blockedStartingAgents), message.RunningAgentSessions, message.RunningAgentTurnStates, message.RunningAgentTurnStatesWatermark) if message.RunningAgentTurnStatesWatermark == nil && len(runningAgents) > 0 && len(message.RunningAgentTurnStates) == 0 && client.logger != nil { // A legacy daemon is online and has running agents but carried no turn // states AND no watermark (so it does not claim modern ordering authority // for an empty snapshot). Recovery applies a bounded grace before treating // these responders as recoverable; log it so the missing snapshot is // observable. A modern daemon always emits the watermark, so an empty // snapshot there is authoritative and not a diagnostic concern. client.logger.Debug("legacy daemon ready carried no turn states; recovery grace will bound the missing snapshot", slog.Int("running_agents", len(runningAgents))) } blockedAgents = appendMissingAgentIDs(blockedAgents, blockedStartingAgents) if !client.hasCapability(agentPermissionModeCapability) { if err := hub.store.ClearWorkspaceWriteAgentAuthTokensForMachine(ctx, client.machineID); err != nil { return err } } if err := client.stopUnsupportedRunningAgents(blockedAgents); err != nil { return err } if hub.realtime != nil { hub.realtime.EmitMachineCapabilities(client.workspaceID, client.machineID, filteredRuntimes, filteredDefaults, message.Hostname, message.OS, message.DaemonVersion) hub.realtime.EmitMachineStatus(client.workspaceID, client.machineID, "online") hub.realtime.EmitDaemonStatus(client.workspaceID, client.machineID, "online") } return nil } func handleSessionLinkAgentStatusCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentStatusMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent status message: %w", err) } running := message.Status == "active" if running && !hub.agentRuntimeAllowedOnClient(ctx, client, message.AgentID) { if client.logger != nil { client.logger.Warn("ignored running agent status from daemon without permission-mode support", slog.String("agent_id", message.AgentID)) } return nil } // A generation-fenced terminal lifecycle frame (Status "inactive" with an // expected identity) must match the current agent session BEFORE any store, // realtime, running/status/activity, or current-state mutation. A delayed stop // from a replaced generation is dropped entirely. The check runs under // client.mu via agentSessionMatches — the same client/session authority // serialization used by session replacement — and every frame of one // connection is processed serially on its read-agentteam goroutine, so the // check-then-apply is ordered at that boundary (no TOCTOU window across // observation-determining mutations, which are all under client.mu). A frame // without the optional fence (legacy daemon) is always authorized. if !clientTerminalFrameAuthorized(client, message.AgentID, message.ExpectedSessionID) { if client.logger != nil { client.logger.Info("ignored stale terminal agent status from replaced generation", slog.String("agent_id", message.AgentID), slog.String("expected_session_id", message.ExpectedSessionID)) } return nil } updated, err := hub.store.UpdateAgentStatusFromMachine(ctx, client.machineID, message.AgentID, message.Status) if err != nil { return err } if !updated { client.logger.Info("ignored agent status from unexpected machine", slog.String("agent_id", message.AgentID), slog.String("status", message.Status)) return nil } client.setAgentRunning(message.AgentID, running) client.setAgentStatus(message.AgentID, message.Status) if hub.realtime != nil { hub.realtime.EmitAgentStatusWithInstance(client.workspaceID, client.createdByUserID, message.AgentID, message.Status, nil) } return nil } func handleSessionLinkAgentActivityCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentActivityMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent activity message: %w", err) } running := strings.TrimSpace(message.Activity) != "" && strings.TrimSpace(message.Activity) != "offline" && strings.TrimSpace(message.Activity) != "sleeping" if running && !hub.agentRuntimeAllowedOnClient(ctx, client, message.AgentID) { if client.logger != nil { client.logger.Warn("ignored running agent activity from daemon without permission-mode support", slog.String("agent_id", message.AgentID)) } return nil } // Same generation fence as the status handler: a terminal activity frame // (Activity "offline" with an expected identity) must match the current agent // session BEFORE any store, realtime, running/status/activity, or current-state // mutation; a delayed stop from a replaced generation is dropped entirely. if !clientTerminalFrameAuthorized(client, message.AgentID, message.ExpectedSessionID) { if client.logger != nil { client.logger.Info("ignored stale terminal agent activity from replaced generation", slog.String("agent_id", message.AgentID), slog.String("expected_session_id", message.ExpectedSessionID)) } return nil } updated, err := hub.store.UpdateAgentActivityFromMachine(ctx, client.machineID, message.AgentID, message.Activity, message.Detail) if err != nil { return err } if !updated { client.logger.Info("ignored agent activity from unexpected machine", slog.String("agent_id", message.AgentID), slog.String("activity", message.Activity)) return nil } client.setAgentRunning(message.AgentID, running) client.setAgentActivity(message.AgentID, message.Activity, message.Detail) if hub.realtime != nil { hub.realtime.EmitAgentActivityWithInstance(client.workspaceID, client.createdByUserID, message.AgentID, message.Activity, message.Detail, nil) } return nil } func handleSessionLinkAgentSessionCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentSessionMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent session message: %w", err) } running := strings.TrimSpace(message.SessionID) != "" if running && !hub.agentRuntimeAllowedOnClient(ctx, client, message.AgentID) { if client.logger != nil { client.logger.Warn("ignored running agent session from daemon without permission-mode support", slog.String("agent_id", message.AgentID)) } return nil } // A generation-fenced terminal clear (empty session, expected identity) is a // compare-and-clear against the current agent session BEFORE any store, // realtime, or current-session mutation. A delayed stop from a replaced // generation must never clear a replacement session, so when the expected // identity does not match the current session the stale clear is dropped // entirely. A legacy empty clear without the optional fence keeps the // unconditional behavior. if !running && !clientTerminalFrameAuthorized(client, message.AgentID, message.ExpectedSessionID) { if client.logger != nil { client.logger.Info("ignored stale terminal session clear from replaced generation", slog.String("agent_id", message.AgentID), slog.String("expected_session_id", message.ExpectedSessionID)) } return nil } // AgentTeam-owned runtime authority is ephemeral connection state and must not // enter the legacy provider-session mirror. persistLegacySession := !client.hasCapability(agentRuntimeSessionAuthorityCapability) && !client.hasCapability(agentLocalSessionResumeCapability) updated, err := hub.store.UpdateAgentSessionFromMachine(ctx, client.machineID, message.AgentID, message.SessionID, persistLegacySession) if err != nil { return err } if !updated { client.logger.Info("ignored agent session from unexpected machine", slog.String("agent_id", message.AgentID)) } else { client.setAgentSession(message.AgentID, message.SessionID) client.setAgentRunning(message.AgentID, running) if hub.realtime != nil { hub.realtime.EmitAgentSessionWithInstance(client.workspaceID, client.createdByUserID, message.AgentID, message.SessionID, nil) } } return nil } // handleSessionLinkAgentTurnStateCommand stores the daemon-derived real Agent // turn execution state (active/queued/idle) in the Hub overlay. This is the // authoritative recovery working signal: presence/status/activity alone must not // be used to infer working. The state is validated against the machine and, when // the daemon supplies a sessionId, against the current agent session: a stale // turn_state from a replaced session is dropped (the current session's known // state is preserved), never clearing the Hub to unknown. The event transport // epoch, per-Agent session, and monotonic revision/watermark are all validated // INSIDE the single guarded setAgentTurnState critical section (machine guard // write side + client.mu), so old-epoch rejection is atomic with the mutation // and can never race a transport generation change. func handleSessionLinkAgentTurnStateCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentTurnStateMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent turn state message: %w", err) } trimmedAgentID := strings.TrimSpace(message.AgentID) if trimmedAgentID == "" { return fmt.Errorf("agent turn state requires agentId") } trimmedSessionID := strings.TrimSpace(message.SessionID) if hub.store != nil { ownedByMachine, err := hub.store.AgentBelongsToMachine(ctx, client.machineID, trimmedAgentID) if err != nil { return err } if !ownedByMachine { client.logger.Info("ignored agent turn state from unexpected machine", slog.String("agent_id", trimmedAgentID)) return nil } } // Test-only deterministic fence: the event is decoded and ownership-validated // here, just before the guarded apply. The atomic old-epoch barrier test // pauses at this boundary, changes the transport generation, resumes, and // proves the old-epoch event is dropped as a true zero-write. Inert in // production (nil hook). if hub.testTurnStateApplyHook != nil { hub.testTurnStateApplyHook(ctx) } client.setAgentTurnState(trimmedAgentID, strings.TrimSpace(message.TurnState), trimmedSessionID, message.Revision, strings.TrimSpace(message.Epoch)) return nil } func handleSessionLinkAgentWakeStatusCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentWakeStatusMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent wake status message: %w", err) } trimmedAgentID := strings.TrimSpace(message.AgentID) trimmedMessageID := strings.TrimSpace(message.MessageID) trimmedWakeToken := strings.TrimSpace(message.WakeToken) trimmedStatus := strings.TrimSpace(message.Status) trimmedTimestamp := strings.TrimSpace(message.Timestamp) if trimmedAgentID == "" || trimmedMessageID == "" || trimmedWakeToken == "" || trimmedStatus == "" || trimmedTimestamp == "" { return fmt.Errorf("daemon protocol violation: agent wake status requires agentId, messageId, wakeToken, status, and timestamp") } transient := message.Transient && client.hasCapability(transientWakeStatusDesktopCapability) var agentName string var agentDisplayName string if hub.store != nil { var ownedByMachine bool var err error agentName, agentDisplayName, ownedByMachine, err = hub.store.AgentNamesForMachine(ctx, client.machineID, trimmedAgentID) if err != nil { return err } if !ownedByMachine { client.logger.Info("ignored agent wake status from unexpected machine", slog.String("agent_id", trimmedAgentID)) return nil } if runtimeEvent, ok, err := runtimeEventFromWakeStatusMessage(client.workspaceID, client.machineID, message); err != nil { return err } else if ok && !transient { if err := hub.recordAgentRuntimeEvent(ctx, runtimeEvent); err != nil { return err } } } if hub.realtime != nil { hub.realtime.EmitAgentWakeStatus( client.workspaceID, trimmedAgentID, agentName, agentDisplayName, trimmedMessageID, trimmedWakeToken, strings.TrimSpace(message.Target), trimmedStatus, strings.TrimSpace(message.ReplyMessageID), strings.TrimSpace(message.Detail), trimmedTimestamp, ) } return nil } func handleSessionLinkAgentRuntimeEventCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentRuntimeEventMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent runtime event message: %w", err) } trimmedAgentID := strings.TrimSpace(message.AgentID) trimmedEventClass := strings.TrimSpace(message.EventClass) trimmedTimestamp := strings.TrimSpace(message.Timestamp) if trimmedAgentID == "" || trimmedEventClass == "" || trimmedTimestamp == "" { return fmt.Errorf("daemon protocol violation: agent runtime event requires agentId, eventClass, and timestamp") } if hub.store != nil { ownedByMachine, err := hub.store.AgentBelongsToMachine(ctx, client.machineID, trimmedAgentID) if err != nil { return err } if !ownedByMachine { client.logger.Info("ignored agent runtime event from unexpected machine", slog.String("agent_id", trimmedAgentID)) return nil } eventTime, err := time.Parse(time.RFC3339Nano, trimmedTimestamp) if err != nil { return fmt.Errorf("parse agent runtime event timestamp: %w", err) } if err := hub.recordAgentRuntimeEvent(ctx, agent.AgentRuntimeEvent{ WorkspaceID: client.workspaceID, AgentID: trimmedAgentID, MachineID: client.machineID, MessageID: strings.TrimSpace(message.MessageID), WakeToken: strings.TrimSpace(message.WakeToken), EventClass: trimmedEventClass, Target: strings.TrimSpace(message.Target), ReplyMessageID: strings.TrimSpace(message.ReplyMessageID), Detail: strings.TrimSpace(message.Detail), OccurredAt: eventTime, }); err != nil { return err } } if runtimeFailureEventClearsRunning(trimmedEventClass) { client.setAgentRunning(trimmedAgentID, false) } return nil } func handleSessionLinkAgentRecoveryPointPreparedCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentRecoveryPointPreparedMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent recovery point prepared message: %w", err) } if strings.TrimSpace(message.AgentID) == "" || strings.TrimSpace(message.MessageID) == "" || strings.TrimSpace(message.ThreadID) == "" || strings.TrimSpace(message.Strategy) == "" || strings.TrimSpace(message.RecoveryRef) == "" { return fmt.Errorf("daemon protocol violation: agent recovery point prepared requires agentId, messageId, threadId, strategy, and recoveryRef") } if hub.store != nil { trimmedAgentID := strings.TrimSpace(message.AgentID) ownedByMachine, err := hub.store.AgentBelongsToMachine(ctx, client.machineID, trimmedAgentID) if err != nil { return err } if !ownedByMachine { client.logger.Info("ignored agent recovery point from unexpected machine", slog.String("agent_id", trimmedAgentID)) return nil } pruned, err := hub.store.RecordMessageRecoveryPointAndPruneOtherThreads( ctx, client.workspaceID, strings.TrimSpace(message.MessageID), trimmedAgentID, strings.TrimSpace(message.ThreadID), strings.TrimSpace(message.Strategy), strings.TrimSpace(message.RecoveryRef), ) if err != nil { return err } if len(pruned) > 0 { if err := hub.GCRecoveryPoints(client.machineID, trimmedAgentID, pruned); err != nil && !errors.Is(err, runtimecontrol.ErrMachineOffline) { client.logger.Warn("gc pruned recovery points", slog.String("agent_id", trimmedAgentID), slog.Any("error", err)) } } } return nil } func handleSessionLinkAgentTrajectoryCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentTrajectoryMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode agent trajectory message: %w", err) } trimmedAgentID := strings.TrimSpace(message.AgentID) trimmedEntries := bytes.TrimSpace(message.Entries) if trimmedAgentID == "" || len(trimmedEntries) == 0 { return fmt.Errorf("daemon protocol violation: empty agent trajectory message") } if bytes.Equal(trimmedEntries, []byte("null")) { return fmt.Errorf("daemon protocol violation: agent trajectory entries must be an array") } if trimmedEntries[0] != '[' { return fmt.Errorf("daemon protocol violation: agent trajectory entries must be an array") } if hub.store != nil { ownedByMachine, err := hub.store.AgentBelongsToMachine(ctx, client.machineID, trimmedAgentID) if err != nil { return err } if !ownedByMachine { client.logger.Info("ignored agent trajectory from unexpected machine", slog.String("agent_id", trimmedAgentID)) return nil } } if hub.realtime != nil { hub.realtime.EmitAgentTrajectory(client.workspaceID, client.createdByUserID, trimmedAgentID, strings.TrimSpace(message.SessionID), message.Timestamp, trimmedEntries) } return nil } func handleSessionLinkAgentTrajectoryHistoryResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentTrajectoryHistoryResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode agent trajectory history result: %w", err) client.trajectoryHistoryTracker().FailAll(trajectoryHistoryResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: agent trajectory history result requestId is required") client.trajectoryHistoryTracker().FailAll(trajectoryHistoryResult{err: err}) return err } result := trajectoryHistoryResult{items: message.Items} if result.items == nil { result.items = []runtimecontrol.AgentTrajectoryItem{} } if trimmedError := strings.TrimSpace(message.Error); trimmedError != "" { result.err = errors.New(trimmedError) } client.trajectoryHistoryTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentDeliverAckCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message deliverAckMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode deliver ack message: %w", err) } client.deliverDeliverAck(message.AgentID, message.Seq, strings.TrimSpace(message.MessageID), strings.TrimSpace(message.InterruptPolicy), deliverAckResult{ accepted: message.Accepted, detail: strings.TrimSpace(message.Detail), }) return nil } func handleSessionLinkAgentRestoreRecoveryPointResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message agentRestoreRecoveryPointResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode agent restore recovery point result: %w", err) client.restoreRecoveryPointTracker().FailAll(restoreRecoveryPointResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: agent restore recovery point result requestId is required") client.restoreRecoveryPointTracker().FailAll(restoreRecoveryPointResult{err: err}) return err } result := restoreRecoveryPointResult{success: message.Success} if !message.Success { errText := strings.TrimSpace(message.Error) if errText == "" { errText = "agent restore recovery point failed" } result.err = errors.New(errText) } client.restoreRecoveryPointTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentWorkspaceFileTreeCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message fileTreeMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace file tree: %w", err) client.deliverTreeResult(fileTreeResult{err: err}) return err } if message.Files == nil { err := fmt.Errorf("daemon protocol violation: workspace file tree files must be an array") client.deliverTreeResult(fileTreeResult{err: err}) return err } files, err := converter.FromDaemonWorkspaceFiles(message.Files) if err != nil { err := fmt.Errorf("decode workspace file tree: %w", err) client.deliverTreeResult(fileTreeResult{err: err}) return err } client.deliverTreeResult(fileTreeResult{page: agent.WorkspaceFilePage{ Files: files, RootPath: strings.TrimSpace(message.RootPath), Limit: message.Limit, Cursor: strings.TrimSpace(message.Cursor), HasMore: message.HasMore, NextCursor: strings.TrimSpace(message.NextCursor), }}) return nil } func handleSessionLinkAgentWorkspaceFileContentCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message fileContentMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace file content: %w", err) client.fileContentTracker().FailAll(fileContentResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: workspace file content requestId is required") client.fileContentTracker().FailAll(fileContentResult{err: err}) return err } client.fileContentTracker().Deliver(trimmedRequestID, fileContentResult{content: agent.WorkspaceFileContent{Content: message.Content, Binary: message.Binary}}) return nil } func handleSessionLinkAgentWorkspaceApplySeedFilesResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message applySeedFilesResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace apply seed files result: %w", err) client.applySeedFilesTracker().FailAll(applySeedFilesResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: workspace apply seed files result requestId is required") client.applySeedFilesTracker().FailAll(applySeedFilesResult{err: err}) return err } result := applySeedFilesResult{success: message.Success} if !message.Success { errText := strings.TrimSpace(message.Error) if errText == "" { errText = "workspace apply seed files failed" } result.err = errors.New(errText) } client.applySeedFilesTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentWorkspaceDeleteResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message deleteWorkspacePathResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace delete result: %w", err) client.deleteWorkspacePathTracker().FailAll(deleteWorkspacePathResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: workspace delete result requestId is required") client.deleteWorkspacePathTracker().FailAll(deleteWorkspacePathResult{err: err}) return err } result := deleteWorkspacePathResult{success: message.Success} if !message.Success { errText := strings.TrimSpace(message.Error) if errText == "" { errText = "workspace delete failed" } result.err = errors.New(errText) } client.deleteWorkspacePathTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentReviewSummaryResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message reviewSummaryResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode review summary result: %w", err) client.reviewSummaryTracker().FailAll(reviewSummaryResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: review summary result requestId is required") client.reviewSummaryTracker().FailAll(reviewSummaryResult{err: err}) return err } result := reviewSummaryResult{} if trimmedError := strings.TrimSpace(message.Error); trimmedError != "" { result.err = reviewLegacyError(message.ErrorCode, trimmedError) } else if message.Summary == nil { result.err = fmt.Errorf("daemon protocol violation: review summary result summary is required") } else { result.summary = *message.Summary } client.reviewSummaryTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentReviewFilesResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message reviewFilesResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode review files result: %w", err) client.reviewFilesTracker().FailAll(reviewFilesResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: review files result requestId is required") client.reviewFilesTracker().FailAll(reviewFilesResult{err: err}) return err } result := reviewFilesResult{files: message.Files} if result.files == nil { result.files = []agent.AgentReviewFile{} } if trimmedError := strings.TrimSpace(message.Error); trimmedError != "" { result.err = reviewLegacyError(message.ErrorCode, trimmedError) } client.reviewFilesTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkAgentReviewDiffResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message reviewDiffResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode review diff result: %w", err) client.reviewDiffTracker().FailAll(reviewDiffResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: review diff result requestId is required") client.reviewDiffTracker().FailAll(reviewDiffResult{err: err}) return err } result := reviewDiffResult{} if trimmedError := strings.TrimSpace(message.Error); trimmedError != "" { result.err = reviewLegacyError(message.ErrorCode, trimmedError) } else if message.Diff == nil { result.err = fmt.Errorf("daemon protocol violation: review diff result diff is required") } else { result.diff = *message.Diff } client.reviewDiffTracker().Deliver(trimmedRequestID, result) return nil } // reviewLegacyError reconstructs a typed sentinel from the stable ErrorCode // the Desktop legacy handlers emit. The control-char path condition is a // product limitation that must surface to the HTTP layer as a controlled // "unsupported" response (422), not a uniform 500. The Desktop side // already stripped any offending path bytes from the underlying error // before populating Error/ErrorCode; the typed sentinel carries its own // fixed, path-free message. Unknown codes fall through to a generic // errors.New(errorText) so the wire stays forward-compatible. func reviewLegacyError(code string, errorText string) error { if strings.TrimSpace(code) == runtimecontrol.ReviewTargetUnsupportedLegacyErrorCode { return runtimecontrol.ErrReviewTargetUnsupported } return errors.New(errorText) } func handleSessionLinkAgentOutputTargetValidateResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message outputTargetValidateResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode output target validate result: %w", err) client.outputTargetValidateTracker().FailAll(outputTargetValidateResult{err: err}) return err } trimmedRequestID := strings.TrimSpace(message.RequestID) if trimmedRequestID == "" { err := fmt.Errorf("daemon protocol violation: output target validate result requestId is required") client.outputTargetValidateTracker().FailAll(outputTargetValidateResult{err: err}) return err } result := outputTargetValidateResult{valid: message.Valid} if trimmedError := strings.TrimSpace(message.Error); trimmedError != "" { result.err = errors.New(trimmedError) } else if !message.Valid { result.err = errors.New("output target validation failed") } client.outputTargetValidateTracker().Deliver(trimmedRequestID, result) return nil } func handleSessionLinkMachineWorkspaceScanResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message scanResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace scan result: %w", err) client.deliverScanResult(scanResult{err: err}) return err } if message.Directories == nil { err := fmt.Errorf("daemon protocol violation: workspace scan directories must be an array") client.deliverScanResult(scanResult{err: err}) return err } client.deliverScanResult(scanResult{page: agent.WorkspaceDirectoryPage{ Directories: message.Directories, Limit: message.Limit, Cursor: strings.TrimSpace(message.Cursor), HasMore: message.HasMore, NextCursor: strings.TrimSpace(message.NextCursor), }}) return nil } func handleSessionLinkMachineWorkspaceDeleteResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message deleteResultMessage if err := json.Unmarshal(payload, &message); err != nil { err := fmt.Errorf("decode workspace delete result: %w", err) client.deliverDeleteResult(deleteResult{err: err}) return err } client.deliverDeleteResult(deleteResult{success: message.Success}) return nil } func handleSessionLinkMachineWorkspacesListCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message machineWorkspacesRequestMessage if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode machine workspaces request: %w", err) } return hub.handleMachineWorkspacesRequest(ctx, client, message) } func handleSessionLinkMachineAgentListCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { agents, err := hub.store.ListAgentsForMachine(ctx, client.machineID) if err != nil { client.logger.Warn("list agents for machine", slog.Any("error", err)) return nil } projectedAgents, err := projectMachineAgentsForDaemon(agents, hub.publicBaseURL, client.workspaceID, false, client.hasCapability(agentMaaSDirectCapability), hub.maasRuntimeBaseURL, client.hasCapability(agentVisibleReplyContractCapability)) if err != nil { client.logger.Warn("project agents for daemon", slog.Any("error", err)) return nil } projectedAgents = client.filterProjectedAgentsByPermissionModeCapability(projectedAgents) if hub.hintReader != nil && client.workspaceID != "" { for i := range projectedAgents { hint, hintErr := hub.hintReader(ctx, client.workspaceID, projectedAgents[i].ID) if hintErr != nil { return fmt.Errorf("agent list: read extension catalog hint for agent %s: %w", projectedAgents[i].ID, hintErr) } projectedAgents[i].ExtensionCatalog = hint } } return client.sessionLinkSender().Send(newMachineAgentListResponseCommand(projectedAgents)) } func handleSessionLinkRemoteRequestResultCommand(hub *Hub, ctx context.Context, client *clientState, payload []byte) error { var message RemoteRequestResult[json.RawMessage] if err := json.Unmarshal(payload, &message); err != nil { return fmt.Errorf("decode remote request result: %w", err) } if hub.elasticRequestLinks != nil && hub.elasticRequestLinks.CompleteRemoteRequest(message) { return nil } requestID := strings.TrimSpace(message.RequestID) if requestID == "" { requestID = strings.TrimSpace(message.RemoteRequestID) } client.remoteRequestTracker().Deliver(requestID, message) return nil } func (hub *Hub) handleMachineWorkspacesRequest(ctx context.Context, client *clientState, message machineWorkspacesRequestMessage) error { requestID := strings.TrimSpace(message.RequestID) response := machineWorkspacesResponseMessage{ Type: "machine:workspaces:list:response", RequestID: requestID, Workspaces: []runtimecontrol.MachineWorkspace{}, } if requestID == "" { response.Error = "requestId is required" return client.sessionLinkSender().Send(response) } lister, ok := hub.store.(runtimecontrol.MachineWorkspaceLister) if !ok || lister == nil { response.Error = runtimecontrol.ErrMachineWorkspaceListUnavailable.Error() return client.sessionLinkSender().Send(response) } workspaces, err := lister.ListMachineOwnerWorkspaces(ctx, agent.MachineIdentity{ MachineID: client.machineID, WorkspaceID: client.workspaceID, CreatedByUserID: client.createdByUserID, }) if err != nil { response.Error = err.Error() return client.sessionLinkSender().Send(response) } sort.SliceStable(workspaces, func(i, j int) bool { return strings.ToLower(workspaces[i].Name) < strings.ToLower(workspaces[j].Name) }) response.Success = true response.Workspaces = workspaces return client.sessionLinkSender().Send(response) } func projectMachineAgentsForDaemon(items []agent.MachineAgent, publicBaseURL string, routeWorkspaceID string, _ bool, useDirectMaaS bool, maasRuntimeBaseURL string, visibleReplyContractSupported bool) ([]machineAgentForDaemon, error) { if len(items) == 0 { return []machineAgentForDaemon{}, nil } projected := make([]machineAgentForDaemon, 0, len(items)) for _, item := range items { var llm *agent.DaemonAgentLLM if agent.NormalizeAgentLLMMode(item.LLMMode) != agent.AgentLLMModeAgentTeamHosted || useDirectMaaS { var err error llm, err = agent.ProjectAgentLLMForDaemonWithOptions( agent.NewAgentLLM(item.LLMMode, item.BaseURL, item.APIKey, item.Headers, item.WireAPI), publicBaseURL, item.Runtime, agent.DaemonAgentLLMProjectionOptions{ UseDirectMaaS: useDirectMaaS, DirectMaaSRuntimeBaseURL: maasRuntimeBaseURL, RouteWorkspaceID: routeWorkspaceID, }, ) if err != nil { return nil, fmt.Errorf("project agent %s llm for daemon: %w", strings.TrimSpace(item.ID), err) } } llmMode := "" baseURL := "" apiKey := "" headers := map[string]string{} wireAPI := "" credentialKind := "" if llm != nil { llmMode = strings.TrimSpace(llm.Mode) baseURL = strings.TrimSpace(llm.BaseURL) apiKey = strings.TrimSpace(llm.APIKey) headers = make(map[string]string, len(llm.Headers)) for key, value := range llm.Headers { headers[key] = value } wireAPI = strings.TrimSpace(llm.WireAPI) credentialKind = strings.TrimSpace(llm.CredentialKind) } // VisibleReplyContract is always-on: emit enabled=true for clients that // declare the capability (visibleReplyContractSupported), regardless of // the stale per-agent DB value. effectiveVisibleReplyContract := visibleReplyContractSupported runtimePolicy := runtimePolicySnapshotWithVisibleReplyContract(item.RuntimePolicy, effectiveVisibleReplyContract) projectedAgent := machineAgentForDaemon{ ID: item.ID, Name: item.Name, DisplayName: item.DisplayName, Description: item.Description, Runtime: item.Runtime, PermissionMode: agent.DefaultAgentPermissionMode(item.PermissionMode), RuntimePolicy: runtimePolicyForDaemonFromStore(runtimePolicy), VisibleReplyContract: visibleReplyContractForDaemonFromEnabled(effectiveVisibleReplyContract), RuntimeDebug: runtimeDebugForDaemonFromEnabled(item.RuntimeDebugEnabled), Model: item.Model, ReasoningEffort: item.ReasoningEffort, LLMMode: llmMode, BaseURL: baseURL, APIKey: apiKey, Headers: headers, WireAPI: wireAPI, CredentialKind: credentialKind, Status: item.Status, Activity: item.Activity, } projected = append(projected, projectedAgent) } return projected, nil } func filterReadyRuntimePayload(runtimes []string, defaults map[string]runtime.RuntimeDefaults) ([]string, map[string]runtime.RuntimeDefaults) { filtered := make([]string, 0, len(runtimes)) seen := make(map[string]struct{}, len(runtimes)) for _, runtime := range runtimes { name := strings.TrimSpace(runtime) if name == "" { continue } if !isSupportedMachineRuntime(name) { continue } if _, ok := seen[name]; ok { continue } seen[name] = struct{}{} filtered = append(filtered, name) } filteredDefaults := make(map[string]runtime.RuntimeDefaults, len(filtered)) for _, runtime := range filtered { if value, ok := defaults[runtime]; ok { filteredDefaults[runtime] = value } } return filtered, filteredDefaults } func isSupportedMachineRuntime(runtime string) bool { return catalog.IsSupportedAgentRuntime(runtime) } func (hub *Hub) recordAgentRuntimeEvent(ctx context.Context, event agent.AgentRuntimeEvent) error { if hub.runtimeEvents == nil { return nil } return hub.runtimeEvents.RecordAgentRuntimeEvent(ctx, event) } func runtimeEventFromWakeStatusMessage(workspaceID string, machineID string, message agentWakeStatusMessage) (agent.AgentRuntimeEvent, bool, error) { eventClass := "" switch strings.TrimSpace(message.Status) { case "reply_sent": eventClass = agent.AgentRuntimeEventClassVisibleReplySent case "completed_without_visible_reply": eventClass = agent.AgentRuntimeEventClassCompletedWithoutVisibleReply case "failed": eventClass = agent.AgentRuntimeEventClassWakeFailed default: return agent.AgentRuntimeEvent{}, false, nil } eventTime, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(message.Timestamp)) if err != nil { return agent.AgentRuntimeEvent{}, false, fmt.Errorf("parse agent wake status timestamp: %w", err) } return agent.AgentRuntimeEvent{ WorkspaceID: strings.TrimSpace(workspaceID), AgentID: strings.TrimSpace(message.AgentID), MachineID: strings.TrimSpace(machineID), MessageID: strings.TrimSpace(message.MessageID), WakeToken: strings.TrimSpace(message.WakeToken), EventClass: eventClass, Target: strings.TrimSpace(message.Target), ReplyMessageID: strings.TrimSpace(message.ReplyMessageID), Detail: strings.TrimSpace(message.Detail), OccurredAt: eventTime, }, true, nil } func runtimeFailureEventClearsRunning(eventClass string) bool { switch strings.TrimSpace(eventClass) { case agent.AgentRuntimeEventClassWakeFailed, agent.AgentRuntimeEventClassRuntimeStartFailed, agent.AgentRuntimeEventClassRuntimeExitedUnexpectedly: return true default: return false } } // register swaps the current connection for one machine. Connection replacement // is observation-determining (the current client identity, connectedAt anchor, // and inherited sessions all feed a RecoveryObservation), so it takes the // machine observation guard write side FIRST (lock order guard -> hub.mu -> // client.mu). The superseded connection is closed AFTER releasing hub.mu and the // guard so the close path (which takes client.mu and performs socket I/O) never // runs while holding incompatible locks. func (hub *Hub) register(client *clientState) { var superseded *clientState if client.guard != nil { client.guard.acquireWrite() } hub.mu.Lock() if previous := hub.clients[client.machineID]; previous != nil { superseded = previous // The guard write side is already held above; use the non-locking // internal variant so no nested RWMutex acquisition occurs. client.inheritAgentSessionsLocked(previous) } if client.connectedAt.IsZero() { client.connectedAt = time.Now().UTC() } hub.clients[client.machineID] = client hub.mu.Unlock() if client.guard != nil { client.guard.releaseWrite() } if superseded != nil { superseded.close(websocket.CloseNormalClosure, "superseded by new connection") } } // unregister removes a connection for one machine. Connection removal is // observation-determining (Connected flips to false), so it takes the machine // observation guard write side FIRST (lock order guard -> hub.mu). A recovery // reader holding the read lease across a transaction commit must finish before // the removal lands; taking the write lock while holding hub.mu (hub.mu -> // guard) would deadlock a reader that holds the read lease and re-checks // connectivity under hub.mu, so the guard is always acquired before hub.mu. func (hub *Hub) unregister(client *clientState) bool { if client.guard != nil { client.guard.acquireWrite() defer client.guard.releaseWrite() } hub.mu.Lock() defer hub.mu.Unlock() current := hub.clients[client.machineID] if current != client { return false } delete(hub.clients, client.machineID) return true } func (hub *Hub) client(machineID string) *clientState { hub.mu.RLock() defer hub.mu.RUnlock() return hub.clients[machineID] } // MachineConnectedAt reports when the current authenticated daemon connection // for the machine established. ok=false means the machine is not connected. The // timestamp is anchored once per connection and never moves, so recovery grace // budgeting can rely on it not being extended by duplicate ready or stale events. // A zero timestamp means no anchor was recorded; the recovery reader treats that // as grace elapsed (fail toward recoverable) so an empty Hub can never stall. func (hub *Hub) MachineConnectedAt(machineID string) (time.Time, bool) { client := hub.client(strings.TrimSpace(machineID)) if client == nil { return time.Time{}, false } client.mu.Lock() defer client.mu.Unlock() return client.connectedAt, true } // observationGuard returns (creating lazily) the per-machine observation // read/write lease. It is shared across reconnects so a recovery reader holding // the read lease blocks turn-state replacement from every connection for the // machine. func (hub *Hub) observationGuard(machineID string) *machineObservationGuard { value, _ := hub.observationGuards.LoadOrStore(strings.TrimSpace(machineID), &machineObservationGuard{}) return value.(*machineObservationGuard) } // AcquireObservationGuard takes the machine observation read lease for one // connected machine and returns a release that MUST be called exactly once. // While the lease is held, no authoritative turn-state replacement for the // machine can commit, so an observation sampled under it stays current through // a row-locked RecoveryTick transaction commit. ok=false means the machine is // not connected and no lease is held. func (hub *Hub) AcquireObservationGuard(machineID string) (release func(), ok bool) { machineID = strings.TrimSpace(machineID) if machineID == "" || hub.client(machineID) == nil { return nil, false } guard := hub.observationGuard(machineID) guard.acquireRead() // Re-verify connectivity after acquiring the lease. register/unregister both // take the machine observation guard write side before mutating the client // map (guard -> hub.mu), so a client cannot be removed or replaced while this // read lease is held; the re-check only guards against a removal that already // committed before the read side was granted. if hub.client(machineID) == nil { guard.releaseRead() return nil, false } var once sync.Once release = func() { once.Do(guard.releaseRead) } return release, true } // ObserveAgentRuntime samples the immutable structured runtime observation for // one Agent responder on a connected machine. The caller MUST hold the machine // observation read lease (AcquireObservationGuard) for the whole time the // observation is in use across the recovery transition. ok=false means the // machine is not connected or the responder is not materializable. func (hub *Hub) ObserveAgentRuntime(machineID, agentID string) (run.RecoveryObservation, bool) { machineID = strings.TrimSpace(machineID) agentID = strings.TrimSpace(agentID) client := hub.client(machineID) if client == nil || agentID == "" { return run.RecoveryObservation{}, false } client.mu.Lock() defer client.mu.Unlock() record, known := client.agentTurnStateRecords[agentID] currentGeneration := strings.TrimSpace(client.sessionGeneration) currentSession := strings.TrimSpace(client.agentSessions[agentID]) // The state record is observable as current ONLY when its transport // generation matches the current connection generation AND its Agent session // EXACTLY matches the current session. A legacy no-session record (empty // SessionID) is never a wildcard: it is only observable while the current // Agent session is also empty. A stale prior-session or prior-epoch record is // never surfaced as known, so old active/idle can never drive park/clear. stateKnown := known && record.Generation == currentGeneration && record.SessionID == currentSession return run.RecoveryObservation{ MachineID: machineID, SessionGeneration: currentGeneration, AgentSession: currentSession, StateGeneration: strings.TrimSpace(record.Generation), StateSession: strings.TrimSpace(record.SessionID), Connected: true, ConnectedAt: client.connectedAt, State: strings.TrimSpace(record.State), StateKnown: stateKnown, Revision: record.Revision, HighWater: client.turnStateHighWater, }, true } // AgentTurnStateSnapshotStatus reports whether the current connection has applied // a ready snapshot that carried turn states and when. It backs the recovery // diagnostics and grace observability; the reader itself only needs the // connection anchor from MachineConnectedAt. func (hub *Hub) AgentTurnStateSnapshotStatus(machineID string) (received bool, receivedAt time.Time) { client := hub.client(strings.TrimSpace(machineID)) if client == nil { return false, time.Time{} } client.mu.Lock() defer client.mu.Unlock() return client.turnStateSnapshotReceived, client.turnStateSnapshotAt } // AgentRuntimeDiagnostics returns a compact observability snapshot of the // daemon/Hub runtime facts for one machine: connection anchor, current snapshot // status, and the per-agent turn states. It is read-only and safe under // concurrency; callers must treat it as a point-in-time observation. It reads // every clientState field under the single held client.mu and must never call a // method that takes client.mu again (non-reentrant). func (hub *Hub) AgentRuntimeDiagnostics(machineID string) map[string]any { client := hub.client(strings.TrimSpace(machineID)) if client == nil { return map[string]any{"connected": false} } client.mu.Lock() defer client.mu.Unlock() states := make(map[string]string, len(client.agentTurnStateRecords)) revisions := make(map[string]int64, len(client.agentTurnStateRecords)) for agentID, record := range client.agentTurnStateRecords { states[agentID] = record.State revisions[agentID] = record.Revision } diagnostics := map[string]any{ "connected": true, "connectedAt": client.connectedAt.UTC().Format(time.RFC3339Nano), "snapshotReceived": client.turnStateSnapshotReceived, "snapshotAt": client.turnStateSnapshotAt.UTC().Format(time.RFC3339Nano), "sessionGeneration": client.sessionGeneration, "runningAgents": len(client.runningAgents), "turnStates": states, "turnStatesReported": len(states), "turnStateRevisions": revisions, "turnStateHighWater": client.turnStateHighWater, "turnsSnapshotMissing": !client.turnStateSnapshotReceived, } return diagnostics } func (client *clientState) sendJSON(message any) error { return client.sendJSONWithWait(message, daemonWriteWait) } func (client *clientState) sendJSONWithWait(message any, writeWait time.Duration) error { client.sendMu.Lock() defer client.sendMu.Unlock() return client.writeJSONLocked(message, writeWait) } func (client *clientState) writeJSONLocked(message any, writeWait time.Duration) error { if writeWait <= 0 { writeWait = daemonWriteWait } _ = client.conn.SetWriteDeadline(time.Now().Add(writeWait)) return client.conn.WriteJSON(message) } func (client *clientState) sendPing(writeTimeout time.Duration) error { client.sendMu.Lock() defer client.sendMu.Unlock() _ = client.conn.SetWriteDeadline(time.Now().Add(writeTimeout)) return client.conn.WriteMessage(websocket.PingMessage, nil) } func (client *clientState) close(code int, text string) { client.mu.Lock() if client.closed { client.mu.Unlock() return } client.closed = true client.mu.Unlock() client.sendMu.Lock() defer client.sendMu.Unlock() _ = client.conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, text), time.Now().Add(time.Second)) _ = client.conn.Close() } // replaceRunningAgents applies a running-set replacement to the observation // state. It takes the machine observation guard write side first (guard -> // client.mu): running-set state deletion (including agentTurnStateRecords for agents // that left the running set) is observation-determining and must be fenced // against a recovery reader holding the read lease across a transaction commit. func (client *clientState) replaceRunningAgents(agentIDs []string) { client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceRunningAgentsStateLocked(agentIDs, false, nil) }) } func (client *clientState) replaceReadyRunningAgents(agentIDs []string, blockedStarts map[string]struct{}) { client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceRunningAgentsStateLocked(agentIDs, true, blockedStarts) }) } func (client *clientState) replaceReadyAgentSessions(agentIDs []string, sessions map[string]string) { client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceReadyAgentSessionsStateLocked(agentIDs, sessions) }) } // applyReadySnapshot atomically applies the full daemon ready snapshot to the // observation state: the running-set replacement, the session replacement, and // the turn-state snapshot (with its watermark) are applied under a SINGLE // machine observation guard write acquisition and a SINGLE client lock // acquisition. A recovery reader holding the read lease across a transaction // commit can therefore never observe the intermediate running-set clear before // the corresponding current watermark/state snapshot. func (client *clientState) applyReadySnapshot( agentIDs []string, blockedStarts map[string]struct{}, sessions map[string]string, turnStates map[string]string, watermark *int64, ) { client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceRunningAgentsStateLocked(agentIDs, true, blockedStarts) client.replaceReadyAgentSessionsStateLocked(agentIDs, sessions) client.replaceReadyAgentTurnStatesStateLocked(agentIDs, turnStates, watermark) }) } // clearReadyState atomically clears the daemon ready snapshot (running set and // sessions) under one guard write + client lock, so an error path cannot leave // a reader observing a partially-applied ready state. func (client *clientState) clearReadyState() { client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceRunningAgentsStateLocked(nil, false, nil) client.replaceReadyAgentSessionsStateLocked(nil, nil) }) } // replaceReadyAgentSessionsStateLocked replaces the session map for the running // set. The caller MUST hold the machine observation guard write side and // client.mu (via withObservationGuardWrite + client.mu.Lock). func (client *clientState) replaceReadyAgentSessionsStateLocked(agentIDs []string, sessions map[string]string) { allowed := agentIDSet(agentIDs) next := make(map[string]string, len(sessions)) for agentID, sessionID := range sessions { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { continue } if _, ok := allowed[trimmedAgentID]; !ok { continue } trimmedSessionID := strings.TrimSpace(sessionID) if trimmedSessionID == "" { continue } next[trimmedAgentID] = trimmedSessionID } if client.agentSessions == nil { client.agentSessions = make(map[string]string) } clear(client.agentSessions) for agentID, sessionID := range next { client.agentSessions[agentID] = sessionID } } // replaceReadyAgentTurnStates replays the daemon-derived real Agent turn // execution states across a reconnect so recovery ticks see the correct // active/queued/idle signal before the next pass. States outside the running // agent set are dropped. Recovery grace is anchored to connectedAt regardless, so // a duplicate ready or a stale incremental event cannot extend the bounded window. // // watermark is the same-connection snapshot revision as a PRESENCE-AWARE value: // nil means the ready omitted the field (a legacy daemon that does not claim // modern ordering authority) and the snapshot is applied with the pre-ordering // clear-and-replace semantics; non-nil (including zero) is an authoritative // current-epoch snapshot watermark even when the snapshot is empty, and is // applied by per-agent merge: // // - An agent already carrying a strictly newer increment (its per-agent // revision > watermark) is left untouched (an older snapshot must not // overwrite a newer per-agent increment). // - An agent in the snapshot with an absent or <= watermark per-agent revision // is materialized at the watermark. // - An agent absent from the snapshot whose per-agent revision <= watermark is // cleared, so an explicit empty current snapshot retires stale working state // (e.g. an old "active" from a previous session). // - Only an accepted (non-regressing) snapshot updates the connection // high-water and the snapshot diagnostics. func (client *clientState) replaceReadyAgentTurnStates(agentIDs []string, states map[string]string, watermark *int64) { // Exclusive boundary: a recovery reader holding the machine observation // read lease across a transaction commit must not observe a ready-snapshot // turn-state replacement landing mid-flight. Write side of the per-machine // guard, then the client lock. client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() client.replaceReadyAgentTurnStatesStateLocked(agentIDs, states, watermark) }) } // replaceReadyAgentTurnStatesStateLocked applies the turn-state snapshot to the // observation state. The caller MUST hold the machine observation guard write // side and client.mu (via withObservationGuardWrite + client.mu.Lock). // // Every applied snapshot record is bound to the CURRENT transport generation and // the agent's CURRENT session, so a fresh connection generation can establish // its own revision sequence and a replaced session's stale state is never // observable as current. func (client *clientState) replaceReadyAgentTurnStatesStateLocked(agentIDs []string, states map[string]string, watermark *int64) { allowed := agentIDSet(agentIDs) next := make(map[string]string, len(states)) for agentID, state := range states { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { continue } if _, ok := allowed[trimmedAgentID]; !ok { continue } trimmedState := strings.TrimSpace(state) if trimmedState == "" { continue } next[trimmedAgentID] = trimmedState } currentGeneration := strings.TrimSpace(client.sessionGeneration) if watermark == nil { // Legacy ready without a watermark: clear-and-replace with the // pre-ordering behavior. It never claims modern ordering authority and // never uses the per-agent merge, so an old daemon cannot accidentally // retire newer per-agent increments. The per-agent revision is preserved // (a legacy ready does not reset ordering state), and the snapshot // diagnostics stay aligned with the legacy "missing snapshot" semantics. priorRevisions := make(map[string]int64, len(client.agentTurnStateRecords)) for agentID, record := range client.agentTurnStateRecords { priorRevisions[agentID] = record.Revision } if client.agentTurnStateRecords == nil { client.agentTurnStateRecords = make(map[string]agentTurnStateRecord, len(next)) } clear(client.agentTurnStateRecords) for agentID, state := range next { client.agentTurnStateRecords[agentID] = agentTurnStateRecord{ State: state, Revision: priorRevisions[agentID], Generation: currentGeneration, SessionID: strings.TrimSpace(client.agentSessions[agentID]), } } if len(next) > 0 || len(agentIDs) == 0 { client.turnStateSnapshotReceived = true client.turnStateSnapshotAt = time.Now().UTC() } else { client.turnStateSnapshotReceived = false client.turnStateSnapshotAt = time.Time{} } return } wm := *watermark if client.agentTurnStateRecords == nil { client.agentTurnStateRecords = make(map[string]agentTurnStateRecord, len(next)) } for agentID, state := range next { // Merge: an older snapshot (lower watermark) must not overwrite a newer // per-agent increment. `>=` re-applies at the same watermark (duplicate/ // refresh ready) and applies for absent/<= revisions; a strictly-older // snapshot leaves a newer increment untouched. An absent per-agent revision // means the agent has never been seen on this connection, so even a base // watermark (0) snapshot materializes it. record, seen := client.agentTurnStateRecords[agentID] revision := record.Revision if !seen || wm >= revision { client.agentTurnStateRecords[agentID] = agentTurnStateRecord{ State: state, Revision: wm, Generation: currentGeneration, SessionID: strings.TrimSpace(client.agentSessions[agentID]), } } } // Clear agents absent from the snapshot unless they carry a STRICTLY newer // per-agent revision than this snapshot watermark, so a stale "active" state // from a previous session (or a legacy agent that left the running set) can // never block recovery, while a delayed older empty snapshot never clears an // agent with a newer increment. for agentID, record := range client.agentTurnStateRecords { if _, still := next[agentID]; !still && wm >= record.Revision { delete(client.agentTurnStateRecords, agentID) } } // Only an accepted (non-regressing) current snapshot updates the high-water // and the snapshot diagnostics. A genuinely older empty snapshot that could // not clear newer increments must not regress either. if wm >= client.turnStateHighWater { client.turnStateHighWater = wm client.turnStateSnapshotReceived = true client.turnStateSnapshotAt = time.Now().UTC() } } // replaceRunningAgentsStateLocked applies a running-set replacement to the // observation state. The caller MUST hold the machine observation guard write // side and client.mu (via withObservationGuardWrite + client.mu.Lock). func (client *clientState) replaceRunningAgentsStateLocked(agentIDs []string, preserveStarts bool, blockedStarts map[string]struct{}) { next := make(map[string]bool, len(agentIDs)) for _, agentID := range agentIDs { trimmed := strings.TrimSpace(agentID) if trimmed == "" { continue } next[trimmed] = true } if preserveStarts { for agentID := range next { delete(client.startingAgents, agentID) } for agentID := range blockedStarts { delete(client.startingAgents, agentID) } for agentID := range client.startingAgents { next[agentID] = true } } else { clear(client.startingAgents) } client.runningAgents = next if client.agentStatuses == nil { client.agentStatuses = make(map[string]string) } if client.agentActivities == nil { client.agentActivities = make(map[string]agentActivityState) } for agentID := range client.agentStatuses { if !next[agentID] { delete(client.agentStatuses, agentID) } } for agentID := range client.agentActivities { if !next[agentID] { delete(client.agentActivities, agentID) } } for agentID := range client.agentTurnStateRecords { if !next[agentID] { delete(client.agentTurnStateRecords, agentID) } } for agentID := range next { client.agentStatuses[agentID] = "active" } } func (client *clientState) setCapabilities(capabilities []string) { next := make(map[string]struct{}, len(capabilities)) for _, capability := range capabilities { trimmed := strings.TrimSpace(capability) if trimmed == "" { continue } next[trimmed] = struct{}{} } client.mu.Lock() client.capabilities = next client.mu.Unlock() } func (client *clientState) setDaemonTransportReady(transport daemonTransportReady) { endpointVersions := make(map[string]map[int]struct{}) for _, endpoint := range transport.SupportedEndpoints { endpointID := strings.TrimSpace(endpoint.EndpointID) if endpointID == "" || endpoint.Version <= 0 { continue } if endpointVersions[endpointID] == nil { endpointVersions[endpointID] = make(map[int]struct{}) } endpointVersions[endpointID][endpoint.Version] = struct{}{} } // sessionGeneration is an observation-determining field: a recovery reader // holding the machine observation read lease across a transaction commit must // not observe a mid-flight session-generation swap. Write side of the guard. client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() previousGeneration := strings.TrimSpace(client.sessionGeneration) nextGeneration := strings.TrimSpace(transport.SessionGeneration) client.sessionGeneration = nextGeneration client.transportProtocol = transport.SchemaVersion client.endpointRegistryVersion = strings.TrimSpace(transport.EndpointRegistryVersion) client.supportedEndpointVersions = endpointVersions // A transport generation change retires the previous generation's // generation-scoped turn-state records and resets the connection high-water // so a fresh generation can establish its own revision sequence. Records // bound to an earlier generation must never be observable as current. if previousGeneration != "" && previousGeneration != nextGeneration { for agentID, record := range client.agentTurnStateRecords { if record.Generation != "" && record.Generation != nextGeneration { delete(client.agentTurnStateRecords, agentID) } } client.turnStateHighWater = 0 client.turnStateSnapshotReceived = false client.turnStateSnapshotAt = time.Time{} } }) } func (client *clientState) currentSessionGeneration() string { client.mu.Lock() defer client.mu.Unlock() return strings.TrimSpace(client.sessionGeneration) } func (client *clientState) supportsElasticRequestLinks() bool { client.mu.Lock() defer client.mu.Unlock() return client.hasCapabilityLocked(daemonTransportCapabilityV1) && client.hasCapabilityLocked(daemonTransportElasticRequestCapability) && client.transportProtocol == defaultElasticRequestLinkProtocolVersion && strings.TrimSpace(client.sessionGeneration) != "" } func (client *clientState) supportsSessionRemoteRequests() bool { client.mu.Lock() defer client.mu.Unlock() return client.hasCapabilityLocked(daemonTransportCapabilityV1) && client.transportProtocol == defaultElasticRequestLinkProtocolVersion && strings.TrimSpace(client.sessionGeneration) != "" } func (client *clientState) supportsRemoteEndpoint(endpointID string, version int, registryVersion string) bool { endpointID = strings.TrimSpace(endpointID) registryVersion = strings.TrimSpace(registryVersion) client.mu.Lock() defer client.mu.Unlock() if !client.hasCapabilityLocked(daemonEndpointsCapabilityV1) { return false } if registryVersion != "" && strings.TrimSpace(client.endpointRegistryVersion) != registryVersion { return false } versions := client.supportedEndpointVersions[endpointID] _, ok := versions[version] return ok } // canFallBackToSessionDataPlane reports whether the connected daemon can carry // this endpoint's payload via the typed RemoteRequestAssignment envelope over // the session link. Used as the data-plane fallback for ElasticPreferred // endpoints that have no legacy command (FallbackCommand=none) when the // elastic transport is operationally unavailable — busy, at capacity, // emergency disabled, or dial timeout. Capability/registry mismatch is // intentionally NOT a fallback condition: an old daemon that does not // advertise the endpoint cannot accept the typed envelope either, so the // caller must surface a stable upgrade-required signal instead. func (client *clientState) canFallBackToSessionDataPlane(descriptor RemoteEndpointDescriptor) bool { if client == nil { return false } if strings.TrimSpace(client.currentSessionGeneration()) == "" { return false } if !client.supportsSessionRemoteRequests() { return false } return client.supportsRemoteEndpoint(descriptor.EndpointID, descriptor.Version, descriptor.RegistryVersion) } func (client *clientState) setMachineRuntimeCatalog(runtimes []string, defaults map[string]runtime.RuntimeDefaults) { nextRuntimes := append([]string(nil), runtimes...) nextDefaults := make(map[string]runtime.RuntimeDefaults, len(defaults)) for runtimeName, value := range defaults { nextDefaults[runtimeName] = value } client.mu.Lock() client.runtimeCatalogKnown = true client.runtimes = nextRuntimes client.runtimeDefaults = nextDefaults client.mu.Unlock() } func (client *clientState) machineRuntimeCatalog() ([]string, map[string]runtime.RuntimeDefaults, bool) { client.mu.Lock() defer client.mu.Unlock() if !client.runtimeCatalogKnown { return nil, nil, false } runtimes := append([]string(nil), client.runtimes...) defaults := make(map[string]runtime.RuntimeDefaults, len(client.runtimeDefaults)) for runtimeName, value := range client.runtimeDefaults { defaults[runtimeName] = value } return runtimes, defaults, true } func (client *clientState) workspaceRuntimeMachineSnapshot() (im.WorkspaceRuntimeMachineSnapshot, bool) { if client == nil { return im.WorkspaceRuntimeMachineSnapshot{}, false } machineID := strings.TrimSpace(client.machineID) if machineID == "" { return im.WorkspaceRuntimeMachineSnapshot{}, false } client.mu.Lock() if !client.runtimeCatalogKnown { client.mu.Unlock() return im.WorkspaceRuntimeMachineSnapshot{}, false } liveAgentIDs := make(map[string]struct{}, len(client.runningAgents)+len(client.agentSessions)) sessions := make(map[string]string, len(client.agentSessions)) for agentID, running := range client.runningAgents { trimmedAgentID := strings.TrimSpace(agentID) if !running || trimmedAgentID == "" || client.startingAgents[trimmedAgentID] { continue } liveAgentIDs[trimmedAgentID] = struct{}{} } for agentID, sessionID := range client.agentSessions { trimmedAgentID := strings.TrimSpace(agentID) trimmedSessionID := strings.TrimSpace(sessionID) if trimmedAgentID == "" || trimmedSessionID == "" { continue } liveAgentIDs[trimmedAgentID] = struct{}{} sessions[trimmedAgentID] = trimmedSessionID } client.mu.Unlock() agentIDs := make([]string, 0, len(liveAgentIDs)) for agentID := range liveAgentIDs { agentIDs = append(agentIDs, agentID) } sort.Strings(agentIDs) agents := make([]im.WorkspaceRuntimeAgentSnapshot, 0, len(agentIDs)) for _, agentID := range agentIDs { agents = append(agents, im.WorkspaceRuntimeAgentSnapshot{ AgentID: agentID, Status: "active", SessionID: sessions[agentID], }) } return im.WorkspaceRuntimeMachineSnapshot{ MachineID: machineID, OwnerUserID: strings.TrimSpace(client.createdByUserID), Agents: agents, }, true } func (client *clientState) setTokenRefreshProof(token string) { hash := "" if token = strings.TrimSpace(token); token != "" { hash = secrets.HashToken(token) } client.mu.Lock() client.tokenRefreshProofHash = hash client.mu.Unlock() } func (client *clientState) hasCapability(capability string) bool { client.mu.Lock() ok := client.hasCapabilityLocked(capability) client.mu.Unlock() return ok } func (client *clientState) hasCapabilityLocked(capability string) bool { _, ok := client.capabilities[strings.TrimSpace(capability)] return ok } func (hub *Hub) agentTokenRefreshWorkspaceWriteAllowed(machineID string, proof string) bool { client := hub.client(machineID) if client == nil { return false } return client.agentTokenRefreshWorkspaceWriteAllowed(proof) } func (client *clientState) agentTokenRefreshWorkspaceWriteAllowed(proof string) bool { proof = strings.TrimSpace(proof) if proof == "" { return false } client.mu.Lock() defer client.mu.Unlock() return client.hasCapabilityLocked(agentPermissionModeCapability) && client.tokenRefreshProofHash != "" && secrets.TokenMatchesHash(proof, client.tokenRefreshProofHash) } func (client *clientState) requireWorkspaceTargetPathCapability(targetPath string) error { // Exact-empty check, matching the workspace handlers' semantics: only // the empty string means "use agent default workdir". TrimSpace would // let a whitespace-only targetPath bypass the capability gate while // still flowing through to the handler as a real path. if targetPath == "" { return nil } if client.hasCapability(workspaceTargetPathCapability) { return nil } return runtimecontrol.ErrDaemonWorkspaceTargetPathUpgradeRequired } func (client *clientState) requireAgentPermissionModeCapabilityLocked(permissionMode string) error { if !agentPermissionModeRequiresDaemonCapability(permissionMode) { return nil } if client.hasCapabilityLocked(agentPermissionModeCapability) { return nil } return runtimecontrol.ErrDaemonPermissionModeUpgradeRequired } func (client *clientState) requireMaaSDirectCapabilityLocked(command agent.StartCommand) error { if !startCommandRequiresMaaSDirect(command) { return nil } if client.hasCapabilityLocked(agentMaaSDirectCapability) { return nil } return runtimecontrol.ErrDaemonMaaSDirectUpgradeRequired } func startCommandRequiresMaaSDirect(command agent.StartCommand) bool { return strings.TrimSpace(command.Config.CredentialKind) == agent.DaemonLLMCredentialKindMaaSCapability } func (hub *Hub) filterRunnableAgentIDs(ctx context.Context, client *clientState, agentIDs []string) ([]string, []string, error) { normalized := make([]string, 0, len(agentIDs)) seen := make(map[string]struct{}, len(agentIDs)) for _, agentID := range agentIDs { trimmed := strings.TrimSpace(agentID) if trimmed == "" { continue } if _, ok := seen[trimmed]; ok { continue } seen[trimmed] = struct{}{} normalized = append(normalized, trimmed) } if len(normalized) > daemonReadyRunningAgentsLimit { return nil, nil, fmt.Errorf("daemon ready runningAgents exceeds limit %d", daemonReadyRunningAgentsLimit) } if len(normalized) == 0 || client.hasCapability(agentPermissionModeCapability) { return normalized, nil, nil } filtered := make([]string, 0, len(normalized)) blocked := make([]string, 0) for _, agentID := range normalized { permissionMode, found, err := hub.store.AgentPermissionModeForMachine(ctx, client.machineID, agentID) if err != nil { return nil, nil, err } if !found || agentPermissionModeRequiresDaemonCapability(permissionMode) { if found { blocked = append(blocked, agentID) } continue } filtered = append(filtered, agentID) } return filtered, blocked, nil } func (hub *Hub) blockedStartingAgentIDs(ctx context.Context, client *clientState) ([]string, error) { if client.hasCapability(agentPermissionModeCapability) { return nil, nil } agentIDs := client.startingAgentIDs() if len(agentIDs) == 0 { return nil, nil } blocked := make([]string, 0) for _, agentID := range agentIDs { permissionMode, found, err := hub.store.AgentPermissionModeForMachine(ctx, client.machineID, agentID) if err != nil { return nil, err } if !found || agentPermissionModeRequiresDaemonCapability(permissionMode) { blocked = append(blocked, agentID) } } return blocked, nil } func (client *clientState) startingAgentIDs() []string { client.mu.Lock() defer client.mu.Unlock() agentIDs := make([]string, 0, len(client.startingAgents)) for agentID := range client.startingAgents { agentIDs = append(agentIDs, agentID) } return agentIDs } func agentIDSet(agentIDs []string) map[string]struct{} { if len(agentIDs) == 0 { return nil } set := make(map[string]struct{}, len(agentIDs)) for _, agentID := range agentIDs { set[agentID] = struct{}{} } return set } func appendMissingAgentIDs(agentIDs []string, additional []string) []string { if len(additional) == 0 { return agentIDs } seen := make(map[string]struct{}, len(agentIDs)+len(additional)) for _, agentID := range agentIDs { seen[agentID] = struct{}{} } for _, agentID := range additional { if _, ok := seen[agentID]; ok { continue } seen[agentID] = struct{}{} agentIDs = append(agentIDs, agentID) } return agentIDs } func (hub *Hub) agentRuntimeAllowedOnClient(ctx context.Context, client *clientState, agentID string) bool { if err := hub.requireAgentRuntimeAllowedOnClient(ctx, client, agentID); err != nil { if client.logger != nil { if errors.Is(err, runtimecontrol.ErrDaemonPermissionModeUpgradeRequired) { client.logger.Warn("ignored running agent from daemon without permission-mode support", slog.String("agent_id", agentID)) } else { client.logger.Warn("load agent permission mode for running-state gate", slog.String("agent_id", agentID), slog.Any("error", err)) } } return false } return true } func (hub *Hub) requireAgentRuntimeAllowedOnClient(ctx context.Context, client *clientState, agentID string) error { if client.hasCapability(agentPermissionModeCapability) { return nil } permissionMode, found, err := hub.store.AgentPermissionModeForMachine(ctx, client.machineID, agentID) if err != nil { return err } if !found || agentPermissionModeRequiresDaemonCapability(permissionMode) { return runtimecontrol.ErrDaemonPermissionModeUpgradeRequired } return nil } func (client *clientState) filterProjectedAgentsByPermissionModeCapability(items []machineAgentForDaemon) []machineAgentForDaemon { if client.hasCapability(agentPermissionModeCapability) { return items } filtered := make([]machineAgentForDaemon, 0, len(items)) for _, item := range items { if agentPermissionModeRequiresDaemonCapability(item.PermissionMode) { continue } filtered = append(filtered, item) } return filtered } func (client *clientState) stopUnsupportedRunningAgents(agentIDs []string) error { for _, agentID := range agentIDs { if err := client.sessionLinkSender().Send(newAgentStopCommand(agentID, false)); err != nil { return err } } return nil } func agentPermissionModeRequiresDaemonCapability(permissionMode string) bool { return agent.DefaultAgentPermissionMode(permissionMode) != agent.AgentPermissionModeFullAccess } func startPermissionModeError(err error) error { if errors.Is(err, agent.ErrPermissionModeCapabilityRequired) { return runtimecontrol.ErrDaemonPermissionModeUpgradeRequired } return err } type agentStartState struct { wasRunning bool wasStarting bool wasStopped bool } func (client *clientState) markAgentStarting(agentID string, clearStopped bool) agentStartState { client.mu.Lock() defer client.mu.Unlock() state := agentStartState{ wasRunning: client.runningAgents[agentID], wasStarting: client.startingAgents[agentID], wasStopped: client.stoppedAgents[agentID], } if client.startingAgents == nil { client.startingAgents = make(map[string]bool) } if client.runningAgents == nil { client.runningAgents = make(map[string]bool) } client.runningAgents[agentID] = true client.startingAgents[agentID] = true if clearStopped { delete(client.stoppedAgents, agentID) } return state } func (client *clientState) restoreAgentStartState(agentID string, state agentStartState) { client.mu.Lock() defer client.mu.Unlock() if state.wasRunning { client.runningAgents[agentID] = true } else { delete(client.runningAgents, agentID) } if state.wasStarting { if client.startingAgents == nil { client.startingAgents = make(map[string]bool) } client.startingAgents[agentID] = true } else { delete(client.startingAgents, agentID) } if state.wasStopped { client.stoppedAgents[agentID] = true } else { delete(client.stoppedAgents, agentID) } } func (client *clientState) setAgentRunning(agentID string, running bool) { client.mu.Lock() defer client.mu.Unlock() if running { if client.runningAgents == nil { client.runningAgents = make(map[string]bool) } client.runningAgents[agentID] = true delete(client.startingAgents, agentID) delete(client.stoppedAgents, agentID) return } delete(client.runningAgents, agentID) delete(client.startingAgents, agentID) delete(client.agentStatuses, agentID) delete(client.agentActivities, agentID) } func (client *clientState) setAgentStatus(agentID string, status string) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return } trimmedStatus := strings.TrimSpace(status) client.mu.Lock() defer client.mu.Unlock() if trimmedStatus == "" { delete(client.agentStatuses, trimmedAgentID) return } if client.agentStatuses == nil { client.agentStatuses = make(map[string]string) } client.agentStatuses[trimmedAgentID] = trimmedStatus } // agentStatus returns the stored agent lifecycle status for one agent. func (client *clientState) agentStatus(agentID string) string { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return "" } client.mu.Lock() defer client.mu.Unlock() return client.agentStatuses[trimmedAgentID] } func (client *clientState) setAgentActivity(agentID string, activity string, detail string) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return } trimmedActivity := strings.TrimSpace(activity) trimmedDetail := strings.TrimSpace(detail) client.mu.Lock() defer client.mu.Unlock() if trimmedActivity == "" && trimmedDetail == "" { delete(client.agentActivities, trimmedAgentID) return } if client.agentActivities == nil { client.agentActivities = make(map[string]agentActivityState) } client.agentActivities[trimmedAgentID] = agentActivityState{ Activity: trimmedActivity, Detail: trimmedDetail, } } // agentActivity returns the stored agent activity and detail for one agent. func (client *clientState) agentActivity(agentID string) (string, string) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return "", "" } client.mu.Lock() defer client.mu.Unlock() state := client.agentActivities[trimmedAgentID] return state.Activity, state.Detail } // setAgentSession replaces or clears the current Agent session. A session // replacement/clear is observation-determining (the per-Agent session feeds the // turn-state staleness fence and the RecoveryObservation identity match), so it // takes the machine guard write side + client.mu. On replacement or clear, any // prior-session turn-state record is atomically retired so the stale state can // never be observable as current. func (client *clientState) setAgentSession(agentID string, sessionID string) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return } trimmedSessionID := strings.TrimSpace(sessionID) client.withObservationGuardWrite(func() { client.mu.Lock() defer client.mu.Unlock() current := strings.TrimSpace(client.agentSessions[trimmedAgentID]) if trimmedSessionID == "" { delete(client.agentSessions, trimmedAgentID) } else { if client.agentSessions == nil { client.agentSessions = make(map[string]string) } client.agentSessions[trimmedAgentID] = trimmedSessionID } // On EVERY Agent session replacement or clear (including empty→non-empty // and non-empty→empty), atomically retire any existing turn-state record // that is not provably bound to the exact new session. A legacy no-session // record (empty SessionID) is retired when a non-empty session is set; a // prior-session record is retired when the session is cleared or replaced. // A stale active/idle can never survive to park/clear a later session. if current != trimmedSessionID { if record, ok := client.agentTurnStateRecords[trimmedAgentID]; ok && record.SessionID != trimmedSessionID { delete(client.agentTurnStateRecords, trimmedAgentID) } } }) } // agentSessionMatches reports whether the current agent session for one agent // equals the expected session identity. It backs the generation-fenced terminal // compare-and-clear: a clear is applied only while the current session still // matches the dying generation, so a delayed stop from a replaced generation is // dropped instead of deleting the replacement's session. func (client *clientState) agentSessionMatches(agentID string, expectedSessionID string) bool { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return false } client.mu.Lock() defer client.mu.Unlock() return strings.TrimSpace(client.agentSessions[trimmedAgentID]) == strings.TrimSpace(expectedSessionID) } // clientTerminalFrameAuthorized reports whether a generation-fenced terminal // lifecycle frame (inactive status, offline activity, or empty session clear) // may be applied. A frame with an expected session identity is authorized only // while the Hub client's current agent session still matches; a frame without // the optional fence (legacy daemon) is always authorized. The check runs under // client.mu — the same client/session authority serialization used by session // replacement — and every frame of one connection is processed serially on its // read-agentteam goroutine, so the check-then-apply is ordered at that boundary. func clientTerminalFrameAuthorized(client *clientState, agentID string, expectedSessionID string) bool { if strings.TrimSpace(expectedSessionID) == "" { return true } return client.agentSessionMatches(agentID, expectedSessionID) } // setAgentTurnState records the daemon-derived real Agent turn execution state // (active/queued/idle) for one agent, bound to the exact transport generation and // Agent session that produced it. An empty state clears the entry so the Hub // reports unknown for that agent (old daemon / no session). // // epoch is the event's transport session generation. When non-empty it MUST // match the current connection's sessionGeneration; an old-epoch event is a // true zero-write (state, revision, high-water, and diagnostics all unchanged). // A legacy event with an empty epoch is bound to the current generation at // application (explicitly documented compatibility behavior) so its record is // only observable while the current generation owns the connection. // // sessionID, when non-empty, must match the current agent session; otherwise the // turn_state is a stale event from a replaced session and is DROPPED — the // current session's known state is preserved and never cleared to unknown. // // revision is the same-connection monotonic ordering revision. When non-zero, an // event whose revision is older than or equal to the applied high-water is a // stale/duplicate increment that must not overwrite a newer snapshot or event and // is DROPPED. On apply the per-agent revision and the connection high-water // advance. A zero revision is a legacy (pre-ordering) daemon event: ordering // cannot be proven, so only the epoch/session fence guards it. // // The epoch, session, and revision comparisons all run under a SINGLE critical // section (machine guard write side + client.mu), so old-epoch rejection is // atomic with the mutation. func (client *clientState) setAgentTurnState(agentID, turnState, sessionID string, revision int64, epoch string) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return } trimmedState := strings.TrimSpace(turnState) trimmedSessionID := strings.TrimSpace(sessionID) trimmedEpoch := strings.TrimSpace(epoch) // Exclusive boundary: a recovery reader holding the machine observation // read lease across a transaction commit must not observe a turn-state // replacement landing mid-flight. Write side of the per-machine guard. if client.guard != nil { client.guard.acquireWrite() defer client.guard.releaseWrite() } client.mu.Lock() defer client.mu.Unlock() // Atomic old-epoch rejection: a NON-EMPTY event transport generation must be // EXACTLY equal to the current connection generation — including when the // current generation is EMPTY (a non-empty event epoch against an // empty-generation connection is a stale/foreign-connection event). It is a // true zero-write — no state, revision, high-water, or diagnostics change. A // legacy event with an empty epoch is bound to the current generation at // application. currentEpoch := strings.TrimSpace(client.sessionGeneration) if trimmedEpoch != "" && trimmedEpoch != currentEpoch { return } // Bound the record to the generation that produced it: legacy events (empty // epoch) are bound to the current generation at application. boundGeneration := trimmedEpoch if boundGeneration == "" { boundGeneration = currentEpoch } if trimmedSessionID != "" && strings.TrimSpace(client.agentSessions[trimmedAgentID]) != trimmedSessionID { // A stale turn_state from a replaced session: drop it and preserve the // current session's known state (never clear to unknown here). return } // A legacy event with an EMPTY session id is bound to the current Agent // session at application (or stays no-session only when the current Agent also // has no session). The stored record is then only observable while the exact // session matches, so it can never survive or authorize a later session. boundSession := trimmedSessionID if boundSession == "" { boundSession = strings.TrimSpace(client.agentSessions[trimmedAgentID]) } if revision > 0 { // An older/equal increment must not overwrite a newer snapshot or event. if revision <= client.turnStateHighWater { return } client.turnStateHighWater = revision } else { // Legacy (pre-ordering) event: no revision to order by, but still track the // agent at the current high-water so a later snapshot merge knows it has // been observed and can drop it when the agent leaves the running set. revision = client.turnStateHighWater } if trimmedState == "" { delete(client.agentTurnStateRecords, trimmedAgentID) return } if client.agentTurnStateRecords == nil { client.agentTurnStateRecords = make(map[string]agentTurnStateRecord) } client.agentTurnStateRecords[trimmedAgentID] = agentTurnStateRecord{ State: trimmedState, Revision: revision, Generation: boundGeneration, SessionID: boundSession, } } // agentTurnState returns the stored real Agent turn execution state for one // agent and whether it is known. Unknown means the daemon does not support the // turn-state overlay (old daemon), has not reported it, or the machine is not // connected. func (client *clientState) agentTurnState(agentID string) (string, bool) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return "", false } client.mu.Lock() record, ok := client.agentTurnStateRecords[trimmedAgentID] client.mu.Unlock() return strings.TrimSpace(record.State), ok } // inheritAgentSessions carries the previous connection's active sessions into // this (new) connection across a reconnect. It is observation-determining // (agentSessions feeds the session staleness of turn-state events), so it takes // the machine observation guard write side. func (client *clientState) inheritAgentSessions(previous *clientState) { client.withObservationGuardWrite(func() { client.inheritAgentSessionsLocked(previous) }) } // inheritAgentSessionsLocked carries the previous connection's active sessions // into this (new) connection. The caller MUST already hold the machine // observation guard write side (register does); the method itself acquires only // each client's own client.mu, never the guard, so it can be composed inside an // outer guarded lifecycle helper without nested RWMutex acquisition. func (client *clientState) inheritAgentSessionsLocked(previous *clientState) { if previous == nil { return } // A reconnecting daemon replays ready/session state after registration. // Carry the last active runtime sessions through that short transition; // accepted ready/session replay still replaces or clears this map. sessions := previous.agentSessionSnapshot() if len(sessions) == 0 { return } client.mu.Lock() defer client.mu.Unlock() if client.agentSessions == nil { client.agentSessions = make(map[string]string, len(sessions)) } for agentID, sessionID := range sessions { if strings.TrimSpace(client.agentSessions[agentID]) == "" { client.agentSessions[agentID] = sessionID } } } func (client *clientState) agentSessionSnapshot() map[string]string { client.mu.Lock() defer client.mu.Unlock() if len(client.agentSessions) == 0 { return nil } sessions := make(map[string]string, len(client.agentSessions)) for agentID, sessionID := range client.agentSessions { trimmedAgentID := strings.TrimSpace(agentID) trimmedSessionID := strings.TrimSpace(sessionID) if trimmedAgentID == "" || trimmedSessionID == "" { continue } sessions[trimmedAgentID] = trimmedSessionID } return sessions } func (client *clientState) agentSession(agentID string) string { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return "" } client.mu.Lock() sessionID := client.agentSessions[trimmedAgentID] client.mu.Unlock() return strings.TrimSpace(sessionID) } func (client *clientState) agentRuntimeStatus(agentID string) (string, string, string, bool) { trimmedAgentID := strings.TrimSpace(agentID) if trimmedAgentID == "" { return "", "", "", false } client.mu.Lock() status := strings.TrimSpace(client.agentStatuses[trimmedAgentID]) activityState := client.agentActivities[trimmedAgentID] client.mu.Unlock() activity := strings.TrimSpace(activityState.Activity) detail := strings.TrimSpace(activityState.Detail) return status, activity, detail, status != "" || activity != "" || detail != "" } func (client *clientState) setAgentStopped(agentID string, stopped bool) { client.mu.Lock() defer client.mu.Unlock() if stopped { if client.stoppedAgents == nil { client.stoppedAgents = make(map[string]bool) } client.stoppedAgents[agentID] = true return } delete(client.stoppedAgents, agentID) } func (client *clientState) isAgentStopped(agentID string) bool { client.mu.Lock() stopped := client.stoppedAgents[agentID] client.mu.Unlock() return stopped } func (client *clientState) isAgentRunning(agentID string) bool { client.mu.Lock() running := client.runningAgents[agentID] || client.startingAgents[agentID] client.mu.Unlock() return running } func (client *clientState) deliverToRunningAgent(ctx context.Context, agentID string, seq int64, message message.ReceiveMessage) (deliverAckResult, error) { key := deliverAckKey(agentID, seq, message.MessageID, message.InterruptPolicy) payload := newAgentDeliverCommand(agentID, seq, message) payload.InterruptPolicy = strings.TrimSpace(payload.InterruptPolicy) return client.deliverAckTracker().Wait(ctx, key, func() error { return client.sessionLinkSender().Send(payload) }) } func (client *clientState) startAgentForMessageAndWaitAck(ctx context.Context, command agent.StartCommand) (deliverAckResult, error) { if command.WakeMessage == nil { return deliverAckResult{}, errors.New("agent start wake message is required") } key := deliverAckKey(command.AgentID, 0, command.WakeMessage.MessageID, command.WakeMessage.InterruptPolicy) return client.deliverAckTracker().Wait(ctx, key, func() error { return client.sendStartCommandIfAllowed(command) }) } func (client *clientState) deliverDeliverAck(agentID string, seq int, messageID string, interruptPolicy string, result deliverAckResult) { key := deliverAckKey(agentID, int64(seq), messageID, interruptPolicy) client.deliverAckTracker().Deliver(key, result) } func deliverAckKey(agentID string, seq int64, messageID string, interruptPolicy string) string { trimmedAgentID := strings.TrimSpace(agentID) return strings.Join([]string{ trimmedAgentID, fmt.Sprintf("%d", seq), strings.TrimSpace(messageID), strings.TrimSpace(interruptPolicy), }, "\x00") } func (client *clientState) listWorkspace(ctx context.Context, agentID string, targetPath string, dirPath string, opts agent.WorkspaceListOptions) (agent.WorkspaceFilePage, error) { client.workspaceRequestMu.Lock() defer client.workspaceRequestMu.Unlock() resultCh := make(chan fileTreeResult, 1) client.workspaceStateMu.Lock() client.pendingTree = resultCh client.workspaceStateMu.Unlock() defer func() { client.workspaceStateMu.Lock() client.pendingTree = nil client.workspaceStateMu.Unlock() }() opts = normalizeWorkspaceListOptions(opts) command := newAgentWorkspaceListCommand(agentID, strings.TrimSpace(targetPath), dirPath, opts) if err := client.sessionLinkSender().Send(command); err != nil { return agent.WorkspaceFilePage{}, err } select { case <-ctx.Done(): return agent.WorkspaceFilePage{}, ctx.Err() case result := <-resultCh: page := result.page if page.Limit <= 0 { page.Limit = opts.Limit } if page.Cursor == "" { page.Cursor = opts.Cursor } return page, result.err } } func normalizeWorkspaceListOptions(opts agent.WorkspaceListOptions) agent.WorkspaceListOptions { if opts.Limit <= 0 { opts.Limit = agent.DefaultWorkspaceListLimit } if opts.Limit > agent.MaxWorkspaceListLimit { opts.Limit = agent.MaxWorkspaceListLimit } opts.Cursor = strings.TrimSpace(opts.Cursor) return opts } func (client *clientState) readWorkspaceFile(ctx context.Context, agentID string, targetPath string, path string) (agent.WorkspaceFileContent, error) { requestID := fmt.Sprintf("req_%d", time.Now().UnixNano()) result, err := client.fileContentTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentWorkspaceReadCommand(agentID, strings.TrimSpace(targetPath), path, requestID)) }) if err != nil { return agent.WorkspaceFileContent{}, err } if result.err != nil { return agent.WorkspaceFileContent{}, result.err } return result.content, nil } func (client *clientState) applySeedFiles(ctx context.Context, agentID string, seedFiles []agent.WorkspaceSeedFile) error { requestID := fmt.Sprintf("seed_%d", time.Now().UnixNano()) result, err := client.applySeedFilesTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentWorkspaceApplySeedFilesCommand(strings.TrimSpace(agentID), requestID, seedFiles)) }) if err != nil { return err } if result.err != nil { return result.err } if !result.success { return errors.New("workspace apply seed files failed") } return nil } func (client *clientState) deleteWorkspacePath(ctx context.Context, agentID string, path string) error { requestID := fmt.Sprintf("delete_%d", time.Now().UnixNano()) result, err := client.deleteWorkspacePathTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentWorkspaceDeleteCommand(strings.TrimSpace(agentID), strings.TrimSpace(path), requestID)) }) if err != nil { return err } if result.err != nil { return result.err } if !result.success { return errors.New("workspace delete failed") } return nil } func (client *clientState) reviewSummary(ctx context.Context, agentID string, targetPath string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewSummary, error) { requestID := fmt.Sprintf("review_summary_%d", time.Now().UnixNano()) result, err := client.reviewSummaryTracker().Wait(ctx, requestID, func() error { // targetPath is a raw filesystem path: do not TrimSpace. The Desktop // handler treats "" as "use agent default workdir" and any non-empty // value as exact — trimming would silently rewrite "/repo " to // "/repo" and point at a different worktree. Refs/mode are still // trimmed (enum-like control strings). return client.sessionLinkSender().Send(newAgentReviewSummaryCommand(strings.TrimSpace(agentID), requestID, targetPath, strings.TrimSpace(baseRef), strings.TrimSpace(headRef), strings.TrimSpace(comparisonMode))) }) if err != nil { return agent.AgentReviewSummary{}, err } return result.summary, result.err } func (client *clientState) reviewFiles(ctx context.Context, agentID string, targetPath string, scope string, baseRef string, headRef string, comparisonMode string) ([]agent.AgentReviewFile, error) { requestID := fmt.Sprintf("review_files_%d", time.Now().UnixNano()) result, err := client.reviewFilesTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentReviewFilesCommand(strings.TrimSpace(agentID), requestID, targetPath, strings.TrimSpace(scope), strings.TrimSpace(baseRef), strings.TrimSpace(headRef), strings.TrimSpace(comparisonMode))) }) if err != nil { return nil, err } return result.files, result.err } func (client *clientState) reviewDiff(ctx context.Context, agentID string, targetPath string, scope string, path string, baseRef string, headRef string, comparisonMode string) (agent.AgentReviewDiff, error) { requestID := fmt.Sprintf("review_diff_%d", time.Now().UnixNano()) result, err := client.reviewDiffTracker().Wait(ctx, requestID, func() error { // Both targetPath (worktree root) and path (Git file path) are raw — // see reviewSummary for the rationale. return client.sessionLinkSender().Send(newAgentReviewDiffCommand(strings.TrimSpace(agentID), requestID, targetPath, strings.TrimSpace(scope), path, strings.TrimSpace(baseRef), strings.TrimSpace(headRef), strings.TrimSpace(comparisonMode))) }) if err != nil { return agent.AgentReviewDiff{}, err } return result.diff, result.err } func (client *clientState) validateOutputTarget(ctx context.Context, agentID string, kind string, targetPath string) error { requestID := fmt.Sprintf("validate_output_target_%d", time.Now().UnixNano()) result, err := client.outputTargetValidateTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentOutputTargetValidateCommand(strings.TrimSpace(agentID), requestID, strings.TrimSpace(kind), strings.TrimSpace(targetPath))) }) if err != nil { return err } if result.err != nil { return result.err } if !result.valid { return errors.New("output target validation failed") } return nil } func (client *clientState) trajectoryHistory(ctx context.Context, agentID string, limit int, before int64) ([]runtimecontrol.AgentTrajectoryItem, error) { requestID := fmt.Sprintf("trajectory_history_%d", time.Now().UnixNano()) result, err := client.trajectoryHistoryTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentTrajectoryHistoryCommand(strings.TrimSpace(agentID), requestID, limit, before)) }) if err != nil { return nil, err } return result.items, result.err } func (client *clientState) restoreAgentRecoveryPoint(ctx context.Context, request runtimecontrol.RestoreAgentRecoveryPointRequest) error { requestID := fmt.Sprintf("restore_%d", time.Now().UnixNano()) result, err := client.restoreRecoveryPointTracker().Wait(ctx, requestID, func() error { return client.sessionLinkSender().Send(newAgentRestoreRecoveryPointCommand(requestID, trimRestoreAgentRecoveryPointRequest(request))) }) if err != nil { return err } return result.err } func (client *clientState) scanWorkspace(ctx context.Context, opts agent.WorkspaceListOptions) (agent.WorkspaceDirectoryPage, error) { client.workspaceRequestMu.Lock() defer client.workspaceRequestMu.Unlock() resultCh := make(chan scanResult, 1) client.workspaceStateMu.Lock() client.pendingScan = resultCh client.workspaceStateMu.Unlock() defer func() { client.workspaceStateMu.Lock() client.pendingScan = nil client.workspaceStateMu.Unlock() }() opts = normalizeWorkspaceListOptions(opts) if err := client.sessionLinkSender().Send(newMachineWorkspaceScanCommand(opts)); err != nil { return agent.WorkspaceDirectoryPage{}, err } select { case <-ctx.Done(): return agent.WorkspaceDirectoryPage{}, ctx.Err() case result := <-resultCh: page := result.page if page.Limit <= 0 { page.Limit = opts.Limit } if page.Cursor == "" { page.Cursor = opts.Cursor } return page, result.err } } func (client *clientState) deleteWorkspaceDirectory(ctx context.Context, directoryName string) (bool, error) { client.workspaceRequestMu.Lock() defer client.workspaceRequestMu.Unlock() resultCh := make(chan deleteResult, 1) client.workspaceStateMu.Lock() client.pendingDelete = resultCh client.workspaceStateMu.Unlock() defer func() { client.workspaceStateMu.Lock() client.pendingDelete = nil client.workspaceStateMu.Unlock() }() if err := client.sessionLinkSender().Send(newMachineWorkspaceDeleteCommand(directoryName)); err != nil { return false, err } select { case <-ctx.Done(): return false, ctx.Err() case result := <-resultCh: return result.success, result.err } } func (client *clientState) deliverTreeResult(result fileTreeResult) { client.workspaceStateMu.Lock() defer client.workspaceStateMu.Unlock() if client.pendingTree != nil { client.pendingTree <- result } } func (client *clientState) deliverScanResult(result scanResult) { client.workspaceStateMu.Lock() defer client.workspaceStateMu.Unlock() if client.pendingScan != nil { client.pendingScan <- result } } func (client *clientState) deliverDeleteResult(result deleteResult) { client.workspaceStateMu.Lock() defer client.workspaceStateMu.Unlock() if client.pendingDelete != nil { client.pendingDelete <- result } }