From e8f46d3f5b85979112bccf7990c558eac750b331 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:16:29 +0000 Subject: [PATCH 1/2] fix(server): Codex goals wait for a delegated task instead of re-prompting Codex continues an active goal as soon as a goal turn completes, and the adapter ran every continuation for the 5 s grace window with no check for a T3 delegated task that will wake the thread. A parent waiting on an async delegate_task child kept starting turns that only reported the child was still running, spending goal tokens until it finished. When a delegated task will still wake the thread, the goal hold now pauses the goal at once and settles the run. The next ordinary turn (the child's wake or a user message) re-activates the goal through the same path as /goal resume. The wake check is the hasPendingWake session field shared with the Claude fix, counted only while the task's result can be delivered. Fixes pingdotgg/t3code#26857 AI-assisted; verified with the tests and tools listed above. --- .../Adapters/CodexAdapterV2.test.ts | 98 +++++++++++++ .../Adapters/CodexAdapterV2.ts | 134 +++++++++++------- .../ProviderSessionManager.test.ts | 40 ++++++ .../ProviderSessionManager.ts | 37 ++++- .../provider-core/src/server/mcpSession.ts | 3 + 5 files changed, 262 insertions(+), 50 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b4648fe578bc9..52146a6f72098 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9882,6 +9882,104 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("pauses a goal at once while a delegated task will wake the thread", () => + Effect.gen(function* () { + const turnStart = (id: number, text: string) => + Effect.map( + CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + }), + (params): CodexReplay.CodexAppServerReplayEntry => ({ + type: "expect_outbound", + label: "turn/start", + frame: { id, method: "turn/start", params }, + }), + ); + const transcript = makeCodexReplayTranscript({ + scenario: "goal-paused-for-wake", + entries: [ + ...sessionStart, + yield* turnStart(3, "Delegate and wait"), + ...withReplayRequestId( + codexReplayPreamble({ nativeThreadId, nativeTurnId: "goal-turn", prompt: "" }).slice( + 6, + ), + 3, + ), + notification("goal active", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "goal-turn", + goal: codexGoal("active", 10), + }), + notification("turn done", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn", status: "completed" }), + }), + ...request( + 4, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 10) }, + ), + yield* turnStart(5, "Child finished"), + ...withReplayRequestId( + codexReplayPreamble({ nativeThreadId, nativeTurnId: "wake-turn", prompt: "" }).slice( + 6, + ), + 5, + ), + ...request( + 6, + "thread/goal/set", + { threadId: nativeThreadId, status: "active" }, + { goal: codexGoal("active", 10) }, + ), + notification("goal resumed", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "wake-turn", + goal: codexGoal("active", 10), + }), + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("test"), + threadId: harness.threadId, + providerSessionId: "wake-session", + providerInstanceId: ProviderInstanceId.make("codex"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer test", + browserToolsAvailable: true, + hasPendingWake: Effect.succeed(true), + }); + yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "Delegate and wait")); + // Settles without waiting out the continuation grace. + yield* harness.firstTerminal; + assert.deepEqual(providerGoals(harness.events), ["active", "paused"]); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + + yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "Child finished")); + yield* harness.until(() => providerGoals(harness.events).at(-1) === "active"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect("stops a turn Codex starts while the expiry pause is in flight", () => Effect.gen(function* () { const prompt = "Keep going"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 5cfd93980a887..4ce8c6ac8525a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2045,6 +2045,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const goalActivations = new Map(); // Held runs must not complete while Stop resolves its target. const goalStops = new Map(); + // Goals paused because a delegated task will wake the thread; the next ordinary turn resumes them. + const goalsPausedForWake = new Set(); const latestRunTurnId = (providerTurnId: ProviderTurnId) => continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; const isContinuedRunDescendant = ( @@ -6257,8 +6259,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( completedTurn: input.goalHoldTurn, next: yield* Deferred.make(), }); - yield* Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe( - Effect.andThen(pauseUncontinuedGoal(nativeThreadId, event)), + const hasPendingWake = (yield* mcpSessions.read(input.context.input.threadId)) + ?.hasPendingWake; + // A continuation turn would only report that the delegated task is still running. + yield* (hasPendingWake ?? Effect.succeed(false)).pipe( + Effect.tap((waitsForWake) => + waitsForWake ? Effect.void : Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE), + ), + Effect.flatMap((waitsForWake) => + pauseUncontinuedGoal(nativeThreadId, event, waitsForWake), + ), Effect.andThen( turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)), ), @@ -6276,6 +6286,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const pauseUncontinuedGoal = Effect.fnUntraced(function* ( nativeThreadId: string, expected: CodexRootTerminalEvent, + waitsForWake: boolean, ) { const hold = goalHolds.get(nativeThreadId); if ( @@ -6288,9 +6299,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( goalHolds.delete(nativeThreadId); // Keeps a turn Codex starts before settlement an orphan once the goal reads paused. goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); - yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", { - nativeThreadId, - }); + if (!waitsForWake) { + yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", { + nativeThreadId, + }); + } yield* client .request("thread/goal/set", { threadId: nativeThreadId, status: "paused" }) .pipe( @@ -6298,6 +6311,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.timeout("10 seconds"), Effect.flatMap(({ goal }) => { goalsByNativeThread.set(nativeThreadId, providerGoalFromCodex(goal)); + if (waitsForWake && goal.status === "paused") + goalsPausedForWake.add(nativeThreadId); return emitGoalUpdate(nativeThreadId); }), Effect.catch((cause) => @@ -7247,6 +7262,47 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), ); + /** Starts a turn and then activates its paused goal; Stop cancels activation. */ + const activateGoal = Effect.fnUntraced(function* ( + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + threadId: string, + codexInput: ReturnType, + ) { + goalsPausedForWake.delete(threadId); + // Hold fast first turns until activation. + const activation = { stopped: false }; + goalActivations.set(threadId, activation); + yield* Effect.gen(function* () { + yield* startNativeTurn(turnInput, yield* codexInput); + if (activation.stopped) return; + const { goal } = yield* client.request("thread/goal/set", { + threadId, + status: "active", + }); + goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); + if (activation.stopped) { + const paused = yield* client.request("thread/goal/set", { + threadId, + status: "paused", + }); + goalsByNativeThread.set(threadId, providerGoalFromCodex(paused.goal)); + } + }).pipe( + Effect.ensuring( + turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + if (goalActivations.get(threadId) === activation) { + goalActivations.delete(threadId); + } + if (goalsByNativeThread.get(threadId)?.status !== "active") { + yield* releaseGoalHold(threadId); + } + }), + ), + ), + ); + }); + /** * Setting or resuming a goal starts the first goal turn with this run's * turn configuration (approvals, sandbox, model, plan mode), since a @@ -7260,6 +7316,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (command.type === "set") yield* validateCodexGoalObjective(command.objective); const threadId = yield* getNativeThreadId(turnInput.providerThread); rootProviderThreads.set(threadId, turnInput.providerThread); + goalsPausedForWake.delete(threadId); const readGoal = client .request("thread/goal/get", { threadId }) .pipe( @@ -7302,50 +7359,20 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); } - // Hold fast first turns until activation; Stop cancels activation. - const activation = { stopped: false }; - goalActivations.set(threadId, activation); - yield* Effect.gen(function* () { - // A bare objective gets echoed back, and "start this goal" gets a - // create_goal call that Codex rejects because the goal already exists. - yield* startNativeTurn( - turnInput, - command.type === "set" - ? yield* toCodexInput({ - ...turnInput, - message: { - ...turnInput.message, - text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, - }, - }) - : [], - ); - if (activation.stopped) return; - const { goal } = yield* client.request("thread/goal/set", { - threadId, - status: "active", - }); - goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); - if (activation.stopped) { - const paused = yield* client.request("thread/goal/set", { - threadId, - status: "paused", - }); - goalsByNativeThread.set(threadId, providerGoalFromCodex(paused.goal)); - } - }).pipe( - Effect.ensuring( - turnTerminalizationPermit.withPermits(1)( - Effect.gen(function* () { - if (goalActivations.get(threadId) === activation) { - goalActivations.delete(threadId); - } - if (goalsByNativeThread.get(threadId)?.status !== "active") { - yield* releaseGoalHold(threadId); - } - }), - ), - ), + // A bare objective gets echoed back, and "start this goal" gets a + // create_goal call that Codex rejects because the goal already exists. + yield* activateGoal( + turnInput, + threadId, + command.type === "set" + ? toCodexInput({ + ...turnInput, + message: { + ...turnInput.message, + text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, + }, + }) + : Effect.succeed([]), ); }, (effect, turnInput) => @@ -7641,7 +7668,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? parseGoalCommand(turnInput.message) : null; if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand); + const nativeThreadId = turnInput.providerThread.nativeThreadRef?.nativeId; return Effect.gen(function* () { + if ( + turnInput.restartContinuationOfRunId === undefined && + nativeThreadId != null && + goalsPausedForWake.has(nativeThreadId) && + goalsByNativeThread.get(nativeThreadId)?.status === "paused" + ) { + return yield* activateGoal(turnInput, nativeThreadId, toCodexInput(turnInput)); + } yield* startNativeTurn( turnInput, turnInput.restartContinuationOfRunId === undefined diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5e494b8b31904..50019ad69f468 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -17,6 +17,7 @@ import { ProviderDriverKind, ProviderInstanceId, type ProviderSessionId, + RunId, ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -2277,6 +2278,43 @@ it.effect("ProviderSessionManagerV2 fails a turn whose provider retry never resu }), ); +it("ProviderSessionManagerV2 counts a delegated task as a wake only while its result can be delivered", () => { + const runId = RunId.make("run-delegating"); + const thread = { archivedAt: null, deletedAt: null }; + const task = { + origin: "app_owned" as const, + status: "running" as const, + runId, + completionDelivery: undefined, + }; + const run = (disposition: "open" | "disposed") => ({ + id: runId, + delegatedCompletion: { disposition, nextGeneration: 1, delivery: null }, + }); + assert.isTrue( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("open")], + subagents: [task], + }), + ); + // An interrupted completion wake disposes the cohort, so the result never wakes the thread. + assert.isFalse( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("disposed")], + subagents: [task], + }), + ); + assert.isFalse( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("open")], + subagents: [{ ...task, origin: "provider_native" }], + }), + ); +}); + it.effect( "ProviderSessionManagerV2 issues MCP credentials before opening and revokes them on close", () => @@ -2322,6 +2360,8 @@ it.effect( new Set(["preview", "orchestration", "worktree", "pull-requests"]), ); + assert.isFalse(yield* captured!.hasPendingWake!); + yield* manager.close(providerSessionId); assert.isUndefined(yield* mcpSessions.read(threadId)); assert.isUndefined(yield* registry.resolve(token!)); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 33088785615a4..9b2a61d0dd2d6 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -4,11 +4,14 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, ModelSelection, + type OrchestrationV2AppThread, OrchestrationV2DomainEvent, OrchestrationV2ProviderSession, type OrchestrationV2PromptSuggestion, type OrchestrationV2ProviderThread, + type OrchestrationV2Run, OrchestrationV2RuntimeRequest, + type OrchestrationV2Subagent, ProviderInstanceId, ProviderSessionId, ThreadId, @@ -86,6 +89,30 @@ const WORKING_RUN_STATUSES: ReadonlySet = new Set([ "waiting", ]); +/** Matches planDelegatedCompletionDelivery: true while a delegated task's result still wakes the thread. */ +export const delegatedTaskWillWake = (records: { + readonly thread: Pick; + readonly runs: ReadonlyArray>; + readonly subagents: ReadonlyArray< + Pick + >; +}): boolean => + records.thread.archivedAt === null && + records.thread.deletedAt === null && + records.subagents.some((task) => { + const run = records.runs.find((candidate) => candidate.id === task.runId); + const delivery = task.completionDelivery?.state; + return ( + task.origin === "app_owned" && + isOrchestrationV2WorkActive(task.status) && + run !== undefined && + (run.delegatedCompletion?.disposition ?? "open") === "open" && + delivery !== "acknowledged" && + delivery !== "delivered" && + delivery !== "disposed" + ); + }); + export const ProviderSessionReleaseReason = Schema.Literals([ "idle_timeout", "runtime_error", @@ -551,7 +578,15 @@ export const layerWithOptions = ( capabilities, subagent, }); - yield* mcpSessions.set(credential.config); + yield* mcpSessions.set({ + ...credential.config, + hasPendingWake: projectionStore + .getThreadRecords(threadId, ["runs", "subagents"]) + .pipe( + Effect.map(delegatedTaskWillWake), + Effect.orElseSucceed(() => false), + ), + }); reserveMcpCredential(threadId, credential.config.providerSessionId); return { mcpCredentialId: credential.config.providerSessionId, issued: true }; }), diff --git a/packages/provider-core/src/server/mcpSession.ts b/packages/provider-core/src/server/mcpSession.ts index e8eba0bfeeeb6..20d17df610fd7 100644 --- a/packages/provider-core/src/server/mcpSession.ts +++ b/packages/provider-core/src/server/mcpSession.ts @@ -1,4 +1,5 @@ import type { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import type * as Effect from "effect/Effect"; /** * Sent by MCP clients that cannot defer tools themselves. The T3 MCP server then @@ -17,6 +18,8 @@ export interface McpProviderSessionConfig { readonly browserToolsAvailable: boolean; readonly capabilities?: ReadonlySet; readonly agentDeviceEnvironment?: Readonly>; + /** True while a delegated task this thread started will wake it with its result. */ + readonly hasPendingWake?: Effect.Effect; } export function withAgentDeviceEnvironment( From 29919b3eda7687ccc7252b7a59f951fadc7e7732 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:34:15 +0000 Subject: [PATCH 2/2] fix(server): decide a Codex goal's pending wake before a continuation can claim it The pending-wake lookup ran in a fork after the goal hold was published, so a continuation Codex started meanwhile could adopt the hold and keep the loop going. The lookup now runs under the terminalization permit before the hold exists, and a pending wake claims the hold in the same critical section, so a racing continuation is stopped as an orphan. Inspecting the goal with /goal no longer drops the marker that resumes a goal paused for a delegated task's wake. AI-assisted; verified with the tests and tools listed above. --- .../Adapters/CodexAdapterV2.test.ts | 49 +++++++++++++++++-- .../Adapters/CodexAdapterV2.ts | 43 ++++++++++------ 2 files changed, 73 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 52146a6f72098..15fc4d4eb5697 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9884,6 +9884,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { it.effect("pauses a goal at once while a delegated task will wake the thread", () => Effect.gen(function* () { + const lookupStarted = yield* Deferred.make(); + const lookupAnswered = yield* Deferred.make(); const turnStart = (id: number, text: string) => Effect.map( CodexAdapterV2.buildCodexTurnStartParams({ @@ -9919,21 +9921,44 @@ describe("CodexAdapterV2 post-settle continuation", () => { threadId: nativeThreadId, turn: makeCodexReplayTurn({ id: "goal-turn", status: "completed" }), }), + // Codex continues before the wake lookup answers. + notification("continuation", "turn/started", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn-2", status: "inProgress" }), + }), ...request( 4, "thread/goal/set", { threadId: nativeThreadId, status: "paused" }, { goal: codexGoal("paused", 10) }, ), - yield* turnStart(5, "Child finished"), + ...request( + 5, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 10) }, + ), + ...request( + 6, + "turn/interrupt", + { threadId: nativeThreadId, turnId: "goal-turn-2" }, + {}, + ), + ...request( + 7, + "thread/goal/get", + { threadId: nativeThreadId }, + { goal: codexGoal("paused", 10) }, + ), + yield* turnStart(8, "Child finished"), ...withReplayRequestId( codexReplayPreamble({ nativeThreadId, nativeTurnId: "wake-turn", prompt: "" }).slice( 6, ), - 5, + 8, ), ...request( - 6, + 9, "thread/goal/set", { threadId: nativeThreadId, status: "active" }, { goal: codexGoal("active", 10) }, @@ -9954,9 +9979,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { endpoint: "http://127.0.0.1:43123/mcp", authorizationHeader: "Bearer test", browserToolsAvailable: true, - hasPendingWake: Effect.succeed(true), + hasPendingWake: Deferred.succeed(lookupStarted, undefined).pipe( + Effect.andThen(Deferred.await(lookupAnswered)), + Effect.as(true), + ), }); yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "Delegate and wait")); + yield* Deferred.await(lookupStarted); + yield* Deferred.succeed(lookupAnswered, undefined); // Settles without waiting out the continuation grace. yield* harness.firstTerminal; assert.deepEqual(providerGoals(harness.events), ["active", "paused"]); @@ -9964,7 +9994,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { harness.terminalEvents().map((event) => event.status), ["completed"], ); + yield* harness.until(() => harness.terminalEvents().length === 1); + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running" + ? [event.providerTurn.ordinal] + : [], + ), + [1], + ); + // Inspecting the goal must not cancel its resume on the wake. + yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "/goal")); yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "Child finished")); yield* harness.until(() => providerGoals(harness.events).at(-1) === "active"); }).pipe( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4ce8c6ac8525a..95c6eca8c0c7a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6253,21 +6253,30 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( event.status === "completed" && nativeThreadId != null ) { + const hasPendingWake = (yield* mcpSessions.read(input.context.input.threadId)) + ?.hasPendingWake; + // Decided under the permit before the hold exists, so no continuation can adopt it first. + const waitsForWake = hasPendingWake === undefined ? false : yield* hasPendingWake; goalHolds.set(nativeThreadId, { context: input.context, event, completedTurn: input.goalHoldTurn, next: yield* Deferred.make(), }); - const hasPendingWake = (yield* mcpSessions.read(input.context.input.threadId)) - ?.hasPendingWake; // A continuation turn would only report that the delegated task is still running. - yield* (hasPendingWake ?? Effect.succeed(false)).pipe( - Effect.tap((waitsForWake) => - waitsForWake ? Effect.void : Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE), - ), - Effect.flatMap((waitsForWake) => - pauseUncontinuedGoal(nativeThreadId, event, waitsForWake), + const claimed = waitsForWake ? claimGoalHold(nativeThreadId, event) : undefined; + if (claimed !== undefined) { + yield* pauseClaimedGoal(nativeThreadId, claimed, true).pipe(Effect.forkIn(scope)); + return; + } + yield* Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe( + Effect.andThen( + Effect.suspend(() => { + const hold = claimGoalHold(nativeThreadId, event); + return hold === undefined + ? Effect.void + : pauseClaimedGoal(nativeThreadId, hold, false); + }), ), Effect.andThen( turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)), @@ -6283,22 +6292,26 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // A goal must not read as active once its run ends. Claiming the hold first // makes a turn Codex starts during the pause an orphan, which is stopped. - const pauseUncontinuedGoal = Effect.fnUntraced(function* ( - nativeThreadId: string, - expected: CodexRootTerminalEvent, - waitsForWake: boolean, - ) { + const claimGoalHold = (nativeThreadId: string, expected: CodexRootTerminalEvent) => { const hold = goalHolds.get(nativeThreadId); if ( hold?.event !== expected || goalStops.has(nativeThreadId) || goalsByNativeThread.get(nativeThreadId)?.status !== "active" ) { - return; + return undefined; } goalHolds.delete(nativeThreadId); // Keeps a turn Codex starts before settlement an orphan once the goal reads paused. goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); + return hold; + }; + + const pauseClaimedGoal = Effect.fnUntraced(function* ( + nativeThreadId: string, + hold: CodexGoalHold, + waitsForWake: boolean, + ) { if (!waitsForWake) { yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", { nativeThreadId, @@ -7316,7 +7329,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (command.type === "set") yield* validateCodexGoalObjective(command.objective); const threadId = yield* getNativeThreadId(turnInput.providerThread); rootProviderThreads.set(threadId, turnInput.providerThread); - goalsPausedForWake.delete(threadId); const readGoal = client .request("thread/goal/get", { threadId }) .pipe( @@ -7332,6 +7344,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( current === null ? "No goal is set." : describeCodexGoal(current), ); } + goalsPausedForWake.delete(threadId); if (command.type === "clear") { const { cleared } = yield* client.request("thread/goal/clear", { threadId }); goalsByNativeThread.set(threadId, null);