ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-19 09:01:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(workboard): recover interrupted worker launches (#126170)

Browse Source 
* fix(workboard): recover interrupted worker launches

Persist prepared, accepted, and failed launch phases so Gateway restart reconciliation cannot leave cards permanently running between launch preparation and worker acceptance.

* fix(workboard): require durable terminal evidence

Do not synthesize terminal-session acceptance timing during restart reconciliation; stale same-key terminal rows without updatedAt now fail the prepared launch instead of being adopted.
This commit is contained in:
Peter Steinberger
2026-08-18 22:01:14 -07:00
committed by GitHub
parent 85cec65a19
commit dcdfd737e5
10 changed files with 1006 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/workboard/src/dispatcher-ownership.test.ts
+61 -1
View File
@@ -567,7 +567,7 @@ describe("Workboard dispatcher ownership", () => {
status: "ready",
workspaceAccess: { unrestricted: true },
});
vi.spyOn(store, "enrichExecutionAssociation").mockRejectedValue(
vi.spyOn(store, "acceptExecutionLaunch").mockRejectedValue(
new Error("execution enrichment unavailable"),
);
let provisionalRunId = "";
@@ -584,6 +584,15 @@ describe("Workboard dispatcher ownership", () => {
sessionKey: input.sessionKey,
runId: provisionalRunId,
},
metadata: {
automation: {
launch: {
phase: "prepared",
requestedSessionKey: input.sessionKey,
provisionalRunId,
},
},
},
});
return { sessionKey: canonicalSessionKey, runId: "accepted-run" };
});
@@ -619,6 +628,7 @@ describe("Workboard dispatcher ownership", () => {
runId: provisionalRunId,
execution: { status: "running", runId: provisionalRunId },
metadata: {
automation: { launch: { phase: "prepared", provisionalRunId } },
claim: { ownerId: "workboard-dispatcher" },
workerLogs: [expect.objectContaining({ runId: "accepted-run" })],
},
@@ -642,6 +652,56 @@ describe("Workboard dispatcher ownership", () => {
},
);
it("marks a prepared launch accepted after Gateway acceptance", async () => {
const store = new WorkboardStore(createMemoryStore());
const card = await store.create({
title: "Worker with durable acceptance",
status: "ready",
workspaceAccess: { unrestricted: true },
});
const canonicalSessionKey = `agent:worker:subagent:workboard-default-${card.id}`;
let provisionalRunId = "";
const run = vi.fn().mockImplementation(async (input) => {
provisionalRunId = input.idempotencyKey;
await expect(store.get(card.id)).resolves.toMatchObject({
sessionKey: input.sessionKey,
runId: provisionalRunId,
metadata: {
automation: {
launch: {
phase: "prepared",
requestedSessionKey: input.sessionKey,
provisionalRunId,
},
},
},
});
return { sessionKey: canonicalSessionKey, runId: "accepted-run" };
});
await dispatchAndStartWorkboardCards({
store,
subagent: { run },
options: { maxStarts: 1 },
});
await expect(store.get(card.id)).resolves.toMatchObject({
sessionKey: canonicalSessionKey,
runId: "accepted-run",
metadata: {
automation: {
launch: {
phase: "accepted",
requestedSessionKey: expect.any(String),
provisionalRunId,
acceptedSessionKey: canonicalSessionKey,
acceptedRunId: "accepted-run",
},
},
},
});
});
it.each(["backlog", "todo", "ready"] as const)(
"starts an exact dashboard card from %s",
async (status) => {
extensions/workboard/src/dispatcher.ts
+38 -42
View File
@@ -3,6 +3,7 @@ import path from "node:path";
import type {
WorkboardCard,
WorkboardExecution,
WorkboardLaunchState,
WorkboardWorkspace,
} from "@openclaw/workboard-contract";
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
@@ -71,6 +72,8 @@ type WorkboardDispatchAndStartResult = WorkboardDispatchResult & {
startFailures: WorkboardStartFailure[];
};
type WorkboardPreparedLaunch = Extract<WorkboardLaunchState, { phase: "prepared" }>;
type WorkboardDispatchStartParams = {
store: WorkboardStore;
subagent: WorkboardSubagentRuntime;
@@ -364,8 +367,8 @@ async function runWorkboardDispatch(
let materializedWorkspace: WorkboardWorkspace | undefined;
let implicitWorkspaceCwd: string | undefined;
let runStarted = false;
let launchIntentPersisted = false;
let workspaceMutation: { before: WorkboardCard; after: WorkboardCard } | undefined;
let preparedLaunch: WorkboardPreparedLaunch | undefined;
const requestedWorkspace = card.metadata?.automation?.workspace;
let workspaceAccess: WorkboardWorkspaceAccess;
let targetWorkspace: string | undefined;
@@ -502,28 +505,14 @@ async function runWorkboardDispatch(
);
workspaceMutation = { before: workspaceBase, after: materializedCard };
}
const launchBase = await params.store.get(card.id);
if (!launchBase) {
throw new Error(`card not found: ${card.id}`);
}
const runId = `workboard:${card.id}:${launchBase.updatedAt}`;
const launched = await params.store.update(
card.id,
{
sessionKey,
runId,
execution: buildExecution({
card: launchBase,
sessionKey,
runId,
runtime: undefined,
now,
}),
...(materializedWorkspace ? { workspace: materializedWorkspace } : {}),
},
{ expectedUpdatedAt: launchBase.updatedAt },
);
launchIntentPersisted = true;
const prepared = await params.store.prepareExecutionLaunch(card.id, {
requestedSessionKey: sessionKey,
now,
scope: { ownerId, token: claimValue },
});
const launched = prepared.card;
preparedLaunch = prepared.launch;
const runId = prepared.launch.provisionalRunId;
const run = await params.subagent.run({
sessionKey,
message: buildWorkerPrompt({
@@ -556,15 +545,18 @@ async function runWorkboardDispatch(
runId: run.runId,
execution: acceptedExecution,
};
const updated = await params.store
.enrichExecutionAssociation(card.id, {
expectedSessionKey: sessionKey,
expectedRunId: runId,
sessionKey: acceptedSessionKey,
runId: run.runId,
execution: acceptedExecution,
})
.catch(() => acceptedCard);
const updated =
(await params.store
.acceptExecutionLaunch(card.id, {
expectedLaunch: prepared.launch,
acceptedAt: Math.max(Date.now(), prepared.launch.preparedAt),
expectedSessionKey: sessionKey,
expectedRunId: runId,
sessionKey: acceptedSessionKey,
runId: run.runId,
execution: acceptedExecution,
})
.catch(() => undefined)) ?? acceptedCard;
acceptedStarts += 1;
startedOwners.add(ownerId);
started.push({
@@ -614,16 +606,20 @@ async function runWorkboardDispatch(
continue;
}
try {
await params.store.block(
card.id,
{
ownerId,
token: claimValue,
reason: `Dispatcher could not start worker: ${message}`,
},
{ ownerId, token: claimValue },
{ clearExecutionAssociation: launchIntentPersisted },
);
const reason = `Dispatcher could not start worker: ${message}`;
if (preparedLaunch) {
await params.store.failPreparedLaunch(card.id, {
expectedLaunch: preparedLaunch,
reason,
failedAt: Date.now(),
});
} else {
await params.store.block(
card.id,
{ ownerId, token: claimValue, reason },
{ ownerId, token: claimValue },
);
}
} catch {
// Leave the original start failure visible; dispatch will diagnose stale claims later.
}
extensions/workboard/src/lifecycle-sync.restart.test.ts
+361
View File
@@ -0,0 +1,361 @@
import { describe, expect, it, vi } from "vitest";
import { dispatchAndStartWorkboardCards } from "./dispatcher.js";
import { createWorkboardLifecycleService, syncWorkboardSubagentEnded } from "./lifecycle-sync.js";
import type { PersistedWorkboardCard, WorkboardKeyedStore } from "./persistence-types.js";
import { WorkboardStore } from "./store.js";
function createMemoryStore(): WorkboardKeyedStore {
const entries = new Map<string, PersistedWorkboardCard>();
return {
async register(key, value) {
entries.set(key, value);
},
async lookup(key) {
return entries.get(key);
},
async delete(key) {
return entries.delete(key);
},
async entries() {
return [...entries].map(([key, value]) => ({ key, value }));
},
};
}
function createDeferred<T>() {
let resolve!: (value: T) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, resolve, reject };
}
async function beginPreparedDispatch() {
const keyed = createMemoryStore();
const dispatchStore = new WorkboardStore(keyed);
const card = await dispatchStore.create({
title: "Prepared across restart",
status: "ready",
workspaceAccess: { unrestricted: true },
});
const reachedRun = createDeferred<{ sessionKey: string; provisionalRunId: string }>();
const runResult = createDeferred<{ sessionKey?: string; runId: string }>();
const run = vi.fn().mockImplementation(async (input) => {
reachedRun.resolve({
sessionKey: input.sessionKey,
provisionalRunId: input.idempotencyKey,
});
return await runResult.promise;
});
const dispatch = dispatchAndStartWorkboardCards({
store: dispatchStore,
subagent: { run },
options: { maxStarts: 1 },
});
const prepared = await reachedRun.promise;
return {
card,
dispatch,
prepared,
runResult,
replacementStore: new WorkboardStore(keyed),
};
}
async function rejectInterruptedDispatch(
interrupted: Awaited<ReturnType<typeof beginPreparedDispatch>>,
) {
interrupted.runResult.reject(new Error("simulated dispatcher process loss"));
await interrupted.dispatch;
}
async function startLifecycleSweep(params: {
store: WorkboardStore;
sessions: Array<{
key: string;
updatedAt?: number;
status?: "running" | "done" | "failed" | "killed" | "timeout";
hasActiveRun?: boolean;
}>;
complete: boolean;
}) {
const readSessions = vi.fn().mockResolvedValue({
sessions: params.sessions,
complete: params.complete,
});
const service = createWorkboardLifecycleService({ store: params.store, readSessions });
const context = { logger: { warn: vi.fn() } } as never;
await service.start(context);
return { context, readSessions, service };
}
async function stopLifecycleSweep(
lifecycle: Awaited<ReturnType<typeof startLifecycleSweep>>,
): Promise<void> {
lifecycle.service.onGatewayStop();
await lifecycle.service.stop?.(lifecycle.context);
}
describe("Workboard prepared launch restart recovery", () => {
it("fails a prepared launch absent from the first complete post-restart snapshot", async () => {
const interrupted = await beginPreparedDispatch();
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [],
complete: true,
});
await expect(interrupted.replacementStore.get(interrupted.card.id)).resolves.toMatchObject({
status: "running",
sessionKey: interrupted.prepared.sessionKey,
runId: interrupted.prepared.provisionalRunId,
execution: { status: "running", runId: interrupted.prepared.provisionalRunId },
metadata: {
claim: { ownerId: "workboard-dispatcher" },
attempts: [{ status: "running", runId: interrupted.prepared.provisionalRunId }],
automation: { launch: { phase: "prepared" } },
},
});
await new Promise<void>((resolve) => {
setTimeout(resolve, 0);
});
expect(lifecycle.readSessions).not.toHaveBeenCalled();
lifecycle.service.onGatewayStart();
await vi.waitFor(async () => {
expect((await interrupted.replacementStore.get(interrupted.card.id))?.status).toBe("blocked");
});
await stopLifecycleSweep(lifecycle);
const recovered = await interrupted.replacementStore.get(interrupted.card.id);
expect(recovered).toMatchObject({
status: "blocked",
metadata: {
automation: {
launch: {
phase: "failed",
requestedSessionKey: interrupted.prepared.sessionKey,
provisionalRunId: interrupted.prepared.provisionalRunId,
reason: expect.stringContaining("Gateway"),
},
},
attempts: [
expect.objectContaining({
status: "blocked",
runId: interrupted.prepared.provisionalRunId,
endedAt: expect.any(Number),
}),
],
},
});
expect(recovered?.metadata?.claim).toBeUndefined();
expect(recovered?.sessionKey).toBeUndefined();
expect(recovered?.runId).toBeUndefined();
expect(recovered?.execution).toBeUndefined();
await rejectInterruptedDispatch(interrupted);
expect(
(await interrupted.replacementStore.get(interrupted.card.id))?.metadata?.automation?.launch,
).toMatchObject({ phase: "failed" });
});
it("fails a prepared launch when its persisted session has no active run", async () => {
const interrupted = await beginPreparedDispatch();
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [
{
key: `agent:worker:${interrupted.prepared.sessionKey}`,
status: "running",
hasActiveRun: false,
},
],
complete: true,
});
lifecycle.service.onGatewayStart();
await vi.waitFor(async () => {
expect((await interrupted.replacementStore.get(interrupted.card.id))?.status).toBe("blocked");
});
await stopLifecycleSweep(lifecycle);
await expect(interrupted.replacementStore.get(interrupted.card.id)).resolves.toMatchObject({
status: "blocked",
metadata: { automation: { launch: { phase: "failed" } } },
});
expect(
(await interrupted.replacementStore.get(interrupted.card.id))?.metadata?.claim,
).toBeUndefined();
await rejectInterruptedDispatch(interrupted);
});
it("accepts a prepared launch found under its canonical post-restart session", async () => {
const interrupted = await beginPreparedDispatch();
const canonicalSessionKey = `agent:worker:${interrupted.prepared.sessionKey}`;
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [{ key: canonicalSessionKey, status: "running", hasActiveRun: true }],
complete: true,
});
expect(lifecycle.readSessions).not.toHaveBeenCalled();
lifecycle.service.onGatewayStart();
await vi.waitFor(async () => {
expect(
(await interrupted.replacementStore.get(interrupted.card.id))?.metadata?.automation?.launch
?.phase,
).toBe("accepted");
});
const accepted = await interrupted.replacementStore.get(interrupted.card.id);
expect(accepted).toMatchObject({
status: "running",
sessionKey: canonicalSessionKey,
runId: interrupted.prepared.provisionalRunId,
metadata: {
claim: { ownerId: "workboard-dispatcher" },
attempts: [
expect.objectContaining({
status: "running",
sessionKey: canonicalSessionKey,
runId: interrupted.prepared.provisionalRunId,
}),
],
automation: {
launch: {
phase: "accepted",
acceptedSessionKey: canonicalSessionKey,
},
},
},
});
expect(accepted?.metadata?.automation?.launch).not.toHaveProperty("acceptedRunId");
await syncWorkboardSubagentEnded({
store: interrupted.replacementStore,
event: {
targetSessionKey: canonicalSessionKey,
runId: "accepted-run",
outcome: "ok",
},
});
const terminal = await interrupted.replacementStore.get(interrupted.card.id);
expect(terminal).toMatchObject({
status: "review",
sessionKey: canonicalSessionKey,
runId: "accepted-run",
metadata: {
automation: { launch: { phase: "accepted", acceptedRunId: "accepted-run" } },
attempts: [
expect.objectContaining({
id: "accepted-run",
status: "succeeded",
runId: "accepted-run",
}),
],
},
});
expect(terminal?.metadata?.attempts).toHaveLength(1);
await stopLifecycleSweep(lifecycle);
await rejectInterruptedDispatch(interrupted);
expect((await interrupted.replacementStore.get(interrupted.card.id))?.status).toBe("review");
});
it("accepts a prepared launch from an explicit terminal session snapshot", async () => {
const interrupted = await beginPreparedDispatch();
const canonicalSessionKey = `agent:worker:${interrupted.prepared.sessionKey}`;
const launch = (await interrupted.replacementStore.get(interrupted.card.id))?.metadata
?.automation?.launch;
if (launch?.phase !== "prepared") {
throw new Error("expected prepared launch");
}
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [
{
key: canonicalSessionKey,
status: "done",
hasActiveRun: false,
updatedAt: launch.preparedAt + 1,
},
],
complete: true,
});
lifecycle.service.onGatewayStart();
await vi.waitFor(async () => {
expect((await interrupted.replacementStore.get(interrupted.card.id))?.status).toBe("review");
});
await stopLifecycleSweep(lifecycle);
await expect(interrupted.replacementStore.get(interrupted.card.id)).resolves.toMatchObject({
status: "review",
sessionKey: canonicalSessionKey,
metadata: {
automation: {
launch: { phase: "accepted", acceptedSessionKey: canonicalSessionKey },
},
attempts: [expect.objectContaining({ status: "succeeded" })],
},
});
await rejectInterruptedDispatch(interrupted);
});
it("does not accept a terminal session without durable timing evidence", async () => {
const interrupted = await beginPreparedDispatch();
const canonicalSessionKey = `agent:worker:${interrupted.prepared.sessionKey}`;
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [
{
key: canonicalSessionKey,
status: "done",
hasActiveRun: false,
},
],
complete: true,
});
lifecycle.service.onGatewayStart();
await vi.waitFor(async () => {
expect((await interrupted.replacementStore.get(interrupted.card.id))?.status).toBe("blocked");
});
await stopLifecycleSweep(lifecycle);
await expect(interrupted.replacementStore.get(interrupted.card.id)).resolves.toMatchObject({
status: "blocked",
metadata: { automation: { launch: { phase: "failed" } } },
});
await rejectInterruptedDispatch(interrupted);
});
it("keeps a prepared launch running when the post-restart snapshot is incomplete", async () => {
const interrupted = await beginPreparedDispatch();
const lifecycle = await startLifecycleSweep({
store: interrupted.replacementStore,
sessions: [],
complete: false,
});
lifecycle.service.onGatewayStart();
await vi.waitFor(() => expect(lifecycle.readSessions).toHaveBeenCalledOnce());
await new Promise<void>((resolve) => {
setTimeout(resolve, 0);
});
await expect(interrupted.replacementStore.get(interrupted.card.id)).resolves.toMatchObject({
status: "running",
sessionKey: interrupted.prepared.sessionKey,
runId: interrupted.prepared.provisionalRunId,
metadata: {
claim: { ownerId: "workboard-dispatcher" },
automation: { launch: { phase: "prepared" } },
},
});
await stopLifecycleSweep(lifecycle);
await rejectInterruptedDispatch(interrupted);
});
});
extensions/workboard/src/lifecycle-sync.ts
+47 -18
View File
@@ -27,7 +27,7 @@ const workboardLifecycleGatewayState = resolveGlobalSingleton(
},
);
type WorkboardLifecycleState = "running" | "succeeded" | "failed" | "idle" | "missing" | "stale";
type WorkboardLifecycleState = "running" | "succeeded" | "failed" | "idle" | "stale";
type WorkboardLifecycleObservation = {
state: WorkboardLifecycleState;
@@ -52,6 +52,16 @@ type WorkboardLifecycleSessionSnapshot = {
complete: boolean;
};
function sessionProvesPreparedAcceptance(session: WorkboardLifecycleSession): boolean {
return (
session.hasActiveRun === true ||
session.status === "done" ||
session.status === "failed" ||
session.status === "killed" ||
session.status === "timeout"
);
}
type WorkboardLifecycleSessionReadOptions = {
includeUnknown: boolean;
};
@@ -87,7 +97,6 @@ const LIFECYCLE_TARGETS = {
succeeded: { card: "review", execution: "review" },
failed: { card: "blocked", execution: "blocked" },
idle: { execution: "idle" },
missing: {},
stale: { card: "running", execution: "running" },
} as const satisfies Record<
WorkboardLifecycleState,
@@ -104,6 +113,7 @@ async function syncWorkboardCardLifecycle(params: {
expectedRunId?: string;
sessionKey: string;
runId?: string;
acceptedAt?: number;
};
}): Promise<boolean> {
const target = LIFECYCLE_TARGETS[params.observation.state];
@@ -140,6 +150,7 @@ async function syncWorkboardLifecycleEvent(params: {
...(cardRunId(card) ? { expectedRunId: cardRunId(card) } : {}),
sessionKey: params.source.sessionKey,
...(params.source.runId ? { runId: params.source.runId } : {}),
acceptedAt: params.observation.sourceUpdatedAt ?? params.now,
},
}
: {}),
@@ -286,28 +297,46 @@ async function syncWorkboardLifecycleSessions(params: {
(canUseAgentlessFallback && suffixIndex >= 0
? sessionsByWorkboardSuffix.get(lookupKey.slice(suffixIndex))
: undefined);
const observation = session
? lifecycleFromSession(session, now)
: params.complete
? ({ state: "missing" } as const)
const launch = card.metadata?.automation?.launch;
const preparedAcceptanceAt =
launch?.phase === "prepared" && session && sessionProvesPreparedAcceptance(session)
? session.hasActiveRun === true
? Math.max(now, launch.preparedAt)
: session.updatedAt
: undefined;
if (
observation &&
(await syncWorkboardCardLifecycle({
launch?.phase === "prepared" &&
(preparedAcceptanceAt === undefined || preparedAcceptanceAt < launch.preparedAt)
) {
if (
params.complete &&
(await params.store.failPreparedLaunch(card.id, {
expectedLaunch: launch,
reason: "Gateway did not accept the prepared Workboard session before restart.",
failedAt: now,
}))
) {
count += 1;
}
continue;
}
if (!session) {
continue;
}
const observation = lifecycleFromSession(session, now);
if (
await syncWorkboardCardLifecycle({
store: params.store,
cardId: card.id,
observation,
now,
...(session
? {
association: {
...(cardSessionKey(card) ? { expectedSessionKey: cardSessionKey(card) } : {}),
...(cardRunId(card) ? { expectedRunId: cardRunId(card) } : {}),
sessionKey: session.key,
},
}
: {}),
}))
association: {
...(cardSessionKey(card) ? { expectedSessionKey: cardSessionKey(card) } : {}),
...(cardRunId(card) ? { expectedRunId: cardRunId(card) } : {}),
sessionKey: session.key,
...(preparedAcceptanceAt === undefined ? {} : { acceptedAt: preparedAcceptanceAt }),
},
})
) {
count += 1;
}
extensions/workboard/src/store-core.ts
+2
View File
@@ -78,6 +78,7 @@ import {
} from "./store-normalizers.js";
type WorkboardUpdateCardOptions = {
allowAutomationLaunch?: boolean;
allowMetadataDependencyLinks?: boolean;
enforceStatusHolds?: boolean;
event?: Omit<WorkboardEvent, "id" | "at">;
@@ -786,6 +787,7 @@ export class WorkboardCoreStore {
: syncExecutionSessionKey(existing.execution, sessionKey)
: normalizeExecution(effectivePatch.execution);
let metadata = normalizeMetadata(effectivePatch.metadata, existing.metadata, {
allowAutomationLaunch: options.allowAutomationLaunch,
allowDependencyLinks: options.allowMetadataDependencyLinks !== false,
preserveProofId: options.preserveProofId,
});
extensions/workboard/src/store-normalizers.ts
+64 -8
View File
@@ -31,6 +31,7 @@ import {
type WorkboardExecutionStatus,
type WorkboardLink,
type WorkboardLinkType,
type WorkboardLaunchState,
type WorkboardMetadata,
type WorkboardNotification,
type WorkboardNotificationKind,
@@ -414,11 +415,9 @@ function normalizeWorkspace(
export function normalizeAutomation(
value: unknown,
fallback: WorkboardAutomation = {},
options: { allowLaunchState?: boolean } = {},
): WorkboardAutomation | undefined {
if (!isRecord(value)) {
return Object.keys(fallback).length ? fallback : undefined;
}
const record = value;
const record = isRecord(value) ? value : {};
const tenant = normalizeBoundedString(record.tenant, fallback.tenant, 80, "tenant");
const boardId = Object.hasOwn(record, "boardId")
? normalizeBoardId(record.boardId, fallback.boardId)
@@ -460,8 +459,11 @@ export function normalizeAutomation(
const workspace = Object.hasOwn(record, "workspace")
? normalizeWorkspace(record.workspace, fallback.workspace)
: fallback.workspace;
// Raw metadata preserves host-issued authority but cannot mint or widen it.
// Raw metadata preserves host-issued authority/state but cannot mint or widen either.
const workspaceAccess = fallback.workspaceAccess;
const launch = normalizeLaunchState(
options.allowLaunchState && Object.hasOwn(record, "launch") ? record.launch : fallback.launch,
);
const next = removeUndefinedAutomationFields({
...(tenant ? { tenant } : {}),
...(boardId ? { boardId } : {}),
@@ -477,10 +479,60 @@ export function normalizeAutomation(
...(createdCardIds?.length ? { createdCardIds } : {}),
...(dispatchCount ? { dispatchCount } : {}),
...(lastDispatchAt ? { lastDispatchAt } : {}),
...(launch ? { launch } : {}),
});
return Object.keys(next).length ? next : undefined;
}
function normalizeLaunchTimestamp(value: unknown): number | undefined {
return typeof value === "number" && Number.isFinite(value) && value >= 0
? Math.trunc(value)
: undefined;
}
function normalizeLaunchString(value: unknown, maxLength: number): string | undefined {
const normalized = normalizeOptionalString(value);
return normalized && normalized.length <= maxLength ? normalized : undefined;
}
function normalizeLaunchState(value: unknown): WorkboardLaunchState | undefined {
if (!isRecord(value)) {
return undefined;
}
const requestedSessionKey = normalizeLaunchString(value.requestedSessionKey, 240);
const provisionalRunId = normalizeLaunchString(value.provisionalRunId, 160);
const preparedAt = normalizeLaunchTimestamp(value.preparedAt);
if (!requestedSessionKey || !provisionalRunId || preparedAt === undefined) {
return undefined;
}
const identity = { requestedSessionKey, provisionalRunId, preparedAt };
if (value.phase === "prepared") {
return { phase: "prepared", ...identity };
}
if (value.phase === "accepted") {
const acceptedAt = normalizeLaunchTimestamp(value.acceptedAt);
const acceptedSessionKey = normalizeLaunchString(value.acceptedSessionKey, 240);
const acceptedRunId = normalizeLaunchString(value.acceptedRunId, 160);
return acceptedAt === undefined || !acceptedSessionKey
? undefined
: {
phase: "accepted",
...identity,
acceptedAt,
acceptedSessionKey,
...(acceptedRunId ? { acceptedRunId } : {}),
};
}
if (value.phase === "failed") {
const failedAt = normalizeLaunchTimestamp(value.failedAt);
const reason = normalizeLaunchString(value.reason, 800);
return failedAt === undefined || !reason
? undefined
: { phase: "failed", ...identity, failedAt, reason };
}
return undefined;
}
export function deriveChildIdempotencyKey(
parentKey: string | undefined,
index: number,
@@ -1035,6 +1087,7 @@ export function normalizeMetadata(
options: {
allowDependencyLinks?: boolean;
allowArchivedAt?: boolean;
allowAutomationLaunch?: boolean;
preserveProofId?: string;
} = {},
): WorkboardMetadata {
@@ -1112,9 +1165,11 @@ export function normalizeMetadata(
workerProtocol: Object.hasOwn(record, "workerProtocol")
? normalizeWorkerProtocol(record.workerProtocol, fallback.workerProtocol)
: fallback.workerProtocol,
automation: Object.hasOwn(record, "automation")
? normalizeAutomation(record.automation, fallback.automation)
: fallback.automation,
automation: normalizeAutomation(
Object.hasOwn(record, "automation") ? record.automation : {},
fallback.automation,
{ allowLaunchState: options.allowAutomationLaunch },
),
claim: Object.hasOwn(record, "claim")
? record.claim
? normalizeClaim(record.claim, fallback.claim)
@@ -1240,6 +1295,7 @@ function removeUndefinedAutomationFields(automation: WorkboardAutomation): Workb
"createdCardIds",
"dispatchCount",
"lastDispatchAt",
"launch",
] as const) {
const value = next[key];
if (
extensions/workboard/src/store-workflow.ts
+46 -35
View File
@@ -4,6 +4,7 @@ import type {
WorkboardArtifact,
WorkboardCard,
WorkboardClaim,
WorkboardMetadata,
WorkboardNotification,
WorkboardRunAttempt,
} from "@openclaw/workboard-contract";
@@ -32,6 +33,7 @@ import {
} from "./store-constants.js";
import type {
WorkboardBlockInput,
WorkboardCardPatch,
WorkboardClaimInput,
WorkboardClaimOptions,
WorkboardCompleteInput,
@@ -330,6 +332,49 @@ export class WorkboardWorkflowStore extends WorkboardPromoteStore {
);
}
protected buildBlockedCardPatch(
existing: WorkboardCard,
reason: string,
now: number,
options: { clearExecutionAssociation?: boolean } = {},
): WorkboardCardPatch & { metadata: WorkboardMetadata } {
const metadata = existing.metadata ?? {};
const notification: WorkboardNotification = {
id: randomUUID(),
kind: "failed",
createdAt: now,
sequence: this.nextNotificationSequence(now),
message: capText(reason, 240) ?? "Workboard card blocked.",
...(cardSessionKey(existing) ? { sessionKey: cardSessionKey(existing) } : {}),
...(cardRunId(existing) ? { runId: cardRunId(existing) } : {}),
};
const execution =
existing.execution?.status === "running"
? { ...existing.execution, status: "blocked" as const, updatedAt: now }
: existing.execution;
return {
status: "blocked",
...(options.clearExecutionAssociation
? { sessionKey: null, runId: null, execution: null }
: execution
? { execution }
: {}),
metadata: {
...metadata,
claim: undefined,
attempts: closeRunningAttempts(metadata.attempts, now, "blocked", reason),
failureCount: (metadata.failureCount ?? 0) + 1,
comments: [
...(metadata.comments ?? []),
{ id: randomUUID(), body: reason, createdAt: now },
].slice(-MAX_CARD_COMMENTS),
notifications: [...(metadata.notifications ?? []), notification].slice(
-MAX_CARD_NOTIFICATIONS,
),
},
};
}
async block(
id: string,
input: WorkboardBlockInput = {},
@@ -346,41 +391,7 @@ export class WorkboardWorkflowStore extends WorkboardPromoteStore {
const reason =
normalizeBoundedString(input.reason, undefined, 2000, "block reason") ??
"Workboard card blocked.";
const metadata = existing.metadata ?? {};
const notification: WorkboardNotification = {
id: randomUUID(),
kind: "failed",
createdAt: now,
sequence: this.nextNotificationSequence(now),
message: capText(reason, 240) ?? "Workboard card blocked.",
...(cardSessionKey(existing) ? { sessionKey: cardSessionKey(existing) } : {}),
...(cardRunId(existing) ? { runId: cardRunId(existing) } : {}),
};
const execution =
existing.execution?.status === "running"
? { ...existing.execution, status: "blocked" as const, updatedAt: now }
: existing.execution;
return await this.updateCard(id, {
status: "blocked",
...(options.clearExecutionAssociation
? { sessionKey: null, runId: null, execution: null }
: execution
? { execution }
: {}),
metadata: {
...metadata,
claim: undefined,
attempts: closeRunningAttempts(metadata.attempts, now, "blocked", reason),
failureCount: (metadata.failureCount ?? 0) + 1,
comments: [
...(metadata.comments ?? []),
{ id: randomUUID(), body: reason, createdAt: now },
].slice(-MAX_CARD_COMMENTS),
notifications: [...(metadata.notifications ?? []), notification].slice(
-MAX_CARD_NOTIFICATIONS,
),
},
});
return await this.updateCard(id, this.buildBlockedCardPatch(existing, reason, now, options));
});
}
extensions/workboard/src/store.test.ts
+89
View File
@@ -1093,6 +1093,95 @@ describe("WorkboardStore", () => {
expect(untrusted.metadata?.automation?.workspaceAccess).toBeUndefined();
});
it("only accepts launch state from store-owned transitions", async () => {
const store = new WorkboardStore(createMemoryStore());
const card = await store.create({
title: "Trusted launch",
status: "ready",
workspaceAccess: { unrestricted: true },
metadata: {
automation: {
launch: {
phase: "prepared",
requestedSessionKey: "injected-session",
provisionalRunId: "injected-run",
preparedAt: 1,
},
},
},
});
expect(card.metadata?.automation?.launch).toBeUndefined();
const claimed = await store.claim(card.id, { ownerId: "worker", token: "claim-token" });
const prepared = await store.prepareExecutionLaunch(card.id, {
requestedSessionKey: "subagent:workboard-default-trusted",
now: 100,
scope: { ownerId: "worker", token: claimed.token },
});
const updated = await store.update(card.id, {
metadata: {
automation: {
launch: {
...prepared.launch,
phase: "accepted",
acceptedAt: 101,
acceptedSessionKey: "agent:injected:subagent:workboard-default-trusted",
acceptedRunId: "injected-accepted-run",
},
},
},
});
expect(updated.metadata?.automation?.launch).toEqual(prepared.launch);
});
it("does not let a delayed launch failure overwrite a newer retry", async () => {
const store = new WorkboardStore(createMemoryStore());
const card = await store.create({ title: "Retried launch", status: "ready" });
const firstClaim = await store.claim(card.id, { ownerId: "worker" });
const first = await store.prepareExecutionLaunch(card.id, {
requestedSessionKey: "subagent:workboard-default-retried",
now: 100,
scope: { ownerId: "worker", token: firstClaim.token },
});
await store.failPreparedLaunch(card.id, {
expectedLaunch: first.launch,
reason: "first launch failed",
failedAt: 101,
});
await store.unblock(card.id);
const retryClaim = await store.claim(card.id, { ownerId: "worker" });
const retry = await store.prepareExecutionLaunch(card.id, {
requestedSessionKey: "subagent:workboard-default-retried",
now: 200,
scope: { ownerId: "worker", token: retryClaim.token },
});
await expect(
store.failPreparedLaunch(card.id, {
expectedLaunch: first.launch,
reason: "delayed first failure",
failedAt: 201,
}),
).resolves.toBe(false);
await expect(
store.syncLifecycle(card.id, {
targetStatus: "review",
executionStatus: "review",
sourceUpdatedAt: undefined,
stale: undefined,
now: 202,
association: {
expectedSessionKey: retry.launch.requestedSessionKey,
expectedRunId: retry.launch.provisionalRunId,
sessionKey: `agent:worker:${retry.launch.requestedSessionKey}`,
acceptedAt: 150,
},
}),
).resolves.toBe(false);
expect((await store.get(card.id))?.metadata?.automation?.launch).toEqual(retry.launch);
});
it("moves cards and records lifecycle timestamps", async () => {
const store = new WorkboardStore(createMemoryStore());
const card = await store.create({ title: "Ship workboard" });
extensions/workboard/src/store.ts
+277 -81
View File
@@ -6,6 +6,7 @@ import type {
WorkboardDiagnostic,
WorkboardExecution,
WorkboardExecutionStatus,
WorkboardLaunchState,
WorkboardMetadata,
WorkboardStaleState,
WorkboardStatus,
@@ -19,6 +20,8 @@ import type {
import { createWorkboardSqliteStores } from "./sqlite-store.js";
import {
buildWorkerContext,
assertCanMutateClaimedCard,
capText,
cardBoardId,
cardRunId,
cardSessionKey,
@@ -44,6 +47,7 @@ import type {
WorkboardDiagnosticsResult,
WorkboardDispatchOptions,
WorkboardDispatchResult,
WorkboardMutationScope,
} from "./store-inputs.js";
import { normalizeBoardId, normalizeTimestamp } from "./store-normalizers.js";
import { WorkboardNotificationStore } from "./store-notifications.js";
@@ -58,15 +62,73 @@ type WorkboardExecutionAssociationInput = {
runId?: string;
execution: WorkboardExecution;
};
type WorkboardExecutionAssociationPatchInput = WorkboardExecutionAssociationInput & {
launch?: WorkboardLaunchState;
};
type WorkboardLifecycleAssociation = Omit<WorkboardExecutionAssociationInput, "execution">;
type WorkboardLifecycleAssociation = Omit<WorkboardExecutionAssociationInput, "execution"> & {
acceptedAt?: number;
};
type WorkboardExecutionAssociationPatch = WorkboardCardPatch & {
metadata?: WorkboardMetadata;
};
type WorkboardPreparedLaunch = Extract<WorkboardLaunchState, { phase: "prepared" }>;
function preparedLaunchMatchesCard(
card: WorkboardCard,
expected: WorkboardPreparedLaunch,
): boolean {
const launch = card.metadata?.automation?.launch;
return (
launch?.phase === "prepared" &&
launch.requestedSessionKey === expected.requestedSessionKey &&
launch.provisionalRunId === expected.provisionalRunId &&
launch.preparedAt === expected.preparedAt &&
card.sessionKey === expected.requestedSessionKey &&
card.runId === expected.provisionalRunId &&
card.execution?.sessionKey === expected.requestedSessionKey &&
card.execution?.runId === expected.provisionalRunId
);
}
function acceptedLaunchForAssociation(
card: WorkboardCard,
association: WorkboardLifecycleAssociation,
): WorkboardLaunchState | undefined {
const launch = card.metadata?.automation?.launch;
if (launch?.phase === "prepared") {
if (
!preparedLaunchMatchesCard(card, launch) ||
association.acceptedAt === undefined ||
association.acceptedAt < launch.preparedAt
) {
return undefined;
}
return {
...launch,
phase: "accepted",
acceptedAt: association.acceptedAt,
acceptedSessionKey: association.sessionKey,
...(association.runId ? { acceptedRunId: association.runId } : {}),
};
}
if (
launch?.phase !== "accepted" ||
(launch.acceptedSessionKey === association.sessionKey &&
(!association.runId || launch.acceptedRunId === association.runId))
) {
return undefined;
}
return {
...launch,
acceptedSessionKey: association.sessionKey,
...(association.runId ? { acceptedRunId: association.runId } : {}),
};
}
function executionAssociationPatch(
card: WorkboardCard,
input: WorkboardExecutionAssociationInput,
input: WorkboardExecutionAssociationPatchInput,
): WorkboardExecutionAssociationPatch | undefined {
if (
cardSessionKey(card) !== input.expectedSessionKey ||
@@ -94,11 +156,21 @@ function executionAssociationPatch(
};
}
}
const metadata =
attemptIndex >= 0 || input.launch
? {
...card.metadata,
...(attemptIndex >= 0 ? { attempts } : {}),
...(input.launch
? { automation: { ...card.metadata?.automation, launch: input.launch } }
: {}),
}
: undefined;
return {
sessionKey: input.sessionKey,
...(input.runId ? { runId: input.runId } : {}),
execution: input.execution,
...(attemptIndex >= 0 ? { metadata: { ...card.metadata, attempts } } : {}),
...(metadata ? { metadata } : {}),
};
}
@@ -126,15 +198,123 @@ function lifecycleExecution(params: {
// Capability layers split review boundaries only; the core still owns persistence and mutation order.
export class WorkboardStore extends WorkboardNotificationStore {
async enrichExecutionAssociation(
async prepareExecutionLaunch(
id: string,
input: WorkboardExecutionAssociationInput,
): Promise<WorkboardCard> {
input: {
requestedSessionKey: string;
now: number;
scope: WorkboardMutationScope;
},
): Promise<{ card: WorkboardCard; launch: WorkboardPreparedLaunch }> {
return await this.enqueueMutation(async () => {
const result = await this.updateLatestCard(id, (current) =>
executionAssociationPatch(current, input),
const result = await this.updateLatestCard(
id,
(card) => {
assertCanMutateClaimedCard(card, input.scope);
const provisionalRunId = `workboard:${card.id}:${card.updatedAt}`;
const launch: WorkboardPreparedLaunch = {
phase: "prepared",
requestedSessionKey: input.requestedSessionKey,
provisionalRunId,
preparedAt: card.updatedAt,
};
return {
sessionKey: input.requestedSessionKey,
runId: provisionalRunId,
execution: {
id: card.execution?.id ?? `${card.id}:agent-session`,
kind: "agent-session",
mode: "autonomous",
status: "running",
sessionKey: input.requestedSessionKey,
runId: provisionalRunId,
startedAt: input.now,
updatedAt: input.now,
},
metadata: {
...card.metadata,
automation: { ...card.metadata?.automation, launch },
},
};
},
{ allowAutomationLaunch: true },
);
return result.card;
const launch = result.card.metadata?.automation?.launch;
if (launch?.phase !== "prepared") {
throw new Error("prepared Workboard launch was not persisted");
}
return { card: result.card, launch };
});
}
async acceptExecutionLaunch(
id: string,
input: WorkboardExecutionAssociationInput & {
expectedLaunch: WorkboardPreparedLaunch;
acceptedAt: number;
},
): Promise<WorkboardCard | undefined> {
return await this.enqueueMutation(async () => {
const result = await this.updateLatestCard(
id,
(card) => {
if (
!preparedLaunchMatchesCard(card, input.expectedLaunch) ||
input.acceptedAt < input.expectedLaunch.preparedAt
) {
return undefined;
}
const launch: WorkboardLaunchState = {
...input.expectedLaunch,
phase: "accepted",
acceptedAt: input.acceptedAt,
acceptedSessionKey: input.sessionKey,
...(input.runId ? { acceptedRunId: input.runId } : {}),
};
return executionAssociationPatch(card, { ...input, launch });
},
{ allowAutomationLaunch: true },
);
return result.updated ? result.card : undefined;
});
}
async failPreparedLaunch(
id: string,
input: { expectedLaunch: WorkboardPreparedLaunch; reason: string; failedAt: number },
): Promise<boolean> {
const failedAt = Math.max(input.failedAt, input.expectedLaunch.preparedAt);
const reason = capText(input.reason, 2000) ?? "Dispatcher could not start worker.";
const launchReason = capText(reason, 800) ?? "Prepared launch failed.";
return await this.enqueueMutation(async () => {
const result = await this.updateLatestCard(
id,
(card) => {
if (!preparedLaunchMatchesCard(card, input.expectedLaunch)) {
return undefined;
}
const blocked = this.buildBlockedCardPatch(card, reason, failedAt, {
clearExecutionAssociation: true,
});
return {
...blocked,
metadata: {
...blocked.metadata,
automation: {
...card.metadata?.automation,
launch: {
...input.expectedLaunch,
phase: "failed",
failedAt,
reason: launchReason,
},
},
},
};
},
{ allowAutomationLaunch: true },
);
return result.updated;
});
}
@@ -150,83 +330,99 @@ export class WorkboardStore extends WorkboardNotificationStore {
},
): Promise<boolean> {
return await this.enqueueMutation(async () => {
const result = await this.updateLatestCard(id, (card) => {
if (card.metadata?.archivedAt) {
return undefined;
}
const patch: WorkboardCardPatch = {};
let metadata: Record<string, unknown> | undefined;
const associationIsCurrent =
!input.association ||
((input.sourceUpdatedAt === undefined ||
!shouldSkipPersistedLifecycleStatusUpdate(card, input.sourceUpdatedAt)) &&
cardSessionKey(card) === input.association.expectedSessionKey &&
cardRunId(card) === input.association.expectedRunId);
// Recompute from the latest row after every cross-host CAS conflict.
if (
associationIsCurrent &&
input.sourceUpdatedAt !== undefined &&
shouldSyncWorkboardLifecycleStatus(card, input.targetStatus)
) {
patch.status = input.targetStatus;
metadata = { lifecycleStatusSourceUpdatedAt: input.sourceUpdatedAt };
}
const associationNeedsUpdate =
input.association &&
(card.sessionKey !== input.association.sessionKey ||
(input.association.runId !== undefined && card.runId !== input.association.runId) ||
!card.execution ||
card.execution.sessionKey !== input.association.sessionKey ||
(input.association.runId !== undefined &&
card.execution.runId !== input.association.runId) ||
(input.executionStatus !== undefined &&
card.execution.status !== input.executionStatus));
if (associationIsCurrent && input.association && associationNeedsUpdate) {
const associationPatch = executionAssociationPatch(card, {
...input.association,
execution: lifecycleExecution({
card,
association: input.association,
status: input.executionStatus,
now: input.now,
}),
});
if (associationPatch) {
Object.assign(patch, associationPatch);
metadata = { ...associationPatch.metadata, ...metadata };
const result = await this.updateLatestCard(
id,
(card) => {
if (card.metadata?.archivedAt) {
return undefined;
}
} else if (
!input.association &&
card.execution &&
input.executionStatus &&
card.execution.status !== input.executionStatus
) {
patch.execution = {
...card.execution,
status: input.executionStatus,
updatedAt: input.now,
};
}
if (associationIsCurrent && input.stale) {
const existing = card.metadata?.stale;
const patch: WorkboardCardPatch = {};
let metadata: Record<string, unknown> | undefined;
const launch = card.metadata?.automation?.launch;
const associationIsCurrent =
!input.association ||
((input.sourceUpdatedAt === undefined ||
!shouldSkipPersistedLifecycleStatusUpdate(card, input.sourceUpdatedAt)) &&
(launch?.phase !== "prepared" ||
(input.association.acceptedAt !== undefined &&
input.association.acceptedAt >= launch.preparedAt)) &&
cardSessionKey(card) === input.association.expectedSessionKey &&
cardRunId(card) === input.association.expectedRunId);
// Recompute from the latest row after every cross-host CAS conflict.
if (
!existing ||
existing.lastSessionUpdatedAt !== input.stale.lastSessionUpdatedAt ||
existing.reason !== input.stale.reason
associationIsCurrent &&
input.sourceUpdatedAt !== undefined &&
shouldSyncWorkboardLifecycleStatus(card, input.targetStatus)
) {
metadata = {
...metadata,
stale: { ...input.stale, detectedAt: existing?.detectedAt ?? input.stale.detectedAt },
patch.status = input.targetStatus;
metadata = { lifecycleStatusSourceUpdatedAt: input.sourceUpdatedAt };
}
const acceptedLaunch = input.association
? acceptedLaunchForAssociation(card, input.association)
: undefined;
const associationNeedsUpdate =
input.association &&
(card.sessionKey !== input.association.sessionKey ||
(input.association.runId !== undefined && card.runId !== input.association.runId) ||
!card.execution ||
card.execution.sessionKey !== input.association.sessionKey ||
(input.association.runId !== undefined &&
card.execution.runId !== input.association.runId) ||
(input.executionStatus !== undefined &&
card.execution.status !== input.executionStatus) ||
Boolean(acceptedLaunch));
if (associationIsCurrent && input.association && associationNeedsUpdate) {
const associationPatch = executionAssociationPatch(card, {
...input.association,
execution: lifecycleExecution({
card,
association: input.association,
status: input.executionStatus,
now: input.now,
}),
...(acceptedLaunch ? { launch: acceptedLaunch } : {}),
});
if (associationPatch) {
Object.assign(patch, associationPatch);
metadata = { ...associationPatch.metadata, ...metadata };
}
} else if (
!input.association &&
card.execution &&
input.executionStatus &&
card.execution.status !== input.executionStatus
) {
patch.execution = {
...card.execution,
status: input.executionStatus,
updatedAt: input.now,
};
}
} else if (associationIsCurrent && card.metadata?.stale) {
metadata = { ...metadata, stale: null };
}
if (metadata) {
patch.metadata = metadata;
}
return Object.keys(patch).length === 0 ? undefined : patch;
});
if (associationIsCurrent && input.stale) {
const existing = card.metadata?.stale;
if (
!existing ||
existing.lastSessionUpdatedAt !== input.stale.lastSessionUpdatedAt ||
existing.reason !== input.stale.reason
) {
metadata = {
...metadata,
stale: {
...input.stale,
detectedAt: existing?.detectedAt ?? input.stale.detectedAt,
},
};
}
} else if (associationIsCurrent && card.metadata?.stale) {
metadata = { ...metadata, stale: null };
}
if (metadata) {
patch.metadata = metadata;
}
return Object.keys(patch).length === 0 ? undefined : patch;
},
{ allowAutomationLaunch: true },
);
return result.updated;
});
}
packages/workboard-contract/src/index.ts
+21
View File
@@ -250,6 +250,26 @@ export type WorkboardWorkspaceAccess =
| { unrestricted: true }
| { unrestricted: false; roots: string[]; writable: boolean };
type WorkboardLaunchIdentity = {
requestedSessionKey: string;
provisionalRunId: string;
preparedAt: number;
};
export type WorkboardLaunchState =
| (WorkboardLaunchIdentity & { phase: "prepared" })
| (WorkboardLaunchIdentity & {
phase: "accepted";
acceptedAt: number;
acceptedSessionKey: string;
acceptedRunId?: string;
})
| (WorkboardLaunchIdentity & {
phase: "failed";
failedAt: number;
reason: string;
});
export type WorkboardAutomation = {
tenant?: string;
boardId?: string;
@@ -265,6 +285,7 @@ export type WorkboardAutomation = {
createdCardIds?: string[];
dispatchCount?: number;
lastDispatchAt?: number;
launch?: WorkboardLaunchState;
};
export type WorkboardBoardMetadata = {