feat(durable): adapter-aware decision delivery, scoped to the tenant workspace

.beads/export-state.json

@@ -1 +1 @@
-{"last_dolt_commit":"fb1t0tec2l8cs1b4jr63mqbir0gi8ttb","timestamp":"2026-07-28T02:16:02.5366637-07:00","issues":456,"memories":0}
+{"last_dolt_commit":"q91afok34ortb9m4q349elh1ms8sllel","timestamp":"2026-07-28T03:15:25.5118813-07:00","issues":458,"memories":0}

packages/agents/scripts/durable-filesystem-spike.ts

@@ -37,12 +37,15 @@ async function main() {
   //    default, so this proves the env-selected path Foreman would ship.
   process.env.ZAPIER_DURABLE_ADAPTER = "filesystem";
   process.env.ZAPIER_DURABLE_FS_DIR = stateDir;
-  configureDurable({ adapter: "filesystem", filesystem: { baseDir: stateDir } });
+  // The key is `fsDir`. An unknown key is silently ignored and `fsDir` falls
+  // back to ~/.config/zapier-sdk/durable — writing into the real home dir.
+  configureDurable({ adapter: "filesystem", fsDir: stateDir });
 
   log("[1] config");
   const cfg = getConfig();
   log("adapter:", cfg.adapter);
   log("stateDir:", stateDir);
+  if (cfg.fsDir !== stateDir) throw new Error(`fsDir did not take: ${cfg.fsDir}`);
 
   const client = createClient();
   log("client:", client.constructor.name);

packages/agents/src/lib/automations/service.ts

@@ -1,11 +1,9 @@
 import { randomUUID } from "node:crypto";
 import {
-  cancelDurableRun,
   deleteAutomation as deleteZapierWorkflow,
+  deliveryForActiveAdapter,
   deployAutomation,
   getTriggerRunStatus,
-  postCallback,
-  resolveCallbackUrl,
   setAutomationEnabled,
   triggerAutomation,
 } from "../durable";
@@ -367,10 +365,17 @@ export async function cancelRunForUser(
     return { cancelled: true, status: "cancelled" };
   }
 
-  const sdk = await getExperimentalSdkForUser(userId);
-  const status = await cancelDurableRun(sdk, run.durable_run_id);
-  await store.updateRun(run.id, { status });
-  return { cancelled: status === "cancelled", status };
+  // Same adapter-aware seam as the approval path (foreman-gk6k) — a local run
+  // has no Zapier client to cancel against.
+  const delivery = await deliveryForActiveAdapter(() => getExperimentalSdkForUser(userId), {
+    tenantKey: workspaceId,
+  });
+  const outcome = await delivery.deliver(run.durable_run_id, { cancel: true });
+  if (!outcome.runStatus) {
+    return { cancelled: false, status: run.status };
+  }
+  await store.updateRun(run.id, { status: outcome.runStatus });
+  return { cancelled: outcome.ok, status: outcome.runStatus };
 }
 
 export interface CallbackResponseInput {
@@ -410,20 +415,20 @@ export async function respondToCallbackForUser(
     return { ok: false, action: "none", reason: "run is not waiting on a callback" };
   }
 
-  const sdk = await getExperimentalSdkForUser(userId);
-
-  if (input.cancel) {
-    const status = await cancelDurableRun(sdk, run.durable_run_id);
-    await store.updateRun(run.id, { status });
-    return { ok: status === "cancelled", action: "cancelled" };
-  }
+  // Adapter-specific mechanics live behind `deliverDecision` (foreman-gk6k) —
+  // on Zapier this resolves + POSTs a callback URL, on the filesystem adapter
+  // there is no URL to POST to and it goes through the local store.
+  const delivery = await deliveryForActiveAdapter(() => getExperimentalSdkForUser(userId), {
+    tenantKey: workspaceId,
+  });
+  const outcome = await delivery.deliver(run.durable_run_id, input);
 
-  const cb = await resolveCallbackUrl(sdk, run.durable_run_id, input.callbackName);
-  if (!cb) {
-    return { ok: false, action: "none", reason: "no open callback URL for this run" };
+  // Persist whatever status the adapter reported, not an assumed "cancelled" —
+  // a run can legitimately have finished before the cancel landed.
+  if (outcome.action === "cancelled" && outcome.runStatus) {
+    await store.updateRun(run.id, { status: outcome.runStatus });
   }
-  const res = await postCallback(cb.url, input.payload ?? {});
-  return { ok: res.ok, action: "resumed", status: res.status };
+  return outcome;
 }
 
 export interface UpdateInput {

packages/agents/src/lib/durable/delivery.ts

@@ -0,0 +1,229 @@
+import type { ExperimentalZapierSdk } from "../zapier/sdk";
+import { cancelDurableRun, postCallback, resolveCallbackUrl } from "./deploy";
+
+/**
+ * Adapter-aware delivery of a human decision to a durable's approval gate
+ * (foreman-gk6k).
+ *
+ * Approving a run means "get this payload to the waiting callback". HOW that
+ * happens is entirely adapter-specific:
+ *
+ * - **zapier**: the gate lives on Zapier's servers. `getDurableRun` will not
+ *   expose the callback URL, so it is recovered from the step the durable
+ *   authors to report it (`humanApprovalGate` in `./source`), then POSTed.
+ * - **filesystem**: the gate lives on local disk. `callbackBaseUrl` is a
+ *   `file://` URL, so there is nothing to POST to — delivery goes through the
+ *   adapter client's `callback(token, payload)`.
+ *
+ * A caller that reaches for `resolveCallbackUrl` + `postCallback` directly
+ * works on Zapier and SILENTLY does nothing on the filesystem adapter. Route
+ * and service code should only ever call `deliverDecision`.
+ */
+
+export type DurableAdapter = "zapier" | "filesystem";
+
+export interface DecisionInput {
+  /** Payload delivered to the gate (approve). Ignored when `cancel` is set. */
+  payload?: unknown;
+  /** Hard deny — terminate the run instead of resuming it. */
+  cancel?: boolean;
+  /** Which gate, when the run has more than one open callback. */
+  callbackName?: string;
+}
+
+export interface DecisionResult {
+  ok: boolean;
+  action: "resumed" | "cancelled" | "none";
+  /** HTTP status of the callback POST. Zapier adapter only — local delivery has no HTTP. */
+  status?: number;
+  /**
+   * Run status the adapter reports after a cancel. Persisted verbatim rather
+   * than assumed to be "cancelled" — Zapier can legitimately return something
+   * else (a run that finished before the cancel landed).
+   */
+  runStatus?: string;
+  /** Why it could not act, when `ok` is false. */
+  reason?: string;
+}
+
+/** One way of getting a decision to a waiting gate. */
+export interface DecisionDelivery {
+  readonly adapter: DurableAdapter;
+  deliver(durableRunId: string, input: DecisionInput): Promise<DecisionResult>;
+}
+
+/**
+ * Which runtime durables execute on.
+ *
+ * Defaults to `"zapier"` — deliberately NOT the package's own `"filesystem"`
+ * default. Every Foreman durable today runs on the Zapier adapter, so
+ * inheriting the upstream default would silently repoint production at an
+ * empty local store. `foreman-2qbk` owns making this a real, validated setting.
+ */
+export function activeDurableAdapter(): DurableAdapter {
+  return process.env.ZAPIER_DURABLE_ADAPTER === "filesystem" ? "filesystem" : "zapier";
+}
+
+/** Delivery against Zapier's hosted runtime — the path Foreman ships today. */
+export function zapierDelivery(sdk: ExperimentalZapierSdk): DecisionDelivery {
+  return {
+    adapter: "zapier",
+    async deliver(durableRunId, input) {
+      if (input.cancel) {
+        const status = await cancelDurableRun(sdk, durableRunId);
+        return { ok: status === "cancelled", action: "cancelled", runStatus: status };
+      }
+      const cb = await resolveCallbackUrl(sdk, durableRunId, input.callbackName);
+      if (!cb) {
+        return { ok: false, action: "none", reason: "no open callback URL for this run" };
+      }
+      const res = await postCallback(cb.url, input.payload ?? {});
+      return { ok: res.ok, action: "resumed", status: res.status };
+    },
+  };
+}
+
+/**
+ * Minimal view of `FilesystemClient` this module needs. Structural, so tests can
+ * pass a fake and so importing `@zapier/zapier-durable/node` stays lazy.
+ *
+ * `getOperations` is a synchronous, LEASE-FREE read. Do not reach for
+ * `checkout()` to inspect state — it acquires the runner's lease and returns
+ * `{ leased: false }` when the runner already holds it.
+ */
+export interface LocalDurableStore {
+  getOperations(executionId: string): Array<{
+    name: string;
+    type: string;
+    status: string;
+    callback_token?: string;
+  }>;
+  callback(token: string, payload: unknown): Promise<{ ok: true } | { error: string }>;
+  checkout(
+    executionId: string,
+    req?: { lease_token?: string },
+  ): Promise<{ leased: true; lease_token: string } | { leased: false }>;
+  release(
+    executionId: string,
+    req: { lease_token: string; status: string; error?: unknown },
+  ): Promise<{ ok: true; done: boolean }>;
+}
+
+/**
+ * Find the still-open callback gate for a local execution.
+ *
+ * Mirrors `resolveCallbackUrl`'s selection rule — named gate, else the sole
+ * pending one — but reads local operations instead of the Zapier wire. Local
+ * `OperationStatus` has no `"waiting"` member (that is an *execution* status),
+ * so a live gate is exactly `type: "callback"` + `status: "pending"`.
+ */
+export function findOpenLocalGate(
+  store: LocalDurableStore,
+  executionId: string,
+  callbackName?: string,
+): { token: string; name: string } | null {
+  const open = store
+    .getOperations(executionId)
+    .filter((o) => o.type === "callback" && o.status === "pending" && o.callback_token);
+  const gate = callbackName ? open.find((o) => o.name === callbackName) : open[0];
+  return gate?.callback_token ? { token: gate.callback_token, name: gate.name } : null;
+}
+
+/** Delivery against a local filesystem store — no Zapier account, no network. */
+export function filesystemDelivery(store: LocalDurableStore): DecisionDelivery {
+  return {
+    adapter: "filesystem",
+    async deliver(executionId, input) {
+      if (input.cancel) {
+        // No cancel primitive on the adapter. A suspended run holds no lease,
+        // so take it and release the execution as terminally failed. If the
+        // runner IS mid-tick the lease is refused and we report that honestly
+        // rather than half-cancelling.
+        const lease = await store.checkout(executionId);
+        if (!lease.leased) {
+          return { ok: false, action: "none", reason: "run is leased; cannot cancel mid-tick" };
+        }
+        await store.release(executionId, {
+          lease_token: lease.lease_token,
+          status: "failed",
+          error: { name: "Cancelled", message: "Cancelled by user from Foreman" },
+        });
+        return { ok: true, action: "cancelled", runStatus: "cancelled" };
+      }
+
+      const gate = findOpenLocalGate(store, executionId, input.callbackName);
+      if (!gate) {
+        return { ok: false, action: "none", reason: "no open callback for this run" };
+      }
+      // `CallbackRequest` is the raw payload — wrapping it as `{ payload }`
+      // fails edge validation against the gate's payloadSchema.
+      const res = await store.callback(gate.token, input.payload ?? {});
+      if ("error" in res) {
+        return { ok: false, action: "none", reason: `callback rejected: ${res.error}` };
+      }
+      return { ok: true, action: "resumed" };
+    },
+  };
+}
+
+/**
+ * Where one tenant's local durable state lives — INSIDE that tenant's agent
+ * workspace directory, alongside its files.
+ *
+ * Mirrors `mastra/agents/workspace.ts` (`FOREMAN_WORKSPACE_PATH`, default
+ * `./data/workspace`, one directory per `workspace_id`). Keeping durable state
+ * under the same per-tenant root means one workspace's suspended runs can never
+ * be read or resumed from another's, and nothing lands in the developer's home
+ * directory.
+ *
+ * This is path CO-LOCATION, not integration: the adapter writes with plain
+ * `node:fs`, not through Mastra's filesystem abstraction. Whether that keeps
+ * working under the sandbox-provider work (foreman-zlru) depends on WHERE the
+ * durable process runs:
+ *
+ * - **Inside the sandbox** — the per-tenant workspace FS is mounted there
+ *   (symlink locally, s3fs/gcsfuse in cloud), so a plain `node:fs` write to the
+ *   mount path lands on the real workspace FS. Works unchanged.
+ * - **On the host, with a CLOUD workspace FS** (foreman-udo9) — the host has no
+ *   mount, so `node:fs` writes to a local path that is no longer the workspace.
+ *   Diverges silently.
+ *
+ * Foreman runs durables in the agent-server process today, so the second case
+ * is the one that will actually bite. The durable package supports custom
+ * adapters; an adapter backed by the workspace filesystem would make this
+ * correct by construction rather than by path coincidence (foreman-1uz7).
+ */
+export function durableStateDirFor(tenantKey: string): string {
+  const root = process.env.FOREMAN_WORKSPACE_PATH ?? "./data/workspace";
+  const safe = tenantKey.replace(/[^a-zA-Z0-9_-]/g, "") || "_shared";
+  return `${root}/${safe}/.durable`;
+}
+
+/**
+ * Build the delivery for the active adapter.
+ *
+ * The Zapier SDK is passed as a thunk so the filesystem path never mints a
+ * Zapier client (it has no account), and the durable package is imported
+ * lazily so the Zapier path never loads the local adapter.
+ */
+export async function deliveryForActiveAdapter(
+  getSdk: () => Promise<ExperimentalZapierSdk>,
+  opts: { tenantKey?: string } = {},
+): Promise<DecisionDelivery> {
+  if (activeDurableAdapter() === "filesystem") {
+    // Import from the ROOT entry point, never `@zapier/zapier-durable/node`.
+    // The subpath re-exports the same names but carries SEPARATE config state:
+    // `configureDurable` there reports success via its own `getConfig()` while
+    // the durable runtime keeps writing to the default `fsDir`. Measured
+    // 2026-07-28 on 0.11.0.
+    const { FilesystemClient } = await import("@zapier/zapier-durable");
+    // Construct the client explicitly instead of `createClient()`. `createClient`
+    // reads PROCESS-GLOBAL config, which cannot be per-tenant on a server serving
+    // several workspaces at once — flipping `fsDir` per request is a race. An
+    // explicit `baseDir` (and `DurableCallOptions.client` on the run side) keeps
+    // tenancy per-call, with no global state involved.
+    const baseDir = durableStateDirFor(opts.tenantKey ?? "_shared");
+    return filesystemDelivery(new FilesystemClient({ baseDir }) as unknown as LocalDurableStore);
+  }
+  return zapierDelivery(await getSdk());
+}

packages/agents/src/lib/durable/index.ts

@@ -3,6 +3,18 @@
  * Zapier SDK surface. The execution substrate for the trigger/workflow rebuild.
  */
 
+export {
+  activeDurableAdapter,
+  type DecisionDelivery,
+  type DecisionInput,
+  type DecisionResult,
+  type DurableAdapter,
+  deliveryForActiveAdapter,
+  filesystemDelivery,
+  findOpenLocalGate,
+  type LocalDurableStore,
+  zapierDelivery,
+} from "./delivery";
 export {
   type AutomationSummary,
   cancelDurableRun,

packages/agents/src/lib/durable/source.ts

@@ -1,3 +1,4 @@
+import { activeDurableAdapter, type DurableAdapter } from "./delivery";
 import type { AutomationSpec } from "./types";
 
 /**
@@ -17,9 +18,22 @@ import type { AutomationSpec } from "./types";
  * POST to the URL. Returns source lines that: create the gate, report its `{ callbackUrl,
  * callbackName }` via a step, and await the decision into `<id>Decision`.
  */
-export function humanApprovalGate(name: string): string {
+export function humanApprovalGate(name: string, adapter?: DurableAdapter): string {
   const q = (v: unknown) => JSON.stringify(v);
   const id = name.replace(/[^a-zA-Z0-9_$]/g, "_").replace(/^([0-9])/, "_$1");
+  const target = adapter ?? activeDurableAdapter();
+
+  // On the filesystem adapter the run is local, so Foreman reads the callback
+  // token straight off the execution's operations (`findOpenLocalGate`). The
+  // URL never has to cross a wire, so neither the binding nor the reporting
+  // step is emitted — one less step to journal and replay per approval.
+  if (target === "filesystem") {
+    return [
+      `  const [${id}Approval] = await ctx.createCallback(${q(name)});`,
+      `  const ${id}Decision = await ${id}Approval;`,
+    ].join("\n");
+  }
+
   return [
     `  const [${id}Approval, ${id}Url] = await ctx.createCallback(${q(name)});`,
     `  await ctx.step(${q(`__report_callback_url_${name}`)}, async () => ({ callbackUrl: ${id}Url, callbackName: ${q(name)} }));`,

packages/agents/tests/unit/automations-service.test.ts

@@ -15,7 +15,11 @@ vi.mock("@/lib/automations/schedules", () => ({
   unregisterAutomationSchedule: vi.fn(async () => {}),
   assertValidCron: vi.fn(),
 }));
-vi.mock("@/lib/durable", () => ({
+// Mocked one level down, at the module that actually owns these calls, so the
+// REAL delivery seam (`deliveryForActiveAdapter` → `zapierDelivery`) runs on
+// top of them (foreman-gk6k). Mocking `@/lib/durable` wholesale would stub out
+// the seam itself, and these assertions would stop proving the Zapier path.
+vi.mock("@/lib/durable/deploy", () => ({
   deployAutomation: vi.fn(async () => ({
     workflowId: "wf_1",
     versionId: "ver_1",
@@ -296,7 +300,9 @@ describe("respondToCallbackForUser (foreman-zfnj)", () => {
       durable_run_id: "dr_1",
     } as never);
     const r = await respondToCallbackForUser("user-1", "run_1", { cancel: true });
-    expect(r).toEqual({ ok: true, action: "cancelled" });
+    // `runStatus` carries what the adapter actually reported, so the caller
+    // persists the truth rather than assuming "cancelled" (foreman-gk6k).
+    expect(r).toEqual({ ok: true, action: "cancelled", runStatus: "cancelled" });
     expect(cancelDurableRun).toHaveBeenCalledWith(expect.anything(), "dr_1");
     expect(store.updateRun).toHaveBeenCalledWith("run_1", { status: "cancelled" });
     expect(postCallback).not.toHaveBeenCalled();