diff --git a/src/agents/subagent-orphan-recovery.test.ts b/src/agents/subagent-orphan-recovery.test.ts index 853b0193be05..51ecef41d1d4 100644 --- a/src/agents/subagent-orphan-recovery.test.ts +++ b/src/agents/subagent-orphan-recovery.test.ts @@ -5,7 +5,6 @@ import * as config from "../config/config.js"; import * as sessions from "../config/sessions.js"; import * as sessionAccessor from "../config/sessions/session-accessor.js"; import type { GatewayRecoveryRuntime } from "../gateway/server-instance-runtime.types.js"; -import * as sessionUtils from "../gateway/session-transcript-readers.js"; import { getActiveGatewayRootWorkCount, resetGatewayWorkAdmission, @@ -28,6 +27,7 @@ const loggerMocks = vi.hoisted(() => ({ const dispatchAgent = vi.fn(async (_payload: Record, _timeoutMs?: number) => ({ runId: "test-run-id", })); +const readSessionMessages = vi.fn(async () => [] as unknown[]); const gatewayRuntime: GatewayRecoveryRuntime = { dispatchAgent: dispatchAgent as GatewayRecoveryRuntime["dispatchAgent"], waitForAgent: vi.fn(), @@ -35,15 +35,29 @@ const gatewayRuntime: GatewayRecoveryRuntime = { }; function recoverOrphanedSubagentSessions( - params: Omit[0], "gatewayRuntime">, + params: Omit< + Parameters[0], + "gatewayRuntime" | "readSessionMessages" + >, ) { - return recoverOrphanedSubagentSessionsWithRuntime({ ...params, gatewayRuntime }); + return recoverOrphanedSubagentSessionsWithRuntime({ + ...params, + gatewayRuntime, + readSessionMessages, + }); } function scheduleOrphanRecovery( - params: Omit[0], "getGatewayRuntime">, + params: Omit< + Parameters[0], + "getGatewayRuntime" | "readSessionMessages" + >, ) { - return scheduleOrphanRecoveryWithRuntime({ ...params, getGatewayRuntime: () => gatewayRuntime }); + return scheduleOrphanRecoveryWithRuntime({ + ...params, + getGatewayRuntime: () => gatewayRuntime, + readSessionMessages, + }); } // Mocks are installed before importing the recovery module so registry/runtime @@ -116,10 +130,6 @@ vi.mock("../config/sessions/session-accessor.js", () => ({ patchSessionEntry: sessionMocks.patchSessionEntry, })); -vi.mock("../gateway/session-transcript-readers.js", () => ({ - readSessionMessagesAsync: vi.fn(async () => []), -})); - vi.mock("./subagent-announce-delivery.js", () => ({ deliverSubagentAnnouncement: vi.fn(async () => ({ delivered: true, path: "direct" })), isInternalAnnounceRequesterSession: vi.fn(() => false), @@ -210,6 +220,8 @@ describe("subagent-orphan-recovery", () => { resetGatewayWorkAdmission(); dispatchAgent.mockReset(); dispatchAgent.mockResolvedValue({ runId: "test-run-id" }); + readSessionMessages.mockReset(); + readSessionMessages.mockResolvedValue([]); vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun) .mockReset() .mockResolvedValue(1); @@ -740,7 +752,7 @@ describe("subagent-orphan-recovery", () => { it("includes last human message in resume when available", async () => { mockSingleAbortedSession({ sessionFile: "session-abc.jsonl" }); - vi.mocked(sessionUtils.readSessionMessagesAsync).mockResolvedValue([ + readSessionMessages.mockResolvedValue([ { role: "user", content: [{ type: "text", text: "Please build feature Y" }] }, { role: "assistant", content: [{ type: "text", text: "Working on it..." }] }, { role: "user", content: [{ type: "text", text: "Also add tests for it" }] }, @@ -759,7 +771,7 @@ describe("subagent-orphan-recovery", () => { it("adds config change hint when assistant messages reference config modifications", async () => { mockSingleAbortedSession(); - vi.mocked(sessionUtils.readSessionMessagesAsync).mockResolvedValue([ + readSessionMessages.mockResolvedValue([ { role: "user", content: "Update the config" }, { role: "assistant", content: "I've modified openclaw.json to add the new setting." }, ]); @@ -916,6 +928,7 @@ describe("subagent-orphan-recovery", () => { scheduleOrphanRecoveryWithRuntime({ getGatewayRuntime: () => currentRuntime, getActiveRuns: () => createActiveRuns(createTestRunRecord()), + readSessionMessages, delayMs: 1, maxRetries: 0, }); diff --git a/src/agents/subagent-orphan-recovery.ts b/src/agents/subagent-orphan-recovery.ts index 4b0794e65e66..63ff4ecef230 100644 --- a/src/agents/subagent-orphan-recovery.ts +++ b/src/agents/subagent-orphan-recovery.ts @@ -231,6 +231,8 @@ async function resumeOrphanedSession(params: { export async function recoverOrphanedSubagentSessions(params: { gatewayRuntime: GatewayRecoveryRuntime; getActiveRuns: () => Map; + /** Test seam for transcript reads; production uses the canonical reader. */ + readSessionMessages?: typeof readSessionMessagesAsync; /** Persisted across retries so already-resumed sessions are not resumed again. */ resumedSessionKeys?: Set; /** Exact stale generations whose terminal transition must retry without session state. */ @@ -249,6 +251,7 @@ export async function recoverOrphanedSubagentSessions(params: { }; const resumedSessionKeys = params.resumedSessionKeys ?? new Set(); const pendingStaleFinalizations = params.pendingStaleFinalizations ?? new Map(); + const readSessionMessages = params.readSessionMessages ?? readSessionMessagesAsync; const configChangePattern = /openclaw\.json|openclaw gateway restart|config\.patch/i; try { @@ -431,7 +434,7 @@ export async function recoverOrphanedSubagentSessions(params: { log.info(`found orphaned subagent session: ${childSessionKey} (run=${runId})`); - const messages = await readSessionMessagesAsync( + const messages = await readSessionMessages( { agentId: resolveAgentIdFromSessionKey(childSessionKey), sessionEntry: entry, @@ -592,6 +595,8 @@ async function finalizeInterruptedRunWithRetry(params: { export function scheduleOrphanRecovery(params: { getGatewayRuntime: () => GatewayRecoveryRuntime | undefined; getActiveRuns: () => Map; + /** Test seam for transcript reads; production uses the canonical reader. */ + readSessionMessages?: typeof readSessionMessagesAsync; delayMs?: number; maxRetries?: number; }): void { @@ -617,6 +622,7 @@ export function scheduleOrphanRecovery(params: { const result = await recoverOrphanedSubagentSessions({ gatewayRuntime, getActiveRuns: params.getActiveRuns, + readSessionMessages: params.readSessionMessages ?? readSessionMessagesAsync, resumedSessionKeys, pendingStaleFinalizations, });