vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-17 06:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tasks): fail closed when task-flow state cannot restore (#105088)

Browse Source 
* fix(tasks): fail closed on task-flow restore errors

* fix(tasks): keep maintenance checks behind owner seam

* fix(tasks): gate runtime startup on flow restore

* test(tasks): prove runtime restore gate failures

* fix(tasks): gate cancellation on flow readiness
This commit is contained in:
Vincent Koc
2026-07-12 16:08:29 +08:00
committed by GitHub
parent c293fef3a8
commit d6d7c19c9d
21 changed files with 832 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/cli/gateway-cli/lifecycle.runtime.ts
View File

@@ -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";

54
src/cli/gateway-cli/run-loop.test.ts
View File

@@ -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();
}
});

4
src/cli/gateway-cli/run-loop.ts
View File

@@ -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");
});

41
src/commands/tasks-json.test.ts
View File

@@ -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);
});
});
});

4
src/commands/tasks-json.ts
View File

@@ -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,

109
src/commands/tasks.test.ts
View File

@@ -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();

2
src/commands/tasks.ts
View File

@@ -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

2
src/gateway/server-methods/tasks.ts
View File

@@ -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";

31
src/gateway/server-startup-early.test.ts
View File

@@ -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);

6
src/gateway/server-startup-early.ts
View File

@@ -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),
);

20
src/tasks/runtime-internal.ts
View File

@@ -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,

200
src/tasks/task-executor.test.ts
View File

@@ -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({

26
src/tasks/task-executor.ts
View File

@@ -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) {

22
src/tasks/task-flow-registry.audit.test.ts
View File

@@ -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", () => {

4
src/tasks/task-flow-registry.audit.ts
View File

@@ -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({

10
src/tasks/task-flow-registry.maintenance.ts
View File

@@ -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" ||

74
src/tasks/task-flow-registry.test.ts
View File

@@ -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");

90
src/tasks/task-flow-registry.ts
View File

@@ -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<string, TaskFlowRecord>();
let restoreAttempted = false;
let restoreFailureMessage: string | null = null;
let flows = new Map<string, TaskFlowRecord>();
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<string, TaskFlowRecord>();
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();

2
src/tasks/task-flow-runtime-internal.ts
View File

@@ -3,11 +3,13 @@ export {
createTaskFlowForTask,
createManagedTaskFlow,
deleteTaskFlowRecordById,
ensureTaskFlowRegistryReady,
failFlow,
finishFlow,
getTaskFlowById,
listTaskFlowRecords,
requestFlowCancel,
reloadTaskFlowRegistryFromStore,
resolveTaskFlowForLookupToken,
resetTaskFlowRegistryForTests,
resumeFlow,

148
src/tasks/task-registry.test.ts
View File

@@ -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<string, TaskRecord>(),
deliveryStates: new Map<string, TaskDeliveryState>(),
}));
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();

44
src/tasks/task-registry.ts
View File

@@ -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<TaskRecord, "parentFlowId">): 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<TaskRecord> = {};
const requesterOrigin = normalizeDeliveryContext(params.requesterOrigin);
const currentDeliveryState = taskDeliveryStates.get(existing.taskId);
@@ -1295,6 +1321,8 @@ function updateTask(taskId: string, patch: Partial<TaskRecord>): 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.