diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b41e6e0c13345..aa05957a50a95 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -7,6 +7,8 @@ import { import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { + ChatAttachmentId, + ComposerContextId, CommandId, CheckpointId, CodexSettings, @@ -46,9 +48,14 @@ import * as Predicate from "effect/Predicate"; import * as FileSystem from "effect/FileSystem"; import { MCP_APP_OUTPUT_KEY, readMcpAppReference } from "@t3tools/shared/mcpApp"; import { toolOutputImages } from "@t3tools/shared/toolOutput"; -import { resolveAttachmentPathById } from "../../attachmentStore.ts"; +import { + createAttachmentId, + resolveAttachmentPath, + resolveAttachmentPathById, +} from "../../attachmentStore.ts"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -81,6 +88,27 @@ import * as CodexAdapterV2 from "./CodexAdapterV2.ts"; import { withCodexReplayChildMetadata } from "./CodexAdapterV2.testkit.ts"; import * as CodexAdapterV2Testkit from "./CodexAdapterV2.testkit.ts"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlitePersistence from "../../persistence/Sqlite.ts"; +import * as ServerSettings from "../../serverSettings.ts"; +import * as EventSink from "../EventSink.ts"; +import * as EventStore from "../EventStore.ts"; +import * as EffectOutbox from "../EffectOutbox.ts"; +import * as ProjectionStore from "../ProjectionStore.ts"; +import * as ProviderRuntimeRecovery from "../ProviderRuntimeRecoveryService.ts"; +import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; + +const decodeCodexTurnInput = Schema.decodeUnknownEffect( + Schema.Struct({ + input: Schema.Array( + Schema.Union([ + Schema.Struct({ type: Schema.Literal("text"), text: Schema.String }), + Schema.Struct({ type: Schema.Literal("localImage"), path: Schema.String }), + Schema.Struct({ type: Schema.Literal("image"), url: Schema.String }), + ]), + ), + }), +); const replayTranscriptJson = Schema.fromJsonString(CodexReplay.CodexAppServerReplayTranscript); const encodeReplayTranscriptJson = Schema.encodeEffect(replayTranscriptJson); const decodeReplayTranscriptJson = Schema.decodeUnknownEffect(replayTranscriptJson); @@ -4412,6 +4440,637 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each([ + { + scenario: "image and context after one restart", + restarts: 1, + completed: false, + files: false, + }, + { + scenario: "image and context after two restarts", + restarts: 2, + completed: false, + files: false, + }, + { + scenario: "a steer accepted before its provider turn completed", + restarts: 1, + completed: true, + files: false, + }, + { + scenario: "two recovered siblings after the first completes and the second is cut", + restarts: 2, + completed: true, + files: true, + siblingState: "completed", + }, + { + scenario: "user-interrupted sibling before the second is cut", + restarts: 2, + completed: true, + files: true, + siblingState: "interrupted", + }, + ])( + "delivers queued steering to Codex after recovery: $scenario", + ({ scenario, restarts, completed, files, siblingState }) => + Effect.gen(function* () { + const name = "restart-steering-" + scenario.replaceAll(" ", "-"); + const threadId = ThreadId.make(name); + const nativeThreadId = "native-" + name; + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-restart-steering-" }); + const serverConfig = yield* ProviderReplayHarness.makeReplayServerConfig(name).pipe( + Effect.orDie, + ); + yield* Effect.addFinalizer(() => + fs.remove(serverConfig.baseDir, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const attachments = yield* Effect.forEach( + files ? ["first.txt", "second.txt"] : ["steering.png"], + (name) => + Effect.gen(function* () { + const attachment = { + type: files ? ("file" as const) : ("image" as const), + id: ChatAttachmentId.make(createAttachmentId(threadId)!), + name, + mimeType: files ? "text/plain" : "image/png", + sizeBytes: 4, + }; + const path = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + assert.isNotNull(path); + yield* fs.makeDirectory((yield* Path.Path).dirname(path), { recursive: true }); + yield* fs.writeFile(path, new Uint8Array([1, 2, 3, 4])); + return { attachment, path }; + }), + ); + const completeTurns: Array< + ( + turnId: string, + status?: "completed" | "interrupted", + ) => Effect.Effect + > = []; + const interrupts: Array = []; + const sent: Array = []; + let sessions = 0; + const adapter = yield* CodexAdapterV2.makeCodexAdapterV2({ + instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CODEX_SETTINGS, + environment: {}, + digest: yield* Digest.Digest, + fileSystem: fs, + idAllocator: yield* IdAllocator.IdAllocatorV2, + serverConfig, + clientFactory: { + open: (input) => + Effect.gen(function* () { + const firstSession = sessions++ === 0; + const preamble = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused", + prompt: "unused", + }); + const transcript = makeCodexReplayTranscript({ + scenario: name, + entries: firstSession + ? preamble.slice(0, 5) + : [ + ...preamble.slice(0, 3), + { + type: "expect_outbound", + frame: { + id: 2, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + frame: { + id: 2, + result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } }, + }, + }, + ], + }); + const context = yield* Layer.build(CodexReplay.layerReplay(transcript)); + const client = yield* CodexClient.CodexAppServerClient.pipe( + Effect.provide(context), + ); + // Capture real adapter output while stubbing the external Codex response. + const request = ((method, params) => + method === "config/read" + ? Effect.succeed({ config: {}, origins: {} }) + : method === "turn/start" + ? Effect.sync(() => { + sent.push(params); + return { + turn: makeCodexReplayTurn({ + id: "turn-" + sent.length, + status: "inProgress", + }), + }; + }) + : method === "turn/interrupt" + ? Effect.gen(function* () { + interrupts.push(params); + yield* completeTurns.at(-1)!("turn-" + sent.length, "interrupted"); + return {}; + }) + : client.request(method, params)) as typeof client.request; + const handleServerNotification: typeof client.handleServerNotification = ( + method, + handler, + ) => + client.handleServerNotification(method, handler).pipe( + Effect.tap(() => + method === "turn/completed" + ? Effect.sync(() => { + completeTurns.push((turnId, status = "completed") => + handler({ + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: turnId, status }), + } as Parameters[0]), + ); + }) + : Effect.void, + ), + ); + return { ...client, request, handleServerNotification }; + }).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterOpenSessionError({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + providerSessionId: input.providerSessionId, + cause, + }), + ), + ), + }, + }); + let database = SqlitePersistence.layerFromPath(directory + "/state.sqlite").pipe( + Layer.provide(NodeServices.layer), + ); + const snapshot = (file: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.unsafe("VACUUM INTO ?", [file]); + }).pipe(Effect.provide(database)); + const runtime = () => + ProviderReplayHarness.layerWithRegistry( + { name, runtimePolicyOverride: { cwd: directory } }, + ProviderAdapterRegistry.layerSingle(adapter), + { + databaseLayer: database, + runEffectWorker: false, + continueThreadsAfterServerUpdate: true, + }, + ).pipe(Layer.provideMerge(EffectOutbox.layer.pipe(Layer.provide(database)))); + const recovery = () => + ProviderRuntimeRecovery.layer.pipe( + Layer.provide( + EventSink.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + EventStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + IdAllocator.layer, + ).pipe(Layer.provideMerge(database)), + ), + ), + ), + Layer.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true })), + ); + yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(name + ":create"), + threadId, + projectId: ProjectId.make(name), + title: name, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: directory, + createdBy: "user", + creationSource: "web", + }); + const started = yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(name + ":first"), + threadId, + messageId: MessageId.make(name + ":first"), + text: "Start working", + attachments: [], + modelSelection: CODEX_TEST_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(started); + const projection = yield* orchestrator.getThreadProjection(threadId); + for (const [index, { attachment }] of attachments.entries()) { + const steerId = name + ":steer-" + index; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(steerId), + threadId, + messageId: MessageId.make(steerId), + text: + "STEERING_MARKER_" + + index + + " use [terminal](t3-context://v1/terminal/restart-terminal)", + attachments: [attachment], + context: { + version: 1, + records: [ + { + version: 1, + contextId: ComposerContextId.make("restart-terminal"), + kind: "terminal", + label: "terminal", + terminalId: "terminal-" + index, + terminalLabel: "Shell", + lineStart: 1, + lineEnd: 1, + text: files ? "CONTEXT_MARKER_" + index : "CONTEXT_MARKER", + }, + ], + }, + modelSelection: CODEX_TEST_MODEL_SELECTION, + dispatchMode: { type: "steer_active", targetRunId: projection.runs[0]!.id }, + createdBy: "user", + creationSource: "web", + }); + const outbox = yield* EffectOutbox.EffectOutboxV2; + const effects = yield* outbox.listByCommandId(CommandId.make(steerId)); + assert.isTrue( + effects.some( + (effect) => + effect.request.type === "provider-turn.steer" && effect.status === "pending", + ), + ); + } + // No daemon is running: the steer is committed but has not reached the provider. + assert.lengthOf(sent, 1); + if (completed) { + const settled = yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => event.type === "run.updated" && event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* completeTurns.at(-1)!("turn-1"); + yield* Fiber.join(settled); + const finished = yield* orchestrator.getThreadProjection(threadId); + assert.equal(finished.runs[0]?.status, "waiting"); + assert.equal(finished.providerTurns[0]?.status, "completed"); + const outbox = yield* EffectOutbox.EffectOutboxV2; + const checkpoints = yield* outbox.listByCommandId( + CommandId.make("command:effect:checkpoint.capture:" + finished.runs[0]!.id), + ); + assert.isTrue( + checkpoints.some( + (effect) => + effect.request.type === "checkpoint.capture" && effect.status === "pending", + ), + ); + const effects = yield* outbox.listByCommandId(CommandId.make(name + ":steer-0")); + assert.isTrue( + effects.some( + (effect) => + effect.request.type === "provider-turn.steer" && effect.status === "pending", + ), + ); + } + // Snapshot before scope finalizers perform a graceful session shutdown. + yield* snapshot(directory + "/crash-1.sqlite"); + }).pipe(Effect.provide(runtime())), + ); + for (let restart = 1; restart <= restarts; restart++) { + yield* TestClock.adjust("1 milli"); + database = SqlitePersistence.layerFromPath( + directory + "/crash-" + restart + ".sqlite", + ).pipe(Layer.provide(NodeServices.layer)); + yield* Effect.scoped( + ProviderRuntimeRecovery.ProviderRuntimeRecoveryService.use((service) => + service.reconcile("startup"), + ).pipe(Effect.provide(recovery())), + ); + yield* Effect.scoped( + Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const finishLatestTurn = Effect.gen(function* () { + const projection = yield* orchestrator.getThreadProjection(threadId); + const active = projection.runs.find( + (run) => run.status === "starting" || run.status === "running", + ); + assert.isDefined(active); + yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && + event.runId === active!.id && + event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + yield* completeTurns.at(-1)!("turn-" + sent.length); + yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === active!.id && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + ); + }); + if (restart < restarts && siblingState) { + // Settle the source checkpoint, then admit both recovered steers without starting S1. + assert.isTrue(yield* worker.runOnce()); + assert.isTrue(yield* worker.runOnce()); + const split = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + split.runs.map((run) => run.status), + ["completed", "starting", "queued"], + ); + assert.deepEqual( + split.runs.slice(1).map((run) => run.userMessageId), + [MessageId.make(name + ":steer-0"), MessageId.make(name + ":steer-1")], + ); + assert.lengthOf(sent, 1); + yield* worker.drain(); + if (siblingState === "interrupted") { + yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && + event.runId === split.runs[1]!.id && + event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make(name + ":interrupt"), + threadId, + runId: split.runs[1]!.id, + }); + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && + event.runId === split.runs[1]!.id && + event.payload.status === "interrupted", + ), + Stream.take(1), + Stream.runDrain, + ); + assert.deepEqual(interrupts, [{ threadId: nativeThreadId, turnId: "turn-2" }]); + } else { + yield* finishLatestTurn; + } + const secondStarted = yield* orchestrator.streamStoredEventsFrom({ threadId }).pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.runId === split.runs[2]!.id && + event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* worker.drain(); + yield* Fiber.join(secondStarted); + const running = yield* orchestrator.getThreadProjection(threadId); + assert.equal(running.runs[1]?.status, siblingState); + assert.equal(running.runs[2]?.status, "running"); + assert.equal(running.providerTurns[1]?.status, siblingState); + assert.equal(running.providerTurns[2]?.status, "running"); + assert.lengthOf(sent, 3); + yield* snapshot(directory + "/crash-" + (restart + 1) + ".sqlite"); + } else if (restart < restarts) { + // Admit the continuation, then lose the process before its provider-start effect. + assert.isTrue(yield* worker.runOnce()); + const projection = yield* (yield* Orchestrator.OrchestratorV2).getThreadProjection( + threadId, + ); + assert.equal(projection.runs.at(-1)?.status, "starting"); + assert.lengthOf(sent, 1); + yield* snapshot(directory + "/crash-" + (restart + 1) + ".sqlite"); + } else { + yield* worker.drain(); + if (siblingState) { + const projection = yield* orchestrator.getThreadProjection(threadId); + if (siblingState === "interrupted") { + assert.equal(projection.runs[1]?.status, "interrupted"); + assert.equal(projection.runs[2]?.status, "cancelled"); + assert.equal( + projection.runs.at(-1)?.restartContinuationOfRunId, + projection.runs[2]?.id, + ); + } + const delivered = yield* decodeCodexTurnInput(sent.at(-1)); + const outbound = JSON.stringify(delivered.input); + assert.lengthOf(sent, 4); + for (const marker of ["STEERING_MARKER_", "CONTEXT_MARKER_"]) { + assert.include(outbound, marker + 1); + assert.notInclude(outbound, marker + 0); + } + assert.include(outbound, attachments[1]!.path); + assert.notInclude(outbound, attachments[0]!.path); + assert.lengthOf( + sent.filter((params) => JSON.stringify(params).includes("STEERING_MARKER_0")), + 1, + "S1 is delivered once", + ); + } else if (completed) { + yield* worker.drain(); + const projection = + yield* (yield* Orchestrator.OrchestratorV2).getThreadProjection(threadId); + assert.lengthOf( + projection.messages.filter( + (message) => message.id === MessageId.make(name + ":steer-0"), + ), + 1, + ); + assert.equal(projection.runs[0]?.status, "completed"); + assert.lengthOf(sent, 2, "accepted steer starts exactly one follow-up"); + assert.lengthOf(projection.runs, 2); + } + } + }).pipe(Effect.provide(runtime())), + ); + } + if (siblingState) return; + assert.lengthOf(sent, 2); + const delivered = yield* decodeCodexTurnInput(sent[1]); + assert.include(JSON.stringify(delivered.input), "STEERING_MARKER"); + assert.include(JSON.stringify(delivered.input), "CONTEXT_MARKER"); + const bytes = yield* Effect.forEach( + delivered.input.filter((item) => item.type !== "text"), + (item) => + item.type === "localImage" + ? fs + .readFile(item.path) + .pipe(Effect.map((data) => Buffer.from(data).toString("base64"))) + : Effect.succeed(item.url.replace(/^data:image\/png;base64,/, "")), + ); + assert.deepEqual(bytes, ["AQIDBA=="]); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect.each([{ goalStatus: "active" }, { goalStatus: "paused" }] as const)( + "retries an EmptyInput continuation with a prompt only for an active goal ($goalStatus)", + ({ goalStatus }) => + Effect.gen(function* () { + const nativeThreadId = "native-empty-continuation"; + const prompt = "Continue where you left off."; + const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId: "unused", prompt }); + const turnStart = preamble[5]?.type === "expect_outbound" ? preamble[5].frame : undefined; + assert.isTrue(Predicate.isObject(turnStart) && Predicate.isObject(turnStart.params)); + const params = (turnStart as { readonly params: Record }).params; + const request = (id: number, method: string, params: unknown, result: unknown) => [ + { type: "expect_outbound" as const, frame: { id, method, params } }, + { type: "emit_inbound" as const, frame: { id, result } }, + ]; + const methods: string[] = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "goal-empty-continuation", + entries: [ + ...preamble.slice(0, 5), + { + type: "expect_outbound", + frame: { id: 3, method: "turn/start", params: { ...params, input: [] } }, + }, + { + type: "emit_inbound", + frame: { + id: 3, + error: { code: -32603, message: "failed to submit turn input: EmptyInput" }, + }, + }, + ...request( + 4, + "thread/goal/get", + { threadId: nativeThreadId }, + { + goal: { + threadId: nativeThreadId, + objective: "Finish the work", + status: goalStatus, + tokenBudget: null, + tokensUsed: 10, + timeUsedSeconds: 3, + createdAt: 1782622440, + updatedAt: 1782622450, + }, + }, + ), + ...(goalStatus === "active" + ? request( + 5, + "turn/start", + { + ...params, + input: [{ type: "text", text: prompt }], + }, + { turn: makeCodexReplayTurn({ id: "continued-goal", status: "inProgress" }) }, + ) + : []), + ], + }), + () => Effect.void, + (method) => + Effect.sync(() => { + methods.push(method); + }), + ); + const result = yield* harness.runtime + .startTurn({ + ...makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("empty-continuation"), + text: prompt, + }), + restartContinuationOfRunId: RunId.make("interrupted-goal"), + }) + .pipe(Effect.result); + assert.equal(result._tag, goalStatus === "active" ? "Success" : "Failure"); + assert.deepEqual( + methods.filter((method) => method !== "config/read").slice(2), + goalStatus === "active" + ? ["turn/start", "thread/goal/get", "turn/start"] + : ["turn/start", "thread/goal/get"], + ); + if (result._tag === "Failure") { + assert.nestedPropertyVal( + result.failure, + "cause.errorMessage", + "failed to submit turn input: EmptyInput", + ); + } + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect("resolves retryable app-server errors on resumed provider activity", () => Effect.gen(function* () { const scenario = "codex-provider-api-retry"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 1589b7d1ea46f..d6c98fc670bb9 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7891,9 +7891,27 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ) { return yield* activateGoal(turnInput, nativeThreadId, toCodexInput(turnInput)); } - yield* startNativeTurn( - turnInput, - !continuation ? yield* toCodexInput(turnInput) : [], + const input = !continuation ? yield* toCodexInput(turnInput) : []; + yield* startNativeTurn(turnInput, input).pipe( + // Codex rejects an empty continuation while a goal is active; send the prompt instead. + Effect.catchTags({ + CodexAppServerRequestError: (cause) => + input.length !== 0 || + turnInput.restartContinuationOfRunId === undefined || + cause.code !== -32603 || + cause.errorMessage !== "failed to submit turn input: EmptyInput" || + turnInput.message.text.trim() === "" + ? Effect.fail(cause) + : 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* cause; + yield* startNativeTurn(turnInput, yield* toCodexInput(turnInput)); + }), + }), ); }).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 0a250bb2b865c..495805475c139 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -36,6 +36,11 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ type: Schema.Literal("provider-runtime.continue"), sourceRunId: RunId, }), + Schema.Struct({ + type: Schema.Literal("provider-turn.steer-follow-up"), + sourceRunId: RunId, + steers: Schema.Array(Schema.Struct({ effectId: Schema.String, messageId: MessageId })), + }), Schema.Struct({ type: Schema.Literal("provider-session.detach"), providerSessionId: ProviderSessionId, @@ -135,6 +140,7 @@ export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.T export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ "provider-runtime.continue", + "provider-turn.steer-follow-up", "provider-session.detach", "provider-thread.rollback", "checkpoint.capture", @@ -349,7 +355,7 @@ export const layer: Layer.Layer = La AND active.rowid < candidate.rowid AND ${ excludeRestartContinuations - ? sql`active.effect_type != 'provider-runtime.continue'` + ? sql`active.effect_type NOT IN ('provider-runtime.continue', 'provider-turn.steer-follow-up')` : sql`1 = 1` } ) @@ -568,7 +574,7 @@ export const layer: Layer.Layer = La excludeRestartContinuations, maxRunningTitleGenerations, )} - AND ${excludeRestartContinuations ? sql`candidate.effect_type != 'provider-runtime.continue'` : sql`1 = 1`} + AND ${excludeRestartContinuations ? sql`candidate.effect_type NOT IN ('provider-runtime.continue', 'provider-turn.steer-follow-up')` : sql`1 = 1`} ORDER BY candidate.available_at ASC, candidate.created_at ASC, candidate.effect_id ASC LIMIT 1 ) diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 19a387eb8a954..0b92be53fff02 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -1,4 +1,5 @@ -import { CommandId } from "@t3tools/contracts"; +import { CommandId, type MessageId, type ThreadId, type RunId } from "@t3tools/contracts"; +import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -31,7 +32,7 @@ import * as RuntimeRequestService from "./RuntimeRequestService.ts"; import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ServerSettings from "../serverSettings.ts"; -import { continueRestartedRun } from "./RestartContinuation.ts"; +import { continueRestartedRun, restartContinuationBlocked } from "./RestartContinuation.ts"; const isProviderTurnControlError = Schema.is(ProviderTurnControlService.ProviderTurnControlError); @@ -102,6 +103,112 @@ export const layerExecutor: Layer.Layer< const threads = yield* ThreadManagementService.ThreadManagementService; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const settings = yield* ServerSettings.ServerSettingsService; + // Reuse message ids and receipts so retries cannot duplicate messages or follow-ups. + const followUpSteer = Effect.fn("EffectWorker.followUpSteer")(function* (input: { + readonly effectId: string; + readonly threadId: ThreadId; + readonly messageId: MessageId; + readonly restartContinuationOfRunId?: RunId; + readonly declined?: Extract< + EffectOutbox.OrchestrationEffectRequestV2, + { type: "provider-turn.steer" } + >; + }) { + const { threadId, declined, restartContinuationOfRunId } = input; + const projection = yield* threads.getThreadRecords(threadId, ["messages", "runs"], { + messageIds: [input.messageId], + }); + const message = projection.messages.find((item) => item.id === input.messageId); + const run = projection.runs.find((item) => item.id === message?.runId); + if (message === undefined || run === undefined) return false; + // Notifications must be queued messages; recovered steering waits behind active work. + const queued = + message.delegatedCompletion !== undefined || + message.notification !== undefined || + declined !== undefined || + restartContinuationOfRunId !== undefined; + yield* threads + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`command:steer-follow-up:${input.effectId}`), + threadId, + messageId: message.id, + ...(declined ? { steerFollowUpOfRunId: run.id } : {}), + ...(declined?.steerFollowUpAfterRunOrdinal !== undefined + ? { steerFollowUpAfterRunOrdinal: declined.steerFollowUpAfterRunOrdinal } + : {}), + ...(restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId }), + text: message.text, + ...(message.context ? { context: message.context } : {}), + attachments: message.attachments, + ...(declined?.modelSelection !== undefined + ? { modelSelection: declined.modelSelection } + : message.delegatedCompletion === undefined + ? {} + : { modelSelection: run.modelSelection }), + dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, + createdBy: message.createdBy, + creationSource: message.creationSource, + ...(message.delegatedCompletion === undefined + ? {} + : { delegatedCompletion: message.delegatedCompletion }), + ...(message.notification === undefined ? {} : { notification: message.notification }), + ...(message.scheduledTaskId === undefined + ? {} + : { scheduledTaskId: message.scheduledTaskId }), + ...(message.senderThreadId === undefined + ? {} + : { senderThreadId: message.senderThreadId }), + ...(message.linkedSender === undefined ? {} : { linkedSender: message.linkedSender }), + }) + .pipe( + // A disabled instance cannot retry this completed turn; its result stands. + Effect.catchTags({ + ThreadManagementProviderInstanceDisabledError: (disabledError) => + Effect.logInfo("orchestration-v2.effect-worker.steer-follow-up-disabled", { + threadId, + messageId: message.id, + projectId: disabledError.projectId, + instanceId: disabledError.instanceId, + }), + }), + ); + return true; + }); + const recoverSteering = Effect.fn("EffectWorker.recoverSteering")(function* ( + threadId: ThreadId, + request: Extract< + EffectOutbox.OrchestrationEffectRequestV2, + { type: "provider-turn.steer-follow-up" } + >, + ) { + const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null)); + if (enabled === null) return; + const projection = yield* threads.getThreadRecords( + threadId, + ["runs", "providerTurns", "runtimeRequests", "turnItems"], + { turnItemTypes: ["secret_request"] }, + ); + if ( + !resolveProjectSettings(enabled, projection.thread.projectId).settings + .continueThreadsAfterServerUpdate + ) + return; + const source = projection.runs.find((run) => run.id === request.sourceRunId); + if ( + source === undefined || + restartContinuationBlocked(projection, source, yield* DateTime.now, true) + ) + return; + for (const steer of request.steers) { + yield* followUpSteer({ + effectId: steer.effectId, + threadId, + messageId: steer.messageId, + restartContinuationOfRunId: source.id, + }); + } + }); return OrchestrationEffectExecutorV2.of({ execute: Effect.fn("OrchestrationEffectExecutorV2.execute")(function (effect, options) { const willRetry = options?.willRetry ?? false; @@ -126,6 +233,17 @@ export const layerExecutor: Layer.Layer< ), ); } + case "provider-turn.steer-follow-up": + return recoverSteering(effect.threadId, effect.request).pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); case "provider-session.detach": return providerSessions .detach({ @@ -298,79 +416,13 @@ export const layerExecutor: Layer.Layer< ) { return yield* error; } - const projection = yield* threads.getThreadRecords( - effect.threadId, - ["messages", "runs"], - { messageIds: [effect.request.messageId] }, - ); - const messageId = effect.request.messageId; - const message = projection.messages.find((item) => item.id === messageId); - const run = projection.runs.find((item) => item.id === message?.runId); - if (message === undefined || run === undefined) return yield* error; - // Notifications must be queued messages. - const queued = - message.delegatedCompletion !== undefined || - message.notification !== undefined || - error.steerDeclined === true; - // Reuse message ids and receipts so retries cannot duplicate messages or follow-ups. - yield* threads - .dispatch({ - type: "message.dispatch", - commandId: CommandId.make(`command:steer-follow-up:${effect.id}`), - threadId: effect.threadId, - messageId: message.id, - ...(error.steerDeclined ? { steerFollowUpOfRunId: run.id } : {}), - ...(error.steerDeclined && - effect.request.steerFollowUpAfterRunOrdinal !== undefined - ? { - steerFollowUpAfterRunOrdinal: - effect.request.steerFollowUpAfterRunOrdinal, - } - : {}), - text: message.text, - ...(message.context ? { context: message.context } : {}), - attachments: message.attachments, - ...(error.steerDeclined && effect.request.modelSelection !== undefined - ? { modelSelection: effect.request.modelSelection } - : message.delegatedCompletion === undefined - ? {} - : { modelSelection: run.modelSelection }), - dispatchMode: { - type: queued ? "queue_after_active" : "start_immediately", - }, - createdBy: message.createdBy, - creationSource: message.creationSource, - ...(message.delegatedCompletion === undefined - ? {} - : { delegatedCompletion: message.delegatedCompletion }), - ...(message.notification === undefined - ? {} - : { notification: message.notification }), - ...(message.scheduledTaskId === undefined - ? {} - : { scheduledTaskId: message.scheduledTaskId }), - ...(message.senderThreadId === undefined - ? {} - : { senderThreadId: message.senderThreadId }), - ...(message.linkedSender === undefined - ? {} - : { linkedSender: message.linkedSender }), - }) - .pipe( - // A disabled instance cannot retry this completed turn; its result stands. - Effect.catchTags({ - ThreadManagementProviderInstanceDisabledError: (disabledError) => - Effect.logInfo( - "orchestration-v2.effect-worker.steer-follow-up-disabled", - { - threadId: effect.threadId, - messageId: message.id, - projectId: disabledError.projectId, - instanceId: disabledError.instanceId, - }, - ), - }), - ); + const followedUp = yield* followUpSteer({ + effectId: effect.id, + threadId: effect.threadId, + messageId: effect.request.messageId, + ...(error.steerDeclined ? { declined: effect.request } : {}), + }); + if (!followedUp) return yield* error; }), ), Effect.mapError( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 226b7d703d0b1..cb9620aaae653 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6820,9 +6820,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (command.restartContinuationOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.restartContinuationOfRunId); + // A completed turn can still own an accepted, undelivered steer. Its + // exact recorded identity is allowed through the normal restart guards. + const pendingSteer = + source?.restartPendingMessageIds?.includes(command.messageId) === true && + (source.status === "waiting" || source.status === "completed") && + latestProviderTurnForAttempt(projection.providerTurns, source.activeAttemptId)?.status === + "completed"; if ( !source || !( + pendingSteer || isRestartContinuationSource(source, projection.providerTurns) || continuationFailedBeforePrompt(source, projection.providerTurns) ) || @@ -6834,7 +6842,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run.id !== source.id && run.status !== "queued" && !cancelledBeforeStart(run) && - runRanAfter(run, source), + runRanAfter(run, source) && + !( + pendingSteer && + source.restartPendingMessageIds?.includes(run.userMessageId) && + run.status !== "interrupted" && + run.status !== "cancelled" && + run.status !== "rolled_back" + ), ) || (yield* stopReachedRun(command, command.threadId, source.id)) ) { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5e01ea08b52b..8afba2f05c0ca 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -327,7 +327,13 @@ export type ProjectionRuntimeRecoveryState = Pick< | "runtimeRequests" | "messages" | "turnItems" ->; +> & { + readonly pendingSteering?: ReadonlyArray<{ + readonly effectId: string; + readonly runId: RunId; + readonly messageId: MessageId; + }>; +}; export type ProjectionPendingUserInputs = Pick< OrchestrationV2ThreadProjection, @@ -841,6 +847,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 } @@ -1282,6 +1292,11 @@ const decodeThreadPayload = Schema.decodeUnknownEffect( const decodeRunPayload = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2RunJsonSchema), ); +const decodePendingSteeringPayload = Schema.decodeUnknownEffect( + Schema.fromJsonString( + Schema.Struct({ effectId: Schema.String, runId: RunId, messageId: MessageId }), + ), +); const decodeRunAttemptPayload = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2RunAttemptJsonSchema), ); @@ -2403,7 +2418,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", )} `; @@ -5232,6 +5250,15 @@ export const layer: Layer.Layer = WHERE latest.thread_id = ${threadId} ORDER BY latest.ordinal DESC LIMIT 1 ) + OR run.run_id IN ( + SELECT message.run_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') + ) -- Retain the latest run for each provider thread with lost -- background work, including owners used before a handoff. OR run.run_id IN ( @@ -5491,6 +5518,22 @@ export const layer: Layer.Layer = ) ORDER BY item.ordinal ASC, item.turn_item_id ASC `, + pendingSteering: sql` + SELECT json_object( + 'effectId', effect.effect_id, + 'runId', message.run_id, + 'messageId', message.message_id + ) AS payload_json + 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 + `, }; const sections = Object.entries(queries).map( ([section, query]) => sql` @@ -5527,7 +5570,12 @@ export const layer: Layer.Layer = decodeRows(decodeMessagePayload, threadId)(readRows("messages")), decodeRows(decodeTurnItemPayload, threadId)(readRows("turnItems")), ]); + const pendingSteering = yield* decodeRows( + decodePendingSteeringPayload, + threadId, + )(readRows("pendingSteering")); return { + pendingSteering, thread, runs, attempts, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 46b739bec5732..8c134b4709e64 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -529,6 +529,14 @@ export const make = Effect.gen(function* () { }); } for (const run of runs) { + const pendingMessageIds = [ + ...new Set([ + ...(run.restartPendingMessageIds ?? []), + ...(projection.pendingSteering ?? []) + .filter((steer) => steer.runId === run.id) + .map((steer) => steer.messageId), + ]), + ]; events.push({ id: yield* allocateEventId(), type: "run.updated", @@ -536,7 +544,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, + ...(pendingMessageIds.length === 0 + ? {} + : { restartPendingMessageIds: pendingMessageIds }), + }, }); for (const attempt of projection.attempts.filter( (candidate) => @@ -832,6 +848,41 @@ export const make = Effect.gen(function* () { }, }); } + // Checkpoint-waiting and completed runs are not cancelled above. Keep + // their accepted input on the source as well, in case another restart + // cuts the follow-up after admission but before provider start. + const steeringFollowUps: Array = []; + for (const run of projection.runs) { + if (runs.some((cancelled) => cancelled.id === run.id)) continue; + if (run.status !== "waiting" && run.status !== "completed") continue; + const steers = (projection.pendingSteering ?? []) + .filter((steer) => steer.runId === run.id) + .map(({ effectId, messageId }) => ({ effectId, messageId })); + if (steers.length === 0) continue; + events.push({ + id: yield* allocateEventId(), + type: "run.updated", + threadId: projection.thread.id, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { + ...run, + restartPendingMessageIds: [ + ...new Set([ + ...(run.restartPendingMessageIds ?? []), + ...steers.map((steer) => steer.messageId), + ]), + ], + }, + }); + steeringFollowUps.push({ + id: "effect:steer-follow-up:" + run.id, + commandId, + threadId: projection.thread.id, + request: { type: "provider-turn.steer-follow-up", sourceRunId: run.id, steers }, + }); + } const continuationRun = continueAfterRestart && trigger === "startup" ? restartContinuationRun( @@ -849,12 +900,13 @@ export const make = Effect.gen(function* () { }, ] : []; + if (continueAfterRestart) effects.unshift(...steeringFollowUps); const stoppedSessions = projection.providerSessions.filter( (candidate) => candidate.status !== "stopped" && candidate.status !== "error", ).length; let retiredEffects: number; let cancelledEffectIds: ReadonlyArray = []; - if (events.length === 0) { + if (events.length === 0 && effects.length === 0) { const retiredEffectIds = yield* outbox .cancelUnsettled({ threadId: projection.thread.id, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index f6ad1471de56b..b5c41e07bb5f6 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -64,6 +64,7 @@ import { import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +import { restartPendingMessageIds } from "./RestartSteering.ts"; import { isRestartNoteContinuation, pendingRestartCancelledBackgroundWork, @@ -792,10 +793,24 @@ export const layer: Layer.Layer< }); let effectiveHandoffs = handoffs; let resumeFellBack = false; - const userText = projectComposerContextForProvider({ - text: providerMessage.text, - records: providerMessage.context?.records ?? [], - }); + const pendingSteeringIds = + run.restartContinuationOfRunId === undefined + ? [] + : restartPendingMessageIds( + run, + projection.runs, + projection.providerTurns, + projection.attempts, + ); + const steeringContinuation = pendingSteeringIds.length > 0; + // A continuation carrying steering already holds the projected text of each steer. + const carriedSteering = steeringContinuation && !pendingSteeringIds.includes(message.id); + const userText = carriedSteering + ? providerMessage.text + : projectComposerContextForProvider({ + text: providerMessage.text, + records: providerMessage.context?.records ?? [], + }); // Native goal commands go verbatim; handoff history waits a turn. const goalDriver = session.driver === "codex" || @@ -803,12 +818,14 @@ export const layer: Layer.Layer< session.providerSession.capabilities.turns.supportsNativeGoals === true; const goalCommand = goalDriver && parseGoalCommand(providerMessage) !== null; // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. - const noteContinuation = isRestartNoteContinuation( - run, - projection.runs, - projection.providerTurns, - projection.attempts, - ); + const noteContinuation = + steeringContinuation || + isRestartNoteContinuation( + run, + projection.runs, + projection.providerTurns, + projection.attempts, + ); const restartCancelledWork = pendingRestartCancelledBackgroundWork({ runs: projection.runs, providerTurns: projection.providerTurns, @@ -821,7 +838,11 @@ export const layer: Layer.Layer< ) .map((candidate) => candidate.id), ), - run, + // A steer delivered as itself takes the next-message path for cancelled work. + run: + steeringContinuation && !carriedSteering + ? { ...run, restartContinuationOfRunId: undefined } + : run, attempts: projection.attempts, }); const restartNote = diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 242e366d4b605..d8be6e00a7e81 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, isOrchestrationV2WorkActive, @@ -16,6 +17,7 @@ import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { isNativeMaintenanceCommand } from "./CommandPolicy.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; +import { restartPendingMessageIds } from "./RestartSteering.ts"; import { awaitingProviderThreadId, cancelledRosterTaskWork, @@ -36,7 +38,7 @@ const CONTINUE_PROMPT = const MAX_CONTINUATION_CHAIN = 3; const CONTINUATION_LOOP_WINDOW_MS = 5 * 60 * 1_000; -function restartContinuationBlocked( +export function restartContinuationBlocked( projection: Pick< ProjectionRuntimeRecoveryState, "thread" | "runtimeRequests" | "providerTurns" | "turnItems" @@ -246,11 +248,17 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta }); return; } + const pendingMessageIds = restartPendingMessageIds( + source, + projection.runs, + projection.providerTurns, + projection.attempts, + ); const sourceRecords = yield* threads.getThreadRecords( input.threadId, ["messages", "turnItems"], { - messageIds: [source.userMessageId], + messageIds: [source.userMessageId, ...pendingMessageIds], turnItemRunIds: [source.id], turnItemTypes: ["run_interrupt_request"], }, @@ -291,6 +299,18 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta (message) => message.id === source.userMessageId, ); if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return; + const pendingMessages = pendingMessageIds.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, @@ -301,10 +321,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", @@ -312,15 +332,19 @@ 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: [], + ...(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/apps/server/src/orchestration-v2/RestartSteering.ts b/apps/server/src/orchestration-v2/RestartSteering.ts new file mode 100644 index 0000000000000..d2c73a0eb9f05 --- /dev/null +++ b/apps/server/src/orchestration-v2/RestartSteering.ts @@ -0,0 +1,56 @@ +import type { + MessageId, + OrchestrationV2ProviderTurn, + OrchestrationV2Run, + OrchestrationV2RunAttempt, +} from "@t3tools/contracts"; + +/** + * Keep steering across continuations cut before delivery. As with restart + * background notes, only a completed turn proves acceptance across adapters. + */ +export function restartPendingMessageIds( + source: OrchestrationV2Run, + runs: ReadonlyArray, + providerTurns: ReadonlyArray, + attempts: ReadonlyArray, +): ReadonlyArray { + const completedAttempts = new Set( + providerTurns.filter((turn) => turn.status === "completed").map((turn) => turn.runAttemptId), + ); + const completedRuns = new Set( + attempts.filter((attempt) => completedAttempts.has(attempt.id)).map((attempt) => attempt.runId), + ); + let current = source; + const visited = new Set([current.id]); + const pending = [...(current.restartPendingMessageIds ?? [])]; + while ( + current.restartContinuationOfRunId !== undefined && + !completedRuns.has(current.id) && + !(current.activeAttemptId !== null && completedAttempts.has(current.activeAttemptId)) + ) { + const previous = runs.find((run) => run.id === current.restartContinuationOfRunId); + if (previous === undefined || visited.has(previous.id)) break; + visited.add(previous.id); + current = previous; + pending.unshift(...(current.restartPendingMessageIds ?? [])); + } + // A sibling may already own this input. Keep its live work, proven acceptance, + // and explicit stop separate from the ambiguous delivery of the cut chain. + return [...new Set(pending)].filter( + (id) => + !runs.some( + (run) => + run.userMessageId === id && + !visited.has(run.id) && + (completedRuns.has(run.id) || + (run.activeAttemptId !== null && completedAttempts.has(run.activeAttemptId)) || + run.status === "queued" || + run.status === "preparing" || + run.status === "starting" || + run.status === "running" || + run.status === "waiting" || + run.status === "interrupted"), + ), + ); +} diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index 10df2ef464f8b..85c6afaf482c8 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -110,10 +110,10 @@ opt-out. The [v2 recovery service](../../apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts) requires matching durable run, provider thread, session, and native resume identity. Queued runs never started, so recovery holds them and continues the run they wait -behind. A settled run qualifies only when the restart cancels work the agent waits on -(a subagent, a background task, or a shell or monitor the agent marked with -`t3_background_wait`) and its session was live, so stale records from an older crash -never wake it. Its continuation prompt is the note of what will not report back. An +behind. A settled run without undelivered steering qualifies only when the restart +cancels work the agent waits on (a subagent, a background task, or a shell or monitor the +agent marked with `t3_background_wait`) and its session was live, so stale records from an +older crash never wake it. Its continuation prompt is the note of what will not report back. An unmarked shell or monitor does not wake a settled run: most are dev servers and their watchdogs that nobody waits on, so that note waits for the next user message. @@ -127,15 +127,25 @@ cannot be overwritten by a stale cancellation. The [continuation handler](../../apps/server/src/orchestration-v2/RestartContinuation.ts) rechecks the preference, provider selection, newer user work, a stop the user requested, and maintenance turns such as `/compact` before dispatching. Intent and delivery both -refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider -turn already ended, and a run waiting on the user: an approval, question, or secret that was pending, +refuse a thread that is archived, settled, or still snoozed, a plain continuation of a +cut turn whose provider turn already ended, and a run waiting on the user: an approval, +question, or secret that was pending, or that recovery closed on the cut turn. Message-capable questions stay answerable, so 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. +Recovery records undelivered steering before retiring its process-bound effects. The +continuation carries its text, attachments, and expanded context. A steer whose provider +already completed uses its accepted message identity and the normal stable follow-up +receipt, including while its run waits for a replayable checkpoint. If another restart +cuts delivery, it follows the continuation chain back to those recorded message IDs. +As with background notes, a completed provider turn proves acceptance; an ambiguous +delivery can be repeated. Codex retries an empty continuation rejected with `EmptyInput` +only when its goal is still active, using the continuation prompt as 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 44e9aa8290e9a..0d34423cf8d8e 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -741,6 +741,7 @@ export const OrchestrationV2Run = Schema.Struct({ checkpointId: Schema.NullOr(CheckpointId), contextHandoffId: Schema.NullOr(ContextHandoffId), restartContinuationOfRunId: Schema.optional(RunId), + restartPendingMessageIds: Schema.optional(Schema.Array(MessageId)), manualContinuationOfRunId: Schema.optional(RunId), usageLimitContinuationOfRunId: Schema.optional(RunId), delegatedTaskId: Schema.optional(NodeId),