packages/forkhorse-gatekeeper/src/worker/workflow.ts

1import { evidence, outcome, uncertain, type Evidence } from "./workflow-evidence.ts";
2import { actionKinds } from "./action-kinds.ts";
3import type { ReadReviewResponse, ReadReceiptsResponse } from "../../../forkhorse-stable/src/index.ts";
4import type { ApprovalQueue, ActionKind } from "@gadgets/workshop-shared/gatekeeper";
5import { SerialTaskQueue } from "@gadgets/gatekeeper-kit/serial-queue";
6
7import type { WorkflowContext } from "../workflow-types.d.ts";
8export type { WorkflowContext } from "../workflow-types.d.ts";
9
10export type WorkflowWrite =
11  | { kind: "join" }
12  | { kind: "claim"; packet: string; repoName?: string }
13  | { kind: "requestReview"; packet: string; repoName: string; sha: string }
14  | { kind: "release"; packet: string };
15
16/** Only the public fields needed to prepare an approval card. */
17export interface WorkflowSubmission {
18  project: { id: string };
19  forks: Pick<WorkflowContext["forks"][number], "agent" | "repoName" | "createdAt">[];
20  packets: Pick<WorkflowContext["packets"][number], "id" | "mode">[];
21}
22
23/** Credential-free seam. Production implementation must use the accepted Stable provider. */
24export interface WorkflowSource {
25  authorize(): Promise<{ agent: string }>;
26  context(): Promise<WorkflowContext>;
27  submission(operation: WorkflowWrite): Promise<WorkflowSubmission>;
28  readReview(review: string): Promise<ReadReviewResponse>;
29  readReceipts(review: string): Promise<ReadReceiptsResponse>;
30  execute(operation: WorkflowWrite): Promise<void>;
31}
32interface Action {
33  operation: WorkflowWrite;
34  actionKind: ActionKind;
35  status: "preparing" | "pending" | "applying" | "retryable" | "applied" | "rejected";
36  before?: Evidence;
37  attempts?: number;
38  copy?: { title: string; description: string };
39}
40type Store = Pick<DurableObjectStorage["kv"], "get" | "put" | "list">;
41const component = (value: string) => value !== "." && value !== ".." && /^[a-zA-Z0-9._-]{1,255}$/.test(value);
42
43function validate(operation: WorkflowWrite): void {
44  if ("packet" in operation && !component(operation.packet)) throw new Error("Invalid packet ID.");
45  if ("repoName" in operation && operation.repoName !== undefined && !component(operation.repoName)) {
46    throw new Error("Invalid fork name.");
47  }
48  if (operation.kind === "requestReview" && (!/^[a-f0-9]{40}$/.test(operation.sha) || /^0+$/.test(operation.sha))) {
49    throw new Error("A full, nonzero commit ID is required.");
50  }
51}
52function sameWrite(requested: WorkflowWrite, stored: WorkflowWrite): boolean {
53  if (requested.kind !== stored.kind) return false;
54  if ("packet" in requested && "packet" in stored && requested.packet !== stored.packet) return false;
55  if (requested.kind === "claim" && stored.kind === "claim") return requested.repoName === undefined || requested.repoName === stored.repoName;
56  if (requested.kind === "requestReview" && stored.kind === "requestReview") return requested.repoName === stored.repoName && requested.sha === stored.sha;
57  return true;
58}
59function description(operation: WorkflowWrite, project: string, agent: string, race: boolean) {
60  switch (operation.kind) {
61    case "join": return { title: "Create an agent fork", description: `Create one new fork of ${project} for ${agent}.` };
62    case "claim": return race ? { title: `Enter race ${operation.packet}`, description: `Enter ${agent} in race ${operation.packet} on fork ${operation.repoName}.` } : { title: `Claim ${operation.packet}`, description: `Claim packet ${operation.packet}${operation.repoName ? ` on fork ${operation.repoName}` : " on the connected agent's latest registered fork"}.` };
63    case "requestReview": return { title: `Request review of ${operation.packet}`, description: `Request review of commit ${operation.sha.slice(0, 7)} on fork ${operation.repoName} for ${operation.packet}.` };
64    case "release": return { title: `Release ${operation.packet}`, description: `Release the connected agent's claim on packet ${operation.packet}.` };
65  }
66}
67
68/**
69 * Durable approval controller. Applied operations use the connected account
70 * and its Stable credential provider, never model-supplied authority.
71 */
72export class WorkflowActions {
73  constructor(
74    private readonly store: Store,
75    private readonly source: WorkflowSource,
76    // Share with push when integrating, including its existing nextAction counter.
77    private readonly serial = new SerialTaskQueue(),
78    private readonly beforeSubmit: () => void | Promise<void> = () => {},
79  ) {}
80
81  has(id: number): boolean { return !!this.store.get<Action>(`workflow:${id}`); }
82  hasPending(): boolean {
83    return [...this.store.list<Action>({ prefix: "workflow:" })]
84      .some(([, action]) => ["preparing", "pending", "applying", "retryable"].includes(action.status));
85  }
86  #kind(operation: WorkflowWrite, context: Pick<WorkflowSubmission, "packets">): ActionKind {
87    if (operation.kind !== "claim") return actionKinds[operation.kind];
88    const packet = context.packets.find(candidate => candidate.id === operation.packet);
89    if (!packet) throw new Error("Packet is unavailable.");
90    return packet.mode === "race" ? actionKinds.raceEntry : actionKinds.claim;
91  }
92  async #reconcile(context?: WorkflowContext): Promise<void> {
93    for (const [key, action] of this.store.list<Action>({ prefix: "workflow:" })) {
94      if (action.status !== "applying" && action.status !== "retryable") continue;
95      const identity = await this.source.authorize();
96      let snapshot: WorkflowContext;
97      try { snapshot = context ?? await this.source.context(); } catch { throw uncertain(action.operation); }
98      await this.source.authorize();
99      if (!action.before || identity.agent !== action.before.agent) continue;
100      const result = outcome(snapshot, action.before, action.operation, action.actionKind.tag === actionKinds.raceEntry.tag);
101      if (result === "applied") this.store.put(key, { ...action, status: "applied" });
102      else this.store.put(key, { ...action, status: result === "absent" && (action.attempts ?? 1) < 2 ? "retryable" : "applying" });
103    }
104  }
105  // Caller must already hold the shared push/workflow serial queue.
106  async reconcileInsideQueue(): Promise<void> { await this.#reconcile(); }
107  async readReview(queue: Pick<ApprovalQueue, "authorizeObservation">, review: string): Promise<ReadReviewResponse> {
108    await queue.authorizeObservation({ title: "Read review", description: "Read the connected agent's review at its recorded commit." });
109    await this.source.authorize();
110    await this.serial.run(() => this.#reconcile());
111    const result = await this.source.readReview(review);
112    await this.source.authorize();
113    return result;
114  }
115  async readReceipts(queue: Pick<ApprovalQueue, "authorizeObservation">, review: string): Promise<ReadReceiptsResponse> {
116    await queue.authorizeObservation({ title: "Read receipt summary", description: "Read the public check summary on the connected agent's review." });
117    await this.source.authorize();
118    await this.serial.run(() => this.#reconcile());
119    const result = await this.source.readReceipts(review);
120    await this.source.authorize();
121    return result;
122  }
123
124  async context(queue: Pick<ApprovalQueue, "authorizeObservation">): Promise<WorkflowContext> {
125    await queue.authorizeObservation({ title: "Read project context", description: "Read project state and the connected agent's work to choose the next step." });
126    return this.serial.run(async () => {
127      await this.source.authorize();
128      const result = await this.source.context();
129      await this.source.authorize();
130      await this.#reconcile(result);
131      return result;
132    });
133  }
134
135  async submit(queue: Pick<ApprovalQueue, "submitAction">, operation: WorkflowWrite): Promise<void> {
136    validate(operation);
137    await this.serial.run(async () => {
138      const identity = await this.source.authorize();
139      await this.beforeSubmit();
140      const prior = [...this.store.list<Action>({ prefix: "workflow:" })].find(([, action]) => action.status === "applying" || action.status === "retryable");
141      await this.#reconcile();
142      if (prior && this.store.get<Action>(prior[0])?.status === "applied" && sameWrite(operation, prior[1].operation)) return;
143      const unresolved = [...this.store.list<Action>({ prefix: "workflow:" })].find(([, action]) => ["preparing", "pending", "applying", "retryable"].includes(action.status));
144      let id: number;
145      let action: Action;
146      if (unresolved) {
147        const [key, existing] = unresolved;
148        if (existing.status === "applying") throw uncertain(existing.operation);
149        // Omitted fork is resolved once, before approval, and stays fixed on retry.
150        if (!sameWrite(operation, existing.operation)) throw new Error("Resolve the pending workflow action before submitting another.");
151        if (existing.status === "pending") return;
152        id = Number(key.slice("workflow:".length));
153        action = existing;
154      } else {
155        const snapshot = await this.source.submission(operation);
156        await this.source.authorize();
157        if (operation.kind === "claim" && !operation.repoName) {
158          const forks = snapshot.forks.filter(fork => fork.agent === identity.agent);
159          const latest = forks.toSorted((a, b) => b.createdAt.localeCompare(a.createdAt))[0];
160          if (!latest) throw new Error("The connected agent has no fork to claim.");
161          operation = { ...operation, repoName: latest.repoName };
162        }
163        const actionKind = this.#kind(operation, snapshot);
164        id = (this.store.get<number>("nextAction") ?? 0) + 1;
165        this.store.put("nextAction", id);
166        action = { operation, actionKind, status: "preparing", copy: description(operation, snapshot.project.id, identity.agent, actionKind.tag === actionKinds.raceEntry.tag) };
167      }
168      this.store.put(`workflow:${id}`, { ...action, status: "preparing" });
169      try {
170        await queue.submitAction(id, { ...action.copy!, implementsRevert: false, awaitDecision: true, actionKind: action.actionKind, autoApprovable: true });
171      } catch {
172        throw new Error("Workflow submission interrupted. Retry the same call to recover its approval request.");
173      }
174      this.store.put(`workflow:${id}`, { ...action, status: "pending" });
175    });
176  }
177
178  async apply(id: number): Promise<void> {
179    await this.serial.run(async () => {
180      let action = this.store.get<Action>(`workflow:${id}`);
181      if (action && action.status !== "applied" && action.status !== "rejected" && (action.attempts ?? 0) > 0) {
182        // Recheck even a previously observed absence immediately before retrying.
183        this.store.put(`workflow:${id}`, { ...action, status: "applying" });
184        await this.#reconcile();
185        action = this.store.get<Action>(`workflow:${id}`);
186      }
187      if (action?.status === "applied") return;
188      if (action?.status === "applying") throw uncertain(action.operation);
189      if (!action || !["preparing", "pending", "retryable"].includes(action.status)) throw new Error("Workflow action is not pending approval.");
190      const identity = await this.source.authorize();
191      const snapshot = await this.source.context();
192      await this.source.authorize();
193      if (this.#kind(action.operation, snapshot).tag !== action.actionKind.tag) throw new Error("Packet action kind changed; discard this action.");
194      if (action.before && (action.attempts ?? 0) > 0) {
195        const result = outcome(snapshot, action.before, action.operation, action.actionKind.tag === actionKinds.raceEntry.tag);
196        if (result === "applied") { this.store.put(`workflow:${id}`, { ...action, status: "applied" }); return; }
197        if (result !== "absent" || (action.attempts ?? 0) >= 2) throw uncertain(action.operation);
198      }
199      const before = evidence(snapshot, identity.agent, action.operation);
200      const sent: Action = { ...action, before, attempts: (action.attempts ?? 0) + 1, status: "applying" };
201      this.store.put(`workflow:${id}`, sent);
202      try {
203        await this.source.execute(action.operation);
204      } catch {
205        throw uncertain(action.operation);
206      }
207      this.store.put(`workflow:${id}`, { ...sent, status: "applied" });
208    });
209  }
210
211  async reject(id: number): Promise<void> {
212    await this.serial.run(async () => {
213      await this.#reconcile();
214      const action = this.store.get<Action>(`workflow:${id}`);
215      if (!action || action.status === "rejected") return;
216      if (action.status === "applied" || action.status === "applying") throw new Error("Workflow action was already sent; inspect context.");
217      this.store.put(`workflow:${id}`, { ...action, status: "rejected" });
218    });
219  }
220}
221