feat(fx): subagent threads show child steps and full replies

apps/server/scripts/record-fx-acp-replay-fixture.ts

@@ -42,6 +42,7 @@ const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown
 const HOME_PLACEHOLDER = "/home/fx-replay";
 const FX_INSTANCE_ID = ProviderInstanceId.make("fx");
 const T3_INSTRUCTIONS_BODY = /<t3_code_instructions>\n[\s\S]*?\n<\/t3_code_instructions>/u;
+const T3_INSTRUCTIONS_PLACEHOLDER = "<t3_code_instructions>\n<any>\n</t3_code_instructions>";
 const EMAIL = /[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}/giu;
 const TOKEN = /\b(?:sk-|gho_|ghp_|xai-|vercel_)[A-Za-z0-9_-]{8,}\b/gu;
 
@@ -212,10 +213,7 @@ function normalizeOutboundFrame(frame: Record<string, unknown>, runtimeInstructi
             isRecord(part) && typeof part.text === "string"
               ? {
                   ...part,
-                  text: part.text.replace(
-                    T3_INSTRUCTIONS_BODY,
-                    "<t3_code_instructions>\n<any>\n</t3_code_instructions>",
-                  ),
+                  text: part.text.replace(T3_INSTRUCTIONS_BODY, T3_INSTRUCTIONS_PLACEHOLDER),
                 }
               : part,
           ),
@@ -227,8 +225,46 @@ function normalizeOutboundFrame(frame: Record<string, unknown>, runtimeInstructi
   }
 }
 
