Skip to content

Commit fc5b293

Browse files
shivamhwpshivamhwp
committed
feat(orchestration-v2): attribute reused subagents to the run driving them
A subagent row keeps the run id it was spawned under. Routing admits a row by run or by parent thread, and the parent thread is excluded by construction, so a later run re-activating that subagent could only see its updates while the spawning run's ingestion fiber happened to still be alive. The Claude adapter already re-attributed on reopen for exactly this reason; this brings the same treatment to routing and to Codex. Reuse arrives on item/completed, not item/started: resumeAgent and sendInput only ever complete, and registerSubagentThreads ignores them because they are not spawnAgent. Hooking the rebind there is what makes it fire — subagent_continue now asserts the subagent is attributed to the run that drove it, where before it stayed pinned to the spawning run. Routing gains a third test: subagent identity, seeded from the projection at turn start. The child thread is adopted only while the agent is live, because a settled row is also re-emitted by trailing traffic — a token usage frame or a collab state sweep — and adopting on those would hand a later run an interrupted agent's thread, readmitting the stale events that post-interrupt recovery exists to exclude. A subagent's row and its timeline item are emitted together, so both route by identity; matching only the row advanced the agent while its item stayed frozen on the spawning run. The reactivation predicate is separate from canRouteRelatedSubagent rather than a widening of it, for the same reason: an interrupted agent must not have its child thread pre-owned, but its identity is still resumable, because the user stopped it rather than losing it. Rehydration seeds turn ordinals alongside the registry — restarting them at 1 would re-derive childRootNodeId for a different native turn — and parks non-terminal leftovers at idle, since a new session can never terminalize an activation it did not drive and would otherwise keep the session pinned open. Not proven: rehydration itself. The fixture that would exercise it is recorded but does not replay to completion, for reasons unrelated to these changes, and is left unregistered with the open question written up in fixtures/index.ts.
1 parent f3666e9 commit fc5b293

12 files changed

Lines changed: 903 additions & 24 deletions

File tree

‎apps/server/scripts/record-codex-app-server-replay-fixture.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,10 @@ import {
2727
SUBAGENT_CONTINUE_PARENT_PROMPT,
2828
SUBAGENT_CONTINUE_PROMPT,
2929
SUBAGENT_PROMPT,
30+
SUBAGENT_REUSE_AFTER_IDLE_PROMPT,
31+
SUBAGENT_REUSE_AFTER_IDLE_RESUME_PROMPT,
32+
SUBAGENT_REUSE_AFTER_INTERRUPT_PROMPT,
33+
SUBAGENT_REUSE_AFTER_INTERRUPT_RESUME_PROMPT,
3034
THREAD_ROLLBACK_AFTER_PROMPT,
3135
THREAD_ROLLBACK_FIRST_PROMPT,
3236
THREAD_ROLLBACK_SECOND_PROMPT,
@@ -71,6 +75,8 @@ const SCENARIO_NAMES = [
7175
"tool_call_restricted_granular",
7276
"subagent",
7377
"subagent_continue",
78+
"subagent_reuse_after_interrupt",
79+
"subagent_reuse_after_idle",
7480
"multi_turn",
7581
"provider_thread_resume",
7682
"todo_list",
@@ -443,6 +449,61 @@ function scenarios(): ReadonlyArray<ReplayScenario> {
443449
},
444450
],
445451
},
452+
{
453+
name: "subagent_reuse_after_interrupt",
454+
fileName: "subagent_reuse_after_interrupt.ndjson",
455+
description:
456+
"A root turn spawns a native Codex subagent and is interrupted while it works, then a later root turn re-activates the same child thread.",
457+
runs: [
458+
{
459+
name: "reuse-after-interrupt",
460+
description:
461+
"Interrupt the spawning turn mid-subagent, then message the same subagent from a new root turn.",
462+
turnDefaults: {
463+
approvalPolicy: "never",
464+
sandboxPolicy: workspaceWriteSandbox(),
465+
},
466+
steps: [
467+
{
468+
type: "interruptedTurn",
469+
label: "spawn-subagent-then-interrupt",
470+
prompt: SUBAGENT_REUSE_AFTER_INTERRUPT_PROMPT,
471+
interruptAfterCommandExecutionStarted: true,
472+
},
473+
{
474+
type: "turn",
475+
label: "reuse-interrupted-subagent",
476+
prompt: SUBAGENT_REUSE_AFTER_INTERRUPT_RESUME_PROMPT,
477+
},
478+
],
479+
},
480+
],
481+
},
482+
{
483+
name: "subagent_reuse_after_idle",
484+
fileName: "subagent_reuse_after_idle.ndjson",
485+
description:
486+
"A root turn spawns a native Codex subagent, then a later root turn re-activates it. Replay advances past the session idle timeout between turns so the adapter registry is empty on reuse.",
487+
runs: [
488+
{
489+
name: "reuse-after-idle",
490+
description:
491+
"Second root turn messages the same subagent after the session would have been reaped.",
492+
steps: [
493+
{
494+
type: "turn",
495+
label: "spawn-subagent",
496+
prompt: SUBAGENT_REUSE_AFTER_IDLE_PROMPT,
497+
},
498+
{
499+
type: "turn",
500+
label: "reuse-subagent-after-idle",
501+
prompt: SUBAGENT_REUSE_AFTER_IDLE_RESUME_PROMPT,
502+
},
503+
],
504+
},
505+
],
506+
},
446507
{
447508
name: "multi_turn",
448509
fileName: "multi_turn.ndjson",

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts‎

Lines changed: 141 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -119,12 +119,6 @@ export const CODEX_DRIVER_KIND = CODEX_PROVIDER;
119119
export const CODEX_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CODEX_DRIVER_KIND);
120120
const DEFAULT_CODEX_SETTINGS = Schema.decodeSync(CodexSettings)({});
121121
const CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS = 50;
122-
const isSettledCodexSubagent = (subagent: OrchestrationV2Subagent) =>
123-
subagent.status === "idle" ||
124-
subagent.status === "completed" ||
125-
subagent.status === "failed" ||
126-
subagent.status === "cancelled" ||
127-
subagent.status === "interrupted";
128122
const CodexBackgroundTerminalTerminateResponse = Schema.Struct({
129123
terminated: Schema.Boolean,
130124
});
@@ -839,7 +833,9 @@ interface DeferredCodexRootTerminal {
839833
}
840834

