Skip to content

Commit edc703b

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
authored
fix(server): a running Claude workflow member shows its answer so far (pingdotgg#20786)
Co-authored-by: Claude Opus 5.5 <[email protected]>
1 parent c0fda31 commit edc703b

2 files changed

Lines changed: 122 additions & 22 deletions

File tree

‎apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts‎

Lines changed: 81 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12148,7 +12148,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
1214812148
);
1214912149
assert.deepEqual(
1215012150
threadMessages(harness.events, member?.childThreadId).map((event) => event.message.text),
12151-
["Reply with exactly: A1"],
12151+
["Reply with exactly: A1", "A1 from the transcript"],
1215212152
);
1215312153
assert.isFalse(
1215412154
harness.events.some(
@@ -13735,6 +13735,86 @@ describe("ClaudeAdapterV2 background wake turns", () => {
1373513735
),
1373613736
);
1373713737

13738+
it.effect("shows a running workflow member's answer so far from its transcript", () =>
13739+
Effect.scoped(
13740+
Effect.gen(function* () {
13741+
const harness = yield* makeWorkflowHarness;
13742+
const now = yield* DateTime.now;
13743+
yield* harness.runtime.startTurn(
13744+
makeClaudeTestTurnInput({
13745+
threadId: harness.threadId,
13746+
providerThread: harness.providerThread,
13747+
now,
13748+
attemptId: RunAttemptId.make("attempt-claude-workflow-live-answer"),
13749+
text: "Run the alpha workflow.",
13750+
attachments: [],
13751+
}),
13752+
);
13753+
yield* launchWorkflow({ harness, snapshotUuid: "00000000-0000-4000-8000-000000001360" });
13754+
const firstThreadId = workflowMemberEvents(harness.events, 1)[0]?.subagent.childThreadId;
13755+
const assistantEvents = () =>
13756+
threadMessages(harness.events, firstThreadId).filter(
13757+
(event) => event.message.role === "assistant",
13758+
);
13759+
const transcriptPath = NodePath.join(harness.transcriptDir, "agent-a1.jsonl");
13760+
const running = (uuid: string) => workflowSnapshot({ uuid, state: "start" });
13761+
13762+
yield* awaitUntil(
13763+
() => assistantEvents().at(-1)?.message.text === "A1 from the transcript",
13764+
"running workflow member answer so far",
13765+
);
13766+
assert.equal(workflowMemberEvents(harness.events, 1).at(-1)?.subagent.status, "running");
13767+
13768+
NodeFS.appendFileSync(
13769+
transcriptPath,
13770+
'{"type":"assistant","message":{"role":"assistant","id":"msg_a1_more","content":[{"type":"text","text":"A1 keeps going"}]}}\n',
13771+
);
13772+
yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001361"));
13773+
assert.lengthOf(assistantEvents(), 1, "a frame inside the throttle window reads nothing");
13774+
13775+
yield* TestClock.adjust("10 seconds");
13776+
yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001362"));
13777+
assert.lengthOf(assistantEvents(), 2);
13778+
assert.equal(
13779+
assistantEvents().at(-1)?.message.text,
13780+
"A1 from the transcript\n\nA1 keeps going",
13781+
);
13782+
13783+
yield* TestClock.adjust("10 seconds");
13784+
yield* harness.offerAndWait(running("00000000-0000-4000-8000-000000001363"));
13785+
assert.lengthOf(assistantEvents(), 2, "an unchanged transcript emits nothing");
13786+
13787+
NodeFS.appendFileSync(
13788+
transcriptPath,
13789+
'{"type":"assistant","message":{"role":"assistant","id":"msg_a1_final","content":[{"type":"text","text":"A1 final"}]}}\n',
13790+
);
13791+
yield* harness.offerAndWait(
13792+
workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001364", state: "done" }),
13793+
);
13794+
assert.equal(workflowMemberEvents(harness.events, 1).at(-1)?.subagent.status, "completed");
13795+
assert.equal(
13796+
assistantEvents().at(-1)?.message.text,
13797+
"A1 from the transcript\n\nA1 keeps going\n\nA1 final",
13798+
);
13799+
assert.lengthOf(
13800+
new Set(assistantEvents().map((event) => event.message.id)),
13801+
1,
13802+
"the settled answer replaces the running one in place",
13803+
);
13804+
13805+
yield* Queue.offer(
13806+
harness.sdkMessages,
13807+
makeResultFrame({ uuid: "00000000-0000-4000-8000-000000001365", result: "Done." }),
13808+
);
13809+
yield* Queue.take(harness.terminalReceipts);
13810+
}).pipe(
13811+
Effect.provide(
13812+
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
13813+
),
13814+
),
13815+
),
13816+
);
13817+
1373813818
it.effect.each([
1373913819
{ backgrounding: "succeeds" },
1374013820
{ backgrounding: "fails" },

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

Lines changed: 41 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -3092,6 +3092,9 @@ function claudeSubagentModelSelection(
30923092

30933093
const PENDING_CLAUDE_SUBAGENT_CAP = 64;
30943094
const PENDING_CLAUDE_SUBAGENT_FRAME_CAP = 256;
3095+
// Workflow progress frames arrive about every 12 seconds; a running member's
3096+
// transcript is re-read at most this often to show its answer so far.
3097+
const WORKFLOW_MEMBER_LIVE_TRANSCRIPT_READ_INTERVAL_MS = 10_000;
30953098

30963099
function rememberPendingClaudeSubagentLaunch(
30973100
pending: Map<string, PendingClaudeSubagentLaunch>,
@@ -3422,6 +3425,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
34223425
readonly attempt: number;
34233426
readonly transcriptAnswerDigest: string | undefined;
34243427
readonly transcriptReadAttempts: number;
3428+
readonly liveTranscriptReadAt: DateTime.Utc | null;
34253429
readonly finalTranscriptRead: boolean;
34263430
readonly result: string | undefined;
34273431
readonly label: string;
@@ -4525,7 +4529,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
45254529
(input.coordinator.task.status === "running"
45264530
? (restarted ? 0 : (previous?.transcriptReadAttempts ?? 0)) < 3
45274531
: restarted || previous?.finalTranscriptRead !== true);
4528-
if (!changed && !readTranscript) continue;
4532+
// A running member streams nothing, so its transcript is the only
4533+
// way to show its answer so far. A restart shows "Retry in
4534+
// progress." and starts a fresh throttle window before the next read.
4535+
const liveTranscriptReadAt = previous?.liveTranscriptReadAt ?? null;
4536+
const readLiveTranscript =
4537+
status === "running" &&
4538+
!restarted &&
4539+
transcriptDir !== undefined &&
4540+
agentId !== undefined &&
4541+
(liveTranscriptReadAt === null ||
4542+
DateTime.toEpochMillis(now) - DateTime.toEpochMillis(liveTranscriptReadAt) >=
4543+
WORKFLOW_MEMBER_LIVE_TRANSCRIPT_READ_INTERVAL_MS);
4544+
if (!changed && !readTranscript && !readLiveTranscript) continue;
45294545

45304546
const nodeId = idAllocator.derive.nodeFromProviderItem({
45314547
driver: CLAUDE_PROVIDER,
@@ -4651,21 +4667,22 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
46514667

46524668
let transcriptAnswerDigest = restarted ? undefined : previous?.transcriptAnswerDigest;
46534669
let transcript: ReadonlyArray<string> = [];
4654-
if (settled) {
4655-
transcript = readTranscript
4656-
? yield* readWorkflowAgentAnswers({
4657-
transcriptDir,
4658-
agentId,
4659-
...(adapterOptions.environment.CLAUDE_CONFIG_DIR === undefined
4660-
? {}
4661-
: {
4662-
configDir: path.resolve(
4663-
session.cwd,
4664-
adapterOptions.environment.CLAUDE_CONFIG_DIR,
4665-
),
4666-
}),
4667-
}).pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []))
4668-
: [];
4670+
if (settled || readLiveTranscript) {
4671+
transcript =
4672+
readTranscript || readLiveTranscript
4673+
? yield* readWorkflowAgentAnswers({
4674+
transcriptDir,
4675+
agentId,
4676+
...(adapterOptions.environment.CLAUDE_CONFIG_DIR === undefined
4677+
? {}
4678+
: {
4679+
configDir: path.resolve(
4680+
session.cwd,
4681+
adapterOptions.environment.CLAUDE_CONFIG_DIR,
4682+
),
4683+
}),
4684+
}).pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []))
4685+
: [];
46694686
const transcriptAnswer = transcript.length > 0 ? transcript.join("\n\n") : undefined;
46704687
const answerDigest =
46714688
transcriptAnswer === undefined
@@ -4677,11 +4694,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
46774694
if (answerDigest !== undefined) transcriptAnswerDigest = answerDigest;
46784695
const answer =
46794696
transcriptAnswer ??
4680-
(member.result !== undefined && changed && previousDigest === undefined
4681-
? `Partial answer from workflow progress:\n\n${member.result}`
4682-
: attempt > 1 && changed && previousDigest === undefined
4683-
? "No answer from this workflow attempt."
4684-
: undefined);
4697+
(!settled
4698+
? undefined
4699+
: member.result !== undefined && changed && previousDigest === undefined
4700+
? `Partial answer from workflow progress:\n\n${member.result}`
4701+
: attempt > 1 && changed && previousDigest === undefined
4702+
? "No answer from this workflow attempt."
4703+
: undefined);
46854704
if (
46864705
answer !== undefined &&
46874706
(transcriptAnswer === undefined || restarted || answerDigest !== previousDigest)
@@ -4710,6 +4729,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
47104729
transcriptReadAttempts:
47114730
(restarted ? 0 : (previous?.transcriptReadAttempts ?? 0)) +
47124731
(readTranscript ? 1 : 0),
4732+
liveTranscriptReadAt: readLiveTranscript || restarted ? now : liveTranscriptReadAt,
47134733
finalTranscriptRead:
47144734
(!restarted && previous?.finalTranscriptRead === true) ||
47154735
(readTranscript &&

0 commit comments

Comments
 (0)