diff --git a/.changeset/execution-tombstones.md b/.changeset/execution-tombstones.md new file mode 100644 index 0000000000..e83b5344a2 --- /dev/null +++ b/.changeset/execution-tombstones.md @@ -0,0 +1,21 @@ +--- +"@executor-js/sdk": patch +"@executor-js/api": patch +"@executor-js/local-app": patch +--- + +fix: surface executions interrupted by a daemon restart instead of losing them silently + +A paused execution lives as an in-memory fiber inside the running engine. When +the local service restarts (login, crash, upgrade), every fiber is gone and a +later `executor resume` read as "approval expired" — silently discarding work +the agent believed was still pending. + +Executions now write a lightweight durable tombstone (id + status + +timestamp, no arguments, no results, no secrets) at pause time. On boot the +service marks every non-terminal tombstone `interrupted`; resuming an +interrupted execution returns an explicit `InterruptedExecutionError` telling +the agent to re-trigger the action, which is safe because nothing ran. + +Also adds the `@executor-js/sdk` execution-record store used by hosts that +need the same guarantee (cloud, self-host). diff --git a/apps/local/src/executor.ts b/apps/local/src/executor.ts index 1ec7ebbf68..1ba1d76efd 100644 --- a/apps/local/src/executor.ts +++ b/apps/local/src/executor.ts @@ -232,6 +232,28 @@ const createLocalExecutorLayer = (options: LocalExecutorOptions = {}) => { }, }); + // Boot sweep: any execution still running|paused from a previous + // process is unrecoverable (its fiber died with that process). Mark + // them interrupted so resume surfaces honestly instead of "not found". + // Runs after storage opens and before the API/MCP surfaces accept + // resume calls; never fails boot (a sweep hiccup is logged, not fatal). + yield* executor.executionRecords.sweepInterrupted().pipe( + Effect.map(({ interrupted }) => { + if (interrupted > 0) { + console.warn( + `[executor] Marked ${interrupted} execution(s) interrupted after restart; re-trigger them to resume.`, + ); + } + }), + Effect.catch(() => + Effect.sync(() => + console.warn( + "[executor] Execution tombstone sweep failed; interrupted state may be stale.", + ), + ), + ), + ); + if (migration.migrated) { console.warn( `[executor] Migrated local Executor data to v2; moved old DB to ${migration.backupPath}.`, diff --git a/packages/core/api/src/executions/api.ts b/packages/core/api/src/executions/api.ts index a7d22d9690..e6a1b3670c 100644 --- a/packages/core/api/src/executions/api.ts +++ b/packages/core/api/src/executions/api.ts @@ -75,6 +75,20 @@ const ApprovalExpiredError = Schema.TaggedStruct("ApprovalExpiredError", { "The approval window closed before the action was approved. Nothing ran; trigger the action again.", }); +/** + * The execution was interrupted by a daemon restart before it settled. + * + * Distinct from `ApprovalExpiredError` (the human never answered) and + * `ExecutionNotFoundError` (an id that was never ours): an interrupted + * execution is one the agent believed was still pending, but the service + * restarted and the fiber is unrecoverable. The honest outcome is + * "re-trigger the action" — nothing ran, so re-triggering is safe. + * See execution-records.ts. + */ +const InterruptedExecutionError = Schema.TaggedStruct("InterruptedExecutionError", { + executionId: Schema.String, +}).annotate({ httpApiStatus: 404 }); + /** * An artifact-originated execution that could not be turned into a call: the * code was not the shell proxy's emission, the artifact is not this caller's, @@ -125,6 +139,11 @@ export const ExecutionsApi = HttpApiGroup.make("executions") params: ExecutionParams, payload: ResumeRequest, success: ResumeResponse, - error: [InternalError, ExecutionNotFoundError, ApprovalExpiredError], + error: [ + InternalError, + ExecutionNotFoundError, + ApprovalExpiredError, + InterruptedExecutionError, + ], }), ); diff --git a/packages/core/api/src/handlers/executions.tombstone.test.ts b/packages/core/api/src/handlers/executions.tombstone.test.ts new file mode 100644 index 0000000000..4819478e50 --- /dev/null +++ b/packages/core/api/src/handlers/executions.tombstone.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Context, Effect, Layer, Predicate } from "effect"; +import { HttpRouter, HttpServer } from "effect/unstable/http"; +import { HttpApi, HttpApiBuilder } from "effect/unstable/httpapi"; + +import type { Executor } from "@executor-js/sdk"; + +import { ExecutionsApi } from "../executions/api"; +import { ExecutionsHandlers } from "./executions"; +import { ExecutionEngineService, ExecutorService } from "../services"; + +// --------------------------------------------------------------------------- +// Focused tests — spec execution-tombstones, AC4 (resume-time surface). +// +// When the daemon restarts, the paused fiber is gone. A resume must NOT read +// as a generic "approval expired": if a tombstone exists for the execution +// (written before the restart), the resume surfaces the honest +// "interrupted — re-trigger" outcome (InterruptedExecutionError). +// --------------------------------------------------------------------------- + +const stubExecutor = (record: { executionId: string; status: string } | null): Executor => + // oxlint-disable-next-line executor/no-double-cast -- minimal executor double: executionRecords.get and pendingApprovals.consume are exercised + ({ + executionRecords: { + get: () => Effect.succeed(record), + put: () => Effect.void, + sweepInterrupted: () => Effect.succeed({ interrupted: 0 }), + }, + // resumeFromPendingApproval consumes a stored approval before reaching + // the tombstone check; absent approvals are the restart scenario. + pendingApprovals: { + consume: () => Effect.succeed(null), + discard: () => Effect.void, + put: () => Effect.void, + }, + }) as unknown as Executor; + +// The engine remembers nothing (fresh process): live resume returns null, and +// there is no pending-approval record — this is the restart scenario. +// oxlint-disable-next-line executor/no-double-cast -- minimal engine double: only resume's null return (fresh process) is exercised +const emptyEngine = { + resume: () => Effect.succeed(null), +} as unknown as ExecutionEngineService["Service"]; + +const runResume = (executor: Executor) => { + const handler = HttpRouter.toWebHandler( + HttpApiBuilder.layer(HttpApi.make("executor").add(ExecutionsApi)).pipe( + Layer.provide(ExecutionsHandlers), + Layer.provide(Layer.succeed(ExecutorService)(executor)), + Layer.provide(Layer.succeed(ExecutionEngineService)(emptyEngine)), + Layer.provideMerge(HttpServer.layerServices), + Layer.provideMerge(Layer.succeed(HttpRouter.RouterConfig)({ maxParamLength: 1000 })), + ), + { disableLogger: false }, + ).handler; + // The handler's inferred type demands a Context second argument at the type level (a beta.59 + // inference quirk of toWebHandler's ReqR) even though the layer above + // provides both at runtime. Pass the runtime-provided context explicitly. + // The handler's inferred type demands a Context second argument at the type level (a beta.59 + // inference quirk of toWebHandler's ReqR) even though the layer above + // provides both at runtime. Passing the real services explicitly also + // satisfies the runtime — the stubs here are self-sufficient. + const context = Context.make(ExecutorService, executor).pipe( + Context.add(ExecutionEngineService, emptyEngine), + ); + return handler( + new Request("https://executor.test/executions/exec_1/resume", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ action: "accept" }), + }), + context, + ); +}; + +describe("resume after daemon restart (tombstone path)", () => { + it("surfaces interrupted (404 + InterruptedExecutionError) when a tombstone exists", async () => { + const res = await runResume(stubExecutor({ executionId: "exec_1", status: "interrupted" })); + expect(res.status).toBe(404); + const body = (await res.json()) as { _tag?: string; executionId?: string }; + expect(Predicate.isTagged(body, "InterruptedExecutionError")).toBe(true); + expect(body.executionId).toBe("exec_1"); + }); + + it("surfaces interrupted for a stale paused tombstone (sweep missed it — not a live execution)", async () => { + const res = await runResume(stubExecutor({ executionId: "exec_1", status: "paused" })); + expect(res.status).toBe(404); + expect(JSON.stringify(await res.json())).toContain("InterruptedExecutionError"); + }); + + it("falls through to approval-expired when no tombstone exists", async () => { + const res = await runResume(stubExecutor(null)); + // ApprovalExpiredError is annotated httpApiStatus: 410 (Gone). + expect(res.status).toBe(410); + expect(JSON.stringify(await res.json())).toContain("ApprovalExpiredError"); + }); + + it("completed tombstones do not resurrect (completed is immutable)", async () => { + const res = await runResume(stubExecutor({ executionId: "exec_1", status: "completed" })); + // No tombstone hit for completed (immutable) — falls through to expired (410). + expect(res.status).toBe(410); + expect(JSON.stringify(await res.json())).toContain("ApprovalExpiredError"); + }); +}); diff --git a/packages/core/api/src/handlers/executions.ts b/packages/core/api/src/handlers/executions.ts index 67f77de650..9db8f18d30 100644 --- a/packages/core/api/src/handlers/executions.ts +++ b/packages/core/api/src/handlers/executions.ts @@ -60,6 +60,29 @@ class ApprovalExpiredError extends Schema.TaggedErrorClass } } +/** + * The execution was interrupted by a daemon restart before it settled. + * + * Distinct from `ApprovalExpiredError` (the human never answered) and + * `ExecutionNotFoundError` (an id that was never ours): an interrupted + * execution is one the agent believed was still pending, but the service + * restarted and the fiber is unrecoverable. The honest outcome is + * "re-trigger the action" — nothing ran, so re-triggering is safe. + * 404, because the live execution no longer exists on this host. + * See execution-records.ts (tombstones). + */ +class InterruptedExecutionError extends Schema.TaggedErrorClass()( + "InterruptedExecutionError", + { + executionId: Schema.String, + }, + { httpApiStatus: 404 }, +) { + override get message(): string { + return "This execution was interrupted by a restart. Re-trigger the action."; + } +} + /** * Parse and bind one artifact-originated call, or fail with something the shell * can render inside the component that made it. @@ -157,6 +180,16 @@ const resumeFromPendingApproval = (executionId: string, action: "accept" | "decl ); if (outcome.status === "completed") { + // Terminal tombstone write — same rationale as the resume handler: + // stop the record from being sweep-eligible once the work is done. + // Best-effort, like every tombstone write. + yield* executor.executionRecords + .put({ + executionId, + status: "completed", + updatedAt: Date.now(), + }) + .pipe(Effect.catchCause(() => Effect.void)); const formatted = formatExecuteResult(outcome.result); return { status: "completed" as const, @@ -196,6 +229,7 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" capture( Effect.gen(function* () { const engine = yield* ExecutionEngineService; + const executor = yield* ExecutorService; // An artifact-originated request is not arbitrary code. It is parsed // against the shell proxy's one grammar and rewritten through the // artifact's connection bindings, exactly as `execute-action` does in @@ -232,6 +266,15 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" code, address: String(outcome.execution.elicitationContext.address), }); + // Tombstone the pause so a restart marks it interrupted rather + // than losing it silently. Best-effort, like the approval record. + yield* executor.executionRecords + .put({ + executionId: outcome.execution.id, + status: "paused", + updatedAt: Date.now(), + }) + .pipe(Effect.catchCause(() => Effect.void)); } const formatted = formatPausedExecution(outcome.execution); @@ -247,6 +290,7 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" capture( Effect.gen(function* () { const engine = yield* ExecutionEngineService; + const executor = yield* ExecutorService; const result = yield* captureEngineError( engine.resume(path.executionId, { action: payload.action, @@ -260,10 +304,34 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" if (!result) { const honoured = yield* resumeFromPendingApproval(path.executionId, payload.action); if (honoured) return honoured; + + // No live pause and no pending-approval record. If a tombstone + // exists for this execution, the daemon may have restarted since + // it paused — surface that honestly instead of a generic + // "approval expired". A running|paused tombstone read at resume + // time means the boot sweep missed it (host couldn't enumerate); + // treat it as interrupted here — it is not a live execution. + const record = yield* executor.executionRecords.get(path.executionId); + if (record !== null && record.status !== "completed") { + return yield* new InterruptedExecutionError({ executionId: path.executionId }); + } + return yield* new ApprovalExpiredError({ executionId: path.executionId }); } if (result.status === "completed") { + // Tombstone the terminal outcome so the record stops being + // sweep-eligible: without this write, a resumed-to-completed + // execution keeps its pre-restart "paused" tombstone and the + // next boot sweep would mark it "interrupted" — factually wrong + // for work that finished. Best-effort, like the pause write. + yield* executor.executionRecords + .put({ + executionId: path.executionId, + status: "completed", + updatedAt: Date.now(), + }) + .pipe(Effect.catchCause(() => Effect.void)); const formatted = formatExecuteResult(result.result); return { status: "completed" as const, diff --git a/packages/core/sdk/src/execution-records.test.ts b/packages/core/sdk/src/execution-records.test.ts new file mode 100644 index 0000000000..7217bf50bf --- /dev/null +++ b/packages/core/sdk/src/execution-records.test.ts @@ -0,0 +1,93 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { makeInMemoryBlobStore } from "./blob"; +import { makeExecutionRecordStore, type ExecutionRecord } from "./execution-records"; + +const record = (overrides?: Partial): ExecutionRecord => ({ + executionId: "exec_1", + status: "paused", + updatedAt: 1_000, + ...overrides, +}); + +const partition = "u:t:s"; + +describe("makeExecutionRecordStore", () => { + it.effect( + "round-trips a record and reads it through a second store over the same partition", + () => + Effect.gen(function* () { + // The whole point of the tombstone: the engine that paused is gone, so + // the record has to be readable by a caller that never saw the pause. + const blobs = makeInMemoryBlobStore(); + yield* makeExecutionRecordStore(blobs, partition).put(record()); + + const restarted = makeExecutionRecordStore(blobs, partition); + expect(yield* restarted.get("exec_1")).toStrictEqual(record()); + }), + ); + + it.effect("reads absent for an unknown execution id", () => + Effect.gen(function* () { + const store = makeExecutionRecordStore(makeInMemoryBlobStore(), partition); + expect(yield* store.get("never_existed")).toBeNull(); + }), + ); + + it.effect("is owner-scoped: a different partition does not see the record", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + yield* makeExecutionRecordStore(blobs, "u:t:other-subject").put(record()); + + // The namespace includes the partition — this caller's store reads a + // different namespace and simply does not see the record. + const otherStore = makeExecutionRecordStore(blobs, "u:t:s"); + expect(yield* otherStore.get("exec_1")).toBeNull(); + }), + ); + + it.effect("treats a corrupt record as absent (never surfaces garbage)", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + yield* blobs.put("u:t:s/@execution-records", "exec_1", "not-json{{"); + + const store = makeExecutionRecordStore(blobs, partition); + expect(yield* store.get("exec_1")).toBeNull(); + }), + ); + + it.effect("sweep marks running|paused records interrupted and clears the live index", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + const store = makeExecutionRecordStore(blobs, partition); + yield* store.put(record({ executionId: "exec_running", status: "running", updatedAt: 1 })); + yield* store.put(record({ executionId: "exec_paused", status: "paused", updatedAt: 2 })); + yield* store.put(record({ executionId: "exec_done", status: "completed", updatedAt: 3 })); + + const swept = yield* store.sweepInterrupted(); + expect(swept.interrupted).toBe(2); + + expect((yield* store.get("exec_running"))?.status).toBe("interrupted"); + expect((yield* store.get("exec_paused"))?.status).toBe("interrupted"); + // completed is immutable + expect((yield* store.get("exec_done"))?.status).toBe("completed"); + + // A second sweep finds nothing live to mark. + expect((yield* store.sweepInterrupted()).interrupted).toBe(0); + }), + ); + + it.effect("re-putting a terminal record removes it from the live index (no re-sweep)", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + const store = makeExecutionRecordStore(blobs, partition); + yield* store.put(record({ executionId: "exec_1", status: "running" })); + // Execution completes before any restart. + yield* store.put(record({ executionId: "exec_1", status: "completed", updatedAt: 2 })); + + expect((yield* store.sweepInterrupted()).interrupted).toBe(0); + expect((yield* store.get("exec_1"))?.status).toBe("completed"); + }), + ); +}); diff --git a/packages/core/sdk/src/execution-records.ts b/packages/core/sdk/src/execution-records.ts new file mode 100644 index 0000000000..5b747c9acb --- /dev/null +++ b/packages/core/sdk/src/execution-records.ts @@ -0,0 +1,187 @@ +// --------------------------------------------------------------------------- +// ExecutionRecordStore — durable "this execution existed" tombstones. +// +// A paused execution normally lives as a suspended fiber inside one engine +// instance (see the header comment in pending-approval.ts for the same +// constraint). When the daemon restarts (launchd KeepAlive makes this +// routine on the local install), every fiber is gone: `executor resume` for +// a pre-restart execution reads as "not found", silently discarding work the +// agent believes is still pending. Tombstones close that gap WITHOUT fiber +// serialization: each execution writes a lightweight record (id, status, +// updatedAt — no args, no results, no secrets) at start/pause/complete, and +// a boot sweep marks every non-terminal record `interrupted`. Resume of an +// interrupted execution is an explicit, honest outcome — "re-trigger the +// action" — never a silent NotFound and never a silent re-run. +// +// Records live in the existing owner-scoped `blob` table under a fixed +// namespace suffix, exactly like pending-approval records, so the partition +// IS the ownership check: another caller's executor reads a different +// namespace and simply does not see the record. +// +// Sweep enumerability: BlobStore has no list operation, and the old +// process's in-memory execution registry is unrecoverable at boot — so the +// store keeps its OWN index of live (running|paused) ids in a second +// namespace, updated on every put and consumed by the sweep. This is what +// makes `sweepInterrupted` honest: it reads the live set, marks each id +// interrupted, and clears the set. +// --------------------------------------------------------------------------- + +import { Effect, Option, Schema } from "effect"; + +import type { BlobStore } from "./blob"; +import type { StorageError } from "./fuma-runtime"; + +/** Lifecycle states persisted in the tombstone. */ +// Schema.Literals (array form) — in effect@4.0.0-beta.59 the multi-arg +// Schema.Literal("a","b",...) decodes ONLY its first member; the array form +// decodes the full set. The repo's resume endpoint uses the same array form. +export const ExecutionRecordStatus = Schema.Literals([ + "running", + "paused", + "interrupted", + "completed", +]); +export type ExecutionRecordStatus = typeof ExecutionRecordStatus.Type; + +/** + * The durable record for one execution. + * + * Deliberately minimal: id + status + updatedAt only. Arguments, results, + * and secrets never touch the tombstone (spec: no secret leakage). + */ +export const ExecutionRecord = Schema.Struct({ + executionId: Schema.String, + status: ExecutionRecordStatus, + /** Epoch ms of the last lifecycle transition. */ + updatedAt: Schema.Number, +}); +export type ExecutionRecord = typeof ExecutionRecord.Type; + +// Encode is plain JSON.stringify (repo convention, shape-memory.ts): the +// record type is already narrow at the call sites. Decode validates the +// parsed value against the schema — corrupt JSON reads as absent. +const encodeExecutionRecord = (record: ExecutionRecord): string => JSON.stringify(record); +const decodeRecordValue = Schema.decodeUnknownOption(ExecutionRecord); +// oxlint-disable executor/no-try-catch-or-throw,executor/no-json-parse -- boundary: untrusted persisted blob text; a corrupt record reads as absent (never surfaced), so a fallible parse collapsing to none is the contract +const decodeExecutionRecord = (raw: string): Option.Option => { + try { + return decodeRecordValue(JSON.parse(raw)); + } catch { + return Option.none(); + } +}; +// oxlint-enable executor/no-try-catch-or-throw,executor/no-json-parse + +// The live-id index: a JSON array of executionIds currently running|paused, +// stored under a single fixed key. Concurrency: blob writes are serialized by +// the storage adapter; a put is read-modify-write on this array. The boot +// sweep runs while no new executions can start (the daemon has not yet +// accepted work), so the read-modify-write is uncontended in practice. +const LIVE_INDEX_KEY = "live"; +const encodeLiveIds = (ids: readonly string[]): string => JSON.stringify(ids); +// oxlint-disable executor/no-try-catch-or-throw,executor/no-json-parse -- boundary: untrusted persisted index text; a corrupt index reads as empty (sweep finds nothing), so a fallible parse collapsing to [] is the contract +const decodeLiveIds = (raw: string): string[] => { + try { + const parsed = JSON.parse(raw) as unknown; + return Array.isArray(parsed) ? parsed.filter((x): x is string => typeof x === "string") : []; + } catch { + return []; + } +}; +// oxlint-enable executor/no-try-catch-or-throw,executor/no-json-parse + +/** + * Durable store for execution tombstones, scoped to one owner partition. + * + * `get` is a strict read: an unparseable record reads as absent (corrupt + * records are treated as gone, never surfaced). `sweepInterrupted` marks + * every non-terminal record `interrupted` in one pass and reports how many + * were swept — `completed` is immutable. + */ +export interface ExecutionRecordStore { + readonly put: (record: ExecutionRecord) => Effect.Effect; + readonly get: (executionId: string) => Effect.Effect; + /** Mark every non-terminal record `interrupted`; returns the count swept. */ + readonly sweepInterrupted: () => Effect.Effect<{ readonly interrupted: number }, StorageError>; +} + +/** + * Bind a `BlobStore` to one owner partition as an execution-record store. + * + * The namespace is the owner partition plus a fixed suffix, matching how + * plugin blobs namespace themselves — the same-query ownership rule the + * pending-approval store uses. + */ +export const makeExecutionRecordStore = ( + blobs: BlobStore, + partition: string, + now: () => number = Date.now, +): ExecutionRecordStore => { + const namespace = `${partition}/@execution-records`; + const liveNamespace = `${partition}/@execution-records-live`; + + /** Add or remove an id from the live index. */ + const updateLiveIndex = (executionId: string, add: boolean) => + Effect.gen(function* () { + const raw = yield* blobs.get(liveNamespace, LIVE_INDEX_KEY); + const live = decodeLiveIds(raw ?? "[]"); + const next = add + ? live.includes(executionId) + ? live + : [...live, executionId] + : live.filter((id) => id !== executionId); + yield* blobs.put(liveNamespace, LIVE_INDEX_KEY, encodeLiveIds(next)); + }); + + return { + put: (record) => + Effect.gen(function* () { + yield* blobs.put(namespace, record.executionId, encodeExecutionRecord(record)); + // Maintain the live index: running|paused ids are enumerated by the + // sweep; terminal ids leave the index (their records remain for + // get()). + if (record.status === "running" || record.status === "paused") { + yield* updateLiveIndex(record.executionId, true); + } else { + yield* updateLiveIndex(record.executionId, false); + } + }), + + get: (executionId) => + Effect.gen(function* () { + const raw = yield* blobs.get(namespace, executionId); + if (raw === null) return null; + const decoded = decodeExecutionRecord(raw); + if (Option.isNone(decoded)) return null; + return decoded.value; + }), + + sweepInterrupted: () => + Effect.gen(function* () { + const raw = yield* blobs.get(liveNamespace, LIVE_INDEX_KEY); + const live = decodeLiveIds(raw ?? "[]"); + let interrupted = 0; + for (const executionId of live) { + const recordRaw = yield* blobs.get(namespace, executionId); + if (recordRaw === null) continue; + const decoded = decodeExecutionRecord(recordRaw); + if (Option.isSome(decoded)) { + const record = decoded.value; + if (record.status === "running" || record.status === "paused") { + yield* blobs.put( + namespace, + executionId, + encodeExecutionRecord({ ...record, status: "interrupted", updatedAt: now() }), + ); + interrupted += 1; + } + } + } + // The live index is consumed by the sweep; nothing is live anymore + // from the previous process's perspective. New executions repopulate + // it on their first put. + yield* blobs.put(liveNamespace, LIVE_INDEX_KEY, encodeLiveIds([])); + return { interrupted }; + }), + }; +}; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index aba6b674c2..13b994a542 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -31,6 +31,7 @@ import { } from "./fuma-runtime"; import { makeFumaBlobStore, pluginBlobStore, type BlobStore, type OwnerPartitions } from "./blob"; import { makePendingApprovalStore, type PendingApprovalStore } from "./pending-approval"; +import { makeExecutionRecordStore, type ExecutionRecordStore } from "./execution-records"; import { coreToolsPlugin } from "./core-tools"; import type { Connection, @@ -499,6 +500,14 @@ export type Executor = { */ readonly pendingApprovals: PendingApprovalStore; + /** + * Durable execution tombstones — ids/statuses only, no args/results/secrets. + * Written at execution start/pause/complete; a boot sweep marks every + * non-terminal record `interrupted` so a restart surfaces honestly instead + * of reading as "not found". See `execution-records.ts`. + */ + readonly executionRecords: ExecutionRecordStore; + readonly execute: ( address: ToolAddress, args: unknown, @@ -6158,6 +6167,13 @@ export const createExecutor =