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

1import { forkAddress, type ReadReviewResponse, type ReadReceiptsResponse } from "../../../forkhorse-stable/src/index.ts";
2import { WorkflowActions } from "./workflow.ts";
3import { WorkflowSession } from "./workflow-session.ts";
4import { actionKinds } from "./action-kinds.ts";
5import { DurableObject, RpcStub } from "cloudflare:workers";
6import { validateRpc } from "capnweb-validate";
7import { SerialTaskQueue } from "@gadgets/gatekeeper-kit/serial-queue";
8import { validateBranchName } from "@gadgets/gatekeeper-kit/git-transport";
9import type {
10  ApprovalQueue, Gatekeeper, GatekeeperUserVerifier, GitCache, GitOid, GitPullHints, ResourceDescription, ActionKind,
11} from "@gadgets/workshop-shared/gatekeeper";
12import type { AgentForkSession } from "../fork-types.d.ts";
13import type { AgentVerifier, Env, UserAccount } from "./account.ts";
14import { oid } from "./core-client.ts";
15import { TYPES } from "./api-types.ts";
16
17interface Props { accountId: string; repoName: string }
18interface Push { branch: string; oldSha: string; newSha: string; status: "preparing" | "pending" | "applying" | "retryable" | "applied" | "rejected"; attempts?: number }
19
20@validateRpc()
21export class ForkGatekeeper extends DurableObject<Env, Props> implements Gatekeeper<AgentForkSession> {
22  readonly #serial = new SerialTaskQueue();
23  #account(): DurableObjectStub<UserAccount> {
24    return this.ctx.exports.UserAccount.get(this.ctx.exports.UserAccount.idFromString(this.ctx.props.accountId));
25  }
26  #workflow(): WorkflowActions {
27    const account = this.#account();
28    return new WorkflowActions(this.ctx.storage.kv, {
29      authorize: async () => { await account.fork(this.ctx.props.repoName); return account.identity(); },
30      context: () => account.workflowContext(),
31      submission: operation => account.workflowSubmission(operation),
32      execute: operation => account.executeWorkflow(operation),
33      readReview: (review): Promise<ReadReviewResponse> => account.readReview(review),
34      readReceipts: (review): Promise<ReadReceiptsResponse> => account.readReceipts(review),
35    }, this.#serial, async () => {
36      // Only uncertain pushes need head/state evidence. Keep the conflict fence
37      // for every pending push without loading /state on a fresh workflow write.
38      if ([...this.ctx.storage.kv.list<Push>({ prefix: "push:" })].some(([, action]) => ["applying", "retryable"].includes(action.status))) {
39        this.#reconcilePush(await this.readHead());
40      }
41      if ([...this.ctx.storage.kv.list<Push>({ prefix: "push:" })].some(([, action]) => ["preparing", "pending", "applying", "retryable"].includes(action.status))) {
42        throw new Error("Resolve the pending push before submitting a workflow action.");
43      }
44    });
45  }
46  async describe(): Promise<ResourceDescription> {
47    const source = this.#account();
48    await source.fork(this.ctx.props.repoName);
49    const meta = await source.metadata();
50    const url = forkAddress({ origin: meta.coreOrigin, project: meta.project, repoName: this.ctx.props.repoName });
51    return { url,
52      title: this.ctx.props.repoName, snippet: `Agent-owned Git fork. Attach this exact address: ${url}`, tsType: "AgentForkSession", suggestedBindingName: "FORK" };
53  }
54  async getTypeScriptTypes(): Promise<string> { return TYPES; }
55  async getAutoApprovableActions(): Promise<ActionKind[]> { return Object.values(actionKinds); }
56  async startSession(queue: RpcStub<ApprovalQueue>): Promise<AgentForkSession> {
57    await this.#account().fork(this.ctx.props.repoName);
58    return new ForkSession(this, queue.dup(), this.#workflow());
59  }
60  async readHead(): Promise<{ repoName: string; branch: string; commitId: string }> {
61    const source = this.#account();
62    const fork = await source.fork(this.ctx.props.repoName);
63    const state = await source.state();
64    return { repoName: fork.repoName, branch: state.project.defaultBranch, commitId: fork.head };
65  }
66  #reconcilePush(head: { branch: string; commitId: string }): void {
67    for (const [key, action] of this.ctx.storage.kv.list<Push>({ prefix: "push:" })) {
68      if (action.status !== "applying" && action.status !== "retryable") continue;
69      if (head.branch !== action.branch) { this.ctx.storage.kv.put(key, { ...action, status: "applying" }); continue; }
70      if (head.commitId === action.newSha) this.ctx.storage.kv.put(key, { ...action, status: "applied" });
71      else this.ctx.storage.kv.put(key, { ...action, status: head.commitId === action.oldSha && (action.attempts ?? 1) < 2 ? "retryable" : "applying" });
72    }
73  }
74  async readAndReconcileHead(): Promise<{ repoName: string; branch: string; commitId: string }> {
75    return this.#serial.run(async () => {
76      const head = await this.readHead();
77      this.#reconcilePush(head);
78      return head;
79    });
80  }
81  async submitPush(queue: RpcStub<ApprovalQueue>, branch: string, commitId: string): Promise<void> {
82    await this.#serial.run(async () => {
83      await this.#workflow().reconcileInsideQueue();
84      if (this.#workflow().hasPending()) throw new Error("Resolve the pending workflow action before pushing.");
85      validateBranchName(branch);
86      if (!oid(commitId) || /^0+$/.test(commitId)) throw new Error("A full, nonzero commit ID is required.");
87      const head = await this.readHead();
88      this.#reconcilePush(head);
89      if (branch !== head.branch) throw new Error("Only the fork's default branch is supported.");
90      if (head.commitId === commitId) return;
91      const actions = [...this.ctx.storage.kv.list<Push>({ prefix: "push:" })];
92      const interrupted = actions.find(([, a]) => a.status === "preparing" || a.status === "retryable");
93      if (interrupted) {
94        const [key, action] = interrupted;
95        if (action.branch !== branch || action.newSha !== commitId) {
96          throw new Error("Retry the interrupted push with the same branch and commit before submitting another.");
97        }
98        // Reuse the gatekeeper action ID after an ambiguous reply/crash. The OS may
99        // queue a duplicate, but applying this ID is serialized and idempotent.
100        await this.#queuePush(queue, Number(key.slice(5)), action, head.repoName);
101        return;
102      }
103      if (actions.some(([, a]) => a.status === "applying")) throw new Error("Push outcome uncertain. Check the fork head against the approved commit; this call will not blindly resend it.");
104      if (actions.some(([, a]) => a.status === "pending")) {
105        throw new Error("Resolve the pending push before submitting another.");
106      }
107      using cache = await queue.getGitCache();
108      if (!/^0+$/.test(head.commitId) && !await cache.isAncestor(head.commitId, commitId)) {
109        throw new Error("Push must be a fast-forward. Rebase on the current fork head.");
110      }
111      const id = (this.ctx.storage.kv.get<number>("nextAction") ?? 0) + 1;
112      this.ctx.storage.kv.put("nextAction", id);
113      const action: Push = { branch, oldSha: head.commitId, newSha: commitId, status: "preparing" };
114      this.ctx.storage.kv.put(`push:${id}`, action);
115      await this.#queuePush(queue, id, action, head.repoName);
116    });
117  }
118  async #queuePush(queue: RpcStub<ApprovalQueue>, id: number, action: Push, repoName: string): Promise<void> {
119    try {
120      await queue.submitAction(id, {
121        title: `Push ${action.newSha.slice(0, 7)} to ${action.branch}`,
122        description: `Move ${action.branch} on fork ${repoName} from ${action.oldSha.slice(0, 7)} to ${action.newSha.slice(0, 7)}. This fails if the branch moved.`,
123        pushedCommits: [action.newSha], implementsRevert: false, awaitDecision: true,
124        actionKind: actionKinds.push, autoApprovable: true,
125      });
126      this.ctx.storage.kv.put(`push:${id}`, { ...action, status: "pending" });
127    } catch {
128      // Keep the intent: the OS may have durably queued it before the reply failed.
129      throw new Error("Push submission was interrupted. Retry the same branch and commit.");
130    }
131  }
132  async gitPull(oids: GitOid[], cache: RpcStub<GitCache>, hints: GitPullHints): Promise<void> {
133    await this.#account().pull(this.ctx.props.repoName, oids, hints, cache);
134  }
135  async applyAction(id: number, cache: RpcStub<GitCache>): Promise<void> {
136    const workflow = this.#workflow();
137    if (workflow.has(id)) return workflow.apply(id);
138    await this.#serial.run(async () => {
139      const head = await this.readHead();
140      this.#reconcilePush(head);
141      const action = this.ctx.storage.kv.get<Push>(`push:${id}`);
142      if (action?.status === "applied") return;
143      if (action?.status === "applying") throw new Error("Push outcome uncertain. Check the fork head against the approved commit; this call will not blindly resend it.");
144      // A queued OS action is still approvable if its submission reply was lost.
145      if (action?.status !== "pending" && action?.status !== "preparing" && action?.status !== "retryable") throw new Error("Push is not pending approval.");
146      // Recover a lost successful reply; the approved exact commit is already present.
147      if (head.commitId !== action.newSha) {
148        if (head.branch !== action.branch || head.commitId !== action.oldSha) throw new Error("Fork moved; discard this push and rebase.");
149        if (!/^0+$/.test(action.oldSha) && !await cache.isAncestor(action.oldSha, action.newSha)) {
150          throw new Error("Push is not a fast-forward.");
151        }
152        const pack = await cache.buildPack();
153        this.ctx.storage.kv.put(`push:${id}`, { ...action, status: "applying", attempts: (action.attempts ?? 0) + 1 });
154        await this.#account().push(this.ctx.props.repoName, action.branch, action.oldSha, action.newSha, pack);
155      }
156      this.ctx.storage.kv.put(`push:${id}`, { ...action, status: "applied" });
157    });
158  }
159  async rejectAction(id: number): Promise<void> {
160    const workflow = this.#workflow();
161    if (workflow.has(id)) return workflow.reject(id);
162    await this.#serial.run(async () => {
163      this.#reconcilePush(await this.readHead());
164      const action = this.ctx.storage.kv.get<Push>(`push:${id}`);
165      if (!action || action.status === "rejected") return;
166      if (action.status === "applied" || action.status === "applying") throw new Error("Push already sent; inspect the fork head.");
167      this.ctx.storage.kv.put(`push:${id}`, { ...action, status: "rejected" });
168    });
169  }
170  async revertAction(): Promise<never> { throw new Error("Create and submit a new reverting commit; history rewrites are not supported."); }
171  async addObserver(_id: string, user: Fetcher<GatekeeperUserVerifier>): Promise<void> {
172    const meta = await this.#account().metadata();
173    if (!await (user as Fetcher<AgentVerifier>).owns(meta.coreOrigin, meta.project, this.ctx.props.repoName)) {
174      throw new Error("Observer cannot access this agent fork.");
175    }
176  }
177  async removeObserver(_id: string): Promise<void> {}
178}
179
180@validateRpc()
181class ForkSession extends WorkflowSession implements AgentForkSession {
182  constructor(private readonly gatekeeper: ForkGatekeeper, queue: RpcStub<ApprovalQueue>, actions: WorkflowActions) { super(actions, queue); }
183  async getHead(): Promise<{ repoName: string; branch: string; commitId: string }> {
184    await this.queue.authorizeObservation({ title: "Read fork head", description: "Read the connected fork's default branch and commit." });
185    const head = await this.gatekeeper.readAndReconcileHead();
186    using cache = await this.queue.getGitCache();
187    if (!/^0+$/.test(head.commitId)) await cache.advertiseCommit(head.commitId);
188    return head;
189  }
190  async push(branch: string, commitId: string): Promise<void> { await this.gatekeeper.submitPush(this.queue, branch, commitId); }
191  [Symbol.dispose](): void { this.queue[Symbol.dispose](); }
192}
193