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

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

@@ -38,6 +38,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;
 
@@ -208,10 +209,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,
           ),
@@ -223,8 +221,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" &&
@@ -310,7 +346,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) }),
@@ -461,7 +497,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.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) }),

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

@@ -1,5 +1,5 @@
 import * as NodeServices from "@effect/platform-node/NodeServices";
-import { describe, it } from "@effect/vitest";
+import { assert, describe, it } from "@effect/vitest";
 import {
   MessageId,
   NodeId,
@@ -13,7 +13,11 @@ import {
 } from "@t3tools/contracts";
 import { ACP_PROTOCOL } from "@t3tools/provider-acp/server/adapter";
 import { FxSettings } from "@t3tools/provider-fx/settings";
-import { FX_DRIVER_KIND, makeFxAdapterV2 } from "@t3tools/provider-fx/testing";
+import {
+  FX_DRIVER_KIND,
+  FX_HOOKS_SESSION_META,
+  makeFxAdapterV2,
+} from "@t3tools/provider-fx/testing";
 import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
 import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
 import { ProviderAdapterV2RuntimePolicy } from "@t3tools/provider-core/server/ProviderAdapter";
@@ -115,6 +119,64 @@ const fxSession = (input: {
   modes: { currentModeId: "ask", availableModes: input.modes.map((id) => ({ id, name: id })) },
 });
 
+const update = (sessionId: string, sessionUpdate: Record<string, unknown>): Frame => ({
+  type: "emit_inbound",
+  label: `session/update:${String(sessionUpdate.sessionUpdate)}`,
+  frame: {
+    kind: "notification",
+    method: "session/update",
+    params: { sessionId, update: sessionUpdate },
+  },
+});
+/** A `libfx/hook` event as fx 0.0.15 sends it once a session opts in. */
+const hook = (sessionId: string, type: string, data: Record<string, unknown>): Frame => ({
+  type: "emit_inbound",
+  label: `libfx/hook:${type}`,
+  frame: {
+    kind: "notification",
+    method: "libfx/hook",
+    params: {
+      version: 1,
+      sessionId,
+      rootSessionId: SESSION_ID,
+      parentSessionId: sessionId === SESSION_ID ? null : SESSION_ID,
+      sourceSessionId: null,
+      workId: null,
+      turnId: "turn-1",
+      executionId: `execution-${sessionId}`,
+      stepIndex: 1,
+      timestamp: 1_793_283_383_907,
+      type,
+      data,
+      contentCaptured: true,
+      contentTruncated: false,
+    },
+  },
+});
+const subagentReply = (reply: string) =>
+  JSON.stringify({ ok: true, result: reply, error_code: null });
+const subagentCall = (callId: string, task: string) =>
+  update(SESSION_ID, {
+    sessionUpdate: "tool_call",
+    toolCallId: callId,
+    name: "subagent",
+    title: `Subagent · ${task}`,
+    kind: "other",
+    status: "pending",
+    rawInput: { action: "run", task },
+    _meta: { fx: { toolCall: { internal: false } } },
+  });
+/** fx's 200-byte ACP preview of a finished `subagent` call. */
+const subagentPreview = (callId: string, reply: string) =>
+  update(SESSION_ID, {
+    sessionUpdate: "tool_call_update",
+    toolCallId: callId,
+    status: "completed",
+    content: [
+      { type: "content", content: { type: "text", text: subagentReply(reply).slice(0, 199) } },
+    ],
+  });
+
 const initialize: ReadonlyArray<Frame> = [
   outbound("initialize"),
   answer("initialize", {
@@ -184,6 +246,7 @@ const openFxSession = Effect.fn("openFxSession")(function* (input: {
       ),
       childProcessSpawner,
       fileSystem,
+      sessionMeta: FX_HOOKS_SESSION_META,
     }),
   });
   const threadId = ThreadId.make(`thread-fx-${input.scenario}`);
@@ -199,15 +262,15 @@ const openFxSession = Effect.fn("openFxSession")(function* (input: {
     interactionMode: "default",
     cwd: replayDir,
   });
-  yield* Effect.gen(function* () {
+  const events = yield* Effect.gen(function* () {
     const session = yield* adapter.openSession({
       threadId,
       providerSessionId: ProviderSessionId.make(`provider-session-fx-${input.scenario}`),
       modelSelection,
       runtimePolicy,
       ...(input.loadSessionId === undefined ? {} : { initialNativeThreadId: input.loadSessionId }),
     });
-    if (input.turnModel === undefined) return;
+    if (input.turnModel === undefined) return [];
     const providerThread = yield* session.ensureThread({ threadId, modelSelection, runtimePolicy });
     const now = yield* DateTime.now;
     const turnSelection = selectionOn(input.turnModel);
@@ -252,12 +315,13 @@ const openFxSession = Effect.fn("openFxSession")(function* (input: {
       modelSelection: turnSelection,
       runtimePolicy,
     });
-    yield* session.events.pipe(
+    return yield* session.events.pipe(
       Stream.takeUntil((event) => event.type === "turn.terminal"),
-      Stream.runDrain,
+      Stream.runCollect,
     );
   }).pipe(Effect.scoped);
   yield* makeAcpReplayCompletenessAssertion(fileSystem, statusPath, transcript);
+  return events;
 });
 
 describe("FxAdapterV2", () => {
@@ -273,7 +337,11 @@ describe("FxAdapterV2", () => {
           runtimeMode: "auto",
           frames: [
             ...initialize,
-            outbound("session/new", { cwd: "<workspace>", mcpServers: "<any>" }),
+            outbound("session/new", {
+              cwd: "<workspace>",
+              mcpServers: "<any>",
+              _meta: FX_HOOKS_SESSION_META,
+            }),
             answer("session/new", fxSession({ model: FX_DEFAULT_MODEL, modes })),
             ...setConfig(
               "mode",
@@ -365,4 +433,118 @@ describe("FxAdapterV2", () => {
       ],
     }).pipe(Effect.provide(testLayer), Effect.scoped),
   );
+  // A restart reloads the session: fx replays the finished call from history
+  // with its 200-byte preview and none of the child's steps, which T3 already
+  // holds. The reload confirms hooks, so the next sub-agent shows its steps.
+  it.effect("leaves a reloaded sub-agent's thread alone and follows the next one", () =>
+    Effect.gen(function* () {
+      const earlierReply = `package.json names fx-hooks-fixture. ${"It has five scripts. ".repeat(12)}`;
+      const reply = `tsconfig.json targets ES2023. ${"Strict mode is on. ".repeat(12)}`;
+      const childSessionId = "fx-session-04";
+      const events = yield* openFxSession({
+        scenario: "reload-subagent",
+        runtimeMode: "approval-required",
+        loadSessionId: SESSION_ID,
+        turnModel: "default",
+        frames: [
+          ...initialize,
+          outbound("session/load", {
+            sessionId: SESSION_ID,
+            cwd: "<workspace>",
+            mcpServers: "<any>",
+            _meta: FX_HOOKS_SESSION_META,
+          }),
+          update(SESSION_ID, {
+            sessionUpdate: "user_message_chunk",
+            content: { type: "text", text: "Spawn a subagent to read package.json" },
+          }),
+          subagentCall("call_earlier", "Read package.json"),
+          subagentPreview("call_earlier", earlierReply),
+          update(SESSION_ID, {
+            sessionUpdate: "agent_message_chunk",
+            content: { type: "text", text: "The subagent read package.json." },
+          }),
+          answer("session/load", {
+            ...fxSession({ model: FX_DEFAULT_MODEL, modes: FX_MAIN_MODES }),
+            _meta: { fx: { hooksVersion: 1 } },
+          }),
+          ...setConfig("mode", "ask", fxConfig({ model: FX_DEFAULT_MODEL, modes: FX_MAIN_MODES })),
+          outbound("session/prompt"),
+          subagentCall("call_next", "Read tsconfig.json"),
+          hook(SESSION_ID, "delegation", {
+            callId: "call_next",
+            childSessionId,
+            workId: "work-1",
+            operationId: null,
+            phase: "dispatched",
+          }),
+          hook(childSessionId, "tool.start", {
+            call: {
+              callId: "call_read",
+              name: "read_file",
+              provenance: "fx_local",
+              argumentsJson: '{"path":"tsconfig.json"}',
+            },
+            rerun: false,
+          }),
+          hook(childSessionId, "tool.end", {
+            call: {
+              callId: "call_read",
+              name: "read_file",
+              provenance: "fx_local",
+              argumentsJson: '{"path":"tsconfig.json"}',
+            },
+            output: '{"compilerOptions":{"target":"ES2023"}}',
+            status: "success",
+            error: null,
+          }),
+          hook(SESSION_ID, "delegation", {
+            callId: "call_next",
+            childSessionId,
+            workId: "work-1",
+            operationId: null,
+            phase: "completed",
+          }),
+          subagentPreview("call_next", reply),
+          hook(SESSION_ID, "tool.end", {
+            call: { callId: "call_next", name: "subagent", provenance: "model" },
+            output: subagentReply(reply),
+            status: "success",
+            error: null,
+          }),
+          answer("session/prompt", { stopReason: "end_turn" }),
+        ],
+      });
+
+      const subagents = events.flatMap((event) =>
+        event.type === "subagent.updated" ? [event.subagent] : [],
+      );
+      assert.deepEqual(
+        [...new Set(subagents.map((subagent) => subagent.nativeTaskRef?.nativeId))],
+        ["call_next"],
+        "the reload projects nothing for the earlier sub-agent",
+      );
+      assert.lengthOf(
+        events.filter((event) => event.type === "app_thread.created"),
+        1,
+      );
+      const settled = subagents.at(-1);
+      assert.equal(settled?.status, "completed");
+      assert.equal(settled?.result, reply);
+      const childItems = events.flatMap((event) =>
+        event.type === "turn_item.updated" && event.turnItem.threadId === settled?.childThreadId
+          ? [event.turnItem]
+          : [],
+      );
+      assert.deepInclude(childItems.at(-1), { type: "assistant_message", text: reply });
+      assert.isTrue(
+        childItems.some(
+          (item) =>
+            item.type === "dynamic_tool" &&
+            item.status === "completed" &&
+            item.title === "Read tsconfig.json",
+        ),
+      );
+    }).pipe(Effect.provide(testLayer), Effect.scoped),
+  );
 });

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

@@ -3,6 +3,7 @@ import { FxSettings } from "@t3tools/provider-fx/settings";
 import {
   FX_ACP_CANCEL_BEHAVIOR,
   FX_DRIVER_KIND,
+  FX_HOOKS_SESSION_META,
   makeFxAdapterV2,
 } from "@t3tools/provider-fx/testing";
 import { ProviderInstanceId } from "@t3tools/contracts";
@@ -64,6 +65,7 @@ function layerFxProviderAdapterRegistryReplay(
             childProcessSpawner,
             fileSystem,
             cancelBehavior: FX_ACP_CANCEL_BEHAVIOR,
+            sessionMeta: FX_HOOKS_SESSION_META,
             ...(replayGate === undefined ? {} : { replayGate }),
           })(runtimeInput),
         assertComplete: makeAcpReplayCompletenessAssertion(fileSystem, statusPath, transcript),