Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
Blame · Line-by-line history

repo-chat.ts

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

Gluecron.com/src/lib/repo-chat.ts
repo-chat.tsBlame614 lines · 1 contributor
Normal viewRaw
38d31d3Claude1/**
2 * AI rubber-duck chat — repo-grounded conversation backed by the
3 * continuous semantic index (src/lib/semantic-index.ts) and Claude
4 * streaming.
5 *
6 * Why a new module instead of extending `ai-chat.ts`?
7 *
8 * - `ai-chat.ts` is built around a single-blob JSON history in
9 * `ai_chats.messages`. Streaming partials, per-message citations,
10 * and token-cost accounting would require rewriting the whole
11 * blob on each turn. The 0060 migration introduces dedicated
12 * `repo_chats` + `repo_chat_messages` tables, one row per
13 * message, which makes streaming and citations first-class.
14 * - Repo chat retrieves grounding context via the per-push semantic
15 * index, which is fundamentally different from the static
16 * README+tree summarisation in `ai-chat.ts`. Keeping the two
17 * codepaths separate avoids muddling the abstraction.
18 *
19 * Public surface:
20 *
21 * - `createChat({ repositoryId, ownerUserId, title? })` →
22 * creates a row in `repo_chats` and returns it.
23 * - `appendUserMessage(chatId, content)` → stores a `role:'user'`
24 * row in `repo_chat_messages` and returns it.
25 * - `streamAssistantReply({ chatId, repoId, userMessage, onChunk })`
26 * → resolves grounding context, streams Claude tokens via the
27 * `onChunk` callback, persists the final assistant message with
28 * citations, returns the stored row.
29 * - `__setStreamerForTests` → test seam so unit tests can replace
30 * the Claude streaming call with a deterministic generator.
31 *
32 * Hard rules:
33 * - Never throw at the boundary. Every external dependency
34 * (semantic index, git blob fetch, Claude API, DB) is wrapped;
35 * on failure we fall back to a safe default and continue.
36 * - Graceful when the semantic index is empty: fall back to a
37 * truncated tree-of-paths summary so the assistant still has
38 * *some* signal about the repo's shape.
39 * - All DB writes are best-effort; a DB outage degrades the chat
40 * to ephemeral mode rather than throwing.
41 */
42
43import { and, asc, desc, eq } from "drizzle-orm";
44import { db } from "../db";
45import {
46 repoChats,
47 repoChatMessages,
48 repositories,
49 users,
50 type RepoChat,
51 type RepoChatMessage,
52} from "../db/schema";
53import { searchSemantic, type SemanticHit } from "./semantic-index";
54import {
55 getBlob,
56 getDefaultBranch,
57 getTreeRecursive,
58} from "../git/repository";
59import { getAnthropic, isAiAvailable, MODEL_SONNET } from "./ai-client";
60
61// ---------------------------------------------------------------------------
62// Constants
63// ---------------------------------------------------------------------------
64
65/** Max semantic hits we feed the assistant. */
66const DEFAULT_TOP_K = 8;
67
68/** Max chars of file content per snippet (after path + heading). */
69const MAX_SNIPPET_CHARS = 1500;
70
71/** Max paths in the empty-index fallback tree summary. */
72const FALLBACK_TREE_CAP = 120;
73
74/** Hard cap on stored title length. */
75const TITLE_LIMIT = 80;
76
77/** Hard cap on the assistant reply we'll persist. */
78const ASSISTANT_REPLY_CAP = 32_000;
79
80// ---------------------------------------------------------------------------
81// Types
82// ---------------------------------------------------------------------------
83
84export interface Citation {
85 file_path: string;
86 blob_sha: string;
87}
88
89export interface CreateChatOpts {
90 repositoryId: string;
91 ownerUserId: string;
92 title?: string | null;
93}
94
95export interface StreamReplyOpts {
96 chatId: string;
97 repoId: string;
98 userMessage: string;
99 /**
100 * Called for each token / text delta emitted by the assistant.
101 * Implementations may ignore the chunk (e.g. tests that only care
102 * about the final reply) — the helper still accumulates and stores
103 * the full reply regardless.
104 */
105 onChunk?: (chunk: string) => void;
106 /**
107 * Optional override of the top-K semantic hits. Useful for tests
108 * + the SSE endpoint where the caller wants to thin the context
109 * for cost reasons.
110 */
111 topK?: number;
112}
113
114/**
115 * Test seam — replace the streaming call with a deterministic generator.
116 * Pass `null` to reset. Each chunk yielded becomes one token in the
117 * persisted reply (and one `onChunk` invocation).
118 */
119export type StreamerFn = (args: {
120 systemPrompt: string;
121 userMessage: string;
122}) => AsyncIterable<string>;
123
124let _streamerOverride: StreamerFn | null = null;
125
126export function __setStreamerForTests(fn: StreamerFn | null): void {
127 _streamerOverride = fn;
128}
129
130// ---------------------------------------------------------------------------
131// Public API
132// ---------------------------------------------------------------------------
133
134/**
135 * Create a new chat row scoped to (repository, owner_user). The title
136 * is optional; if absent we leave it null and the route layer can fill
137 * it in from the first user message.
138 */
139export async function createChat(opts: CreateChatOpts): Promise<RepoChat | null> {
140 if (!opts.repositoryId || !opts.ownerUserId) return null;
141 try {
142 const [row] = await db
143 .insert(repoChats)
144 .values({
145 repositoryId: opts.repositoryId,
146 ownerUserId: opts.ownerUserId,
147 title: (opts.title || "").slice(0, TITLE_LIMIT) || null,
148 })
149 .returning();
150 return row || null;
151 } catch (err) {
152 if (process.env.DEBUG_REPO_CHAT === "1") {
153 console.error("[repo-chat] createChat failed:", err);
154 }
155 return null;
156 }
157}
158
159/**
160 * Append a `role:'user'` row to the chat and bump `repo_chats.updated_at`
161 * so the chat list ordering stays fresh.
162 */
163export async function appendUserMessage(
164 chatId: string,
165 content: string
166): Promise<RepoChatMessage | null> {
167 if (!chatId || !content) return null;
168 try {
169 const [row] = await db
170 .insert(repoChatMessages)
171 .values({
172 chatId,
173 role: "user",
174 content,
175 citations: [],
176 tokenCost: 0,
177 })
178 .returning();
179 // Best-effort: keep the parent chat's `updated_at` warm.
180 try {
181 await db
182 .update(repoChats)
183 .set({ updatedAt: new Date() })
184 .where(eq(repoChats.id, chatId));
185 } catch {
186 /* tolerate */
187 }
188 return row || null;
189 } catch (err) {
190 if (process.env.DEBUG_REPO_CHAT === "1") {
191 console.error("[repo-chat] appendUserMessage failed:", err);
192 }
193 return null;
194 }
195}
196
197/**
198 * The full pipeline: resolve repo context (semantic hits → snippets →
199 * system prompt), stream Claude's reply token-by-token via `onChunk`,
200 * persist the final assistant row with citations + a coarse token
201 * cost estimate, and return it.
202 *
203 * Never throws. On any failure inside grounding/streaming we still
204 * try to persist a short advisory assistant message so the UI never
205 * shows a "phantom" user message with no reply.
206 */
207export async function streamAssistantReply(
208 opts: StreamReplyOpts
209): Promise<RepoChatMessage | null> {
210 const { chatId, repoId, userMessage } = opts;
211 const topK = Math.max(1, Math.min(opts.topK ?? DEFAULT_TOP_K, 20));
212
213 // 1. Resolve grounding context.
214 const { citations, contextBlock } = await buildGroundingContext({
215 repoId,
216 userMessage,
217 topK,
218 });
219
220 // 2. Build the system prompt.
221 const systemPrompt = await buildSystemPrompt({
222 repoId,
223 contextBlock,
224 });
225
226 // 3. Stream Claude's reply (or canned tokens in tests).
227 let reply = "";
228 try {
229 const stream = _streamerOverride
230 ? _streamerOverride({ systemPrompt, userMessage })
231 : claudeStream({ systemPrompt, userMessage });
232
233 for await (const chunk of stream) {
234 if (!chunk) continue;
235 reply += chunk;
236 if (opts.onChunk) {
237 try {
238 opts.onChunk(chunk);
239 } catch {
240 // Caller-supplied callback errors mustn't kill the stream.
241 }
242 }
243 // Safety cap — Claude can produce very long outputs; we don't
244 // want a runaway response to blow up our row.
245 if (reply.length >= ASSISTANT_REPLY_CAP) break;
246 }
247 } catch (err) {
248 if (process.env.DEBUG_REPO_CHAT === "1") {
249 console.error("[repo-chat] stream failed:", err);
250 }
251 if (!reply) {
252 reply =
253 "Sorry — I couldn't reach the AI service to answer that. Please retry in a moment.";
254 }
255 }
256
257 // Cap the persisted reply.
258 if (reply.length > ASSISTANT_REPLY_CAP) {
259 reply = reply.slice(0, ASSISTANT_REPLY_CAP);
260 }
261
262 // 4. Coarse token-cost estimate: ~4 chars/token. Stored as integer.
263 const tokenCost = Math.ceil(
264 (systemPrompt.length + userMessage.length + reply.length) / 4
265 );
266
267 // 5. Persist.
268 try {
269 const [row] = await db
270 .insert(repoChatMessages)
271 .values({
272 chatId,
273 role: "assistant",
274 content: reply,
275 citations,
276 tokenCost,
277 })
278 .returning();
279 // Bump parent chat freshness so list ordering reflects the answer.
280 try {
281 await db
282 .update(repoChats)
283 .set({ updatedAt: new Date() })
284 .where(eq(repoChats.id, chatId));
285 } catch {
286 /* tolerate */
287 }
288 return row || null;
289 } catch (err) {
290 if (process.env.DEBUG_REPO_CHAT === "1") {
291 console.error("[repo-chat] persist assistant failed:", err);
292 }
293 return null;
294 }
295}
296
297/**
298 * List all chats for a (user, repo) pair, ordered by most-recently
299 * updated. Returns [] on any DB failure.
300 */
301export async function listChatsForRepo(
302 ownerUserId: string,
303 repositoryId: string,
304 limit = 30
305): Promise<RepoChat[]> {
306 if (!ownerUserId || !repositoryId) return [];
307 try {
308 return await db
309 .select()
310 .from(repoChats)
311 .where(
312 and(
313 eq(repoChats.ownerUserId, ownerUserId),
314 eq(repoChats.repositoryId, repositoryId)
315 )
316 )
317 .orderBy(desc(repoChats.updatedAt))
318 .limit(Math.max(1, Math.min(limit, 100)));
319 } catch {
320 return [];
321 }
322}
323
324/**
325 * Fetch all messages in a chat, oldest first. Empty array on DB error.
326 */
327export async function listMessages(
328 chatId: string
329): Promise<RepoChatMessage[]> {
330 if (!chatId) return [];
331 try {
332 return await db
333 .select()
334 .from(repoChatMessages)
335 .where(eq(repoChatMessages.chatId, chatId))
336 .orderBy(asc(repoChatMessages.createdAt));
337 } catch {
338 return [];
339 }
340}
341
342/**
343 * Verify the chat belongs to the user. Returns the chat row or null.
344 * Used by the route handlers + the SSE endpoint to authorise access.
345 */
346export async function getChatForUser(
347 chatId: string,
348 ownerUserId: string
349): Promise<RepoChat | null> {
350 if (!chatId || !ownerUserId) return null;
351 try {
352 const [row] = await db
353 .select()
354 .from(repoChats)
355 .where(
356 and(eq(repoChats.id, chatId), eq(repoChats.ownerUserId, ownerUserId))
357 )
358 .limit(1);
359 return row || null;
360 } catch {
361 return null;
362 }
363}
364
365// ---------------------------------------------------------------------------
366// Grounding context — semantic hits + snippet fetch, with tree fallback.
367// ---------------------------------------------------------------------------
368
369async function buildGroundingContext(args: {
370 repoId: string;
371 userMessage: string;
372 topK: number;
373}): Promise<{ citations: Citation[]; contextBlock: string }> {
374 const { repoId, userMessage, topK } = args;
375
376 let hits: SemanticHit[] = [];
377 try {
378 hits = await searchSemantic({
379 repositoryId: repoId,
380 query: userMessage,
381 limit: topK,
382 });
383 } catch {
384 hits = [];
385 }
386
387 if (!hits.length) {
388 // Fallback: tree-of-paths summary so Claude still has *some*
389 // sense of repo shape. Best-effort — empty string on failure.
390 const treeSummary = await buildTreeFallback(repoId);
391 return {
392 citations: [],
393 contextBlock: treeSummary
394 ? `No semantic index is available for this repo (yet). Repo layout:\n\n${treeSummary}`
395 : "",
396 };
397 }
398
399 // Resolve owner/name for getBlob lookups.
400 const repoMeta = await loadRepoMeta(repoId);
401
402 const citations: Citation[] = [];
403 const sections: string[] = [];
404
405 for (const hit of hits) {
406 // The semantic-index already stores a snippet; prefer that for
407 // cost, but fetch a slightly larger window via getBlob when meta
408 // is available so the model sees real code, not a 500-char preview.
409 let snippet = hit.snippet || "";
410 if (repoMeta) {
411 try {
412 const blob = await getBlob(
413 repoMeta.owner,
414 repoMeta.name,
415 repoMeta.defaultBranch,
416 hit.filePath
417 );
418 if (blob && !blob.isBinary && blob.content) {
419 snippet = blob.content.slice(0, MAX_SNIPPET_CHARS);
420 }
421 } catch {
422 // tolerate; fall back to whatever snippet we already have
423 }
424 }
425 if (!snippet) continue;
426
427 citations.push({
428 file_path: hit.filePath,
429 blob_sha: hit.blobSha,
430 });
431 sections.push(`### ${hit.filePath}\n\`\`\`\n${snippet}\n\`\`\``);
432 }
433
434 return {
435 citations,
436 contextBlock: sections.join("\n\n"),
437 };
438}
439
440async function buildTreeFallback(repoId: string): Promise<string> {
441 const meta = await loadRepoMeta(repoId);
442 if (!meta) return "";
443 try {
444 const tree = await getTreeRecursive(
445 meta.owner,
446 meta.name,
447 meta.defaultBranch,
448 FALLBACK_TREE_CAP * 2
449 );
450 if (!tree) return "";
451 const blobs = tree.tree
452 .filter((e) => e.type === "blob")
453 .slice(0, FALLBACK_TREE_CAP)
454 .map((e) => `- ${e.path}`);
455 return blobs.join("\n");
456 } catch {
457 return "";
458 }
459}
460
461async function buildSystemPrompt(args: {
462 repoId: string;
463 contextBlock: string;
464}): Promise<string> {
465 const meta = await loadRepoMeta(args.repoId);
466 const owner = meta?.owner || "unknown";
467 const name = meta?.name || "repo";
468
469 const header = [
470 `You are Gluecron's repo chat for ${owner}/${name}.`,
471 `Here are the most relevant files based on the user's question:`,
472 "",
473 args.contextBlock || "(no grounding context available)",
474 "",
475 `Answer concisely. Cite files as [path](/${owner}/${name}/blob/HEAD/path).`,
476 `Prefer code snippets over prose when explaining concrete behaviour.`,
477 `If the grounding context doesn't cover the question, say so plainly rather than guessing.`,
478 ];
479 return header.join("\n");
480}
481
482// ---------------------------------------------------------------------------
483// Repo metadata helper — owner/name/defaultBranch for git lookups.
484// ---------------------------------------------------------------------------
485
486interface RepoMeta {
487 owner: string;
488 name: string;
489 defaultBranch: string;
490}
491
492async function loadRepoMeta(repoId: string): Promise<RepoMeta | null> {
493 if (!repoId) return null;
494 try {
495 const [row] = await db
496 .select({
497 name: repositories.name,
498 defaultBranch: repositories.defaultBranch,
499 owner: users.username,
500 })
501 .from(repositories)
502 .innerJoin(users, eq(repositories.ownerId, users.id))
503 .where(eq(repositories.id, repoId))
504 .limit(1);
505 if (!row) return null;
506 let defaultBranch = row.defaultBranch || "main";
507 // If the configured default branch isn't actually resolvable on
508 // disk, fall back to whatever HEAD points at.
509 try {
510 const real = await getDefaultBranch(row.owner, row.name);
511 if (real) defaultBranch = real;
512 } catch {
513 /* tolerate */
514 }
515 return { owner: row.owner, name: row.name, defaultBranch };
516 } catch {
517 return null;
518 }
519}
520
521// ---------------------------------------------------------------------------
522// Claude streaming — thin wrapper that yields text deltas.
523// ---------------------------------------------------------------------------
524
525async function* claudeStream(args: {
526 systemPrompt: string;
527 userMessage: string;
528}): AsyncGenerator<string, void, unknown> {
529 if (!isAiAvailable()) {
530 yield "AI is not configured on this Gluecron instance — set ANTHROPIC_API_KEY to enable rubber-duck chat.";
531 return;
532 }
533
534 const client = getAnthropic();
535 // Anthropic SDK exposes `.stream(...)` returning an event emitter
536 // with an async iterator over RawMessageStreamEvent chunks. We
537 // extract `content_block_delta` text deltas.
538 const stream = client.messages.stream({
539 model: MODEL_SONNET,
540 max_tokens: 2048,
541 system: args.systemPrompt,
542 messages: [{ role: "user", content: args.userMessage }],
543 });
544
545 // The SDK's stream object is itself async-iterable over events.
0c3eee5Claude546 // Anthropic emits `message_start` / `message_delta` events that carry
547 // usage data — sample those so we can attribute cost without buffering
548 // the whole response.
549 let inputTokens = 0;
550 let outputTokens = 0;
38d31d3Claude551 for await (const event of stream as AsyncIterable<unknown>) {
0c3eee5Claude552 const ev = event as Record<string, unknown> | null;
553 if (ev && typeof ev === "object") {
554 // message_start.message.usage.input_tokens
555 const msg = (ev as { message?: { usage?: { input_tokens?: number; output_tokens?: number } } }).message;
556 if (msg && msg.usage) {
557 if (typeof msg.usage.input_tokens === "number")
558 inputTokens = msg.usage.input_tokens;
559 if (typeof msg.usage.output_tokens === "number")
560 outputTokens = msg.usage.output_tokens;
561 }
562 // message_delta.usage.output_tokens (incremental output tokens)
563 const usage = (ev as { usage?: { input_tokens?: number; output_tokens?: number } }).usage;
564 if (usage) {
565 if (typeof usage.input_tokens === "number")
566 inputTokens = usage.input_tokens;
567 if (typeof usage.output_tokens === "number")
568 outputTokens = usage.output_tokens;
569 }
570 }
38d31d3Claude571 const delta = extractTextDelta(event);
572 if (delta) yield delta;
573 }
0c3eee5Claude574
575 // Record cost AFTER the stream completes so output_tokens is final.
576 try {
577 const { recordAiCost } = await import("./ai-cost-tracker");
578 await recordAiCost({
579 model: MODEL_SONNET,
580 inputTokens,
581 outputTokens,
582 category: "chat",
583 sourceKind: "repo_chat",
584 });
585 } catch {
586 /* swallow — best-effort */
587 }
38d31d3Claude588}
589
590function extractTextDelta(event: unknown): string {
591 if (!event || typeof event !== "object") return "";
592 const e = event as Record<string, unknown>;
593 if (e.type !== "content_block_delta") return "";
594 const delta = e.delta as Record<string, unknown> | undefined;
595 if (!delta) return "";
596 if (delta.type === "text_delta" && typeof delta.text === "string") {
597 return delta.text;
598 }
599 return "";
600}
601
602// ---------------------------------------------------------------------------
603// Test-only exports.
604// ---------------------------------------------------------------------------
605
606export const __test = {
607 buildGroundingContext,
608 buildSystemPrompt,
609 buildTreeFallback,
610 extractTextDelta,
611 ASSISTANT_REPLY_CAP,
612 MAX_SNIPPET_CHARS,
613 FALLBACK_TREE_CAP,
614};