Repository navigation
fix(server): Codex goals wait for a delegated task instead of re-prompting #34898
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2045,6 +2045,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( | |
| const goalActivations = new Map<string, { stopped: boolean }>(); | ||
| // Held runs must not complete while Stop resolves its target. | ||
| const goalStops = new Map<string, number>(); | ||
| // Goals paused because a delegated task will wake the thread; the next ordinary turn resumes them. | ||
| const goalsPausedForWake = new Set<string>(); | ||
| const latestRunTurnId = (providerTurnId: ProviderTurnId) => | ||
| continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; | ||
| const isContinuedRunDescendant = ( | ||
|
|
@@ -6251,14 +6253,31 @@ 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<ActiveCodexTurnContext | undefined>(), | ||
| }); | ||
| // A continuation turn would only report that the delegated task is still running. | ||
| 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(pauseUncontinuedGoal(nativeThreadId, event)), | ||
| 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)), | ||
| ), | ||
|
|
@@ -6273,31 +6292,40 @@ 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, | ||
| ) { | ||
| 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); | ||
| yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", { | ||
| nativeThreadId, | ||
| }); | ||
| 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, | ||
| }); | ||
| } | ||
| yield* client | ||
| .request("thread/goal/set", { threadId: nativeThreadId, status: "paused" }) | ||
| .pipe( | ||
| // The hold is already claimed, so an unanswered pause must not keep the run open. | ||
| 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 +7275,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<typeof toCodexInput>, | ||
| ) { | ||
| goalsPausedForWake.delete(threadId); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Keep the wake marker until the goal activation succeeds.
Clear the marker only after activation succeeds. Clear it also when Stop cancels activation, so a stopped goal is not reactivated later. A failed activation then keeps the marker for the next ordinary turn. Suggested fix
|
||
| // 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 | ||
|
|
@@ -7275,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); | ||
|
|
@@ -7302,50 +7372,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 +7681,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 | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Persist the paused-for-wake state across adapter sessions.
goalsPausedForWakelives only in one Codex adapter session. The session manager can release an idle session while a delegated task is still pending. The idle probe does not count an open delegated completion. After the release, the wake starts its turn on a new adapter with an empty set. ThestartTurncheck at line 7689 then skipsactivateGoal, and the goal stays paused. A server restart clears the set in the same way.Store the paused-for-wake state outside the adapter session, or rebuild it from durable state. The activation check also requires the goal status to read
"paused"fromgoalsByNativeThread. A new adapter does not read the goal state on this path, so the fix must cover that status as well.