diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 4a7f18fffe..7c360101ac 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -128,6 +128,7 @@ import { getTotalCost } from "@/common/utils/tokens/usageAggregator"; import type { CompactionCompletionMetadata } from "@/common/types/compaction"; import { CompactionHandler } from "./compactionHandler"; import { RetryManager, type RetryFailureError, type RetryStatusEvent } from "./retryManager"; +import type { EffectRunner } from "./di/effectRunner"; import type { TelemetryService } from "./telemetryService"; import type { BackgroundProcessManager } from "./backgroundProcessManager"; @@ -543,6 +544,13 @@ export interface AgentSessionStreamManager { isStreaming(workspaceId: string): boolean; getStreamInfo(workspaceId: string): AgentSessionActiveStreamInfo | undefined; replayStream(workspaceId: string, options?: { afterTimestamp?: number }): Promise; + /** + * The runner the stream manager's clock-driven fibers use; the session's + * `RetryManager` schedules its backoff on the same clock. Absent on test + * doubles and the AIService fallback, which leaves RetryManager on the + * global runtime. + */ + readonly effectRunner?: EffectRunner; } /** Keeps AgentSession coupled only to the AI operations and events it consumes. */ @@ -952,7 +960,8 @@ export class AgentSession { async () => { await this.retryActiveStream(); }, - (event) => this.handleRetryStatusChange(event) + (event) => this.handleRetryStatusChange(event), + this.streamManager.effectRunner ); this.attachAiListeners(); diff --git a/src/node/services/di/layers/app.ts b/src/node/services/di/layers/app.ts index 2ef424fc95..bfb3221321 100644 --- a/src/node/services/di/layers/app.ts +++ b/src/node/services/di/layers/app.ts @@ -4,7 +4,7 @@ import { AppFiberScopeLive } from "@/node/services/di/appFiberScope"; import { EffectRunnerLive } from "@/node/services/di/effectRunner"; import type { AppTags } from "@/node/services/di/tags"; import { CoreLive, MemoryMetaLive } from "./core"; -import { CoreOptionsFromDesktopLive, CrossCuttingLive } from "./desktop"; +import { CoreOptionsFromDesktopLive, CrossCuttingLive, DesktopLive } from "./desktop"; import { StoresLive } from "./stores"; /** @@ -20,14 +20,16 @@ import { StoresLive } from "./stores"; * captures its building context, so placing it there keeps that context to the * stores plus references (`Clock`, …). Above them the graph replays the * constructor's former order: memory metadata, the cross-cutting services, the - * core options derived from them, then the staged core graph (which reads - * `MemoryMeta` and `WorkspaceMcpOverrides` from those layers directly). + * core options derived from them, the staged core graph (which reads + * `MemoryMeta` and `WorkspaceMcpOverrides` from those layers directly), and + * finally the desktop group layers with their wiring. */ export function AppLive(stores: ConfigStores): Layer.Layer { const runtimeSeams = AppFiberScopeLive.pipe( Layer.provideMerge(EffectRunnerLive.pipe(Layer.provideMerge(StoresLive(stores)))) ); - return CoreLive.pipe( + return DesktopLive.pipe( + Layer.provideMerge(CoreLive), Layer.provideMerge(CoreOptionsFromDesktopLive), Layer.provideMerge(CrossCuttingLive), Layer.provideMerge(MemoryMetaLive), diff --git a/src/node/services/di/layers/core.ts b/src/node/services/di/layers/core.ts index 63737554a9..a1ba9b894f 100644 --- a/src/node/services/di/layers/core.ts +++ b/src/node/services/di/layers/core.ts @@ -13,7 +13,7 @@ import { AIService } from "@/node/services/aiService"; import { BackgroundProcessManager } from "@/node/services/backgroundProcessManager"; import type { CoreOptions, CoreServices, CoreServicesOptions } from "@/node/services/coreServices"; import { AppFiberScopeLive } from "@/node/services/di/appFiberScope"; -import { EffectRunnerLive } from "@/node/services/di/effectRunner"; +import { EffectRunnerLive, EffectRunnerTag } from "@/node/services/di/effectRunner"; import { AI, BackgroundProcessManagerTag, @@ -90,11 +90,18 @@ export class CoreOptionsTag extends Context.Service /** * What the roots must provide beneath `CoreLive`: the stores, the options, - * and the two always-present collaborators the desktop builds elsewhere - * (`MemoryMetaLive`; `WorkspaceMcpOverrides` from `CrossCuttingLive`). CLI - * roots supply the defaults (`MemoryMetaLive`, `WorkspaceMcpOverridesDefaultLive`). + * the runtime's `EffectRunner` (the base seam in both roots; StreamManager's + * clock-driven fibers run through it), and the two always-present + * collaborators the desktop builds elsewhere (`MemoryMetaLive`; + * `WorkspaceMcpOverrides` from `CrossCuttingLive`). CLI roots supply the + * defaults (`MemoryMetaLive`, `WorkspaceMcpOverridesDefaultLive`). */ -export type CoreInputTags = StoreTags | CoreOptionsTag | MemoryMeta | WorkspaceMcpOverrides; +export type CoreInputTags = + | StoreTags + | CoreOptionsTag + | EffectRunnerTag + | MemoryMeta + | WorkspaceMcpOverrides; /** Memory metadata sidecar; scope root derives from the xum home (`config.rootDir`). */ export const MemoryMetaLive: Layer.Layer = Layer.effect( @@ -220,8 +227,13 @@ export const StreamManagerLive = Layer.effect( StreamManagerTag, Effect.gen(function* () { const providerService = yield* Provider; - return new StreamManager(yield* History, yield* SessionUsage, () => - providerService.getConfig() + return new StreamManager( + yield* History, + yield* SessionUsage, + () => providerService.getConfig(), + // Default event sink: AIService installs itself as the sink (S3). + undefined, + yield* EffectRunnerTag ); }) ); diff --git a/src/node/services/di/layers/desktop.ts b/src/node/services/di/layers/desktop.ts index 8fcb187eae..e4537cb50a 100644 --- a/src/node/services/di/layers/desktop.ts +++ b/src/node/services/di/layers/desktop.ts @@ -1,30 +1,179 @@ import * as path from "path"; import { Context, Effect, Layer } from "effect"; -import { AnalyticsService } from "@/node/services/analytics/analyticsService"; +import { DEFAULT_CODER_ARCHIVE_BEHAVIOR } from "@/common/config/coderArchiveBehavior"; +import { DEFAULT_WORKTREE_ARCHIVE_BEHAVIOR } from "@/common/config/worktreeArchiveBehavior"; +import { EXPERIMENT_IDS } from "@/common/constants/experiments"; +import type { + ErrorEvent, + ReasoningDeltaEvent, + StreamAbortEvent, + StreamDeltaEvent, + StreamEndEvent, + StreamStartEvent, + ToolCallDeltaEvent, + ToolCallEndEvent, + ToolCallStartEvent, +} from "@/common/types/stream"; +import { + createCoderArchiveHook, + createCoderUnarchiveHook, +} from "@/node/runtime/coderLifecycleHooks"; +import { setGlobalCoderService } from "@/node/runtime/runtimeFactory"; +import { + createRuntimeForWorkspace, + resolveWorkspaceExecutionPath, +} from "@/node/runtime/runtimeHelpers"; +import { setSshPromptService as setSSH2SshPromptService } from "@/node/runtime/SSH2ConnectionPool"; +import { setSshPromptService } from "@/node/runtime/sshConnectionPool"; +import { createWorktreeArchiveHook } from "@/node/runtime/worktreeLifecycleHooks"; +import { AgentPluginInstallService } from "@/node/services/agentPlugins/installService"; +import { AgentStatusService } from "@/node/services/agentStatusService"; +import { + AnalyticsService, + type IngestWorkspaceMeta, +} from "@/node/services/analytics/analyticsService"; +import { createBackupGitRepo, createBackupPayloadStore } from "@/node/services/backup/adapters"; +import { BackupService } from "@/node/services/backup/backupService"; +import { AgentBrowserSessionDiscoveryService } from "@/node/services/browser/AgentBrowserSessionDiscoveryService"; +import { BrowserBridgeServer } from "@/node/services/browser/BrowserBridgeServer"; +import { BrowserBridgeTokenManager } from "@/node/services/browser/BrowserBridgeTokenManager"; +import { BrowserControlService } from "@/node/services/browser/BrowserControlService"; +import { BrowserSessionStateHub } from "@/node/services/browser/BrowserSessionStateHub"; +import { CoderOauthService } from "@/node/services/coderOauthService"; +import { coderService as coderServiceSingleton } from "@/node/services/coderService"; +import { CodexOauthService } from "@/node/services/codexOauthService"; +import { CopilotOauthService } from "@/node/services/copilotOauthService"; +import { DesktopBridgeServer } from "@/node/services/desktop/DesktopBridgeServer"; +import { DesktopSessionManager } from "@/node/services/desktop/DesktopSessionManager"; +import { DesktopTokenManager } from "@/node/services/desktop/DesktopTokenManager"; import { DevToolsService } from "@/node/services/devToolsService"; +import { EffectRunnerTag } from "@/node/services/di/effectRunner"; import { + AgentBrowserSessionDiscovery, + AgentPluginInstall, + AgentStatus, + AI, Analytics, + BackgroundProcessManagerTag, + Backup, + BrowserBridgeServerTag, + BrowserBridgeTokenManagerTag, + BrowserControl, + BrowserSessionStateHubTag, + Coder, + CoderOauth, + CodexOauth, ConfigTag, + CopilotOauth, + DesktopBridgeServerTag, + DesktopSessionManagerTag, + DesktopTokenManagerTag, DevTools, + Editor, Experiments, + ExtensionMetadata, + FileLeaseManagerTag, + Heartbeat, + History, + IdleCompaction, + IdleDispatcherTag, + Instructions, + MCPConfig, + McpOauth, + MCPServerManagerTag, + Memory, + MemoryConsolidation, + MemoryMeta, + MenuEvent, + MuxGatewayOauth, + MuxGovernorOauth, Policy, + Project, + Provider, + ProvidersConfigStoreTag, + PTY, + QuickJSRuntimeFactoryTag, + Refine, + SecretsStoreTag, + Server, + ServerAuth, SessionTiming, + SessionUsage, + SshPrompt, + Task, Telemetry, + Terminal, + Timeline, + Tokenizer, + TurnRequestBuilderBindingsTag, + Update, + Voice, + WindowTag, + Workspace, + WorkspaceGoal, + WorkspaceLifecycleHooksTag, WorkspaceMcpOverrides, + WorktreeArchiveSnapshot, + type BrowserTags, + type CoreTags, type CrossCuttingTags, + type DesktopBridgeTags, + type DesktopTags, + type MiscDesktopTags, + type OauthTags, + type StoreTags, + type TerminalEditorTags, + type WorkerTags, } from "@/node/services/di/tags"; +import { EditorService } from "@/node/services/editorService"; import { ExperimentsService } from "@/node/services/experimentsService"; +import { HeartbeatService } from "@/node/services/heartbeatService"; +import { IdleCompactionService } from "@/node/services/idleCompactionService"; +import { InstructionsService } from "@/node/services/instructionsService"; +import { McpOauthService } from "@/node/services/mcpOauthService"; +import { MenuEventService } from "@/node/services/menuEventService"; +import { MuxGatewayOauthService } from "@/node/services/muxGatewayOauthService"; +import { MuxGovernorOauthService } from "@/node/services/muxGovernorOauthService"; import { PolicyService } from "@/node/services/policyService"; +import { ProjectService } from "@/node/services/projectService"; +import { QuickJSRuntimeFactory } from "@/node/services/ptc/quickjsRuntime"; +import { PTYService } from "@/node/services/ptyService"; +import { RefineService } from "@/node/services/refinement/refineService"; +import { ServerAuthService } from "@/node/services/serverAuthService"; +import { ServerService } from "@/node/services/serverService"; import { SessionTimingService } from "@/node/services/sessionTimingService"; +import { SshPromptService } from "@/node/services/sshPromptService"; import { TelemetryService } from "@/node/services/telemetryService"; +import { TerminalService } from "@/node/services/terminalService"; +import { TimelineService } from "@/node/services/timelineService"; +import { TokenizerService } from "@/node/services/tokenizerService"; +import { UpdateService } from "@/node/services/updateService"; +import { VoiceService } from "@/node/services/voiceService"; +import { WindowService } from "@/node/services/windowService"; +import { WorkspaceLifecycleHooks } from "@/node/services/workspaceLifecycleHooks"; import { WorkspaceMcpOverridesService } from "@/node/services/workspaceMcpOverridesService"; +import { WorktreeArchiveSnapshotService } from "@/node/services/worktreeArchiveSnapshotService"; import { CoreOptionsTag } from "./core"; /** - * Desktop/server-only layers (`ServiceContainer` roots). Only the services the - * core graph's options derive from live here so far; the remaining desktop - * constructions stay in the `ServiceContainer` constructor until they get - * their own group layers. + * Desktop/server-only layers (`ServiceContainer` roots; Effect migration + * Phase 11). + * + * `CrossCuttingLive` and `CoreOptionsFromDesktopLive` sit beneath the core + * graph (its options derive from them). The remaining desktop services are + * built above the core graph by six **group layers** — one `Layer.effectContext` + * per group, constructing several services in the order the `ServiceContainer` + * constructor used — rather than one layer per service: cold build cost grows + * with the number of layers, and the desktop tail has no per-service swap + * needs. Groups that depend on each other are staged with + * `Layer.provideMerge`; only true siblings share a `Layer.mergeAll` + * (siblings may build in any order, so nothing relies on sibling order). + * Bodies are synchronous and register no finalizers (DI contract in + * `../appRuntime.ts`); teardown stays explicit in `ServiceContainer.dispose()`. + * + * The former constructor's post-construction wiring — setters, event + * listeners, global registrations — is replayed, statement for statement in + * its original order, by `DesktopWiringLive` once every service exists. */ /** @@ -83,3 +232,598 @@ export const CoreOptionsFromDesktopLive: Layer.Layer< }; }) ); + +// --------------------------------------------------------------------------- +// Desktop group layers. Requirements (the `R` annotations) are the DAG: the +// four base groups need only stores, cross-cutting and core services; OAuth +// additionally needs the WindowService (Misc); the workers need the +// WindowService (Misc) and the TokenizerService (TerminalEditor). +// --------------------------------------------------------------------------- + +/** Browser automation bridge: token manager → discovery → control → state hub → server. */ +export const BrowserLive: Layer.Layer = Layer.effectContext( + Effect.map(ConfigTag, (config) => { + const browserBridgeTokenManager = new BrowserBridgeTokenManager(); + const browserSessionDiscoveryService = new AgentBrowserSessionDiscoveryService({ + resolveWorkspaceCandidatePathsFn: async (workspaceId: string) => { + const allWorkspaceMetadata = await config.getAllWorkspaceMetadata(); + const workspaceMetadata = + allWorkspaceMetadata.find((candidate) => candidate.id === workspaceId) ?? null; + if (workspaceMetadata == null) { + return []; + } + + const runtime = createRuntimeForWorkspace(workspaceMetadata); + const workspacePath = resolveWorkspaceExecutionPath(workspaceMetadata, runtime); + return [workspaceMetadata.projectPath, workspacePath].filter( + (candidatePath): candidatePath is string => candidatePath.trim().length > 0 + ); + }, + }); + const browserControlService = new BrowserControlService({ + browserSessionDiscoveryService, + resolveSessionEnvFn: () => Promise.resolve(process.env), + }); + const browserSessionStateHub = new BrowserSessionStateHub({ + browserControlService, + }); + const browserBridgeServer = new BrowserBridgeServer({ + browserSessionDiscoveryService, + browserBridgeTokenManager, + browserSessionStateHub, + }); + return Context.empty().pipe( + Context.add(BrowserBridgeTokenManagerTag, browserBridgeTokenManager), + Context.add(AgentBrowserSessionDiscovery, browserSessionDiscoveryService), + Context.add(BrowserControl, browserControlService), + Context.add(BrowserSessionStateHubTag, browserSessionStateHub), + Context.add(BrowserBridgeServerTag, browserBridgeServer) + ); + }) +); + +/** Desktop companion bridge: session manager → token manager → server. */ +export const DesktopBridgeLive: Layer.Layer< + DesktopBridgeTags, + never, + ConfigTag | Experiments | Workspace +> = Layer.effectContext( + Effect.gen(function* () { + const desktopSessionManager = new DesktopSessionManager({ + config: yield* ConfigTag, + experimentsService: yield* Experiments, + workspaceService: yield* Workspace, + }); + const desktopTokenManager = new DesktopTokenManager(); + const desktopBridgeServer = new DesktopBridgeServer({ + desktopSessionManager, + desktopTokenManager, + }); + return Context.empty().pipe( + Context.add(DesktopSessionManagerTag, desktopSessionManager), + Context.add(DesktopTokenManagerTag, desktopTokenManager), + Context.add(DesktopBridgeServerTag, desktopBridgeServer) + ); + }) +); + +/** Terminal (PTY → terminal), editor, and token budgeting (tokenizer → instructions). */ +export const TerminalEditorLive: Layer.Layer< + TerminalEditorTags, + never, + ConfigTag | SecretsStoreTag | Workspace | SessionUsage | AI | Provider +> = Layer.effectContext( + Effect.gen(function* () { + const config = yield* ConfigTag; + const aiService = yield* AI; + // Terminal services - PTYService is cross-platform + const ptyService = new PTYService(); + const terminalService = new TerminalService(config, ptyService, yield* SecretsStoreTag); + // Editor service for opening workspaces in code editors + const editorService = new EditorService(config, yield* Workspace); + const tokenizerService = new TokenizerService(yield* SessionUsage, aiService, yield* Provider); + const instructionsService = new InstructionsService(config, aiService, tokenizerService); + return Context.empty().pipe( + Context.add(PTY, ptyService), + Context.add(Terminal, terminalService), + Context.add(Editor, editorService), + Context.add(Tokenizer, tokenizerService), + Context.add(Instructions, instructionsService) + ); + }) +); + +/** + * The remaining desktop services: leaves over the stores/core/cross-cutting + * graph, plus `ProjectService` (needs `SshPromptService`, built first here). + */ +export const MiscDesktopLive: Layer.Layer< + MiscDesktopTags, + never, + | ConfigTag + | SecretsStoreTag + | ProvidersConfigStoreTag + | Experiments + | Policy + | Provider + | MCPServerManagerTag + | WorkspaceMcpOverrides +> = Layer.effectContext( + Effect.gen(function* () { + const config = yield* ConfigTag; + const experimentsService = yield* Experiments; + const policyService = yield* Policy; + const providerService = yield* Provider; + const providersConfigStore = yield* ProvidersConfigStoreTag; + const workflowRuntimeFactory = new QuickJSRuntimeFactory(); + const sshPromptService = new SshPromptService(); + const windowService = new WindowService(); + const backupService = new BackupService(config, { + gitRepo: createBackupGitRepo({ + cacheRoot: path.join(config.rootDir, "backup-cache"), + }), + payload: createBackupPayloadStore({ config }), + }); + // Managed Agent Plugin installer (agent-plugins experiment). Gated on the + // backend ExperimentsService exactly like the plugin MCP provider; the + // MCP manager dependency lets update/uninstall recycle running plugin + // servers whose content changed behind an unchanged command line. + const agentPluginInstallService = new AgentPluginInstallService(config, { + isEnabled: () => experimentsService.isExperimentEnabled(EXPERIMENT_IDS.AGENT_PLUGINS), + mcpServerManager: yield* MCPServerManagerTag, + workspaceMcpOverridesService: yield* WorkspaceMcpOverrides, + }); + const projectService = new ProjectService(config, sshPromptService, yield* SecretsStoreTag); + const updateService = new UpdateService(config); + const serverService = new ServerService(); + const menuEventService = new MenuEventService(); + const voiceService = new VoiceService( + config, + providerService, + policyService, + providersConfigStore + ); + const serverAuthService = new ServerAuthService(config); + const workspaceLifecycleHooks = new WorkspaceLifecycleHooks(); + const worktreeArchiveSnapshotService = new WorktreeArchiveSnapshotService(config); + return Context.empty().pipe( + Context.add(QuickJSRuntimeFactoryTag, workflowRuntimeFactory), + Context.add(SshPrompt, sshPromptService), + Context.add(WindowTag, windowService), + Context.add(Backup, backupService), + Context.add(AgentPluginInstall, agentPluginInstallService), + Context.add(Project, projectService), + Context.add(Update, updateService), + Context.add(Server, serverService), + Context.add(MenuEvent, menuEventService), + Context.add(Voice, voiceService), + Context.add(Coder, coderServiceSingleton), + Context.add(ServerAuth, serverAuthService), + Context.add(WorkspaceLifecycleHooksTag, workspaceLifecycleHooks), + Context.add(WorktreeArchiveSnapshot, worktreeArchiveSnapshotService) + ); + }) +); + +/** OAuth flows; every one hands off to the browser through the WindowService (Misc). */ +export const OauthLive: Layer.Layer< + OauthTags, + never, + | ConfigTag + | ProvidersConfigStoreTag + | FileLeaseManagerTag + | MCPConfig + | Provider + | Policy + | Telemetry + | WindowTag +> = Layer.effectContext( + Effect.gen(function* () { + const config = yield* ConfigTag; + const windowService = yield* WindowTag; + const providersConfigStore = yield* ProvidersConfigStoreTag; + const providerService = yield* Provider; + const policyService = yield* Policy; + const mcpOauthService = new McpOauthService( + config, + yield* MCPConfig, + windowService, + yield* Telemetry + ); + const muxGatewayOauthService = new MuxGatewayOauthService( + providersConfigStore, + providerService, + windowService + ); + const muxGovernorOauthService = new MuxGovernorOauthService( + config, + windowService, + policyService + ); + const codexOauthService = new CodexOauthService( + providersConfigStore, + providerService, + windowService + ); + const coderOauthService = new CoderOauthService( + providersConfigStore, + yield* FileLeaseManagerTag, + providerService, + windowService, + // Policy-aware: an enforced forcedBaseUrl overrides the deployment URL + // for logins, refreshes, and issuer checks. + policyService + ); + const copilotOauthService = new CopilotOauthService(providerService, windowService); + return Context.empty().pipe( + Context.add(McpOauth, mcpOauthService), + Context.add(MuxGatewayOauth, muxGatewayOauthService), + Context.add(MuxGovernorOauth, muxGovernorOauthService), + Context.add(CodexOauth, codexOauthService), + Context.add(CoderOauth, coderOauthService), + Context.add(CopilotOauth, copilotOauthService) + ); + }) +); + +/** + * Clock-driven workers (through the runtime's `EffectRunner`) and the + * timeline/refine pair: idle compaction → heartbeat → timeline → refine → + * agent status. `start()`/`stop()` stay with `ServiceContainer`. + */ +export const WorkersLive: Layer.Layer< + WorkerTags, + never, + | ConfigTag + | EffectRunnerTag + | Experiments + | History + | ExtensionMetadata + | Workspace + | Task + | IdleDispatcherTag + | Memory + | MemoryMeta + | AI + | SessionUsage + | Tokenizer + | WindowTag +> = Layer.effectContext( + Effect.gen(function* () { + const config = yield* ConfigTag; + // Clock-driven workers run their lifecycle fibers through the runtime's + // context-bound runner (unsupervised; see di/effectRunner.ts). + const effectRunner = yield* EffectRunnerTag; + const experimentsService = yield* Experiments; + const historyService = yield* History; + const extensionMetadata = yield* ExtensionMetadata; + const workspaceService = yield* Workspace; + const aiService = yield* AI; + const sessionUsageService = yield* SessionUsage; + // Idle compaction service - auto-compacts workspaces after configured idle period + const idleCompactionService = new IdleCompactionService( + config, + historyService, + extensionMetadata, + (workspaceId) => workspaceService.executeIdleCompaction(workspaceId), + effectRunner + ); + // IdleDispatcher + goal continuation bridge are owned by the core graph + // so the wiring works for `xum run` too. Share the same dispatcher with + // HeartbeatService — its priority ordering ensures an active goal + // suppresses background heartbeats. + const heartbeatService = new HeartbeatService( + config, + extensionMetadata, + workspaceService, + yield* Task, + yield* IdleDispatcherTag, + effectRunner + ); + const timelineService = new TimelineService(config, historyService, experimentsService); + // /refine trajectory distillation (RLM r11). Chat emission routes through + // WorkspaceService so a live session renders the appended summary row + // immediately (the row itself is already durable in chat.jsonl). + const refineService = new RefineService( + config, + yield* Memory, + yield* MemoryMeta, + historyService, + aiService, + experimentsService, + { + timelineService, + sessionUsageService, + emitChatMessage: (workspaceId, message) => + workspaceService.emitChatEvent(workspaceId, { ...message, type: "message" }), + // r40: refine row publication and apply mutations must not interleave + // with a concurrent turn's PREPARING snapshot or split its + // user/assistant pair — hold the session's turn-admission block while + // they land, failing closed when a turn is active. + acquireTurnExclusion: (workspaceId) => + workspaceService.acquireIdleTurnExclusion(workspaceId), + } + ); + // AgentStatusService depends on tokenizer + window focus state; instantiate + // after both are constructed so the small-model status loop can run with + // accurate token budgeting and focus-aware cadence. + const agentStatusService = new AgentStatusService( + config, + historyService, + yield* Tokenizer, + extensionMetadata, + workspaceService, + yield* WindowTag, + aiService, + // Status generation spends tokens outside StreamManager; give it a cost + // telemetry sink so that spend shows up in per-workspace usage, and an + // ingest trigger so the headless-usage sidecar reaches dashboard totals + // even when the workspace has no further stream activity. + { + sessionUsageService, + requestAnalyticsIngest: (workspaceId) => { + workspaceService.emit("analyticsIngest", { workspaceId }); + }, + } + ); + return Context.empty().pipe( + Context.add(IdleCompaction, idleCompactionService), + Context.add(Heartbeat, heartbeatService), + Context.add(Timeline, timelineService), + Context.add(Refine, refineService), + Context.add(AgentStatus, agentStatusService) + ); + }) +); + +// --------------------------------------------------------------------------- +// Wiring — the former `ServiceContainer` constructor's post-construction +// statements (setters, event listeners, global registrations), in their +// original order, once every desktop service exists. Runs after `CoreLive`'s +// wiring, so the analytics listeners below keep their position after the core +// listeners. Synchronous statements only: no finalizers, no forks (I5). +// --------------------------------------------------------------------------- + +export const DesktopWiringLive: Layer.Layer< + never, + never, + ConfigTag | CrossCuttingTags | CoreTags | DesktopTags +> = Layer.effectDiscard( + Effect.gen(function* () { + const config = yield* ConfigTag; + const analyticsService = yield* Analytics; + const sessionTimingService = yield* SessionTiming; + const turnRequestBuilderBindings = yield* TurnRequestBuilderBindingsTag; + const aiService = yield* AI; + const workspaceService = yield* Workspace; + const taskService = yield* Task; + const workspaceGoalService = yield* WorkspaceGoal; + const mcpServerManager = yield* MCPServerManagerTag; + const memoryConsolidationService = yield* MemoryConsolidation; + const backgroundProcessManager = yield* BackgroundProcessManagerTag; + const coderService = yield* Coder; + const backupService = yield* Backup; + const memoryService = yield* Memory; + const projectService = yield* Project; + const sshPromptService = yield* SshPrompt; + const desktopSessionManager = yield* DesktopSessionManagerTag; + const idleCompactionService = yield* IdleCompaction; + const heartbeatService = yield* Heartbeat; + const timelineService = yield* Timeline; + const refineService = yield* Refine; + const mcpOauthService = yield* McpOauth; + const codexOauthService = yield* CodexOauth; + const coderOauthService = yield* CoderOauth; + const terminalService = yield* Terminal; + const workspaceLifecycleHooks = yield* WorkspaceLifecycleHooksTag; + const worktreeArchiveSnapshotService = yield* WorktreeArchiveSnapshot; + + turnRequestBuilderBindings.analyticsService = analyticsService; + + projectService.setWorkspaceService(workspaceService); + projectService.setWorkspaceMetadataRefresher(workspaceService); + projectService.setMcpServerManager(mcpServerManager); + // Backup restores register approved project imports through the same create() path the + // UI uses; setter injection because BackupService is constructed before ProjectService. + backupService.setProjectService(projectService); + // Restored project memory is written directly, so the service announces it through + // MemoryService for the memory browser's change subscription. + backupService.setMemoryNotifier(memoryService); + turnRequestBuilderBindings.desktopSessionManager = desktopSessionManager; + + // Forward terminal idle-compaction outcomes so the loop stops re-attempting a + // persistently failing workspace (immediately on model_not_found, otherwise after + // two consecutive failures). + workspaceService.setIdleCompactionOutcomeListener((workspaceId, outcome) => + idleCompactionService.recordOutcome(workspaceId, outcome) + ); + + // Removal must be able to abort + drain a running /refine pass before it + // deletes the session directory (post-construction wiring: RefineService + // is built after WorkspaceService). + workspaceService.setRefinePassCanceller(refineService); + workspaceService.setTimelineRecorder(timelineService); + taskService.setTimelineRecorder(timelineService); + heartbeatService.setTimelineRecorder(timelineService); + workspaceGoalService.setTimelineRecorder(timelineService); + turnRequestBuilderBindings.timelineService = timelineService; + timelineService.subscribeToWorkspace(workspaceService); + + mcpServerManager.setMcpOauthService(mcpOauthService); + turnRequestBuilderBindings.codexOauthService = codexOauthService; + turnRequestBuilderBindings.coderOauthService = coderOauthService; + + // Wire terminal service to workspace service for cleanup on removal + workspaceService.setTerminalService(terminalService); + workspaceService.setDesktopSessionManager(desktopSessionManager); + // Plugin-override pruning is wired inside the core graph (shared with + // headless CLI registration), using the WorkspaceMcpOverridesService. + + workspaceService.setWorktreeArchiveSnapshotService(worktreeArchiveSnapshotService); + const getArchiveBehavior = () => + config.loadConfigOrDefault().coderWorkspaceArchiveBehavior ?? DEFAULT_CODER_ARCHIVE_BEHAVIOR; + workspaceLifecycleHooks.registerBeforeArchive( + createCoderArchiveHook({ + coderService, + getArchiveBehavior, + // Model-driven archives probe the remote spawn-record layout before stopping a + // running Coder workspace: detached jobs surviving an unclean Xum exit live only in + // those records, which the host-local crash-orphan scans cannot see. + hasUnsettledRemoteBackgroundJobs: async (workspaceMetadata) => { + const runtime = createRuntimeForWorkspace(workspaceMetadata); + return await backgroundProcessManager.hasUnsettledRemoteSpawnRecords( + runtime, + workspaceMetadata.id + ); + }, + }) + ); + workspaceLifecycleHooks.registerAfterUnarchive( + createCoderUnarchiveHook({ + coderService, + getArchiveBehavior, + }) + ); + const getWorktreeArchiveBehavior = () => + config.loadConfigOrDefault().worktreeArchiveBehavior ?? DEFAULT_WORKTREE_ARCHIVE_BEHAVIOR; + workspaceLifecycleHooks.registerAfterArchive( + createWorktreeArchiveHook({ getWorktreeArchiveBehavior }) + ); + workspaceService.setWorkspaceLifecycleHooks(workspaceLifecycleHooks); + + // Register globally so all createRuntime calls can create CoderSSHRuntime + setGlobalCoderService(coderService); + setSshPromptService(sshPromptService); + setSSH2SshPromptService(sshPromptService); + + // Backend timing stats. + aiService.on("stream-start", (data: StreamStartEvent) => + sessionTimingService.handleStreamStart(data) + ); + aiService.on("stream-delta", (data: StreamDeltaEvent) => + sessionTimingService.handleStreamDelta(data) + ); + aiService.on("reasoning-delta", (data: ReasoningDeltaEvent) => + sessionTimingService.handleReasoningDelta(data) + ); + aiService.on("tool-call-start", (data: ToolCallStartEvent) => + sessionTimingService.handleToolCallStart(data) + ); + aiService.on("tool-call-delta", (data: ToolCallDeltaEvent) => + sessionTimingService.handleToolCallDelta(data) + ); + aiService.on("tool-call-end", (data: ToolCallEndEvent) => + sessionTimingService.handleToolCallEnd(data) + ); + // Newly created sub-agent workspaces are ingested here before a full rebuild, + // so keep workspaceName + parentWorkspaceId to avoid NULL analytics attribution. + // Multi-project workspaces stay stored under _multi in config, but analytics should + // still attribute spend to the workspace's first real project path. + const ingestWorkspaceAnalytics = (workspaceId: string) => { + const workspaceLookup = config.findWorkspace(workspaceId); + const sessionDir = path.join(config.sessionsDir, workspaceId); + const analyticsProjectPath = + workspaceLookup?.attributionProjectPath ?? workspaceLookup?.projectPath; + analyticsService.ingestWorkspace(workspaceId, sessionDir, { + projectPath: analyticsProjectPath, + projectName: analyticsProjectPath ? path.basename(analyticsProjectPath) : undefined, + workspaceName: workspaceLookup?.workspaceName, + parentWorkspaceId: workspaceLookup?.parentWorkspaceId, + }); + }; + aiService.on("stream-end", (data: StreamEndEvent) => { + sessionTimingService.handleStreamEnd(data); + ingestWorkspaceAnalytics(data.workspaceId); + }); + // Billable usage persisted outside StreamManager stream-end requests its + // own incremental ingest pass. + workspaceService.on("analyticsIngest", (event) => { + ingestWorkspaceAnalytics(event.workspaceId); + }); + // Memory consolidation/harvest spend rides the headless-usage sidecar + // without any chat activity; ingest promptly so background sweeps reach + // dashboard totals instead of stranding until an unrelated stream-end + // or app restart. + memoryConsolidationService.on("analyticsIngest", (event: { workspaceId: string }) => { + ingestWorkspaceAnalytics(event.workspaceId); + }); + // WorkspaceService emits metadata:null after successful remove(). + // Clear analytics rows immediately so deleted workspaces disappear from stats + // without waiting for a future ingest pass. + workspaceService.on("metadata", (event) => { + if (event.metadata !== null) { + return; + } + + // Removed sub-agent children archive their transcript into the parent's + // session dir before this event fires. Re-ingest the parent (chained after + // the clear) so the child's spend is restored from the archive instead of + // vanishing from analytics until the parent's next stream-end. + let reingestAfterClear: + | { workspaceId: string; sessionDir: string; meta: IngestWorkspaceMeta } + | undefined; + const parentWorkspaceId = event.removedParentWorkspaceId; + if (parentWorkspaceId) { + const parentLookup = config.findWorkspace(parentWorkspaceId); + const parentProjectPath = parentLookup?.attributionProjectPath ?? parentLookup?.projectPath; + reingestAfterClear = { + workspaceId: parentWorkspaceId, + sessionDir: path.join(config.sessionsDir, parentWorkspaceId), + meta: { + projectPath: parentProjectPath, + projectName: parentProjectPath ? path.basename(parentProjectPath) : undefined, + workspaceName: parentLookup?.workspaceName, + parentWorkspaceId: parentLookup?.parentWorkspaceId, + }, + }; + } + + analyticsService.clearWorkspace(event.workspaceId, { reingestAfterClear }); + }); + + aiService.on("stream-abort", (data: StreamAbortEvent) => { + sessionTimingService.handleStreamAbort(data); + // Aborted turns persist their spend before this event fires (same async + // chain): normal aborts commit the usage-stamped partial to chat.jsonl + // (or the headless sidecar for non-commit-worthy partials); abandoned + // aborts (edit/discard) write only the sidecar. Ingest both, or the + // interrupted turn's spend stays out of dashboards until the next + // stream-end. + ingestWorkspaceAnalytics(data.workspaceId); + }); + // Errored turns whose partial would be dropped at commit time route their + // usage to the headless sidecar (persistStreamError). The sidecar write + // precedes this event in the same async chain, so ingest here keeps the + // dashboard current instead of waiting for the next stream or restart. + aiService.on("error", (data: ErrorEvent) => { + ingestWorkspaceAnalytics(data.workspaceId); + }); + }) +); + +// --------------------------------------------------------------------------- +// Staged composition (the group DAG). Base groups are true siblings: none of +// their constructors takes another desktop service (audited per constructor; +// only `CoderOauthService` subscribes to a collaborator — the core +// ProviderService — in its constructor, and the two token managers start their +// own unref'd cleanup intervals). OAuth and the workers need base services. +// --------------------------------------------------------------------------- + +const DesktopBase = Layer.mergeAll( + MiscDesktopLive, + BrowserLive, + DesktopBridgeLive, + TerminalEditorLive +); +const DesktopUpper = Layer.mergeAll(OauthLive, WorkersLive).pipe(Layer.provideMerge(DesktopBase)); + +/** + * Every desktop-only service, wired. Built above the core graph: it needs the + * stores, the runtime's `EffectRunner`, the cross-cutting services and every + * core tag (`AppLive` in ./app.ts provides them beneath). + */ +export const DesktopLive: Layer.Layer< + DesktopTags, + never, + StoreTags | EffectRunnerTag | CrossCuttingTags | CoreTags +> = DesktopWiringLive.pipe(Layer.provideMerge(DesktopUpper)); diff --git a/src/node/services/di/tags.ts b/src/node/services/di/tags.ts index dc3eac3d37..761dbe6dcb 100644 --- a/src/node/services/di/tags.ts +++ b/src/node/services/di/tags.ts @@ -19,33 +19,71 @@ import type { SecretsStore, WorkspaceSessionLocator, } from "@/node/config"; +import type { AgentPluginInstallService } from "@/node/services/agentPlugins/installService"; +import type { AgentStatusService } from "@/node/services/agentStatusService"; import type { AIService } from "@/node/services/aiService"; import type { AnalyticsService } from "@/node/services/analytics/analyticsService"; import type { BackgroundProcessManager } from "@/node/services/backgroundProcessManager"; +import type { BackupService } from "@/node/services/backup/backupService"; +import type { AgentBrowserSessionDiscoveryService } from "@/node/services/browser/AgentBrowserSessionDiscoveryService"; +import type { BrowserBridgeServer } from "@/node/services/browser/BrowserBridgeServer"; +import type { BrowserBridgeTokenManager } from "@/node/services/browser/BrowserBridgeTokenManager"; +import type { BrowserControlService } from "@/node/services/browser/BrowserControlService"; +import type { BrowserSessionStateHub } from "@/node/services/browser/BrowserSessionStateHub"; +import type { CoderOauthService } from "@/node/services/coderOauthService"; +import type { CoderService } from "@/node/services/coderService"; +import type { CodexOauthService } from "@/node/services/codexOauthService"; +import type { CopilotOauthService } from "@/node/services/copilotOauthService"; +import type { DesktopBridgeServer } from "@/node/services/desktop/DesktopBridgeServer"; +import type { DesktopSessionManager } from "@/node/services/desktop/DesktopSessionManager"; +import type { DesktopTokenManager } from "@/node/services/desktop/DesktopTokenManager"; import type { DevToolsService } from "@/node/services/devToolsService"; +import type { EditorService } from "@/node/services/editorService"; import type { ExperimentsService } from "@/node/services/experimentsService"; import type { ExtensionMetadataService } from "@/node/services/ExtensionMetadataService"; +import type { HeartbeatService } from "@/node/services/heartbeatService"; import type { HistoryService } from "@/node/services/historyService"; +import type { IdleCompactionService } from "@/node/services/idleCompactionService"; import type { IdleDispatcher } from "@/node/services/idleDispatcher"; import type { InitStateManager } from "@/node/services/initStateManager"; +import type { InstructionsService } from "@/node/services/instructionsService"; import type { MCPConfigService } from "@/node/services/mcpConfigService"; +import type { McpOauthService } from "@/node/services/mcpOauthService"; import type { MCPServerManager } from "@/node/services/mcpServerManager"; import type { MemoryConsolidationService } from "@/node/services/memoryConsolidationService"; import type { MemoryMetaService } from "@/node/services/memoryMeta"; import type { MemoryService } from "@/node/services/memoryService"; +import type { MenuEventService } from "@/node/services/menuEventService"; +import type { MuxGatewayOauthService } from "@/node/services/muxGatewayOauthService"; +import type { MuxGovernorOauthService } from "@/node/services/muxGovernorOauthService"; import type { PolicyService } from "@/node/services/policyService"; +import type { ProjectService } from "@/node/services/projectService"; import type { ProviderService } from "@/node/services/providerService"; +import type { QuickJSRuntimeFactory } from "@/node/services/ptc/quickjsRuntime"; +import type { PTYService } from "@/node/services/ptyService"; +import type { RefineService } from "@/node/services/refinement/refineService"; +import type { ServerAuthService } from "@/node/services/serverAuthService"; +import type { ServerService } from "@/node/services/serverService"; import type { SessionTimingService } from "@/node/services/sessionTimingService"; import type { SessionUsageService } from "@/node/services/sessionUsageService"; +import type { SshPromptService } from "@/node/services/sshPromptService"; import type { StreamManager } from "@/node/services/streamManager"; import type { TaskService } from "@/node/services/taskService"; import type { TelemetryService } from "@/node/services/telemetryService"; import type { TerminalAttentionStore } from "@/node/services/terminalAttentionStore"; +import type { TerminalService } from "@/node/services/terminalService"; +import type { TimelineService } from "@/node/services/timelineService"; +import type { TokenizerService } from "@/node/services/tokenizerService"; import type { TurnRequestBuilderBindings } from "@/node/services/turnRequestBuilder"; +import type { UpdateService } from "@/node/services/updateService"; +import type { VoiceService } from "@/node/services/voiceService"; +import type { WindowService } from "@/node/services/windowService"; import type { WorkspaceGoalService } from "@/node/services/workspaceGoalService"; +import type { WorkspaceLifecycleHooks } from "@/node/services/workspaceLifecycleHooks"; import type { WorkspaceMcpOverridesService } from "@/node/services/workspaceMcpOverridesService"; import type { WorkspaceService } from "@/node/services/workspaceService"; import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager"; +import type { WorktreeArchiveSnapshotService } from "@/node/services/worktreeArchiveSnapshotService"; import type { AppFiberScopeTag } from "./appFiberScope"; import type { EffectRunnerTag } from "./effectRunner"; import type { CoreOptionsTag } from "./layers/core"; @@ -142,6 +180,107 @@ export class WorkspaceMcpOverrides extends Context.Service< WorkspaceMcpOverridesService >()("xum/WorkspaceMcpOverrides") {} +// Desktop/server-only services (`DesktopLive` group layers in ./layers/desktop.ts). +// Browser automation bridge. +export class BrowserBridgeTokenManagerTag extends Context.Service< + BrowserBridgeTokenManagerTag, + BrowserBridgeTokenManager +>()("xum/BrowserBridgeTokenManager") {} +export class AgentBrowserSessionDiscovery extends Context.Service< + AgentBrowserSessionDiscovery, + AgentBrowserSessionDiscoveryService +>()("xum/AgentBrowserSessionDiscovery") {} +export class BrowserControl extends Context.Service()( + "xum/BrowserControl" +) {} +export class BrowserSessionStateHubTag extends Context.Service< + BrowserSessionStateHubTag, + BrowserSessionStateHub +>()("xum/BrowserSessionStateHub") {} +export class BrowserBridgeServerTag extends Context.Service< + BrowserBridgeServerTag, + BrowserBridgeServer +>()("xum/BrowserBridgeServer") {} +// Desktop companion bridge. +export class DesktopSessionManagerTag extends Context.Service< + DesktopSessionManagerTag, + DesktopSessionManager +>()("xum/DesktopSessionManager") {} +export class DesktopTokenManagerTag extends Context.Service< + DesktopTokenManagerTag, + DesktopTokenManager +>()("xum/DesktopTokenManager") {} +export class DesktopBridgeServerTag extends Context.Service< + DesktopBridgeServerTag, + DesktopBridgeServer +>()("xum/DesktopBridgeServer") {} +// Terminal, editor and token budgeting. +export class PTY extends Context.Service()("xum/PTY") {} +export class Terminal extends Context.Service()("xum/Terminal") {} +export class Editor extends Context.Service()("xum/Editor") {} +export class Tokenizer extends Context.Service()("xum/Tokenizer") {} +export class Instructions extends Context.Service()( + "xum/Instructions" +) {} +// Leaves and the remaining desktop services. +/** `WindowTag`: the bare name would shadow the DOM `Window` global. */ +export class WindowTag extends Context.Service()("xum/Window") {} +export class SshPrompt extends Context.Service()("xum/SshPrompt") {} +export class QuickJSRuntimeFactoryTag extends Context.Service< + QuickJSRuntimeFactoryTag, + QuickJSRuntimeFactory +>()("xum/QuickJSRuntimeFactory") {} +export class Backup extends Context.Service()("xum/Backup") {} +export class AgentPluginInstall extends Context.Service< + AgentPluginInstall, + AgentPluginInstallService +>()("xum/AgentPluginInstall") {} +export class Project extends Context.Service()("xum/Project") {} +export class Update extends Context.Service()("xum/Update") {} +export class Server extends Context.Service()("xum/Server") {} +export class MenuEvent extends Context.Service()("xum/MenuEvent") {} +export class Voice extends Context.Service()("xum/Voice") {} +/** The module singleton `coderService`, provided under a tag like every other field. */ +export class Coder extends Context.Service()("xum/Coder") {} +export class ServerAuth extends Context.Service()( + "xum/ServerAuth" +) {} +export class WorkspaceLifecycleHooksTag extends Context.Service< + WorkspaceLifecycleHooksTag, + WorkspaceLifecycleHooks +>()("xum/WorkspaceLifecycleHooks") {} +export class WorktreeArchiveSnapshot extends Context.Service< + WorktreeArchiveSnapshot, + WorktreeArchiveSnapshotService +>()("xum/WorktreeArchiveSnapshot") {} +// OAuth flows (all need the WindowService for the browser hand-off). +export class McpOauth extends Context.Service()("xum/McpOauth") {} +export class MuxGatewayOauth extends Context.Service()( + "xum/MuxGatewayOauth" +) {} +export class MuxGovernorOauth extends Context.Service()( + "xum/MuxGovernorOauth" +) {} +export class CodexOauth extends Context.Service()( + "xum/CodexOauth" +) {} +export class CoderOauth extends Context.Service()( + "xum/CoderOauth" +) {} +export class CopilotOauth extends Context.Service()( + "xum/CopilotOauth" +) {} +// Clock-driven workers and the timeline/refine pair they record into. +export class IdleCompaction extends Context.Service()( + "xum/IdleCompaction" +) {} +export class Heartbeat extends Context.Service()("xum/Heartbeat") {} +export class Timeline extends Context.Service()("xum/Timeline") {} +export class Refine extends Context.Service()("xum/Refine") {} +export class AgentStatus extends Context.Service()( + "xum/AgentStatus" +) {} + /** The process's config stores (`ConfigStores`), one tag per store. */ export type StoreTags = | ConfigTag @@ -203,5 +342,48 @@ export type CrossCuttingTags = | DevTools | WorkspaceMcpOverrides; +/** The desktop-only services provided by the `DesktopLive` group layers, by group. */ +export type BrowserTags = + | BrowserBridgeTokenManagerTag + | AgentBrowserSessionDiscovery + | BrowserControl + | BrowserSessionStateHubTag + | BrowserBridgeServerTag; +export type DesktopBridgeTags = + | DesktopSessionManagerTag + | DesktopTokenManagerTag + | DesktopBridgeServerTag; +export type TerminalEditorTags = PTY | Terminal | Editor | Tokenizer | Instructions; +export type MiscDesktopTags = + | WindowTag + | SshPrompt + | QuickJSRuntimeFactoryTag + | Backup + | AgentPluginInstall + | Project + | Update + | Server + | MenuEvent + | Voice + | Coder + | ServerAuth + | WorkspaceLifecycleHooksTag + | WorktreeArchiveSnapshot; +export type OauthTags = + | McpOauth + | MuxGatewayOauth + | MuxGovernorOauth + | CodexOauth + | CoderOauth + | CopilotOauth; +export type WorkerTags = IdleCompaction | Heartbeat | Timeline | Refine | AgentStatus; +export type DesktopTags = + | BrowserTags + | DesktopBridgeTags + | TerminalEditorTags + | MiscDesktopTags + | OauthTags + | WorkerTags; + /** Every service the desktop/server app graph (`AppLive`) provides. */ -export type AppTags = CoreRootTags | CrossCuttingTags; +export type AppTags = CoreRootTags | CrossCuttingTags | DesktopTags; diff --git a/src/node/services/retryManager.ts b/src/node/services/retryManager.ts index f736e217c2..b667a0840f 100644 --- a/src/node/services/retryManager.ts +++ b/src/node/services/retryManager.ts @@ -67,9 +67,9 @@ export class RetryManager { private readonly onStatusChange: (event: RetryStatusEvent) => void, /** * Runs the retry fiber fork and its interrupt. The global runtime by - * default (the streamManager call site keeps it until the runtime seam - * reaches StreamManager); a context-bound runner puts the backoff sleep on - * the runtime's `Clock` — a `TestClock` in tests. + * default (direct construction in tests); AgentSession passes its stream + * manager's runner, so the backoff sleep shares the stream's `Clock` — the + * app runtime's in production, a `TestClock` in tests. */ private readonly runner: EffectRunner = defaultEffectRunner ) { diff --git a/src/node/services/serviceContainer.test.ts b/src/node/services/serviceContainer.test.ts index 35dacf73b2..4ce8b01984 100644 --- a/src/node/services/serviceContainer.test.ts +++ b/src/node/services/serviceContainer.test.ts @@ -6,37 +6,148 @@ import { Context, Duration, Effect, Layer } from "effect"; import { TestClock } from "effect/testing"; import { MULTI_PROJECT_CONFIG_KEY } from "@/common/constants/multiProject"; import { createConfigStores, type Config, type ConfigStores } from "@/node/config"; +import type { ORPCContext } from "@/node/orpc/context"; +import { isInteractiveHostKeyApprovalAvailable } from "@/node/runtime/sshConnectionPool"; import { AppFiberScopeTag } from "@/node/services/di/appFiberScope"; import { EffectRunnerTag } from "@/node/services/di/effectRunner"; import * as appLayers from "@/node/services/di/layers/app"; import { CoreOptionsTag } from "@/node/services/di/layers/core"; import { + AgentBrowserSessionDiscovery, + AgentPluginInstall, AI, Analytics, + Backup, + BrowserBridgeServerTag, + BrowserBridgeTokenManagerTag, + BrowserControl, + BrowserSessionStateHubTag, + Coder, + CoderOauth, + CodexOauth, + ConfigTag, + CopilotOauth, + DesktopBridgeServerTag, + DesktopSessionManagerTag, + DesktopTokenManagerTag, DevTools, + Editor, Experiments, + FileLeaseManagerTag, + History, IdleDispatcherTag, InitStateManagerTag, + Instructions, MCPConfig, + McpOauth, MCPServerManagerTag, Memory, MemoryConsolidation, MemoryMeta, + MenuEvent, + MuxGatewayOauth, + MuxGovernorOauth, Policy, + Project, Provider, + ProvidersConfigStoreTag, + QuickJSRuntimeFactoryTag, + Refine, + SecretsStoreTag, + Server, + ServerAuth, + SessionLocatorTag, SessionTiming, SessionUsage, + SshPrompt, StreamManagerTag, Task, Telemetry, + Terminal, + Timeline, + Tokenizer, + TurnRequestBuilderBindingsTag, + Update, + Voice, + WindowTag, Workspace, WorkspaceGoal, + WorkspaceLifecycleHooksTag, WorkspaceMcpOverrides, - WorkspaceTurnManagerTag, + WorktreeArchiveSnapshot, type AppTags, } from "@/node/services/di/tags"; import { ServiceContainer } from "./serviceContainer"; +/** + * Independent field → tag listing for every ORPC context field (the production + * mapping lives in the Layer files); `Record` keeps it exhaustive, so + * a field added to `ORPCContext` without a tag fails to compile here. + */ +const ORPC_FIELD_TAGS: Record< + keyof Omit, + Context.Key +> = { + config: ConfigTag, + sessionLocator: SessionLocatorTag, + providersConfigStore: ProvidersConfigStoreTag, + secretsStore: SecretsStoreTag, + fileLeaseManager: FileLeaseManagerTag, + aiService: AI, + historyService: History, + streamManager: StreamManagerTag, + initStateManager: InitStateManagerTag, + projectService: Project, + workspaceService: Workspace, + taskService: Task, + providerService: Provider, + muxGatewayOauthService: MuxGatewayOauth, + muxGovernorOauthService: MuxGovernorOauth, + codexOauthService: CodexOauth, + coderOauthService: CoderOauth, + copilotOauthService: CopilotOauth, + backupService: Backup, + terminalService: Terminal, + editorService: Editor, + windowService: WindowTag, + updateService: Update, + tokenizerService: Tokenizer, + serverService: Server, + menuEventService: MenuEvent, + voiceService: Voice, + mcpConfigService: MCPConfig, + mcpOauthService: McpOauth, + workspaceMcpOverridesService: WorkspaceMcpOverrides, + mcpServerManager: MCPServerManagerTag, + agentPluginInstallService: AgentPluginInstall, + sessionTimingService: SessionTiming, + timelineService: Timeline, + telemetryService: Telemetry, + experimentsService: Experiments, + memoryService: Memory, + memoryMetaService: MemoryMeta, + memoryConsolidationService: MemoryConsolidation, + refineService: Refine, + sessionUsageService: SessionUsage, + instructionsService: Instructions, + workspaceGoalService: WorkspaceGoal, + devToolsService: DevTools, + browserSessionDiscoveryService: AgentBrowserSessionDiscovery, + browserBridgeTokenManager: BrowserBridgeTokenManagerTag, + browserBridgeServer: BrowserBridgeServerTag, + browserControlService: BrowserControl, + browserSessionStateHub: BrowserSessionStateHubTag, + policyService: Policy, + coderService: Coder, + serverAuthService: ServerAuth, + sshPromptService: SshPrompt, + analyticsService: Analytics, + desktopSessionManager: DesktopSessionManagerTag, + desktopTokenManager: DesktopTokenManagerTag, + desktopBridgeServer: DesktopBridgeServerTag, + workflowRuntimeFactory: QuickJSRuntimeFactoryTag, +}; + describe("ServiceContainer", () => { let tempDir: string; let config: Config; @@ -283,37 +394,20 @@ describe("ServiceContainer", () => { services.idleCompactionService.stop(); }); - it("serves the layer-built core and cross-cutting services through the fields and the Effect context", () => { + it("serves every ORPC context field through its tag (one instance each)", () => { services = new ServiceContainer(stores); - const effectContext = services.toORPCContext()["effect/context"]; + const orpcContext = services.toORPCContext(); + const effectContext = orpcContext["effect/context"]; - const fieldTags: Array<[keyof ServiceContainer, Context.Key]> = [ - ["aiService", AI], - ["streamManager", StreamManagerTag], - ["initStateManager", InitStateManagerTag], - ["workspaceService", Workspace], - ["taskService", Task], - ["workspaceTurnManager", WorkspaceTurnManagerTag], - ["providerService", Provider], - ["mcpConfigService", MCPConfig], - ["mcpServerManager", MCPServerManagerTag], - ["sessionUsageService", SessionUsage], - ["workspaceGoalService", WorkspaceGoal], - ["memoryService", Memory], - ["memoryMetaService", MemoryMeta], - ["memoryConsolidationService", MemoryConsolidation], - ["idleDispatcher", IdleDispatcherTag], - ["policyService", Policy], - ["telemetryService", Telemetry], - ["experimentsService", Experiments], - ["sessionTimingService", SessionTiming], - ["analyticsService", Analytics], - ["devToolsService", DevTools], - ["workspaceMcpOverridesService", WorkspaceMcpOverrides], - ]; - for (const [field, tag] of fieldTags) { - expect(Context.get(effectContext, tag)).toBe(services[field]); + for (const [field, tag] of Object.entries(ORPC_FIELD_TAGS) as Array< + [keyof typeof ORPC_FIELD_TAGS, Context.Key] + >) { + expect(Context.get(effectContext, tag)).toBe(orpcContext[field]); } + expect(services.runtime.get(IdleDispatcherTag)).toBe(services.idleDispatcher); + expect(services.runtime.get(StreamManagerTag).effectRunner).toBe( + services.runtime.get(EffectRunnerTag) + ); // The core graph's options are derived from the layer-built cross-cutting // instances, so core constructors received the same objects the fields expose. const coreOptions = services.runtime.get(CoreOptionsTag); @@ -321,6 +415,136 @@ describe("ServiceContainer", () => { expect(coreOptions.experimentsService).toBe(services.experimentsService); }); + it("wires the desktop services like the constructor did (each line has an observable effect)", () => { + services = new ServiceContainer(stores); + + // turnRequestBuilderBindings: the desktop-only collaborators. + const bindings = services.runtime.get(TurnRequestBuilderBindingsTag); + expect(bindings.analyticsService).toBe(services.analyticsService); + expect(bindings.desktopSessionManager).toBe(services.desktopSessionManager); + expect(bindings.timelineService).toBe(services.timelineService); + expect(bindings.codexOauthService).toBe(services.codexOauthService); + expect(bindings.coderOauthService).toBe(services.coderOauthService); + + // Setter-provided collaborators (the former `set*` lines). + const workspaceInternals = services.workspaceService as unknown as { + terminalService?: unknown; + desktopSessionManager?: unknown; + refinePassCanceller?: unknown; + timelineRecorder?: unknown; + worktreeArchiveSnapshotService?: unknown; + workspaceLifecycleHooks?: unknown; + }; + expect(workspaceInternals.terminalService).toBe(services.terminalService); + expect(workspaceInternals.desktopSessionManager).toBe(services.desktopSessionManager); + expect(workspaceInternals.refinePassCanceller).toBe(services.refineService); + expect(workspaceInternals.timelineRecorder).toBe(services.timelineService); + expect(workspaceInternals.worktreeArchiveSnapshotService).toBe( + services.runtime.get(WorktreeArchiveSnapshot) + ); + expect(workspaceInternals.workspaceLifecycleHooks).toBe( + services.runtime.get(WorkspaceLifecycleHooksTag) + ); + for (const recorderOwner of [ + services.taskService, + services.heartbeatService, + services.workspaceGoalService, + ]) { + expect((recorderOwner as unknown as { timelineRecorder?: unknown }).timelineRecorder).toBe( + services.timelineService + ); + } + const projectInternals = services.projectService as unknown as { + workspaceService?: unknown; + workspaceMetadataRefresher?: unknown; + mcpServerManager?: unknown; + }; + expect(projectInternals.workspaceService).toBe(services.workspaceService); + expect(projectInternals.workspaceMetadataRefresher).toBe(services.workspaceService); + expect(projectInternals.mcpServerManager).toBe(services.mcpServerManager); + expect( + (services.mcpServerManager as unknown as { mcpOauthService?: unknown }).mcpOauthService + ).toBe(services.mcpOauthService); + const backupInternals = services.backupService as unknown as { + projectRegistrar?: unknown; + memoryNotifier?: unknown; + }; + expect(backupInternals.projectRegistrar).toBe(services.projectService); + expect(backupInternals.memoryNotifier).toBe(services.memoryService); + + // Idle-compaction outcomes reach the idle compaction service. + const recordOutcomeSpy = spyOn(services.idleCompactionService, "recordOutcome"); + const outcomeListener = ( + services.workspaceService as unknown as { + idleCompactionOutcomeListener?: (workspaceId: string, outcome: unknown) => void; + } + ).idleCompactionOutcomeListener; + outcomeListener?.("ws-1", { success: true }); + expect(recordOutcomeSpy).toHaveBeenCalledWith("ws-1", { success: true }); + + // Global registrations: the SSH connection pools consult this container's + // prompt service for interactive host-key approval. + const responderSpy = spyOn(services.sshPromptService, "hasInteractiveResponder"); + responderSpy.mockReturnValue(true); + expect(isInteractiveHostKeyApprovalAvailable()).toBe(true); + responderSpy.mockReturnValue(false); + expect(isInteractiveHostKeyApprovalAvailable()).toBe(false); + + // Timeline subscribed to the workspace service, and the workers' timing + // listeners registered: a stream-start reaches the session timing service. + const timingSpy = spyOn(services.sessionTimingService, "handleStreamStart").mockImplementation( + () => undefined + ); + services.aiService.emit("stream-start", { + type: "stream-start", + workspaceId: "ws-1", + messageId: "m-1", + model: "openai:gpt-4o", + historySequence: 1, + startTime: Date.now(), + mode: "exec", + }); + expect(timingSpy).toHaveBeenCalledTimes(1); + }); + + it("tears down in the fixed dispose() and shutdown() order", async () => { + const order: string[] = []; + const record = (step: string) => () => { + order.push(step); + return Promise.resolve(undefined); + }; + services = new ServiceContainer(stores); + spyOn(services.desktopBridgeServer, "stop").mockImplementation(record("bridge.stop")); + spyOn(services.desktopSessionManager, "closeAll").mockImplementation( + record("sessions.closeAll") + ); + spyOn(services.browserBridgeServer, "stop").mockImplementation(record("browserBridge.stop")); + spyOn(services.analyticsService, "dispose").mockImplementation(record("analytics.dispose")); + spyOn(services.timelineService, "flush").mockImplementation(record("timeline.flush")); + spyOn(services.telemetryService, "shutdown").mockImplementation(record("telemetry.shutdown")); + + await services.dispose(); + // §5: bridge before sessions; browser bridge before analytics; timeline flush last. + expect(order).toEqual([ + "bridge.stop", + "sessions.closeAll", + "browserBridge.stop", + "analytics.dispose", + "timeline.flush", + ]); + + order.length = 0; + await services.shutdown(); + expect(order).toEqual([ + "bridge.stop", + "sessions.closeAll", + "browserBridge.stop", + "timeline.flush", + "analytics.dispose", + "telemetry.shutdown", + ]); + }); + it("surfaces a throwing layer as a synchronous constructor throw", () => { const realAppLive = appLayers.AppLive; const appLiveSpy = spyOn(appLayers, "AppLive").mockImplementation((appStores) => diff --git a/src/node/services/serviceContainer.ts b/src/node/services/serviceContainer.ts index 071ecd6ac2..bb1f36db87 100644 --- a/src/node/services/serviceContainer.ts +++ b/src/node/services/serviceContainer.ts @@ -1,85 +1,51 @@ -import * as path from "path"; -import { DEFAULT_CODER_ARCHIVE_BEHAVIOR } from "@/common/config/coderArchiveBehavior"; -import { DEFAULT_WORKTREE_ARCHIVE_BEHAVIOR } from "@/common/config/worktreeArchiveBehavior"; import { log } from "@/node/services/log"; import type { Config, ConfigStores, WorkspaceSessionLocator } from "@/node/config"; import type { FileLeaseManager, ProvidersConfigStore, SecretsStore } from "@/node/config"; import type { CoreServices } from "@/node/services/coreServices"; -import { PTYService } from "@/node/services/ptyService"; import type { TerminalWindowManager } from "@/desktop/terminalWindowManager"; -import { ProjectService } from "@/node/services/projectService"; -import { MuxGatewayOauthService } from "@/node/services/muxGatewayOauthService"; -import { MuxGovernorOauthService } from "@/node/services/muxGovernorOauthService"; -import { CodexOauthService } from "@/node/services/codexOauthService"; -import { CoderOauthService } from "@/node/services/coderOauthService"; -import { CopilotOauthService } from "@/node/services/copilotOauthService"; -import { TerminalService } from "@/node/services/terminalService"; -import { BackupService } from "@/node/services/backup/backupService"; -import { createBackupGitRepo, createBackupPayloadStore } from "@/node/services/backup/adapters"; -import { EditorService } from "@/node/services/editorService"; -import { WindowService } from "@/node/services/windowService"; -import { UpdateService } from "@/node/services/updateService"; -import { TokenizerService } from "@/node/services/tokenizerService"; -import { InstructionsService } from "@/node/services/instructionsService"; -import { ServerService } from "@/node/services/serverService"; -import { MenuEventService } from "@/node/services/menuEventService"; -import { VoiceService } from "@/node/services/voiceService"; +import type { ProjectService } from "@/node/services/projectService"; +import type { MuxGatewayOauthService } from "@/node/services/muxGatewayOauthService"; +import type { MuxGovernorOauthService } from "@/node/services/muxGovernorOauthService"; +import type { CodexOauthService } from "@/node/services/codexOauthService"; +import type { CoderOauthService } from "@/node/services/coderOauthService"; +import type { CopilotOauthService } from "@/node/services/copilotOauthService"; +import type { TerminalService } from "@/node/services/terminalService"; +import type { BackupService } from "@/node/services/backup/backupService"; +import type { EditorService } from "@/node/services/editorService"; +import type { WindowService } from "@/node/services/windowService"; +import type { UpdateService } from "@/node/services/updateService"; +import type { TokenizerService } from "@/node/services/tokenizerService"; +import type { InstructionsService } from "@/node/services/instructionsService"; +import type { ServerService } from "@/node/services/serverService"; +import type { MenuEventService } from "@/node/services/menuEventService"; +import type { VoiceService } from "@/node/services/voiceService"; import type { TelemetryService } from "@/node/services/telemetryService"; -import type { - ErrorEvent, - ReasoningDeltaEvent, - StreamAbortEvent, - StreamDeltaEvent, - StreamEndEvent, - StreamStartEvent, - ToolCallDeltaEvent, - ToolCallEndEvent, - ToolCallStartEvent, -} from "@/common/types/stream"; -import { BrowserBridgeServer } from "@/node/services/browser/BrowserBridgeServer"; -import { AgentBrowserSessionDiscoveryService } from "@/node/services/browser/AgentBrowserSessionDiscoveryService"; -import { BrowserBridgeTokenManager } from "@/node/services/browser/BrowserBridgeTokenManager"; -import { BrowserControlService } from "@/node/services/browser/BrowserControlService"; -import { BrowserSessionStateHub } from "@/node/services/browser/BrowserSessionStateHub"; +import type { BrowserBridgeServer } from "@/node/services/browser/BrowserBridgeServer"; +import type { AgentBrowserSessionDiscoveryService } from "@/node/services/browser/AgentBrowserSessionDiscoveryService"; +import type { BrowserBridgeTokenManager } from "@/node/services/browser/BrowserBridgeTokenManager"; +import type { BrowserControlService } from "@/node/services/browser/BrowserControlService"; +import type { BrowserSessionStateHub } from "@/node/services/browser/BrowserSessionStateHub"; import type { DevToolsService } from "@/node/services/devToolsService"; import type { SessionTimingService } from "@/node/services/sessionTimingService"; -import { TimelineService } from "@/node/services/timelineService"; -import type { - AnalyticsService, - IngestWorkspaceMeta, -} from "@/node/services/analytics/analyticsService"; +import type { TimelineService } from "@/node/services/timelineService"; +import type { AnalyticsService } from "@/node/services/analytics/analyticsService"; import type { ExperimentsService } from "@/node/services/experimentsService"; import type { WorkspaceMcpOverridesService } from "@/node/services/workspaceMcpOverridesService"; -import { AgentPluginInstallService } from "@/node/services/agentPlugins/installService"; -import { EXPERIMENT_IDS } from "@/common/constants/experiments"; -import { McpOauthService } from "@/node/services/mcpOauthService"; -import { HeartbeatService } from "@/node/services/heartbeatService"; -import { AgentStatusService } from "@/node/services/agentStatusService"; -import { IdleCompactionService } from "@/node/services/idleCompactionService"; +import type { AgentPluginInstallService } from "@/node/services/agentPlugins/installService"; +import type { McpOauthService } from "@/node/services/mcpOauthService"; +import type { HeartbeatService } from "@/node/services/heartbeatService"; +import type { AgentStatusService } from "@/node/services/agentStatusService"; +import type { IdleCompactionService } from "@/node/services/idleCompactionService"; import type { IdleDispatcher } from "@/node/services/idleDispatcher"; -import { coderService, type CoderService } from "@/node/services/coderService"; -import { SshPromptService } from "@/node/services/sshPromptService"; -import { WorkspaceLifecycleHooks } from "@/node/services/workspaceLifecycleHooks"; -import { WorktreeArchiveSnapshotService } from "@/node/services/worktreeArchiveSnapshotService"; -import { - createCoderArchiveHook, - createCoderUnarchiveHook, -} from "@/node/runtime/coderLifecycleHooks"; -import { createWorktreeArchiveHook } from "@/node/runtime/worktreeLifecycleHooks"; -import { QuickJSRuntimeFactory } from "@/node/services/ptc/quickjsRuntime"; -import { RefineService } from "@/node/services/refinement/refineService"; -import { setGlobalCoderService } from "@/node/runtime/runtimeFactory"; -import { setSshPromptService } from "@/node/runtime/sshConnectionPool"; -import { setSshPromptService as setSSH2SshPromptService } from "@/node/runtime/SSH2ConnectionPool"; -import { - createRuntimeForWorkspace, - resolveWorkspaceExecutionPath, -} from "@/node/runtime/runtimeHelpers"; +import type { CoderService } from "@/node/services/coderService"; +import type { SshPromptService } from "@/node/services/sshPromptService"; +import type { QuickJSRuntimeFactory } from "@/node/services/ptc/quickjsRuntime"; +import type { RefineService } from "@/node/services/refinement/refineService"; import type { PolicyService } from "@/node/services/policyService"; -import { ServerAuthService } from "@/node/services/serverAuthService"; -import { DesktopBridgeServer } from "@/node/services/desktop/DesktopBridgeServer"; -import { DesktopSessionManager } from "@/node/services/desktop/DesktopSessionManager"; -import { DesktopTokenManager } from "@/node/services/desktop/DesktopTokenManager"; +import type { ServerAuthService } from "@/node/services/serverAuthService"; +import type { DesktopBridgeServer } from "@/node/services/desktop/DesktopBridgeServer"; +import type { DesktopSessionManager } from "@/node/services/desktop/DesktopSessionManager"; +import type { DesktopTokenManager } from "@/node/services/desktop/DesktopTokenManager"; import type { ORPCContext } from "@/node/orpc/context"; import type { Scope } from "effect"; import { AppFiberScopeTag } from "@/node/services/di/appFiberScope"; @@ -89,30 +55,85 @@ import { makeAppRuntime, type AppRuntime, } from "@/node/services/di/appRuntime"; -import { EffectRunnerTag } from "@/node/services/di/effectRunner"; import { AppLive } from "@/node/services/di/layers/app"; -import { coreServicesFromContext } from "@/node/services/di/layers/core"; import { + AgentBrowserSessionDiscovery, + AgentPluginInstall, + AgentStatus, + AI, Analytics, + BackgroundProcessManagerTag, + Backup, + BrowserBridgeServerTag, + BrowserBridgeTokenManagerTag, + BrowserControl, + BrowserSessionStateHubTag, + Coder, + CoderOauth, + CodexOauth, + ConfigTag, + CopilotOauth, + DesktopBridgeServerTag, + DesktopSessionManagerTag, + DesktopTokenManagerTag, DevTools, + Editor, Experiments, + ExtensionMetadata, + FileLeaseManagerTag, + Heartbeat, + History, + IdleCompaction, + IdleDispatcherTag, + InitStateManagerTag, + Instructions, + MCPConfig, + McpOauth, + MCPServerManagerTag, + Memory, + MemoryConsolidation, + MemoryMeta, + MenuEvent, + MuxGatewayOauth, + MuxGovernorOauth, Policy, + Project, + Provider, + ProvidersConfigStoreTag, + QuickJSRuntimeFactoryTag, + Refine, + SecretsStoreTag, + Server, + ServerAuth, + SessionLocatorTag, SessionTiming, + SessionUsage, + SshPrompt, + StreamManagerTag, + Task, Telemetry, + Terminal, + Timeline, + Tokenizer, + Update, + Voice, + WindowTag, + Workspace, + WorkspaceGoal, WorkspaceMcpOverrides, + WorkspaceTurnManagerTag, type AppTags, } from "@/node/services/di/tags"; /** * ServiceContainer - Central dependency container for all backend services. * - * This class instantiates and wires together all services needed by the ORPC router. - * Services are accessed via the ORPC context object. - * - * Services provided by the Effect Layer graph (`di/layers/app.ts`: the stores, - * the runtime seams, the cross-cutting services and the whole core graph) are - * built first by `runtime` and handed to the constructor-wired desktop - * remainder; the migration moves services into the graph incrementally (see - * the DI contract in `di/appRuntime.ts`). + * Every service is built by the Effect Layer graph (`di/layers/app.ts`: the + * stores, the runtime seams, the cross-cutting services, the core graph shared + * with the CLI roots, and the desktop group layers with their wiring). The + * constructor builds that graph once, eagerly and synchronously, and exposes + * the services as plain fields for the ORPC context; startup (`initialize()`) + * and the hand-ordered teardown (`dispose()`/`shutdown()`) stay here (DI + * contract in `di/appRuntime.ts`). */ export class ServiceContainer { public readonly runtime: AppRuntime; @@ -121,7 +142,7 @@ export class ServiceContainer { * Closed early in `dispose()`; no production occupant yet. */ public readonly appFiberScope: Scope.Closeable; - public readonly workflowRuntimeFactory = new QuickJSRuntimeFactory(); + public readonly workflowRuntimeFactory: QuickJSRuntimeFactory; public readonly config: Config; public readonly sessionLocator: WorkspaceSessionLocator; public readonly providersConfigStore: ProvidersConfigStore; @@ -147,7 +168,7 @@ export class ServiceContainer { public readonly refineService: RefineService; private readonly extensionMetadata: CoreServices["extensionMetadata"]; private readonly backgroundProcessManager: CoreServices["backgroundProcessManager"]; - // Desktop-only services + // Desktop-only services (`di/layers/desktop.ts`) public readonly projectService: ProjectService; public readonly muxGatewayOauthService: MuxGatewayOauthService; public readonly muxGovernorOauthService: MuxGovernorOauthService; @@ -184,8 +205,7 @@ export class ServiceContainer { public readonly desktopSessionManager: DesktopSessionManager; public readonly desktopTokenManager: DesktopTokenManager; public readonly desktopBridgeServer: DesktopBridgeServer; - public readonly sshPromptService = new SshPromptService(); - private readonly ptyService: PTYService; + public readonly sshPromptService: SshPromptService; public readonly idleCompactionService: IdleCompactionService; public readonly idleDispatcher: IdleDispatcher; public readonly heartbeatService: HeartbeatService; @@ -201,421 +221,76 @@ export class ServiceContainer { private disposePromise: Promise | null = null; constructor(stores: ConfigStores) { - // Built eagerly and synchronously (a layer body that throws fails the - // constructor, like any service constructor) before the constructor-wired - // services so layer-provided instances can be passed into them. + // Built eagerly and synchronously: a layer body that throws fails the + // constructor, like any service constructor did before the graph existed. this.runtime = makeAppRuntime(AppLive(stores)); - this.appFiberScope = this.runtime.get(AppFiberScopeTag); - // Clock-driven workers run their lifecycle fibers through the runtime's - // context-bound runner (unsupervised; see di/effectRunner.ts). - const effectRunner = this.runtime.get(EffectRunnerTag); - const config = stores.config; - this.config = config; - this.sessionLocator = stores.sessionLocator; - this.providersConfigStore = stores.providersConfigStore; - this.secretsStore = stores.secretsStore; - this.fileLeaseManager = stores.fileLeaseManager; - - // Cross-cutting services: layer-built (`CrossCuttingLive`) ahead of the - // core graph, whose options derive from them (`CoreOptionsFromDesktopLive`). - this.policyService = this.runtime.get(Policy); - this.telemetryService = this.runtime.get(Telemetry); - this.experimentsService = this.runtime.get(Experiments); - this.backupService = new BackupService(config, { - gitRepo: createBackupGitRepo({ - cacheRoot: path.join(config.rootDir, "backup-cache"), - }), - payload: createBackupPayloadStore({ config }), - }); - this.sessionTimingService = this.runtime.get(SessionTiming); - this.analyticsService = this.runtime.get(Analytics); - this.devToolsService = this.runtime.get(DevTools); - this.browserBridgeTokenManager = new BrowserBridgeTokenManager(); - this.workspaceMcpOverridesService = this.runtime.get(WorkspaceMcpOverrides); - - // The core graph (shared with the `xum run`/`xum workflow` roots) is built by - // `CoreLive`; read it back as the plain object the wiring below uses. - const core = coreServicesFromContext(this.runtime.context); - - // Spread core services into class fields - this.historyService = core.historyService; - this.aiService = core.aiService; - this.streamManager = core.streamManager; - this.initStateManager = core.initStateManager; - core.turnRequestBuilderBindings.analyticsService = this.analyticsService; - this.browserSessionDiscoveryService = new AgentBrowserSessionDiscoveryService({ - resolveWorkspaceCandidatePathsFn: async (workspaceId: string) => { - const allWorkspaceMetadata = await config.getAllWorkspaceMetadata(); - const workspaceMetadata = - allWorkspaceMetadata.find((candidate) => candidate.id === workspaceId) ?? null; - if (workspaceMetadata == null) { - return []; - } - - const runtime = createRuntimeForWorkspace(workspaceMetadata); - const workspacePath = resolveWorkspaceExecutionPath(workspaceMetadata, runtime); - return [workspaceMetadata.projectPath, workspacePath].filter( - (candidatePath): candidatePath is string => candidatePath.trim().length > 0 - ); - }, - }); - this.browserControlService = new BrowserControlService({ - browserSessionDiscoveryService: this.browserSessionDiscoveryService, - resolveSessionEnvFn: () => Promise.resolve(process.env), - }); - this.browserSessionStateHub = new BrowserSessionStateHub({ - browserControlService: this.browserControlService, - }); - this.browserBridgeServer = new BrowserBridgeServer({ - browserSessionDiscoveryService: this.browserSessionDiscoveryService, - browserBridgeTokenManager: this.browserBridgeTokenManager, - browserSessionStateHub: this.browserSessionStateHub, - }); - this.workspaceService = core.workspaceService; - this.taskService = core.taskService; - this.workspaceTurnManager = core.workspaceTurnManager; - this.providerService = core.providerService; - this.mcpConfigService = core.mcpConfigService; - this.mcpServerManager = core.mcpServerManager; - this.sessionUsageService = core.sessionUsageService; - this.workspaceGoalService = core.workspaceGoalService; - this.memoryService = core.memoryService; - this.memoryMetaService = core.memoryMetaService; - this.memoryConsolidationService = core.memoryConsolidationService; - this.extensionMetadata = core.extensionMetadata; - this.backgroundProcessManager = core.backgroundProcessManager; - - // Managed Agent Plugin installer (agent-plugins experiment). Gated on the - // backend ExperimentsService exactly like the plugin MCP provider; the - // MCP manager dependency lets update/uninstall recycle running plugin - // servers whose content changed behind an unchanged command line. - this.agentPluginInstallService = new AgentPluginInstallService(config, { - isEnabled: () => this.experimentsService.isExperimentEnabled(EXPERIMENT_IDS.AGENT_PLUGINS), - mcpServerManager: this.mcpServerManager, - workspaceMcpOverridesService: this.workspaceMcpOverridesService, - }); - - this.projectService = new ProjectService(config, this.sshPromptService, this.secretsStore); - this.projectService.setWorkspaceService(this.workspaceService); - this.projectService.setWorkspaceMetadataRefresher(this.workspaceService); - this.projectService.setMcpServerManager(this.mcpServerManager); - // Backup restores register approved project imports through the same create() path the - // UI uses; setter injection because BackupService is constructed before ProjectService. - this.backupService.setProjectService(this.projectService); - // Restored project memory is written directly, so the service announces it through - // MemoryService for the memory browser's change subscription. - this.backupService.setMemoryNotifier(this.memoryService); - this.desktopSessionManager = new DesktopSessionManager({ - config, - experimentsService: this.experimentsService, - workspaceService: this.workspaceService, - }); - core.turnRequestBuilderBindings.desktopSessionManager = this.desktopSessionManager; - this.desktopTokenManager = new DesktopTokenManager(); - this.desktopBridgeServer = new DesktopBridgeServer({ - desktopSessionManager: this.desktopSessionManager, - desktopTokenManager: this.desktopTokenManager, - }); - - // Idle compaction service - auto-compacts workspaces after configured idle period - this.idleCompactionService = new IdleCompactionService( - config, - this.historyService, - this.extensionMetadata, - (workspaceId) => this.workspaceService.executeIdleCompaction(workspaceId), - effectRunner - ); - // Forward terminal idle-compaction outcomes so the loop stops re-attempting a - // persistently failing workspace (immediately on model_not_found, otherwise after - // two consecutive failures). - this.workspaceService.setIdleCompactionOutcomeListener((workspaceId, outcome) => - this.idleCompactionService.recordOutcome(workspaceId, outcome) - ); - // IdleDispatcher + goal continuation bridge are owned by the core graph - // so the wiring works for `xum run` too. Share the same dispatcher with - // HeartbeatService — its priority ordering ensures an active goal - // suppresses background heartbeats. - this.idleDispatcher = core.idleDispatcher; - this.heartbeatService = new HeartbeatService( - config, - this.extensionMetadata, - this.workspaceService, - this.taskService, - this.idleDispatcher, - effectRunner - ); - this.timelineService = new TimelineService( - config, - this.historyService, - this.experimentsService - ); - // /refine trajectory distillation (RLM r11). Chat emission routes through - // WorkspaceService so a live session renders the appended summary row - // immediately (the row itself is already durable in chat.jsonl). - this.refineService = new RefineService( - config, - this.memoryService, - this.memoryMetaService, - this.historyService, - this.aiService, - this.experimentsService, - { - timelineService: this.timelineService, - sessionUsageService: this.sessionUsageService, - emitChatMessage: (workspaceId, message) => - this.workspaceService.emitChatEvent(workspaceId, { ...message, type: "message" }), - // r40: refine row publication and apply mutations must not interleave - // with a concurrent turn's PREPARING snapshot or split its - // user/assistant pair — hold the session's turn-admission block while - // they land, failing closed when a turn is active. - acquireTurnExclusion: (workspaceId) => - this.workspaceService.acquireIdleTurnExclusion(workspaceId), - } - ); - // Removal must be able to abort + drain a running /refine pass before it - // deletes the session directory (post-construction wiring: RefineService - // is built after WorkspaceService). - this.workspaceService.setRefinePassCanceller(this.refineService); - this.workspaceService.setTimelineRecorder(this.timelineService); - this.taskService.setTimelineRecorder(this.timelineService); - this.heartbeatService.setTimelineRecorder(this.timelineService); - this.workspaceGoalService.setTimelineRecorder(this.timelineService); - core.turnRequestBuilderBindings.timelineService = this.timelineService; - this.timelineService.subscribeToWorkspace(this.workspaceService); - this.windowService = new WindowService(); - this.mcpOauthService = new McpOauthService( - config, - this.mcpConfigService, - this.windowService, - this.telemetryService - ); - this.mcpServerManager.setMcpOauthService(this.mcpOauthService); - - this.muxGatewayOauthService = new MuxGatewayOauthService( - this.providersConfigStore, - this.providerService, - this.windowService - ); - this.muxGovernorOauthService = new MuxGovernorOauthService( - config, - this.windowService, - this.policyService - ); - this.codexOauthService = new CodexOauthService( - this.providersConfigStore, - this.providerService, - this.windowService - ); - core.turnRequestBuilderBindings.codexOauthService = this.codexOauthService; - this.coderOauthService = new CoderOauthService( - this.providersConfigStore, - this.fileLeaseManager, - this.providerService, - this.windowService, - // Policy-aware: an enforced forcedBaseUrl overrides the deployment URL - // for logins, refreshes, and issuer checks. - this.policyService - ); - core.turnRequestBuilderBindings.coderOauthService = this.coderOauthService; - this.copilotOauthService = new CopilotOauthService(this.providerService, this.windowService); - // Terminal services - PTYService is cross-platform - this.ptyService = new PTYService(); - this.terminalService = new TerminalService(config, this.ptyService, this.secretsStore); - // Wire terminal service to workspace service for cleanup on removal - this.workspaceService.setTerminalService(this.terminalService); - this.workspaceService.setDesktopSessionManager(this.desktopSessionManager); - // Plugin-override pruning is wired inside the core graph (shared with - // headless CLI registration), using this.workspaceMcpOverridesService. - // Editor service for opening workspaces in code editors - this.editorService = new EditorService(config, this.workspaceService); - this.updateService = new UpdateService(this.config); - this.tokenizerService = new TokenizerService( - this.sessionUsageService, - this.aiService, - this.providerService - ); - this.instructionsService = new InstructionsService( - config, - this.aiService, - this.tokenizerService - ); - // AgentStatusService depends on tokenizer + window focus state; instantiate - // after both are constructed so the small-model status loop can run with - // accurate token budgeting and focus-aware cadence. - this.agentStatusService = new AgentStatusService( - config, - this.historyService, - this.tokenizerService, - this.extensionMetadata, - this.workspaceService, - this.windowService, - this.aiService, - // Status generation spends tokens outside StreamManager; give it a cost - // telemetry sink so that spend shows up in per-workspace usage, and an - // ingest trigger so the headless-usage sidecar reaches dashboard totals - // even when the workspace has no further stream activity. - { - sessionUsageService: this.sessionUsageService, - requestAnalyticsIngest: (workspaceId) => { - this.workspaceService.emit("analyticsIngest", { workspaceId }); - }, - } - ); - this.serverService = new ServerService(); - this.menuEventService = new MenuEventService(); - this.voiceService = new VoiceService( - config, - this.providerService, - this.policyService, - this.providersConfigStore - ); - this.coderService = coderService; - - this.serverAuthService = new ServerAuthService(config); - - const workspaceLifecycleHooks = new WorkspaceLifecycleHooks(); - const worktreeArchiveSnapshotService = new WorktreeArchiveSnapshotService(this.config); - this.workspaceService.setWorktreeArchiveSnapshotService(worktreeArchiveSnapshotService); - const getArchiveBehavior = () => - this.config.loadConfigOrDefault().coderWorkspaceArchiveBehavior ?? - DEFAULT_CODER_ARCHIVE_BEHAVIOR; - workspaceLifecycleHooks.registerBeforeArchive( - createCoderArchiveHook({ - coderService: this.coderService, - getArchiveBehavior, - // Model-driven archives probe the remote spawn-record layout before stopping a - // running Coder workspace: detached jobs surviving an unclean Xum exit live only in - // those records, which the host-local crash-orphan scans cannot see. - hasUnsettledRemoteBackgroundJobs: async (workspaceMetadata) => { - const runtime = createRuntimeForWorkspace(workspaceMetadata); - return await this.backgroundProcessManager.hasUnsettledRemoteSpawnRecords( - runtime, - workspaceMetadata.id - ); - }, - }) - ); - workspaceLifecycleHooks.registerAfterUnarchive( - createCoderUnarchiveHook({ - coderService: this.coderService, - getArchiveBehavior, - }) - ); - const getWorktreeArchiveBehavior = () => - this.config.loadConfigOrDefault().worktreeArchiveBehavior ?? - DEFAULT_WORKTREE_ARCHIVE_BEHAVIOR; - workspaceLifecycleHooks.registerAfterArchive( - createWorktreeArchiveHook({ getWorktreeArchiveBehavior }) - ); - this.workspaceService.setWorkspaceLifecycleHooks(workspaceLifecycleHooks); - - // Register globally so all createRuntime calls can create CoderSSHRuntime - setGlobalCoderService(this.coderService); - setSshPromptService(this.sshPromptService); - setSSH2SshPromptService(this.sshPromptService); - - // Backend timing stats. - this.aiService.on("stream-start", (data: StreamStartEvent) => - this.sessionTimingService.handleStreamStart(data) - ); - this.aiService.on("stream-delta", (data: StreamDeltaEvent) => - this.sessionTimingService.handleStreamDelta(data) - ); - this.aiService.on("reasoning-delta", (data: ReasoningDeltaEvent) => - this.sessionTimingService.handleReasoningDelta(data) - ); - this.aiService.on("tool-call-start", (data: ToolCallStartEvent) => - this.sessionTimingService.handleToolCallStart(data) - ); - this.aiService.on("tool-call-delta", (data: ToolCallDeltaEvent) => - this.sessionTimingService.handleToolCallDelta(data) - ); - this.aiService.on("tool-call-end", (data: ToolCallEndEvent) => - this.sessionTimingService.handleToolCallEnd(data) - ); - // Newly created sub-agent workspaces are ingested here before a full rebuild, - // so keep workspaceName + parentWorkspaceId to avoid NULL analytics attribution. - // Multi-project workspaces stay stored under _multi in config, but analytics should - // still attribute spend to the workspace's first real project path. - const ingestWorkspaceAnalytics = (workspaceId: string) => { - const workspaceLookup = this.config.findWorkspace(workspaceId); - const sessionDir = path.join(this.config.sessionsDir, workspaceId); - const analyticsProjectPath = - workspaceLookup?.attributionProjectPath ?? workspaceLookup?.projectPath; - this.analyticsService.ingestWorkspace(workspaceId, sessionDir, { - projectPath: analyticsProjectPath, - projectName: analyticsProjectPath ? path.basename(analyticsProjectPath) : undefined, - workspaceName: workspaceLookup?.workspaceName, - parentWorkspaceId: workspaceLookup?.parentWorkspaceId, - }); - }; - this.aiService.on("stream-end", (data: StreamEndEvent) => { - this.sessionTimingService.handleStreamEnd(data); - ingestWorkspaceAnalytics(data.workspaceId); - }); - // Billable usage persisted outside StreamManager stream-end requests its - // own incremental ingest pass. - this.workspaceService.on("analyticsIngest", (event) => { - ingestWorkspaceAnalytics(event.workspaceId); - }); - // Memory consolidation/harvest spend rides the headless-usage sidecar - // without any chat activity; ingest promptly so background sweeps reach - // dashboard totals instead of stranding until an unrelated stream-end - // or app restart. - this.memoryConsolidationService.on("analyticsIngest", (event: { workspaceId: string }) => { - ingestWorkspaceAnalytics(event.workspaceId); - }); - // WorkspaceService emits metadata:null after successful remove(). - // Clear analytics rows immediately so deleted workspaces disappear from stats - // without waiting for a future ingest pass. - this.workspaceService.on("metadata", (event) => { - if (event.metadata !== null) { - return; - } - - // Removed sub-agent children archive their transcript into the parent's - // session dir before this event fires. Re-ingest the parent (chained after - // the clear) so the child's spend is restored from the archive instead of - // vanishing from analytics until the parent's next stream-end. - let reingestAfterClear: - | { workspaceId: string; sessionDir: string; meta: IngestWorkspaceMeta } - | undefined; - const parentWorkspaceId = event.removedParentWorkspaceId; - if (parentWorkspaceId) { - const parentLookup = this.config.findWorkspace(parentWorkspaceId); - const parentProjectPath = parentLookup?.attributionProjectPath ?? parentLookup?.projectPath; - reingestAfterClear = { - workspaceId: parentWorkspaceId, - sessionDir: path.join(this.config.sessionsDir, parentWorkspaceId), - meta: { - projectPath: parentProjectPath, - projectName: parentProjectPath ? path.basename(parentProjectPath) : undefined, - workspaceName: parentLookup?.workspaceName, - parentWorkspaceId: parentLookup?.parentWorkspaceId, - }, - }; - } - - this.analyticsService.clearWorkspace(event.workspaceId, { reingestAfterClear }); - }); - - this.aiService.on("stream-abort", (data: StreamAbortEvent) => { - this.sessionTimingService.handleStreamAbort(data); - // Aborted turns persist their spend before this event fires (same async - // chain): normal aborts commit the usage-stamped partial to chat.jsonl - // (or the headless sidecar for non-commit-worthy partials); abandoned - // aborts (edit/discard) write only the sidecar. Ingest both, or the - // interrupted turn's spend stays out of dashboards until the next - // stream-end. - ingestWorkspaceAnalytics(data.workspaceId); - }); - // Errored turns whose partial would be dropped at commit time route their - // usage to the headless sidecar (persistStreamError). The sidecar write - // precedes this event in the same async chain, so ingest here keeps the - // dashboard current instead of waiting for the next stream or restart. - this.aiService.on("error", (data: ErrorEvent) => { - ingestWorkspaceAnalytics(data.workspaceId); - }); + const get = this.runtime.get; + this.appFiberScope = get(AppFiberScopeTag); + this.workflowRuntimeFactory = get(QuickJSRuntimeFactoryTag); + this.config = get(ConfigTag); + this.sessionLocator = get(SessionLocatorTag); + this.providersConfigStore = get(ProvidersConfigStoreTag); + this.secretsStore = get(SecretsStoreTag); + this.fileLeaseManager = get(FileLeaseManagerTag); + this.historyService = get(History); + this.aiService = get(AI); + this.streamManager = get(StreamManagerTag); + this.initStateManager = get(InitStateManagerTag); + this.workspaceService = get(Workspace); + this.taskService = get(Task); + this.workspaceTurnManager = get(WorkspaceTurnManagerTag); + this.providerService = get(Provider); + this.mcpConfigService = get(MCPConfig); + this.mcpServerManager = get(MCPServerManagerTag); + this.sessionUsageService = get(SessionUsage); + this.workspaceGoalService = get(WorkspaceGoal); + this.memoryService = get(Memory); + this.memoryMetaService = get(MemoryMeta); + this.memoryConsolidationService = get(MemoryConsolidation); + this.refineService = get(Refine); + this.extensionMetadata = get(ExtensionMetadata); + this.backgroundProcessManager = get(BackgroundProcessManagerTag); + this.projectService = get(Project); + this.muxGatewayOauthService = get(MuxGatewayOauth); + this.muxGovernorOauthService = get(MuxGovernorOauth); + this.codexOauthService = get(CodexOauth); + this.coderOauthService = get(CoderOauth); + this.copilotOauthService = get(CopilotOauth); + this.backupService = get(Backup); + this.terminalService = get(Terminal); + this.editorService = get(Editor); + this.windowService = get(WindowTag); + this.updateService = get(Update); + this.tokenizerService = get(Tokenizer); + this.instructionsService = get(Instructions); + this.serverService = get(Server); + this.menuEventService = get(MenuEvent); + this.voiceService = get(Voice); + this.mcpOauthService = get(McpOauth); + this.workspaceMcpOverridesService = get(WorkspaceMcpOverrides); + this.agentPluginInstallService = get(AgentPluginInstall); + this.telemetryService = get(Telemetry); + this.sessionTimingService = get(SessionTiming); + this.timelineService = get(Timeline); + this.devToolsService = get(DevTools); + this.browserSessionDiscoveryService = get(AgentBrowserSessionDiscovery); + this.browserBridgeTokenManager = get(BrowserBridgeTokenManagerTag); + this.browserBridgeServer = get(BrowserBridgeServerTag); + this.browserControlService = get(BrowserControl); + this.browserSessionStateHub = get(BrowserSessionStateHubTag); + this.analyticsService = get(Analytics); + this.experimentsService = get(Experiments); + this.policyService = get(Policy); + this.coderService = get(Coder); + this.serverAuthService = get(ServerAuth); + this.desktopSessionManager = get(DesktopSessionManagerTag); + this.desktopTokenManager = get(DesktopTokenManagerTag); + this.desktopBridgeServer = get(DesktopBridgeServerTag); + this.sshPromptService = get(SshPrompt); + this.idleCompactionService = get(IdleCompaction); + this.idleDispatcher = get(IdleDispatcherTag); + this.heartbeatService = get(Heartbeat); + this.agentStatusService = get(AgentStatus); } async initialize(): Promise { diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 5d7ef88925..cd100e1885 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -44,6 +44,7 @@ import * as modelStatsModule from "@/common/utils/tokens/modelStats"; import { SessionUsageService } from "./sessionUsageService"; import type { HistoryService } from "./historyService"; import { createTestHistoryService } from "./testHistoryService"; +import { makeTestEffectRunner } from "./di/testEffectRunner"; import { createAnthropic } from "@ai-sdk/anthropic"; import { countTokens } from "@/node/utils/main/tokenizer"; import { shouldRunIntegrationTests, validateApiKeys } from "../../../tests/testUtils"; @@ -1109,6 +1110,49 @@ describe("StreamManager - stream resource scope", () => { await new Promise((resolve) => setTimeout(resolve, throttleMs + 200)); expect(writePartialSpy.mock.calls.length).toBe(writesAtStreamEnd); }); + + test("runs the partial-write debounce on the injected runner's clock", async () => { + // The debounce fiber must sleep on the injected EffectRunner (the app + // runtime's clock in production), not the global runtime: a TestClock + // runner fires the flush only when the test clock advances. + const testRunner = makeTestEffectRunner(); + try { + const streamManager = new StreamManager( + historyService, + undefined, + undefined, + undefined, + testRunner.runner + ); + expect(streamManager.effectRunner).toBe(testRunner.runner); + const workspaceId = "runner-debounce-workspace"; + // Inside the throttle window, so the write is debounced rather than immediate. + const streamInfo = createStreamInfoForTests({ lastPartialWriteTime: Date.now() }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + const schedulePartialWrite = getPrivateMethodForTests< + (workspaceId: string, streamInfo: Record) => Promise + >(streamManager, "schedulePartialWrite"); + const writePartialSpy = spyOn(historyService, "writePartial"); + const throttleMs: unknown = Reflect.get(streamManager, "PARTIAL_WRITE_THROTTLE_MS"); + if (typeof throttleMs !== "number") { + throw new Error("Expected StreamManager.PARTIAL_WRITE_THROTTLE_MS to be a number"); + } + + await schedulePartialWrite.call(streamManager, workspaceId, streamInfo); + expect(streamInfo.partialWriteFiber).toBeDefined(); + // Real time passes; the virtual clock has not, so nothing flushes. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(writePartialSpy).not.toHaveBeenCalled(); + + await testRunner.adjust(throttleMs); + // The flush's Effect.promise settles on the next macrotask. + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(writePartialSpy).toHaveBeenCalledTimes(1); + expect(streamInfo.partialWriteFiber).toBeUndefined(); + } finally { + await testRunner.dispose(); + } + }); }); describe("StreamManager - stopWhen configuration", () => { diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index b9bc6f8dda..8284b05f5e 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -97,6 +97,7 @@ import { } from "@/common/utils/providers/modelEntries"; import { clampErrorMessage, getErrorMessage } from "@/common/utils/errors"; import { runLanguageModelCleanup } from "./languageModelCleanup"; +import { defaultEffectRunner, type EffectRunner } from "./di/effectRunner"; import { shellQuote } from "@/common/utils/shell"; import { classify429Capacity } from "@/common/utils/errors/classify429Capacity"; import { extractChunkDeltaText } from "@/common/utils/ai/streamChunks"; @@ -778,6 +779,14 @@ export class StreamManager { private readonly sessionUsageService?: SessionUsageService; private readonly getProvidersConfig: () => ProvidersConfigMap | null; private eventSink: TurnEngineEventSink; + /** + * Runs the clock-driven lifecycle fibers (partial-write debounce and its + * interrupt) and is handed to per-session workers that must share this + * stream's clock (`RetryManager`, via `AgentSessionStreamManager`). The app + * runtime's context-bound runner in production — a `TestClock` in tests — + * and the global runtime wherever nothing is injected (di/effectRunner.ts). + */ + public readonly effectRunner: EffectRunner; // Token tracker for live streaming statistics private tokenTracker = new StreamingTokenTracker(); // Track OpenAI previousResponseIds that have been invalidated @@ -793,12 +802,14 @@ export class StreamManager { historyService: HistoryService, sessionUsageService?: SessionUsageService, getProvidersConfig?: () => ProvidersConfigMap | null, - eventSink: TurnEngineEventSink = () => undefined + eventSink: TurnEngineEventSink = () => undefined, + runner: EffectRunner = defaultEffectRunner ) { this.historyService = historyService; this.sessionUsageService = sessionUsageService; this.getProvidersConfig = getProvidersConfig ?? (() => null); this.eventSink = eventSink; + this.effectRunner = runner; } setEventSink(eventSink: TurnEngineEventSink): void { @@ -1138,9 +1149,9 @@ export class StreamManager { // to the sleep, so the debounce delay is registered before this method // returns (same observable ordering as the previous setTimeout call). streamInfo.partialWriteFiber = streamInfo.resourceScope - ? Effect.runSync(Effect.forkIn(delayedFlush, streamInfo.resourceScope)) + ? this.effectRunner.runSync(Effect.forkIn(delayedFlush, streamInfo.resourceScope)) : // Whitebox test fixtures register stream infos without a resource scope. - Effect.runFork(delayedFlush); + this.effectRunner.runFork(delayedFlush); } /** @@ -1155,7 +1166,7 @@ export class StreamManager { return; } streamInfo.partialWriteFiber = undefined; - Effect.runFork(Fiber.interrupt(fiber)); + this.effectRunner.runFork(Fiber.interrupt(fiber)); } private async awaitPendingPartialWrite(streamInfo: WorkspaceStreamInfo): Promise {