packages/forkhorse-gatekeeper/__tests__/workflow.test.ts

1import { expect, it, vi } from "vitest";
2import { WorkflowActions, type WorkflowContext, type WorkflowWrite } from "../src/worker/workflow.ts";
3import { workflowSource } from "../src/worker/workflow-source.ts";
4
5// Local workerd fixture only. Production provider integration is covered by vendor tests.
6function fixture() {
7  const data = new Map<string, unknown>();
8  const store = {
9    get<T>(key: string) { return data.get(key) as T | undefined; },
10    put<T>(key: string, value: T) { data.set(key, structuredClone(value)); },
11    list<T>({ prefix }: { prefix?: string } = {}) {
12      return new Map([...data].filter(([key]) => !prefix || key.startsWith(prefix))) as Map<string, T>;
13    },
14  };
15  const context = { project: { id: "fixture" }, packets: [{ id: "fixture-packet", mode: "standard" }], myWork: [], forks: [{ repoName: "fixture-fork", agent: "fixture-agent", createdAt: "2026-01-01" }], claims: [], races: [], reviews: [] } as unknown as WorkflowContext;
16  const identity = vi.fn(async () => ({ agent: "fixture-agent", work: context.myWork, forks: context.forks.filter(fork => fork.agent === "fixture-agent") }));
17  const request = vi.fn(async (_path: string, _init?: RequestInit): Promise<unknown> => context);
18  const source = workflowSource({ project: "fixture", identity, request });
19  const queue = { authorizeObservation: vi.fn(async () => {}), submitAction: vi.fn(async (_id: number, _description: unknown) => {}) };
20  const actions = new WorkflowActions(store as unknown as DurableObjectStorage["kv"], source);
21  return { data, store, context, identity, request, source, queue, actions };
22}
23const cases: [WorkflowWrite, string, string, unknown][] = [
24  [{ kind: "join" }, "/join", "POST", { agent: "fixture-agent", ttlSeconds: 60 }],
25  [{ kind: "claim", packet: "fixture-packet", repoName: "fixture-fork" }, "/packets/fixture-packet/claim", "POST", { repoName: "fixture-fork" }],
26  [{ kind: "claim", packet: "fixture-packet" }, "/packets/fixture-packet/claim", "POST", { repoName: "fixture-fork" }],
27  [{ kind: "requestReview", packet: "fixture-packet", repoName: "fixture-fork", sha: "a".repeat(40) }, "/reviews", "POST", { packet: "fixture-packet", repoName: "fixture-fork", sha: "a".repeat(40) }],
28  [{ kind: "release", packet: "fixture-packet" }, "/packets/fixture-packet/claim", "DELETE", undefined],
29];
30
31it("authorizes context as an observation, retaining the core resume shape", async () => {
32  const f = fixture();
33  expect(await f.actions.context(f.queue)).toEqual(f.context);
34  expect(f.queue.authorizeObservation).toHaveBeenCalledOnce();
35  expect(f.queue.submitAction).not.toHaveBeenCalled();
36  expect(f.request).toHaveBeenCalledExactlyOnceWith("/state");
37  expect(f.identity).toHaveBeenCalledTimes(3);
38});
39
40for (const [operation, path, method, body] of cases) {
41  it(`queues ${operation.kind}${"repoName" in operation ? " with a fork" : ""} and sends the exact core route only on apply`, async () => {
42    const f = fixture();
43    const sentinel = "fixture-token-must-never-escape";
44    f.request.mockImplementation(async (_path, init) => init ? { token: sentinel, repoName: "new-fixture-fork" } : f.context);
45    const log = vi.spyOn(console, "log");
46    const error = vi.spyOn(console, "error");
47    try {
48      expect(await f.actions.submit(f.queue, operation)).toBeUndefined();
49      expect(f.request.mock.calls.filter(([, init]) => init)).toEqual([]);
50      const [id, description] = f.queue.submitAction.mock.calls[0]!;
51      expect(description).toMatchObject({ implementsRevert: false, awaitDecision: true });
52      expect(description).toHaveProperty("autoApprovable", true);
53      expect(description).toHaveProperty("actionKind");
54      expect(await f.actions.apply(id)).toBeUndefined();
55      expect(f.request.mock.calls.filter(([, init]) => init)).toEqual([[path, {
56        method, ...(body ? { body: JSON.stringify(body) } : {}),
57      }]]);
58      await f.actions.apply(id);
59      expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
60      expect(JSON.stringify([...f.data])).not.toContain(sentinel);
61      expect(JSON.stringify(f.queue.submitAction.mock.calls)).not.toContain(sentinel);
62      expect(log).not.toHaveBeenCalled();
63      expect(error).not.toHaveBeenCalled();
64    } finally { log.mockRestore(); error.mockRestore(); }
65  });
66}
67
68it("recovers a lost submission with the same action ID, including after controller recreation", async () => {
69  const f = fixture();
70  f.queue.submitAction.mockRejectedValueOnce(new Error("fixture-secret-in-error"));
71  await expect(f.actions.submit(f.queue, { kind: "join" })).rejects.toThrow("submission interrupted");
72  const resumed = new WorkflowActions(f.store as unknown as DurableObjectStorage["kv"], f.source);
73  await resumed.submit(f.queue, { kind: "join" });
74  await resumed.submit(f.queue, { kind: "join" });
75  expect(f.queue.submitAction.mock.calls.map(([id]) => id)).toEqual([1, 1]);
76  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(0);
77  await resumed.apply(1);
78  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
79});
80
81it("can approve an action whose queue reply was lost", async () => {
82  const f = fixture();
83  f.queue.submitAction.mockRejectedValueOnce(new Error("fixture-lost-reply"));
84  await expect(f.actions.submit(f.queue, { kind: "join" })).rejects.toThrow("interrupted");
85  await f.actions.apply(1);
86  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
87});
88
89it("keeps the fence when context is unavailable after a lost response", async () => {
90  const f = fixture();
91  await f.actions.submit(f.queue, { kind: "join" });
92  f.request.mockImplementation(async (_path, init) => { if (init) throw new Error("fixture-secret"); return f.context; });
93  await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
94  f.request.mockRejectedValue(new Error("fixture-secret"));
95  const resumed = new WorkflowActions(f.store as unknown as DurableObjectStorage["kv"], f.source);
96  await expect(resumed.apply(1)).rejects.toThrow("outcome uncertain");
97  await expect(resumed.submit(f.queue, { kind: "join" })).rejects.toThrow("outcome uncertain");
98  await expect(resumed.reject(1)).rejects.toThrow("outcome uncertain");
99  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
100  expect(JSON.stringify([...f.data])).not.toContain("fixture-secret");
101});
102
103const claim = { packet: "fixture-packet", agent: "fixture-agent", repoName: "fixture-fork", createdAt: "2026-01-01", paths: [], usage: { reported: null, metered: null } };
104for (const kind of ["claim", "race", "join", "requestReview", "release"] as const) {
105  function setup() {
106    const f = fixture();
107    const operation: WorkflowWrite = kind === "race" || kind === "claim" ? { kind: "claim", packet: "fixture-packet", repoName: "fixture-fork" } : kind === "release" ? { kind, packet: "fixture-packet" } : kind === "requestReview" ? { kind, packet: "fixture-packet", repoName: "fixture-fork", sha: "a".repeat(40) } : { kind };
108    if (kind === "race") {
109      f.context.packets[0]!.mode = "race";
110      f.context.races = [{ packet: "fixture-packet", entrants: [] }] as unknown as WorkflowContext["races"];
111    }
112    if (kind === "release") f.context.claims.push(claim);
113    function applied() {
114      if (kind === "claim") f.context.claims.push(claim);
115      if (kind === "race") f.context.races[0]!.entrants.push({ ...claim, claimedAt: claim.createdAt, closedAt: null } as unknown as WorkflowContext["races"][number]["entrants"][number]);
116      if (kind === "join") f.context.forks.push({ repoName: "fixture-new-fork", agent: "fixture-agent" } as WorkflowContext["forks"][number]);
117      if (kind === "requestReview") f.context.reviews.push({ ...claim, sha: "a".repeat(40) } as unknown as WorkflowContext["reviews"][number]);
118      if (kind === "release") f.context.claims = [];
119    }
120    return { ...f, operation, applied };
121  }
122  it(`reconciles an applied ${kind} from context after a lost reply without another write`, async () => {
123    const f = setup();
124    await f.actions.submit(f.queue, f.operation);
125    f.request.mockImplementation(async (_path, init) => { if (init) { f.applied(); throw new Error("fixture-lost-reply"); } return f.context; });
126    await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
127    const resumed = new WorkflowActions(f.store as unknown as DurableObjectStorage["kv"], f.source);
128    await resumed.context(f.queue);
129    expect(resumed.hasPending()).toBe(false);
130    await resumed.apply(1);
131    expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
132  });
133  it(`allows only one evidence-based retry of an absent ${kind}`, async () => {
134    const f = setup();
135    await f.actions.submit(f.queue, f.operation);
136    f.request.mockImplementation(async (_path, init) => { if (init) throw new Error("fixture-not-applied"); return f.context; });
137    await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
138    await f.actions.context(f.queue);
139    await f.actions.submit(f.queue, f.operation);
140    await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
141    await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
142    expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(2);
143    expect(f.actions.hasPending()).toBe(true);
144    f.applied();
145    await f.actions.context(f.queue);
146    expect(f.actions.hasPending()).toBe(false);
147  });
148}
149
150it("does not mistake existing or multiple newly discovered forks for one successful join", async () => {
151  const f = fixture();
152  await f.actions.submit(f.queue, { kind: "join" });
153  f.request.mockImplementation(async (_path, init) => { if (init) throw new Error("fixture-lost-reply"); return f.context; });
154  await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
155  f.context.forks.push(...["new-one", "new-two"].map(repoName => ({ repoName, agent: "fixture-agent" } as WorkflowContext["forks"][number])));
156  await f.actions.context(f.queue);
157  await expect(f.actions.apply(1)).rejects.toThrow("outcome uncertain");
158  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
159});
160
161it("denial prevents writes and revoked authorization is rechecked on apply", async () => {
162  const f = fixture();
163  await f.actions.submit(f.queue, { kind: "join" });
164  await f.actions.reject(1);
165  await expect(f.actions.apply(1)).rejects.toThrow("not pending");
166  await f.actions.submit(f.queue, { kind: "release", packet: "fixture" });
167  f.identity.mockRejectedValue(new Error("Fixture revoked"));
168  await expect(f.actions.apply(2)).rejects.toThrow("revoked");
169  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(0);
170});
171
172it("refuses conflicting pending actions and malformed paths or commit IDs before queuing", async () => {
173  const f = fixture();
174  for (const operation of [
175    { kind: "claim", packet: "../owner" },
176    { kind: "claim", packet: ".." },
177    { kind: "claim", packet: "fixture", repoName: "other/fork" },
178    { kind: "requestReview", packet: "fixture", repoName: "fixture", sha: "0".repeat(40) },
179  ] as WorkflowWrite[]) await expect(f.actions.submit(f.queue, operation)).rejects.toThrow();
180  expect(f.queue.submitAction).not.toHaveBeenCalled();
181  await f.actions.submit(f.queue, { kind: "join" });
182  await expect(f.actions.submit(f.queue, { kind: "release", packet: "fixture" })).rejects.toThrow("pending workflow");
183  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(0);
184});
185
186it("uses authenticated /me work instead of any untrusted state summary", async () => {
187  const f = fixture();
188  f.request.mockResolvedValue({ project: { id: "fixture" }, myWork: ["foreign-work"] });
189  expect((await f.actions.context(f.queue)).myWork).toEqual([]);
190});
191
192it("denied observations do not read and revocation during a context read prevents disclosure", async () => {
193  const f = fixture();
194  f.queue.authorizeObservation.mockRejectedValueOnce(new Error("Fixture denied"));
195  await expect(f.actions.context(f.queue)).rejects.toThrow("denied");
196  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(0);
197  f.request.mockImplementationOnce(async () => {
198    f.identity.mockRejectedValue(new Error("Fixture revoked during read"));
199    return f.context;
200  });
201  await expect(f.actions.context(f.queue)).rejects.toThrow("revoked during read");
202});
203
204
205it("serializes reconciliation behind an in-flight write and retains its pre-write evidence", async () => {
206  const f = fixture();
207  let complete!: () => void;
208  let started!: () => void;
209  const inFlight = new Promise<void>(resolve => { started = resolve; });
210  const reply = new Promise<void>(resolve => { complete = resolve; });
211  await f.actions.submit(f.queue, { kind: "join" });
212  f.request.mockImplementation(async (_path, init) => {
213    if (init) {
214      started(); await reply;
215      f.context.forks.push({ repoName: "new-fork", agent: "fixture-agent" } as WorkflowContext["forks"][number]);
216      throw new Error("Fixture lost reply");
217    }
218    return structuredClone(f.context);
219  });
220  const applying = f.actions.apply(1).catch(error => error);
221  await inFlight;
222  const reading = f.actions.context(f.queue);
223  complete();
224  expect(String(await applying)).toContain("outcome uncertain");
225  await reading;
226  expect(f.actions.hasPending()).toBe(false);
227  expect(f.request.mock.calls.filter(([, init]) => init)).toHaveLength(1);
228});
229
230it("keeps a claim's resolved fork fixed while waiting for approval", async () => {
231  const f = fixture();
232  await f.actions.submit(f.queue, { kind: "claim", packet: "fixture-packet" });
233  f.context.forks.push({ repoName: "newer-fork", agent: "fixture-agent", createdAt: "2026-02-01" } as WorkflowContext["forks"][number]);
234  await f.actions.apply(1);
235  const write = f.request.mock.calls.find(([, init]) => init)!;
236  expect(JSON.parse(String(write[1]!.body))).toEqual({ repoName: "fixture-fork" });
237});
238
239it("prepares a claim from /packets and the caller's latest fork without reading full context", async () => {
240  const f = fixture();
241  f.context.forks.unshift({ repoName: "latest-fork", agent: "fixture-agent", createdAt: "2026-03-01" } as WorkflowContext["forks"][number]);
242  f.context.forks.push({ repoName: "foreign-newer-fork", agent: "other-agent", createdAt: "2026-04-01" } as WorkflowContext["forks"][number]);
243  await f.actions.submit(f.queue, { kind: "claim", packet: "fixture-packet" });
244  expect(f.request).toHaveBeenCalledExactlyOnceWith("/packets");
245  expect(f.queue.submitAction.mock.calls[0]![1]).toMatchObject({
246    title: "Claim fixture-packet", description: "Claim packet fixture-packet on fork latest-fork.",
247  });
248});
249