-function normalizeInboundFrame(frame: Record<string, unknown>): Record<string, unknown> {
+/** A model call's request content: up to 64 KiB per hook that T3 never reads. */
+const HOOK_MODEL_REQUEST_FIELDS = new Set(["messages", "instructions", "tools", "toolsJson"]);
+
+function normalizeInboundFrame(
+  frame: Record<string, unknown>,
+  runtimeInstructions: string,
+): Record<string, unknown> {
   const params = isRecord(frame.params) ? frame.params : undefined;
+  if (frame.method === "libfx/hook" && params?.type === "model.start" && isRecord(params.data)) {
+    const data = params.data;
+    return {
+      ...frame,
+      params: {
+        ...params,
+        data: Object.fromEntries(
+          Object.entries(data).filter(([key]) => !HOOK_MODEL_REQUEST_FIELDS.has(key)),
+        ),
+      },
+    };
+  }
+  // fx echoes the whole prompt as the turn's input, T3's own text included.
+  if (
+    frame.method === "libfx/hook" &&
+    params?.type === "turn.start" &&
+    isRecord(params.data) &&
+    typeof params.data.input === "string"
+  ) {
+    return {
+      ...frame,
+      params: {
+        ...params,
+        data: {
+          ...params.data,
+          input: params.data.input
+            .replace(T3_INSTRUCTIONS_BODY, T3_INSTRUCTIONS_PLACEHOLDER)
+            .replace(runtimeInstructions, "<any>"),
+        },
+      },
+    };
+  }
   const update = isRecord(params?.update) ? params.update : undefined;
   if (
     update?.sessionUpdate === "available_commands_update" &&
@@ -314,7 +350,7 @@ function normalizeEntries(input: {
     const frame =
       entry.type === "expect_outbound"
         ? normalizeOutboundFrame(entry.frame, input.runtimeInstructions)
-        : normalizeInboundFrame(entry.frame);
+        : normalizeInboundFrame(entry.frame, input.runtimeInstructions);
     return {
       ...entry,
       ...(entry.label === undefined ? {} : { label: replaceAll(entry.label) }),
@@ -480,7 +516,7 @@ const recordScenario = Effect.fn("recordFxScenario")(function* (fixtureName: str
       generatedBy: "live-fx-recorder",
       fxVersion: initializeAgent.version ?? "unknown",
       normalization:
-        "Session ids are fx-session-NN, the workspace is <workspace>, HOME is /home/fx-replay and the recording user is fx-replay. T3-owned prompt text, MCP servers and initialize params other than clientCapabilities and _meta are <any>. User skill names, emails and token-shaped strings are removed. Timestamps are kept as recorded.",
+        "Session ids are fx-session-NN, the workspace is <workspace>, HOME is /home/fx-replay and the recording user is fx-replay. T3-owned prompt text, MCP servers and initialize params other than clientCapabilities and _meta are <any>. User skill names, emails and token-shaped strings are removed. libfx/hook model.start events keep their ids and purpose but not their request content. Timestamps are kept as recorded.",
       droppedFrames,
     },
     entries: [

apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts

@@ -6641,7 +6641,7 @@ describe("AcpAdapterV2", () => {
   );
 
   it.effect(
-    "projects a child terminal in the completed-root finalization window exactly once",
+    "projects changed full results in the completed-root finalization window once per result",
     () =>
       Effect.gen(function* () {
         const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
@@ -6672,6 +6672,7 @@ describe("AcpAdapterV2", () => {
           const secondChildSessionId = "mock-child-session-finalize-window-second";
           let firstSubagentStatus: "running" | "completed" = "running";
           let secondSubagentStatus: "waiting" | "completed" = "waiting";
+          let secondSubagentResult: string | null = "SECOND_DONE";
           type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"];
           let sessionUpdateHandler:
             | Parameters<RuntimeService["handleSessionUpdate"]>[0]
@@ -6709,7 +6710,7 @@ describe("AcpAdapterV2", () => {
                     model: null,
                     status: secondSubagentStatus,
                     childSessionId: secondChildSessionId,
-                    result: secondSubagentStatus === "completed" ? "SECOND_DONE" : null,
+                    result: secondSubagentStatus === "completed" ? secondSubagentResult : null,
                     suppressNormalTool: true,
                   } as AcpAdapterV2SubagentUpdate;
                 }
@@ -6782,10 +6783,6 @@ describe("AcpAdapterV2", () => {
               rawOutput: { content: "FIRST_DONE" },
             },
           }).pipe(Effect.provideService(Clock.Clock, blockingClock));
-          for (let attempt = 0; attempt < 10; attempt += 1) {
-            yield* Effect.yieldNow;
-          }
-
           yield* sessionUpdateHandler!({
             sessionId: "mock-session-1",
             update: {
@@ -6799,8 +6796,6 @@ describe("AcpAdapterV2", () => {
           });
           assert.isTrue(yield* hasPendingBackgroundWork);
 
-          while (Option.isSome(yield* Queue.poll(events))) {}
-
           blockNextClockRead = true;
           const adjustFiber = yield* TestClock.adjust("3 seconds").pipe(Effect.forkScoped);
           yield* Deferred.await(finalizationClockRead);
@@ -6816,35 +6811,71 @@ describe("AcpAdapterV2", () => {
               rawOutput: { content: "SECOND_DONE" },
             },
           });
+          for (const result of ["SECOND_FULL_RESULT", "SECOND_FULL_RESULT", null]) {
+            secondSubagentResult = result;
+            yield* sessionUpdateHandler!({
+              sessionId: secondChildSessionId,
+              update: {
+                sessionUpdate: "tool_call_update",
+                toolCallId: "tool-call-generic-2",
+                title: "second background subagent",
+                kind: "other",
+                status: "completed",
+                rawOutput: { content: result },
+              },
+            });
+          }
           yield* Deferred.succeed(releaseFinalizationClockRead, undefined);
           yield* Fiber.join(adjustFiber);
-          yield* Effect.yieldNow;
-          yield* Effect.yieldNow;
 
           const providerTurnId = idAllocator.derive.providerTurn({
             driver: ACP_TEST_DRIVER,
             nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:1"),
           });
-          let completedSubagentUpdates = 0;
-          let rootTerminalStatus: string | null = null;
-          let polled = yield* Queue.poll(events);
-          while (Option.isSome(polled)) {
-            const event = polled.value;
-            if (
-              event.type === "turn_item.updated" &&
-              event.turnItem.type === "subagent" &&
-              event.turnItem.nativeItemRef?.nativeId === "task-finalize-window-second" &&
-              event.turnItem.status === "completed"
-            ) {
-              completedSubagentUpdates += 1;
-            }
-            if (event.type === "turn.terminal" && event.providerTurnId === providerTurnId) {
-              rootTerminalStatus = event.status;
-            }
-            polled = yield* Queue.poll(events);
-          }
-          assert.equal(rootTerminalStatus, "completed");
-          assert.equal(completedSubagentUpdates, 1);
+          const observed = yield* Stream.fromQueue(events).pipe(
+            Stream.takeUntil(
+              (event) => event.type === "turn.terminal" && event.providerTurnId === providerTurnId,
+            ),
+            Stream.runCollect,
+          );
+          const terminal = observed.at(-1);
+          assert.equal(
+            terminal?.type === "turn.terminal" ? terminal.status : undefined,
+            "completed",
+          );
+          const results = observed.flatMap((event) =>
+            event.type === "turn_item.updated" &&
+            event.turnItem.type === "subagent" &&
+            event.turnItem.nativeItemRef?.nativeId === "task-finalize-window-second" &&
+            event.turnItem.status === "completed"
+              ? [event.turnItem.result]
+              : [],
+          );
+          assert.deepEqual(results, ["SECOND_DONE", "SECOND_FULL_RESULT"]);
+          const childUpdates = observed.flatMap((event) =>
+            event.type === "subagent.updated" &&
+            event.subagent.nativeTaskRef?.nativeId === "task-finalize-window-second" &&
+            event.subagent.status === "completed"
+              ? [event.subagent]
+              : [],
+          );
+          assert.deepEqual(
+            childUpdates.map(({ result }) => result),
+            results,
+          );
+          assert.equal(childUpdates[1]?.completedAt, childUpdates[0]?.completedAt);
+          const replies = observed.flatMap((event) =>
+            event.type === "message.updated" &&
+            event.message.threadId === childUpdates[0]?.childThreadId &&
+            event.message.role === "assistant"
+              ? [event.message]
+              : [],
+          );
+          assert.deepEqual(
+            replies.map(({ text }) => text),
+            results,
+          );
+          assert.equal(replies[1]?.id, replies[0]?.id);
           assert.isFalse(
             yield* hasPendingBackgroundWork,
             "finalize-window terminal must clear the carryover pin",

apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts

@@ -196,6 +196,7 @@ export function makeAcpReplayRuntime(input: {
   readonly cancelMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelMeta"];
   readonly cancelBehavior?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelBehavior"];
   readonly initializeMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["initializeMeta"];
+  readonly sessionMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["sessionMeta"];
   readonly replayGate?: ProviderReplayGate;
 }): (
   runtimeInput: AcpAdapterV2RuntimeInput,
@@ -238,6 +239,7 @@ export function makeAcpReplayRuntime(input: {
           ...(input.cancelMeta === undefined ? {} : { cancelMeta: input.cancelMeta }),
           ...(input.cancelBehavior === undefined ? {} : { cancelBehavior: input.cancelBehavior }),
           ...(input.initializeMeta === undefined ? {} : { initializeMeta: input.initializeMeta }),
+          ...(input.sessionMeta === undefined ? {} : { sessionMeta: input.sessionMeta }),
           ...(input.replayGate === undefined
             ? {}
             : { transformStdout: holdGatedReplayLines(input.transcript, input.replayGate) }),