diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..5f3b0c4289c9e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2242,6 +2242,130 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("asks T3 to record a root turn Codex starts from its own queue", () => + Effect.gen(function* () { + const nativeThreadId = "queue-thread"; + const nativeTurnId = "queue-first-turn"; + const queuedTurnId = "queue-external-turn"; + const prompt = "Reply with the word READY."; + const queued = ( + label: string, + method: "turn/started" | "item/completed" | "turn/completed", + params: object, + ): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label, + frame: { method, params }, + }); + const transcript = makeCodexReplayTranscript({ + scenario: "codex-queue-turn", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + queued("turn/completed/first", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }), + queued("turn/started/queued", "turn/started", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: queuedTurnId, status: "inProgress" }), + }), + queued("item/completed/queued-prompt", "item/completed", { + item: { + type: "userMessage", + id: "queued-prompt", + content: [{ type: "text", text: "Reply with the word QUEUED." }], + }, + threadId: nativeThreadId, + turnId: queuedTurnId, + completedAtMs: 1782622500000, + }), + queued("item/completed/queued-answer", "item/completed", { + item: { + type: "agentMessage", + id: "queued-answer", + text: "QUEUED", + phase: "final_answer", + memoryCitation: null, + }, + threadId: nativeThreadId, + turnId: queuedTurnId, + completedAtMs: 1782622501000, + }), + queued("turn/completed/queued", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: queuedTurnId, status: "completed" }), + }), + { + type: "emit_inbound", + label: "marker", + frame: { method: "thread/closed", params: { threadId: "queue-marker-thread" } }, + }, + ], + }); + const marker = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + transcript, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + (method) => + method === "thread/closed" ? Deferred.succeed(marker, undefined) : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("queue-first-attempt"), + text: prompt, + }), + ); + yield* harness.awaitTerminal(1); + yield* Deferred.await(marker); + + assert.isEmpty(harness.subagentUpdates()); + assert.lengthOf(harness.continuationRequests, 1); + assert.equal(harness.continuationRequests[0]?.threadId, harness.threadId); + assert.equal(harness.continuationRequests[0]?.providerThreadId, harness.providerThread.id); + + // The continuation run takes over the running native turn instead of sending turn/start. + const continuation = makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("queue-continuation-attempt"), + text: "Background task completed.", + }); + yield* harness.runtime.startTurn({ + ...continuation, + providerTurnOrdinal: 2, + message: { ...continuation.message, createdBy: "agent", creationSource: "provider" }, + }); + yield* harness.awaitTerminal(2); + const queuedMessages = harness.events.flatMap((event) => + event.type === "message.updated" && event.message.runId === continuation.runId + ? [[event.message.role, event.message.text]] + : [], + ); + assert.deepInclude(queuedMessages, ["user", "Reply with the word QUEUED."]); + assert.deepInclude(queuedMessages, ["assistant", "QUEUED"]); + assert.equal(harness.terminalEvents()[1]?.status, "completed"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index efa5d0a9c28f1..f46db7eb4cbbd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1224,6 +1224,8 @@ interface ActiveCodexTurnContext { hookStopReason?: string; capacityRetryAttempt?: number; capacityRetryPending?: Deferred.Deferred; + /** Codex started this root turn from its own queue, so its prompt is not a T3 message. */ + startedByCodex?: boolean; closedCommandItemIds?: ReadonlySet; latestProviderFailure?: { readonly nativeMessage: string; @@ -1872,7 +1874,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ), ); - const client = yield* clientFactory.open({ + const openedClient = yield* clientFactory.open({ instanceId: adapterOptions.instanceId, threadId: input.threadId, providerSessionId: input.providerSessionId, @@ -1880,6 +1882,38 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( settings: resolvedRuntime?.config ?? adapterOptions.settings, environment: resolvedRuntime?.environment ?? adapterOptions.environment, }); + // Root turns Codex started on its own (`codex queue`) wait here for T3 to open a run. + // Their notifications are held in order and replayed once the run owns the turn. + const codexStartedTurns = new Map< + string, + { + readonly nativeThreadId: string; + readonly nativeTurnId: string; + readonly startedAt: DateTime.Utc; + readonly held: Array>; + readonly registered: Deferred.Deferred; + } + >(); + const client: typeof openedClient = { + ...openedClient, + handleServerNotification: (method, handler) => + openedClient.handleServerNotification(method, (payload) => { + const turnId = + "turnId" in payload && typeof payload.turnId === "string" + ? payload.turnId + : "turn" in payload && + typeof payload.turn === "object" && + payload.turn !== null && + "id" in payload.turn && + typeof payload.turn.id === "string" + ? payload.turn.id + : undefined; + const waiting = turnId === undefined ? undefined : codexStartedTurns.get(turnId); + if (waiting === undefined) return handler(payload); + waiting.held.push(handler(payload)); + return Effect.void; + }), + }; const additionalContextByThread = yield* Ref.make( new Map>(), ); @@ -2432,6 +2466,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (context !== undefined || attemptsRemaining <= 0) { return context; } + // Approvals for a queue-started turn wait for its run; held notifications never reach here. + const codexStarted = codexStartedTurns.get(nativeTurnId); + if (codexStarted !== undefined) { + yield* Deferred.await(codexStarted.registered); + return (yield* Ref.get(activeTurns)).get(nativeTurnId); + } yield* Effect.yieldNow; return yield* awaitActiveTurn(nativeTurnId, attemptsRemaining - 1); }); @@ -2447,6 +2487,68 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return context === undefined ? undefined : ({ context, settled: false } as const); }); + const releaseCodexStartedTurn = Effect.fnUntraced(function* (nativeTurnId: string) { + const waiting = codexStartedTurns.get(nativeTurnId); + if (waiting === undefined) return; + codexStartedTurns.delete(nativeTurnId); + yield* Deferred.succeed(waiting.registered, false); + }); + + const awaitRunForCodexStartedTurn = Effect.fn("CodexAdapterV2.awaitRunForCodexStartedTurn")( + function* (input: { + readonly appThreadId: ThreadId; + readonly providerThread: OrchestrationV2ProviderThread; + readonly nativeThreadId: string; + readonly nativeTurnId: string; + readonly startedAt: DateTime.Utc; + }) { + if (continuationRequests === undefined) return; + codexStartedTurns.set(input.nativeTurnId, { + nativeThreadId: input.nativeThreadId, + nativeTurnId: input.nativeTurnId, + startedAt: input.startedAt, + held: [], + registered: yield* Deferred.make(), + }); + yield* continuationRequests.offer({ + threadId: input.appThreadId, + providerThreadId: input.providerThread.id, + driver: CODEX_PROVIDER, + detail: null, + clearIfCurrent: () => releaseCodexStartedTurn(input.nativeTurnId), + onRunEndedWithoutStarting: () => releaseCodexStartedTurn(input.nativeTurnId), + }); + }, + ); + + /** Gives a queue-started Codex turn to the continuation run T3 opened for it. */ + const adoptCodexStartedTurn = Effect.fn("CodexAdapterV2.adoptCodexStartedTurn")(function* ( + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + ) { + if ( + turnInput.message.createdBy !== "agent" || + turnInput.message.creationSource !== "provider" + ) + return false; + const nativeThreadId = turnInput.providerThread.nativeThreadRef?.nativeId; + const waiting = [...codexStartedTurns.values()].find( + (candidate) => candidate.nativeThreadId === nativeThreadId, + ); + if (waiting === undefined || (yield* Deferred.isDone(waiting.registered))) return false; + const context = yield* registerRootTurn({ + turnInput, + nativeTurnId: waiting.nativeTurnId, + startedAt: waiting.startedAt, + }); + context.startedByCodex = true; + yield* Deferred.succeed(waiting.registered, true); + while (waiting.held.length > 0) { + yield* waiting.held.shift()!.pipe(Effect.ignore); + } + codexStartedTurns.delete(waiting.nativeTurnId); + return true; + }); + const trackRunningCommandItem = (nativeTurnId: string, item: TrackedRunningCommandItem) => Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); @@ -3838,7 +3940,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( >, nativeStartedAt?: DateTime.Utc, ) { - if (context.subagent === null || context.providerTurnOrdinal === 1) { + if ( + context.subagent === null + ? context.startedByCodex !== true + : context.providerTurnOrdinal === 1 + ) { return false; } const text = codexUserMessageText(item.content); @@ -3851,7 +3957,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeStartedAt, ); const artifacts = makeSubagentConversationArtifacts({ - senderThreadId: context.subagent.parentContext.projectionThreadId, + ...(context.subagent === null + ? {} + : { senderThreadId: context.subagent.parentContext.projectionThreadId }), messageId: idAllocator.derive.messageFromProviderItem({ driver: CODEX_PROVIDER, nativeItemId: item.id, @@ -3870,15 +3978,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ordinal, now: startedAt, }); + const runId = context.subagent === null ? context.projectionRunId : null; yield* emitProviderEvent({ type: "message.updated", driver: CODEX_PROVIDER, - message: artifacts.message, + message: { ...artifacts.message, runId }, }); yield* emitProviderEvent({ type: "turn_item.updated", driver: CODEX_PROVIDER, - turnItem: artifacts.turnItem, + turnItem: { ...artifacts.turnItem, runId }, }); return true; }); @@ -5026,6 +5135,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); return; } + const rootProviderThread = rootProviderThreads.get(payload.threadId); + if ( + continuationRequests !== undefined && + rootProviderThread?.appThreadId != null && + !(yield* Ref.get(subagentThreads)).has(payload.threadId) && + (yield* findActiveTurnByNativeThreadId(payload.threadId)) === undefined + ) { + yield* awaitRunForCodexStartedTurn({ + appThreadId: rootProviderThread.appThreadId, + providerThread: rootProviderThread, + nativeThreadId: payload.threadId, + nativeTurnId: payload.turn.id, + startedAt: codexTimestamp(payload.turn.startedAt), + }); + return; + } yield* rememberSubagentTurnStarted({ nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, @@ -7814,6 +7939,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand); const nativeThreadId = turnInput.providerThread.nativeThreadRef?.nativeId; return Effect.gen(function* () { + if (yield* adoptCodexStartedTurn(turnInput)) return; if ( turnInput.restartContinuationOfRunId === undefined && nativeThreadId != null &&