Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/context/src/index.ts

1,424 lines66,578 bytesCodeBlame
1/**
2 * The context service: the context hub's catalog, search and scorecards.
3 *
4 * - **Catalog.** What a workspace builds and runs, as entities (projects,
5 * apps, APIs, packages, languages, owners, environments, integrations,
6 * docs) and relations between them. Built by itself: on every push to a
7 * project's default branch it reads the project's manifests and docs
8 * whose blobs changed (at most `MAX_READS` files a push) and joins them
9 * with what projects, deployments, identity and integrations know.
10 * Rebuilding is idempotent: entity ids are hashes of what they are.
11 * - **Search.** Docs, catalog entities, issues, pull requests and kept
12 * memory, embedded with Workers AI into a Vectorize index whose rows
13 * carry their workspace, project and whether they are private, so a
14 * query reads only what its reader may. Matching words in D1 answers
15 * when the index cannot, and fills in after it.
16 * - **Backfill.** Once per workspace (and again on request): the catalog
17 * for every project, memory candidates from docs, manifests and the last
18 * merged pull requests, and the search index filled. One queued job per
19 * project, capped and metered.
20 * - **Run context.** The Context section every g1t agent run starts with.
21 *
22 * Reached through service bindings: `POST /rpc/<method>`.
23 */
24
25import {
26 type Backfill,
27 type CaptureItem,
28 type Catalog,
29 type ContextStatus,
30 type Entity,
31 type EntityDetail,
32 type EntityKind,
33 ENTITY_KINDS,
34 type G1tEvent,
35 type Memory,
36 type Project,
37 type ProjectDeploys,
38 type RelationKind,
39 type RunContext,
40 type Scorecard,
41 type SearchHit,
42 type SearchKind,
43 type SearchResult,
44 type ServiceBinding,
45 type User,
46 type Viewer,
47 ComputeGate,
48 billingClient,
49 embeddingEstimateMicros,
50 localRefusal,
51 currentMovedPath,
52 currentWorkspaceSlug,
53 repoMove,
54 staleMovedPaths,
55 deploymentsClient,
56 fail,
57 granted,
58 identityClient,
59 integrationsClient,
60 memoryReviewClient,
61 ok,
62 projectsClient,
63 reposClient,
64 securityClient,
65 staleSlugs,
66 workClient,
67 type Result,
68} from "@g1t/contracts";
69
70import { assemble, authorsOf, integrationEntities, type EntityDraft, type FileRecord, type ProjectInput, type Surroundings } from "./assemble";
71import { extract, interesting, type FileFacts } from "./extract";
72import { composeRunContext, type ContextNote, type ProjectContext } from "./runcontext";
73import { evaluate } from "./scorecards";
74import { allowedKinds, countVisible, indexFilter, memoryReadable, merge, projectReadable, readable, runMemoryReadable, type IndexMeta, type Reader } from "./visibility";
75
76type Job =
77 | { type: "backfill_project"; workspace: string; slug: string }
78 | { type: "backfill_memory"; workspace: string };
79
80type Env = {
81 DB: D1Database;
82 AI?: Ai;
83 VECTORS?: Vectorize;
84 JOBS: Queue<Job>;
85 REPOS: ServiceBinding;
86 PROJECTS: ServiceBinding;
87 WORK: ServiceBinding;
88 IDENTITY: ServiceBinding;
89 DEPLOYMENTS: ServiceBinding;
90 INTEGRATIONS: ServiceBinding;
91 SECURITY?: ServiceBinding;
92 /** Told the month's embedding cost so far, which it charges once the month is over. */
93 BILLING?: ServiceBinding;
94};
95
96/** Workers AI's embedding model: 768 dimensions, as the index was made with. */
97const EMBED_MODEL = "@cf/baai/bge-base-en-v1.5";
98/** What it costs, in millionths of a dollar per token ($0.067 per million). */
99const MICROS_PER_TOKEN = 0.067;
100/** Past this many tokens in a month, a workspace's new text goes unindexed; text search still finds it. */
101const MONTHLY_TOKENS = 20_000_000;
102/** The most text embedded for one row; the model reads 512 tokens. */
103const EMBED_CHARS = 2000;
104const EMBED_BATCH = 50;
105/** The most files read from a repository for one project's rebuild. */
106const MAX_READS = 15;
107/** Of them, workflows. */
108const MAX_WORKFLOW_READS = 3;
109/** The most projects one push rebuilds. */
110const MAX_PROJECTS_PER_PUSH = 5;
111/** A backfill's caps. */
112const BACKFILL_PROJECTS = 50;
113const BACKFILL_PULLS = 20;
114const BACKFILL_ITEMS = 30;
115/** A backfill still marked running after this long is taken to have died. */
116const BACKFILL_STALE_MS = 30 * 60 * 1000;
117const ITEM_CHARS = 4000;
118const SNIPPET_CHARS = 280;
119/** Kinds every project shares rather than owns. */
120const SHARED: Set<EntityKind> = new Set(["owner", "language", "integration"]);
121
122/** One compute gate per isolate, so entitlements are kept between calls. */
123let computeGate: ComputeGate | null = null;
124function gateFor(billing: ServiceBinding): ComputeGate {
125 computeGate ??= new ComputeGate(billing);
126 return computeGate;
127}
128
129const now = () => new Date().toISOString();
130const month = () => now().slice(0, 7);
131
132function isMember(viewer: Viewer, workspace: string): boolean {
133 return !!viewer?.workspaces?.some((m) => m.slug === workspace.toLowerCase());
134}
135
136async function sha(text: string): Promise<string> {
137 const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text));
138 return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, "0")).join("");
139}
140
141/** An entity's id: the same for the same thing, however often it is rebuilt. */
142export async function entityId(workspace: string, kind: string, key: string): Promise<string> {
143 return `ent_${(await sha(`${workspace}\u0000${kind}\u0000${key.toLowerCase()}`)).slice(0, 24)}`;
144}
145
146function snippet(text: string | null | undefined): string {
147 const line = (text ?? "").replace(/\s+/g, " ").trim();
148 return line.length <= SNIPPET_CHARS ? line : `${line.slice(0, SNIPPET_CHARS - 1)}…`;
149}
150
151type EntityRow = {
152 id: string;
153 workspace: string;
154 kind: EntityKind;
155 key: string;
156 name: string;
157 summary: string | null;
158 project_id: string | null;
159 project: string | null;
160 repo_id: string | null;
161 private: number;
162 data: string;
163 source: string;
164 ref: string | null;
165 updated_at: string;
166};
167
168type ItemRow = {
169 id: string;
170 workspace: string;
171 kind: "doc" | "issue" | "pull";
172 entity_id: string | null;
173 project: string | null;
174 private: number;
175 title: string;
176 text: string;
177 url: string | null;
178 by: string | null;
179 updated_at: string;
180};
181
182function toEntity(row: EntityRow): Entity {
183 let data: Record<string, unknown> = {};
184 try {
185 data = JSON.parse(row.data);
186 } catch {}
187 return {
188 id: row.id,
189 workspace: row.workspace,
190 kind: row.kind,
191 key: row.key,
192 name: row.name,
193 summary: row.summary,
194 project: row.project,
195 private: !!row.private,
196 data,
197 source: row.source,
198 ref: row.ref,
199 updatedAt: row.updated_at,
200 };
201}
202
203/** Where a memory came from, in a few words. */
204function memorySource(memory: Memory): string {
205 const ref = memory.source.reference ?? "";
206 const number = memory.source.number;
207 switch (memory.source.kind) {
208 case "doc":
209 return ref.split(":").slice(2).join(":") || "a doc";
210 case "review":
211 return number ? `review on #${number}` : "a review";
212 case "pr":
213 return number ? `#${number}` : "a merged pull request";
214 case "run":
215 return number ? `agent run on #${number}` : "an agent run";
216 default:
217 return `added by ${memory.createdBy}`;
218 }
219}
220
221/** What one workspace's rebuilds share: asked of other services once. */
222class WorkspaceCache {
223 private members?: Promise<string[]>;
224 private deploys?: Promise<ProjectDeploys[]>;
225 private connections?: Promise<Surroundings["integrations"]>;
226 constructor(
227 private readonly env: Env,
228 readonly workspace: string,
229 readonly actor: User,
230 ) {}
231
232 memberNames(): Promise<string[]> {
233 this.members ??= identityClient(this.env.IDENTITY)
234 .listMembers(this.workspace, this.actor)
235 .then((found) => (found.ok ? found.value.map((member) => member.username) : []))
236 .catch(() => []);
237 return this.members;
238 }
239
240 deployments(): Promise<ProjectDeploys[]> {
241 this.deploys ??= deploymentsClient(this.env.DEPLOYMENTS)
242 .overview(this.workspace, this.actor)
243 .then((found) => (found.ok ? found.value : []))
244 .catch(() => []);
245 return this.deploys;
246 }
247
248 integrations(): Promise<Surroundings["integrations"]> {
249 this.connections ??= integrationsClient(this.env.INTEGRATIONS)
250 .list(this.workspace, this.actor)
251 .then((found) =>
252 found.ok
253 ? found.value
254 .filter((connection) => connection.kind !== "models")
255 .map((connection) => ({
256 id: connection.id,
257 provider: connection.provider,
258 name: connection.name,
259 kind: connection.kind,
260 repo: connection.config.repo ?? null,
261 }))
262 : [],
263 )
264 .catch(() => []);
265 return this.connections;
266 }
267}
268
269type ScanStats = { entities: number; candidates: number; kept: number; indexed: number };
270
271class Context {
272 constructor(private readonly env: Env) {}
273
274 private get db() {
275 return this.env.DB;
276 }
277
278 private async workspaceActor(slug: string): Promise<User | null> {
279 const workspace = await identityClient(this.env.IDENTITY).getWorkspace(slug);
280 if (!workspace) return null;
281 return {
282 id: workspace.id,
283 username: workspace.slug,
284 kind: "workspace",
285 verified: true,
286 workspaces: [{ slug: workspace.slug, role: "member" }],
287 };
288 }
289
290 // ---- Search index --------------------------------------------------------
291
292 /**
293 * Whether semantic search is open to a workspace: embeddings are compute,
294 * so only on a paid plan or the trial (`localRefusal`). Text search
295 * answers for everyone else. Closed when billing cannot say, which
296 * leaves text search; open where there is no billing service at all.
297 */
298 private async semanticOpen(workspace: string): Promise<boolean> {
299 if (!this.env.BILLING) return true;
300 const ent = await gateFor(this.env.BILLING).entitlements(workspace);
301 return ent != null && localRefusal(ent, "embedding", false) == null;
302 }
303
304 /**
305 * Embeds and stores rows in the search index, within the workspace's
306 * monthly allowance, reserving what it costs with billing first and
307 * settling what it did. Never throws.
308 */
309 private async index(workspace: string, rows: { id: string; text: string; meta: IndexMeta }[]): Promise<number> {
310 const { AI, VECTORS } = this.env;
311 if (!AI || !VECTORS || rows.length === 0) return 0;
312 if (!(await this.semanticOpen(workspace))) return 0;
313 let reservation: string | null = null;
314 let spentTokens = 0;
315 if (this.env.BILLING) {
316 const estimate = rows.reduce((sum, row) => sum + Math.ceil(Math.min(row.text.length, EMBED_CHARS) / 4), 0);
317 const admitted = await gateFor(this.env.BILLING).admit({
318 workspace,
319 repo: { namespace: workspace, name: rows[0].meta.project ?? "" },
320 public: rows.every((row) => !row.meta.private),
321 kind: "embedding",
322 estimateMicros: embeddingEstimateMicros(estimate),
323 });
324 if (!admitted.ok) {
325 console.log("embeddings not started for", workspace, admitted.code, admitted.message);
326 return 0;
327 }
328 reservation = admitted.reservation?.id ?? null;
329 }
330 try {
331 const used = await this.db
332 .prepare("SELECT tokens FROM usage WHERE workspace = ? AND month = ?")
333 .bind(workspace, month())
334 .first<{ tokens: number }>();
335 if ((used?.tokens ?? 0) >= MONTHLY_TOKENS) return 0;
336 let stored = 0;
337 for (let at = 0; at < rows.length; at += EMBED_BATCH) {
338 const batch = rows.slice(at, at + EMBED_BATCH);
339 const texts = batch.map((row) => row.text.slice(0, EMBED_CHARS));
340 const embedded = (await AI.run(EMBED_MODEL, { text: texts })) as { data?: number[][] };
341 const vectors = (embedded.data ?? []).map((values, i) => ({
342 id: batch[i].id,
343 values,
344 metadata: batch[i].meta as unknown as Record<string, VectorizeVectorMetadata>,
345 }));
346 if (vectors.length) await VECTORS.upsert(vectors);
347 stored += vectors.length;
348 const tokens = (rows: { text: string }[]) => rows.reduce((sum, row) => sum + Math.ceil(row.text.length / 4), 0);
349 const all = texts.map((text) => ({ text }));
350 spentTokens += tokens(all);
351 await this.meter(workspace, tokens(all), tokens(all.filter((_, i) => batch[i].meta.private)));
352 }
353 return stored;
354 } catch (error) {
355 console.error("could not index", rows.length, "rows for", workspace, error);
356 return 0;
357 } finally {
358 if (reservation && this.env.BILLING) {
359 await gateFor(this.env.BILLING).settle(reservation, embeddingEstimateMicros(spentTokens));
360 }
361 }
362 }
363
364 /** Drops a repository's rows kept under any of `workspaces`, and their index entries. */
365 private async forgetRepo(repoId: string, workspaces: string[]): Promise<void> {
366 const marks = workspaces.map(() => "?").join(", ");
367 const items = await this.db
368 .prepare(`SELECT id FROM items WHERE repo_id = ? AND workspace IN (${marks})`)
369 .bind(repoId, ...workspaces)
370 .all<{ id: string }>();
371 for (let i = 0; i < items.results.length; i += 100) {
372 await this.unindex(items.results.slice(i, i + 100).map((row) => row.id));
373 }
374 const projects = `SELECT project_id FROM entities WHERE repo_id = ? AND workspace IN (${marks}) AND project_id IS NOT NULL`;
375 await this.db.batch([
376 this.db.prepare(`DELETE FROM relations WHERE project_id IN (${projects})`).bind(repoId, ...workspaces),
377 this.db.prepare(`DELETE FROM files WHERE project_id IN (${projects})`).bind(repoId, ...workspaces),
378 this.db.prepare(`DELETE FROM items WHERE repo_id = ? AND workspace IN (${marks})`).bind(repoId, ...workspaces),
379 this.db.prepare(`DELETE FROM scans WHERE repo_id = ? AND workspace IN (${marks})`).bind(repoId, ...workspaces),
380 this.db.prepare(`DELETE FROM entities WHERE repo_id = ? AND workspace IN (${marks})`).bind(repoId, ...workspaces),
381 ]);
382 }
383
384 /** A deleted workspace's hub: everything but its usage, which billing has already read. */
385 private async forgetWorkspace(slug: string): Promise<void> {
386 const items = await this.db.prepare("SELECT id FROM items WHERE workspace = ?").bind(slug).all<{ id: string }>();
387 for (let i = 0; i < items.results.length; i += 100) {
388 await this.unindex(items.results.slice(i, i + 100).map((row) => row.id));
389 }
390 await this.db.batch(
391 ["items", "relations", "entities", "scans", "backfills"].map((table) =>
392 this.db.prepare(`DELETE FROM ${table} WHERE workspace = ?`).bind(slug),
393 ),
394 );
395 }
396
397 private async unindex(ids: string[]): Promise<void> {
398 if (!this.env.VECTORS || ids.length === 0) return;
399 try {
400 await this.env.VECTORS.deleteByIds(ids);
401 } catch (error) {
402 console.error("could not take", ids.length, "rows out of the index", error);
403 }
404 }
405
406 /**
407 * Records what embedding cost. Of it, `billableTokens` are private
408 * text's: public repositories' text and searches are never charged.
409 * Billing is told the month's billable total, which it charges at cost
410 * plus its margin once the month is over (its `embedding_tokens` meter).
411 * A failure to tell billing only delays it: the next call sends the
412 * whole month again.
413 */
414 private async meter(workspace: string, tokens: number, billableTokens = 0): Promise<void> {
415 const total = await this.db
416 .prepare(
417 `INSERT INTO usage (workspace, month, tokens, cost_micros, billable_micros) VALUES (?1, ?2, ?3, ?4, ?5)
418 ON CONFLICT (workspace, month) DO UPDATE SET
419 tokens = tokens + ?3, cost_micros = cost_micros + ?4, billable_micros = billable_micros + ?5
420 RETURNING billable_micros`,
421 )
422 .bind(workspace, month(), tokens, Math.ceil(tokens * MICROS_PER_TOKEN), billableTokens * MICROS_PER_TOKEN)
423 .first<{ billable_micros: number }>();
424 if (this.env.BILLING && total && billableTokens > 0) {
425 await billingClient(this.env.BILLING)
426 .notePending(workspace, "context", Math.ceil(total.billable_micros))
427 .catch((error) => console.error("could not tell billing what embedding cost", workspace, error));
428 }
429 }
430
431 /**
432 * A query's embedding, for semantic search. Too small to reserve one by
433 * one (a few hundred tokens, a fraction of a cent): it needs the plan or
434 * the trial, as indexing does, and is metered with the month's usage.
435 */
436 private async embedQuery(workspace: string, query: string): Promise<number[] | null> {
437 if (!this.env.AI || !this.env.VECTORS) return null;
438 if (!(await this.semanticOpen(workspace))) return null;
439 try {
440 const embedded = (await this.env.AI.run(EMBED_MODEL, { text: [query.slice(0, EMBED_CHARS)] })) as { data?: number[][] };
441 await this.meter(workspace, Math.ceil(query.length / 4));
442 return embedded.data?.[0] ?? null;
443 } catch (error) {
444 console.error("could not embed a query", error);
445 return null;
446 }
447 }
448
449 // ---- Building the catalog --------------------------------------------------
450
451 /** The files of a project worth reading, with their blobs, found in a few tree reads. */
452 private async candidates(project: Project, actor: User, ref: string): Promise<{ files: { path: string; hash: string }[]; siblings: string[]; head: string | null } | null> {
453 if (project.source.kind !== "hosted") return null;
454 const repos = reposClient(this.env.REPOS);
455 const { repo, rootDir: root } = project.source;
456 const at = (path: string) => [root, path].filter(Boolean).join("/");
457 const top = await repos.tree(repo, actor, ref, root);
458 if (!top.ok) return null;
459 const files: { path: string; hash: string }[] = [];
460 const siblings = top.value.entries.map((entry) => entry.name);
461 const blobs = (entries: { name: string; hash: string; kind: string }[], dir: string) => {
462 for (const entry of entries) {
463 if (entry.kind !== "blob" && entry.kind !== "exec") continue;
464 const path = dir ? `${dir}/${entry.name}` : entry.name;
465 if (interesting(path)) files.push({ path, hash: entry.hash });
466 }
467 };
468 blobs(top.value.entries, "");
469 const dirs = new Set(top.value.entries.filter((entry) => entry.kind === "tree").map((entry) => entry.name));
470 const look = async (dir: string) => {
471 const found = await repos.tree(repo, actor, ref, at(dir)).catch(() => null);
472 return found?.ok ? found.value.entries : [];
473 };
474 for (const dir of ["docs", "doc", "runbooks"]) if (dirs.has(dir)) blobs(await look(dir), dir);
475 for (const dir of [".g1t", ".github"]) {
476 if (!dirs.has(dir)) continue;
477 const inside = await look(dir);
478 blobs(inside, dir);
479 if (inside.some((entry) => entry.name === "workflows" && entry.kind === "tree")) blobs(await look(`${dir}/workflows`), `${dir}/workflows`);
480 }
481 return { files, siblings, head: top.value.head?.hash ?? null };
482 }
483
484 /**
485 * Builds one project's place in the catalog from its default branch (or
486 * `commit`), reading only files whose blobs changed unless `force`.
487 * Indexes what changed and sends what its docs say to memory.
488 */
489 async scan(project: Project, cache: WorkspaceCache, commit: string | null, force: boolean): Promise<ScanStats> {
490 const stats: ScanStats = { entities: 0, candidates: 0, kept: 0, indexed: 0 };
491 if (project.source.kind !== "hosted") return stats;
492 const { repo, repoId, rootDir, defaultBranch } = project.source;
493 const ref = commit ?? defaultBranch;
494 const found = await this.candidates(project, cache.actor, ref);
495 if (!found) return stats;
496 const workspace = project.workspace;
497
498 // What each file says: stored for unchanged blobs, read for the rest.
499 const stored = new Map(
500 (
501 await this.db.prepare("SELECT path, hash, facts FROM files WHERE project_id = ?").bind(project.id).all<{ path: string; hash: string; facts: string }>()
502 ).results.map((row) => [row.path, row]),
503 );
504 const files: FileRecord[] = [];
505 const changed: string[] = [];
506 const writes: D1PreparedStatement[] = [];
507 let reads = 0;
508 let workflowReads = 0;
509 const repos = reposClient(this.env.REPOS);
510 for (const file of found.files) {
511 const before = stored.get(file.path);
512 const workflow = file.path.includes("/workflows/");
513 const fresh = before && before.hash === file.hash && !force;
514 const canRead = reads < MAX_READS && (!workflow || workflowReads < MAX_WORKFLOW_READS);
515 if (fresh || !canRead) {
516 if (before) files.push({ path: file.path, facts: JSON.parse(before.facts) as FileFacts });
517 continue;
518 }
519 reads++;
520 if (workflow) workflowReads++;
521 const blob = await repos.blob(repo, cache.actor, found.head ?? ref, [rootDir, file.path].filter(Boolean).join("/")).catch(() => null);
522 if (!blob?.ok || blob.value.text == null) continue;
523 const facts = extract(file.path, blob.value.text, { project: project.name, siblings: found.siblings });
524 files.push({ path: file.path, facts });
525 changed.push(file.path);
526 writes.push(
527 this.db
528 .prepare(
529 `INSERT INTO files (project_id, path, hash, facts, read_at) VALUES (?, ?, ?, ?, ?)
530 ON CONFLICT (project_id, path) DO UPDATE SET hash = excluded.hash, facts = excluded.facts, read_at = excluded.read_at`,
531 )
532 .bind(project.id, file.path, file.hash, JSON.stringify(facts), now()),
533 );
534 }
535 const present = new Set(found.files.map((file) => file.path));
536 for (const path of stored.keys()) {
537 if (!present.has(path)) writes.push(this.db.prepare("DELETE FROM files WHERE project_id = ? AND path = ?").bind(project.id, path));
538 }
539
540 // What g1t knows about it besides its files.
541 const [members, deploys, integrations, graph, log] = await Promise.all([
542 cache.memberNames(),
543 cache.deployments(),
544 cache.integrations(),
545 projectsClient(this.env.PROJECTS).graph(project.id).catch(() => ({ dependsOn: [], usedBy: [] })),
546 repos.log(repo, cache.actor, found.head ?? ref, 100).catch(() => null),
547 ]);
548 const deploy = deploys.find((d) => d.slug === project.slug) ?? null;
549 const around: Surroundings = {
550 owners: authorsOf(log?.ok ? log.value : [], members),
551 dependsOn: graph.dependsOn.map((dep) => ({ slug: dep.slug, as: dep.as })),
552 deploy: deploy
553 ? {
554 enabled: deploy.enabled,
555 production: deploy.production ? { url: deploy.production.url, commit: deploy.production.commit, deployedAt: deploy.production.deployedAt } : null,
556 previews: deploy.previews,
557 latest: deploy.latest ? { status: deploy.latest.status, kind: deploy.latest.kind, error: deploy.latest.error, createdAt: deploy.latest.createdAt } : null,
558 }
559 : null,
560 integrations,
561 };
562 const input: ProjectInput = {
563 id: project.id,
564 workspace,
565 slug: project.slug,
566 name: project.name,
567 description: project.description,
568 private: project.private,
569 repoId,
570 repo,
571 rootDir,
572 defaultBranch,
573 };
574 const built = assemble(input, files, around);
575 const drafts: EntityDraft[] = [...built.entities, ...integrationEntities(integrations)];
576
577 // Entities: upserted; the project's own that it no longer has, removed.
578 const previous = new Map(
579 (
580 await this.db
581 .prepare("SELECT id, name, summary FROM entities WHERE workspace = ? AND (project_id = ? OR project_id IS NULL)")
582 .bind(workspace, project.id)
583 .all<{ id: string; name: string; summary: string | null }>()
584 ).results.map((row) => [row.id, row]),
585 );
586 const ids: string[] = [];
587 const toIndex: { id: string; text: string; meta: IndexMeta }[] = [];
588 const at = now();
589 for (const draft of drafts) {
590 const id = await entityId(workspace, draft.kind, draft.key);
591 ids.push(id);
592 const shared = SHARED.has(draft.kind);
593 const source = draft.kind === "app" || draft.kind === "environment" ? "deployments" : draft.kind === "integration" ? "integrations" : "scan";
594 writes.push(
595 this.db
596 .prepare(
597 `INSERT INTO entities (id, workspace, kind, key, name, summary, project_id, project, repo_id, private, data, source, ref, updated_at)
598 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
599 ON CONFLICT (workspace, kind, key) DO UPDATE SET name = excluded.name, summary = excluded.summary,
600 project_id = excluded.project_id, project = excluded.project, repo_id = excluded.repo_id, private = excluded.private,
601 data = excluded.data, source = excluded.source, ref = excluded.ref, updated_at = excluded.updated_at`,
602 )
603 .bind(
604 id,
605 workspace,
606 draft.kind,
607 draft.key,
608 draft.name,
609 draft.summary,
610 shared ? null : project.id,
611 shared ? null : project.slug,
612 shared ? null : repoId,
613 shared ? 0 : project.private ? 1 : 0,
614 JSON.stringify(draft.data),
615 source,
616 draft.ref,
617 at,
618 ),
619 );
620 const before = previous.get(id);
621 if (force || !before || before.name !== draft.name || before.summary !== draft.summary) {
622 toIndex.push({
623 id,
624 text: `${draft.kind} ${draft.name}. ${draft.summary ?? ""}`,
625 meta: {
626 workspace,
627 kind: draft.kind,
628 project: shared ? "" : project.slug,
629 private: shared ? false : project.private,
630 title: draft.name,
631 snippet: snippet(draft.summary),
632 url: draft.ref ?? "",
633 source: "catalog",
634 by: "",
635 at,
636 },
637 });
638 }
639 }
640 const gone = (
641 await this.db
642 .prepare("SELECT id FROM entities WHERE project_id = ? AND id NOT IN (SELECT value FROM json_each(?))")
643 .bind(project.id, JSON.stringify(ids))
644 .all<{ id: string }>()
645 ).results.map((row) => row.id);
646 if (gone.length) {
647 writes.push(this.db.prepare("DELETE FROM entities WHERE id IN (SELECT value FROM json_each(?))").bind(JSON.stringify(gone)));
648 writes.push(this.db.prepare("DELETE FROM items WHERE entity_id IN (SELECT value FROM json_each(?))").bind(JSON.stringify(gone)));
649 }
650
651 // Relations: the project's own, replaced whole.
652 writes.push(this.db.prepare("DELETE FROM relations WHERE project_id = ?").bind(project.id));
653 for (const relation of built.relations) {
654 const [from, to] = await Promise.all([entityId(workspace, relation.from.kind, relation.from.key), entityId(workspace, relation.to.kind, relation.to.key)]);
655 writes.push(
656 this.db
657 .prepare("INSERT OR REPLACE INTO relations (workspace, from_id, kind, to_id, project_id, updated_at) VALUES (?, ?, ?, ?, ?, ?)")
658 .bind(workspace, from, relation.kind, to, project.id, at),
659 );
660 }
661
662 // Docs that changed: their pieces, for search.
663 const repoPath = `/${repo.namespace}/${repo.name}`;
664 const chunkIds: string[] = [];
665 for (const file of files) {
666 const doc = file.facts.doc;
667 if (!doc || (!changed.includes(file.path) && !force)) continue;
668 const docId = await entityId(workspace, "doc", `${project.slug}:${doc.path}`);
669 writes.push(this.db.prepare("DELETE FROM items WHERE entity_id = ?").bind(docId));
670 const url = `${repoPath}/blob/${defaultBranch}/${[rootDir, doc.path].filter(Boolean).join("/")}`;
671 doc.chunks.forEach((text, i) => {
672 const id = `${docId}:${i}`;
673 chunkIds.push(id);
674 writes.push(
675 this.db
676 .prepare(
677 `INSERT OR REPLACE INTO items (id, workspace, kind, entity_id, project_id, project, repo_id, private, title, text, url, by, updated_at)
678 VALUES (?, ?, 'doc', ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)`,
679 )
680 .bind(id, workspace, docId, project.id, project.slug, repoId, project.private ? 1 : 0, `${doc.title} (${doc.path})`, text, url, at),
681 );
682 toIndex.push({
683 id,
684 text: `${doc.title}\n${text}`,
685 meta: { workspace, kind: "doc", project: project.slug, private: project.private, title: `${doc.title} (${doc.path})`, snippet: snippet(text), url, source: doc.path, by: "", at },
686 });
687 });
688 }
689 writes.push(
690 this.db
691 .prepare(
692 `INSERT INTO scans (project_id, workspace, repo_id, "commit", tests, scanned_at) VALUES (?, ?, ?, ?, ?, ?)
693 ON CONFLICT (project_id) DO UPDATE SET workspace = excluded.workspace, "commit" = excluded."commit", tests = excluded.tests, scanned_at = excluded.scanned_at`,
694 )
695 .bind(project.id, workspace, repoId, found.head, built.tests ? 1 : 0, at),
696 );
697 for (let i = 0; i < writes.length; i += 50) await this.db.batch(writes.slice(i, i + 50));
698 await this.unindex(gone);
699 stats.entities = drafts.length;
700 stats.indexed = await this.index(workspace, toIndex);
701
702 // What changed files say worth remembering, as memory candidates.
703 const items: CaptureItem[] = built.hints
704 .filter((hint) => force || changed.includes(hint.path))
705 .map((hint) => ({
706 scope: "project",
707 repoId,
708 kind: hint.kind,
709 text: hint.text,
710 confidence: hint.confidence,
711 source: "doc",
712 reference: `doc:${repoId}:${[rootDir, hint.path].filter(Boolean).join("/")}`,
713 evidence: hint.evidence,
714 }));
715 for (let i = 0; i < items.length; i += 50) {
716 const captured = await memoryReviewClient(this.env.WORK)
717 .captureMemories(workspace, items.slice(i, i + 50), "g1t")
718 .catch(() => null);
719 stats.candidates += captured?.added ?? 0;
720 stats.kept += captured?.kept ?? 0;
721 }
722 return stats;
723 }
724
725 // ---- Issues, pull requests and memory in search ------------------------
726
727 private async repoOf(repoId: string): Promise<{ namespace: string; name: string } | null> {
728 const response = await this.env.REPOS.fetch("https://repos/rpc/path_by_id", {
729 method: "POST",
730 headers: { "content-type": "application/json" },
731 body: JSON.stringify({ id: repoId }),
732 });
733 return response.ok ? ((await response.json()) as { namespace: string; name: string } | null) : null;
734 }
735
736 /** Stores and indexes issues and pull requests as search rows. */
737 private async putItems(
738 workspace: string,
739 project: Project | null,
740 repoId: string,
741 rows: { kind: "issue" | "pull"; number: number; title: string; body: string | null; by: string | null; updatedAt: string; url: string; private: boolean }[],
742 ): Promise<number> {
743 const writes: D1PreparedStatement[] = [];
744 const toIndex: { id: string; text: string; meta: IndexMeta }[] = [];
745 for (const row of rows) {
746 const id = `${row.kind}:${repoId}#${row.number}`;
747 const title = `#${row.number} ${row.title}`;
748 const text = (row.body ?? "").slice(0, ITEM_CHARS);
749 writes.push(
750 this.db
751 .prepare(
752 `INSERT OR REPLACE INTO items (id, workspace, kind, entity_id, project_id, project, repo_id, private, title, text, url, by, updated_at)
753 VALUES (?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
754 )
755 .bind(id, workspace, row.kind, project?.id ?? null, project?.slug ?? null, repoId, row.private ? 1 : 0, title, text, row.url, row.by, row.updatedAt),
756 );
757 toIndex.push({
758 id,
759 text: `${row.title}\n${text}`,
760 meta: { workspace, kind: row.kind, project: project?.slug ?? "", private: row.private, title, snippet: snippet(text || row.title), url: row.url, source: row.kind === "issue" ? "issue" : "pull request", by: row.by ?? "", at: row.updatedAt },
761 });
762 }
763 for (let i = 0; i < writes.length; i += 50) await this.db.batch(writes.slice(i, i + 50));
764 return this.index(workspace, toIndex);
765 }
766
767 private async indexIssueOrPull(kind: "issue" | "pull", repoId: string, number: number): Promise<void> {
768 const path = await this.repoOf(repoId);
769 if (!path) return;
770 const workspace = path.namespace.toLowerCase();
771 const actor = await this.workspaceActor(workspace);
772 if (!actor) return;
773 const [repo, projects] = await Promise.all([reposClient(this.env.REPOS).get(path, actor), projectsClient(this.env.PROJECTS).byRepo(repoId)]);
774 if (!repo.ok || repo.value.forkOf) return;
775 const work = workClient(this.env.WORK);
776 const base = `/${path.namespace}/${path.name}`;
777 if (kind === "issue") {
778 const found = await work.getIssue(path, number, actor);
779 if (!found.ok) return;
780 const issue = found.value.issue;
781 await this.putItems(workspace, projects[0] ?? null, repoId, [
782 { kind, number, title: issue.title, body: issue.body, by: issue.author.username, updatedAt: issue.updatedAt, url: `${base}/issues/${number}`, private: repo.value.isPrivate },
783 ]);
784 } else {
785 const found = await work.getPull(path, number, actor);
786 if (!found.ok) return;
787 const pull = found.value.pull as { title: string; body: string | null; agent: string; updatedAt?: string; author?: { username: string } };
788 await this.putItems(workspace, projects[0] ?? null, repoId, [
789 { kind, number, title: pull.title, body: pull.body, by: pull.author?.username ?? pull.agent, updatedAt: pull.updatedAt ?? now(), url: `${base}/pull/${number}`, private: repo.value.isPrivate },
790 ]);
791 }
792 }
793
794 /** A memory in search while it is kept, and out of it otherwise. */
795 private async indexMemories(workspace: string, memories: Memory[]): Promise<number> {
796 const kept = memories.filter((memory) => (memory.status ?? "kept") === "kept");
797 await this.unindex(memories.filter((memory) => !kept.includes(memory)).map((memory) => `memory:${memory.id}`));
798 const projects = await projectsClient(this.env.PROJECTS)
799 .list(workspace, (await this.workspaceActor(workspace)) ?? null)
800 .then((found) => (found.ok ? found.value : []))
801 .catch(() => [] as Project[]);
802 const slugOf = (memory: Memory) =>
803 memory.scope === "project" && memory.repo
804 ? (projects.find(
805 (project) =>
806 project.source.kind === "hosted" &&
807 project.primary &&
808 project.source.repo.namespace.toLowerCase() === memory.repo!.namespace.toLowerCase() &&
809 project.source.repo.name.toLowerCase() === memory.repo!.name.toLowerCase(),
810 )?.slug ?? "")
811 : "";
812 return this.index(
813 workspace,
814 kept.map((memory) => ({
815 id: `memory:${memory.id}`,
816 text: `${memory.kind}: ${memory.text}`,
817 meta: {
818 workspace,
819 kind: "memory",
820 project: slugOf(memory),
821 // Memory is for members, whatever its project.
822 private: true,
823 title: `${memory.kind[0].toUpperCase()}${memory.kind.slice(1)}${memory.scope === "workspace" ? " (workspace)" : ""}`,
824 snippet: snippet(memory.text),
825 url: memory.repo ? `/${memory.repo.namespace}/${memory.repo.name}/memory` : `/${workspace}/-/memory`,
826 source: memorySource(memory),
827 by: memory.createdBy,
828 at: memory.updatedAt,
829 },
830 })),
831 );
832 }
833
834 // ---- Events --------------------------------------------------------------
835
836 async onEvent(event: G1tEvent): Promise<void> {
837 switch (event.type) {
838 case "git.push": {
839 if (!event.data.defaultBranch) return;
840 const projects = (await projectsClient(this.env.PROJECTS).byRepo(event.data.repoId)).slice(0, MAX_PROJECTS_PER_PUSH);
841 if (projects.length === 0) return;
842 const actor = await this.workspaceActor(projects[0].workspace);
843 if (!actor) return;
844 const cache = new WorkspaceCache(this.env, projects[0].workspace, actor);
845 for (const project of projects) await this.scan(project, cache, event.data.after, false);
846 return;
847 }
848 case "issue.opened":
849 case "issue.updated":
850 case "issue.closed":
851 return this.indexIssueOrPull("issue", event.data.repoId, event.data.number);
852 case "pull.ready":
853 case "pull.merged":
854 return this.indexIssueOrPull("pull", event.data.repoId, event.data.number);
855 case "memory.changed": {
856 const { memoryId, workspace, status } = event.data;
857 if (status !== "kept") return this.unindex([`memory:${memoryId}`]);
858 const memories = await memoryReviewClient(this.env.WORK).memoriesById(workspace, [memoryId]);
859 await this.indexMemories(workspace, memories);
860 return;
861 }
862 case "repo.transferred":
863 case "repo.renamed": {
864 // What the hub knew about the repository under its old path goes,
865 // index included (its rows carry the path: a transfer's old
866 // workspace, a rename's old name); the workspace it is in now is
867 // built again, which reads it where and as it is now. Nothing here
868 // is the only copy.
869 const move = repoMove(event)!;
870 const current = await currentMovedPath(this.env.REPOS, move);
871 const stale = staleMovedPaths(move, current);
872 if (stale.length === 0) return;
873 const workspace = current.split("/")[0]!.toLowerCase();
874 await this.forgetRepo(move.repoId, [...new Set(stale.map((path) => path.split("/")[0]!.toLowerCase()))]);
875 const actor = await this.workspaceActor(workspace);
876 if (actor) await this.backfill({ actor, workspace });
877 return;
878 }
879 case "repo.deleted":
880 case "repo.purged": {
881 // Deleted, it is hidden: what the hub knew of it goes, index
882 // included, so no search or agent finds it. A restore builds it
883 // again; a purge finds nothing left.
884 await this.forgetRepo(event.data.repoId, [event.data.namespace.toLowerCase()]);
885 return;
886 }
887 case "repo.restored": {
888 const workspace = event.data.namespace.toLowerCase();
889 const actor = await this.workspaceActor(workspace);
890 if (actor) await this.backfill({ actor, workspace });
891 return;
892 }
893 case "workspace.deleted": {
894 await this.forgetWorkspace(event.data.slug);
895 return;
896 }
897 case "workspace.renamed": {
898 const current = await currentWorkspaceSlug(this.env.IDENTITY, event.data);
899 for (const old of staleSlugs(event.data, current)) {
900 await this.db.batch(
901 ["entities", "relations", "items", "scans", "usage"].map((table) =>
902 this.db.prepare(`UPDATE OR IGNORE ${table} SET workspace = ? WHERE workspace = ?`).bind(current, old),
903 ),
904 );
905 await this.db.prepare("DELETE FROM backfills WHERE workspace = ?").bind(old).run();
906 }
907 // The index's rows carry the slug: build them again under the new one.
908 const actor = await this.workspaceActor(current);
909 if (actor) await this.backfill({ actor, workspace: current });
910 return;
911 }
912 default:
913 return;
914 }
915 }
916
917 // ---- Backfill ------------------------------------------------------------
918
919 private async backfillRow(workspace: string): Promise<Backfill | null> {
920 const row = await this.db.prepare("SELECT * FROM backfills WHERE workspace = ?").bind(workspace).first<Record<string, unknown>>();
921 if (!row) return null;
922 return {
923 workspace,
924 status: row.status as Backfill["status"],
925 by: String(row.by),
926 projects: Number(row.projects),
927 done: Number(row.done),
928 entities: Number(row.entities),
929 candidates: Number(row.candidates),
930 kept: Number(row.kept),
931 indexed: Number(row.indexed),
932 error: (row.error as string | null) ?? null,
933 startedAt: String(row.started_at),
934 finishedAt: (row.finished_at as string | null) ?? null,
935 };
936 }
937
938 async backfill(a: { actor: User; workspace: string }): Promise<Result<Backfill>> {
939 const workspace = a.workspace.toLowerCase();
940 if (!isMember(a.actor, workspace)) return fail("forbidden", "Only members can rebuild a workspace's context.");
941 const running = await this.backfillRow(workspace);
942 if (running?.status === "running" && Date.now() - Date.parse(running.startedAt) < BACKFILL_STALE_MS) return ok(running);
943 const actor = (await this.workspaceActor(workspace)) ?? a.actor;
944 const listed = await projectsClient(this.env.PROJECTS).list(workspace, actor);
945 if (!listed.ok) return listed;
946 const projects = listed.value.filter((project) => project.source.kind === "hosted").slice(0, BACKFILL_PROJECTS);
947 const at = now();
948 await this.db
949 .prepare(
950 `INSERT OR REPLACE INTO backfills (workspace, status, by, projects, done, entities, candidates, kept, indexed, error, started_at, finished_at)
951 VALUES (?, ?, ?, ?, 0, 0, 0, 0, 0, NULL, ?, ?)`,
952 )
953 .bind(workspace, projects.length ? "running" : "done", a.actor.username, projects.length, at, projects.length ? null : at)
954 .run();
955 const jobs: { body: Job }[] = [
956 ...projects.map((project) => ({ body: { type: "backfill_project", workspace, slug: project.slug } as Job })),
957 { body: { type: "backfill_memory", workspace } },
958 ];
959 for (let i = 0; i < jobs.length; i += 100) await this.env.JOBS.sendBatch(jobs.slice(i, i + 100));
960 return ok((await this.backfillRow(workspace))!);
961 }
962
963 async runJob(job: Job): Promise<void> {
964 const actor = await this.workspaceActor(job.workspace);
965 if (!actor) return;
966 if (job.type === "backfill_memory") {
967 const memories = await memoryReviewClient(this.env.WORK).searchMemories(job.workspace, null, { limit: 100 });
968 const indexed = await this.indexMemories(job.workspace, memories);
969 await this.db.prepare("UPDATE backfills SET indexed = indexed + ? WHERE workspace = ?").bind(indexed, job.workspace).run();
970 return;
971 }
972 const stats: ScanStats = { entities: 0, candidates: 0, kept: 0, indexed: 0 };
973 let error: string | null = null;
974 try {
975 const found = await projectsClient(this.env.PROJECTS).get(job.workspace, job.slug, actor);
976 if (found.ok && found.value.source.kind === "hosted") {
977 const project = found.value;
978 const source = project.source as Extract<Project["source"], { kind: "hosted" }>;
979 const cache = new WorkspaceCache(this.env, job.workspace, actor);
980 Object.assign(stats, await this.scan(project, cache, null, true));
981 if (project.primary) {
982 // Decisions from merged pull requests, and people's corrections in their reviews.
983 const seeded = await memoryReviewClient(this.env.WORK).seedFromPulls(source.repoId, BACKFILL_PULLS).catch(() => null);
984 stats.candidates += seeded?.added ?? 0;
985 stats.kept += seeded?.kept ?? 0;
986 const work = workClient(this.env.WORK);
987 const base = `/${source.repo.namespace}/${source.repo.name}`;
988 const [issues, pulls] = await Promise.all([
989 work.listIssues(source.repo, actor).catch(() => null),
990 work.listPulls(source.repo, actor, "closed").catch(() => null),
991 ]);
992 stats.indexed += await this.putItems(job.workspace, project, source.repoId, [
993 ...(issues?.ok ? issues.value.slice(0, BACKFILL_ITEMS) : []).map((issue) => ({
994 kind: "issue" as const,
995 number: issue.number,
996 title: issue.title,
997 body: issue.body,
998 by: issue.author.username,
999 updatedAt: issue.updatedAt,
1000 url: `${base}/issues/${issue.number}`,
1001 private: project.private,
1002 })),
1003 ...(pulls?.ok ? pulls.value.filter((pull) => pull.status === "merged").slice(0, BACKFILL_ITEMS) : []).map((pull) => ({
1004 kind: "pull" as const,
1005 number: pull.number,
1006 title: pull.title,
1007 body: pull.body,
1008 by: (pull as { author?: { username: string } }).author?.username ?? pull.agent,
1009 updatedAt: pull.mergedAt ?? now(),
1010 url: `${base}/pull/${pull.number}`,
1011 private: project.private,
1012 })),
1013 ]);
1014 }
1015 }
1016 } catch (caught) {
1017 error = String(caught).slice(0, 300);
1018 console.error("backfill of", job.workspace, job.slug, "failed", caught);
1019 }
1020 await this.db
1021 .prepare(
1022 `UPDATE backfills SET done = done + 1, entities = entities + ?, candidates = candidates + ?, kept = kept + ?, indexed = indexed + ?,
1023 error = COALESCE(?, error),
1024 status = CASE WHEN done + 1 >= projects THEN 'done' ELSE status END,
1025 finished_at = CASE WHEN done + 1 >= projects THEN ? ELSE finished_at END
1026 WHERE workspace = ?`,
1027 )
1028 .bind(stats.entities, stats.candidates, stats.kept, stats.indexed, error, now(), job.workspace)
1029 .run();
1030 }
1031
1032 // ---- Reading -------------------------------------------------------------
1033
1034 /**
1035 * Who is reading, and what of the workspace they may see. Someone who can
1036 * read every repository in it (an owner, a member while its base
1037 * permission is Read or more, its own token) sees everything; anyone else
1038 * sees the projects the projects service lists for them, which are those
1039 * whose repositories they can read.
1040 */
1041 private async reader(workspace: string, viewer: Viewer): Promise<Reader> {
1042 const member = isMember(viewer, workspace);
1043 const full = member && !!viewer && granted(viewer, { id: "", namespace: workspace, isPrivate: true }) != null;
1044 if (full) return { workspace, member, full, visible: new Set() };
1045 const listed = await projectsClient(this.env.PROJECTS).list(workspace, viewer).catch(() => null);
1046 const projects = listed?.ok ? listed.value : [];
1047 return {
1048 workspace,
1049 member,
1050 full,
1051 visible: new Set(projects.map((p) => p.slug)),
1052 privateVisible: projects.some((p) => p.private),
1053 repos: new Set(
1054 projects.flatMap((p) => (p.source.kind === "hosted" ? [`${p.source.repo.namespace}/${p.source.repo.name}`.toLowerCase()] : [])),
1055 ),
1056 };
1057 }
1058
1059 private visibleRow(row: { private: number; project: string | null }, reader: Reader): boolean {
1060 return readable({ workspace: reader.workspace, kind: "entity", project: row.project ?? "", private: !!row.private }, reader);
1061 }
1062
1063 async status(a: { workspace: string; viewer: Viewer }): Promise<Result<ContextStatus>> {
1064 const workspace = a.workspace.toLowerCase();
1065 if (!isMember(a.viewer, workspace)) return fail("not_found", "There is no such workspace.");
1066 let backfill = await this.backfillRow(workspace);
1067 // The hub is never empty for long: a workspace's first look starts it.
1068 if (!backfill) {
1069 const actor = await this.workspaceActor(workspace);
1070 if (actor) {
1071 const started = await this.backfill({ actor: { ...actor, username: "g1t" }, workspace });
1072 backfill = started.ok ? started.value : null;
1073 }
1074 }
1075 // Counted over what the viewer may read: a member whose base permission
1076 // is None counts only the projects they were given.
1077 const reader = await this.reader(workspace, a.viewer);
1078 const [counted, usage] = await Promise.all([
1079 this.db
1080 .prepare("SELECT kind, project, private, COUNT(*) AS n FROM entities WHERE workspace = ? GROUP BY kind, project, private")
1081 .bind(workspace)
1082 .all<{ kind: EntityKind; project: string | null; private: number; n: number }>(),
1083 this.db.prepare("SELECT tokens, cost_micros FROM usage WHERE workspace = ? AND month = ?").bind(workspace, month()).first<{ tokens: number; cost_micros: number }>(),
1084 ]);
1085 return ok({
1086 backfill,
1087 counts: countVisible(counted.results, reader),
1088 usage: { month: month(), tokens: usage?.tokens ?? 0, costMicros: usage?.cost_micros ?? 0 },
1089 // Embeddings are compute: a paid plan or the trial (semanticOpen).
1090 semantic: !!(this.env.AI && this.env.VECTORS) && (await this.semanticOpen(workspace)),
1091 });
1092 }
1093
1094 async catalog(a: { workspace: string; viewer: Viewer; kind?: EntityKind | null; project?: string | null }): Promise<Result<Catalog>> {
1095 const workspace = a.workspace.toLowerCase();
1096 const reader = await this.reader(workspace, a.viewer);
1097 let sql = "SELECT * FROM entities WHERE workspace = ?";
1098 const params: unknown[] = [workspace];
1099 if (a.kind) {
1100 sql += " AND kind = ?";
1101 params.push(a.kind);
1102 }
1103 if (a.project) {
1104 sql += " AND (project = ? OR project IS NULL)";
1105 params.push(a.project.toLowerCase());
1106 }
1107 sql += " ORDER BY kind, name COLLATE NOCASE LIMIT 2000";
1108 const rows = (await this.db.prepare(sql).bind(...params).all<EntityRow>()).results.filter((row) => this.visibleRow(row, reader));
1109 const ids = new Set(rows.map((row) => row.id));
1110 const relations = (
1111 await this.db.prepare("SELECT from_id, kind, to_id FROM relations WHERE workspace = ? LIMIT 10000").bind(workspace).all<{ from_id: string; kind: RelationKind; to_id: string }>()
1112 ).results
1113 .filter((row) => ids.has(row.from_id) && ids.has(row.to_id))
1114 .map((row) => ({ from: row.from_id, kind: row.kind, to: row.to_id }));
1115 const built = await this.db.prepare("SELECT MAX(scanned_at) AS at FROM scans WHERE workspace = ?").bind(workspace).first<{ at: string | null }>();
1116 return ok({ entities: rows.map(toEntity), relations, builtAt: built?.at ?? null });
1117 }
1118
1119 async entity(a: { workspace: string; viewer: Viewer; kind: EntityKind; id: string }): Promise<Result<EntityDetail>> {
1120 const workspace = a.workspace.toLowerCase();
1121 if (!ENTITY_KINDS.includes(a.kind)) return fail("invalid", `kind is one of ${ENTITY_KINDS.join(", ")}.`);
1122 const reader = await this.reader(workspace, a.viewer);
1123 const row =
1124 (await this.db.prepare("SELECT * FROM entities WHERE workspace = ? AND id = ?").bind(workspace, a.id).first<EntityRow>()) ??
1125 (await this.db.prepare("SELECT * FROM entities WHERE workspace = ? AND kind = ? AND key = ? COLLATE NOCASE").bind(workspace, a.kind, a.id).first<EntityRow>());
1126 if (!row || row.kind !== a.kind || !this.visibleRow(row, reader)) return fail("not_found", `There is no such ${a.kind} in ${workspace}'s catalog.`);
1127 const [out, into] = await Promise.all([
1128 this.db
1129 .prepare("SELECT r.kind AS rel, e.* FROM relations r JOIN entities e ON e.id = r.to_id WHERE r.from_id = ? LIMIT 200")
1130 .bind(row.id)
1131 .all<EntityRow & { rel: RelationKind }>(),
1132 this.db
1133 .prepare("SELECT r.kind AS rel, e.* FROM relations r JOIN entities e ON e.id = r.from_id WHERE r.to_id = ? LIMIT 200")
1134 .bind(row.id)
1135 .all<EntityRow & { rel: RelationKind }>(),
1136 ]);
1137 const relations = [
1138 ...out.results.filter((r) => this.visibleRow(r, reader)).map((r) => ({ kind: r.rel, direction: "out" as const, entity: toEntity(r) })),
1139 ...into.results.filter((r) => this.visibleRow(r, reader)).map((r) => ({ kind: r.rel, direction: "in" as const, entity: toEntity(r) })),
1140 ];
1141 return ok({ entity: toEntity(row), relations });
1142 }
1143
1144 async search(a: { workspace: string; viewer: Viewer; query: string; project?: string | null; kinds?: SearchKind[] | null; limit?: number | null }): Promise<Result<SearchResult>> {
1145 const workspace = a.workspace.toLowerCase();
1146 const query = (a.query ?? "").trim().slice(0, 500);
1147 if (!query) return fail("invalid", "Say what to search for.");
1148 const reader = await this.reader(workspace, a.viewer);
1149 if (!reader.full && !reader.member && reader.visible.size === 0) return ok({ query, hits: [], mode: "text" });
1150 const limit = Math.min(Math.max(a.limit ?? 20, 1), 50);
1151 const kinds = allowedKinds(reader, a.kinds);
1152 const wants = (kind: SearchKind) => !kinds || kinds.includes(kind);
1153 const project = a.project?.toLowerCase() || null;
1154
1155 // Semantic: the index, filtered to what this reader may see.
1156 let semantic: SearchHit[] = [];
1157 let mode: SearchResult["mode"] = "text";
1158 const vector = await this.embedQuery(workspace, query);
1159 if (vector && this.env.VECTORS) {
1160 try {
1161 const found = await this.env.VECTORS.query(vector, { topK: 20, returnMetadata: "all", filter: indexFilter(reader, { project, kinds }) as VectorizeVectorMetadataFilter });
1162 mode = "semantic";
1163 semantic = found.matches
1164 .map((match) => ({ id: match.id, score: match.score, meta: match.metadata as unknown as IndexMeta }))
1165 .filter((match) => match.meta && readable(match.meta, reader))
1166 .map((match) => ({
1167 kind: match.meta.kind as SearchKind,
1168 id: match.id.startsWith("memory:") ? match.id.slice(7) : match.id,
1169 title: match.meta.title,
1170 snippet: match.meta.snippet,
1171 project: match.meta.project || null,
1172 url: match.meta.url || null,
1173 score: match.score,
1174 source: match.meta.source,
1175 by: match.meta.by || null,
1176 updatedAt: match.meta.at || null,
1177 }));
1178 // Memory changes after it is indexed: show only what is still kept.
1179 const memoryIds = semantic.filter((hit) => hit.kind === "memory").map((hit) => hit.id);
1180 if (memoryIds.length) {
1181 const kept = new Set(
1182 (await memoryReviewClient(this.env.WORK).memoriesById(workspace, memoryIds))
1183 .filter((memory) => (memory.status ?? "kept") === "kept" && memoryReadable(memory.repo, reader))
1184 .map((memory) => memory.id),
1185 );
1186 semantic = semantic.filter((hit) => hit.kind !== "memory" || kept.has(hit.id));
1187 }
1188 } catch (error) {
1189 console.error("semantic search failed; matching words instead", error);
1190 }
1191 }
1192
1193 // Text: every word, in the catalog, docs, issues and pull requests, and memory.
1194 const words = query.toLowerCase().split(/\s+/).filter(Boolean).slice(0, 6);
1195 const like = (columns: string) => words.map(() => `(${columns}) LIKE ?`).join(" AND ");
1196 const patterns = words.map((word) => `%${word.replace(/[%_]/g, "")}%`);
1197 const text: SearchHit[] = [];
1198 const entityKinds = ENTITY_KINDS.filter(wants);
1199 if (entityKinds.length) {
1200 const rows = await this.db
1201 .prepare(
1202 `SELECT * FROM entities WHERE workspace = ? AND kind IN (SELECT value FROM json_each(?)) ${project ? "AND (project = ? OR project IS NULL)" : ""}
1203 AND ${like("lower(name || ' ' || COALESCE(summary, ''))")} LIMIT 30`,
1204 )
1205 .bind(workspace, JSON.stringify(entityKinds), ...(project ? [project] : []), ...patterns)
1206 .all<EntityRow>();
1207 for (const row of rows.results.filter((r) => this.visibleRow(r, reader))) {
1208 text.push({ kind: row.kind, id: row.id, title: row.name, snippet: snippet(row.summary), project: row.project, url: row.ref, score: 0.3, source: "catalog", by: null, updatedAt: row.updated_at });
1209 }
1210 }
1211 const itemKinds = (["doc", "issue", "pull"] as const).filter(wants);
1212 if (itemKinds.length) {
1213 const rows = await this.db
1214 .prepare(
1215 `SELECT * FROM items WHERE workspace = ? AND kind IN (SELECT value FROM json_each(?)) ${project ? "AND project = ?" : ""}
1216 AND ${like("lower(title || ' ' || text)")} ORDER BY updated_at DESC LIMIT 30`,
1217 )
1218 .bind(workspace, JSON.stringify(itemKinds), ...(project ? [project] : []), ...patterns)
1219 .all<ItemRow>();
1220 for (const row of rows.results.filter((r) => this.visibleRow(r, reader))) {
1221 text.push({ kind: row.kind, id: row.id, title: row.title, snippet: snippet(row.text), project: row.project, url: row.url, score: 0.2, source: row.kind === "doc" ? "doc" : row.kind, by: row.by, updatedAt: row.updated_at });
1222 }
1223 }
1224 if (reader.member && wants("memory")) {
1225 const memories = await memoryReviewClient(this.env.WORK).searchMemories(workspace, query, { limit: 10 }).catch(() => [] as Memory[]);
1226 for (const memory of memories.filter((m) => memoryReadable(m.repo, reader))) {
1227 text.push({
1228 kind: "memory",
1229 id: memory.id,
1230 title: `${memory.kind[0].toUpperCase()}${memory.kind.slice(1)}${memory.scope === "workspace" ? " (workspace)" : ""}`,
1231 snippet: snippet(memory.text),
1232 project: null,
1233 url: memory.repo ? `/${memory.repo.namespace}/${memory.repo.name}/memory` : `/${workspace}/-/memory`,
1234 score: memory.pinned ? 0.35 : 0.25,
1235 source: memorySource(memory),
1236 by: memory.createdBy,
1237 updatedAt: memory.updatedAt,
1238 });
1239 }
1240 }
1241 return ok({ query, hits: merge(semantic, text, limit), mode });
1242 }
1243
1244 async scorecards(a: { workspace: string; viewer: Viewer }): Promise<Result<Scorecard[]>> {
1245 const workspace = a.workspace.toLowerCase();
1246 if (!isMember(a.viewer, workspace)) return fail("forbidden", "Scorecards are for members of the workspace.");
1247 const listed = await projectsClient(this.env.PROJECTS).list(workspace, a.viewer);
1248 if (!listed.ok) return listed;
1249 const [entities, scans, deploys, security] = await Promise.all([
1250 this.db
1251 .prepare("SELECT * FROM entities WHERE workspace = ? AND kind IN ('project', 'doc')")
1252 .bind(workspace)
1253 .all<EntityRow>(),
1254 this.db.prepare("SELECT project_id, tests FROM scans WHERE workspace = ?").bind(workspace).all<{ project_id: string; tests: number }>(),
1255 deploymentsClient(this.env.DEPLOYMENTS)
1256 .overview(workspace, a.viewer)
1257 .then((found) => (found.ok ? found.value : []))
1258 .catch(() => [] as ProjectDeploys[]),
1259 this.env.SECURITY
1260 ? securityClient(this.env.SECURITY)
1261 .workspace(workspace, a.viewer)
1262 .then((found) => (found.ok ? found.value : null))
1263 .catch(() => null)
1264 : Promise.resolve(null),
1265 ]);
1266 const tests = new Map(scans.results.map((row) => [row.project_id, !!row.tests]));
1267 const cards: Scorecard[] = [];
1268 for (const project of listed.value) {
1269 if (project.source.kind !== "hosted") continue;
1270 const own = entities.results.filter((row) => row.project_id === project.id);
1271 const entry = own.find((row) => row.kind === "project");
1272 const data = entry ? toEntity(entry).data : {};
1273 const deploy = deploys.find((d) => d.slug === project.slug);
1274 const repoId = project.source.repoId;
1275 const findings = security?.find((repo) => repo.repoId === repoId);
1276 const rules = evaluate({
1277 name: project.name,
1278 owners: Array.isArray(data.owners) ? (data.owners as string[]) : [],
1279 docs: own.filter((row) => row.kind === "doc").map((row) => String(toEntity(row).data.path ?? "")),
1280 tests: tests.get(project.id) ?? false,
1281 testCommand: Array.isArray(data.testCommands) && data.testCommands.length ? String(data.testCommands[0]) : null,
1282 deploy: deploy
1283 ? {
1284 enabled: deploy.enabled,
1285 production: deploy.production ? { url: deploy.production.url } : null,
1286 latest: deploy.latest ? { kind: deploy.latest.kind, status: deploy.latest.status, error: deploy.latest.error } : null,
1287 }
1288 : null,
1289 secretFindings: security ? (findings?.secrets ?? 0) : null,
1290 });
1291 const applies = rules.filter((rule) => rule.status !== "na");
1292 cards.push({
1293 project: project.slug,
1294 name: project.name,
1295 repo: project.source.repo,
1296 passed: applies.filter((rule) => rule.status === "pass").length,
1297 total: applies.length,
1298 rules,
1299 });
1300 }
1301 return ok(cards);
1302 }
1303
1304 /**
1305 * The Context section for an agent starting work, holding only what the
1306 * person it acts for (`requester`) may read: an outside collaborator's run
1307 * is told the project's memory, never the workspace's, and only the
1308 * projects around it they can read. No requester is the workspace's own
1309 * step. Never fails a run: on any trouble, nothing.
1310 */
1311 async runContext(a: { repoId: string; task?: string; budget?: number; requester?: Viewer }): Promise<RunContext> {
1312 try {
1313 const projects = (await projectsClient(this.env.PROJECTS).byRepo(a.repoId)).slice(0, 2);
1314 if (projects.length === 0) return { text: null, sources: [] };
1315 const workspace = projects[0].workspace;
1316 const actor = await this.workspaceActor(workspace);
1317 if (!actor) return { text: null, sources: [] };
1318 const reader = a.requester ? await this.reader(workspace, a.requester) : null;
1319 const home = projects.find((project) => project.source.kind === "hosted");
1320 const repoKey = home?.source.kind === "hosted" ? `${home.source.repo.namespace}/${home.source.repo.name}` : "";
1321 const cache = new WorkspaceCache(this.env, workspace, actor);
1322 const deploys = await cache.deployments();
1323 const live = new Map(deploys.map((d) => [d.slug, d]));
1324 const contexts: ProjectContext[] = [];
1325 for (const project of projects) {
1326 if (project.source.kind !== "hosted") continue;
1327 const rows = (await this.db.prepare("SELECT * FROM entities WHERE workspace = ? AND project_id = ?").bind(workspace, project.id).all<EntityRow>()).results.map(toEntity);
1328 const entry = rows.find((row) => row.kind === "project");
1329 const graph = await projectsClient(this.env.PROJECTS).graph(project.id).catch(() => ({ dependsOn: [], usedBy: [] }));
1330 const deploy = live.get(project.slug);
1331 contexts.push({
1332 slug: project.slug,
1333 name: project.name,
1334 repo: `${project.source.repo.namespace}/${project.source.repo.name}`,
1335 rootDir: project.source.rootDir,
1336 languages: Array.isArray(entry?.data.languages) ? (entry!.data.languages as string[]) : [],
1337 packages: rows.filter((row) => row.kind === "package").map((row) => row.name).slice(0, 5),
1338 testCommands: Array.isArray(entry?.data.testCommands) ? (entry!.data.testCommands as string[]) : [],
1339 owners: Array.isArray(entry?.data.owners) ? (entry!.data.owners as string[]) : [],
1340 dependsOn: graph.dependsOn
1341 .filter((dep) => projectReadable(dep.slug, reader))
1342 .map((dep) => ({ slug: dep.slug, as: dep.as, url: live.get(dep.slug)?.production?.url ?? null })),
1343 usedBy: graph.usedBy.filter((dep) => projectReadable(dep.slug, reader)).map((dep) => ({ slug: dep.slug })),
1344 environments: deploy?.enabled
1345 ? [{ name: "Production", url: deploy.production?.url ?? null, status: deploy.latest?.kind === "production" ? deploy.latest.status : deploy.production ? "ready" : null }]
1346 : [],
1347 docs: rows.filter((row) => row.kind === "doc").map((row) => String(row.data.path ?? row.name)),
1348 });
1349 }
1350 // Workspace memory (no project) only for a run its members may be told it.
1351 const slugs = new Set([...(reader && !reader.member ? [] : [""]), ...projects.map((project) => project.slug)]);
1352 const review = memoryReviewClient(this.env.WORK);
1353 // The memories closest to the task, less the pinned ones every run already has.
1354 let memories: ContextNote[] = [];
1355 const task = (a.task ?? "").trim();
1356 const vector = task ? await this.embedQuery(workspace, task.slice(0, 1500)) : null;
1357 if (vector && this.env.VECTORS) {
1358 const found = await this.env.VECTORS.query(vector, { topK: 20, returnMetadata: "all", filter: { workspace, kind: "memory" } }).catch(() => null);
1359 const ids = (found?.matches ?? [])
1360 .filter((match) => slugs.has(String((match.metadata as { project?: string } | undefined)?.project ?? "")) && match.score >= 0.5)
1361 .map((match) => match.id.slice("memory:".length));
1362 if (ids.length) {
1363 const byId = new Map((await review.memoriesById(workspace, ids)).map((memory) => [memory.id, memory]));
1364 memories = ids
1365 .map((id) => byId.get(id))
1366 .filter((memory): memory is Memory => !!memory && (memory.status ?? "kept") === "kept" && !memory.pinned && runMemoryReadable(memory, reader, repoKey))
1367 .slice(0, 6)
1368 .map((memory) => ({ id: memory.id, kind: memory.kind, text: memory.text, source: memorySource(memory) }));
1369 }
1370 }
1371 const shown = new Set(memories.map((note) => note.id));
1372 const decisions = (await review.searchMemories(workspace, null, { repoIds: [a.repoId], limit: 100 }).catch(() => [] as Memory[]))
1373 .filter((memory) => memory.kind === "decision" && !memory.pinned && !shown.has(memory.id) && runMemoryReadable(memory, reader, repoKey))
1374 .sort((x, y) => y.updatedAt.localeCompare(x.updatedAt))
1375 .slice(0, 3)
1376 .map((memory) => ({ id: memory.id, kind: memory.kind, text: memory.text, source: memorySource(memory) }));
1377 return composeRunContext({ projects: contexts, memories, decisions, budget: Math.min(Math.max(a.budget ?? 4000, 500), 12_000) });
1378 } catch (error) {
1379 console.error("no run context for", a.repoId, error);
1380 return { text: null, sources: [] };
1381 }
1382 }
1383}
1384
1385export default {
1386 async fetch(request: Request, env: Env): Promise<Response> {
1387 const match = new URL(request.url).pathname.match(/^\/rpc\/([a-z_]+)$/);
1388 if (request.method !== "POST" || !match) return new Response("Not found\n", { status: 404 });
1389 const service = new Context(env);
1390 const args = (await request.json().catch(() => ({}))) as any;
1391 switch (match[1]) {
1392 case "catalog":
1393 return Response.json(await service.catalog(args));
1394 case "entity":
1395 return Response.json(await service.entity(args));
1396 case "search":
1397 return Response.json(await service.search(args));
1398 case "scorecards":
1399 return Response.json(await service.scorecards(args));
1400 case "backfill":
1401 return Response.json(await service.backfill(args));
1402 case "status":
1403 return Response.json(await service.status(args));
1404 case "run_context":
1405 return Response.json(await service.runContext(args));
1406 default:
1407 return new Response("Unknown method\n", { status: 404 });
1408 }
1409 },
1410
1411 async queue(batch: MessageBatch<G1tEvent | Job>, env: Env): Promise<void> {
1412 const service = new Context(env);
1413 for (const message of batch.messages) {
1414 try {
1415 if (batch.queue === "g1t-context-jobs") await service.runJob(message.body as Job);
1416 else await service.onEvent(message.body as G1tEvent);
1417 message.ack();
1418 } catch (error) {
1419 console.error("context could not handle", (message.body as { type?: string }).type, error);
1420 message.retry();
1421 }
1422 }
1423 },
1424} satisfies ExportedHandler<Env, G1tEvent | Job>;