diff --git a/src/cli/gateway-cli/lifecycle.runtime.ts b/src/cli/gateway-cli/lifecycle.runtime.ts index 0411f481abd5..e0e5a400dd54 100644 --- a/src/cli/gateway-cli/lifecycle.runtime.ts +++ b/src/cli/gateway-cli/lifecycle.runtime.ts @@ -47,5 +47,5 @@ export { waitForActiveTasks, } from "../../process/command-queue.js"; export { getInspectableActiveTaskRestartBlockers } from "../../tasks/task-registry.maintenance.js"; -export { reloadTaskRegistryFromStore } from "../../tasks/runtime-internal.js"; +export { reloadTaskRuntimeStateFromStore } from "../../tasks/runtime-internal.js"; export { abortPendingChannelReloads } from "../../gateway/server-reload-handlers.js"; diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index 67795b7a4614..18fcb9e2fdd1 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -67,7 +67,7 @@ const waitForActiveCronJobs = vi.fn(async (_timeoutMs?: number) => ({ drained: true, active: 0, })); -const reloadTaskRegistryFromStore = vi.fn(); +const reloadTaskRuntimeStateFromStore = vi.fn(); const rotateAgentEventLifecycleGeneration = vi.fn(); const clearRuntimeConfigSnapshot = vi.fn(); const restartGatewayProcessWithFreshPid = vi.fn< @@ -180,7 +180,7 @@ vi.mock("../../tasks/cron-task-cancel.js", () => ({ })); vi.mock("../../tasks/runtime-internal.js", () => ({ - reloadTaskRegistryFromStore: () => reloadTaskRegistryFromStore(), + reloadTaskRuntimeStateFromStore: () => reloadTaskRuntimeStateFromStore(), })); vi.mock("../../infra/agent-events.js", () => ({ @@ -947,7 +947,10 @@ describe("runGatewayLoop", () => { expect(clearRuntimeConfigSnapshot).toHaveBeenCalledTimes(1); expect(resetGatewayRestartStateForInProcessRestart).toHaveBeenCalledTimes(1); expect(rotateAgentEventLifecycleGeneration).toHaveBeenCalledTimes(1); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(1); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(1); + expect(reloadTaskRuntimeStateFromStore.mock.invocationCallOrder[0] ?? Infinity).toBeLessThan( + start.mock.invocationCallOrder[1] ?? Infinity, + ); expect( rotateAgentEventLifecycleGeneration.mock.invocationCallOrder[0] ?? Infinity, ).toBeLessThan(resetAllLanes.mock.invocationCallOrder[0] ?? Infinity); @@ -983,7 +986,7 @@ describe("runGatewayLoop", () => { expect(clearRuntimeConfigSnapshot).toHaveBeenCalledTimes(2); expect(resetGatewayRestartStateForInProcessRestart).toHaveBeenCalledTimes(2); expect(rotateAgentEventLifecycleGeneration).toHaveBeenCalledTimes(2); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(2); expect(acquireGatewayLock).toHaveBeenCalledTimes(3); sigterm(); @@ -1095,7 +1098,7 @@ describe("runGatewayLoop", () => { expect(markGatewayDraining).toHaveBeenCalledTimes(1); expect(resetAllLanes).toHaveBeenCalledTimes(1); expect(resetGatewayRestartStateForInProcessRestart).toHaveBeenCalledTimes(1); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(1); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(1); } finally { sigterm(); await expect(exited).resolves.toBe(0); @@ -1284,7 +1287,7 @@ describe("runGatewayLoop", () => { expect(markGatewayDraining).toHaveBeenCalledTimes(2); expect(resetAllLanes).toHaveBeenCalledTimes(2); expect(resetGatewayRestartStateForInProcessRestart).toHaveBeenCalledTimes(2); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(2); expect(acquireGatewayLock).toHaveBeenCalledTimes(3); expect(gatewayLog.error).toHaveBeenCalledWith( expect.stringContaining("gateway startup failed: restart startup failed."), @@ -1356,7 +1359,7 @@ describe("runGatewayLoop", () => { expect(markGatewayDraining).toHaveBeenCalledTimes(2); expect(resetAllLanes).toHaveBeenCalledTimes(2); expect(resetGatewayRestartStateForInProcessRestart).toHaveBeenCalledTimes(2); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(2); expect(acquireGatewayLock).toHaveBeenCalledTimes(3); } finally { sigterm(); @@ -1365,12 +1368,16 @@ describe("runGatewayLoop", () => { }); }); - it("keeps the process alive and retries after task-registry restore fails during restart", async () => { + it("keeps the process alive and retries after task runtime state restores fail", async () => { vi.clearAllMocks(); - reloadTaskRegistryFromStore.mockReset(); - reloadTaskRegistryFromStore.mockImplementationOnce(() => { - throw new Error("task registry restore failed"); - }); + reloadTaskRuntimeStateFromStore.mockReset(); + reloadTaskRuntimeStateFromStore + .mockImplementationOnce(() => { + throw new Error("task-flow registry restore failed"); + }) + .mockImplementationOnce(() => { + throw new Error("task registry restore failed"); + }); peekGatewaySigusr1RestartReason.mockReturnValue(undefined); respawnGatewayProcessForUpdate.mockReturnValue({ mode: "disabled", @@ -1407,6 +1414,22 @@ describe("runGatewayLoop", () => { const sigterm = captureSignal("SIGTERM"); try { + sigusr1(); + await waitForLoopCondition( + () => + gatewayLog.error.mock.calls.some(([message]) => + String(message).includes( + "gateway startup failed: task-flow registry restore failed.", + ), + ), + "expected failed task-flow registry restore to be logged", + ); + + expectRestartCloseCall(closeFirst, 90_000); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(1); + expect(start).toHaveBeenCalledTimes(1); + expect(runtime.exit).not.toHaveBeenCalled(); + sigusr1(); await waitForLoopCondition( () => @@ -1416,15 +1439,14 @@ describe("runGatewayLoop", () => { "expected failed task-registry restore to be logged", ); - expectRestartCloseCall(closeFirst, 90_000); - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(1); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(2); expect(start).toHaveBeenCalledTimes(1); expect(runtime.exit).not.toHaveBeenCalled(); sigusr1(); await startedSecond; - expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2); + expect(reloadTaskRuntimeStateFromStore).toHaveBeenCalledTimes(3); expect(start).toHaveBeenCalledTimes(2); expect(runtime.exit).not.toHaveBeenCalled(); } finally { @@ -1438,7 +1460,7 @@ describe("runGatewayLoop", () => { }); }); } finally { - reloadTaskRegistryFromStore.mockReset(); + reloadTaskRuntimeStateFromStore.mockReset(); } }); diff --git a/src/cli/gateway-cli/run-loop.ts b/src/cli/gateway-cli/run-loop.ts index e534839ebeab..dce977128339 100644 --- a/src/cli/gateway-cli/run-loop.ts +++ b/src/cli/gateway-cli/run-loop.ts @@ -885,7 +885,7 @@ export async function runGatewayLoop(params: { const { abortActiveCronTaskRuns, advanceCronActiveJobGeneration, - reloadTaskRegistryFromStore, + reloadTaskRuntimeStateFromStore, retireActiveCronTaskRunTracking, resetCronActiveJobs, resetAllLanes, @@ -910,7 +910,7 @@ export async function runGatewayLoop(params: { resetAllLanes(); clearRuntimeConfigSnapshot(); resetGatewayRestartStateForInProcessRestart(); - reloadTaskRegistryFromStore(); + reloadTaskRuntimeStateFromStore(); markGatewayRestartTrace("restart.next-start"); }); diff --git a/src/commands/tasks-json.test.ts b/src/commands/tasks-json.test.ts index 8c68d016375a..780a3c1ba631 100644 --- a/src/commands/tasks-json.test.ts +++ b/src/commands/tasks-json.test.ts @@ -5,6 +5,7 @@ import { createManagedTaskFlow as createManagedTaskFlowOrNull, resetTaskFlowRegistryForTests, } from "../tasks/task-flow-registry.js"; +import { configureTaskFlowRegistryRuntime } from "../tasks/task-flow-registry.store.js"; import type { TaskFlowRecord } from "../tasks/task-flow-registry.types.js"; import { createTaskRecord as createTaskRecordOrNull, @@ -239,4 +240,44 @@ describe("tasks JSON commands", () => { }); }); }); + + it("keeps task-flow restore failures inspectable in audit JSON", async () => { + await withTaskJsonStateDir(async () => { + const loadSnapshot = vi.fn(() => { + throw new Error("SQLITE_IOERR: task-flow audit restore failed"); + }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + }, + }); + const runtime = createRuntime(); + + await tasksAuditJsonCommand({ json: true }, runtime); + + expect(readJsonLog(runtime)).toMatchObject({ + count: 1, + summary: { + taskFlows: { + total: 1, + errors: 1, + byCode: { + restore_failed: 1, + }, + }, + }, + findings: [ + { + kind: "task_flow", + severity: "error", + code: "restore_failed", + detail: + "task-flow registry restore failed: SQLITE_IOERR: task-flow audit restore failed", + }, + ], + }); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + }); + }); }); diff --git a/src/commands/tasks-json.ts b/src/commands/tasks-json.ts index 1dc10190cae6..99846e534e5b 100644 --- a/src/commands/tasks-json.ts +++ b/src/commands/tasks-json.ts @@ -6,7 +6,6 @@ import type { RuntimeEnv } from "../runtime.js"; import { writeRuntimeJson } from "../runtime.js"; import { listTaskRecords } from "../tasks/runtime-internal.js"; import { listTaskFlowAuditFindings } from "../tasks/task-flow-registry.audit.js"; -import { listTaskFlowRecords } from "../tasks/task-flow-runtime-internal.js"; import { listTaskAuditFindings } from "../tasks/task-registry.audit.js"; import type { TaskRecord } from "../tasks/task-registry.types.js"; import { @@ -40,9 +39,8 @@ function toSystemAuditFindings(params: { codeFilter?: TaskSystemAuditCode; }) { const tasks = listTaskJsonRecords(); - const flows = listTaskFlowRecords(); const taskFindings = listTaskAuditFindings({ tasks }); - const flowFindings = listTaskFlowAuditFindings({ flows }); + const flowFindings = listTaskFlowAuditFindings(); const result = buildTaskSystemAuditFindings({ taskFindings, flowFindings, diff --git a/src/commands/tasks.test.ts b/src/commands/tasks.test.ts index 3ffb31b5abd7..f4e2192797ac 100644 --- a/src/commands/tasks.test.ts +++ b/src/commands/tasks.test.ts @@ -12,9 +12,12 @@ import { createManagedTaskFlow as createManagedTaskFlowOrNull, resetTaskFlowRegistryForTests, } from "../tasks/task-flow-registry.js"; +import { configureTaskFlowRegistryRuntime } from "../tasks/task-flow-registry.store.js"; import type { TaskFlowRecord } from "../tasks/task-flow-registry.types.js"; import { createTaskRecord as createTaskRecordOrNull, + getTaskById, + reloadTaskRegistryFromStore, resetTaskRegistryDeliveryRuntimeForTests, resetTaskRegistryForTests, } from "../tasks/task-registry.js"; @@ -212,6 +215,55 @@ describe("tasks commands", () => { }); }); + it("keeps task-flow restore failures inspectable in full audit output", async () => { + await withTaskCommandStateDir(async () => { + const loadSnapshot = vi.fn(() => { + throw new Error("SQLITE_IOERR: task-flow command audit restore failed"); + }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + }, + }); + + const jsonRuntime = createRuntime(); + await tasksAuditCommand({ json: true }, jsonRuntime); + expect(readFirstJsonLog(jsonRuntime)).toMatchObject({ + count: 1, + summary: { + taskFlows: { + total: 1, + errors: 1, + byCode: { + restore_failed: 1, + }, + }, + }, + findings: [ + { + kind: "task_flow", + severity: "error", + code: "restore_failed", + detail: + "task-flow registry restore failed: SQLITE_IOERR: task-flow command audit restore failed", + }, + ], + }); + + const textRuntime = createRuntime(); + await tasksAuditCommand({ json: false }, textRuntime); + const output = vi + .mocked(textRuntime.log) + .mock.calls.map(([line]) => String(line)) + .join("\n"); + expect(output).toContain("TaskFlow"); + expect(output).toContain("restore_failed"); + expect(output).toContain("task-flow registry restore failed"); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + }); + }); + it("reports blank list filters as absent in command JSON output", async () => { await withTaskCommandStateDir(async () => { const task = createTaskRecord({ @@ -716,6 +768,63 @@ describe("tasks commands", () => { }); }); + it.each([false, true])( + "refuses all maintenance when task-flow restore fails (apply=%s)", + async (apply) => { + await withTaskCommandStateDir(async (state) => { + const now = Date.now(); + vi.useFakeTimers(); + vi.setSystemTime(now - 8 * 24 * 60 * 60_000); + const staleTask = createTaskRecord({ + runtime: "cli", + ownerKey: "agent:main:main", + scopeKind: "session", + runId: `stale-task-${String(apply)}`, + task: "Task that maintenance would prune", + status: "succeeded", + deliveryStatus: "not_applicable", + }); + vi.setSystemTime(now); + const storePath = path.join(state.sessionsDir("main"), "sessions.json"); + const staleSessionKey = "agent:main:cron:done-job:run:old-run"; + await writeSessionEntries(storePath, { + [staleSessionKey]: { + sessionId: "old-run", + updatedAt: Date.now() - 8 * 24 * 60 * 60_000, + }, + }); + const loadSnapshot = vi.fn(() => { + throw new Error("SQLITE_CORRUPT: task-flow maintenance restore failed"); + }); + const saveSnapshot = vi.fn(); + const upsertFlow = vi.fn(); + const deleteFlow = vi.fn(); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot, + upsertFlow, + deleteFlow, + }, + }); + const runtime = createRuntime(); + + await expect(tasksMaintenanceCommand({ json: true, apply }, runtime)).rejects.toThrow( + "Task-flow registry restore failed: SQLITE_CORRUPT: task-flow maintenance restore failed. Refusing task maintenance.", + ); + + expect(loadSnapshot).toHaveBeenCalledTimes(1); + expect(saveSnapshot).not.toHaveBeenCalled(); + expect(upsertFlow).not.toHaveBeenCalled(); + expect(deleteFlow).not.toHaveBeenCalled(); + expect(runtime.log).not.toHaveBeenCalled(); + expect(loadSessionEntry({ sessionKey: staleSessionKey, storePath })).toBeDefined(); + reloadTaskRegistryFromStore(); + expect(getTaskById(staleTask.taskId)?.taskId).toBe(staleTask.taskId); + }); + }, + ); + it("applies a conservative session registry sweep for stale cron run sessions", async () => { await withTaskCommandStateDir(async (state) => { const now = Date.now(); diff --git a/src/commands/tasks.ts b/src/commands/tasks.ts index 569bf250368b..62e5b8b3b45f 100644 --- a/src/commands/tasks.ts +++ b/src/commands/tasks.ts @@ -19,6 +19,7 @@ import { getTaskById, updateTaskNotifyPolicyById } from "../tasks/runtime-intern import { cancelDetachedTaskRunById } from "../tasks/task-executor.js"; import { listTaskFlowAuditFindings } from "../tasks/task-flow-registry.audit.js"; import { + assertTaskFlowRegistryMaintenanceReady, getInspectableTaskFlowAuditSummary, previewTaskFlowRegistryMaintenance, runTaskFlowRegistryMaintenance, @@ -587,6 +588,7 @@ export async function tasksMaintenanceCommand( runtime: RuntimeEnv, ) { configureTaskMaintenanceFromConfig(); + assertTaskFlowRegistryMaintenanceReady(); const auditBefore = getInspectableTaskAuditSummary(); const flowAuditBefore = getInspectableTaskFlowAuditSummary(); const taskMaintenance = opts.apply diff --git a/src/gateway/server-methods/tasks.ts b/src/gateway/server-methods/tasks.ts index ac47a553cb9b..dec7c8f9f318 100644 --- a/src/gateway/server-methods/tasks.ts +++ b/src/gateway/server-methods/tasks.ts @@ -12,8 +12,8 @@ import { validateTasksListParams, } from "../../../packages/gateway-protocol/src/index.js"; import { parseAgentSessionKey } from "../../routing/session-key.js"; -import { cancelDetachedTaskRunById } from "../../tasks/detached-task-runtime.js"; import { getTaskById, listTaskRecordsUnsorted } from "../../tasks/runtime-internal.js"; +import { cancelDetachedTaskRunById } from "../../tasks/task-executor.js"; import type { TaskRecord, TaskStatus } from "../../tasks/task-registry.types.js"; import { mapTaskSummary, taskUpdatedAt } from "./task-summary.js"; import type { GatewayRequestHandlers } from "./types.js"; diff --git a/src/gateway/server-startup-early.test.ts b/src/gateway/server-startup-early.test.ts index d9e1db32d127..69bc78addc2d 100644 --- a/src/gateway/server-startup-early.test.ts +++ b/src/gateway/server-startup-early.test.ts @@ -15,6 +15,7 @@ const mocks = vi.hoisted(() => ({ registerSkillsChangeListener: vi.fn(), skillsChangeUnsub: vi.fn(), ensureContextWindowCacheLoaded: vi.fn(), + ensureTaskRuntimeStateReady: vi.fn(), configureTaskRegistryMaintenance: vi.fn(), startTaskRegistryMaintenance: vi.fn(), getInspectableActiveTaskRestartBlockers: vi.fn(), @@ -42,6 +43,10 @@ vi.mock("../agents/context.js", () => ({ ensureContextWindowCacheLoaded: mocks.ensureContextWindowCacheLoaded, })); +vi.mock("../tasks/runtime-internal.js", () => ({ + ensureTaskRuntimeStateReady: mocks.ensureTaskRuntimeStateReady, +})); + vi.mock("../tasks/task-registry.maintenance.js", () => ({ configureTaskRegistryMaintenance: mocks.configureTaskRegistryMaintenance, startTaskRegistryMaintenance: mocks.startTaskRegistryMaintenance, @@ -97,6 +102,7 @@ describe("startGatewayEarlyRuntime", () => { mocks.skillsChangeUnsub.mockReset(); mocks.ensureContextWindowCacheLoaded.mockReset(); mocks.ensureContextWindowCacheLoaded.mockResolvedValue(undefined); + mocks.ensureTaskRuntimeStateReady.mockReset(); mocks.configureTaskRegistryMaintenance.mockReset(); mocks.startTaskRegistryMaintenance.mockReset(); mocks.getInspectableActiveTaskRestartBlockers.mockReset(); @@ -124,8 +130,15 @@ describe("startGatewayEarlyRuntime", () => { await Promise.resolve(); expect(mocks.ensureContextWindowCacheLoaded).toHaveBeenCalledWith({}); expect(mocks.primeRemoteSkillsCache).toHaveBeenCalledTimes(1); + expect(mocks.ensureTaskRuntimeStateReady).toHaveBeenCalledTimes(1); expect(mocks.configureTaskRegistryMaintenance).toHaveBeenCalledTimes(1); expect(mocks.startTaskRegistryMaintenance).toHaveBeenCalledTimes(1); + expect(mocks.ensureTaskRuntimeStateReady.mock.invocationCallOrder[0] ?? Infinity).toBeLessThan( + mocks.startGatewayDiscovery.mock.invocationCallOrder[0] ?? Infinity, + ); + expect(mocks.startGatewayDiscovery.mock.invocationCallOrder[0] ?? Infinity).toBeLessThan( + mocks.startTaskRegistryMaintenance.mock.invocationCallOrder[0] ?? Infinity, + ); expect(mocks.registerSkillsChangeListener).toHaveBeenCalledTimes(1); expect(earlyRuntime.getActiveTaskCount()).toBe(1); @@ -149,6 +162,24 @@ describe("startGatewayEarlyRuntime", () => { expect(earlyRuntime).toHaveProperty("startMaintenance"); }); + it("fails before discovery and task maintenance when task state cannot restore", async () => { + mocks.ensureTaskRuntimeStateReady.mockImplementationOnce(() => { + throw new Error("task-flow registry restore failed"); + }); + + await expect( + startGatewayEarlyRuntime( + earlyRuntimeInput({ + minimalTestGateway: false, + }), + ), + ).rejects.toThrow("task-flow registry restore failed"); + + expect(mocks.startGatewayDiscovery).not.toHaveBeenCalled(); + expect(mocks.configureTaskRegistryMaintenance).not.toHaveBeenCalled(); + expect(mocks.startTaskRegistryMaintenance).not.toHaveBeenCalled(); + }); + it("starts discovery with the current plugin registry services", async () => { const stop = vi.fn(async () => {}); mocks.startGatewayDiscovery.mockResolvedValueOnce({ bonjourStop: stop } as never); diff --git a/src/gateway/server-startup-early.ts b/src/gateway/server-startup-early.ts index 18082c0ab5e9..ecf4effe9386 100644 --- a/src/gateway/server-startup-early.ts +++ b/src/gateway/server-startup-early.ts @@ -111,6 +111,12 @@ export async function startGatewayEarlyRuntime(params: { getRuntimeConfig: () => OpenClawConfig; startupTrace?: GatewayStartupTrace; }) { + if (!params.minimalTestGateway) { + await measureStartup(params.startupTrace, "runtime.early.task-state", async () => { + const { ensureTaskRuntimeStateReady } = await import("../tasks/runtime-internal.js"); + ensureTaskRuntimeStateReady(); + }); + } const bonjourStop = await measureStartup(params.startupTrace, "runtime.early.discovery", () => startGatewayPluginDiscovery(params), ); diff --git a/src/tasks/runtime-internal.ts b/src/tasks/runtime-internal.ts index d65b6b84f9ab..53f9124fc0bb 100644 --- a/src/tasks/runtime-internal.ts +++ b/src/tasks/runtime-internal.ts @@ -1,5 +1,25 @@ // Internal task registry facade used by runtime modules without exposing public SDK surface. +import { + ensureTaskFlowRegistryReady, + reloadTaskFlowRegistryFromStore, +} from "./task-flow-runtime-internal.js"; +import { + ensureTaskRegistryReady as ensureTaskRegistryReadyInternal, + reloadTaskRegistryFromStore as reloadTaskRegistryFromStoreInternal, +} from "./task-registry.js"; + +export function ensureTaskRuntimeStateReady(): void { + ensureTaskFlowRegistryReady(); + ensureTaskRegistryReadyInternal(); +} + +export function reloadTaskRuntimeStateFromStore(): void { + reloadTaskFlowRegistryFromStore(); + reloadTaskRegistryFromStoreInternal(); +} + export { + assertTaskCancellationReadyById, cancelTaskById, createTaskRecord, deleteTaskRecordById, diff --git a/src/tasks/task-executor.test.ts b/src/tasks/task-executor.test.ts index edeea01631a1..6ad5c02a0598 100644 --- a/src/tasks/task-executor.test.ts +++ b/src/tasks/task-executor.test.ts @@ -31,12 +31,14 @@ import { listTaskFlowRecords, resetTaskFlowRegistryForTests, } from "./task-flow-registry.js"; +import { configureTaskFlowRegistryRuntime } from "./task-flow-registry.store.js"; import type { TaskFlowRecord } from "./task-flow-registry.types.js"; import { setTaskRegistryDeliveryRuntimeForTests, getTaskById, findLatestTaskForFlowId, findTaskByRunId, + markTaskTerminalById, resetTaskRegistryControlRuntimeForTests, resetTaskRegistryDeliveryRuntimeForTests, resetTaskRegistryForTests, @@ -363,6 +365,39 @@ describe("task-executor", () => { }); }); + it("keeps detached tasks standalone when task-flow restore fails", async () => { + await withTaskExecutorStateDir(async () => { + const loadSnapshot = vi.fn(() => { + throw new Error("SQLITE_IOERR: task-flow restore failed"); + }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + }, + }); + + const created = createRunningTaskRun({ + runtime: "subagent", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey: "agent:codex:subagent:standalone", + runId: "run-executor-flow-restore-failed", + task: "Continue without a one-task flow", + startedAt: 10, + deliveryStatus: "pending", + }); + + expect(created.parentFlowId).toBeUndefined(); + expect(getTaskById(created.taskId)).toMatchObject({ + taskId: created.taskId, + status: "running", + parentFlowId: undefined, + }); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + }); + }); + it("promotes a provisional kill in an already-cancelled one-task flow", async () => { await withTaskExecutorStateDir(async () => { const child = createRunningTaskRun({ @@ -743,6 +778,75 @@ describe("task-executor", () => { }); }); + it("dispatches cancellation for tasks owned only by the registered runtime", async () => { + await withTaskExecutorStateDir(async () => { + const cancelDetachedTaskRunByIdSpy = vi.fn(async () => ({ + found: true, + cancelled: true, + })); + setDetachedTaskLifecycleRuntime({ + ...getDetachedTaskLifecycleRuntime(), + cancelDetachedTaskRunById: cancelDetachedTaskRunByIdSpy, + }); + + const cancelled = await cancelDetachedTaskRunById({ + cfg: {} as never, + taskId: "runtime-owned-task", + reason: "operator request", + }); + + expect(cancelDetachedTaskRunByIdSpy).toHaveBeenCalledWith({ + cfg: {} as never, + taskId: "runtime-owned-task", + reason: "operator request", + }); + expect(cancelled).toEqual({ + found: true, + cancelled: true, + }); + }); + }); + + it("checks linked flow readiness before invoking a registered cancellation runtime", async () => { + await withTaskExecutorStateDir(async () => { + const child = createRunningAcpChildTaskRun({ + runId: "run-external-cancel-restore-failed", + }); + expect(child.parentFlowId).toBeTruthy(); + const cancelDetachedTaskRunByIdSpy = spyOnRuntimeCancel(); + + resetTaskFlowRegistryForTests({ persist: false }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot: () => { + throw new Error("SQLITE_IOERR: cancellation flow restore failed"); + }, + saveSnapshot: () => {}, + }, + }); + + const cancelled = await cancelDetachedTaskRunById({ + cfg: {} as never, + taskId: child.taskId, + }); + + expect(cancelDetachedTaskRunByIdSpy).not.toHaveBeenCalled(); + expect(hoisted.cancelSessionMock).not.toHaveBeenCalled(); + expect(cancelled).toMatchObject({ + found: true, + cancelled: false, + reason: expect.stringContaining( + "Task-flow registry restore failed: SQLITE_IOERR: cancellation flow restore failed", + ), + task: { + taskId: child.taskId, + status: "running", + }, + }); + expect(getTaskById(child.taskId)?.status).toBe("running"); + }); + }); + it("falls back to the legacy canceller when the registered runtime declines task ownership", async () => { await withTaskExecutorStateDir(async () => { hoisted.cancelSessionMock.mockResolvedValue(undefined); @@ -813,6 +917,102 @@ describe("task-executor", () => { }); }); + it("dispatches provisional terminal projections to their registered runtime", async () => { + await withTaskExecutorStateDir(async () => { + const child = createRunningTaskRun({ + runtime: "subagent", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey: "agent:codex:subagent:provisional", + runId: "run-provisional-runtime-owned", + task: "Cancel provisional runtime task", + startedAt: 10, + deliveryStatus: "not_applicable", + }); + markTaskTerminalById({ + taskId: child.taskId, + status: "cancelled", + endedAt: 20, + error: SUBAGENT_KILL_TASK_ERROR, + }); + const cancelDetachedTaskRunByIdSpy = vi.fn(async () => ({ + found: true, + cancelled: true, + })); + setDetachedTaskLifecycleRuntime({ + ...getDetachedTaskLifecycleRuntime(), + cancelDetachedTaskRunById: cancelDetachedTaskRunByIdSpy, + }); + + const cancelled = await cancelDetachedTaskRunById({ + cfg: {} as never, + taskId: child.taskId, + }); + + expect(cancelDetachedTaskRunByIdSpy).toHaveBeenCalledWith({ + cfg: {} as never, + taskId: child.taskId, + }); + expect(cancelled).toEqual({ + found: true, + cancelled: true, + }); + }); + }); + + it("checks linked flow readiness before dispatching provisional terminal projections", async () => { + await withTaskExecutorStateDir(async () => { + const child = createRunningTaskRun({ + runtime: "subagent", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey: "agent:codex:subagent:provisional-restore-failed", + runId: "run-provisional-runtime-restore-failed", + task: "Gate provisional runtime cancellation", + startedAt: 10, + deliveryStatus: "pending", + }); + expect(child.parentFlowId).toBeTruthy(); + markTaskTerminalById({ + taskId: child.taskId, + status: "cancelled", + endedAt: 20, + error: SUBAGENT_KILL_TASK_ERROR, + }); + const cancelDetachedTaskRunByIdSpy = vi.fn(async () => ({ + found: true, + cancelled: true, + })); + setDetachedTaskLifecycleRuntime({ + ...getDetachedTaskLifecycleRuntime(), + cancelDetachedTaskRunById: cancelDetachedTaskRunByIdSpy, + }); + resetTaskFlowRegistryForTests({ persist: false }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot: () => { + throw new Error("SQLITE_IOERR: provisional cancellation restore failed"); + }, + saveSnapshot: () => {}, + }, + }); + + const cancelled = await cancelDetachedTaskRunById({ + cfg: {} as never, + taskId: child.taskId, + }); + + expect(cancelDetachedTaskRunByIdSpy).not.toHaveBeenCalled(); + expect(cancelled).toMatchObject({ + found: true, + cancelled: false, + reason: expect.stringContaining( + "Task-flow registry restore failed: SQLITE_IOERR: provisional cancellation restore failed", + ), + }); + }); + }); + it("cancels active subagent child tasks", async () => { await withTaskExecutorStateDir(async () => { hoisted.killSubagentRunAdminMock.mockResolvedValue({ diff --git a/src/tasks/task-executor.ts b/src/tasks/task-executor.ts index 4846cbe3ec48..19e5c3fcb526 100644 --- a/src/tasks/task-executor.ts +++ b/src/tasks/task-executor.ts @@ -1,5 +1,6 @@ // Executes task records through configured runtimes and updates registry state. import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { formatErrorMessage } from "../infra/errors.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import type { DetachedRunningTaskCreateParams, @@ -8,6 +9,7 @@ import type { } from "./detached-task-runtime-contract.js"; import { getRegisteredDetachedTaskLifecycleRuntime } from "./detached-task-runtime-state.js"; import { + assertTaskCancellationReadyById, cancelTaskById, createTaskRecord, getTaskById, @@ -595,12 +597,32 @@ export async function cancelFlowByIdForOwner(params: { }); } -export async function cancelDetachedTaskRunById(params: { cfg: OpenClawConfig; taskId: string }) { +export async function cancelDetachedTaskRunById(params: { + cfg: OpenClawConfig; + taskId: string; + reason?: string; +}) { const task = getTaskById(params.taskId); + const registeredRuntime = getRegisteredDetachedTaskLifecycleRuntime(); if (!task) { + if (registeredRuntime) { + const cancelled = await registeredRuntime.cancelDetachedTaskRunById(params); + if (cancelled.found) { + return cancelled; + } + } return cancelTaskById(params); } - const registeredRuntime = getRegisteredDetachedTaskLifecycleRuntime(); + try { + assertTaskCancellationReadyById(task.taskId); + } catch (error) { + return { + found: true, + cancelled: false, + reason: formatErrorMessage(error), + task, + }; + } if (registeredRuntime) { const cancelled = await registeredRuntime.cancelDetachedTaskRunById(params); if (cancelled.found) { diff --git a/src/tasks/task-flow-registry.audit.test.ts b/src/tasks/task-flow-registry.audit.test.ts index 90a3aa21fd1c..6dc430fca8c6 100644 --- a/src/tasks/task-flow-registry.audit.test.ts +++ b/src/tasks/task-flow-registry.audit.test.ts @@ -1,5 +1,5 @@ // Covers managed task-flow audit summaries and stale-flow classification. -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { captureEnv } from "../test-utils/env.js"; import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js"; @@ -93,20 +93,24 @@ describe("task-flow-registry audit", () => { }); it("surfaces restore failures as task-flow audit findings", () => { + const loadSnapshot = vi.fn(() => { + throw new Error("boom"); + }); configureTaskFlowRegistryRuntime({ store: { - loadSnapshot: () => { - throw new Error("boom"); - }, + loadSnapshot, saveSnapshot: () => {}, }, }); - const findings = listTaskFlowAuditFindings(); - expect(findings).toHaveLength(1); - expect(findings[0]?.severity).toBe("error"); - expect(findings[0]?.code).toBe("restore_failed"); - expect(findings[0]?.detail).toContain("boom"); + for (let attempt = 0; attempt < 2; attempt += 1) { + const findings = listTaskFlowAuditFindings(); + expect(findings).toHaveLength(1); + expect(findings[0]?.severity).toBe("error"); + expect(findings[0]?.code).toBe("restore_failed"); + expect(findings[0]?.detail).toContain("boom"); + } + expect(loadSnapshot).toHaveBeenCalledTimes(1); }); it("clears restore-failed findings after a clean reset and restore", () => { diff --git a/src/tasks/task-flow-registry.audit.ts b/src/tasks/task-flow-registry.audit.ts index 67165be7a825..593ca5f25a02 100644 --- a/src/tasks/task-flow-registry.audit.ts +++ b/src/tasks/task-flow-registry.audit.ts @@ -142,7 +142,8 @@ function createEmptyTaskFlowAuditSummary(): TaskFlowAuditSummary { export function listTaskFlowAuditFindings( options: TaskFlowAuditOptions = {}, ): TaskFlowAuditFinding[] { - const flows = options.flows ?? listTaskFlowRecords(); + const restoreFailure = getTaskFlowRegistryRestoreFailure(); + const flows = options.flows ?? (restoreFailure ? [] : listTaskFlowRecords()); const now = options.now ?? Date.now(); const staleRunningMs = options.staleRunningMs ?? DEFAULT_STALE_RUNNING_MS; const staleWaitingMs = options.staleWaitingMs ?? DEFAULT_STALE_WAITING_MS; @@ -150,7 +151,6 @@ export function listTaskFlowAuditFindings( const cancelStuckMs = options.cancelStuckMs ?? DEFAULT_CANCEL_STUCK_MS; const findings: TaskFlowAuditFinding[] = []; - const restoreFailure = getTaskFlowRegistryRestoreFailure(); if (restoreFailure) { findings.push( createFinding({ diff --git a/src/tasks/task-flow-registry.maintenance.ts b/src/tasks/task-flow-registry.maintenance.ts index 2c3552b84760..4d905f35b737 100644 --- a/src/tasks/task-flow-registry.maintenance.ts +++ b/src/tasks/task-flow-registry.maintenance.ts @@ -9,6 +9,7 @@ import { import { deleteTaskFlowRecordById, getTaskFlowById, + getTaskFlowRegistryRestoreFailure, listTaskFlowRecords, updateFlowRecordByIdExpectedRevision, } from "./task-flow-registry.js"; @@ -22,6 +23,15 @@ type TaskFlowRegistryMaintenanceSummary = { pruned: number; }; +export function assertTaskFlowRegistryMaintenanceReady(): void { + const restoreFailure = getTaskFlowRegistryRestoreFailure(); + if (restoreFailure) { + throw new Error( + `Task-flow registry restore failed: ${restoreFailure}. Refusing task maintenance.`, + ); + } +} + function isTerminalFlow(flow: TaskFlowRecord): boolean { return ( flow.status === "succeeded" || diff --git a/src/tasks/task-flow-registry.test.ts b/src/tasks/task-flow-registry.test.ts index 026a476f092a..ef406758d62a 100644 --- a/src/tasks/task-flow-registry.test.ts +++ b/src/tasks/task-flow-registry.test.ts @@ -6,10 +6,12 @@ import { createTaskFlowForTask as createTaskFlowForTaskOrNull, createManagedTaskFlow as createManagedTaskFlowOrNull, deleteTaskFlowRecordById, + getTaskFlowRegistryRestoreFailure, failFlow, getTaskFlowById, listTaskFlowRecords, requestFlowCancel, + reloadTaskFlowRegistryFromStore, resetTaskFlowRegistryForTests, resumeFlow, setFlowWaiting, @@ -239,6 +241,78 @@ describe("task-flow-registry", () => { expect(events[2]?.flowId).toBe(created.flowId); }); + it("keeps restore failures sticky until an explicit reload succeeds", () => { + const hiddenFlow: TaskFlowRecord = { + flowId: "hidden-flow", + syncMode: "managed", + ownerKey: "agent:main:main", + controllerId: "tests/hidden-flow", + revision: 4, + status: "running", + notifyPolicy: "done_only", + goal: "Existing durable flow", + createdAt: 10, + updatedAt: 20, + }; + const loadSnapshot = vi + .fn() + .mockImplementationOnce(() => { + throw new Error("SQLITE_CORRUPT: task-flow restore failed"); + }) + .mockReturnValue({ + flows: new Map([[hiddenFlow.flowId, hiddenFlow]]), + }); + const upsertFlow = vi.fn(); + const deleteFlow = vi.fn(); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + upsertFlow, + deleteFlow, + }, + }); + + expect(() => listTaskFlowRecords()).toThrow( + "Task-flow registry restore failed: SQLITE_CORRUPT: task-flow restore failed", + ); + expect(getTaskFlowRegistryRestoreFailure()).toBe("SQLITE_CORRUPT: task-flow restore failed"); + expect(() => + createManagedTaskFlowOrNull({ + ownerKey: "agent:main:main", + controllerId: "tests/restore-failure", + goal: "Must not persist over hidden flows", + }), + ).toThrow("Task-flow registry restore failed: SQLITE_CORRUPT: task-flow restore failed"); + expect(() => + updateFlowRecordByIdExpectedRevision({ + flowId: hiddenFlow.flowId, + expectedRevision: hiddenFlow.revision, + patch: { currentStep: "must not overwrite hidden state" }, + }), + ).toThrow("Task-flow registry restore failed: SQLITE_CORRUPT: task-flow restore failed"); + expect(() => deleteTaskFlowRecordById(hiddenFlow.flowId)).toThrow( + "Task-flow registry restore failed: SQLITE_CORRUPT: task-flow restore failed", + ); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + expect(upsertFlow).not.toHaveBeenCalled(); + expect(deleteFlow).not.toHaveBeenCalled(); + + reloadTaskFlowRegistryFromStore(); + + expect(loadSnapshot).toHaveBeenCalledTimes(2); + expect(getTaskFlowRegistryRestoreFailure()).toBeNull(); + expect(listTaskFlowRecords()).toEqual([expect.objectContaining(hiddenFlow)]); + expect( + createManagedTaskFlowOrNull({ + ownerKey: "agent:main:main", + controllerId: "tests/restore-recovery", + goal: "Create after explicit recovery", + }), + ).not.toBeNull(); + expect(upsertFlow).toHaveBeenCalledTimes(1); + }); + it("does not throw or register memory when flow create persistence fails", () => { const upsertFlow = vi.fn((_flow: TaskFlowRecord) => { throw new Error("SQLITE_FULL: database or disk is full"); diff --git a/src/tasks/task-flow-registry.ts b/src/tasks/task-flow-registry.ts index 7333df1e272f..158b006350f6 100644 --- a/src/tasks/task-flow-registry.ts +++ b/src/tasks/task-flow-registry.ts @@ -18,9 +18,13 @@ import type { import type { TaskNotifyPolicy, TaskRecord } from "./task-registry.types.js"; const log = createSubsystemLogger("tasks/task-flow-registry"); -const flows = new Map(); -let restoreAttempted = false; -let restoreFailureMessage: string | null = null; +let flows = new Map(); +type TaskFlowRegistryRestoreState = + | { status: "uninitialized" } + | { status: "restoring" } + | { status: "ready" } + | { status: "failed"; error: Error; message: string }; +let taskFlowRegistryRestoreState: TaskFlowRegistryRestoreState = { status: "uninitialized" }; type FlowRecordPatch = Omit< Partial< @@ -244,23 +248,42 @@ function resolveTaskMirroredFlowTiming( return { updatedAt: endedAt, endedAt }; } -function ensureFlowRegistryReady() { - if (restoreAttempted) { - return; +function restoreTaskFlowRegistryOnce(): void { + switch (taskFlowRegistryRestoreState.status) { + case "ready": + return; + case "failed": + throw taskFlowRegistryRestoreState.error; + case "restoring": + throw new Error("Task-flow registry restore is already in progress."); + case "uninitialized": + break; } - restoreAttempted = true; + taskFlowRegistryRestoreState = { status: "restoring" }; try { const restored = getTaskFlowRegistryStore().loadSnapshot(); - flows.clear(); + const restoredFlows = new Map(); for (const [flowId, flow] of restored.flows) { - flows.set(flowId, normalizeRestoredFlowRecord(flow)); + restoredFlows.set(flowId, normalizeRestoredFlowRecord(flow)); } - restoreFailureMessage = null; + flows = restoredFlows; + taskFlowRegistryRestoreState = { status: "ready" }; } catch (error) { - flows.clear(); - restoreFailureMessage = formatErrorMessage(error); - log.warn("Failed to restore task-flow registry", { error }); - return; + flows = new Map(); + const message = formatErrorMessage(error); + const restoreError = new Error(`Task-flow registry restore failed: ${message}`, { + cause: error, + }); + taskFlowRegistryRestoreState = { + status: "failed", + error: restoreError, + message, + }; + log.warn("Failed to restore task-flow registry", { + error: message, + consoleMessage: `Failed to restore task-flow registry: ${message}`, + }); + throw restoreError; } emitFlowRegistryObserverEvent(() => ({ kind: "restored", @@ -268,9 +291,25 @@ function ensureFlowRegistryReady() { })); } +export function ensureTaskFlowRegistryReady(): void { + restoreTaskFlowRegistryOnce(); +} + export function getTaskFlowRegistryRestoreFailure(): string | null { - ensureFlowRegistryReady(); - return restoreFailureMessage; + try { + ensureTaskFlowRegistryReady(); + return null; + } catch { + return taskFlowRegistryRestoreState.status === "failed" + ? taskFlowRegistryRestoreState.message + : "Task-flow registry restore did not complete."; + } +} + +export function reloadTaskFlowRegistryFromStore(): void { + flows = new Map(); + taskFlowRegistryRestoreState = { status: "uninitialized" }; + ensureTaskFlowRegistryReady(); } function createFlowSnapshotWith(next?: TaskFlowRecord, deletedFlowId?: string) { @@ -430,7 +469,7 @@ function writeFlowRecord(next: TaskFlowRecord, previous?: TaskFlowRecord): TaskF } export function createFlowRecord(params: CreateFlowRecordParams): TaskFlowRecord | null { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const record = buildFlowRecord(params); return writeFlowRecord(record); } @@ -491,7 +530,7 @@ function updateFlowRecordByIdUnchecked( flowId: string, patch: FlowRecordPatch, ): TaskFlowRecord | null { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const current = flows.get(flowId); if (!current) { return null; @@ -504,7 +543,7 @@ export function updateFlowRecordByIdExpectedRevision(params: { expectedRevision: number; patch: FlowRecordPatch; }): TaskFlowUpdateResult { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const current = flows.get(params.flowId); if (!current) { return { @@ -725,13 +764,13 @@ export function syncFlowFromTask( } export function getTaskFlowById(flowId: string): TaskFlowRecord | undefined { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const flow = flows.get(flowId); return flow ? cloneFlowRecord(flow) : undefined; } export function listTaskFlowsForOwnerKey(ownerKey: string): TaskFlowRecord[] { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const normalizedOwnerKey = ownerKey.trim(); if (!normalizedOwnerKey) { return []; @@ -756,14 +795,14 @@ export function resolveTaskFlowForLookupToken(token: string): TaskFlowRecord | u } export function listTaskFlowRecords(): TaskFlowRecord[] { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); return [...flows.values()] .map((flow) => cloneFlowRecord(flow)) .toSorted((left, right) => right.createdAt - left.createdAt); } export function deleteTaskFlowRecordById(flowId: string): boolean { - ensureFlowRegistryReady(); + ensureTaskFlowRegistryReady(); const current = flows.get(flowId); if (!current) { return false; @@ -781,9 +820,8 @@ export function deleteTaskFlowRecordById(flowId: string): boolean { } export function resetTaskFlowRegistryForTests(opts?: { persist?: boolean }) { - flows.clear(); - restoreAttempted = false; - restoreFailureMessage = null; + flows = new Map(); + taskFlowRegistryRestoreState = { status: "uninitialized" }; resetTaskFlowRegistryRuntimeForTests(); if (opts?.persist !== false) { persistFlowRegistry(); diff --git a/src/tasks/task-flow-runtime-internal.ts b/src/tasks/task-flow-runtime-internal.ts index 24671b49045e..b8118f80024b 100644 --- a/src/tasks/task-flow-runtime-internal.ts +++ b/src/tasks/task-flow-runtime-internal.ts @@ -3,11 +3,13 @@ export { createTaskFlowForTask, createManagedTaskFlow, deleteTaskFlowRecordById, + ensureTaskFlowRegistryReady, failFlow, finishFlow, getTaskFlowById, listTaskFlowRecords, requestFlowCancel, + reloadTaskFlowRegistryFromStore, resolveTaskFlowForLookupToken, resetTaskFlowRegistryForTests, resumeFlow, diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index e07c38800e5f..0bd749f73ebd 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -26,6 +26,7 @@ import { withTempDir } from "../test-helpers/temp-dir.js"; import { withEnvAsync } from "../test-utils/env.js"; import { registerActiveCronTaskRun, resetActiveCronTaskRunsForTests } from "./cron-task-cancel.js"; import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js"; +import { ensureTaskRuntimeStateReady } from "./runtime-internal.js"; import { createTaskFlowForTask as createTaskFlowForTaskOrNull, createManagedTaskFlow as createManagedTaskFlowOrNull, @@ -38,6 +39,7 @@ import type { TaskFlowRecord } from "./task-flow-registry.types.js"; import { cancelTaskById, createTaskRecord as createTaskRecordOrNull, + deleteTaskRecordById, finalizeTaskRunByRunId, findLatestTaskForRelatedSessionKey, findTaskByRunId, @@ -1223,6 +1225,152 @@ describe("task-registry", () => { }); }); + it("does not persist linked task changes while task-flow restore is failed", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + const taskStore = createInMemoryTaskRegistryStore(); + const taskUpsert = vi.spyOn(taskStore, "upsertTaskWithDeliveryState"); + const taskDelete = vi.spyOn(taskStore, "deleteTaskWithDeliveryState"); + const deliveryUpsert = vi.spyOn(taskStore, "upsertDeliveryState"); + configureTaskRegistryRuntime({ store: taskStore }); + configureTaskFlowRegistryRuntime({ + store: createInMemoryTaskFlowRegistryStore(), + }); + + const task = createTaskRecord({ + runtime: "acp", + ownerKey: "agent:main:main", + scopeKind: "session", + runId: "flow-restore-failed-task", + task: "Preserve linked task state", + status: "running", + }); + const flow = createTaskFlowForTask({ task }); + expect( + linkTaskToFlowById({ + taskId: task.taskId, + flowId: flow.flowId, + })?.parentFlowId, + ).toBe(flow.flowId); + taskUpsert.mockClear(); + deliveryUpsert.mockClear(); + + resetTaskFlowRegistryForTests({ persist: false }); + const loadSnapshot = vi.fn(() => { + throw new Error("SQLITE_IOERR: task-flow restore failed"); + }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + }, + }); + + expect(() => + markTaskTerminalById({ + taskId: task.taskId, + status: "succeeded", + endedAt: 200, + }), + ).toThrow("Task-flow registry restore failed: SQLITE_IOERR: task-flow restore failed"); + expect(taskUpsert).not.toHaveBeenCalled(); + expect(requireTaskById(task.taskId).status).toBe("running"); + + expect(() => deleteTaskRecordById(task.taskId)).toThrow( + "Task-flow registry restore failed: SQLITE_IOERR: task-flow restore failed", + ); + expect(taskDelete).not.toHaveBeenCalled(); + expect(requireTaskById(task.taskId).taskId).toBe(task.taskId); + + expect(() => + createTaskRecord({ + runtime: "acp", + ownerKey: "agent:main:main", + scopeKind: "session", + requesterOrigin: { + channel: "notifychat", + to: "notifychat:123", + }, + runId: task.runId, + task: task.task, + status: "running", + }), + ).toThrow("Task-flow registry restore failed: SQLITE_IOERR: task-flow restore failed"); + expect(deliveryUpsert).not.toHaveBeenCalled(); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + + const standalone = createTaskRecord({ + runtime: "cli", + ownerKey: "agent:main:main", + scopeKind: "session", + runId: "standalone-during-flow-restore-failure", + task: "Keep standalone task state available", + status: "running", + deliveryStatus: "not_applicable", + }); + expect( + markTaskTerminalById({ + taskId: standalone.taskId, + status: "succeeded", + endedAt: 300, + })?.status, + ).toBe("succeeded"); + }); + }); + + it("restores task-flow state before activating the task registry", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + const loadTaskSnapshot = vi.fn(() => ({ + tasks: new Map(), + deliveryStates: new Map(), + })); + configureTaskRegistryRuntime({ + store: { + loadSnapshot: loadTaskSnapshot, + saveSnapshot: () => {}, + }, + }); + configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot: () => { + throw new Error("SQLITE_CORRUPT: task-flow startup restore failed"); + }, + saveSnapshot: () => {}, + }, + }); + + expect(() => ensureTaskRuntimeStateReady()).toThrow( + "Task-flow registry restore failed: SQLITE_CORRUPT: task-flow startup restore failed", + ); + expect(loadTaskSnapshot).not.toHaveBeenCalled(); + }); + }); + + it("propagates task registry restore failures through the runtime gate", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + configureTaskFlowRegistryRuntime({ + store: createInMemoryTaskFlowRegistryStore(), + }); + configureTaskRegistryRuntime({ + store: { + loadSnapshot: () => { + throw new Error("SQLITE_IOERR: task startup restore failed"); + }, + saveSnapshot: () => {}, + }, + }); + + expect(() => ensureTaskRuntimeStateReady()).toThrow( + "Task registry restore failed: SQLITE_IOERR: task startup restore failed", + ); + }); + }); + it("reports task update success and retries when task-mirrored flow sync persistence fails", async () => { await withTaskRegistryTempDir(async () => { vi.useFakeTimers(); diff --git a/src/tasks/task-registry.ts b/src/tasks/task-registry.ts index 66b9d7159ea6..d39d89c40ff3 100644 --- a/src/tasks/task-registry.ts +++ b/src/tasks/task-registry.ts @@ -25,7 +25,10 @@ import { isDeliverableMessageChannel } from "../utils/message-channel.js"; import { cancelActiveCronTaskRun } from "./cron-task-cancel.js"; import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js"; import { isChildlessNativeSubagentTask } from "./native-subagent-task.js"; -import { isTaskFlowCancellationPending } from "./task-cancellation-state.js"; +import { + isProvisionalSubagentKillTask, + isTaskFlowCancellationPending, +} from "./task-cancellation-state.js"; import { formatTaskBlockedFollowupMessage, formatTaskStateChangeMessage, @@ -38,6 +41,7 @@ import { } from "./task-executor-policy.js"; import type { TaskFlowRecord } from "./task-flow-registry.types.js"; import { + ensureTaskFlowRegistryReady, getTaskFlowById, syncFlowFromTaskResult, updateFlowRecordByIdExpectedRevision, @@ -214,6 +218,27 @@ function assertParentFlowLinkAllowed(params: { } } +function ensureLinkedTaskFlowRegistryReady(task: Pick): void { + if (task.parentFlowId?.trim()) { + ensureTaskFlowRegistryReady(); + } +} + +function ensureTaskCancellationReady(task: TaskRecord): void { + const runId = task.runId?.trim(); + const linkedTasks = + runId && (task.runtime === "acp" || task.runtime === "subagent") + ? getTasksByRunScope({ + runId, + runtime: task.runtime, + sessionKey: task.childSessionKey, + }) + : [task]; + for (const linkedTask of linkedTasks.length > 0 ? linkedTasks : [task]) { + ensureLinkedTaskFlowRegistryReady(linkedTask); + } +} + function cloneTaskRecord(record: TaskRecord): TaskRecord { return { ...record }; } @@ -938,6 +963,7 @@ function mergeExistingTaskForCreate( notifyPolicy?: TaskNotifyPolicy; }, ): TaskRecord | null { + ensureLinkedTaskFlowRegistryReady(existing); const patch: Partial = {}; const requesterOrigin = normalizeDeliveryContext(params.requesterOrigin); const currentDeliveryState = taskDeliveryStates.get(existing.taskId); @@ -1295,6 +1321,8 @@ function updateTask(taskId: string, patch: Partial): TaskRecord | nu normalizeOptionalString(current.childSessionKey) !== normalizeOptionalString(next.childSessionKey); const parentFlowIndexChanged = current.parentFlowId?.trim() !== next.parentFlowId?.trim(); + ensureLinkedTaskFlowRegistryReady(current); + ensureLinkedTaskFlowRegistryReady(next); // Persist before mutating memory. If the store rejects the write, keep the // in-memory mirror at the durable value and report that no mutation applied. if (!tryPersistTaskUpsert(next, "update")) { @@ -2312,6 +2340,7 @@ export async function cancelTaskById(params: { } const childSessionKey = task.childSessionKey?.trim(); try { + ensureTaskCancellationReady(task); // A direct kill is only a provisional terminal projection. Re-read the // owning subagent run before promotion so its canonical completion can win. if (task.runtime !== "cli") { @@ -2500,6 +2529,18 @@ export async function cancelTaskById(params: { } } +export function assertTaskCancellationReadyById(taskId: string): TaskRecord | null { + ensureTaskRegistryReady(); + const task = tasks.get(taskId.trim()); + if (!task) { + return null; + } + if (!isTerminalTaskStatus(task.status) || isProvisionalSubagentKillTask(task)) { + ensureTaskCancellationReady(task); + } + return cloneTaskRecord(task); +} + // Callers that provide their own order use this cloned snapshot to avoid paying // for listTaskRecords' createdAt sort before immediately discarding that order. export function listTaskRecordsUnsorted(): TaskRecord[] { @@ -2678,6 +2719,7 @@ export function deleteTaskRecordById(taskId: string): boolean { if (!current) { return false; } + ensureLinkedTaskFlowRegistryReady(current); // Persist the delete before mutating memory, as a single atomic store // operation. If persistence fails, leave the in-memory record intact and // report that no delete was applied.