From 1010ec6c22979f19c67c6ca0c128bd4b2a6bb345 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:00:53 -0700 Subject: [PATCH] fix(server): a running Claude workflow member shows its answer so far Workflow members never stream, so a running member's thread showed only its prompt and a "Thinking" row until it finished. Progress frames now re-read a running member's transcript at most every 10 seconds and project the answer so far into the same assistant message the settled answer later replaces. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 82 ++++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 62 +++++++++----- 2 files changed, 122 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 00d6f36be378..2c0d13aa0b32 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12148,7 +12148,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); assert.deepEqual( threadMessages(harness.events, member?.childThreadId).map((event) => event.message.text), - ["Reply with exactly: A1"], + ["Reply with exactly: A1", "A1 from the transcript"], ); assert.isFalse( harness.events.some( @@ -13735,6 +13735,86 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("shows a running workflow member's answer so far from its transcript", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWorkflowHarness; + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-workflow-live-answer"), + text: "Run the alpha workflow.", + attachments: [], + }), + ); + yield* launchWorkflow({ harness, snapshotUuid: "00000000-0000-4000-8000-000000001360" }); + const firstThreadId = workflowMemberEvents(harness.events, 1)[0]?.subagent.childThreadId; + const assistantEvents = () => + threadMessages(harness.events, firstThreadId).filter( + (event) => event.message.role === "assistant", + ); + const transcriptPath = NodePath.join(harness.transcriptDir, "agent-a1.jsonl"); + const running = (uuid: string) => workflowSnapshot({ uuid, state: "start" }); + + yield* awaitUntil( + () => assistantEvents().at(-1)?.message.text === "A1 from the transcript", + "running workflow member answer so far", + ); + assert.equal(workflowMemberEvents(harness.events, 1).at(-1)?.subagent.status, "running"); + + NodeFS.appendFileSync( + transcriptPath, + '{"type":"assistant","message":{"role":"assistant","id":"msg_a1_more","content":[{"type":"text","text":"A1 keeps going"}]}}\n', + ); + yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001361")); + assert.lengthOf(assistantEvents(), 1, "a frame inside the throttle window reads nothing"); + + yield* TestClock.adjust("10 seconds"); + yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001362")); + assert.lengthOf(assistantEvents(), 2); + assert.equal( + assistantEvents().at(-1)?.message.text, + "A1 from the transcript\n\nA1 keeps going", + ); + + yield* TestClock.adjust("10 seconds"); + yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001363")); + assert.lengthOf(assistantEvents(), 2, "an unchanged transcript emits nothing"); + + NodeFS.appendFileSync( + transcriptPath, + '{"type":"assistant","message":{"role":"assistant","id":"msg_a1_final","content":[{"type":"text","text":"A1 final"}]}}\n', + ); + yield* harness.offerAndWait( + workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001364", state: "done" }), + ); + assert.equal(workflowMemberEvents(harness.events, 1).at(-1)?.subagent.status, "completed"); + assert.equal( + assistantEvents().at(-1)?.message.text, + "A1 from the transcript\n\nA1 keeps going\n\nA1 final", + ); + assert.lengthOf( + new Set(assistantEvents().map((event) => event.message.id)), + 1, + "the settled answer replaces the running one in place", + ); + + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000001365", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect.each([ { backgrounding: "succeeds" }, { backgrounding: "fails" }, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 75a3bbd8912c..0a289a522663 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3092,6 +3092,9 @@ function claudeSubagentModelSelection( const PENDING_CLAUDE_SUBAGENT_CAP = 64; const PENDING_CLAUDE_SUBAGENT_FRAME_CAP = 256; +// Workflow progress frames arrive about every 12 seconds; a running member's +// transcript is re-read at most this often to show its answer so far. +const WORKFLOW_MEMBER_LIVE_TRANSCRIPT_READ_INTERVAL_MS = 10_000; function rememberPendingClaudeSubagentLaunch( pending: Map, @@ -3422,6 +3425,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( readonly attempt: number; readonly transcriptAnswerDigest: string | undefined; readonly transcriptReadAttempts: number; + readonly liveTranscriptReadAt: DateTime.Utc | null; readonly finalTranscriptRead: boolean; readonly result: string | undefined; readonly label: string; @@ -4525,7 +4529,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (input.coordinator.task.status === "running" ? (restarted ? 0 : (previous?.transcriptReadAttempts ?? 0)) < 3 : restarted || previous?.finalTranscriptRead !== true); - if (!changed && !readTranscript) continue; + // A running member streams nothing, so its transcript is the only + // way to show its answer so far. A restart shows "Retry in + // progress." and starts a fresh throttle window before the next read. + const liveTranscriptReadAt = previous?.liveTranscriptReadAt ?? null; + const readLiveTranscript = + status === "running" && + !restarted && + transcriptDir !== undefined && + agentId !== undefined && + (liveTranscriptReadAt === null || + DateTime.toEpochMillis(now) - DateTime.toEpochMillis(liveTranscriptReadAt) >= + WORKFLOW_MEMBER_LIVE_TRANSCRIPT_READ_INTERVAL_MS); + if (!changed && !readTranscript && !readLiveTranscript) continue; const nodeId = idAllocator.derive.nodeFromProviderItem({ driver: CLAUDE_PROVIDER, @@ -4651,21 +4667,22 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( let transcriptAnswerDigest = restarted ? undefined : previous?.transcriptAnswerDigest; let transcript: ReadonlyArray = []; - if (settled) { - transcript = readTranscript - ? yield* readWorkflowAgentAnswers({ - transcriptDir, - agentId, - ...(adapterOptions.environment.CLAUDE_CONFIG_DIR === undefined - ? {} - : { - configDir: path.resolve( - session.cwd, - adapterOptions.environment.CLAUDE_CONFIG_DIR, - ), - }), - }).pipe(Effect.orElseSucceed((): ReadonlyArray => [])) - : []; + if (settled || readLiveTranscript) { + transcript = + readTranscript || readLiveTranscript + ? yield* readWorkflowAgentAnswers({ + transcriptDir, + agentId, + ...(adapterOptions.environment.CLAUDE_CONFIG_DIR === undefined + ? {} + : { + configDir: path.resolve( + session.cwd, + adapterOptions.environment.CLAUDE_CONFIG_DIR, + ), + }), + }).pipe(Effect.orElseSucceed((): ReadonlyArray => [])) + : []; const transcriptAnswer = transcript.length > 0 ? transcript.join("\n\n") : undefined; const answerDigest = transcriptAnswer === undefined @@ -4677,11 +4694,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (answerDigest !== undefined) transcriptAnswerDigest = answerDigest; const answer = transcriptAnswer ?? - (member.result !== undefined && changed && previousDigest === undefined - ? `Partial answer from workflow progress:\n\n${member.result}` - : attempt > 1 && changed && previousDigest === undefined - ? "No answer from this workflow attempt." - : undefined); + (!settled + ? undefined + : member.result !== undefined && changed && previousDigest === undefined + ? `Partial answer from workflow progress:\n\n${member.result}` + : attempt > 1 && changed && previousDigest === undefined + ? "No answer from this workflow attempt." + : undefined); if ( answer !== undefined && (transcriptAnswer === undefined || restarted || answerDigest !== previousDigest) @@ -4710,6 +4729,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( transcriptReadAttempts: (restarted ? 0 : (previous?.transcriptReadAttempts ?? 0)) + (readTranscript ? 1 : 0), + liveTranscriptReadAt: readLiveTranscript || restarted ? now : liveTranscriptReadAt, finalTranscriptRead: (!restarted && previous?.finalTranscriptRead === true) || (readTranscript &&