fix(server): a running Claude workflow member shows its answer so far

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" },

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<string, PendingClaudeSubagentLaunch>,
@@ -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<string> = [];
-            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<string> => []))
-                : [];
+            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<string> => []))
+                  : [];
               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 &&