From 997a564c281ec7cd7f2a896d75c45a6bf256d732 Mon Sep 17 00:00:00 2001 From: pick-cat Date: Sat, 11 Jul 2026 04:56:57 +0800 Subject: [PATCH] fix(agents): apply stale-run liveness check to aborted subagent orphan recovery (#90817) * fix(agents): apply stale-run liveness to aborted subagent orphan recovery Skip stale unended subagent runs during orphan recovery and registry restore, even when they carry abortedLastRun. Previously, restart-aborted runs were exempt from the stale-unended age check, allowing hours-old aborted child sessions to be resurrected after long downtime. Fixes #90766 Co-Authored-By: Claude Opus 4.7 * fix(agents): finalize stale aborted runs instead of only skipping them Previously the stale-run check in recoverOrphanedSubagentSessions only incremented the skipped counter. Stale active runs were left unended because scheduleOrphanRecovery only retries failedRuns, not skipped runs. Now stale runs are finalized via finalizeInterruptedSubagentRun so they don't remain orphaned in the registry. Ref: #90766 review feedback Co-Authored-By: Claude Opus 4.7 * fix(agents): await stale-run finalization in orphan recovery Await finalizeInterruptedSubagentRun for stale aborted runs and report failedRuns when finalization does not update the registry, so the scheduler retry path can recover from finalization failures. Co-authored-by: Cursor * test(agents): faithful restart-path proof for stale orphan recovery Drive the real recoverOrphanedSubagentSessions against the real subagent registry, the real isStaleUnendedSubagentRun policy, and a real on-disk session store, mocking only the outbound gateway transport and transcript reader. Proves finalizeInterruptedSubagentRun actually ends the stale aborted run in the registry (endedAt set, outcome error) instead of resuming it, while a fresh aborted run still resumes. Co-Authored-By: Claude Opus 4.8 * chore: amend author email * fix(agents): scope orphan finalization to run generation * fix(agents): preserve stale restart recovery ownership * test(agents): isolate restart recovery scheduling * fix(agents): make stale restart finalization durable * fix(agents): keep stale retries generation-scoped * test(agents): await restart recovery scheduling * fix(agents): verify interrupted finalization * fix(agents): defer interrupted finalization during restart * fix(agents): preserve lifecycle type narrowing * style(agents): use nonmutating recovery ordering * chore: keep release note in PR body --------- Co-authored-by: Claude Opus 4.7 Co-authored-by: Cursor Co-authored-by: Pick-cat <266665499+Pick-cat@users.noreply.github.com> Co-authored-by: Peter Steinberger --- docs/tools/subagents.md | 8 +- src/agents/subagent-delivery-state.test.ts | 19 ++ src/agents/subagent-delivery-state.ts | 8 + ...rphan-recovery.restart-integration.test.ts | 267 +++++++++++++++++ src/agents/subagent-orphan-recovery.test.ts | 270 +++++++++++++++++- src/agents/subagent-orphan-recovery.ts | 171 ++++++++++- src/agents/subagent-registry-helpers.ts | 6 +- .../subagent-registry-lifecycle.test.ts | 235 +++++++++++++++ src/agents/subagent-registry-lifecycle.ts | 130 +++++++-- src/agents/subagent-registry-run-manager.ts | 2 + src/agents/subagent-registry-steer-runtime.ts | 5 +- .../subagent-registry.persistence.test.ts | 48 +++- .../subagent-registry.store.sqlite.test.ts | 8 +- src/agents/subagent-registry.test.ts | 131 ++++++++- src/agents/subagent-registry.ts | 110 ++++--- src/agents/subagent-registry.types.ts | 2 + 16 files changed, 1322 insertions(+), 98 deletions(-) create mode 100644 src/agents/subagent-orphan-recovery.restart-integration.test.ts diff --git a/docs/tools/subagents.md b/docs/tools/subagents.md index d3d3c26b134e..0d2c5cadfced 100644 --- a/docs/tools/subagents.md +++ b/docs/tools/subagents.md @@ -611,10 +611,10 @@ status summaries, descendant completion gating, and per-session concurrency checks. After a gateway restart, stale unended restored runs are pruned unless -their child session is marked `abortedLastRun: true`. Those -restart-aborted child sessions remain recoverable through the sub-agent -orphan recovery flow, which sends a synthetic resume message before -clearing the aborted marker. +their child session is marked `abortedLastRun: true`. Restart-aborted +runs remain registered for the sub-agent orphan recovery flow: stale +runs are finalized without a resume, while fresh child sessions receive +a synthetic resume message before the aborted marker is cleared. Automatic restart recovery is bounded per child session. If the same sub-agent child is accepted for orphan recovery repeatedly inside the diff --git a/src/agents/subagent-delivery-state.test.ts b/src/agents/subagent-delivery-state.test.ts index 8df09a226c76..c716bf8112fb 100644 --- a/src/agents/subagent-delivery-state.test.ts +++ b/src/agents/subagent-delivery-state.test.ts @@ -79,6 +79,25 @@ describe("normalizeSubagentRunState", () => { expect(entry.killReconciliation).toBeUndefined(); }); + it("keeps only complete interrupted-recovery terminal ownership", () => { + const terminal = { + endedAt: 200, + endedReason: "subagent-error" as const, + outcome: { status: "error" as const, error: "restart interrupted run" }, + terminalOwner: "interrupted-recovery" as const, + }; + const valid = normalizeSubagentRunState(baseRun(terminal)); + const malformed = [ + baseRun({ ...terminal, endedAt: undefined }), + baseRun({ ...terminal, outcome: { status: "ok" } }), + baseRun({ ...terminal, endedReason: "subagent-complete" }), + baseRun({ ...terminal, pauseReason: "sessions_yield" }), + ].map((entry) => normalizeSubagentRunState(entry)); + + expect(valid.terminalOwner).toBe("interrupted-recovery"); + expect(malformed.every((entry) => entry.terminalOwner === undefined)).toBe(true); + }); + it("migrates legacy pending delivery fields into nested completion and delivery state", () => { // Restored runs may still carry flat pendingFinalDelivery fields from older // builds; normalization must preserve retry payloads before stripping them. diff --git a/src/agents/subagent-delivery-state.ts b/src/agents/subagent-delivery-state.ts index 1c366fac43cd..f32be6ab91f3 100644 --- a/src/agents/subagent-delivery-state.ts +++ b/src/agents/subagent-delivery-state.ts @@ -60,6 +60,14 @@ export function normalizeSubagentRunState(entry: SubagentRunRecord): SubagentRun ? entry.deleteCleanupDispatchedAt : undefined; entry.suppressCompletionDelivery = entry.suppressCompletionDelivery === true ? true : undefined; + entry.terminalOwner = + entry.terminalOwner === "interrupted-recovery" && + Number.isFinite(entry.endedAt) && + entry.outcome?.status === "error" && + entry.endedReason === "subagent-error" && + entry.pauseReason !== "sessions_yield" + ? "interrupted-recovery" + : undefined; const killReconciliation = entry.killReconciliation; if ( !killReconciliation || diff --git a/src/agents/subagent-orphan-recovery.restart-integration.test.ts b/src/agents/subagent-orphan-recovery.restart-integration.test.ts new file mode 100644 index 000000000000..6f19ba8c4a34 --- /dev/null +++ b/src/agents/subagent-orphan-recovery.restart-integration.test.ts @@ -0,0 +1,267 @@ +// Faithful restart-path integration proof for stale-aborted subagent orphan +// recovery. Unlike subagent-orphan-recovery.test.ts (which stubs the session +// store and finalize), this drives the REAL recovery pass against the REAL +// subagent registry, the REAL liveness policy, and a REAL on-disk session +// store. Only the outbound gateway transport and the transcript file reader are +// mocked, because they are the genuine process boundaries a single-process test +// cannot stand up. It exists to prove that finalize actually ends the real +// registry run (not a stubbed counter) and that the fresh run still resumes. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { setRuntimeConfigSnapshot } from "../config/config.js"; +import { + clearSessionStoreCacheForTest, + drainSessionStoreWriterQueuesForTest, +} from "../config/sessions/store.js"; +import { callGateway } from "../gateway/call.js"; +import { createRunningTaskRun } from "../tasks/detached-task-runtime.js"; +import { resetTaskFlowRegistryForTests } from "../tasks/task-flow-registry.js"; +import { findTaskByRunId, resetTaskRegistryForTests } from "../tasks/task-registry.js"; +import { captureEnv } from "../test-utils/env.js"; +import { recoverOrphanedSubagentSessions } from "./subagent-orphan-recovery.js"; +import { + addSubagentRunForTests, + finalizeInterruptedSubagentRun, + getSubagentRunByChildSessionKey, + listSubagentRunsForRequester, + resetSubagentRegistryForTests, + testing, +} from "./subagent-registry.js"; +import { + createSubagentRegistryTestDeps, + readSubagentSessionStore, + writeSubagentSessionEntry, +} from "./subagent-registry.persistence.test-support.js"; +import type { SubagentRunRecord } from "./subagent-registry.types.js"; + +vi.mock("../gateway/call.js", () => ({ + callGateway: vi.fn(async () => ({ runId: "resumed-run-id" })), +})); + +vi.mock("../gateway/session-utils.fs.js", () => ({ + readSessionMessagesAsync: vi.fn(async () => []), +})); + +const TWO_HOURS_MS = 2 * 60 * 60 * 1_000; + +function makeRunRecord(overrides: Partial): SubagentRunRecord { + return { + runId: "run", + childSessionKey: "agent:main:subagent:child", + requesterSessionKey: "agent:main:main", + requesterDisplayKey: "main", + task: "restart-recoverable work", + cleanup: "keep", + createdAt: Date.now(), + startedAt: Date.now(), + ...overrides, + } as SubagentRunRecord; +} + +describe("subagent orphan recovery — faithful restart path", () => { + const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); + let tempStateDir: string | null = null; + + beforeEach(async () => { + resetTaskRegistryForTests({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + tempStateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-orphan-integ-")); + process.env.OPENCLAW_STATE_DIR = tempStateDir; + setRuntimeConfigSnapshot({ session: { store: undefined } } as never); + // Real registry wiring: only the delivery/announce/cleanup seams (true + // external side effects) are recorded so completeSubagentRun runs in-process. + testing.setDepsForTest({ + ...createSubagentRegistryTestDeps(), + runSubagentAnnounceFlow: vi.fn(async () => true), + onAgentEvent: vi.fn(() => () => undefined), + }); + vi.mocked(callGateway).mockClear(); + vi.mocked(callGateway).mockResolvedValue({ runId: "resumed-run-id" } as never); + }); + + afterEach(async () => { + testing.setDepsForTest(); + resetSubagentRegistryForTests({ persist: false }); + await drainSessionStoreWriterQueuesForTest(); + clearSessionStoreCacheForTest(); + resetTaskRegistryForTests({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + if (tempStateDir) { + await fs.rm(tempStateDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + tempStateDir = null; + } + envSnapshot.restore(); + }); + + it("finalizes a stale (>2h) aborted run in the real registry instead of resuming it", async () => { + const now = Date.now(); + const childSessionKey = "agent:main:subagent:stale-aborted"; + const runId = "run-stale-aborted"; + const storePath = await writeSubagentSessionEntry({ + stateDir: tempStateDir!, + agentId: "main", + sessionKey: childSessionKey, + sessionId: "sess-stale-aborted", + updatedAt: now, + abortedLastRun: true, + defaultSessionId: "sess-stale-aborted", + }); + const record = makeRunRecord({ + runId, + childSessionKey, + createdAt: now - 3 * TWO_HOURS_MS, + startedAt: now - 3 * TWO_HOURS_MS, + }); + expect( + createRunningTaskRun({ + runtime: "subagent", + sourceId: runId, + ownerKey: record.requesterSessionKey, + scopeKind: "session", + childSessionKey, + runId, + task: record.task, + deliveryStatus: "pending", + startedAt: record.startedAt, + lastEventAt: record.startedAt, + }), + ).not.toBeNull(); + addSubagentRunForTests(record); + + const before = getSubagentRunByChildSessionKey(childSessionKey); + console.log( + `[proof] before recovery: stale run endedAt=${before?.endedAt ?? "undefined"} outcome=${ + before?.outcome?.status ?? "undefined" + }`, + ); + + const result = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => new Map([[runId, record]]), + }); + + const after = getSubagentRunByChildSessionKey(childSessionKey); + console.log( + `[proof] after recovery: result=${JSON.stringify(result)} endedAt=${ + after?.endedAt ?? "undefined" + } outcome=${after?.outcome?.status ?? "undefined"}`, + ); + + // Stale aborted run was finalized in the real registry, not resumed. + expect(vi.mocked(callGateway)).not.toHaveBeenCalled(); + expect(after?.endedAt).toBeTypeOf("number"); + expect(after?.outcome?.status).toBe("error"); + expect(result.recovered).toBe(0); + expect(findTaskByRunId(runId)).toMatchObject({ + status: "failed", + endedAt: expect.any(Number), + error: expect.stringContaining("stale aborted subagent run not resumed"), + }); + + // The task finalizer and session projection are durable, not only + // in-memory side effects of the recovery pass. + resetTaskRegistryForTests({ persist: false }); + expect(findTaskByRunId(runId)).toMatchObject({ status: "failed" }); + await drainSessionStoreWriterQueuesForTest(); + const persistedSession = (await readSubagentSessionStore(storePath))[childSessionKey]; + expect(persistedSession).toMatchObject({ + status: "failed", + endedAt: expect.any(Number), + }); + expect(persistedSession?.abortedLastRun).toBeUndefined(); + }); + + it("resumes a fresh (<2h) aborted run through the real recovery pass", async () => { + const now = Date.now(); + const childSessionKey = "agent:main:subagent:fresh-aborted"; + const runId = "run-fresh-aborted"; + await writeSubagentSessionEntry({ + stateDir: tempStateDir!, + agentId: "main", + sessionKey: childSessionKey, + sessionId: "sess-fresh-aborted", + updatedAt: now, + abortedLastRun: true, + defaultSessionId: "sess-fresh-aborted", + }); + const record = makeRunRecord({ + runId, + childSessionKey, + createdAt: now - 60_000, + startedAt: now - 55_000, + }); + addSubagentRunForTests(record); + + const result = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => new Map([[runId, record]]), + }); + + console.log( + `[proof] fresh recovery: result=${JSON.stringify(result)} gatewayCalls=${ + vi.mocked(callGateway).mock.calls.length + }`, + ); + + // Fresh aborted run passed the stale gate and reached a real resume call. + const agentCalls = vi + .mocked(callGateway) + .mock.calls.filter((args) => (args[0] as { method?: string })?.method === "agent"); + expect(agentCalls).toHaveLength(1); + expect(result.recovered).toBe(1); + }); + + it("finalizes only a stale predecessor when a fresh generation shares its child session", async () => { + const now = Date.now(); + const childSessionKey = "agent:main:subagent:shared-generation"; + const staleRecord = makeRunRecord({ + runId: "run-stale-generation", + childSessionKey, + generation: 1, + createdAt: now - 3 * 60 * 60 * 1_000, + startedAt: now - 3 * 60 * 60 * 1_000, + sessionStartedAt: now - 3 * 60 * 60 * 1_000, + }); + const freshRecord = makeRunRecord({ + runId: "run-fresh-generation", + childSessionKey, + generation: 2, + createdAt: now - 60_000, + startedAt: now - 55_000, + sessionStartedAt: now - 60_000, + }); + for (const record of [staleRecord, freshRecord]) { + expect( + createRunningTaskRun({ + runtime: "subagent", + sourceId: record.runId, + ownerKey: record.requesterSessionKey, + scopeKind: "session", + childSessionKey, + runId: record.runId, + task: record.task, + deliveryStatus: "pending", + startedAt: record.startedAt, + lastEventAt: record.startedAt, + }), + ).not.toBeNull(); + } + addSubagentRunForTests(staleRecord); + addSubagentRunForTests(freshRecord); + + const updated = await finalizeInterruptedSubagentRun({ + runId: staleRecord.runId, + error: "stale predecessor interrupted by restart", + endedAt: now, + }); + + const runs = listSubagentRunsForRequester("agent:main:main"); + expect(updated).toBe(1); + expect(callGateway).not.toHaveBeenCalled(); + expect(runs.some((entry) => entry.runId === staleRecord.runId)).toBe(false); + expect(runs).toContainEqual(expect.objectContaining({ runId: freshRecord.runId })); + expect(runs.find((entry) => entry.runId === freshRecord.runId)?.endedAt).toBeUndefined(); + expect(findTaskByRunId(staleRecord.runId)).toMatchObject({ status: "failed" }); + expect(findTaskByRunId(freshRecord.runId)).toMatchObject({ status: "running" }); + }); +}); diff --git a/src/agents/subagent-orphan-recovery.test.ts b/src/agents/subagent-orphan-recovery.test.ts index 56bf01289bae..3bdfc6bd17b5 100644 --- a/src/agents/subagent-orphan-recovery.test.ts +++ b/src/agents/subagent-orphan-recovery.test.ts @@ -1,6 +1,7 @@ // Subagent orphan-recovery tests cover restart recovery for child sessions whose // embedded run was interrupted while the registry still considers them active. import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import * as config from "../config/config.js"; import * as sessions from "../config/sessions.js"; import * as gateway from "../gateway/call.js"; import * as sessionUtils from "../gateway/session-transcript-readers.js"; @@ -18,6 +19,11 @@ import { import * as subagentRegistrySteerRuntime from "./subagent-registry-steer-runtime.js"; import type { SubagentRunRecord } from "./subagent-registry.types.js"; +const loggerMocks = vi.hoisted(() => ({ + info: vi.fn(), + warn: vi.fn(), +})); + // Mocks are installed before importing the recovery module so registry/runtime // helpers resolve to deterministic restart fixtures. vi.mock("../config/config.js", () => ({ @@ -26,6 +32,10 @@ vi.mock("../config/config.js", () => ({ })), })); +vi.mock("../logging/subsystem.js", () => ({ + createSubsystemLogger: () => loggerMocks, +})); + vi.mock("../config/sessions.js", () => ({ loadSessionStore: vi.fn(() => ({})), resolveAgentIdFromSessionKey: vi.fn(() => "main"), @@ -136,6 +146,9 @@ describe("subagent-orphan-recovery", () => { vi.useFakeTimers(); vi.clearAllMocks(); resetGatewayWorkAdmission(); + vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun) + .mockReset() + .mockResolvedValue(1); }); afterEach(() => { @@ -208,6 +221,143 @@ describe("subagent-orphan-recovery", () => { }); }); + it("finalizes stale aborted runs instead of resuming them", async () => { + mockSingleAbortedSession(); + const now = Date.now(); + const staleSessionStartedAt = now - 3 * 60 * 60 * 1_000; + const activeRuns = createActiveRuns( + createTestRunRecord({ + createdAt: staleSessionStartedAt, + startedAt: now - 60_000, + sessionStartedAt: staleSessionStartedAt, + }), + ); + + const result = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + }); + + expect(result.recovered).toBe(0); + expect(result.failed).toBe(0); + expect(result.skipped).toBe(1); + expect(gateway.callGateway).not.toHaveBeenCalled(); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenCalledOnce(); + const finalizeParams = requireRecord( + firstCallParam( + vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).mock.calls, + "stale finalize", + ), + "stale finalize params", + ); + expect(finalizeParams).toEqual({ + runId: "run-1", + error: "stale aborted subagent run not resumed (10800s old, exceeds stale-run window)", + }); + }); + + it("reports stale finalization failures for scheduler retry", async () => { + mockSingleAbortedSession(); + vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).mockResolvedValueOnce(0); + const staleStartedAt = Date.now() - 3 * 60 * 60 * 1_000; + const activeRuns = createActiveRuns( + createTestRunRecord({ + createdAt: staleStartedAt, + startedAt: staleStartedAt, + }), + ); + + const result = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + }); + + expect(result.recovered).toBe(0); + expect(result.failed).toBe(1); + expect(result.skipped).toBe(0); + expect(result.failedRuns).toEqual([ + { + runId: "run-1", + childSessionKey: "agent:main:subagent:test-session-1", + error: expect.stringContaining("stale aborted subagent run not resumed"), + }, + ]); + expect(gateway.callGateway).not.toHaveBeenCalled(); + }); + + it("retries a stale predecessor after its same-session successor resumes", async () => { + const now = Date.now(); + const staleStartedAt = now - 3 * 60 * 60 * 1_000; + const childSessionKey = "agent:main:subagent:test-session-1"; + const store = { + [childSessionKey]: { + sessionId: "session-abc", + updatedAt: now, + abortedLastRun: true, + }, + }; + vi.mocked(sessions.loadSessionStore).mockReturnValue(store); + vi.mocked(sessions.updateSessionStore).mockImplementation(async (_storePath, update) => { + update(store); + }); + const activeRuns = createActiveRuns( + createTestRunRecord({ + runId: "fresh-run", + childSessionKey, + createdAt: now - 60_000, + startedAt: now - 55_000, + sessionStartedAt: now - 2 * 60 * 60 * 1_000 + 60_000, + }), + createTestRunRecord({ + runId: "stale-run", + childSessionKey, + createdAt: staleStartedAt, + startedAt: staleStartedAt, + }), + ); + vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun) + .mockResolvedValueOnce(0) + .mockResolvedValueOnce(1); + vi.mocked(subagentRegistrySteerRuntime.replaceSubagentRunAfterSteer).mockImplementation( + ({ previousRunId, nextRunId, fallback }) => { + const previous = activeRuns.get(previousRunId) ?? fallback; + if (!previous) { + return false; + } + activeRuns.delete(previousRunId); + activeRuns.set(nextRunId, { ...previous, runId: nextRunId }); + return true; + }, + ); + const resumedSessionKeys = new Set(); + const pendingStaleFinalizations = new Map(); + + const first = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + resumedSessionKeys, + pendingStaleFinalizations, + }); + expect(first).toMatchObject({ recovered: 1, failed: 1, skipped: 0 }); + expect(store[childSessionKey].abortedLastRun).toBe(false); + Reflect.deleteProperty(store, childSessionKey); + vi.setSystemTime(now + 2 * 60_000); + + const second = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + resumedSessionKeys, + pendingStaleFinalizations, + }); + + expect(second).toMatchObject({ recovered: 0, failed: 0, skipped: 2 }); + expect(gateway.callGateway).toHaveBeenCalledOnce(); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenCalledTimes(2); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ runId: "stale-run" }), + ); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).not.toHaveBeenCalledWith( + expect.objectContaining({ runId: "test-run-id" }), + ); + }); + it("skips runs that have already ended", async () => { const activeRuns = new Map(); activeRuns.set( @@ -235,14 +385,15 @@ describe("subagent-orphan-recovery", () => { }, }); - const activeRuns = createActiveRuns( - createTestRunRecord({ - endedAt: Date.now() - 1_000, - outcome: { - status: "timeout", - }, - }), - ); + const legacyTimeout = createTestRunRecord({ + endedAt: Date.now() - 1_000, + endedReason: "subagent-complete", + outcome: { + status: "timeout", + }, + terminalOwner: "interrupted-recovery", + }); + const activeRuns = createActiveRuns(legacyTimeout); const result = await recoverOrphanedSubagentSessions({ getActiveRuns: () => activeRuns, @@ -252,6 +403,40 @@ describe("subagent-orphan-recovery", () => { expect(result.failed).toBe(0); expect(result.skipped).toBe(0); expect(gateway.callGateway).toHaveBeenCalledOnce(); + expect(legacyTimeout.terminalOwner).toBeUndefined(); + }); + + it("replays interrupted terminal ownership before config or session lookup", async () => { + const run = createTestRunRecord({ + endedAt: 2_000, + endedReason: "subagent-error", + outcome: { status: "error", error: "restart interrupted run" }, + terminalOwner: "interrupted-recovery", + completion: { required: false, resultText: null, capturedAt: 2_000 }, + }); + const activeRuns = createActiveRuns(run); + const resumedSessionKeys = new Set([run.childSessionKey]); + + const first = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + resumedSessionKeys, + }); + const second = await recoverOrphanedSubagentSessions({ + getActiveRuns: () => activeRuns, + resumedSessionKeys, + }); + + expect(first).toMatchObject({ recovered: 0, failed: 0, skipped: 1 }); + expect(second).toMatchObject({ recovered: 0, failed: 0, skipped: 1 }); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenCalledTimes(2); + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenNthCalledWith(1, { + runId: "run-1", + error: "restart interrupted run", + endedAt: 2_000, + }); + expect(config.getRuntimeConfig).not.toHaveBeenCalled(); + expect(sessions.loadSessionStore).not.toHaveBeenCalled(); + expect(gateway.callGateway).not.toHaveBeenCalled(); }); it("handles multiple orphaned sessions", async () => { @@ -673,12 +858,12 @@ describe("subagent-orphan-recovery", () => { ), "interrupted run finalization params", ); - expect(finalizeParams.runId).toBe("run-1"); - expect(finalizeParams.childSessionKey).toBe("agent:main:subagent:test-session-1"); - expect(finalizeParams.error).toContain("Automatic recovery failed after 2 attempts"); - expect(finalizeParams.error).toContain("service restart"); + expect(finalizeParams).toEqual({ + runId: "run-1", + error: + "Subagent run was interrupted by a gateway restart or connection loss. Automatic recovery failed after 2 attempts. Please retry. (service restart)", + }); }); - it("waits for suspension to reopen before mutating an orphaned session", async () => { mockSingleAbortedSession(); vi.mocked(gateway.callGateway).mockResolvedValue({ runId: "resumed-run" }); @@ -703,4 +888,63 @@ describe("subagent-orphan-recovery", () => { expect(sessions.updateSessionStore).toHaveBeenCalledOnce(); expect(getActiveGatewayRootWorkCount()).toBe(0); }); + + it.each(["returns zero", "rejects"])( + "retries the exact interrupted terminal when finalization first %s", + async (mode) => { + mockSingleAbortedSession(); + vi.mocked(gateway.callGateway).mockRejectedValueOnce(new Error("service restart")); + if (mode === "returns zero") { + vi.mocked( + subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun, + ).mockResolvedValueOnce(0); + } else { + vi.mocked( + subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun, + ).mockRejectedValueOnce(new Error("registry unavailable")); + } + + scheduleOrphanRecovery({ + getActiveRuns: () => createActiveRuns(createTestRunRecord()), + delayMs: 1, + maxRetries: 0, + }); + + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersByTimeAsync(1); + await Promise.resolve(); + await Promise.resolve(); + + const finalize = vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun); + expect(finalize).toHaveBeenCalledTimes(2); + expect(finalize.mock.calls[1]).toEqual(finalize.mock.calls[0]); + expect(loggerMocks.warn).not.toHaveBeenCalledWith( + expect.stringContaining("interrupted terminal projection(s) incomplete"), + expect.anything(), + ); + }, + ); + + it("logs an incomplete interrupted terminal after its retry budget is exhausted", async () => { + mockSingleAbortedSession(); + vi.mocked(gateway.callGateway).mockRejectedValueOnce(new Error("service restart")); + vi.mocked(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).mockResolvedValue(0); + + scheduleOrphanRecovery({ + getActiveRuns: () => createActiveRuns(createTestRunRecord()), + delayMs: 1, + maxRetries: 0, + }); + + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersByTimeAsync(3); + await Promise.resolve(); + await Promise.resolve(); + + expect(subagentRegistrySteerRuntime.finalizeInterruptedSubagentRun).toHaveBeenCalledTimes(3); + expect(loggerMocks.warn).toHaveBeenCalledWith( + "orphan recovery exhausted with 1 interrupted terminal projection(s) incomplete", + { runIds: ["run-1"] }, + ); + }); }); diff --git a/src/agents/subagent-orphan-recovery.ts b/src/agents/subagent-orphan-recovery.ts index 92c6c947e822..a1f6817bf7e8 100644 --- a/src/agents/subagent-orphan-recovery.ts +++ b/src/agents/subagent-orphan-recovery.ts @@ -36,6 +36,8 @@ import { replaceSubagentRunAfterSteer, } from "./subagent-registry-steer-runtime.js"; import type { SubagentRunRecord } from "./subagent-registry.types.js"; +import { isStaleUnendedSubagentRun } from "./subagent-run-liveness.js"; +import { getSubagentSessionStartedAt } from "./subagent-session-metrics.js"; const log = createSubsystemLogger("subagent-interrupted-resume"); @@ -67,6 +69,7 @@ function reclassifyLegacyRestartInterruptedRun(runRecord: SubagentRunRecord): vo runRecord.endedAt = undefined; runRecord.endedReason = undefined; runRecord.outcome = undefined; + runRecord.terminalOwner = undefined; } /** @@ -192,6 +195,8 @@ export async function recoverOrphanedSubagentSessions(params: { getActiveRuns: () => Map; /** Persisted across retries so already-resumed sessions are not resumed again. */ resumedSessionKeys?: Set; + /** Exact stale generations whose terminal transition must retry without session state. */ + pendingStaleFinalizations?: Map; }): Promise<{ recovered: number; failed: number; @@ -205,6 +210,7 @@ export async function recoverOrphanedSubagentSessions(params: { failedRuns: [] as Array<{ runId: string; childSessionKey: string; error?: string }>, }; const resumedSessionKeys = params.resumedSessionKeys ?? new Set(); + const pendingStaleFinalizations = params.pendingStaleFinalizations ?? new Map(); const configChangePattern = /openclaw\.json|openclaw gateway restart|config\.patch/i; try { @@ -213,21 +219,81 @@ export async function recoverOrphanedSubagentSessions(params: { return result; } - const cfg = getRuntimeConfig(); + let cfg: ReturnType | undefined; const storeCache = new Map>(); + const scanNow = Date.now(); + const runEntries = [...activeRuns.entries()].toSorted(([, left], [, right]) => { + const leftIsStale = isStaleUnendedSubagentRun(left, scanNow); + const rightIsStale = isStaleUnendedSubagentRun(right, scanNow); + return Number(rightIsStale) - Number(leftIsStale); + }); - for (const [runId, runRecord] of activeRuns.entries()) { + for (const [runId, runRecord] of runEntries) { const childSessionKey = runRecord.childSessionKey?.trim(); if (!childSessionKey) { continue; } - const now = Date.now(); + const now = scanNow; + if ( + runRecord.terminalOwner === "interrupted-recovery" && + Number.isFinite(runRecord.endedAt) && + runRecord.outcome?.status === "error" && + runRecord.endedReason === "subagent-error" && + runRecord.pauseReason !== "sessions_yield" + ) { + const recoveryError = + runRecord.outcome?.status === "error" + ? (runRecord.outcome.error ?? "subagent run interrupted by gateway restart") + : "subagent run interrupted by gateway restart"; + try { + const updated = await finalizeInterruptedSubagentRun({ + runId, + error: recoveryError, + endedAt: runRecord.endedAt, + }); + if (updated === 0) { + result.failed++; + result.failedRuns.push({ runId, childSessionKey, error: recoveryError }); + } else { + pendingStaleFinalizations.delete(runId); + result.skipped++; + } + } catch (err: unknown) { + const error = formatErrorMessage(err); + log.warn(`replay interrupted terminal ${runId}: ${error}`); + result.failed++; + result.failedRuns.push({ runId, childSessionKey, error }); + } + continue; + } + const pendingStaleError = pendingStaleFinalizations.get(runId); + if (pendingStaleError) { + try { + const updated = await finalizeInterruptedSubagentRun({ + runId, + error: pendingStaleError, + }); + if (updated === 0) { + result.failed++; + result.failedRuns.push({ runId, childSessionKey, error: pendingStaleError }); + } else { + pendingStaleFinalizations.delete(runId); + result.skipped++; + } + } catch (err: unknown) { + const error = formatErrorMessage(err); + log.warn(`retry stale terminal ${runId}: ${error}`); + result.failed++; + result.failedRuns.push({ runId, childSessionKey, error }); + } + continue; + } if (resumedSessionKeys.has(childSessionKey)) { result.skipped++; continue; } - try { + cfg ??= getRuntimeConfig(); const agentId = resolveAgentIdFromSessionKey(childSessionKey); const storePath = resolveStorePath(cfg.session?.store, { agentId }); @@ -255,12 +321,49 @@ export async function recoverOrphanedSubagentSessions(params: { continue; } - // Check if this session was aborted by the restart if (!entry.abortedLastRun) { result.skipped++; continue; } + // Runs that are too old to be worth recovering must be finalized + // so they don't remain in an unended state. The scheduler only + // retries failedRuns; a plain skip would leave the run orphaned. + if (isStaleUnendedSubagentRun(runRecord, now)) { + const staleStartedAt = getSubagentSessionStartedAt(runRecord) ?? now; + const staleAgeSeconds = Math.round((now - staleStartedAt) / 1000); + const staleError = `stale aborted subagent run not resumed (${staleAgeSeconds}s old, exceeds stale-run window)`; + try { + const updated = await finalizeInterruptedSubagentRun({ + runId, + error: staleError, + }); + if (updated === 0) { + pendingStaleFinalizations.set(runId, staleError); + result.failed++; + result.failedRuns.push({ + runId, + childSessionKey, + error: staleError, + }); + } else { + pendingStaleFinalizations.delete(runId); + result.skipped++; + } + } catch (err: unknown) { + const error = formatErrorMessage(err); + log.warn(`finalize stale run ${runId}: ${error}`); + pendingStaleFinalizations.set(runId, staleError); + result.failed++; + result.failedRuns.push({ + runId, + childSessionKey, + error, + }); + } + continue; + } + const recoveryGate = evaluateSubagentRecoveryGate(entry, now); if (!recoveryGate.allowed) { if (recoveryGate.shouldMarkWedged) { @@ -409,6 +512,8 @@ export async function recoverOrphanedSubagentSessions(params: { const MAX_RECOVERY_RETRIES = 3; /** Backoff multiplier between retries (exponential). */ const RETRY_BACKOFF_MULTIPLIER = 2; +/** Separate durable-terminal attempts after session recovery is exhausted. */ +const MAX_TERMINAL_FINALIZE_ATTEMPTS = 3; function buildRecoveryFailureMessage(params: { attempts: number; error?: string }): string { const base = @@ -422,6 +527,35 @@ function buildRecoveryFailureMessage(params: { attempts: number; error?: string return `${base} (${detail})`; } +async function finalizeInterruptedRunWithRetry(params: { + runId: string; + error: string; + initialDelayMs: number; +}): Promise { + let delayMs = Math.max(1, params.initialDelayMs); + for (let attempt = 1; attempt <= MAX_TERMINAL_FINALIZE_ATTEMPTS; attempt += 1) { + try { + const updated = await finalizeInterruptedSubagentRun({ + runId: params.runId, + error: params.error, + }); + if (updated > 0) { + return true; + } + } catch { + // The outer scheduler owns this exact-run retry budget. + } + if (attempt < MAX_TERMINAL_FINALIZE_ATTEMPTS) { + await new Promise((resolve) => { + const timer = setTimeout(resolve, delayMs); + timer.unref?.(); + }); + delayMs *= RETRY_BACKOFF_MULTIPLIER; + } + } + return false; +} + /** * Schedule orphan recovery after a delay, with retry logic. * The delay gives the gateway time to fully bootstrap after restart. @@ -436,6 +570,7 @@ export function scheduleOrphanRecovery(params: { const maxRetries = params.maxRetries ?? MAX_RECOVERY_RETRIES; const resumedSessionKeys = new Set(); + const pendingStaleFinalizations = new Map(); const attemptRecovery = (attempt: number, delay: number) => { setTimeout(() => { // Every delayed/retry scan owns a fresh root lease. Keep terminal @@ -444,6 +579,7 @@ export function scheduleOrphanRecovery(params: { const result = await recoverOrphanedSubagentSessions({ ...params, resumedSessionKeys, + pendingStaleFinalizations, }); if (result.failed > 0 && attempt < maxRetries) { const nextDelay = delay * RETRY_BACKOFF_MULTIPLIER; @@ -457,18 +593,25 @@ export function scheduleOrphanRecovery(params: { return; } const attempts = attempt + 1; - await Promise.allSettled( - result.failedRuns.map((run) => - finalizeInterruptedSubagentRun({ + const terminalResults = await Promise.all( + result.failedRuns.map(async (run) => ({ + runId: run.runId, + completed: await finalizeInterruptedRunWithRetry({ runId: run.runId, - childSessionKey: run.childSessionKey, - error: buildRecoveryFailureMessage({ - attempts, - error: run.error, - }), + error: buildRecoveryFailureMessage({ attempts, error: run.error }), + initialDelayMs: delay, }), - ), + })), ); + const incomplete = terminalResults + .filter((terminal) => !terminal.completed) + .map((terminal) => terminal.runId); + if (incomplete.length > 0) { + log.warn( + `orphan recovery exhausted with ${incomplete.length} interrupted terminal projection(s) incomplete`, + { runIds: incomplete }, + ); + } }).catch((err: unknown) => { if (attempt < maxRetries) { const nextDelay = delay * RETRY_BACKOFF_MULTIPLIER; diff --git a/src/agents/subagent-registry-helpers.ts b/src/agents/subagent-registry-helpers.ts index f6c5e4311313..f680a4ace3f7 100644 --- a/src/agents/subagent-registry-helpers.ts +++ b/src/agents/subagent-registry-helpers.ts @@ -321,9 +321,9 @@ export function reconcileOrphanedRestoredRuns(params: { const now = Date.now(); let changed = false; for (const [runId, entry] of params.runs.entries()) { - if (entry.killReconciliation) { - // Provider completion may still repair this provisional kill. The - // sweeper owns its bounded reconciliation even when the session vanished. + if (entry.killReconciliation || entry.terminalOwner === "interrupted-recovery") { + // Provider completion or interrupted recovery still owns these rows. + // Their bounded reconciliation runs even when the session vanished. continue; } const orphanReason = resolveSubagentRunOrphanReason({ diff --git a/src/agents/subagent-registry-lifecycle.test.ts b/src/agents/subagent-registry-lifecycle.test.ts index a7077b0f2309..8b609ca5d76b 100644 --- a/src/agents/subagent-registry-lifecycle.test.ts +++ b/src/agents/subagent-registry-lifecycle.test.ts @@ -473,6 +473,241 @@ describe("subagent registry lifecycle hardening", () => { expect(taskExecutorMocks.completeTaskRunByRunId).not.toHaveBeenCalled(); }); + it("keeps a provider terminal when it acquires the completion lock first", async () => { + let finishCapture: ((value: string) => void) | undefined; + const entry = createRunEntry(); + const controller = createLifecycleController({ + entry, + captureSubagentCompletionReply: vi.fn( + () => + new Promise((resolve) => { + finishCapture = resolve; + }), + ), + }); + const providerCompletion = controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "ok" }, + reason: SUBAGENT_ENDED_REASON_COMPLETE, + triggerCleanup: false, + }); + await vi.waitFor(() => expect(finishCapture).toBeTypeOf("function")); + const interruptedRecovery = controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_001, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + + finishCapture?.("provider result"); + await Promise.all([providerCompletion, interruptedRecovery]); + + expect(entry.outcome?.status).toBe("ok"); + expect(entry.endedReason).toBe(SUBAGENT_ENDED_REASON_COMPLETE); + expect(entry.terminalOwner).toBeUndefined(); + expect(taskExecutorMocks.failTaskRunByRunId).not.toHaveBeenCalled(); + }); + + it("persists interrupted recovery before task projection and rejects late provider or yield", async () => { + const entry = createRunEntry(); + const persistOrThrow = vi.fn(); + const controller = createLifecycleController({ entry, persistOrThrow }); + await controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + const recovered = structuredClone(entry); + + await controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_001, + outcome: { status: "ok" }, + reason: SUBAGENT_ENDED_REASON_COMPLETE, + triggerCleanup: false, + }); + + expect(markSubagentRunPausedAfterYield({ entry, endedAt: 4_002 })).toBe(false); + expect(entry).toEqual(recovered); + expect(entry).toMatchObject({ + endedAt: 4_000, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + terminalOwner: "interrupted-recovery", + outcome: { status: "error", error: "restart interrupted run" }, + completion: { resultText: null, capturedAt: 4_000 }, + }); + expect(persistOrThrow).toHaveBeenCalledOnce(); + expect(persistOrThrow.mock.invocationCallOrder[0]).toBeLessThan( + taskExecutorMocks.failTaskRunByRunId.mock.invocationCallOrder[0]!, + ); + }); + + it("rolls interrupted recovery back when registry persistence fails", async () => { + const entry = createRunEntry(); + const original = structuredClone(entry); + const controller = createLifecycleController({ + entry, + persistOrThrow: vi.fn(() => { + throw new Error("registry store boom"); + }), + }); + + await expect( + controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }), + ).rejects.toThrow("registry store boom"); + + expect(entry).toEqual(original); + expect(taskExecutorMocks.failTaskRunByRunId).not.toHaveBeenCalled(); + }); + + it.each([ + ["provisional", { killReconciliation: { killedAt: 4_000 } }], + ["stable", {}], + ])("keeps %s killed state unchanged during interrupted recovery", async (_name, extra) => { + const entry = createRunEntry({ + endedAt: 4_000, + endedReason: SUBAGENT_ENDED_REASON_KILLED, + outcome: { status: "error", error: "agent run aborted" }, + suppressAnnounceReason: "killed", + ...extra, + }); + const original = structuredClone(entry); + const persistOrThrow = vi.fn(); + await createLifecycleController({ entry, persistOrThrow }).completeSubagentRun({ + runId: entry.runId, + endedAt: 4_001, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + + expect(entry).toEqual(original); + expect(persistOrThrow).not.toHaveBeenCalled(); + expect(taskExecutorMocks.failTaskRunByRunId).not.toHaveBeenCalled(); + expect(taskExecutorMocks.completeTaskRunByRunId).not.toHaveBeenCalled(); + }); + + it("does not overwrite partial terminal evidence during interrupted recovery", async () => { + const terminalEvidence: Array> = [ + { endedAt: 4_000 }, + { outcome: { status: "error", error: "existing failure" } }, + { endedReason: SUBAGENT_ENDED_REASON_ERROR }, + { execution: { status: "terminal", endedAt: 4_000 } }, + { + endedAt: 4_000, + outcome: { status: "error", error: "existing failure" }, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + }, + ]; + for (const evidence of terminalEvidence) { + const entry = createRunEntry(evidence); + const original = structuredClone(entry); + const persistOrThrow = vi.fn(); + await createLifecycleController({ entry, persistOrThrow }).completeSubagentRun({ + runId: entry.runId, + endedAt: 4_001, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + + expect(entry).toEqual(original); + expect(persistOrThrow).not.toHaveBeenCalled(); + } + expect(taskExecutorMocks.failTaskRunByRunId).not.toHaveBeenCalled(); + expect(taskExecutorMocks.completeTaskRunByRunId).not.toHaveBeenCalled(); + }); + + it("drains exact interrupted terminal evidence after restart admission reopens", async () => { + const interruptedOutcome = { + status: "error" as const, + error: "restart interrupted run", + startedAt: 2_000, + endedAt: 4_000, + elapsedMs: 2_000, + }; + const entry = createRunEntry({ + endedAt: 4_000, + outcome: interruptedOutcome, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + execution: { + status: "terminal", + startedAt: 2_000, + endedAt: 4_000, + outcome: interruptedOutcome, + }, + }); + const persistOrThrow = vi.fn(); + await createLifecycleController({ entry, persistOrThrow }).completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + + expect(entry).toMatchObject({ + endedAt: 4_000, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + terminalOwner: "interrupted-recovery", + outcome: { status: "error", error: "restart interrupted run" }, + execution: { status: "terminal", endedAt: 4_000 }, + }); + expect(persistOrThrow).toHaveBeenCalled(); + }); + + it("preserves conflicting nested terminal evidence during interrupted recovery", async () => { + const interruptedOutcome = { + status: "error" as const, + error: "restart interrupted run", + startedAt: 2_000, + endedAt: 4_000, + elapsedMs: 2_000, + }; + const entry = createRunEntry({ + endedAt: 4_000, + outcome: interruptedOutcome, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + execution: { + status: "terminal", + startedAt: 2_000, + endedAt: 3_999, + outcome: { status: "error", error: "provider failure" }, + }, + }); + const original = structuredClone(entry); + const persistOrThrow = vi.fn(); + await createLifecycleController({ entry, persistOrThrow }).completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + }); + + expect(entry).toEqual(original); + expect(persistOrThrow).not.toHaveBeenCalled(); + expect(taskExecutorMocks.failTaskRunByRunId).not.toHaveBeenCalled(); + expect(taskExecutorMocks.completeTaskRunByRunId).not.toHaveBeenCalled(); + }); + it("restores a provisional kill when canonical task projection fails", async () => { const entry = createRunEntry({ endedAt: 4_000, diff --git a/src/agents/subagent-registry-lifecycle.ts b/src/agents/subagent-registry-lifecycle.ts index ee2db88606dc..957164db5c16 100644 --- a/src/agents/subagent-registry-lifecycle.ts +++ b/src/agents/subagent-registry-lifecycle.ts @@ -42,6 +42,7 @@ import { } from "./subagent-delivery-state.js"; import { SUBAGENT_ENDED_REASON_COMPLETE, + SUBAGENT_ENDED_REASON_ERROR, SUBAGENT_ENDED_REASON_KILLED, type SubagentLifecycleEndedReason, } from "./subagent-lifecycle-events.js"; @@ -1443,6 +1444,7 @@ export function createSubagentRegistryLifecycleController(params: { startedAt?: number; suppressSessionEffects?: boolean; completionSnapshot?: { resultText: string | null; capturedAt: number }; + recoverInterrupted?: true; }; const completeSubagentRunAttempt = async (completeParams: CompleteSubagentRunParams) => { @@ -1474,6 +1476,95 @@ export function createSubagentRegistryLifecycleController(params: { } Object.assign(target, snapshot); }; + const recoveryRequested = completeParams.recoverInterrupted === true; + if (!recoveryRequested && entry.terminalOwner === "interrupted-recovery") { + // Restart recovery already persisted the terminal winner for this exact + // run. Late provider/lifecycle callbacks cannot reopen that decision. + return; + } + if (recoveryRequested) { + const ownsInterruptedRecovery = entry.terminalOwner === "interrupted-recovery"; + // Mismatched partial terminal evidence is an existing winner and must + // not be overwritten. Exact normalized evidence may be the same recovery + // request deferred by restart admission, so drain it. + const hasTerminalEvidence = + typeof entry.endedAt === "number" || + entry.outcome !== undefined || + entry.endedReason !== undefined || + entry.execution?.status === "terminal"; + const expectedElapsedMs = + typeof currentEntry.startedAt === "number" && typeof completeParams.endedAt === "number" + ? Math.max(0, completeParams.endedAt - currentEntry.startedAt) + : undefined; + const outcomeMatchesInterruptedRecovery = (outcome: SubagentRunOutcome | undefined) => + completeParams.outcome.status === "error" && + outcome?.status === "error" && + outcome.error === completeParams.outcome.error && + (outcome.startedAt === undefined || outcome.startedAt === currentEntry.startedAt) && + (outcome.endedAt === undefined || outcome.endedAt === completeParams.endedAt) && + (outcome.elapsedMs === undefined || outcome.elapsedMs === expectedElapsedMs); + const executionMatchesInterruptedRecovery = + entry.execution?.status !== "terminal" || + (entry.execution.endedAt === completeParams.endedAt && + (entry.execution.startedAt === undefined || + entry.execution.startedAt === currentEntry.startedAt) && + outcomeMatchesInterruptedRecovery(entry.execution.outcome)); + const matchesRequestedInterruptedTerminal = + typeof completeParams.endedAt === "number" && + entry.endedAt === completeParams.endedAt && + outcomeMatchesInterruptedRecovery(entry.outcome) && + entry.endedReason === SUBAGENT_ENDED_REASON_ERROR && + executionMatchesInterruptedRecovery; + if ( + !ownsInterruptedRecovery && + (entry.killReconciliation !== undefined || + entry.endedReason === SUBAGENT_ENDED_REASON_KILLED || + entry.pauseReason === "sessions_yield" || + typeof entry.cleanupCompletedAt === "number" || + (hasTerminalEvidence && !matchesRequestedInterruptedTerminal)) + ) { + return; + } + if (!ownsInterruptedRecovery) { + const endedAt = + typeof completeParams.endedAt === "number" ? completeParams.endedAt : Date.now(); + const outcome = withSubagentOutcomeTiming( + { status: "error", error: completeParams.outcome.error }, + { startedAt: entry.startedAt, endedAt }, + ); + entry.endedAt = endedAt; + entry.outcome = outcome; + entry.endedReason = SUBAGENT_ENDED_REASON_ERROR; + entry.pauseReason = undefined; + entry.execution = { + ...entry.execution, + status: "terminal", + startedAt: entry.startedAt, + endedAt, + outcome, + interruptedAt: undefined, + interruptionReason: undefined, + }; + entry.completion = { + ...ensureCompletionState(entry), + resultText: null, + capturedAt: endedAt, + }; + entry.cleanupHandled = false; + entry.terminalOwner = "interrupted-recovery"; + mutated = true; + try { + params.persistOrThrow(); + } catch (error) { + restoreEntrySnapshot(entrySnapshot); + throw error; + } + // Any later delivery-payload write rolls back to this durable owner, + // never to the pre-recovery running row. + entrySnapshot = structuredClone(entry); + mutated = false; + } + } sessionSuperseded = newerGenerationOwnsSession(currentEntry); if ( completeParams.reason === SUBAGENT_ENDED_REASON_KILLED && @@ -1494,12 +1585,14 @@ export function createSubagentRegistryLifecycleController(params: { ) { return; } - const shouldDrainExistingTerminal = isOlderEquivalentTerminalCallback({ - entry, - endedAt: requestedEndedAt, - outcome: completeParams.outcome, - reason: completeParams.reason, - }); + const shouldDrainExistingTerminal = + recoveryRequested || + isOlderEquivalentTerminalCallback({ + entry, + endedAt: requestedEndedAt, + outcome: completeParams.outcome, + reason: completeParams.reason, + }); if (shouldDrainExistingTerminal) { // Preserve the newer canonical timing while allowing this duplicate // caller to rescue a stalled cleanup and delivery tail. @@ -1515,11 +1608,13 @@ export function createSubagentRegistryLifecycleController(params: { Number.isFinite(completeParams.startedAt) ? completeParams.startedAt : undefined; - const expiredDeadlineMs = resolveExpiredExplicitRunDeadlineMs({ - entry, - nextEndedAt: endedAt, - observedStartedAt, - }); + const expiredDeadlineMs = recoveryRequested + ? undefined + : resolveExpiredExplicitRunDeadlineMs({ + entry, + nextEndedAt: endedAt, + observedStartedAt, + }); if (expiredDeadlineMs !== undefined) { endedAt = expiredDeadlineMs; completionOutcome = { status: "timeout" }; @@ -1619,10 +1714,13 @@ export function createSubagentRegistryLifecycleController(params: { }; mutated = true; } - const outcome = withSubagentOutcomeTiming(completionOutcome, { - startedAt: entry.startedAt, - endedAt, - }); + const outcome = + recoveryRequested && entry.outcome + ? entry.outcome + : withSubagentOutcomeTiming(completionOutcome, { + startedAt: entry.startedAt, + endedAt, + }); if (shouldUpdateRunOutcome(entry.outcome, outcome)) { entry.outcome = outcome; mutated = true; @@ -1664,7 +1762,7 @@ export function createSubagentRegistryLifecycleController(params: { // A newer generation may share the session key. Its transcript/reply is // not evidence for this older run, so reconcile only the terminal task state. - if (sessionSuperseded) { + if (recoveryRequested || sessionSuperseded) { const completion = ensureCompletionState(entry); if (completion.resultText === undefined) { completion.resultText = null; diff --git a/src/agents/subagent-registry-run-manager.ts b/src/agents/subagent-registry-run-manager.ts index 8481495abb9b..f6cff688718e 100644 --- a/src/agents/subagent-registry-run-manager.ts +++ b/src/agents/subagent-registry-run-manager.ts @@ -107,6 +107,7 @@ export function markSubagentRunPausedAfterYield(params: { }): boolean { const { entry } = params; if ( + entry.terminalOwner === "interrupted-recovery" || entry.endedReason === SUBAGENT_ENDED_REASON_KILLED || entry.suppressAnnounceReason === "killed" || (entry.cleanup === "delete" && Number.isFinite(entry.deleteCleanupDispatchedAt)) @@ -674,6 +675,7 @@ export function createSubagentRunManager(params: { cleanupCompletedAt: undefined, cleanupHandled: false, suppressAnnounceReason: undefined, + terminalOwner: undefined, killReconciliation: undefined, suppressCompletionDelivery: undefined, delivery: { diff --git a/src/agents/subagent-registry-steer-runtime.ts b/src/agents/subagent-registry-steer-runtime.ts index d97e1ce021f8..839b522aaad0 100644 --- a/src/agents/subagent-registry-steer-runtime.ts +++ b/src/agents/subagent-registry-steer-runtime.ts @@ -25,8 +25,7 @@ type ReplaceSubagentRunAfterSteerParams = { type ReplaceSubagentRunAfterSteerFn = (params: ReplaceSubagentRunAfterSteerParams) => boolean; type FinalizeInterruptedSubagentRunParams = { - runId?: string; - childSessionKey?: string; + runId: string; error: string; endedAt?: number; }; @@ -52,7 +51,7 @@ export function replaceSubagentRunAfterSteer(params: ReplaceSubagentRunAfterStee return replaceSubagentRunAfterSteerImpl?.(params) ?? false; } -/** Finalizes interrupted runs through the installed registry hook. */ +/** Finalizes one interrupted run generation through the installed registry hook. */ export async function finalizeInterruptedSubagentRun(params: FinalizeInterruptedSubagentRunParams) { return (await finalizeInterruptedSubagentRunImpl?.(params)) ?? 0; } diff --git a/src/agents/subagent-registry.persistence.test.ts b/src/agents/subagent-registry.persistence.test.ts index 739f5b0e669a..66db24bdd95b 100644 --- a/src/agents/subagent-registry.persistence.test.ts +++ b/src/agents/subagent-registry.persistence.test.ts @@ -13,6 +13,7 @@ import { import { callGateway } from "../gateway/call.js"; import { onAgentEvent } from "../infra/agent-events.js"; import { captureEnv, deleteTestEnvValue, setTestEnvValue, withEnv } from "../test-utils/env.js"; +import { scheduleOrphanRecovery } from "./subagent-orphan-recovery.js"; import { persistSubagentSessionTiming } from "./subagent-registry-helpers.js"; import { getSubagentRunsSnapshotForRead } from "./subagent-registry-state.js"; import { @@ -218,6 +219,7 @@ describe("subagent registry persistence", () => { startedAt: 111, endedAt: 222, }); + vi.mocked(scheduleOrphanRecovery).mockReset(); vi.mocked(onAgentEvent).mockReset(); vi.mocked(onAgentEvent).mockReturnValue(() => undefined); }); @@ -817,6 +819,43 @@ describe("subagent registry persistence", () => { ]); }); + it("preserves restored interrupted-recovery owners for orphan replay", async () => { + const now = Date.now(); + const runId = "run-interrupted-recovery-restore"; + await writePersistedRegistry( + { + version: 2, + runs: { + [runId]: { + runId, + childSessionKey: "agent:main:subagent:interrupted-recovery-restore", + requesterSessionKey: "agent:main:main", + requesterDisplayKey: "main", + task: "replay interrupted terminal", + cleanup: "keep", + createdAt: now - 100, + startedAt: now - 50, + endedAt: now, + endedReason: "subagent-error", + outcome: { status: "error", error: "restart interrupted run" }, + terminalOwner: "interrupted-recovery", + completion: { required: false, resultText: null, capturedAt: now }, + }, + }, + }, + { seedChildSessions: false }, + ); + + restartRegistry(); + await waitForRegistryWork(() => vi.mocked(scheduleOrphanRecovery).mock.calls.length > 0); + + expect(callGateway).not.toHaveBeenCalled(); + expect(scheduleOrphanRecovery).toHaveBeenCalledOnce(); + expect(listSubagentRunsForRequester("agent:main:main")).toEqual([ + expect.objectContaining({ runId, terminalOwner: "interrupted-recovery" }), + ]); + }); + it("reconciles stale unended restored runs that are not restart-recoverable", async () => { const now = Date.now(); const runId = "run-stale-unended-restore"; @@ -850,7 +889,7 @@ describe("subagent registry persistence", () => { expect(listSubagentRunsForRequester("agent:main:main")).toHaveLength(0); }); - it("keeps stale unended restored runs with abortedLastRun for restart recovery", async () => { + it("keeps stale unended restored runs with abortedLastRun for lifecycle recovery", async () => { vi.mocked(callGateway).mockImplementationOnce(async (request) => { expectFields(request, { method: "agent.wait", @@ -891,12 +930,17 @@ describe("subagent registry persistence", () => { }); restartRegistry(); - await waitForRegistryWork(() => vi.mocked(callGateway).mock.calls.length > 0); + await waitForRegistryWork( + () => + vi.mocked(callGateway).mock.calls.length > 0 && + vi.mocked(scheduleOrphanRecovery).mock.calls.length > 0, + ); expect(callGateway).toHaveBeenCalledTimes(1); const [request] = vi.mocked(callGateway).mock.calls.at(0) ?? []; expectFields(request, { method: "agent.wait" }); expectFields((request as { params?: unknown } | undefined)?.params, { runId }); + expect(scheduleOrphanRecovery).toHaveBeenCalledOnce(); expect( listSubagentRunsForRequester("agent:main:main").some((entry) => entry.runId === runId), ).toBe(true); diff --git a/src/agents/subagent-registry.store.sqlite.test.ts b/src/agents/subagent-registry.store.sqlite.test.ts index 1589d2e7cbe3..d5bb1f06c058 100644 --- a/src/agents/subagent-registry.store.sqlite.test.ts +++ b/src/agents/subagent-registry.store.sqlite.test.ts @@ -79,7 +79,12 @@ describe("subagent registry sqlite store", () => { it("persists subagent runs in the shared sqlite state database", async () => { await withTempStateEnv(async () => { - const run = createRun(); + const run = createRun({ + endedReason: "subagent-error", + outcome: { status: "error", error: "restart interrupted run", endedAt: 250 }, + terminalOwner: "interrupted-recovery", + completion: { required: true, resultText: null, capturedAt: 250 }, + }); saveSubagentRegistryToSqlite(new Map([[run.runId, run]])); @@ -91,6 +96,7 @@ describe("subagent registry sqlite store", () => { task: run.task, endedAt: run.endedAt, outcome: run.outcome, + terminalOwner: "interrupted-recovery", completion: run.completion, delivery: run.delivery, }); diff --git a/src/agents/subagent-registry.test.ts b/src/agents/subagent-registry.test.ts index ff1d84ef11b1..7187edf25a40 100644 --- a/src/agents/subagent-registry.test.ts +++ b/src/agents/subagent-registry.test.ts @@ -5102,8 +5102,9 @@ describe("subagent registry seam flow", () => { createdAt: endedAt - 30_000, startedAt: endedAt - 20_000, endedAt, - endedReason: "subagent-complete", - outcome: { status: "ok" }, + endedReason: "subagent-error", + outcome: { status: "error", error: "restart interrupted run" }, + terminalOwner: "interrupted-recovery", delivery: { status: "suspended", createdAt: endedAt + 1_000, @@ -5142,6 +5143,7 @@ describe("subagent registry seam flow", () => { cleanupHandled: false, }); expect(replacement?.endedAt).toBeUndefined(); + expect(replacement?.terminalOwner).toBeUndefined(); expect(replacement?.delivery?.lastError).toBeUndefined(); expect(replacement?.delivery?.payload).toBeUndefined(); expect(replacement?.delivery?.suspendedAt).toBeUndefined(); @@ -5587,9 +5589,134 @@ describe("subagent registry seam flow", () => { endedAt: 2, elapsedMs: 1, }); + expect(run?.terminalOwner).toBe("interrupted-recovery"); expect(run?.cleanupCompletedAt).toBeTypeOf("number"); + + const announceCalls = mocks.runSubagentAnnounceFlow.mock.calls.length; + await expect( + mod.finalizeInterruptedSubagentRun({ + runId: "run-interrupted", + error: + "Subagent run was interrupted by a gateway restart or connection loss. Automatic recovery failed after 2 attempts. Please retry.", + endedAt: 2, + }), + ).resolves.toBe(1); + expect(run?.terminalOwner).toBe("interrupted-recovery"); + expect(run?.outcome?.error).toContain("Automatic recovery failed after 2 attempts"); + expect(mocks.runSubagentAnnounceFlow).toHaveBeenCalledTimes(announceCalls); }); + it("returns zero without mutating the run or task when recovery persistence fails", async () => { + resetTaskRegistryForTests({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + try { + const runId = "run-interrupted-persist-failure"; + const childSessionKey = "agent:main:subagent:interrupted-persist-failure"; + createRunningTaskRun({ + runtime: "subagent", + sourceId: runId, + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey, + runId, + task: "preserve interrupted task", + deliveryStatus: "pending", + startedAt: 1, + lastEventAt: 1, + }); + const entry = { + runId, + childSessionKey, + requesterSessionKey: "agent:main:main", + requesterDisplayKey: "main", + task: "preserve interrupted task", + cleanup: "keep" as const, + createdAt: 1, + startedAt: 1, + }; + mod.addSubagentRunForTests(entry); + const original = structuredClone(entry); + mocks.persistSubagentRunsToDiskOrThrow.mockImplementationOnce(() => { + throw new Error("registry store boom"); + }); + + await expect( + mod.finalizeInterruptedSubagentRun({ + runId, + error: "restart interrupted run", + endedAt: 2, + }), + ).resolves.toBe(0); + + expect(mocks.persistSubagentRunsToDiskOrThrow).toHaveBeenCalledOnce(); + expect( + mod.listSubagentRunsForRequester("agent:main:main").find((run) => run.runId === runId), + ).toEqual(original); + expect(findTaskByRunIdForStatus(runId)).toMatchObject({ status: "running" }); + } finally { + resetTaskRegistryForTests({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + } + }); + + const completeTerminalOutcome = { + status: "error" as const, + error: "existing failure", + startedAt: 1, + endedAt: 2, + elapsedMs: 1, + }; + const completeTerminalEvidence = { + endedAt: 2, + outcome: completeTerminalOutcome, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + execution: { + status: "terminal" as const, + startedAt: 1, + endedAt: 2, + outcome: completeTerminalOutcome, + }, + }; + it.each([ + ["missing-ended-at", 0, { ...completeTerminalEvidence, endedAt: undefined }, undefined], + ["missing-outcome", 0, { ...completeTerminalEvidence, outcome: undefined }, undefined], + ["missing-ended-reason", 0, { ...completeTerminalEvidence, endedReason: undefined }, undefined], + ["missing-execution", 0, { ...completeTerminalEvidence, execution: undefined }, undefined], + ["cleanup-complete", 1, completeTerminalEvidence, 2], + ["cleanup-partial", 0, { ...completeTerminalEvidence, execution: undefined }, 2], + ])( + "%s terminal evidence returns %i", + async (scenario, expected, evidence, cleanupCompletedAt) => { + const runId = `run-interrupted-${scenario}`; + const entry = { + runId, + childSessionKey: `agent:main:subagent:interrupted-${scenario}`, + requesterSessionKey: "agent:main:main", + requesterDisplayKey: "main", + task: "preserve existing terminal evidence", + cleanup: "keep" as const, + createdAt: 1, + startedAt: 1, + cleanupCompletedAt, + ...evidence, + }; + mod.addSubagentRunForTests(entry); + const original = structuredClone(entry); + + await expect( + mod.finalizeInterruptedSubagentRun({ + runId, + error: "restart interrupted run", + endedAt: 3, + }), + ).resolves.toBe(expected); + + expect( + mod.listSubagentRunsForRequester("agent:main:main").find((run) => run.runId === runId), + ).toEqual(original); + }, + ); + it("removes attachments for released delete-mode runs", async () => { const attachmentsRootDir = await fs.mkdtemp( path.join(os.tmpdir(), "openclaw-release-attachments-"), diff --git a/src/agents/subagent-registry.ts b/src/agents/subagent-registry.ts index 2974919e7529..2703fe4e7aab 100644 --- a/src/agents/subagent-registry.ts +++ b/src/agents/subagent-registry.ts @@ -460,6 +460,7 @@ type CompleteSubagentRunParams = { triggerCleanup: boolean; startedAt?: number; suppressSessionEffects?: boolean; + recoverInterrupted?: true; }; async function completeSubagentRunWithRecoveryAttempt( @@ -836,6 +837,12 @@ function resumeSubagentRun(runId: string) { if (!entry) { return; } + if (entry.terminalOwner === "interrupted-recovery") { + // Startup orphan recovery replays this durable exact-run winner before it + // reads session/config state. Do not prune or resume it through announce. + resumedRuns.add(runId); + return; + } if (entry.cleanupCompletedAt) { return; } @@ -1738,6 +1745,7 @@ export function resetSubagentRegistryForTests(opts?: { persist?: boolean }) { stopSweeper(); sweepInProgress = false; restoreAttempted = false; + lastOrphanRecoveryScheduleAt = 0; if (listenerStop) { listenerStop(); listenerStop = null; @@ -1773,25 +1781,24 @@ export function releaseSubagentRun(runId: string) { subagentRunManager.releaseSubagentRun(runId); } +function hasCompleteSubagentTerminalState(entry: SubagentRunRecord | undefined): boolean { + return ( + entry !== undefined && + typeof entry.endedAt === "number" && + Number.isFinite(entry.endedAt) && + entry.outcome !== undefined && + entry.endedReason !== undefined && + entry.execution?.status === "terminal" + ); +} + export async function finalizeInterruptedSubagentRun(params: { - runId?: string; - childSessionKey?: string; + runId: string; error: string; endedAt?: number; }): Promise { - const runIds = new Set(); - if (typeof params.runId === "string" && params.runId.trim()) { - runIds.add(params.runId.trim()); - } - if (typeof params.childSessionKey === "string" && params.childSessionKey.trim()) { - const childSessionKey = params.childSessionKey.trim(); - for (const [runId, entry] of subagentRuns.entries()) { - if (entry.childSessionKey === childSessionKey) { - runIds.add(runId); - } - } - } - if (runIds.size === 0) { + const runId = params.runId.trim(); + if (!runId) { return 0; } @@ -1799,32 +1806,55 @@ export async function finalizeInterruptedSubagentRun(params: { typeof params.endedAt === "number" && Number.isFinite(params.endedAt) ? params.endedAt : Date.now(); - let updated = 0; - for (const runId of runIds) { - clearPendingLifecycleError(runId); - clearPendingLifecycleTimeout(runId); - const entry = subagentRuns.get(runId); - if (!entry || typeof entry.cleanupCompletedAt === "number") { - continue; - } - await completeSubagentRunWithRecovery( - { - runId, - endedAt, - outcome: { - status: "error", - error: params.error, - }, - reason: SUBAGENT_ENDED_REASON_ERROR, - sendFarewell: true, - accountId: entry.requesterOrigin?.accountId, - triggerCleanup: true, - }, - "explicit-failed-mark", - ); - updated += 1; + clearPendingLifecycleError(runId); + clearPendingLifecycleTimeout(runId); + const entry = subagentRuns.get(runId); + if (!entry) { + return 0; + } + if ( + typeof entry.cleanupCompletedAt === "number" && + entry.terminalOwner !== "interrupted-recovery" + ) { + return hasCompleteSubagentTerminalState(entry) ? 1 : 0; + } + const completionParams: CompleteSubagentRunParams = { + runId, + endedAt, + outcome: { + status: "error", + error: params.error, + }, + reason: SUBAGENT_ENDED_REASON_ERROR, + sendFarewell: true, + accountId: entry.requesterOrigin?.accountId, + triggerCleanup: true, + recoverInterrupted: true, + }; + try { + await completeSubagentRun(completionParams); + // A successfully finalized stale generation can be retired once a newer + // generation owns the session; the captured exact row still has its result. + const finalized = subagentRuns.get(runId) ?? entry; + // Recovery preserves partial terminal evidence instead of overwriting it. + // Keep scheduler retries alive until the exact row is fully terminal. + return hasCompleteSubagentTerminalState(finalized) ? 1 : 0; + } catch (error) { + if (isGatewayRestartDraining() && subagentRuns.get(runId) === entry) { + log.warn("subagent completion deferred during gateway restart", { + source: "explicit-failed-mark", + runId, + }); + scheduleSubagentCompletionRetryAfterRestart(completionParams, "explicit-failed-mark", entry); + return 1; + } + log.warn("failed to durably finalize interrupted subagent run", { + runId, + childSessionKey: entry.childSessionKey, + error, + }); + return 0; } - return updated; } export function resolveRequesterForChildSession(childSessionKey: string): { diff --git a/src/agents/subagent-registry.types.ts b/src/agents/subagent-registry.types.ts index 47ae4e5f1907..1028e084dde5 100644 --- a/src/agents/subagent-registry.types.ts +++ b/src/agents/subagent-registry.types.ts @@ -123,6 +123,8 @@ export type SubagentRunRecord = { cleanupCompletedAt?: number; cleanupHandled?: boolean; suppressAnnounceReason?: "steer-restart" | "killed"; + /** Sticky owner while restart recovery replays this exact terminal run. */ + terminalOwner?: "interrupted-recovery"; /** Present only while a current-version killed run awaits bounded reconciliation. */ killReconciliation?: SubagentKillReconciliationState; /** Durable requester-stop policy until silent completion cleanup finishes. */