test(agents): prove terminal reply boundaries

src/agents/acp-spawn.test.ts

@@ -7,6 +7,9 @@ import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vite
 import type { AcpInitializeSessionInput } from "../acp/control-plane/manager.types.js";
 import type { SessionEntry } from "../config/sessions/types.js";
 import type { OpenClawConfig } from "../config/types.openclaw.js";
+import type { CallGatewayOptions } from "../gateway/call.js";
+import { setGatewayDedupeEntry, waitForAgentJob } from "../gateway/server-methods/agent-job.js";
+import type { DedupeEntry } from "../gateway/server-shared.js";
 import {
   testing as sessionBindingServiceTesting,
   registerSessionBindingAdapter,
@@ -16,7 +19,18 @@ import {
 } from "../infra/outbound/session-binding-service.js";
 import { normalizeSessionDeliveryState } from "../utils/delivery-context.shared.js";
 import { reserveChildAdmissionSlot } from "./child-admission.js";
+import type { AgentRunTerminalReplySnapshot } from "./agent-run-terminal-reply.js";
+import { createAcpVisibleTextAccumulator } from "./command/attempt-execution.helpers.js";
+import {
+  buildAcpResult,
+  createAcpToolLifecycleTracker,
+  emitAcpLifecycleEnd,
+} from "./command/attempt-execution.js";
 import { resolveThinkingDefault } from "./model-selection.js";
+import { SUBAGENT_ENDED_REASON_COMPLETE } from "./subagent-lifecycle-events.js";
+import { createSubagentRegistryLifecycleController } from "./subagent-registry-lifecycle.js";
+import type { RegisterSubagentRunParams } from "./subagent-registry-run-manager.js";
+import type { SubagentRunRecord } from "./subagent-registry.types.js";
 
 type SessionBindingAdapterCapabilities = NonNullable<SessionBindingAdapter["capabilities"]>;
 
@@ -200,7 +214,10 @@ vi.mock("../config/sessions/paths.js", () => ({
   resolveStorePath: hoisted.resolveStorePathMock,
 }));
 
-vi.mock("../config/sessions/session-accessor.js", () => hoisted.createSessionAccessorMock());
+vi.mock("../config/sessions/session-accessor.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../config/sessions/session-accessor.js")>()),
+  ...hoisted.createSessionAccessorMock(),
+}));
 
 vi.mock("../config/sessions.js", () => ({
   loadSessionStore: hoisted.loadSessionStoreMock,
@@ -227,14 +244,16 @@ vi.mock("./acp-spawn-parent-stream.js", () => ({
   startAcpSpawnParentStreamRelay: hoisted.startAcpSpawnParentStreamRelayMock,
 }));
 
-vi.mock("./subagent-registry.js", () => ({
+vi.mock("./subagent-registry.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("./subagent-registry.js")>()),
   countActiveRunsForSession: hoisted.countActiveRunsForSessionMock,
   getSubagentRunByChildSessionKey: hoisted.getSubagentRunByChildSessionKeyMock,
   // ACP registration deliberately moved behind the shared spawn pipeline.
   registerSubagentRun: hoisted.registerSubagentRunMock,
 }));
 
-vi.mock("../tasks/runtime-internal.js", () => ({
+vi.mock("../tasks/runtime-internal.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../tasks/runtime-internal.js")>()),
   listTasksForOwnerKey: hoisted.listTasksForOwnerKeyMock,
 }));
 
@@ -405,6 +424,54 @@ function expectAcceptedSpawn(result: SpawnResult): Extract<SpawnResult, { status
   return result;
 }
 
+async function waitForAgentReplySnapshot(runId: string) {
+  const result = await waitForAgentJob({ runId, timeoutMs: 0 });
+  expect(result).toEqual(expect.objectContaining({ status: "ok" }));
+  return result as { terminalReply?: AgentRunTerminalReplySnapshot };
+}
+
+function createRegisteredRunEntry(registration: RegisterSubagentRunParams): SubagentRunRecord {
+  return {
+    ...registration,
+    createdAt: 100,
+    execution: { status: "running", startedAt: 100 },
+  };
+}
+
+function createBoundaryLifecycleController(params: {
+  entry: SubagentRunRecord;
+  captureSubagentCompletionReply: ReturnType<typeof vi.fn>;
+  runSubagentAnnounceFlow: ReturnType<typeof vi.fn>;
+}) {
+  return createSubagentRegistryLifecycleController({
+    runs: new Map([[params.entry.runId, params.entry]]),
+    resumedRuns: new Set(),
+    subagentAnnounceTimeoutMs: 1_000,
+    getRuntimeConfig: () => ({}),
+    persist: vi.fn(),
+    persistOrThrow: vi.fn(),
+    clearPendingLifecycleError: vi.fn(),
+    countPendingDescendantRuns: () => 0,
+    suppressAnnounceForSteerRestart: () => false,
+    resolveSubagentTask: () => ({ lookup: "unavailable" }),
+    shouldEmitEndedHookForRun: () => false,
+    emitSubagentEndedHookForRun: vi.fn(async () => {}),
+    emitSubagentProgressEndedForRun: vi.fn(async () => {}),
+    notifyContextEngineSubagentEnded: vi.fn(async () => {}),
+    retireSupersededRun: vi.fn(async () => {}),
+    resumeSubagentRun: vi.fn(),
+    callGateway: async <T = Record<string, unknown>>(_opts: CallGatewayOptions): Promise<T> =>
+      ({}) as T,
+    captureSubagentCompletionReply: params.captureSubagentCompletionReply,
+    runSubagentAnnounceFlow: params.runSubagentAnnounceFlow,
+    maybeWakeRequesterAfterAllChildrenSettled: vi.fn(async (wakeParams) => {
+      wakeParams.completeBatch([wakeParams.settledEntry.runId]);
+      return false;
+    }),
+    warn: vi.fn(),
+  });
+}
+
 function expectRecordFields(
   record: unknown,
   expected: Record<string, unknown>,
@@ -2872,6 +2939,118 @@ describe("spawnAcpDirect", () => {
     expect(secondHandle.notifyStarted).toHaveBeenCalledTimes(1);
   });
 
+  it.each([
+    {
+      name: "visible",
+      chunks: ["v".repeat(5_000)],
+      expected: { disposition: "visible", text: `${"v".repeat(4_095)}…` } as const,
+      resultText: `${"v".repeat(4_095)}…`,
+    },
+    {
+      name: "silent",
+      chunks: ["NO_REPLY"],
+      expected: { disposition: "silent" } as const,
+      resultText: "NO_REPLY",
+    },
+    {
+      name: "empty",
+      chunks: [] as string[],
+      expected: { disposition: "empty" } as const,
+      resultText: null,
+    },
+  ])(
+    "carries bounded $name ACP output from spawn pipeline through wait and registry",
+    async ({ name, chunks, expected, resultText }) => {
+      for (const order of ["lifecycle-first", "dedupe-first"] as const) {
+        const runId = `run-acp-boundary-${name}-${order}`;
+        hoisted.callGatewayMock.mockImplementation(async (argsUnknown: unknown) => {
+          const args = argsUnknown as { method?: string };
+          if (args.method === "agent") {
+            return { runId };
+          }
+          if (args.method === "sessions.patch" || args.method === "sessions.delete") {
+            return { ok: true };
+          }
+          return {};
+        });
+
+        const spawned = expectAcceptedSpawn(
+          await spawnAcpDirect(createSpawnRequest(), createRequesterContext()),
+        );
+        expect(spawned).toMatchObject({ runId, mode: "run" });
+        const registration = latestMockCall(
+          hoisted.registerSubagentRunMock,
+          "subagent registration",
+        )[0] as RegisterSubagentRunParams;
+        expect(registration).toMatchObject({ runId, spawnMode: "run" });
+
+        const accumulator = createAcpVisibleTextAccumulator();
+        for (const chunk of chunks) {
+          accumulator.consume(chunk);
+        }
+        const terminalReply = accumulator.finalizeReplySnapshot();
+        expect(terminalReply).toEqual(expected);
+        const result = buildAcpResult({
+          payloadText: accumulator.finalize(),
+          terminalReply,
+          startedAt: 100,
+          resultStatus: "completed",
+        });
+        const dedupe = new Map<string, DedupeEntry>();
+        const observeLifecycle = () =>
+          emitAcpLifecycleEnd({
+            runId,
+            toolTracker: createAcpToolLifecycleTracker(),
+            resultStatus: "completed",
+            terminalReply,
+          });
+        const observeDedupe = () =>
+          setGatewayDedupeEntry({
+            dedupe,
+            key: `agent:${runId}`,
+            entry: {
+              ts: 200,
+              ok: true,
+              payload: { runId, status: "ok", startedAt: 100, endedAt: 200, result },
+            },
+          });
+        for (const observe of order === "lifecycle-first"
+          ? [observeLifecycle, observeDedupe]
+          : [observeDedupe, observeLifecycle]) {
+          observe();
+        }
+
+        const waited = await waitForAgentReplySnapshot(runId);
+        expect(waited.terminalReply).toEqual(expected);
+        const entry = createRegisteredRunEntry(registration);
+        const captureSubagentCompletionReply = vi.fn(async () => undefined);
+        const runSubagentAnnounceFlow = vi.fn(async () => true);
+        await createBoundaryLifecycleController({
+          entry,
+          captureSubagentCompletionReply,
+          runSubagentAnnounceFlow,
+        }).completeSubagentRun({
+          runId,
+          endedAt: 200,
+          outcome: { status: "ok" },
+          reason: SUBAGENT_ENDED_REASON_COMPLETE,
+          triggerCleanup: true,
+          terminalReply: waited.terminalReply,
+        });
+
+        expect(captureSubagentCompletionReply).not.toHaveBeenCalled();
+        expect(entry.completion).toMatchObject({
+          terminalReply: expected,
+          resultText,
+        });
+        expect(runSubagentAnnounceFlow).toHaveBeenCalledTimes(1);
+        expect(runSubagentAnnounceFlow).toHaveBeenCalledWith(
+          expect.objectContaining({ terminalReply: expected }),
+        );
+      }
+    },
+  );
+
   it("implicitly streams mode=run ACP spawns for subagent requester sessions", async () => {
     replaceSpawnConfig({
       ...hoisted.state.cfg,

src/agents/subagent-announce.format.e2e.test.ts

@@ -914,23 +914,48 @@ describe("subagent announce formatting", () => {
     expect(agentSpy).not.toHaveBeenCalled();
   });
 
-  it("records producer-owned empty output without consulting transcript fallback", async () => {
-    const didAnnounce = await runSubagentAnnounceFlow({
-      childSessionKey: "agent:main:subagent:test",
-      childRunId: "run-direct-completion-empty",
-      requesterSessionKey: "agent:main:main",
-      requesterDisplayKey: "main",
-      requesterOrigin: { channel: "slack", to: "channel:C123", accountId: "acct-1" },
-      ...defaultOutcomeAnnounce,
-      expectsCompletionMessage: true,
-      terminalReply: { disposition: "empty" },
-    });
+  it.each([
+    {
+      name: "visible",
+      terminalReply: { disposition: "visible", text: "restored visible reply" } as const,
+      expectedAgentCalls: 1,
+      expectedMessage: "restored visible reply",
+    },
+    {
+      name: "silent",
+      terminalReply: { disposition: "silent" } as const,
+      expectedAgentCalls: 0,
+      expectedMessage: undefined,
+    },
+    {
+      name: "empty",
+      terminalReply: { disposition: "empty" } as const,
+      expectedAgentCalls: 1,
+      expectedMessage: "(no output)",
+    },
+  ])(
+    "replays restored durable $name output without transcript inference",
+    async ({ name, terminalReply, expectedAgentCalls, expectedMessage }) => {
+      const didAnnounce = await runSubagentAnnounceFlow({
+        childSessionKey: "agent:main:subagent:test",
+        childRunId: `run-restored-completion-${name}`,
+        requesterSessionKey: "agent:main:main",
+        requesterDisplayKey: "main",
+        requesterOrigin: { channel: "slack", to: "channel:C123", accountId: "acct-1" },
+        ...defaultOutcomeAnnounce,
+        expectsCompletionMessage: true,
+        terminalReply,
+      });
 
-    expect(didAnnounce).toBe(true);
-    expect(chatHistoryMock).not.toHaveBeenCalled();
-    expect(readLatestAssistantReplyMock).not.toHaveBeenCalled();
-    expect(getAgentCall()?.params?.message).toContain("(no output)");
-  });
+      expect(didAnnounce).toBe(true);
+      expect(chatHistoryMock).not.toHaveBeenCalled();
+      expect(readLatestAssistantReplyMock).not.toHaveBeenCalled();
+      expect(agentSpy).toHaveBeenCalledTimes(expectedAgentCalls);
+      if (expectedMessage) {
+        expect(getAgentCall()?.params?.message).toContain(expectedMessage);
+      }
+    },
+  );
 
   it("uses fallback reply when wake continuation returns NO_REPLY", async () => {
     const didAnnounce = await runSubagentAnnounceFlow({

src/agents/subagent-registry.store.sqlite.test.ts

@@ -139,6 +139,68 @@ describe("subagent registry sqlite store", () => {
     });
   });
 
+  it.each([
+    {
+      name: "visible",
+      terminalReply: { disposition: "visible", text: "restart-visible" } as const,
+      resultText: "restart-visible",
+    },
+    {
+      name: "silent",
+      terminalReply: { disposition: "silent" } as const,
+      resultText: "NO_REPLY",
+    },
+    {
+      name: "empty",
+      terminalReply: { disposition: "empty" } as const,
+      resultText: null,
+    },
+  ])(
+    "restores $name terminal reply in completion and pending delivery after restart",
+    async ({ name, terminalReply, resultText }) => {
+      await withTempStateEnv(async () => {
+        const runId = `run-restart-${name}`;
+        const run = createRun({
+          runId,
+          childSessionKey: `agent:main:subagent:${name}`,
+          completion: {
+            required: true,
+            resultText,
+            capturedAt: 260,
+            terminalReply,
+          },
+          delivery: {
+            status: "pending",
+            createdAt: 270,
+            attemptCount: 0,
+            payload: {
+              requesterSessionKey: "agent:main:main",
+              requesterDisplayKey: "main",
+              childSessionKey: `agent:main:subagent:${name}`,
+              childRunId: runId,
+              task: "check terminal reply restart",
+              startedAt: 110,
+              endedAt: 250,
+              outcome: { status: "ok" },
+              expectsCompletionMessage: true,
+              terminalReply,
+            },
+          },
+        });
+
+        saveSubagentRegistryToSqlite(new Map([[runId, run]]));
+        closeOpenClawStateDatabaseForTest();
+
+        const restored = loadSubagentRegistryFromSqlite().get(runId);
+        expect(restored?.completion).toMatchObject({ terminalReply, resultText });
+        expect(restored?.delivery).toMatchObject({
+          status: "pending",
+          payload: { terminalReply },
+        });
+      });
+    },
+  );
+
   it("uses save calls as whole-registry snapshots", async () => {
     await withTempStateEnv(async () => {
       const first = createRun({ runId: "run-one", childSessionKey: "agent:main:subagent:one" });