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

1import { forkAddress, storedAgentKeyProvider } from "../../../forkhorse-stable/src/index.ts";
2import type { WorkflowWrite } from "./workflow.ts";
3import { DurableObject, RpcStub, RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
4import type {
5  AccountDescription, AppUiContext, ConnectHandoff, Gatekeeper, GatekeeperConnectCallback,
6  GatekeeperUiFrame, GatekeeperUser, GatekeeperUserVerifier, ResourceConfiguratorFrame, SupportedResource,
7} from "@gadgets/workshop-shared/gatekeeper";
8import { generateNonce, isLiveNonce, type TimedNonce } from "@gadgets/gatekeeper-kit/connect-nonce";
9import { isConnectAttempted, markConnectAttempted } from "@gadgets/gatekeeper-kit/connect-handshake";
10import { stageCredentials, commitStagedCredentials } from "@gadgets/gatekeeper-kit/credential-stage";
11import { pullGitObjectsIntoCache, pushGitRefUpdate } from "@gadgets/gatekeeper-kit/git-transport";
12import type { GitCache, GitOid, GitPullHints } from "@gadgets/workshop-shared/gatekeeper";
13import { validateRpc } from "capnweb-validate";
14import { CoreClient } from "./core-client.ts";
15import { checkConnectionDraft, type ConnectionDraft } from "../configuration.ts";
16import { createSidebarFrame, sidebarAccountDescription } from "./sidebar.ts";
17import { sidebarWithCoreOrigin } from "../../app/owner-origin.ts";
18import type { ProjectState } from "../../../forkhorse-dashboard/src/contracts.ts";
19import SIDEBAR_HTML from "../../dist/sidebar.html";
20import PICKER_HTML from "../../dist/picker.html";
21import brand from "../../../../brand.json";
22import { markIconUrl } from "../../../forkhorse-dashboard/src/mark.ts";
23
24export interface Env extends Cloudflare.Env { BASE_URL: string }
25interface Credentials extends ConnectionDraft { agent: string }
26interface AccountProps { accountId: string }
27export const RESOURCE: SupportedResource = {
28  urlPattern: "https://:host/projects/:project/forks/:repoName",
29  title: "Agent fork", description: "A Git fork owned by the connected agent.",
30  icon: { url: markIconUrl(brand.mark) },
31};
32
33declare global {
34  namespace Cloudflare {
35    interface GlobalProps {
36      mainModule: typeof import("./index.ts");
37      durableNamespaces: "UserAccount" | "ForkGatekeeper";
38    }
39  }
40}
41
42/** Private credential store. Only account/fork capabilities carry its unguessable object id. */
43export class UserAccount extends DurableObject<Env> {
44  #generation = 0;
45  #client(): CoreClient {
46    const generation = this.#generation;
47    const creds = this.ctx.storage.kv.get<Credentials>("credentials");
48    if (!creds) throw new Error("Agent connection unavailable.");
49    return new CoreClient(creds, creds.agent, () => {
50      if (generation !== this.#generation || !this.ctx.storage.kv.get("credentials")) {
51        throw new Error("Agent connection unavailable.");
52      }
53    }, storedAgentKeyProvider(() => this.ctx.storage.kv.get<Credentials>("credentials")));
54  }
55  async initiate(callback: Fetcher<GatekeeperConnectCallback>, nonce: string): Promise<void> {
56    if (this.ctx.storage.kv.get("callback")) throw new Error("Connection already initiated.");
57    this.ctx.storage.kv.put("callback", callback);
58    this.ctx.storage.kv.put("nonce", { value: nonce, expiresAt: Date.now() + 600_000 } satisfies TimedNonce);
59    await this.ctx.storage.setAlarm(Date.now() + 3_600_000);
60  }
61  async validNonce(nonce: string): Promise<boolean> {
62    return isLiveNonce(this.ctx.storage.kv.get<TimedNonce>("nonce"), nonce, Date.now());
63  }
64  async completeConnection(nonce: string, draft: ConnectionDraft): Promise<ConnectHandoff> {
65    // Consume before network I/O: concurrent submissions and retries cannot complete twice.
66    if (!isLiveNonce(this.ctx.storage.kv.get<TimedNonce>("nonce"), nonce, Date.now())) {
67      throw new Error("Connection link expired. Start again.");
68    }
69    this.ctx.storage.kv.delete("nonce");
70    const generation = this.#generation;
71    const checked = checkConnectionDraft(draft);
72    const normalized = { ...draft, coreOrigin: checked.coreOrigin };
73    const identity = await new CoreClient(normalized).identity();
74    if (generation !== this.#generation) throw new Error("Connection revoked.");
75    const credentials: Credentials = { ...normalized, agent: identity.agent };
76    const callback = this.ctx.storage.kv.get<Fetcher<GatekeeperConnectCallback>>("callback");
77    if (!callback) throw new Error("Connection callback unavailable.");
78    const existing = this.ctx.storage.kv.get<Credentials>("credentials");
79    if (existing) {
80      if (existing.agent !== credentials.agent || existing.project !== credentials.project ||
81          existing.coreOrigin !== credentials.coreOrigin) throw new Error("Reconnect must retain the same agent and project.");
82      const stageId = stageCredentials(this.ctx.storage.kv, credentials, Date.now());
83      return callback.reconnectComplete(stageId);
84    }
85    if (isConnectAttempted(this.ctx.storage.kv)) throw new Error("Connection already attempted.");
86    markConnectAttempted(this.ctx.storage.kv);
87    this.ctx.storage.kv.put("credentials", credentials);
88    try {
89      const handoff = await callback.complete(this.ctx.exports.AgentAccount({ props: { accountId: this.ctx.id.toString() } }));
90      if (generation !== this.#generation) throw new Error("Connection revoked.");
91      await this.ctx.storage.deleteAlarm();
92      return handoff;
93    } catch {
94      this.#generation++;
95      this.ctx.storage.kv.delete("credentials");
96      throw new Error("Connection completion failed. Start again.");
97    }
98  }
99  async reconnect(): Promise<string> {
100    if (!this.ctx.storage.kv.get("credentials")) throw new Error("Agent connection unavailable.");
101    const nonce = generateNonce();
102    this.ctx.storage.kv.put("nonce", { value: nonce, expiresAt: Date.now() + 600_000 } satisfies TimedNonce);
103    return nonce;
104  }
105  async commitReconnect(stageId: string): Promise<void> {
106    if (!this.ctx.storage.kv.get("credentials")) throw new Error("Agent connection unavailable.");
107    const staged = commitStagedCredentials<Credentials>(this.ctx.storage.kv, Date.now(), stageId);
108    if (!staged) throw new Error("Reconnect stage unavailable.");
109    this.#generation++;
110    this.ctx.storage.kv.put("credentials", staged);
111  }
112  async revoke(): Promise<void> {
113    this.#generation++;
114    this.ctx.storage.kv.delete("credentials");
115    await this.ctx.storage.deleteAll();
116  }
117  async alarm(): Promise<void> { await this.revoke(); }
118  async workflowContext() { return this.#client().context(); }
119  async workflowSubmission(operation: WorkflowWrite) { return this.#client().workflowSubmission(operation); }
120  async executeWorkflow(operation: WorkflowWrite): Promise<void> { await this.#client().executeWorkflow(operation); }
121  async readReview(review: string) { return this.#client().readReview(review); }
122  async readReceipts(review: string) { return this.#client().readReceipts(review); }
123  async identity() { return this.#client().identity(); }
124  async state(): Promise<ProjectState> { return this.#client().state(); }
125  async fork(repoName: string) { return this.#client().fork(repoName); }
126  async metadata(): Promise<{ coreOrigin: string; project: string; agent: string }> {
127    const identity = await this.identity();
128    const creds = this.ctx.storage.kv.get<Credentials>("credentials");
129    if (!creds) throw new Error("Agent connection unavailable.");
130    return { coreOrigin: creds.coreOrigin, project: creds.project, agent: identity.agent };
131  }
132  async pull(repoName: string, oids: GitOid[], hints: GitPullHints, cache: RpcStub<GitCache>): Promise<void> {
133    const client = this.#client();
134    try {
135      await pullGitObjectsIntoCache(body => client.git(repoName, "upload-pack", body), oids, hints, cache);
136    } catch { throw new Error("Git pull failed."); }
137  }
138  async push(repoName: string, branch: string, oldSha: string, newSha: string, pack: ReadableStream<Uint8Array>): Promise<void> {
139    const client = this.#client();
140    try {
141      await pushGitRefUpdate(body => client.git(repoName, "receive-pack", body), { branch, oldSha, newSha }, pack);
142    } catch { throw new Error("Git push failed or was rejected. Refresh the fork head."); }
143  }
144}
145
146function account(exports: Cloudflare.Exports, id: string) {
147  return exports.UserAccount.get(exports.UserAccount.idFromString(id));
148}
149
150@validateRpc()
151class ForkPicker extends RpcTarget {
152  constructor(private readonly source: DurableObjectStub<UserAccount>) { super(); }
153  async list(): Promise<{ url: string; repoName: string; head: string }[]> {
154    const meta = await this.source.metadata();
155    return (await this.source.identity()).forks.map(f => ({
156      url: forkAddress({ origin: meta.coreOrigin, project: meta.project, repoName: f.repoName }),
157      repoName: f.repoName, head: f.head,
158    }));
159  }
160}
161
162export class AgentAccount extends WorkerEntrypoint<Env, AccountProps> implements GatekeeperUser {
163  #account() { return account(this.ctx.exports, this.ctx.props.accountId); }
164  async describe(): Promise<AccountDescription> {
165    const meta = await this.#account().metadata();
166    return { ...sidebarAccountDescription(meta.agent), uniqueName: `${meta.agent} · ${meta.project}` };
167  }
168  async getSupportedResources(): Promise<SupportedResource[]> {
169    const meta = await this.#account().metadata();
170    const address = forkAddress({ origin: meta.coreOrigin, project: meta.project, repoName: `${meta.project}--repo-name` });
171    return [{ ...RESOURCE, description: `An agent-owned Git fork. Attach its exact address: ${address.slice(0, address.lastIndexOf("/") + 1)}<repoName>. Choose one of the connected agent's forks in the resource picker.` }];
172  }
173  async getGatekeeperClassFor(url: string): Promise<{ class: DurableObjectClass<Gatekeeper<any>>; resource: SupportedResource }> {
174    const source = this.#account();
175    const meta = await source.metadata();
176    const prefix = `${meta.coreOrigin}/projects/${meta.project}/forks/`;
177    if (!url.startsWith(prefix)) throw new Error("Fork is outside the connected project.");
178    const repoName = url.slice(prefix.length);
179    if (encodeURIComponent(repoName) !== repoName) throw new Error("Invalid fork URL.");
180    if (forkAddress({ origin: meta.coreOrigin, project: meta.project, repoName }) !== url) throw new Error("Invalid fork URL.");
181    await source.fork(repoName);
182    return { class: this.ctx.exports.ForkGatekeeper({ props: { accountId: this.ctx.props.accountId, repoName } }), resource: RESOURCE };
183  }
184  async startResourceConfigurator(pattern: string): Promise<ResourceConfiguratorFrame> {
185    if (pattern !== RESOURCE.urlPattern) throw new Error("Unsupported resource type.");
186    await this.#account().identity();
187    return { iframeHtml: PICKER_HTML, ui: new RpcStub(new ForkPicker(this.#account())) };
188  }
189  async startAppUi(_context: AppUiContext): Promise<GatekeeperUiFrame> {
190    const source = this.#account();
191    // Existing authenticated metadata; serialize only the public origin, never the credential.
192    // A failed read only omits the owner link; the frame keeps its existing error and retry path.
193    const coreOrigin = await source.metadata().then((meta) => meta.coreOrigin, () => undefined);
194    return createSidebarFrame({
195      authorize: async () => { await source.identity(); },
196      read: () => source.state(),
197      watch(changed, revoked) {
198        let stopped = false;
199        let timer: ReturnType<typeof setTimeout>;
200        let previous = "";
201        const poll = async () => {
202          try {
203            const state = await source.state();
204            if (stopped) return;
205            const key = JSON.stringify(state);
206            if (key !== previous) { previous = key; changed(state); }
207            if (!stopped) timer = setTimeout(() => { void poll(); }, 250);
208          } catch { if (!stopped) revoked(); }
209        };
210        void poll();
211        return () => { stopped = true; clearTimeout(timer); };
212      },
213    }, sidebarWithCoreOrigin(SIDEBAR_HTML, coreOrigin));
214  }
215  async revoke(): Promise<void> { await this.#account().revoke(); }
216  async reconnect(): Promise<{ url: string }> {
217    const nonce = await this.#account().reconnect();
218    return { url: `${this.env.BASE_URL}/${this.ctx.props.accountId}/${nonce}` };
219  }
220  async commitReconnect(stageId: string): Promise<void> { await this.#account().commitReconnect(stageId); }
221  async getAuthenticatedEmail(): Promise<null> { return null; }
222  async ensureResources(patterns: string[]): Promise<{}> {
223    if (patterns.some(p => p !== RESOURCE.urlPattern)) throw new Error("Unsupported resource type.");
224    await this.#account().identity(); return {};
225  }
226  async getVerifier(): Promise<Fetcher<GatekeeperUserVerifier>> {
227    return this.ctx.exports.AgentVerifier({ props: this.ctx.props });
228  }
229}
230
231export class AgentVerifier extends WorkerEntrypoint<Env, AccountProps> implements GatekeeperUserVerifier {
232  async owns(coreOrigin: string, project: string, repoName: string): Promise<boolean> {
233    const source = account(this.ctx.exports, this.ctx.props.accountId);
234    try {
235      const meta = await source.metadata();
236      if (meta.coreOrigin !== coreOrigin || meta.project !== project) return false;
237      await source.fork(repoName); return true;
238    } catch { return false; }
239  }
240}
241