diff --git a/apps/server/src/mcp/McpSessionRegistry.test.ts b/apps/server/src/mcp/McpSessionRegistry.test.ts index a3f74c3e28c24..a90b0d9a88cfe 100644 --- a/apps/server/src/mcp/McpSessionRegistry.test.ts +++ b/apps/server/src/mcp/McpSessionRegistry.test.ts @@ -59,6 +59,26 @@ it.effect("stores only a token hash, resolves the bearer token, and revokes by t }), ); +it.effect("keeps the request namespace across credential reissue and splits it per thread", () => + Effect.gen(function* () { + const registry = yield* makeRegistry(() => 1_000); + const request = { + threadId: ThreadId.make("thread-restarted"), + providerInstanceId: ProviderInstanceId.make("codex"), + }; + const scopeOf = (issued: { readonly config: { readonly authorizationHeader: string } }) => + registry.resolve(issued.config.authorizationHeader.replace(/^Bearer\s+/, "")); + const first = yield* scopeOf(yield* registry.issue(request)); + const second = yield* scopeOf(yield* registry.issue(request)); + const other = yield* scopeOf( + yield* registry.issue({ ...request, threadId: ThreadId.make("thread-other") }), + ); + expect(second?.thread.providerSessionId).not.toBe(first?.thread.providerSessionId); + expect(second?.requestNamespace).toBe(first?.requestNamespace); + expect(other?.requestNamespace).not.toBe(first?.requestNamespace); + }), +); + it.effect("always grants pull-requests and gates browser and device access independently", () => Effect.gen(function* () { const registry = yield* makeRegistry(() => 1_000); diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index b16529924a430..c4fe871ae8e2d 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -113,7 +113,8 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( const browserToolsAvailable = request.browserToolsAvailable ?? true; const scope: McpInvocationContext.McpThreadInvocationScope = { environmentId, - requestNamespace: providerSessionId, + // Retrying an accepted write after a restart must reuse its receipt. + requestNamespace: `thread:${encodeURIComponent(request.threadId)}:${encodeURIComponent(request.providerInstanceId)}`, thread: { threadId: ThreadId.make(request.threadId), providerSessionId, diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 29a8305358011..c6b819682aa11 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -101,6 +101,97 @@ it.effect("attributes a launched thread's first message to the calling thread", }), ); +it.effect( + "a launch retried with the same clientRequestId keeps its command after session rotation", + () => + Effect.gen(function* () { + const sourceThreadId = ThreadId.make("source-thread"); + const projectId = ProjectId.make("project"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const modelSelection = { instanceId: providerInstanceId, model: "gpt-5" }; + const caller = { + id: sourceThreadId, + projectId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + activeRunId: "active-run", + archivedAt: null, + deletedAt: null, + } as OrchestrationV2ThreadShell; + const launches: Array = []; + const launch = (clientRequestId: string | undefined, requestNamespace: string) => + Effect.gen(function* () { + const dependencies = Layer.mergeAll( + ServerSettings.layerTest(), + NodeCrypto.layer, + Layer.succeed(McpInvocationContext.McpInvocationContext, { + environmentId: EnvironmentId.make("environment"), + requestNamespace, + thread: { + threadId: sourceThreadId, + providerSessionId: "session", + providerInstanceId, + }, + client: undefined, + issuedAt: 0, + capabilities: new Set(["orchestration" as const]), + }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(caller), + }), + Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => { + launches.push(input); + return Effect.succeed({ + threadId: input.threadId, + projection: { + thread: { id: input.threadId, projectId, modelSelection }, + runs: [], + }, + resumed: false, + } as unknown as ThreadLaunch.ThreadLaunchResult); + }, + }), + Layer.mock(Project.ProjectService)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + }), + Layer.mock(GitVcsDriver.GitVcsDriver)({}), + NodeServices.layer, + ServerConfig.layerTest(process.cwd(), { prefix: "t3-launch-retry-" }).pipe( + Layer.provide(NodeServices.layer), + ), + ); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe( + Layer.provide(dependencies), + ), + ), + ); + yield* toolkit + .handle("t3_thread_launch", { + title: "Audit", + message: "Review the change", + ...(clientRequestId === undefined ? {} : { clientRequestId }), + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + }); + // The namespace is stable per thread, so a rotated session still matches. + yield* launch("retry-launch", "thread:source-thread:codex"); + yield* launch("retry-launch", "thread:source-thread:codex"); + expect(launches[1]?.commandId).toBe(launches[0]?.commandId); + expect(launches[1]?.threadId).toBe(launches[0]?.threadId); + expect(launches[1]?.initialMessage?.messageId).toBe(launches[0]?.initialMessage?.messageId); + yield* launch("retry-launch", "thread:other-thread:codex"); + yield* launch(undefined, "thread:source-thread:codex"); + yield* launch(undefined, "thread:source-thread:codex"); + expect(new Set(launches.map((input) => input.threadId)).size).toBe(4); + }), +); + it.effect("launches a scratch thread into the Scratch project", () => Effect.gen(function* () { const sourceThreadId = ThreadId.make("source-thread"); diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index ca47192c98e4b..4832c7e6e1902 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -1,4 +1,10 @@ -import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + ThreadId, + OrchestratorMcpFailure, + ProjectId, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; @@ -57,7 +63,12 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { Effect.gen(function* () { const context = yield* readCaller(); const { caller } = context; - const commandId = yield* newCommandId(); + const commandId = + input.clientRequestId === undefined + ? yield* newCommandId() + : CommandId.make( + `command:mcp:${encodeURIComponent(context.scope.requestNamespace)}:thread-launch:${encodeURIComponent(input.clientRequestId)}`, + ); const threadId = ThreadId.make(commandId); const messageId = MessageId.make(commandId); const attachments = input.attachments ?? []; diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index e455c041c4c31..85f04c04d79d4 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -104,8 +104,9 @@ const ProjectCloneTool = Tool.make("t3_project_clone", { const ThreadLaunchTool = Tool.make("t3_thread_launch", { ...shared, description: - 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. For upstream commits, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Each call creates a new launch with no retry key; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.', + 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. For upstream commits, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Pass a stable clientRequestId to retry the same launch after errors or restarts; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.', parameters: Schema.Struct({ + clientRequestId: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(256))), projectId: Schema.optional(ProjectId), scratch: Schema.optional( Schema.Boolean.annotate({ diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index a16df458e2043..c257a63315281 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3458,6 +3458,154 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each(["active-goal", "no-goal"] as const)( + "a continuation Codex rejects as EmptyInput retries only on an active goal (%s)", + (state) => + Effect.scoped( + Effect.gen(function* () { + const scenario = "codex-restart-empty-input"; + const nativeThreadId = "native-restart-empty-input"; + const nativeTurnId = "turn-restart-empty-input"; + const prompt = "Continue where you left off."; + const preamble = codexReplayPreamble({ + nativeThreadId, + nativeTurnId, + prompt: "unused", + }).slice(0, 5); + const turnParams = { + threadId: nativeThreadId, + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + summary: "detailed", + }; + const transcript = makeCodexReplayTranscript({ + scenario, + entries: [ + ...preamble, + { + type: "expect_outbound", + label: "resume", + frame: { + id: 3, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + label: "resume", + frame: { id: 3, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + { + type: "expect_outbound", + label: "empty continue", + frame: { id: 4, method: "turn/start", params: { ...turnParams, input: [] } }, + }, + { + type: "emit_inbound", + label: "empty continue", + frame: { + id: 4, + error: { code: -32603, message: "failed to submit turn input: EmptyInput" }, + }, + }, + { + type: "expect_outbound", + label: "read goal", + frame: { id: 5, method: "thread/goal/get", params: { threadId: nativeThreadId } }, + }, + { + type: "emit_inbound", + label: "read goal", + frame: { + id: 5, + result: { + goal: + state === "no-goal" + ? null + : { + threadId: nativeThreadId, + objective: "Finish the work", + status: "active", + tokenBudget: null, + tokensUsed: 10, + timeUsedSeconds: 3, + createdAt: 1782622440, + updatedAt: 1782622450, + }, + }, + }, + }, + ...(state === "active-goal" + ? [ + { + type: "expect_outbound" as const, + label: "prompted continue", + frame: { + id: 6, + method: "turn/start", + params: { ...turnParams, input: [{ type: "text", text: prompt }] }, + }, + }, + { + type: "emit_inbound" as const, + label: "prompted continue", + frame: { + id: 6, + result: { + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + }, + }, + }, + ] + : []), + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + const resumed = yield* harness.runtime.resumeThread({ + providerThread: harness.providerThread, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + const started = yield* harness.runtime + .startTurn({ + ...makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: resumed, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-restart-empty-input"), + text: prompt, + }), + restartContinuationOfRunId: RunId.make("run-before-restart"), + }) + .pipe(Effect.result); + if (state === "active-goal") { + assert.equal(started._tag, "Success"); + return; + } + assert.equal(started._tag, "Failure"); + if (started._tag === "Failure") { + assert.nestedPropertyVal( + started.failure, + "cause.errorMessage", + "failed to submit turn input: EmptyInput", + ); + } + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("resolves retryable app-server errors on resumed provider activity", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 19d1e7b7ffe29..a38ec89ac5bf6 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6614,11 +6614,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( : null; if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand); return Effect.gen(function* () { - yield* startNativeTurn( - turnInput, + const input = turnInput.restartContinuationOfRunId === undefined ? yield* toCodexInput(turnInput) - : [], + : []; + yield* startNativeTurn(turnInput, input).pipe( + Effect.catchTags({ + CodexAppServerRequestError: (cause) => { + // Codex rejects an empty continuation while a goal turn is + // still active. Send the continue prompt as real input. + if ( + input.length !== 0 || + cause.code !== -32603 || + cause.errorMessage !== "failed to submit turn input: EmptyInput" + ) + return Effect.fail(cause); + return Effect.gen(function* () { + const current = yield* client + .request("thread/goal/get", { + threadId: yield* getNativeThreadId(turnInput.providerThread), + }) + .pipe(Effect.orElseSucceed(() => null)); + if (current?.goal?.status !== "active") return yield* Effect.fail(cause); + yield* startNativeTurn(turnInput, yield* toCodexInput(turnInput)); + }); + }, + }), ); }).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce388..886527a8a0354 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -12,6 +12,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunId, RuntimeRequestId, ThreadId, @@ -966,3 +967,82 @@ it.effect("marks fork descendants unreadable when their source is missing or cor ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("surfaces steering still queued in the outbox and keeps its ids across run upserts", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const now = yield* DateTime.now; + const threadId = yield* createThread("steering"); + const runId = yield* createRun(threadId, "running"); + const steer = (name: string) => { + const messageId = MessageId.make(`message:steering:${name}`); + return { + messageId, + effect: { + id: `effect:steering:${name}`, + commandId: CommandId.make(`command:steering:${name}`), + threadId, + request: { + type: "provider-turn.steer" as const, + providerSessionId: ProviderSessionId.make("session:steering"), + providerThreadId: ProviderThreadId.make("provider-thread:steering"), + providerTurnId: ProviderTurnId.make("provider-turn:steering"), + messageId, + }, + }, + }; + }; + const delivered = steer("delivered"); + const queued = steer("queued"); + for (const { messageId } of [delivered, queued]) { + yield* projections.apply({ + id: EventId.make(`event:${messageId}`), + type: "message.updated", + threadId, + runId, + driver, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: messageId, + threadId, + runId, + nodeId: null, + role: "user", + text: "steer", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + } + yield* outbox.enqueue([delivered.effect]); + const claimed = yield* outbox.claimNext({ workerId: "worker:steering", leaseDurationMs: 1000 }); + assert.isTrue( + yield* outbox.succeed({ effectId: delivered.effect.id, workerId: "worker:steering" }), + ); + assert.isTrue(claimed._tag === "Some"); + yield* outbox.enqueue([queued.effect]); + + const before = yield* projections.getRuntimeRecoveryProjection(threadId); + assert.deepEqual(before.pendingSteering, [{ runId, messageId: queued.messageId }]); + + const run = before.runs[0]!; + const apply = (payload: OrchestrationV2Run, name: string) => + projections.apply({ + id: EventId.make(`event:steering:${name}`), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload, + }); + yield* apply({ ...run, restartPendingMessageIds: [queued.messageId] }, "record"); + yield* apply({ ...run, status: "cancelled", completedAt: now }, "later-upsert"); + const after = yield* projections.getRuntimeRecoveryProjection(threadId); + assert.deepEqual(after.runs[0]?.restartPendingMessageIds, [queued.messageId]); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..1ec3063c22463 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -269,7 +269,13 @@ export type ProjectionRuntimeRecoveryState = Pick< | "runtimeRequests" | "messages" | "turnItems" ->; +> & { + /** Steering whose delivery was still queued in the outbox, per run. */ + readonly pendingSteering?: ReadonlyArray<{ + readonly runId: RunId; + readonly messageId: MessageId; + }>; +}; export type ProjectionPendingUserInputs = Pick< OrchestrationV2ThreadProjection, @@ -665,6 +671,10 @@ function preserveRunRecordedFields( ...(next.delegatedCompletion === undefined && current.delegatedCompletion !== undefined ? { delegatedCompletion: current.delegatedCompletion } : {}), + ...(next.restartPendingMessageIds === undefined && + current.restartPendingMessageIds !== undefined + ? { restartPendingMessageIds: current.restartPendingMessageIds } + : {}), ...(next.restartCancelledBackgroundWork === undefined && current.restartCancelledBackgroundWork !== undefined ? { restartCancelledBackgroundWork: current.restartCancelledBackgroundWork } @@ -1970,7 +1980,10 @@ export const layer: Layer.Layer = requested_at = excluded.requested_at, completed_at = excluded.completed_at, payload_json = ${keepRecordedRunField( - keepRecordedRunField(sql`excluded.payload_json`, "$.delegatedCompletion"), + keepRecordedRunField( + keepRecordedRunField(sql`excluded.payload_json`, "$.delegatedCompletion"), + "$.restartPendingMessageIds", + ), "$.restartCancelledBackgroundWork", )} `; @@ -4368,7 +4381,27 @@ export const layer: Layer.Layer = decodeRows(decodeMessagePayload, threadId)(messageRows), decodeRows(decodeTurnItemPayload, threadId)(turnItemRows), ]); + // Delegated results are redelivered by their own cohort, not by the prompt. + const pendingSteering = yield* sql<{ + readonly run_id: string; + readonly message_id: string; + }>` + SELECT message.run_id, message.message_id + FROM orchestration_v2_effect_outbox AS effect + JOIN orchestration_v2_projection_messages AS message + ON message.thread_id = effect.thread_id + AND message.message_id = json_extract(effect.payload_json, '$.messageId') + WHERE effect.thread_id = ${threadId} AND effect.effect_type = 'provider-turn.steer' + AND effect.status IN ('pending', 'running') + AND message.run_id IS NOT NULL + AND json_extract(message.payload_json, '$.delegatedCompletion') IS NULL + ORDER BY effect.rowid ASC + `; return { + pendingSteering: pendingSteering.map((row) => ({ + runId: RunId.make(row.run_id), + messageId: MessageId.make(row.message_id), + })), thread, runs, attempts, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index e02a65a6a5564..6b7b905a23e72 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -642,6 +642,79 @@ it.effect("closes a secret request card's form when its run is recovered", () => }).pipe(Effect.provide(layer)); }); +it.effect("records undelivered steering on the cancelled run and ends cut streams", () => { + const threadId = ThreadId.make("thread_steer_recovery"); + const runId = RunId.make("run_steer_recovery"); + const steerId = MessageId.make("message_steer_recovery"); + let committedInput: Parameters[0] | null = + null; + const projection = { + thread: { id: threadId }, + pendingSteering: [{ runId, messageId: steerId }], + runtimeRequests: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runs: [{ id: runId, status: "running", providerInstanceId: ProviderInstanceId.make("codex") }], + attempts: [], + nodes: [], + subagents: [], + messages: [], + turnItems: ["assistant_message", "reasoning", "proposed_plan"].map((type) => ({ + id: `turn-item:${type}`, + threadId, + runId, + nodeId: null, + type, + status: "running", + streaming: true, + })), + } as unknown as OrchestrationV2ThreadProjection; + const layer = ProviderRuntimeRecovery.layer.pipe( + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.mergeAll( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([threadId]), + getRuntimeRecoveryProjection: () => Effect.succeed(projection), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (input) => { + committedInput = input; + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ + runRecoveryOnce: Effect.succeed(false), + }), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + listByCommandId: () => Effect.succeed([]), + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + + return Effect.gen(function* () { + yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup"); + const events = committedInput?.events ?? []; + const run = events.find((event) => event.type === "run.updated"); + assert.equal(run?.type === "run.updated" ? run.payload.status : null, "cancelled"); + assert.deepEqual(run?.type === "run.updated" ? run.payload.restartPendingMessageIds : null, [ + steerId, + ]); + const items = events.flatMap((event) => + event.type === "turn-item.updated" ? [event.payload] : [], + ); + assert.lengthOf(items, 3); + for (const item of items) { + assert.equal(item.status, "cancelled"); + assert.isFalse("streaming" in item ? item.streaming : true); + } + }).pipe(Effect.provide(layer)); +}); + it.effect.each(["queued", "preparing"] as const)("recovers $0 work after restart", (status) => { const threadId = ThreadId.make("thread_queued_restart"); const runId = RunId.make("run_queued_restart"); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 32de3b9d6a4f4..8d01372176583 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -149,7 +149,18 @@ const cancelledItem = ( ): OrchestrationV2ThreadProjection["turnItems"][number] => item.type === "secret_request" && item.secretStatus === "pending" ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now } - : { ...item, status: "cancelled", completedAt: now, updatedAt: now }; + : { + ...item, + status: "cancelled", + completedAt: now, + updatedAt: now, + // A cut stream never gets its closing event. + ...(item.type === "reasoning" || + item.type === "assistant_message" || + item.type === "proposed_plan" + ? { streaming: false } + : {}), + }; function latestStartedRun( projection: ProjectionStore.ProjectionRuntimeRecoveryState, @@ -353,6 +364,16 @@ export const make = Effect.gen(function* () { }); } for (const run of runs) { + // Steering still queued in the outbox died with the process; the + // continuation prompt carries it. + const undeliveredSteering = [ + ...new Set([ + ...(run.restartPendingMessageIds ?? []), + ...(projection.pendingSteering ?? []) + .filter((steer) => steer.runId === run.id) + .map((steer) => steer.messageId), + ]), + ]; events.push({ id: yield* allocateEventId(), type: "run.updated", @@ -360,7 +381,15 @@ export const make = Effect.gen(function* () { runId: run.id, providerInstanceId: run.providerInstanceId, occurredAt: now, - payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now }, + payload: { + ...run, + status: "cancelled", + queuePosition: null, + completedAt: now, + ...(undeliveredSteering.length === 0 + ? {} + : { restartPendingMessageIds: undeliveredSteering }), + }, }); for (const attempt of projection.attempts.filter( (candidate) => diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 5c1a27cb47373..1c34f3244cb9c 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -937,12 +937,18 @@ export const layer: Layer.Layer< }); // Delivered once: this run's provider turn marks the work as told. A // restart continuation is prompted by its own text or resumes natively. - const noteContinuation = isRestartNoteContinuation( - run, - projection.runs, - projection.providerTurns, - projection.attempts, - ); + const promptedContinuation = + isRestartNoteContinuation( + run, + projection.runs, + projection.providerTurns, + projection.attempts, + ) || + projection.runs.some( + (source) => + source.id === run.restartContinuationOfRunId && + (source.restartPendingMessageIds?.length ?? 0) > 0, + ); const restartCancelledWork = pendingRestartCancelledBackgroundWork({ runs: projection.runs, providerTurns: projection.providerTurns, @@ -1152,7 +1158,7 @@ export const layer: Layer.Layer< .join("\n\n"); const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput; yield* start({ - ...(noteContinuation ? promptedInput : turnInput), + ...(promptedContinuation ? promptedInput : turnInput), message: { ...turnInput.message, text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`, @@ -1199,7 +1205,7 @@ export const layer: Layer.Layer< effectiveHandoffs.length === 0 && missedItems.length === 0 && restartNote === "" && - !noteContinuation + !promptedContinuation ? session : makeDeliverySession(session, startWithHandoffs); return runExecution.startRootRun({ diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 60f55d8689ec4..3f6c8208d3481 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -1066,6 +1066,61 @@ it.effect("shows a continuation as a T3 Code notice whose detail is the prompt", }), ); +it.effect("carries steering the provider never received into the continuation", () => + Effect.gen(function* () { + const steerId = MessageId.make("message:steer"); + const contextSteerId = MessageId.make("message:context-steer"); + const attachment = { + type: "file" as const, + id: "attachment:steer", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 12, + }; + const [resumed] = yield* continuationDispatches({ + ...cutMidTurn({ restartPendingMessageIds: [steerId, contextSteerId] }), + messages: [ + { id: steerId, text: "Also update the docs.", attachments: [attachment] }, + { + id: contextSteerId, + text: "Check [log](t3-context://v1/terminal/ctx_terminal).", + attachments: [], + context: { + version: 1, + records: [ + { + version: 1, + contextId: "ctx_terminal", + kind: "terminal", + label: "Terminal 1 lines 3-4", + terminalId: "term-1", + terminalLabel: "Terminal 1", + lineStart: 3, + lineEnd: 4, + text: "STATUS=queued\nMODE=restart", + }, + ], + }, + }, + { id: MessageId.make("message:unrelated"), text: "Not carried.", attachments: [] }, + ], + } as unknown as OrchestrationV2ThreadProjection); + assert.include( + resumed?.text ?? "", + "Continue where you left off.\n\nAlso update the docs.\n\nCheck [Terminal: log; ref=ctx_terminal].", + ); + assert.include(resumed?.text ?? "", ''); + assert.include( + resumed?.text ?? "", + "terminal: Terminal 1\n3 | STATUS=queued\n4 | MODE=restart", + ); + assert.notInclude(resumed?.text ?? "", 'unavailable="true"'); + assert.notInclude(resumed?.text ?? "", "Not carried."); + assert.deepEqual(resumed?.attachments, [attachment]); + assert.isUndefined(resumed?.notification); + }), +); + it.effect( "carries the note forward when its continuation was cut before reaching the provider", () => @@ -1104,6 +1159,86 @@ it.effect( }), ); +it.effect("continues every thread that shares one provider session", () => + Effect.gen(function* () { + const secondThreadId = ThreadId.make("thread:restart-second"); + const secondRunId = RunId.make("run:restart-second"); + const secondProviderThreadId = ProviderThreadId.make("provider-thread:restart-second"); + const secondAttemptId = RunAttemptId.make("attempt:restart-second"); + const commits: Array[0]> = []; + // Recovering the first thread stops the session they share. + const read = (id: ThreadId) => { + const base = makeProjection(); + const session = { + ...base.providerSessions[0]!, + status: commits.length === 0 ? ("ready" as const) : ("stopped" as const), + }; + return id === threadId + ? { ...base, providerSessions: [session] } + : { + ...base, + thread: { ...base.thread, id: secondThreadId }, + runs: [ + { + ...base.runs[0]!, + id: secondRunId, + providerThreadId: secondProviderThreadId, + activeAttemptId: secondAttemptId, + }, + ], + providerThreads: [ + { + ...base.providerThreads[0]!, + id: secondProviderThreadId, + appThreadId: secondThreadId, + }, + ], + providerSessions: [session], + providerTurns: [ + { + ...base.providerTurns[0]!, + id: ProviderTurnId.make("turn:restart-second"), + providerThreadId: secondProviderThreadId, + runAttemptId: secondAttemptId, + }, + ], + }; + }; + const recovery = yield* ProviderRuntimeRecovery.make.pipe( + Effect.provide( + Layer.mergeAll( + ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([threadId, secondThreadId]), + getRuntimeRecoveryProjection: (id) => Effect.succeed(read(id)), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (input) => { + commits.push(input); + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ + runRecoveryOnce: Effect.succeed(false), + }), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + yield* recovery.reconcile("startup"); + assert.deepEqual( + commits.flatMap((commit) => commit.effects.map((effect) => effect.request)), + [ + { type: "provider-runtime.continue", sourceRunId: runId }, + { type: "provider-runtime.continue", sourceRunId: secondRunId }, + ], + ); + }), +); + it.effect("prepares later threads' continuations when one thread fails", () => Effect.gen(function* () { const brokenThreadId = ThreadId.make("thread:broken"); diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 3339133c9a969..aec28b0197c29 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -1,5 +1,6 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; +import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { CommandId, latestProviderTurnForAttempt, @@ -246,7 +247,7 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta input.threadId, ["messages", "turnItems"], { - messageIds: [source.userMessageId], + messageIds: [source.userMessageId, ...(source.restartPendingMessageIds ?? [])], turnItemRunIds: [source.id], turnItemTypes: ["run_interrupt_request"], }, @@ -261,6 +262,19 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta (message) => message.id === source.userMessageId, ); if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return; + // Steering the provider never received before the process ended. + const pendingMessages = (source.restartPendingMessageIds ?? []).flatMap((id) => + sourceRecords.messages.filter((message) => message.id === id), + ); + const prompt = [ + CONTINUE_PROMPT, + ...pendingMessages.map((message) => + projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }), + ), + ].join("\n\n"); const note = restartContinuationNote( source, projection.runs, @@ -271,10 +285,10 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work); const text = noteText === undefined - ? CONTINUE_PROMPT - : note.settled + ? prompt + : note.settled && pendingMessages.length === 0 ? noteText - : `${noteText}\n\n${CONTINUE_PROMPT}`; + : `${noteText}\n\n${prompt}`; yield* threads .dispatch({ type: "message.dispatch", @@ -282,15 +296,21 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta threadId: input.threadId, messageId, text, - notification: { - source: { kind: "system" }, - outcome: "updated", - summary: note.settled - ? "T3 Code restarted and stopped background work" - : "T3 Code restarted and resumed this turn", - detail: text, - }, - attachments: [], + // Shown as a work log row; the prompt the agent got is its detail. + // Carried steering stays a visible message so its attachments show. + ...(pendingMessages.length === 0 + ? { + notification: { + source: { kind: "system" as const }, + outcome: "updated" as const, + summary: note.settled + ? "T3 Code restarted and stopped background work" + : "T3 Code restarted and resumed this turn", + detail: text, + }, + } + : {}), + attachments: pendingMessages.flatMap((message) => message.attachments), modelSelection: source.modelSelection, dispatchMode: { type: "start_immediately" }, createdBy: "agent", diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index 5260ea989671e..19f566be13634 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -93,9 +93,17 @@ or that recovery closed on the cut turn. Message-capable questions stay answerab they do not block. Delivery also stops after three consecutive continuations in one `restartContinuationOfRunId` chain, so a turn that crashes the server cannot loop. Stable command and message IDs prevent duplicate submissions after an outbox retry. Codex -resumes without adding provider prompt text unless the turn lost background work; +resumes without adding provider prompt text unless the turn lost background work or steering; other adapters receive the continuation message through their normal turn path. +Steering is process-bound, so a steer still queued when the process ended never reached +the provider. Recovery records its message ids on the cancelled run +(`restartPendingMessageIds`) and the continuation appends their text and attachments to +its prompt. This carries them one hop: a second restart before delivery loses them, and a +steer delivered just before the crash may be repeated once. When Codex rejects the empty +continuation input while a goal turn is still active, the adapter reads the goal and, if +it is active, retries with the continue prompt as real input. + Delegated tasks (`delegate_task` child threads) are reconciled as their own threads, never as the parent's background work. The orchestrator settles child results and completion deliveries in a startup pass after reconciliation, because the terminal-run diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..04812cf27fd20 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -536,6 +536,8 @@ export const OrchestrationV2Run = Schema.Struct({ checkpointId: Schema.NullOr(CheckpointId), contextHandoffId: Schema.NullOr(ContextHandoffId), restartContinuationOfRunId: Schema.optional(RunId), + /** Steering input lost with a provider process, carried into the restart prompt. */ + restartPendingMessageIds: Schema.optional(Schema.Array(MessageId)), delegatedTaskId: Schema.optional(NodeId), workStartedAt: Schema.optional(Schema.DateTimeUtc), restartCancelledBackgroundWork: Schema.optional( diff --git a/packages/provider-core/src/server/orchestrationInstructions.ts b/packages/provider-core/src/server/orchestrationInstructions.ts index 49d6d5440bdc0..67a55b771d587 100644 --- a/packages/provider-core/src/server/orchestrationInstructions.ts +++ b/packages/provider-core/src/server/orchestrationInstructions.ts @@ -28,7 +28,7 @@ For stacked work, set \`baseRef\` to the intended parent branch and \`startFromO \`t3_thread_launch\` is the single-thread launch tool. Use \`create_threads\` only for a batch of threads intentionally sharing the caller's checkout: it always inherits the caller's project, branch, and worktree and has no workspace override. Asking an agent to run \`git worktree add\` or \`cd\` in its prompt does not update T3's thread binding. Select the workspace in the launch call instead. \`t3_worktree_handoff\` moves the calling thread, not another thread, and cannot move a thread already attached to a worktree. -\`t3_thread_launch\` has no idempotency key. Retain its returned threadId and inspect it with \`t3_thread_read\` / \`t3_thread_wait\`; preparation can still be running after acceptance. If a launch fails or its response is lost, inspect \`t3_thread_list\` before retrying, since a thread may already exist.`, +Reuse a stable \`clientRequestId\` when retrying \`t3_thread_launch\`. Retain its returned threadId and inspect it with \`t3_thread_read\` / \`t3_thread_wait\`; preparation can still be running after acceptance. If a launch fails or its response is lost, inspect \`t3_thread_list\` before retrying, since a thread may already exist.`, `Tool names may include a harness-normalized MCP prefix, such as \`mcp__t3_code__delegate_task\`; the semantics are the same. Some harnesses attach optional MCP servers lazily: if an initial tool-catalog scan does not show T3 tools, do not conclude that cross-provider delegation is unavailable. Make one bounded direct attempt using the known T3 tool name on the next tool step. In Codex code mode, for example, call \`tools.mcp__t3_code__orchestrator_capabilities({})\` before reporting that the capability is absent. Keep polling/wait loops bounded, do not duplicate active work, and use stable \`clientRequestId\` values when retrying tools that accept them. ACP fallback: some ACP agents accept the injected MCP server but fail to expose its tools. When the T3 tools are absent and \`T3_ACP_MCP_NODE\` is present, call the same tools through the terminal: \`ELECTRON_RUN_AS_NODE=1 "$T3_ACP_MCP_NODE" \${T3_ACP_MCP_ENTRYPOINT:+"$T3_ACP_MCP_ENTRYPOINT"} acp-mcp-call orchestrator_capabilities '{}'\` (\`T3_ACP_MCP_ENTRYPOINT\` is unset when T3 runs as a standalone executable). Delegate with \`acp-mcp-call delegate_task '{"task":"...","target":{"providerInstanceId":"...","model":"..."},"mode":"async","clientRequestId":"..."}'\`. This is the supported T3 transport fallback, not an ordinary shell-based substitute for delegation.