diff --git a/src/agents/acp-spawn.test.ts b/src/agents/acp-spawn.test.ts index b682f8830732..647449f62491 100644 --- a/src/agents/acp-spawn.test.ts +++ b/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; @@ -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()), + ...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()), 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()), listTasksForOwnerKey: hoisted.listTasksForOwnerKeyMock, })); @@ -405,6 +424,54 @@ function expectAcceptedSpawn(result: SpawnResult): Extract; + runSubagentAnnounceFlow: ReturnType; +}) { + 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 >(_opts: CallGatewayOptions): Promise => + ({}) 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, @@ -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(); + 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, diff --git a/src/agents/subagent-announce.format.e2e.test.ts b/src/agents/subagent-announce.format.e2e.test.ts index c3c60f9530c3..a834e0a25be7 100644 --- a/src/agents/subagent-announce.format.e2e.test.ts +++ b/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({ diff --git a/src/agents/subagent-registry.store.sqlite.test.ts b/src/agents/subagent-registry.store.sqlite.test.ts index d28594852992..68e11fe510fb 100644 --- a/src/agents/subagent-registry.store.sqlite.test.ts +++ b/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" });