From b284a692e9271c0f16500191f626011f13ac7d76 Mon Sep 17 00:00:00 2001 From: HamChowderr Date: Tue, 28 Jul 2026 03:02:41 -0700 Subject: [PATCH 1/2] feat(durable): adapter-aware decision delivery, scoped to the tenant workspace foreman-gk6k. Approve/Deny assumed the callback was an HTTP endpoint, so it worked only on the Zapier adapter. lib/durable/delivery.ts is now the single seam: routes and service code call deliverDecision and never touch resolveCallbackUrl/postCallback directly. - zapier: unchanged mechanics (resolve the reported URL, POST it, cancel via the SDK). The existing service tests now mock @/lib/durable/deploy instead of @/lib/durable, so the REAL seam runs on top of them and those assertions keep proving the Zapier path rather than a stub. - filesystem: no URL to POST to, so delivery goes through the store's callback(token, payload). The token is read with getOperations(), a lease-free read -- checkout() would take the runner's lease. - cancel: the adapter has no cancel primitive. A suspended run holds no lease, so take it and release as failed; if the runner IS mid-tick the lease is refused and that is reported instead of half-cancelling. - DecisionResult carries runStatus so callers persist what the adapter really reported, not an assumed cancelled. State lives at /.durable, per tenant, matching the agent workspace layout. FilesystemClient is constructed with an explicit baseDir rather than createClient(), because createClient reads PROCESS-GLOBAL config that cannot be per-tenant on a server serving several workspaces at once. cancelRunForUser went through the seam too -- it had the same hardcoding. Two traps found by measurement and now guarded: - configureDurable imported from @zapier/zapier-durable/node reports success via its own getConfig() while the runtime keeps writing to the default fsDir. The subpath and the root hold SEPARATE config state. Import from the root only. - The option key is fsDir; an unknown key (e.g. filesystem: {baseDir}) is silently ignored and falls back to ~/.config/zapier-sdk/durable. An earlier revision of these tests wrote into the real home directory because of it. There is now a regression test asserting nothing is written there. Filed foreman-1uz7: node:fs co-location with the workspace holds only while the workspace FS is local, or while the durable runs inside the sandbox mount. 422 tests pass (17 new, incl. a real offline durable), typecheck 0, lint 0. Co-Authored-By: Claude Opus 5 (1M context) --- .beads/export-state.json | 2 +- .../scripts/durable-filesystem-spike.ts | 5 +- .../agents/src/lib/automations/service.ts | 43 +-- packages/agents/src/lib/durable/delivery.ts | 229 ++++++++++++++++ packages/agents/src/lib/durable/index.ts | 12 + .../tests/unit/automations-service.test.ts | 10 +- .../tests/unit/durable-delivery.test.ts | 255 ++++++++++++++++++ 7 files changed, 533 insertions(+), 23 deletions(-) create mode 100644 packages/agents/src/lib/durable/delivery.ts create mode 100644 packages/agents/tests/unit/durable-delivery.test.ts diff --git a/.beads/export-state.json b/.beads/export-state.json index e9d0100..9d19f97 100644 --- a/.beads/export-state.json +++ b/.beads/export-state.json @@ -1 +1 @@ -{"last_dolt_commit":"fb1t0tec2l8cs1b4jr63mqbir0gi8ttb","timestamp":"2026-07-28T02:16:02.5366637-07:00","issues":456,"memories":0} \ No newline at end of file +{"last_dolt_commit":"bo3fi0bggs8cggdci9s9oi5na7pivcn4","timestamp":"2026-07-28T03:01:28.0906914-07:00","issues":457,"memories":0} \ No newline at end of file diff --git a/packages/agents/scripts/durable-filesystem-spike.ts b/packages/agents/scripts/durable-filesystem-spike.ts index c738c3a..ba72a05 100644 --- a/packages/agents/scripts/durable-filesystem-spike.ts +++ b/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); diff --git a/packages/agents/src/lib/automations/service.ts b/packages/agents/src/lib/automations/service.ts index f099f21..7585167 100644 --- a/packages/agents/src/lib/automations/service.ts +++ b/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 { diff --git a/packages/agents/src/lib/durable/delivery.ts b/packages/agents/src/lib/durable/delivery.ts new file mode 100644 index 0000000..12d0acd --- /dev/null +++ b/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; +} + +/** + * 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, + opts: { tenantKey?: string } = {}, +): Promise { + 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()); +} diff --git a/packages/agents/src/lib/durable/index.ts b/packages/agents/src/lib/durable/index.ts index c64f170..10eb18c 100644 --- a/packages/agents/src/lib/durable/index.ts +++ b/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, diff --git a/packages/agents/tests/unit/automations-service.test.ts b/packages/agents/tests/unit/automations-service.test.ts index 68de71c..912e2a7 100644 --- a/packages/agents/tests/unit/automations-service.test.ts +++ b/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(); diff --git a/packages/agents/tests/unit/durable-delivery.test.ts b/packages/agents/tests/unit/durable-delivery.test.ts new file mode 100644 index 0000000..4b93189 --- /dev/null +++ b/packages/agents/tests/unit/durable-delivery.test.ts @@ -0,0 +1,255 @@ +import { existsSync, mkdtempSync, readdirSync, rmSync } from "node:fs"; +import { homedir, tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + activeDurableAdapter, + durableStateDirFor, + filesystemDelivery, + findOpenLocalGate, + type LocalDurableStore, + zapierDelivery, +} from "@/lib/durable/delivery"; + +/** + * foreman-gk6k — the delivery seam must behave identically from the caller's + * point of view on both adapters, while the mechanics underneath differ + * completely (HTTP POST to a resolved URL vs. a local store call). + * + * The filesystem half runs a REAL durable against a REAL FilesystemClient in a + * temp dir: no credentials, no network, no Zapier early-access allowlist. + */ + +const HOME_EXECUTIONS = join(homedir(), ".config", "zapier-sdk", "durable", "executions"); +const homeExecutionsBefore = existsSync(HOME_EXECUTIONS) ? readdirSync(HOME_EXECUTIONS).length : 0; + +const stateDirs: string[] = []; +function tempStateDir(): string { + const dir = mkdtempSync(join(tmpdir(), "foreman-delivery-test-")); + stateDirs.push(dir); + return dir; +} + +afterAll(() => { + for (const dir of stateDirs) rmSync(dir, { recursive: true, force: true }); +}); + +describe("activeDurableAdapter", () => { + const original = process.env.ZAPIER_DURABLE_ADAPTER; + afterAll(() => { + if (original === undefined) delete process.env.ZAPIER_DURABLE_ADAPTER; + else process.env.ZAPIER_DURABLE_ADAPTER = original; + }); + + it("defaults to zapier, NOT the package's own filesystem default", () => { + delete process.env.ZAPIER_DURABLE_ADAPTER; + expect(activeDurableAdapter()).toBe("zapier"); + }); + + it("selects filesystem only on an exact match", () => { + process.env.ZAPIER_DURABLE_ADAPTER = "filesystem"; + expect(activeDurableAdapter()).toBe("filesystem"); + process.env.ZAPIER_DURABLE_ADAPTER = "Filesystem"; + expect(activeDurableAdapter()).toBe("zapier"); + }); +}); + +describe("durableStateDirFor", () => { + const original = process.env.FOREMAN_WORKSPACE_PATH; + afterAll(() => { + if (original === undefined) delete process.env.FOREMAN_WORKSPACE_PATH; + else process.env.FOREMAN_WORKSPACE_PATH = original; + }); + + it("nests durable state inside the tenant's own workspace directory", () => { + delete process.env.FOREMAN_WORKSPACE_PATH; + expect(durableStateDirFor("ws-abc")).toBe("./data/workspace/ws-abc/.durable"); + }); + + it("keeps two workspaces on disjoint paths", () => { + expect(durableStateDirFor("ws-a")).not.toBe(durableStateDirFor("ws-b")); + }); + + it("follows FOREMAN_WORKSPACE_PATH, like the agent workspace does", () => { + process.env.FOREMAN_WORKSPACE_PATH = "/srv/foreman"; + expect(durableStateDirFor("ws-abc")).toBe("/srv/foreman/ws-abc/.durable"); + }); + + it("cannot be escaped by a traversal-shaped tenant key", () => { + delete process.env.FOREMAN_WORKSPACE_PATH; + expect(durableStateDirFor("../../etc")).toBe("./data/workspace/etc/.durable"); + expect(durableStateDirFor("")).toBe("./data/workspace/_shared/.durable"); + }); +}); + +describe("findOpenLocalGate", () => { + const ops = [ + { name: "prepare", type: "step", status: "completed" }, + { name: "first-gate", type: "callback", status: "completed", callback_token: "tok-done" }, + { name: "second-gate", type: "callback", status: "pending", callback_token: "tok-open" }, + { name: "third-gate", type: "callback", status: "pending", callback_token: "tok-other" }, + ]; + const store = { getOperations: () => ops } as unknown as LocalDurableStore; + + it("ignores completed gates and steps, taking the first still-pending one", () => { + expect(findOpenLocalGate(store, "exec-1")).toEqual({ token: "tok-open", name: "second-gate" }); + }); + + it("selects by name when the run has more than one open gate", () => { + expect(findOpenLocalGate(store, "exec-1", "third-gate")).toEqual({ + token: "tok-other", + name: "third-gate", + }); + }); + + it("returns null for an unknown gate name rather than falling back", () => { + expect(findOpenLocalGate(store, "exec-1", "nope")).toBeNull(); + }); + + it("returns null when nothing is open", () => { + const closed = { getOperations: () => [ops[1]] } as unknown as LocalDurableStore; + expect(findOpenLocalGate(closed, "exec-1")).toBeNull(); + }); +}); + +describe("zapierDelivery", () => { + it("resolves the reported callback URL and POSTs the payload", async () => { + const posted: Array<{ url: string; body: unknown }> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (url: string, init: RequestInit) => { + posted.push({ url, body: JSON.parse(init.body as string) }); + return { ok: true, status: 200 } as Response; + }) as typeof fetch; + + const sdk = { + getDurableRun: async () => ({ + data: { + execution: { + operations: [ + { + name: "__report_callback_url_approve", + status: "completed", + result: { callbackUrl: "https://cb.zapier.test/abc", callbackName: "approve" }, + }, + { name: "approve", status: "pending", callback_token: "tok" }, + ], + }, + }, + }), + }; + + try { + const res = await zapierDelivery(sdk as never).deliver("run-1", { + payload: { approved: true }, + }); + expect(res).toMatchObject({ ok: true, action: "resumed", status: 200 }); + expect(posted).toEqual([{ url: "https://cb.zapier.test/abc", body: { approved: true } }]); + } finally { + globalThis.fetch = originalFetch; + } + }); + + it("reports the status the SDK returns on cancel instead of assuming cancelled", async () => { + const sdk = { cancelDurableRun: async () => ({ data: { status: "completed" } }) }; + const res = await zapierDelivery(sdk as never).deliver("run-1", { cancel: true }); + // Finished before the cancel landed: not ok, but the real status is surfaced + // so the caller persists the truth. + expect(res).toEqual({ ok: false, action: "cancelled", runStatus: "completed" }); + }); + + it("does not POST when no gate reported a URL", async () => { + const sdk = { + getDurableRun: async () => ({ data: { execution: { operations: [] } } }), + }; + const res = await zapierDelivery(sdk as never).deliver("run-1", { payload: {} }); + expect(res).toMatchObject({ ok: false, action: "none" }); + }); +}); + +describe("filesystemDelivery (real adapter, offline)", () => { + // Everything comes from the ROOT entry point. `@zapier/zapier-durable/node` + // re-exports the same names but holds SEPARATE config state, so configuring + // through it reports success while the runtime writes to the default fsDir — + // i.e. into the developer's real home directory. Measured on 0.11.0. + let createClient: typeof import("@zapier/zapier-durable").createClient; + let defineDurable: typeof import("@zapier/zapier-durable").defineDurable; + + beforeAll(async () => { + ({ createClient, defineDurable } = await import("@zapier/zapier-durable")); + }); + + async function suspendedRun(baseDir: string) { + const { configureDurable, getConfig } = await import("@zapier/zapier-durable"); + // The key is `fsDir` — an unknown key is silently ignored and falls back to + // the home default. Assert the temp dir actually took. + configureDurable({ adapter: "filesystem", fsDir: baseDir }); + expect(getConfig().fsDir).toBe(baseDir); + + const durable = defineDurable({ + name: "gk6k-approval", + run: async (ctx) => { + const [approval] = await ctx.createCallback("human-approval"); + const decision = (await approval) as { approved: boolean }; + return { approved: decision.approved }; + }, + }); + + const first = await durable({}); + expect(first.done).toBe(false); + return { durable, executionId: first.executionId as string }; + } + + it("delivers an approval to a real suspended run, which then completes", async () => { + const baseDir = tempStateDir(); + const { durable, executionId } = await suspendedRun(baseDir); + + const store = createClient() as unknown as LocalDurableStore; + const res = await filesystemDelivery(store).deliver(executionId, { + payload: { approved: true }, + }); + + // No HTTP status: there is no HTTP on this path. + expect(res).toEqual({ ok: true, action: "resumed" }); + + const resumed = await durable(executionId); + expect(resumed.done).toBe(true); + expect(resumed.result).toEqual({ approved: true }); + }); + + it("reports a clear reason when the run has no open gate", async () => { + const baseDir = tempStateDir(); + const { executionId } = await suspendedRun(baseDir); + const store = createClient() as unknown as LocalDurableStore; + + await filesystemDelivery(store).deliver(executionId, { payload: { approved: true } }); + // Second delivery — the gate is closed now. + const again = await filesystemDelivery(store).deliver(executionId, { + payload: { approved: true }, + }); + expect(again.ok).toBe(false); + expect(again.reason).toMatch(/no open callback/); + }); + + it("cancels a suspended run by releasing it as failed", async () => { + const baseDir = tempStateDir(); + const { executionId } = await suspendedRun(baseDir); + const store = createClient() as unknown as LocalDurableStore; + + const res = await filesystemDelivery(store).deliver(executionId, { cancel: true }); + expect(res).toEqual({ ok: true, action: "cancelled", runStatus: "cancelled" }); + + const client = createClient() as unknown as { + getExecution(id: string): { status: string } | null; + }; + expect(client.getExecution(executionId)?.status).toBe("failed"); + }); + + it("wrote nothing into the developer's real home directory", () => { + // Regression guard. Both known ways to get this wrong (the `filesystem: + // { baseDir }` key that does not exist, and configuring via the `/node` + // subpath) fail SILENTLY by writing to ~/.config/zapier-sdk/durable. + const home = join(homedir(), ".config", "zapier-sdk", "durable", "executions"); + const leaked = existsSync(home) ? readdirSync(home).length : 0; + expect(leaked).toBe(homeExecutionsBefore); + }); +}); From 57155c915c62b4f05690976d00daf0c718b9747c Mon Sep 17 00:00:00 2001 From: HamChowderr Date: Tue, 28 Jul 2026 03:23:09 -0700 Subject: [PATCH 2/2] feat(durable): drop the callback self-report step on the filesystem path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit foreman-2qbk (part 2). humanApprovalGate() emitted a __report_callback_url_* step for one reason: getDurableRun will not expose the callback URL, so the durable had to smuggle it out through a step result. That does not apply locally. The run is in-process, so Foreman reads the token straight off the execution's operations via findOpenLocalGate. On the filesystem adapter the gate now emits neither the reporting step nor the URL binding — an unused binding would be dead weight in generated source — while the Zapier path is byte-identical to before. Adapter comes from activeDurableAdapter() and is overridable per call, so tests pin it explicitly instead of depending on ambient env. 424 tests pass, typecheck 0, lint 0. Co-Authored-By: Claude Opus 5 (1M context) --- .beads/export-state.json | 2 +- packages/agents/src/lib/durable/source.ts | 16 +++++++++++++- .../agents/tests/unit/durable-source.test.ts | 21 +++++++++++++++++-- 3 files changed, 35 insertions(+), 4 deletions(-) diff --git a/.beads/export-state.json b/.beads/export-state.json index 9d19f97..82dc566 100644 --- a/.beads/export-state.json +++ b/.beads/export-state.json @@ -1 +1 @@ -{"last_dolt_commit":"bo3fi0bggs8cggdci9s9oi5na7pivcn4","timestamp":"2026-07-28T03:01:28.0906914-07:00","issues":457,"memories":0} \ No newline at end of file +{"last_dolt_commit":"q91afok34ortb9m4q349elh1ms8sllel","timestamp":"2026-07-28T03:15:25.5118813-07:00","issues":458,"memories":0} \ No newline at end of file diff --git a/packages/agents/src/lib/durable/source.ts b/packages/agents/src/lib/durable/source.ts index 4fd9a75..21759d1 100644 --- a/packages/agents/src/lib/durable/source.ts +++ b/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 `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)} }));`, diff --git a/packages/agents/tests/unit/durable-source.test.ts b/packages/agents/tests/unit/durable-source.test.ts index 6e21a45..028ddd1 100644 --- a/packages/agents/tests/unit/durable-source.test.ts +++ b/packages/agents/tests/unit/durable-source.test.ts @@ -63,7 +63,7 @@ describe("buildDurableSource", () => { describe("humanApprovalGate (foreman-zfnj)", () => { it("creates the gate, reports its URL+name via a step, and awaits the decision", () => { - const gate = humanApprovalGate("approve"); + const gate = humanApprovalGate("approve", "zapier"); expect(gate).toContain( 'const [approveApproval, approveUrl] = await ctx.createCallback("approve");', ); @@ -74,8 +74,25 @@ describe("humanApprovalGate (foreman-zfnj)", () => { expect(gate).toContain("const approveDecision = await approveApproval;"); }); + it("omits the report step on the filesystem adapter (foreman-2qbk)", () => { + const gate = humanApprovalGate("approve", "filesystem"); + // The gate itself is identical… + expect(gate).toContain('await ctx.createCallback("approve")'); + expect(gate).toContain("const approveDecision = await approveApproval;"); + // …but the URL never crosses a wire locally, so nothing reports it and the + // URL is not even bound (an unused binding would be dead weight in the + // generated source). Foreman reads the token off the local operations. + expect(gate).not.toContain("__report_callback_url"); + expect(gate).not.toContain("approveUrl"); + }); + + it("defaults to the active adapter, which is zapier unless opted out", () => { + delete process.env.ZAPIER_DURABLE_ADAPTER; + expect(humanApprovalGate("approve")).toContain("__report_callback_url_approve"); + }); + it("sanitizes a non-identifier name into safe variable names", () => { - const gate = humanApprovalGate("needs-sign-off"); + const gate = humanApprovalGate("needs-sign-off", "zapier"); // createCallback + callbackName keep the original name… expect(gate).toContain('ctx.createCallback("needs-sign-off")'); expect(gate).toContain('callbackName: "needs-sign-off"');