fix(sdk,api,local): surface executions interrupted by a daemon restart

.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).

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}.`,

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,
+      ],
     }),
   );

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<ExecutorService |
+  // ExecutionEngineService> 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<ExecutorService |
+  // ExecutionEngineService> 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");
+  });
+});

packages/core/api/src/handlers/executions.ts

@@ -60,6 +60,29 @@ class ApprovalExpiredError extends Schema.TaggedErrorClass<ApprovalExpiredError>
   }
 }
 
+/**
+ * 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>()(
+  "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,

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>): 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");
+    }),
+  );
+});