841835
interface CodexSubagentThreadContext {
842-
readonly parentContext: ActiveCodexTurnContext;
836+
// Rebound whenever a later root turn re-activates this subagent, so its rows
837+
// are attributed to the run that is actually driving it.
838+
parentContext: ActiveCodexTurnContext;
843839
readonly providerThread: OrchestrationV2ProviderThread;
844840
readonly childThread: OrchestrationV2AppThread;
845841
readonly subagentNodeId: OrchestrationV2ExecutionNode["id"];
@@ -1502,6 +1498,74 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
15021498
const emitProviderEvent = (event: ProviderAdapterV2Event) =>
15031499
Queue.offer(events, event).pipe(Effect.asVoid);
15041500

1501+
/**
1502+
* Rebuild registry entries for subagents that already exist in the
1503+
* projection. The registry is process-local, so without this a session
1504+
* reap or server restart orphans every prior agent: the returning
1505+
* native thread is unrecognised and gets spawned as a new identity.
1506+
* Entries created here emit nothing — they only let a later
1507+
* re-activation resolve to the original subagent.
1508+
*/
1509+
const seedExistingSubagents = (context: ActiveCodexTurnContext) =>
1510+
Effect.gen(function* () {
1511+
const existing = context.input.existingSubagents ?? [];
1512+
if (existing.length === 0) return;
1513+
yield* Ref.update(subagentThreads, (current) => {
1514+
const updated = new Map(current);
1515+
for (const entry of existing) {
1516+
const nativeThreadId = entry.childProviderThread.nativeThreadRef?.nativeId ?? null;
1517+
if (nativeThreadId === null || updated.has(nativeThreadId)) continue;
1518+
const nativeItemId = entry.subagent.nativeTaskRef?.nativeId ?? null;
1519+
if (nativeItemId === null || !nativeItemId.includes(":")) continue;
1520+
updated.set(nativeThreadId, {
1521+
parentContext: context,
1522+
providerThread: entry.childProviderThread,
1523+
childThread: entry.childThread,
1524+
subagentNodeId: entry.subagent.id,
1525+
childRootNodeId: idAllocator.derive.nodeFromProviderItem({
1526+
driver: CODEX_PROVIDER,
1527+
nativeItemId: `${nativeItemId}:thread-root`,
1528+
}),
1529+
childThreadId: entry.childThread.id,
1530+
nativeToolCallId: nativeItemId.slice(0, nativeItemId.indexOf(":")),
1531+
ordinal: entry.ordinal,
1532+
startedAt: entry.subagent.startedAt ?? context.startedAt,
1533+
turnItemId: entry.turnItemId,
1534+
turnItemOrdinal: entry.turnItemOrdinal,
1535+
task: {
1536+
...entry.subagent,
1537+
runId: context.projectionRunId,
1538+
parentNodeId: context.itemParentNodeId,
1539+
// A non-terminal status here is a leftover from the session
1540+
// that died; this one never drove that activation and can
1541+
// never terminalize it. Left as running it would keep
1542+
// hasPendingBackgroundWork true and pin the session open.
1543+
status:
1544+
entry.subagent.status === "pending" ||
1545+
entry.subagent.status === "running" ||
1546+
entry.subagent.status === "waiting"
1547+
? "idle"
1548+
: entry.subagent.status,
1549+
currentActivationId: null,
1550+
},
1551+
});
1552+
}
1553+
return updated;
1554+
});
1555+
// Turn ordinals live in the same lost registry. Restarting them at
1556+
// 1 would re-derive childRootNodeId for a different native turn, so
1557+
// continue past the activations already recorded.
1558+
yield* Ref.update(nextProviderTurnOrdinals, (current) => {
1559+
const updated = new Map(current);
1560+
for (const entry of existing) {
1561+
const key = String(entry.childProviderThread.id);
1562+
if (updated.has(key)) continue;
1563+
updated.set(key, Math.max(1, entry.subagent.activationCount));
1564+
}
1565+
return updated;
1566+
});
1567+
});
1568+
15051569
const registerRootTurn = (input: {
15061570
readonly turnInput: ProviderAdapterV2TurnInput;
15071571
readonly nativeTurnId: string;
@@ -1539,6 +1603,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
15391603
updated.set(input.nativeTurnId, context);
15401604
return updated;
15411605
});
1606+
yield* seedExistingSubagents(context);
15421607
yield* emitProviderEvent({
15431608
type: "provider_turn.updated",
15441609
driver: CODEX_PROVIDER,
@@ -2009,7 +2074,11 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
20092074
}) =>
20102075
Effect.gen(function* () {
20112076
const registeredSubagents = yield* Ref.get(subagentThreads);
2012-
if (registeredSubagents.has(input.nativeThreadId)) {
2077+
const existing = registeredSubagents.get(input.nativeThreadId);
2078+
if (existing !== undefined) {
2079+
// Same agent thread referenced again by a later turn: keep the
2080+
// identity and adopt it into the run now driving it.
2081+
rebindSubagentToContext(existing, input.context);
20132082
return;
20142083
}
20152084

@@ -2254,6 +2323,39 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
22542323
}
22552324
});
22562325

2326+
/**
2327+
* Adopt an already-registered subagent into the root turn that is
2328+
* re-activating it. The task row keeps the run it was spawned under
2329+
* otherwise, and every later row would be attributed to a closed run:
2330+
* routing drops it, and the current run stops counting the subagent as
2331+
* open work and can finalize while it is still going.
2332+
*/
2333+
const rebindSubagentToContext = (
2334+
subagent: CodexSubagentThreadContext,
2335+
context: ActiveCodexTurnContext,
2336+
) => {
2337+
if (subagent.parentContext.projectionRunId === context.projectionRunId) return;
2338+
subagent.parentContext = context;
2339+
subagent.task = {
2340+
...subagent.task,
2341+
runId: context.projectionRunId,
2342+
parentNodeId: context.itemParentNodeId,
2343+
};
2344+
};
2345+
2346+
const rebindSubagentsToCurrentTurn = (input: {
2347+
readonly context: ActiveCodexTurnContext;
2348+
readonly item: CodexCollabAgentToolCallItem;
2349+
}) =>
2350+
Effect.gen(function* () {
2351+
if (input.item.receiverThreadIds.length === 0) return;
2352+
const registered = yield* Ref.get(subagentThreads);
2353+
for (const nativeThreadId of input.item.receiverThreadIds) {
2354+
const subagent = registered.get(nativeThreadId);
2355+
if (subagent !== undefined) rebindSubagentToContext(subagent, input.context);
2356+
}
2357+
});
2358+
22572359
const registerSubagentActivity = (input: {
22582360
readonly context: ActiveCodexTurnContext;
22592361
readonly item: CodexSubAgentActivityItem;
@@ -2287,6 +2389,17 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
22872389
return;
22882390
}
22892391

2392+
// "interacted" is how Codex reports a root turn messaging an agent
2393+
// it did not spawn. It is the only reuse signal on this path — the
2394+
// collab tool call that accompanies it carries neither
2395+
// receiverThreadIds nor agentsStates — and it arrives before the
2396+
// child turn starts, so adopting the identity here is what lets the
2397+
// resulting activation be attributed to the run driving it.
2398+
if (input.item.kind === "interacted") {
2399+
rebindSubagentToContext(subagent, input.context);
2400+
return;
2401+
}
2402+
22902403
if (input.item.kind === "interrupted") {
22912404
yield* emitSubagentTaskUpdate({
22922405
subagent,
@@ -3291,7 +3404,13 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
32913404
yield* client.handleServerNotification("thread/status/changed", (payload) =>
32923405
Effect.gen(function* () {
32933406
const subagent = (yield* Ref.get(subagentThreads)).get(payload.threadId);
3294-
if (subagent === undefined || isSettledCodexSubagent(subagent.task)) return;
3407+
// Live status applies only while an activation is in flight. Every
3408+
// resting and terminal state clears currentActivationId, so this
3409+
// single check drops trailing frames for an idle or finished
3410+
// identity while letting a re-activated one resume immediately.
3411+
if (subagent === undefined || subagent.task.currentActivationId === null) {
3412+
return;
3413+
}
32953414
const status =
32963415
payload.status.type === "active"
32973416
? payload.status.activeFlags.length > 0
@@ -3347,6 +3466,14 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
33473466
}
33483467
}
33493468

3469+
// A collab tool call that targets an existing agent thread is a
3470+
// re-activation; adopt it into this turn before its child turn
3471+
// starts so the activation is attributed to the current run.
3472+
if (payload.item.type === "collabAgentToolCall") {
3473+
yield* rebindSubagentsToCurrentTurn({ context, item: payload.item });
3474+
return;
3475+
}
3476+
33503477
if (payload.item.type === "subAgentActivity") {
33513478
yield* registerSubagentActivity({
33523479
context,
@@ -3608,6 +3735,11 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
36083735
context,
36093736
item: payload.item,
36103737
});
3738+
// Reuse arrives here, not on item/started: resumeAgent and
3739+
// sendInput only ever complete. registerSubagentThreads ignores
3740+
// them because they are not spawnAgent, so without this the
3741+
// returning agent keeps the run it was spawned under.
3742+
yield* rebindSubagentsToCurrentTurn({ context, item: payload.item });
36113743
yield* updateSubagentStates({
36123744
item: payload.item,
36133745
});

‎apps/server/src/orchestration-v2/ProviderAdapter.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -389,6 +389,21 @@ export interface ProviderAdapterV2EnsureThreadInput {
389389
readonly existingProviderThread?: OrchestrationV2ProviderThread;
390390
}
391391

392+
/**
393+
* A reusable subagent that already exists in the projection, supplied so an
394+
* adapter whose in-memory registry was lost (session reap, server restart) can
395+
* still recognise the agent thread and re-activate the same identity instead of
396+
* spawning a duplicate.
397+
*/
398+
export interface ProviderAdapterV2ExistingSubagent {
399+
readonly subagent: OrchestrationV2Subagent;
400+
readonly childThread: OrchestrationV2AppThread;
401+
readonly childProviderThread: OrchestrationV2ProviderThread;
402+
readonly turnItemId: TurnItemId;
403+
readonly turnItemOrdinal: number;
404+
readonly ordinal: number;
405+
}
406+
392407
export interface ProviderAdapterV2TurnInput {
393408
readonly appThread: OrchestrationV2AppThread;
394409
readonly threadId: ThreadId;
@@ -401,6 +416,7 @@ export interface ProviderAdapterV2TurnInput {
401416
readonly message: ProviderAdapterV2TurnMessage;
402417
readonly modelSelection: ModelSelection;
403418
readonly runtimePolicy: ProviderAdapterV2RuntimePolicy;
419+
readonly existingSubagents?: ReadonlyArray<ProviderAdapterV2ExistingSubagent>;
404420
}
405421

406422
export interface ProviderAdapterV2SteerInput {

0 commit comments

Comments
 (0)