Skip to content
Sign up

flagon-io/g1t

public
00

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

g1t.sh
gitagentscloudflarerustforgemcp
301 linesCodeBlameRaw
1/**
2 * @g1t in chat (docs.g1t.sh/guides/ask/): every member has a direct message
3 * with g1t, and a person writing in it starts an answer from the
4 * workspace's own AI provider, through the integrations service's `ask`.
5 * The top bar's Ask is the same conversation.
6 *
7 * What is decided here is pure, so it is tested on its own (g1t.test.ts):
8 * which direct messages are with g1t, which messages start an answer, the
9 * conversation an answer reads, what g1t says when Ask is not set up, and
10 * reading the answer as it streams. The room (src/room.ts) runs answers;
11 * src/index.ts writes the messages.
12 */
13// The modules themselves, not the package's index, so the tests load them.
14import { G1T_KEY, type ChatAnswer, type MemberProfile } from "@g1t/contracts/chat";
15import { ASK_MAX_MESSAGES, ASK_MAX_MESSAGE_CHARS, ASK_MAX_TOTAL_CHARS, type AskMessage, type AskPage } from "@g1t/contracts/integrations";
16
17/** How @g1t shows in chat: the g1t mark, never a person. */
18export const G1T_PROFILE: MemberProfile = {
19 kind: "system",
20 id: "g1t",
21 name: "g1t",
22 display_name: "g1t",
23 avatar: null,
24 role: "Answers with your workspace's own AI provider",
25 title: null,
26 avatar_seed: null,
27};
28
29/** The most steps (tool calls and the like) one answer keeps. */
30export const MAX_STEPS = 50;
31/** The longest one step may be, as JSON; longer ones keep their type and a note. */
32const MAX_STEP_BYTES = 8_192;
33/** The most answers one direct message with g1t starts in a minute: one person asking. */
34export const ANSWERS_PER_MINUTE = 20;
35
36/** What the `meta` column keeps (migration 0005): on an answer, a question from a page, or a new-conversation mark. */
37export type MessageMeta = { answer?: ChatAnswer; page?: AskPage; reset?: true };
38
39export function parseMeta(text: string | null | undefined): MessageMeta {
40 if (!text) return {};
41 try {
42 const meta = JSON.parse(text) as unknown;
43 return meta && typeof meta === "object" && !Array.isArray(meta) ? (meta as MessageMeta) : {};
44 } catch {
45 return {};
46 }
47}
48
49export function metaColumn(meta: MessageMeta): string | null {
50 return meta.answer || meta.page || meta.reset ? JSON.stringify(meta) : null;
51}
52
53/**
54 * The person in a direct message with g1t: one person and g1t, no one
55 * else. Null for any other conversation.
56 */
57export function g1tDmPerson(members: readonly string[]): string | null {
58 if (members.length !== 2 || !members.includes(G1T_KEY)) return null;
59 const person = members.find((member) => member !== G1T_KEY)!;
60 return person.startsWith("user:") ? person : null;
61}
62
63/**
64 * Whether a message just written starts an answer: a person's own words,
65 * at the top of a direct message with g1t. g1t's own messages, cards and
66 * thread replies never do; nor does anything anywhere else (mentioning
67 * @g1t in a channel does not, yet).
68 */
69export function startsAnswer(input: {
70 channelKind: "channel" | "dm";
71 members: readonly string[];
72 author: string;
73 threadRoot: string | null;
74 card: boolean;
75 body: string;
76}): boolean {
77 if (input.channelKind !== "dm" || input.threadRoot || input.card || !input.body.trim()) return false;
78 const person = g1tDmPerson(input.members);
79 return person != null && input.author === person;
80}
81
82/** A message as the conversation reads it: oldest first. */
83export type SaidRow = { id: string; author: string; kind: "text" | "card"; body: string; meta: string | null; deleted_at: string | null };
84
85/**
86 * What an answer reads: the direct message's messages after the latest
87 * "New conversation" mark, the person's as `user` and g1t's answers as
88 * `assistant`, oldest first, alternating, starting and ending with the
89 * person, within Ask's limits (the newest kept). Answers that failed, the
90 * set-up note and the answer being written are left out; several messages
91 * in a row from one side are read as one. Empty when there is nothing to
92 * answer.
93 */
94export function conversation(rows: readonly SaidRow[], answering: string | null = null): AskMessage[] {
95 let start = 0;
96 rows.forEach((row, index) => {
97 if (row.author === G1T_KEY && parseMeta(row.meta).reset) start = index + 1;
98 });
99 const turns: AskMessage[] = [];
100 for (const row of rows.slice(start)) {
101 if (row.deleted_at || row.kind !== "text" || row.id === answering) continue;
102 const text = row.body.trim();
103 if (!text) continue;
104 let role: AskMessage["role"];
105 if (row.author === G1T_KEY) {
106 const answer = parseMeta(row.meta).answer;
107 if (!answer || answer.setup || answer.state === "error" || answer.state === "thinking" || answer.state === "streaming") continue;
108 role = "assistant";
109 } else if (row.author.startsWith("user:")) {
110 role = "user";
111 } else continue;
112 const last = turns.at(-1);
113 if (last?.role === role) last.content = `${last.content}\n\n${text}`;
114 else turns.push({ role, content: text });
115 }
116 for (const turn of turns) {
117 if (turn.content.length > ASK_MAX_MESSAGE_CHARS) {
118 // The person's newest words matter most: keep the end of a long question, the start of a long answer.
119 turn.content = turn.role === "user" ? turn.content.slice(-ASK_MAX_MESSAGE_CHARS) : turn.content.slice(0, ASK_MAX_MESSAGE_CHARS);
120 }
121 }
122 // The latest word is g1t's: already answered.
123 if (turns.at(-1)?.role !== "user") return [];
124 let kept = turns.slice(-ASK_MAX_MESSAGES);
125 let total = kept.reduce((sum, turn) => sum + turn.content.length, 0);
126 while (kept.length && (kept[0]!.role !== "user" || total > ASK_MAX_TOTAL_CHARS)) {
127 total -= kept[0]!.content.length;
128 kept = kept.slice(1);
129 }
130 return kept;
131}
132
133/** The page the newest question was asked from, if it was asked from one. */
134export function latestPage(rows: readonly SaidRow[]): AskPage | null {
135 for (let at = rows.length - 1; at >= 0; at--) {
136 const row = rows[at]!;
137 if (row.author.startsWith("user:") && !row.deleted_at) return parseMeta(row.meta).page ?? null;
138 }
139 return null;
140}
141
142/**
143 * A page as a person's message may carry it: only what Ask reads, of the
144 * shapes it reads. Null when nothing of it is usable.
145 */
146export function pageFrom(value: unknown): AskPage | null {
147 if (!value || typeof value !== "object") return null;
148 const sent = value as Record<string, unknown>;
149 const number = (n: unknown) => (typeof n === "number" && Number.isInteger(n) && n > 0 && n < 2 ** 31 ? n : null);
150 const repo = sent.repo as { namespace?: unknown; name?: unknown } | null | undefined;
151 const page: AskPage = {
152 path: typeof sent.path === "string" && sent.path.startsWith("/") ? sent.path.slice(0, 500) : null,
153 repo:
154 repo && typeof repo.namespace === "string" && typeof repo.name === "string" && repo.namespace && repo.name
155 ? { namespace: repo.namespace.slice(0, 100), name: repo.name.slice(0, 100) }
156 : null,
157 issue: number(sent.issue),
158 pull: number(sent.pull),
159 };
160 return page.path || page.repo ? page : null;
161}
162
163/** What g1t says when nobody chose an AI provider for Ask yet: once, and where to choose one. */
164export function notSetUpText(slug: string): string {
165 return [
166 "I answer with an AI provider your workspace connects, on its own key, and none is chosen for me yet.",
167 "",
168 `An owner can choose one under [Integrations → Your AI providers → Ask g1t](/${slug}/-/integrations/models#ask). If you're not an owner, ask one of them to.`,
169 ].join("\n");
170}
171
172/** What g1t says when Ask is set up but cannot answer now: the reason, short. */
173export function cannotAnswerText(problem: string): string {
174 return `I couldn't answer that: ${problem.trim() || "something went wrong."}`;
175}
176
177/**
178 * An answer as it streams in: the integrations service's newline-delimited
179 * events, read as they come. Events it does not know (a newer service's
180 * tool steps, say) are kept as steps, never shown as text, and never
181 * break the answer.
182 */
183export class AnswerReader {
184 body = "";
185 answer: ChatAnswer = { state: "thinking" };
186 private rest = "";
187 private ended = false;
188
189 /** Reads a chunk; true when something shown changed. */
190 feed(chunk: string): boolean {
191 const lines = (this.rest + chunk).split("\n");
192 this.rest = lines.pop() ?? "";
193 let changed = false;
194 for (const line of lines) changed = this.line(line) || changed;
195 return changed;
196 }
197
198 /** The stream ended: whatever is left of a line, and an answer that never said it was done. */
199 finish(): void {
200 if (this.rest) this.line(this.rest);
201 this.rest = "";
202 if (!this.ended) {
203 this.ended = true;
204 if (this.body.trim()) this.answer = { ...this.answer, state: "error", error: "The answer broke off." };
205 else this.answer = { ...this.answer, state: "error", error: this.answer.error ?? "No answer came." };
206 }
207 }
208
209 /** Stopped (Stop, or a newer question): what came stays. */
210 stop(): void {
211 if (this.ended) return;
212 this.ended = true;
213 this.answer = { ...this.answer, state: "stopped" };
214 }
215
216 get over(): boolean {
217 return this.ended;
218 }
219
220 private line(line: string): boolean {
221 if (!line.trim() || this.ended) return false;
222 let event: Record<string, unknown>;
223 try {
224 event = JSON.parse(line) as Record<string, unknown>;
225 } catch {
226 return false;
227 }
228 if (!event || typeof event !== "object" || typeof event.type !== "string") return false;
229 switch (event.type) {
230 case "meta":
231 this.answer = {
232 ...this.answer,
233 provider_label: typeof event.provider_label === "string" ? event.provider_label : null,
234 model: typeof event.model === "string" ? event.model : null,
235 };
236 return true;
237 case "text":
238 if (typeof event.text !== "string" || !event.text) return false;
239 this.body += event.text;
240 this.answer = { ...this.answer, state: "streaming" };
241 return true;
242 case "done":
243 this.ended = true;
244 this.answer = { ...this.answer, state: "done", stop_reason: typeof event.stop_reason === "string" ? event.stop_reason : null };
245 return true;
246 case "error":
247 this.ended = true;
248 this.answer = { ...this.answer, state: "error", error: typeof event.message === "string" ? event.message : "Something went wrong." };
249 return true;
250 default: {
251 const steps = this.answer.steps ?? [];
252 const json = JSON.stringify(event);
253 const step = json.length > MAX_STEP_BYTES ? { type: event.type, id: event.id, truncated: true } : event;
254 // A tool step comes again with its id when it ends, and sources come
255 // whole: each replaces the line it updates, so a step is kept once.
256 const at = steps.findIndex((kept) =>
257 event.type === "sources" ? kept.type === "sources" : event.type === "tool" && typeof event.id === "string" && kept.type === "tool" && kept.id === event.id,
258 );
259 if (at !== -1) {
260 this.answer = { ...this.answer, steps: steps.map((kept, index) => (index === at ? step : kept)) };
261 return true;
262 }
263 if (steps.length >= MAX_STEPS) return false;
264 this.answer = { ...this.answer, steps: [...steps, step] };
265 return true;
266 }
267 }
268 }
269}
270
271/**
272 * The text an answer is stored with at its end: what came, or, when nothing
273 * did, a short line saying why, so g1t never answers with silence.
274 */
275export function finalBody(body: string, answer: ChatAnswer): string {
276 if (body.trim()) return body;
277 if (answer.state === "error") return cannotAnswerText(answer.error ?? "");
278 if (answer.state === "stopped") return "_Stopped before answering._";
279 return "_No answer came._";
280}
281
282/** Whether `times` (ms, oldest first) leave room for another answer now; and the times to keep. */
283export function answerAllowed(times: readonly number[], now: number): { ok: boolean; times: number[] } {
284 const recent = times.filter((at) => now - at < 60_000);
285 if (recent.length >= ANSWERS_PER_MINUTE) return { ok: false, times: recent };
286 return { ok: true, times: [...recent, now] };
287}
288
289/**
290 * An answer older than this that still says it is being written was cut
291 * off (the room that wrote it went away): it shows as stopped.
292 */
293export const STALE_ANSWER_MS = 16 * 60_000;
294
295export function shownAnswer(answer: ChatAnswer | undefined, createdAt: string, now: number): ChatAnswer | null {
296 if (!answer) return null;
297 if ((answer.state === "thinking" || answer.state === "streaming") && now - Date.parse(createdAt) > STALE_ANSWER_MS) {
298 return { ...answer, state: "stopped" };
299 }
300 return answer;
301}