🤖 refactor: split turn preparation into a TurnRequestBuilder pipeline

src/cli/run.ts

@@ -651,6 +651,8 @@ async function main(): Promise<number> {
     workspaceService,
     workspaceGoalService,
     idleDispatcher,
+    streamManager,
+    turnRequestBuilderBindings,
   } = createCoreServices({
     config,
     policyService,
@@ -675,7 +677,7 @@ async function main(): Promise<number> {
   // Codex OAuth explicitly to ensure Codex-routed OpenAI requests can load/refresh
   // OAuth tokens from providers.jsonc.
   const codexOauthService = new CodexOauthService(config, providerService);
-  aiService.setCodexOauthService(codexOauthService);
+  turnRequestBuilderBindings.codexOauthService = codexOauthService;
   // Same for Coder OAuth: coder:* models need per-request token loading/refresh.
   // Bind it to the REAL config (not the ephemeral tempDir copy): Coder rotates
   // the refresh token on every use, so persisting rotations only to tempDir
@@ -690,7 +692,7 @@ async function main(): Promise<number> {
     // token refreshes/issuer checks, and denied providers fail closed.
     policyService
   );
-  aiService.setCoderOauthService(coderOauthService);
+  turnRequestBuilderBindings.coderOauthService = coderOauthService;
 
   // CLI-only exit code control: allows agent to set the process exit code
   // Useful for CI workflows where the agent should block merge on failure
@@ -715,13 +717,14 @@ async function main(): Promise<number> {
       return { success: true, exit_code };
     },
   });
-  aiService.setExtraTools({ set_exit_code: setExitCodeTool });
+  turnRequestBuilderBindings.extraTools = { set_exit_code: setExitCodeTool };
 
   const session = new AgentSession({
     workspaceId,
     config,
     historyService,
     aiService,
+    streamManager,
     initStateManager,
     backgroundProcessManager,
     workspaceGoalService,

src/cli/workflow.ts

@@ -368,7 +368,7 @@ async function createWorkflowContext(options: {
       mcpConfig: realConfig,
     });
     codexOauthService = new CodexOauthService(config, services.providerService);
-    services.aiService.setCodexOauthService(codexOauthService);
+    services.turnRequestBuilderBindings.codexOauthService = codexOauthService;
     // Bind Coder OAuth to the REAL config (not the ephemeral tempDir copy):
     // Coder rotates the refresh token on every use, so persisting rotations
     // only to tempDir would strand ~/.xum/providers.jsonc with a consumed
@@ -382,7 +382,7 @@ async function createWorkflowContext(options: {
       // for token refreshes/issuer checks, and denied providers fail closed.
       policyService
     );
-    services.aiService.setCoderOauthService(coderOauthService);
+    services.turnRequestBuilderBindings.coderOauthService = coderOauthService;
 
     // Const capture: `services` is a `let`, so the deferred sanitize closure
     // below would lose TypeScript's definite-assignment narrowing.
@@ -392,6 +392,7 @@ async function createWorkflowContext(options: {
       config,
       historyService: services.historyService,
       aiService: services.aiService,
+      streamManager: services.streamManager,
       initStateManager: services.initStateManager,
       backgroundProcessManager: services.backgroundProcessManager,
       workspaceGoalService: services.workspaceGoalService,

src/node/orpc/context.ts

@@ -3,6 +3,8 @@ import type { IncomingHttpHeaders } from "http";
 import type { Config } from "@/node/config";
 import type { AIService } from "@/node/services/aiService";
 import type { HistoryService } from "@/node/services/historyService";
+import type { InitStateManager } from "@/node/services/initStateManager";
+import type { StreamManager } from "@/node/services/streamManager";
 import type { ProjectService } from "@/node/services/projectService";
 import type { WorkspaceService } from "@/node/services/workspaceService";
 import type { MuxGatewayOauthService } from "@/node/services/muxGatewayOauthService";
@@ -56,6 +58,8 @@ export interface ORPCContext {
   config: Config;
   aiService: AIService;
   historyService: HistoryService;
+  streamManager: StreamManager;
+  initStateManager: InitStateManager;
   projectService: ProjectService;
   workspaceService: WorkspaceService;
   taskService: TaskService;

src/node/orpc/router.test.ts

@@ -55,8 +55,8 @@ describe("router agent skill routes", () => {
 
       const context = {
         config: new Config(tempDir),
+        initStateManager: { waitForInit: mock(async () => undefined) },
         aiService: {
-          waitForInit: mock(async () => undefined),
           resolveXumToolScopeForWorkspace: mock(() => ({
             type: "project",
             xumHome: tempDir,
@@ -123,8 +123,8 @@ describe("router agent skill routes", () => {
 
       const context = {
         config: new Config(tempDir),
+        initStateManager: { waitForInit: mock(async () => undefined) },
         aiService: {
-          waitForInit: mock(async () => undefined),
           resolveXumToolScopeForWorkspace: mock(() => ({
             type: "project",
             xumHome: tempDir,

src/node/services/agentDefinitions/agentDefinitionsService.ts

@@ -942,7 +942,7 @@ export async function resolveAgentFrontmatter(
 
 export type AgentDefinitionsContext = Pick<
   ORPCContext,
-  "config" | "aiService" | "experimentsService"
+  "config" | "aiService" | "experimentsService" | "initStateManager"
 >;
 
 export async function resolveAgentDiscoveryContext(
@@ -980,7 +980,7 @@ export async function listAgentDefinitions(
     includeDisabled?: boolean;
   }
 ) {
-  if (input.workspaceId) await context.aiService.waitForInit(input.workspaceId);
+  if (input.workspaceId) await context.initStateManager.waitForInit(input.workspaceId);
   const { runtime, discoveryPath } = await resolveAgentDiscoveryContext(context, input);
   const includeAgentPlugins = context.experimentsService.isExperimentEnabled(
     EXPERIMENT_IDS.AGENT_PLUGINS
@@ -1041,7 +1041,7 @@ export async function getAgentDefinition(
     agentId: AgentId;
   }
 ) {
-  if (input.workspaceId) await context.aiService.waitForInit(input.workspaceId);
+  if (input.workspaceId) await context.initStateManager.waitForInit(input.workspaceId);
   const { runtime, discoveryPath } = await resolveAgentDiscoveryContext(context, input);
   return readAgentDefinition(runtime, discoveryPath, input.agentId, {
     includeAgentPlugins: context.experimentsService.isExperimentEnabled(

src/node/services/agentPlugins/workspacePluginOperations.ts

@@ -70,7 +70,7 @@ export async function listWorkspaceMcpPrompts(
   workspaceId: string,
   signal?: AbortSignal
 ) {
-  await context.aiService.waitForInit(workspaceId, signal);
+  await context.initStateManager.waitForInit(workspaceId, signal);
   const metadataResult = await context.aiService.getWorkspaceMetadata(workspaceId);
   if (!metadataResult.success) throw new Error(metadataResult.error);
   const metadata = metadataResult.data;
@@ -152,7 +152,7 @@ export async function listWorkspacePluginSlashCommands(
   signal?: AbortSignal
 ) {
   if (!context.experimentsService.isExperimentEnabled(EXPERIMENT_IDS.AGENT_PLUGINS)) return [];
-  await context.aiService.waitForInit(workspaceId, signal);
+  await context.initStateManager.waitForInit(workspaceId, signal);
   const metadataResult = await context.aiService.getWorkspaceMetadata(workspaceId);
   if (!metadataResult.success) throw new Error(metadataResult.error);
   const metadata = metadataResult.data;
@@ -173,7 +173,7 @@ export async function getWorkspacePluginComposition(
   workspaceId: string,
   signal?: AbortSignal
 ) {
-  await context.aiService.waitForInit(workspaceId, signal);
+  await context.initStateManager.waitForInit(workspaceId, signal);
   const metadataResult = await context.aiService.getWorkspaceMetadata(workspaceId);
   if (!metadataResult.success) throw new Error(metadataResult.error);
   const metadata = metadataResult.data;

src/node/services/agentSession.admissionGates.test.ts

@@ -8,6 +8,7 @@ import type { SendMessageError } from "@/common/types/errors";
 import { createMuxMessage } from "@/common/types/message";
 import { Ok } from "@/common/types/result";
 import { AgentSession, CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE } from "./agentSession";
+import { createStreamLifecycleMocks } from "./agentSession.testHarness";
 import { createTestHistoryService } from "./testHistoryService";
 
 const TEST_MODEL = "anthropic:claude-3-5-sonnet-latest";
@@ -32,6 +33,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
 
     const streamMessage = mock(() => Promise.resolve(Ok(undefined)));
     const aiService = Object.assign(new EventEmitter(), {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream: mock((_workspaceId: string) => Promise.resolve(Ok(undefined))),
       streamMessage: streamMessage as unknown as AIService["streamMessage"],

src/node/services/agentSession.autoCompaction.test.ts

@@ -19,7 +19,11 @@ import type { BackgroundProcessManager } from "@/node/services/backgroundProcess
 import type { InitStateManager } from "@/node/services/initStateManager";
 import { AgentSession } from "./agentSession";
 import type { CompactionMonitor } from "./compactionMonitor";
-import { createAgentSessionHarness, createStartedTurnHandle } from "./agentSession.testHarness";
+import {
+  createAgentSessionHarness,
+  createStartedTurnHandle,
+  createStreamLifecycleMocks,
+} from "./agentSession.testHarness";
 import { createTestHistoryService } from "./testHistoryService";
 
 describe("AgentSession on-send auto-compaction snapshot deferral", () => {
@@ -824,6 +828,7 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
     });
 
     const aiService = Object.assign(aiEmitter, {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream: mock((_workspaceId: string) => Promise.resolve(Ok(undefined))),
       streamMessage: streamMessage as unknown as (
@@ -944,6 +949,7 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
       Promise.resolve(Ok(createStartedTurnHandle()))
     );
     const aiService = Object.assign(aiEmitter, {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream: mock((_workspaceId: string) => Promise.resolve(Ok(undefined))),
       streamMessage: streamMessage as unknown as (
@@ -1053,6 +1059,7 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
     });
 
     const aiService = Object.assign(aiEmitter, {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream,
       streamMessage: streamMessage as unknown as (
@@ -1201,6 +1208,7 @@ describe("AgentSession on-send auto-compaction snapshot deferral", () => {
     });
 
     const aiService = Object.assign(aiEmitter, {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream,
       streamMessage: streamMessage as unknown as (

src/node/services/agentSession.continueMessageAgentId.test.ts

@@ -4,6 +4,7 @@ import type { CompactionFollowUpRequest, MuxMessage } from "@/common/types/messa
 import type { FilePart, SendMessageOptions } from "@/common/orpc/types";
 import type { Config } from "@/node/config";
 import { AgentSession } from "./agentSession";
+import { createStreamLifecycleMocks } from "./agentSession.testHarness";
 import type { AIService } from "./aiService";
 import type { BackgroundProcessManager } from "./backgroundProcessManager";
 import type { InitStateManager } from "./initStateManager";
@@ -104,6 +105,7 @@ function createAiService(): AIService {
     off() {
       return this;
     },
+    ...createStreamLifecycleMocks(),
     isStreaming: () => false,
     stopStream: mock(() => Promise.resolve({ success: true as const, data: undefined })),
   } as unknown as AIService;

src/node/services/agentSession.disposeRace.test.ts

@@ -17,7 +17,7 @@ import {
   startAbandonedBranchSummaryInBackground,
   type BranchSummaryAiService,
 } from "./branchSummary";
-import { createAgentSessionHarness } from "./agentSession.testHarness";
+import { createAgentSessionHarness, createStreamLifecycleMocks } from "./agentSession.testHarness";
 import type { StreamMessageOptions } from "./aiService";
 import type { TurnCompletion } from "./streamManager";
 
@@ -39,6 +39,7 @@ describe("AgentSession disposal race conditions", () => {
     const streamMessage = mock(() => Promise.resolve(Ok(undefined)));
 
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
         aiHandlers.set(String(eventName), listener);
         return this;
@@ -136,6 +137,7 @@ describe("AgentSession disposal race conditions", () => {
   test("bails out of a send parked on the branch-summary await when removal disposes the session", async () => {
     const streamMessage = mock(() => Promise.resolve(Ok(undefined)));
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(_eventName: string | symbol, _listener: (...args: unknown[]) => void) {
         return this;
       },
@@ -254,6 +256,7 @@ describe("AgentSession disposal race conditions", () => {
     const aiHandlers = new Map<string, (...args: unknown[]) => void>();
 
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
         aiHandlers.set(String(eventName), listener);
         return this;
@@ -340,6 +343,7 @@ describe("AgentSession disposal race conditions", () => {
     const aiHandlers = new Map<string, (...args: unknown[]) => void>();
 
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
         aiHandlers.set(String(eventName), listener);
         return this;
@@ -434,6 +438,7 @@ describe("AgentSession disposal race conditions", () => {
 
   test("does not reset auto-retry intent for synthetic or rejected sends", async () => {
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(_eventName: string | symbol, _listener: (...args: unknown[]) => void) {
         return this;
       },
@@ -578,6 +583,7 @@ describe("AgentSession disposal race conditions", () => {
 
   test("preserves synthetic flag when flushing queued messages", () => {
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(_eventName: string | symbol, _listener: (...args: unknown[]) => void) {
         return this;
       },

src/node/services/agentSession.editMessageId.test.ts

@@ -8,7 +8,7 @@ import { createMuxMessage } from "@/common/types/message";
 import { Ok } from "@/common/types/result";
 import { AgentSession } from "./agentSession";
 import { createTestHistoryService } from "./testHistoryService";
-import { createStartedTurnHandle } from "./agentSession.testHarness";
+import { createStartedTurnHandle, createStreamLifecycleMocks } from "./agentSession.testHarness";
 
 type StreamMessageHandler = AIService["streamMessage"];
 
@@ -42,6 +42,7 @@ describe("AgentSession.sendMessage (editMessageId)", () => {
 
     const streamMessage = mock(streamHandler);
     const aiService = Object.assign(new EventEmitter(), {
+      ...createStreamLifecycleMocks(),
       isStreaming: mock((_workspaceId: string) => false),
       stopStream: mock((_workspaceId: string) => Promise.resolve(Ok(undefined))),
       streamMessage: streamMessage as unknown as AIService["streamMessage"],

src/node/services/agentSession.fileChangeNotification.test.ts

@@ -12,7 +12,7 @@ import type { AIService, StreamMessageOptions } from "./aiService";
 import type { BackgroundProcessManager } from "./backgroundProcessManager";
 import type { InitStateManager } from "./initStateManager";
 import { createTestHistoryService } from "./testHistoryService";
-import { createStartedTurnHandle } from "./agentSession.testHarness";
+import { createStartedTurnHandle, createStreamLifecycleMocks } from "./agentSession.testHarness";
 
 /**
  * Log purity: externally-edited files must produce a durable <system-file-update>
@@ -51,6 +51,7 @@ describe("AgentSession file-change notification (turn start)", () => {
       return Promise.resolve(Ok(createStartedTurnHandle()));
     });
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on: mock(() => aiService),
       off: mock(() => aiService),
       stopStream: mock(() => Promise.resolve(Ok(undefined))),

src/node/services/agentSession.memoryContext.test.ts

@@ -8,6 +8,7 @@ import type { Config } from "@/node/config";
 import type { AIService } from "./aiService";
 import type { MemorySessionContext } from "./memoryService";
 import { AgentSession } from "./agentSession";
+import { createStreamLifecycleMocks } from "./agentSession.testHarness";
 import type { BackgroundProcessManager } from "./backgroundProcessManager";
 import type { HistoryService } from "./historyService";
 import type { InitStateManager } from "./initStateManager";
@@ -28,6 +29,7 @@ function createSession(args: {
 }): AgentSession {
   const aiEmitter = new EventEmitter();
   const aiService: AIService = {
+    ...createStreamLifecycleMocks(),
     on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
       aiEmitter.on(String(eventName), listener);
       return this;

src/node/services/agentSession.postCompactionAttachments.test.ts

@@ -14,6 +14,7 @@ import type { Config } from "@/node/config";
 
 import type { AIService } from "./aiService";
 import { AgentSession } from "./agentSession";
+import { createStreamLifecycleMocks } from "./agentSession.testHarness";
 import type { BackgroundProcessManager } from "./backgroundProcessManager";
 import type { HistoryService } from "./historyService";
 import type { InitStateManager } from "./initStateManager";
@@ -104,6 +105,7 @@ function getAttachmentTypes(
 function createSessionForHistory(historyService: HistoryService, sessionDir: string): AgentSession {
   const aiEmitter = new EventEmitter();
   const aiService: AIService = {
+    ...createStreamLifecycleMocks(),
     on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
       aiEmitter.on(String(eventName), listener);
       return this;

src/node/services/agentSession.postCompactionRefresh.test.ts

@@ -8,7 +8,7 @@ import { createTestHistoryService } from "./testHistoryService";
 import type { CompactionCompletionMetadata } from "@/common/types/compaction";
 import { createMuxMessage } from "@/common/types/message";
 import type { StreamEndEvent } from "@/common/types/stream";
-import { createAgentSessionHarness } from "./agentSession.testHarness";
+import { createAgentSessionHarness, createStreamLifecycleMocks } from "./agentSession.testHarness";
 
 // NOTE: These tests focus on the event wiring (tool-call-end -> callback).
 // The actual post-compaction state computation is covered elsewhere.
@@ -173,6 +173,7 @@ describe("AgentSession post-compaction refresh trigger", () => {
     const handlers = new Map<string, (...args: unknown[]) => void>();
 
     const aiService: AIService = {
+      ...createStreamLifecycleMocks(),
       on(eventName: string | symbol, listener: (...args: unknown[]) => void) {
         handlers.set(String(eventName), listener);
         return this;