History

fix(orchestration-v2): preserve subagent progress state

Shivam Sharma committed 3 months ago

Commit c0a38521 parent 3f829675 files · +71 −15Browse files

Unified · Split

5 files
  1. apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +2 −2
  2. apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +24 −0
  3. apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +30 −12
  4. apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson +1 −1
  5. apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts +14 −0
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+2−2
@@ -2790,11 +2790,11 @@ export function makeClaudeAdapterV2( · 2789 unmodified lines
27902790 status: input.status,
27912791 title: "Subagent progress",
27922792 startedAt: progressStartedAt,
2793− completedAt: input.status === "running" ? null : now,
2793+ completedAt: settled ? now : null,
27942794 updatedAt: now,
27952795 type: "reasoning",
27962796 text: progress,
2797− streaming: input.status === "running",
2797+ streaming: !settled,
27982798 },
27992799 });
28002800 }
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts+24−0
@@ -3577,6 +3577,21 @@ describe("CodexAdapterV2 post-settle continuation", () => { · 3576 unmodified lines
35773577 assert.equal(firstCompletion?.subagent.activationCount, 1);
35783578 assert.equal(firstCompletion?.subagent.usage?.totalTokens, 100);
35793579 assert.isNull(firstCompletion?.subagent.currentActivationId);
3580+ const firstActivationUpdates = harness
3581+ .subagentActivationUpdates()
3582+ .filter((event) => event.activation.ordinal === 1);
3583+ const firstActivation = firstActivationUpdates.at(-1)?.activation;
3584+ assert.equal(firstActivation?.ordinal, 1);
3585+ assert.equal(firstActivation?.status, "completed");
3586+ assert.equal(firstActivation?.usage?.totalTokens, 100);
3587+ assert.equal(
3588+ firstActivation?.completedAt,
3589+ firstActivationUpdates.at(-2)?.activation.completedAt,
3590+ );
3591+ assert.equal(
3592+ firstActivation?.updatedAt,
3593+ firstActivationUpdates.at(-2)?.activation.updatedAt,
3594+ );
35803595 assert.lengthOf(
35813596 new Set(
35823597 settledUpdates
@@ -3629,6 +3644,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { · 46 unmodified lines
36293644 [RESUME_CHILD_TURN_1, RESUME_CHILD_TURN_2],
36303645 );
36313646 assert.equal(activations.at(-1)?.activation.status, "completed");
3647+ assert.equal(activations.at(-1)?.activation.usage?.totalTokens, 80);
3648+ assert.equal(
3649+ activations.at(-1)?.activation.completedAt,
3650+ activations.at(-2)?.activation.completedAt,
3651+ );
3652+ assert.equal(
3653+ activations.at(-1)?.activation.updatedAt,
3654+ activations.at(-2)?.activation.updatedAt,
3655+ );
36323656 assert.isFalse(yield* harness.hasPendingBackgroundWork);
36333657 assert.lengthOf(harness.terminalEvents(), 1);
36343658 assert.lengthOf(harness.continuationRequests, 0);
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts+30−12
@@ -1517,6 +1517,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi · 1516 unmodified lines
15171517 const pendingRootTurns = yield* Ref.make(new Map<string, ProviderAdapterV2TurnInput>());
15181518 const turnWaiters = yield* Ref.make(new Map<string, Deferred.Deferred<void, never>>());
15191519 const subagentThreads = yield* Ref.make(new Map<string, CodexSubagentThreadContext>());
1520+ const subagentActivationsByNativeTurnId = yield* Ref.make(
1521+ new Map<string, OrchestrationV2SubagentActivation>(),
1522+ );
15201523 const pendingSubagentTurns = yield* Ref.make(
15211524 new Map<string, ReadonlyArray<PendingCodexSubagentTurnStarted>>(),
15221525 );
@@ -1958,6 +1961,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi · 435 unmodified lines
19581961 updatedAt: turn.startedAt,
19591962 } satisfies OrchestrationV2SubagentActivation;
19601963 activeContext.subagentActivation = activation;
1964+ yield* Ref.update(subagentActivationsByNativeTurnId, (current) =>
1965+ new Map(current).set(turn.nativeTurnId, activation),
1966+ );
19611967 yield* Ref.update(activeTurns, (current) => {
19621968 const updated = new Map(current);
19631969 updated.set(turn.nativeTurnId, activeContext);
@@ -3380,30 +3386,36 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi · 1416 unmodified lines
33803386 });
33813387
33823388 const activeContext = (yield* Ref.get(activeTurns)).get(payload.turnId);
3383− if (
3384− activeContext === undefined ||
3385− activeContext.subagent !== subagent ||
3386− activeContext.subagentActivation === null
3387− ) {
3389+ const activation =
3390+ activeContext?.subagent === subagent
3391+ ? activeContext.subagentActivation
3392+ : ((yield* Ref.get(subagentActivationsByNativeTurnId)).get(payload.turnId) ?? null);
3393+ if (activation === null) {
33883394 return;
33893395 }
33903396 const last = payload.tokenUsage.last;
3391− const activation = {
3392− ...activeContext.subagentActivation,
3393− usage: {
3397+ const now = yield* DateTime.now;
3398+ const updatedActivation = {
3399+ ...activation,
3400+ usage: mergeCumulativeSubagentUsage(activation.usage, {
33943401 totalTokens: last.totalTokens,
33953402 inputTokens: last.inputTokens,
33963403 cachedInputTokens: last.cachedInputTokens,
33973404 outputTokens: last.outputTokens,
33983405 reasoningOutputTokens: last.reasoningOutputTokens,
3399− },
3400− updatedAt: yield* DateTime.now,
3406+ }),
3407+ updatedAt: activation.completedAt === null ? now : activation.updatedAt,
34013408 } satisfies OrchestrationV2SubagentActivation;
3402− activeContext.subagentActivation = activation;
3409+ if (activeContext?.subagent === subagent) {
3410+ activeContext.subagentActivation = updatedActivation;
3411+ }
3412+ yield* Ref.update(subagentActivationsByNativeTurnId, (current) =>
3413+ new Map(current).set(payload.turnId, updatedActivation),
3414+ );
34033415 yield* emitProviderEvent({
34043416 type: "subagent_activation.updated",
34053417 driver: CODEX_PROVIDER,
3406− activation,
3418+ activation: updatedActivation,
34073419 });
34083420 }).pipe(Effect.orDie),
34093421 );
@@ -3435,6 +3447,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi · 25 unmodified lines
34353447 updatedAt: now,
34363448 } satisfies OrchestrationV2SubagentActivation;
34373449 activeContext.subagentActivation = activation;
3450+ yield* Ref.update(subagentActivationsByNativeTurnId, (current) =>
3451+ new Map(current).set(activeContext.nativeTurnId, activation),
3452+ );
34383453 yield* emitProviderEvent({
34393454 type: "subagent_activation.updated",
34403455 driver: CODEX_PROVIDER,
@@ -4392,6 +4407,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi · 951 unmodified lines
43924407 updatedAt: input.completedAt,
43934408 } satisfies OrchestrationV2SubagentActivation;
43944409 input.context.subagentActivation = activation;
4410+ yield* Ref.update(subagentActivationsByNativeTurnId, (current) =>
4411+ new Map(current).set(input.nativeTurnId, activation),
4412+ );
43954413 yield* emitProviderEvent({
43964414 type: "subagent_activation.updated",
43974415 driver: CODEX_PROVIDER,
apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson+1−1
@@ -4,7 +4,7 @@ · 3 unmodified lines
44 {"type": "emit_inbound", "label": "system", "frame": {"type": "system", "subtype": "init", "agents": [], "apiKeySource": "none", "claude_code_version": "2.1.183", "cwd": "/tmp/claude-replay-claude_workflow", "tools": [], "mcp_servers": [], "model": "claude-sonnet-4-6", "permissionMode": "bypassPermissions", "slash_commands": [], "output_style": "default", "skills": [], "plugins": [], "fast_mode_state": "off", "uuid": "af3bf4cc-1385-4e19-a5fa-fcbe2a0ca201", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
55 {"type": "emit_inbound", "label": "assistant:commentary", "frame": {"type": "assistant", "message": {"model": "claude-sonnet-4-6", "id": "msg_01claudeworkflow1", "type": "message", "role": "assistant", "content": [{"type": "text", "text": "Starting the two-phase workflow."}], "stop_reason": null, "stop_sequence": null, "stop_details": null, "usage": {"input_tokens": 1, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0, "output_tokens": 1, "service_tier": "standard", "inference_geo": "not_available"}, "context_management": null}, "parent_tool_use_id": null, "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201", "uuid": "97898e28-ee4e-4fcf-80c3-5c790c502201"}}
66 {"type": "emit_inbound", "label": "task_started:local_workflow", "frame": {"type": "system", "subtype": "task_started", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Research and implement", "task_type": "local_workflow", "prompt": "Research the parser, then implement the fix.", "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73201", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
7−{"type": "emit_inbound", "label": "task_progress:phase-1", "frame": {"type": "system", "subtype": "task_progress", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Research phase running", "usage": {"input_tokens": 300, "output_tokens": 200, "total_tokens": 500, "tool_uses": 4}, "workflow_progress": [{"type": "workflow_phase", "index": 0, "title": "Research"}, {"type": "workflow_phase", "index": 1, "title": "Implement"}, {"type": "workflow_agent", "index": 0, "label": "Researcher", "state": "active", "phaseIndex": 0, "attempt": 0, "model": "claude-sonnet-4-6", "lastToolName": "Read", "tokens": 200, "toolCalls": 2}], "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73202", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
7+{"type": "emit_inbound", "label": "task_progress:phase-1", "frame": {"type": "system", "subtype": "task_progress", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Research phase running", "usage": {"input_tokens": 300, "output_tokens": 200, "total_tokens": 500, "tool_uses": 4}, "workflow_progress": [{"type": "workflow_phase", "index": 0, "title": "Research"}, {"type": "workflow_phase", "index": 1, "title": "Implement"}, {"type": "workflow_agent", "index": 0, "label": "Researcher", "state": "waiting", "phaseIndex": 0, "attempt": 0, "model": "claude-sonnet-4-6", "lastToolName": "Read", "tokens": 200, "toolCalls": 2}], "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73202", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
88 {"type": "emit_inbound", "label": "task_progress:phase-2", "frame": {"type": "system", "subtype": "task_progress", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Implement phase running", "usage": {"input_tokens": 700, "output_tokens": 500, "total_tokens": 1200, "tool_uses": 9}, "workflow_progress": [{"type": "workflow_phase", "index": 0, "title": "Research"}, {"type": "workflow_phase", "index": 1, "title": "Implement"}, {"type": "workflow_agent", "index": 0, "label": "Researcher", "state": "done", "phaseIndex": 0, "attempt": 0, "model": "claude-sonnet-4-6", "lastToolName": "Read", "tokens": 400, "toolCalls": 3}, {"type": "workflow_agent", "index": 1, "label": "Implementer", "state": "active", "phase_index": 1, "attempt": 0, "model": "claude-opus-4-1", "last_tool_name": "Edit", "tokens": 500, "tool_calls": 4}], "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73203", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
99 {"type": "emit_inbound", "label": "task_notification:local_workflow", "frame": {"type": "system", "subtype": "task_notification", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "status": "completed", "output_file": "", "summary": "Workflow finished both phases.", "usage": {"input_tokens": 900, "output_tokens": 600, "total_tokens": 1500, "tool_uses": 11}, "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73204", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}
1010 {"type": "emit_inbound", "label": "assistant:commentary", "frame": {"type": "assistant", "message": {"model": "claude-sonnet-4-6", "id": "msg_01claudeworkflow2", "type": "message", "role": "assistant", "content": [{"type": "text", "text": "claude workflow fixture complete"}], "stop_reason": null, "stop_sequence": null, "stop_details": null, "usage": {"input_tokens": 1, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0, "output_tokens": 1, "service_tier": "standard", "inference_geo": "not_available"}, "context_management": null}, "parent_tool_use_id": null, "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201", "uuid": "89be9e65-097b-47e8-849b-be723386b201"}}
apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts+14−0
@@ -72,6 +72,20 @@ export function assertClaudeWorkflowOutput( · 71 unmodified lines
7272 0,
7373 "a member without membership would render outside its workflow",
7474 );
75+ const waitingProgressItem = result.domainEvents.find(
76+ (event) =>
77+ event.type === "turn-item.updated" &&
78+ event.payload.type === "reasoning" &&
79+ event.payload.text === "Read" &&
80+ event.payload.status === "waiting",
81+ );
82+ assert.isDefined(waitingProgressItem);
83+ if (waitingProgressItem?.type === "turn-item.updated") {
84+ assert.isNull(waitingProgressItem.payload.completedAt);
85+ assert.isTrue(
86+ waitingProgressItem.payload.type === "reasoning" && waitingProgressItem.payload.streaming,
87+ );
88+ }
7589 // The coordinator's usage already covers its members, so the panel subtracts
7690 // theirs from it. That only stays non-negative if the provider's totals are
7791 // carried through unmodified.