Skip to content
Sign up

flagon-io/g1t

public
0
1,170 linesCodeBlameRaw
1/**
2 * Sessions (docs.g1t.sh/guides/agent-sessions/): the work an agent spins off
3 * from a conversation, a routine's run, or its part in another session.
4 *
5 * A conversation with an agent is never one long context. Replies read the
6 * last few messages, what the agent remembers, and its recent sessions in
7 * that conversation. Real work happens in a session:
8 *
9 * - **Bounded.** A session has a goal, its own working context of turns
10 * (compacted as it grows: the goal, a summary of earlier steps, and the
11 * latest turns), a step limit and a spend cap.
12 * - **Visible.** It posts a live card where it was asked, updates the card
13 * in place as it works, posts progress in the card's thread, and reports
14 * back in the conversation when done. Its page has the full transcript.
15 * - **Steerable.** A reply in the card's thread, or from its page, reaches
16 * it at its next step, or wakes it again once it is done.
17 * - **A tree.** It can hand parts to its subagents or bring colleagues in;
18 * each is a child session whose result comes back to it. Everything in
19 * the tree is paid by the agent at the root, within the root's cap.
20 * - **Stoppable.** Anyone who can see it can stop it and everything under it.
21 *
22 * Each step runs on the agent's desk (desk.ts) through `metered` (meter.ts),
23 * so a step is billed, capped and recorded exactly as a reply is.
24 */
25import {
26 type AgentLook,
27 type AgentRef,
28 type AgentSession,
29 type AgentSessionKind,
30 type AgentSessionStatus,
31 type AskerAccess,
32 type MessageCard,
33 type ModelTier,
34 type ServiceBinding,
35 type SessionEvent,
36 type SessionOutput,
37 type SubagentDef,
38 type User,
39 chatClient,
40 identityClient,
41 agentRef,
42 newId,
43 workClient,
44} from "@g1t/contracts";
45
46import { readLook } from "../../../packages/contracts/src/agent-look.ts";
47import { CHAT_MAX_HOPS } from "../../../packages/contracts/src/chat.ts";
48import { Audience } from "./audience.ts";
49import { type MeterEnv, metered } from "./meter.ts";
50import { type RecallPlace, MAX_FACTS, cleanFact, memorySection, recall, scopeFor } from "./memory.ts";
51import { readPolicy } from "./policy.ts";
52import { type PortsEnv, audiencePorts, loadTeams, teamsOfAgents, toolPorts } from "./ports.ts";
53import { systemPrompt } from "./prompt.ts";
54import { loadShelf, skillsSection, teamSlugs } from "./skills.ts";
55import { readVersion } from "./skill-library.ts";
56import { conversationFrom } from "./surface.ts";
57import { type Row, definitionOf, periods } from "./store.ts";
58import { abilitiesWithConnectors, abilitiesSection, abilityPorts, pendingRequests, saidText } from "./abilities.ts";
59import { computerPorts, hasComputer } from "./computer.ts";
60import { commandEvent, eventStatement } from "./transcript.ts";
61import { type ActionPorts, type ToolCall, ToolBox } from "./tools.ts";
62import { type ModelMessage, SESSION_LIMITS, runTurn } from "./turn.ts";
63import { effortOf, effortPlan, higherEffort, isLevel } from "./routing.ts";
64import { recallQuery, recallSection } from "./recall.ts";
65import { rosterLines } from "./orchestrator.ts";
66import { teamsSection } from "./teammates.ts";
67import { dollars } from "./money.ts";
68import { postDraft } from "./cards.ts";
69import { sessionActions } from "./card-views.ts";
70import type { Desk } from "./desk.ts";
71
72export type SessionEnv = MeterEnv &
73 PortsEnv & {
74 CHAT: ServiceBinding;
75 IDENTITY: ServiceBinding;
76 WORK: ServiceBinding;
77 NOTIFY?: ServiceBinding;
78 /** The workspace's connections, for abilities outside g1t (abilities.ts). */
79 INTEGRATIONS: ServiceBinding;
80 /** The audit log, for refusals. */
81 EVENTS: ServiceBinding;
82 /** The runner, for agents' own computers (computer.ts). Absent: sessions have none. */
83 RUNNER?: ServiceBinding;
84 DESKS: DurableObjectNamespace<Desk>;
85 };
86
87/**
88 * Steps one session takes at most before it must report, at medium effort;
89 * its agent's effort setting moves it (routing.ts `SESSION_STEPS`).
90 */
91export const MAX_STEPS = 8;
92/** Children one session may have running at once. */
93export const MAX_CHILDREN = 4;
94/** How deep a tree of sessions may go. */
95export const MAX_DEPTH = 3;
96/** Past this many characters of working context, it is compacted. */
97const CONTEXT_LIMIT = 120_000;
98/** Turns kept whole when compacting. */
99const KEEP_TURNS = 6;
100/** The longest report posted in chat. */
101const MAX_REPORT = 12_000;
102
103export const LIVE: AgentSessionStatus[] = ["queued", "working", "waiting", "needs_approval"];
104const OVER: AgentSessionStatus[] = ["done", "failed", "stopped"];
105
106export type SessionRow = {
107 id: string;
108 workspace_id: string;
109 agent_id: string;
110 subagent: string | null;
111 kind: string;
112 parent_id: string | null;
113 root_id: string;
114 payer_agent_id: string;
115 title: string;
116 goal: string;
117 status: string;
118 status_note: string | null;
119 summary: string | null;
120 workspace: string;
121 channel_id: string;
122 channel_kind: string;
123 channel_name: string | null;
124 thread_root: string | null;
125 message_id: string | null;
126 card_message_id: string | null;
127 asked_by: string | null;
128 asked_by_username: string | null;
129 asker: string | null;
130 routine_id: string | null;
131 chain: string;
132 hops: number;
133 context: string;
134 inbox: string;
135 steps: number;
136 tool_calls: number;
137 input_tokens: number;
138 output_tokens: number;
139 cost_micros: number;
140 charged_micros: number;
141 cap_micros: number | null;
142 model: string | null;
143 /** The effort level it ran at (the highest, when Auto raised it). Null on sessions from before. */
144 effort?: string | null;
145 tier?: string | null;
146 outputs: string;
147 step_started_at: string | null;
148 created_at: string;
149 updated_at: string;
150 finished_at: string | null;
151};
152
153/** One turn of a session's working context: plain text, never tool blocks. */
154type Turn = { role: "user" | "assistant"; content: string };
155/** Something that arrived for a session while it worked. */
156type Inbound = { kind: "steer" | "child"; by: string; body: string };
157
158function json<T>(raw: string | null | undefined, fallback: T): T {
159 if (!raw) return fallback;
160 try {
161 return JSON.parse(raw) as T;
162 } catch {
163 return fallback;
164 }
165}
166
167const iso = () => new Date().toISOString();
168
169/** A session as the contract shows it; `visible` false hides what it was about. */
170export function toSession(row: SessionRow, agent: { handle: string; display_name: string; avatar_seed: string; look?: AgentLook | null } | null, visible: boolean): AgentSession {
171 return {
172 id: row.id,
173 workspace_id: row.workspace_id,
174 agent_id: row.agent_id,
175 agent_handle: agent?.handle ?? "agent",
176 agent_name: agent?.display_name ?? "An agent",
177 agent_avatar_seed: agent?.avatar_seed ?? agent?.handle ?? row.agent_id,
178 agent_look: agent?.look ?? null,
179 subagent: row.subagent,
180 kind: row.kind as AgentSessionKind,
181 parent_id: row.parent_id,
182 root_id: row.root_id,
183 payer_agent_id: row.payer_agent_id,
184 title: visible ? row.title : "A private session",
185 goal: visible ? row.goal : "",
186 status: row.status as AgentSessionStatus,
187 status_note: visible ? row.status_note : null,
188 summary: visible ? row.summary : null,
189 channel_id: row.channel_id,
190 channel_kind: row.channel_kind === "dm" ? "dm" : "channel",
191 channel_name: visible ? row.channel_name : null,
192 card_message_id: visible ? row.card_message_id : null,
193 asked_by: row.asked_by,
194 asked_by_username: visible ? row.asked_by_username : null,
195 routine_id: row.routine_id,
196 steps: row.steps,
197 tool_calls: row.tool_calls,
198 input_tokens: row.input_tokens,
199 output_tokens: row.output_tokens,
200 charged_micros: row.charged_micros,
201 cost_micros: row.cost_micros,
202 cap_micros: row.cap_micros,
203 model: row.model,
204 effort: isLevel(row.effort) ? row.effort : null,
205 outputs: visible ? json<SessionOutput[]>(row.outputs, []) : [],
206 created_at: row.created_at,
207 updated_at: row.updated_at,
208 finished_at: row.finished_at,
209 visible,
210 };
211}
212
213/** A session's card, as its conversation shows it. */
214export function cardFor(
215 row: Pick<SessionRow, "id" | "title" | "status" | "steps" | "tool_calls" | "charged_micros" | "status_note"> & Partial<Pick<SessionRow, "cap_micros" | "summary" | "goal">>,
216 slug: string,
217 handle: string,
218 children = 0,
219): MessageCard {
220 const state: Record<string, string> = {
221 queued: "Queued",
222 working: "Working",
223 waiting: children === 1 ? "Waiting on a helper" : "Waiting on helpers",
224 needs_approval: "Needs approval",
225 done: "Done",
226 failed: "Failed",
227 stopped: "Stopped",
228 };
229 const parts = [
230 row.steps ? `Step ${row.steps}` : null,
231 row.tool_calls ? `${row.tool_calls} tool${row.tool_calls === 1 ? "" : "s"}` : null,
232 row.charged_micros ? dollars(row.charged_micros) : null,
233 ].filter(Boolean);
234 const note = row.status === "needs_approval" || row.status === "failed" || row.status === "stopped" ? row.status_note : null;
235 const href = `/${slug}/-/agents/${handle}/sessions/${row.id}`;
236 return {
237 kind: "session",
238 title: row.title,
239 detail: [parts.join(" · ") || "Starting", note].filter(Boolean).join(" — ").slice(0, 480),
240 state: state[row.status] ?? row.status,
241 href,
242 ...(row.status === "done" && row.summary ? { body: row.summary.length > 600 ? `${row.summary.slice(0, 600)}…` : row.summary } : {}),
243 fields: [
244 ...(row.cap_micros ? [{ label: "Spent", value: `${dollars(row.charged_micros)} of ${dollars(row.cap_micros)}` }] : []),
245 ...(children ? [{ label: "Helpers", value: `${children} working` }] : []),
246 ],
247 actions: sessionActions(row.status, row.cap_micros ?? null, row.charged_micros, href),
248 owner: "agents",
249 ref: row.id,
250 };
251}
252
253// Appending to a session's transcript lives in transcript.ts, so that the
254// cards and the computer can write entries without needing sessions.ts.
255export { eventStatement } from "./transcript.ts";
256
257/**
258 * The working context, kept bounded: the goal, then a summary of what is
259 * cut, then the latest turns whole. What is cut stays in the transcript.
260 */
261export function compact(turns: Turn[], limit = CONTEXT_LIMIT, keep = KEEP_TURNS): Turn[] {
262 const size = (list: Turn[]) => list.reduce((n, t) => n + t.content.length, 0);
263 if (size(turns) <= limit || turns.length <= keep + 1) return turns;
264 const [goal, ...rest] = turns;
265 let tail = rest.slice(-keep);
266 // The kept part starts with someone else's turn, as the model needs.
267 while (tail.length && tail[0].role === "assistant") tail = tail.slice(1);
268 const cut = rest.slice(0, rest.length - tail.length);
269 const notes = cut
270 .filter((t) => t.role === "assistant")
271 .map((t, i) => `- Step ${i + 1}: ${t.content.replace(/\s+/g, " ").slice(0, 600)}`)
272 .join("\n");
273 const earlier: Turn = { role: "user", content: `${goal.content}\n\n(Earlier in this session, now summarised:\n${notes || "- nothing to note"})` };
274 return [earlier, ...tail];
275}
276
277/** Turns as the Messages API takes them: alternating, someone else's first. */
278function alternate(turns: Turn[]): ModelMessage[] {
279 const out: Turn[] = [];
280 for (const turn of turns) {
281 const last = out[out.length - 1];
282 if (last && last.role === turn.role) last.content += `\n\n${turn.content}`;
283 else out.push({ ...turn });
284 }
285 while (out.length && out[0].role === "assistant") out.shift();
286 if (out.length && out[out.length - 1].role === "assistant") out.push({ role: "user", content: "(Go on with the session.)" });
287 return out;
288}
289
290async function agentRow(db: D1Database, id: string): Promise<Row | null> {
291 return db.prepare("SELECT * FROM agents WHERE id = ?").bind(id).first<Row>();
292}
293
294export async function sessionRow(db: D1Database, id: string): Promise<SessionRow | null> {
295 return db.prepare("SELECT * FROM agent_sessions WHERE id = ?").bind(id).first<SessionRow>();
296}
297
298/** Hands a session to its agent's desk to work its next step. */
299export async function wake(env: Pick<SessionEnv, "DESKS">, agentId: string, sessionId: string): Promise<void> {
300 await env.DESKS.get(env.DESKS.idFromName(agentId)).session(sessionId, agentId);
301}
302
303export type NewSession = {
304 agent: Row;
305 kind: AgentSessionKind;
306 subagent?: SubagentDef | null;
307 parent?: SessionRow | null;
308 title: string;
309 goal: string;
310 workspace: string;
311 channel_id: string;
312 channel_kind: "channel" | "dm";
313 channel_name: string | null;
314 thread_root: string | null;
315 message_id: string | null;
316 asked_by: string | null;
317 asked_by_username: string | null;
318 asker: AskerAccess | null;
319 routine_id?: string | null;
320 chain: string[];
321 hops: number;
322};
323
324/**
325 * Starts a session: its row, its card where it was asked (a child's card
326 * goes in its parent's thread), and its first step on the desk. A child's
327 * cap is what its root has left; a root's is the agent's per-task cap or
328 * the workspace's default for sessions, whichever is lower.
329 */
330export async function startSession(env: SessionEnv, input: NewSession): Promise<SessionRow> {
331 const db = env.DB;
332 const now = iso();
333 const id = newId("asn");
334 const parent = input.parent ?? null;
335 const root = parent ? ((await sessionRow(db, parent.root_id)) ?? parent) : null;
336 let cap: number | null;
337 if (root) {
338 const tree = await db.prepare("SELECT COALESCE(SUM(charged_micros), 0) AS spent FROM agent_sessions WHERE root_id = ?").bind(root.id).first<{ spent: number }>();
339 cap = root.cap_micros != null ? Math.max(1, root.cap_micros - (tree?.spent ?? 0)) : null;
340 } else {
341 const policy = await readPolicy(db, input.agent.workspace_id, periods(new Date())[0]);
342 const task = definitionOf(input.agent).budget.task_micros;
343 cap = Math.min(policy.default_session_micros, task && task > 0 ? task : Number.POSITIVE_INFINITY);
344 }
345 const subagent = input.subagent ?? null;
346 const goal = [
347 input.goal,
348 subagent ? `\n(You are working as ${input.agent.display_name}'s subagent "${subagent.name}": ${subagent.description}\n\n${subagent.instructions})` : "",
349 ].join("");
350 const context: Turn[] = [{ role: "user", content: `Your session: ${input.title}\n\n${goal}` }];
351 await db.batch([
352 db
353 .prepare(
354 `INSERT INTO agent_sessions (id, workspace_id, agent_id, subagent, kind, parent_id, root_id, payer_agent_id, title, goal, status,
355 workspace, channel_id, channel_kind, channel_name, thread_root, message_id, asked_by, asked_by_username, asker, routine_id,
356 chain, hops, context, cap_micros, created_at, updated_at)
357 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
358 )
359 .bind(
360 id,
361 input.agent.workspace_id,
362 input.agent.id,
363 subagent?.name ?? null,
364 input.kind,
365 parent?.id ?? null,
366 root?.id ?? id,
367 root?.payer_agent_id ?? input.agent.id,
368 input.title.slice(0, 120),
369 input.goal.slice(0, 8000),
370 input.workspace.toLowerCase(),
371 input.channel_id,
372 input.channel_kind,
373 input.channel_name,
374 input.thread_root,
375 input.message_id,
376 input.asked_by,
377 input.asked_by_username,
378 input.asker ? JSON.stringify(input.asker) : null,
379 input.routine_id ?? null,
380 JSON.stringify(input.chain),
381 input.hops,
382 JSON.stringify(context),
383 cap === Number.POSITIVE_INFINITY ? null : cap,
384 now,
385 now,
386 ),
387 eventStatement(db, id, "goal", input.asked_by_username, `${input.title}\n\n${input.goal}`),
388 ]);
389 let row = (await sessionRow(db, id))!;
390 if (parent) {
391 await addOutput(db, parent.id, { kind: "session", id, agent_handle: subagent ? `${input.agent.handle}/${subagent.name}` : input.agent.handle, title: row.title });
392 await db.batch([eventStatement(db, parent.id, "child", input.agent.handle, `${subagent ? `Subagent ${subagent.name}` : `@${input.agent.handle}`} started: ${row.title}`)]);
393 } else {
394 // A root session's card, where it was asked.
395 const posted = await chatClient(env.CHAT)
396 .postAsAgent(row.workspace, row.channel_id, row.agent_id, {
397 body: "",
398 card: cardFor(row, row.workspace, input.agent.handle),
399 thread_root: row.thread_root,
400 hops: row.hops,
401 asked_by: row.asked_by,
402 asker: input.asker,
403 chain: input.chain,
404 })
405 .catch(() => null);
406 if (posted?.ok) {
407 await db.prepare("UPDATE agent_sessions SET card_message_id = ? WHERE id = ?").bind(posted.value.id, id).run();
408 row = { ...row, card_message_id: posted.value.id };
409 }
410 }
411 await wake(env, row.agent_id, id);
412 return row;
413}
414
415async function addOutput(db: D1Database, id: string, output: SessionOutput): Promise<void> {
416 const row = await db.prepare("SELECT outputs FROM agent_sessions WHERE id = ?").bind(id).first<{ outputs: string }>();
417 const list = json<SessionOutput[]>(row?.outputs, []);
418 list.push(output);
419 await db.prepare("UPDATE agent_sessions SET outputs = ? WHERE id = ?").bind(JSON.stringify(list.slice(-50)), id).run();
420}
421
422/** The root session of a tree, whose card and agent speak for it in chat. */
423async function speaker(db: D1Database, row: SessionRow): Promise<{ root: SessionRow; agent: Row | null }> {
424 const root = row.root_id === row.id ? row : ((await sessionRow(db, row.root_id)) ?? row);
425 return { root, agent: await agentRow(db, root.agent_id) };
426}
427
428/** Brings the root's card up to date with the tree. Never throws. */
429export async function refreshCard(env: SessionEnv, row: SessionRow): Promise<void> {
430 try {
431 const db = env.DB;
432 const { root, agent } = await speaker(db, row);
433 if (!root.card_message_id || !agent) return;
434 const fresh = (await sessionRow(db, root.id)) ?? root;
435 const tree = await db
436 .prepare("SELECT COUNT(*) AS n, COALESCE(SUM(charged_micros), 0) AS spent, COALESCE(SUM(tool_calls), 0) AS tools, SUM(CASE WHEN status IN ('queued','working','waiting') AND id <> root_id THEN 1 ELSE 0 END) AS live FROM agent_sessions WHERE root_id = ?")
437 .bind(root.id)
438 .first<{ n: number; spent: number; tools: number; live: number }>();
439 const shown = { ...fresh, charged_micros: tree?.spent ?? fresh.charged_micros, tool_calls: tree?.tools ?? fresh.tool_calls };
440 await chatClient(env.CHAT).updateAsAgent(fresh.workspace, fresh.channel_id, fresh.agent_id, root.card_message_id, { card: cardFor(shown, fresh.workspace, agent.handle, tree?.live ?? 0) });
441 } catch (error) {
442 console.error("agents: a session card was not updated", row.id, String(error));
443 }
444}
445
446/** Posts in the root card's thread, as the root's agent; a child's note names who it is from. */
447async function postInThread(env: SessionEnv, row: SessionRow, by: Row, text: string): Promise<boolean> {
448 const db = env.DB;
449 const { root } = await speaker(db, row);
450 if (!root.card_message_id) return false;
451 const prefix = root.id === row.id ? "" : `**${row.subagent ? `${by.display_name} · ${row.subagent}` : by.display_name}:** `;
452 const posted = await chatClient(env.CHAT)
453 .postAsAgent(root.workspace, root.channel_id, root.agent_id, {
454 body: `${prefix}${text}`.slice(0, 8000),
455 thread_root: root.card_message_id,
456 hops: root.hops,
457 asked_by: root.asked_by,
458 asker: json<AskerAccess | null>(root.asker, null),
459 chain: json<string[]>(root.chain, []),
460 })
461 .catch(() => null);
462 return !!posted?.ok;
463}
464
465/** Posts a card (a draft issue) in the root card's thread, as the root's agent; its id. */
466async function postCardInThread(env: SessionEnv, row: SessionRow, by: Row, card: MessageCard): Promise<string | null> {
467 const { root } = await speaker(env.DB, row);
468 const posted = await chatClient(env.CHAT)
469 .postAsAgent(root.workspace, root.channel_id, root.agent_id, {
470 body: root.id === row.id ? "" : `**${by.display_name}** drafted this:`,
471 card,
472 thread_root: root.card_message_id ?? root.thread_root,
473 hops: root.hops,
474 asked_by: root.asked_by,
475 asker: json<AskerAccess | null>(root.asker, null),
476 chain: json<string[]>(root.chain, []),
477 })
478 .catch(() => null);
479 return posted?.ok ? posted.value.id : null;
480}
481
482/** Sets a session's status, records why, and brings its card along. */
483async function setStatus(env: SessionEnv, row: SessionRow, status: AgentSessionStatus, note: string | null, extra: Record<string, string | number | null> = {}): Promise<SessionRow> {
484 const db = env.DB;
485 const names = Object.keys(extra);
486 const finished = OVER.includes(status) ? iso() : null;
487 await db
488 .prepare(
489 `UPDATE agent_sessions SET status = ?, status_note = ?, updated_at = ?, finished_at = COALESCE(?, finished_at)${names.map((n) => `, ${n} = ?`).join("")} WHERE id = ?`,
490 )
491 .bind(status, note, iso(), finished, ...names.map((n) => extra[n]), row.id)
492 .run();
493 const fresh = (await sessionRow(db, row.id))!;
494 await refreshCard(env, fresh);
495 return fresh;
496}
497
498/** The person who asked, resolved, for acting on their behalf. */
499async function askerUser(env: SessionEnv, row: SessionRow): Promise<User | null> {
500 if (!row.asked_by) return null;
501 const [user] = await identityClient(env.IDENTITY)
502 .usersForAudience([row.asked_by])
503 .catch(() => [] as User[]);
504 return user ?? null;
505}
506
507/**
508 * What an agent may do here: remember and forget within where it is,
509 * file issues as the person who asked, and (in a session) post updates,
510 * use subagents and bring colleagues in. Shared by replies and sessions.
511 */
512export function actionPorts(
513 env: SessionEnv,
514 input: {
515 agent: Row;
516 place: RecallPlace;
517 source: { kind: "message" | "session"; ref: string; label: string; channel_id: string };
518 asker: { id: string | null; username: string | null };
519 workspace: string;
520 session?: SessionRow | null;
521 /** From a reply: starts a session for the conversation. */
522 spinOff?: (title: string, goal: string) => Promise<{ ok: boolean; message: string }>;
523 /** Posts a card where this work reports (a draft issue); its message id, or null. */
524 postCard: (card: MessageCard) => Promise<string | null>;
525 },
526): ActionPorts {
527 const db = env.DB;
528 const { agent, place } = input;
529 const session = input.session ?? null;
530 const ports: ActionPorts = {
531 async remember(body, wanted, onlyForAsker) {
532 const fact = cleanFact(body);
533 if (!fact) return { ok: false, message: "Say what to remember." };
534 const count = await db.prepare("SELECT COUNT(*) AS n FROM agent_memories WHERE agent_id = ?").bind(agent.id).first<{ n: number }>();
535 if ((count?.n ?? 0) >= MAX_FACTS) return { ok: false, message: "Your memory is full. Forget something out of date first." };
536 const privately = !!onlyForAsker && !!input.asker.id;
537 const { scope, ref } = scopeFor(place, wanted, privately ? input.asker.id : null);
538 const id = newId("mem");
539 const now = iso();
540 const label = scope === "person" ? input.asker.username : scope === "channel" ? input.source.label : null;
541 await db
542 .prepare(
543 `INSERT INTO agent_memories (id, agent_id, workspace_id, scope, scope_ref, scope_label, body, source_kind, source_ref, source_label, source_channel_id, created_by, created_by_kind, created_at, updated_at)
544 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'agent', ?, ?)`,
545 )
546 .bind(id, agent.id, agent.workspace_id, scope, ref, label, fact, input.source.kind, input.source.ref, input.source.label, input.source.channel_id, agent.handle, now, now)
547 .run();
548 if (session) await addOutput(db, session.id, { kind: "memory", id, body: fact });
549 const where = scope === "workspace" ? "for the whole workspace" : scope === "person" ? "for this person" : "for this conversation";
550 const narrowed = privately
551 ? " (only for them: you read an artifact not everyone in the workspace can)"
552 : wanted && wanted !== scope
553 ? ` (${wanted} wasn't allowed from here)`
554 : "";
555 return { ok: true, message: `Remembered ${where}${narrowed}: ${fact}` };
556 },
557 async forget(id) {
558 const row = await db.prepare("SELECT scope, scope_ref FROM agent_memories WHERE id = ? AND agent_id = ?").bind(id, agent.id).first<{ scope: string; scope_ref: string }>();
559 // Only what could be recalled here can be forgotten from here.
560 const here = row && (row.scope === "workspace" ? place.kind === "public" : row.scope === "channel" ? row.scope_ref === place.channel_id : place.kind === "dm" && place.people.length === 1 && place.people[0] === row.scope_ref);
561 if (!row || !here) return { ok: false, message: "There is no such note you can forget here." };
562 await db.prepare("DELETE FROM agent_memories WHERE id = ?").bind(id).run();
563 return { ok: true, message: "Forgotten." };
564 },
565 async comment(repo, asker, number, body) {
566 const made = await workClient(env.WORK).workspaceAgentComment({ namespace: repo.namespace, name: repo.name }, number, refOf(agent), asker, body);
567 if (!made.ok) return { ok: false, message: `It couldn't be posted: ${made.error.message}` };
568 return { ok: true, message: `Commented on ${repo.namespace}/${repo.name}#${number}.` };
569 },
570 async review(repo, asker, number, verdict, body) {
571 const made = await workClient(env.WORK).workspaceAgentReview({ namespace: repo.namespace, name: repo.name }, number, refOf(agent), asker, verdict, body);
572 if (!made.ok) return { ok: false, message: `The review couldn't be posted: ${made.error.message}` };
573 const what = verdict === "approve" ? "Approved" : verdict === "request_changes" ? "Requested changes on" : "Reviewed";
574 return { ok: true, message: `${what} ${repo.namespace}/${repo.name}#${number} (advisory). Link it in your report: /${repo.namespace}/${repo.name}/pull/${number}` };
575 },
576 async draftIssue(repo, issue) {
577 const draft = await postDraft(
578 env,
579 {
580 agent_id: agent.id,
581 workspace_id: agent.workspace_id,
582 workspace: input.workspace,
583 channel_id: input.source.channel_id,
584 session_id: session?.id ?? null,
585 repo_id: repo.id,
586 repo: `${repo.namespace}/${repo.name}`,
587 title: issue.title,
588 body: issue.body,
589 labels: issue.labels,
590 asked_by: input.asker.id,
591 },
592 input.postCard,
593 );
594 if (!draft) return { ok: false, message: "The draft couldn't be posted; give it in your answer instead." };
595 return { ok: true, message: "The draft is in the conversation as a card with File issue and Discard. Tell them in a sentence; don't repeat it." };
596 },
597 };
598 if (input.spinOff) ports.startSession = input.spinOff;
599 if (session) {
600 ports.postUpdate = async (text) => {
601 const posted = await postInThread(env, session, agent, text);
602 if (posted) await db.batch([eventStatement(db, session.id, "update", agent.handle, text)]);
603 return posted ? { ok: true, message: "Posted." } : { ok: false, message: "It couldn't be posted; carry on." };
604 };
605 const child = async (target: Row, subagent: SubagentDef | null, brief: string) => {
606 const depth = await treeDepth(db, session);
607 if (depth >= MAX_DEPTH) return { ok: false, message: "This work is already deep enough; do this part yourself." };
608 const live = await db
609 .prepare("SELECT COUNT(*) AS n FROM agent_sessions WHERE parent_id = ? AND status IN ('queued','working','waiting','needs_approval')")
610 .bind(session.id)
611 .first<{ n: number }>();
612 if ((live?.n ?? 0) >= MAX_CHILDREN) return { ok: false, message: `You already have ${MAX_CHILDREN} helpers working; wait for them.` };
613 const title = brief.split("\n")[0].slice(0, 100) || "Helping";
614 await startSession(env, {
615 agent: target,
616 kind: subagent ? "subagent" : "helper",
617 subagent,
618 parent: session,
619 title,
620 goal: `${agent.display_name} (@${agent.handle}) asked for your help with part of the session "${session.title}".\n\n${brief}\n\nWhen you're done, answer with your result for ${agent.display_name}: findings, links, and anything left open.`,
621 workspace: session.workspace,
622 channel_id: session.channel_id,
623 channel_kind: session.channel_kind === "dm" ? "dm" : "channel",
624 channel_name: session.channel_name,
625 thread_root: session.thread_root,
626 message_id: session.message_id,
627 asked_by: session.asked_by,
628 asked_by_username: session.asked_by_username,
629 asker: json<AskerAccess | null>(session.asker, null),
630 chain: [...json<string[]>(session.chain, []), agent.id],
631 hops: session.hops + 1,
632 });
633 return { ok: true, message: `${subagent ? `Your subagent ${subagent.name}` : `@${target.handle}`} is on it. End this step with what you're waiting for; their result comes back to you before your next step.` };
634 };
635 ports.useSubagent = async (name, brief) => {
636 const subagent = definitionOf(agent).subagents.find((s) => s.name === name);
637 if (!subagent) return { ok: false, message: `You have no subagent called ${name}.` };
638 return child(agent, subagent, brief);
639 };
640 ports.bringIn = async (handle, brief) => {
641 const colleague = await db
642 .prepare("SELECT * FROM agents WHERE workspace_id = ? AND handle = ? AND archived_at IS NULL AND scope = 'workspace'")
643 .bind(agent.workspace_id, handle)
644 .first<Row>();
645 if (!colleague || colleague.id === agent.id) return { ok: false, message: `There is no other agent called @${handle} here.` };
646 if (json<string[]>(session.chain, []).includes(colleague.id)) return { ok: false, message: `@${handle} is already part of this work.` };
647 if (session.hops + 1 > CHAT_MAX_HOPS) return { ok: false, message: "This work has been passed along too many times; do it yourself." };
648 return child(colleague, null, brief);
649 };
650 }
651 return ports;
652}
653
654/** How work names an agent on issues and pull requests. */
655function refOf(agent: Row): AgentRef {
656 return agentRef({ id: agent.id, handle: agent.handle, display_name: agent.display_name, avatar_seed: agent.avatar_seed || agent.handle, look: readLook(agent.look ?? null) });
657}
658
659async function treeDepth(db: D1Database, row: SessionRow): Promise<number> {
660 let depth = 0;
661 let at: SessionRow | null = row;
662 while (at?.parent_id && depth < 10) {
663 depth++;
664 at = await sessionRow(db, at.parent_id);
665 }
666 return depth;
667}
668
669/** The section of the system prompt that says what a session is and how to finish. */
670function sessionSection(row: SessionRow, asker: string, steps = MAX_STEPS): string {
671 const report =
672 row.kind === "helper" || row.kind === "subagent"
673 ? "Your final answer goes back to the agent who asked for your help, not into chat."
674 : row.kind === "routine"
675 ? `Your final answer is posted in ${row.channel_kind === "dm" ? "the direct message" : `#${row.channel_name ?? "the channel"}`} as this routine's report.`
676 : `Your final answer is posted for ${asker} in the conversation where they asked.`;
677 return [
678 "## This session",
679 "",
680 `You are working a session: "${row.title}". It is bounded: work through it with your tools, step by step, and finish within ${steps} steps.`,
681 `- ${report} Make it the report: what you found or did, with links (issues, files, threads), and anything left open.`,
682 "- Use post_update for real milestones or a question for the people following, not for every step.",
683 "- When part of the work belongs to a subagent or a colleague, hand it over with use_subagent or bring_in and end your step saying what you're waiting for; their results come back to you.",
684 "- Never claim to have done or checked something you didn't. If you can't do something from here, say so in the report.",
685 ].join("\n");
686}
687
688/**
689 * Works one step of a session, on its agent's desk. Reads what arrived
690 * (steering, helpers' results), runs one metered model turn with the
691 * session's tools, and decides what comes next: done, waiting on helpers,
692 * another step, or stopped at a limit. Never throws.
693 */
694export async function advance(env: SessionEnv, id: string): Promise<void> {
695 const db = env.DB;
696 let row = await sessionRow(db, id);
697 if (!row || OVER.includes(row.status as AgentSessionStatus) || row.status === "needs_approval") return;
698 const agent = await agentRow(db, row.agent_id);
699 const payer = await agentRow(db, row.payer_agent_id);
700 if (!agent || !payer || agent.archived_at) {
701 await setStatus(env, row, "failed", "Its agent was archived.");
702 return finished(env, row.id);
703 }
704 // Waiting on helpers: only once every child is over.
705 const pending = await db
706 .prepare("SELECT COUNT(*) AS n FROM agent_sessions WHERE parent_id = ? AND status IN ('queued','working','waiting','needs_approval')")
707 .bind(row.id)
708 .first<{ n: number }>();
709 if ((pending?.n ?? 0) > 0) {
710 if (row.status !== "waiting") await setStatus(env, row, "waiting", null);
711 return;
712 }
713 // Its effort setting: the tier a step starts on, how hard the model
714 // reasons, and how many steps it may take. Auto works harder on a
715 // session someone had to steer.
716 const steered = json<Inbound[]>(row.inbox, []).some((item) => item.kind === "steer") || row.effort === "high" || row.effort === "max";
717 const setting = effortOf(definitionOf(agent).routing);
718 const plan = effortPlan(setting, "large", { raised: setting === "auto" && steered });
719 if (row.steps >= plan.steps && json<Inbound[]>(row.inbox, []).length === 0) {
720 await setStatus(env, row, "done", null, { summary: row.summary ?? "Stopped at the step limit." });
721 return finished(env, row.id);
722 }
723 // The session's cap, counting its whole tree for a root.
724 if (row.cap_micros != null && row.charged_micros >= row.cap_micros) {
725 await setStatus(env, row, "needs_approval", `Reached its cap of ${dollars(row.cap_micros)}.`);
726 await notifyApproval(env, row, agent);
727 return;
728 }
729
730 // What arrived meanwhile becomes the next turn.
731 const inbox = json<Inbound[]>(row.inbox, []);
732 let context = json<Turn[]>(row.context, []);
733 if (inbox.length) {
734 const lines = inbox.map((item) => (item.kind === "steer" ? `@${item.by} says: ${item.body}` : `Result from ${item.by}:\n${item.body}`));
735 context.push({ role: "user", content: lines.join("\n\n") });
736 } else if (row.steps > 0) {
737 context.push({ role: "user", content: "(Go on with the session.)" });
738 }
739 context = compact(context);
740 const startedAt = iso();
741 await db
742 .prepare("UPDATE agent_sessions SET status = 'working', status_note = NULL, inbox = '[]', context = ?, step_started_at = ?, updated_at = ? WHERE id = ? AND status <> 'stopped'")
743 .bind(JSON.stringify(context), startedAt, startedAt, row.id)
744 .run();
745 row = (await sessionRow(db, id))!;
746 if (row.status === "stopped") return;
747 await refreshCard(env, row);
748
749 const slug = row.workspace;
750 const definition = definitionOf(agent);
751 const subagent = row.subagent ? definition.subagents.find((s) => s.name === row!.subagent) ?? null : null;
752 const events: D1PreparedStatement[] = [];
753 const calls: ToolCall[] = [];
754 const startTier: ModelTier = plan.start;
755 const asker = { id: row.asked_by, username: row.asked_by_username };
756 const current = row;
757 // Its teams, from their pages: told every step, and their budgets apply.
758 const teamsHere = await loadTeams(env, slug, agent.workspace_id, { id: agent.id }).catch(() => null);
759
760 const outcome = await metered(
761 env,
762 {
763 row: agent,
764 payer,
765 slug,
766 task: "session",
767 start: startTier,
768 effort: plan.level,
769 askerName: row.asked_by_username,
770 person: row.asked_by_username,
771 leftMicros: row.cap_micros != null ? row.cap_micros - row.charged_micros : null,
772 limits: subagent ? subagent.routing : null,
773 teams: teamsHere,
774 },
775 async (model) => {
776 // The audience: who reads what this session posts. Without one it reads nothing but its own context.
777 let toolbox: ToolBox | null = null;
778 let place: RecallPlace = { channel_id: current.channel_id, kind: current.channel_kind === "dm" ? "dm" : "private", people: current.asked_by ? [current.asked_by] : [] };
779 try {
780 if (current.asked_by) {
781 const audience = await Audience.build(slug, current.asked_by, audiencePorts(env, slug, current.channel_id));
782 place = { channel_id: current.channel_id, kind: audience.kind, people: audience.shared ? (current.asked_by ? [current.asked_by] : []) : audience.members.map((m) => m.id) };
783 const noConsult: ToolPorts["consult"] = async () => ({ ok: false, message: "In a session, bring a colleague in with bring_in instead." });
784 const sourceLabel = current.channel_kind === "dm" ? "a direct message" : `#${current.channel_name ?? "a channel"}`;
785 toolbox = new ToolBox(
786 audience,
787 toolPorts(env, slug, agent.workspace_id, current.channel_id, noConsult, agent.id),
788 {
789 agentId: agent.id,
790 notConsult: [agent.handle],
791 hops: current.hops,
792 maxHops: CHAT_MAX_HOPS,
793 session: true,
794 onCall: (call) => {
795 calls.push(call);
796 events.push(eventStatement(db, current.id, "tool", agent.handle, call.args, call.tool, call.outcome));
797 },
798 },
799 [],
800 actionPorts(env, {
801 agent,
802 place,
803 source: { kind: "session", ref: current.id, label: `the session "${current.title}" in ${sourceLabel}`, channel_id: current.channel_id },
804 asker,
805 workspace: slug,
806 session: current,
807 postCard: (card) => postCardInThread(env, current, agent, card),
808 }),
809 );
810 }
811 } catch (error) {
812 console.error("agents: no audience for a session step, so no tools", current.id, String(error));
813 }
814 // Its abilities outside g1t (abilities.ts), for the person it acts for: offered and enforced by the tool box.
815 let abilitiesText: string | null = null;
816 // Its own computer (computer.ts), in sessions only: offered when its shell or files ability isn't Never.
817 let computerCwd: string | null = null;
818 if (toolbox && current.asked_by) {
819 const askerUser = await identityClient(env.IDENTITY)
820 .usersForAudience([current.asked_by])
821 .then((users) => users[0] ?? null)
822 .catch(() => null);
823 const resolved = askerUser ? await abilitiesWithConnectors(env, { agent, definition, workspace: slug, asker: askerUser, teams: teamSlugs(teamsHere) }).catch(() => null) : null;
824 const sections = resolved?.sections ?? null;
825 if (sections && resolved) {
826 toolbox.useAbilities(
827 sections,
828 abilityPorts(env, {
829 agent,
830 workspace: slug,
831 channel_id: current.channel_id,
832 session: { id: current.id, title: current.title },
833 asker,
834 postCard: (card) => postCardInThread(env, current, agent, card),
835 connectors: resolved.connectors,
836 }),
837 saidText([current.goal, ...inbox.filter((item) => item.kind === "steer").map((item) => item.body)]),
838 resolved.servers,
839 );
840 abilitiesText = abilitiesSection(sections);
841 if (hasComputer(sections)) {
842 const computer = computerPorts(env, {
843 agent,
844 workspace: slug,
845 session: { id: current.id },
846 asker,
847 onCommand: (command) => events.push(commandEvent(db, current.id, agent.handle, command)),
848 });
849 if (computer) {
850 toolbox.useComputer(computer);
851 computerCwd = computer.cwd;
852 }
853 }
854 }
855 }
856 // What the workspace's artifacts say about the work: its goal, and whatever arrived for this step.
857 const asked = [current.goal, ...inbox.map((item) => item.body)].reverse();
858 const [facts, passages, shelf] = await Promise.all([
859 recall(db, agent.id, place).catch(() => []),
860 toolbox ? toolbox.recall(recallQuery(asked, 800), definition.reading ?? []) : Promise.resolve([]),
861 // Its skills: named in the prompt, read with use_skill (skills.ts).
862 toolbox ? loadShelf(db, agent.workspace_id, { id: agent.id, skills_off: definition.skills_off }, teamSlugs(teamsHere)) : Promise.resolve([]),
863 ]);
864 toolbox?.useShelf(shelf, (skillId, version) => readVersion(db, skillId, version));
865 const [team, onTeams, here] = await Promise.all([
866 db
867 .prepare("SELECT id, handle, display_name, role, title, responsibilities FROM agents WHERE workspace_id = ? AND archived_at IS NULL AND id <> ? AND scope = 'workspace' ORDER BY builtin DESC, handle LIMIT 50")
868 .bind(agent.workspace_id, agent.id)
869 .all<{ id: string; handle: string; display_name: string; role: string; title: string; responsibilities: string }>(),
870 teamsOfAgents(env, slug),
871 // Who reads what this session posts: said every step, as in a reply. A helper may not be a member: then not said.
872 chatClient(env.CHAT)
873 .conversationForAgent(slug, current.channel_id, agent.id, current.asked_by)
874 .then((found) => (found.ok ? conversationFrom(found.value) : null))
875 .catch(() => null),
876 ]);
877 const roster = rosterLines(
878 team.results.map((a) => ({
879 handle: a.handle,
880 display_name: a.display_name,
881 role: a.role,
882 title: a.title,
883 teams: onTeams.get(a.id) ?? [],
884 responsibilities: json<string[]>(a.responsibilities, []),
885 status: "idle",
886 spent_month_micros: 0,
887 monthly_micros: null,
888 })),
889 );
890 const access = json<AskerAccess | null>(current.asker, null);
891 const system = [
892 systemPrompt({
893 agent: { ...definition, id: agent.id, teams: teamsHere?.teams.map((t) => t.name) ?? [] },
894 workspace: slug,
895 channel: { kind: current.channel_kind === "dm" ? "dm" : "channel", name: current.channel_name },
896 asker: { name: current.asked_by_username ?? "someone", display_name: null, access },
897 today: new Date(),
898 tools: toolbox ? { code: toolbox.definitions().some((tool) => tool.name === "read_file") } : null,
899 colleagues: roster,
900 teams: teamsSection(agent.id, teamsHere, new Date()),
901 session: true,
902 conversation: here,
903 skills: skillsSection(shelf, toolbox?.definitions().map((tool) => tool.name) ?? []),
904 abilities: abilitiesText,
905 computer: computerCwd ? { cwd: computerCwd } : null,
906 }),
907 sessionSection(current, current.asked_by_username ? `@${current.asked_by_username}` : "the person who asked", plan.steps),
908 memorySection(facts),
909 recallSection(passages),
910 ]
911 .filter(Boolean)
912 .join("\n\n");
913 const result = await runTurn(model.send, {
914 model: model.model.model,
915 system,
916 messages: alternate(context),
917 tools: toolbox,
918 price: model.ownModel ? null : model.model.price,
919 effort: model.effort,
920 maxRounds: SESSION_LIMITS.rounds,
921 inputBudget: SESSION_LIMITS.input,
922 maxOutput: SESSION_LIMITS.output,
923 onText: (text) => events.push(eventStatement(db, current.id, "text", agent.handle, text)),
924 stopped: async () => (await db.prepare("SELECT status FROM agent_sessions WHERE id = ?").bind(current.id).first<{ status: string }>())?.status === "stopped",
925 });
926 return { ...result, cost: model.ownModel ? 0 : result.cost };
927 },
928 ).catch((error: unknown) => ({ ok: false as const, reason: "error", message: error instanceof Error ? error.message : String(error) }));
929
930 if (events.length) await db.batch(events).catch((error: unknown) => console.error("agents: transcript not written", id, String(error)));
931 row = (await sessionRow(db, id))!;
932
933 if (!outcome.ok) {
934 if (outcome.reason === "error") {
935 console.error("agents: a session step failed", id, outcome.message);
936 await db.batch([eventStatement(db, id, "note", null, `This step failed: ${outcome.message}`)]);
937 await setStatus(env, row, "failed", "Something went wrong on g1t's side.");
938 } else {
939 await db.batch([eventStatement(db, id, "note", null, outcome.message)]);
940 await setStatus(env, row, "stopped", outcome.message);
941 }
942 return finished(env, id);
943 }
944
945 const answer = outcome.value;
946 const tokens = outcome.tokens;
947 context.push({ role: "assistant", content: answer.text || "(no text)" });
948 await db
949 .prepare(
950 `UPDATE agent_sessions SET steps = steps + 1, tool_calls = tool_calls + ?, input_tokens = input_tokens + ?, output_tokens = output_tokens + ?,
951 cost_micros = cost_micros + ?, charged_micros = charged_micros + ?, model = ?, effort = ?, tier = ?, context = ?, step_started_at = NULL, updated_at = ? WHERE id = ?`,
952 )
953 .bind(
954 calls.length,
955 tokens.input + tokens.cacheRead + tokens.cacheWrite,
956 tokens.output,
957 outcome.cost,
958 outcome.charged,
959 outcome.model,
960 higherEffort(isLevel(row.effort) ? row.effort : null, plan.level),
961 outcome.tier,
962 JSON.stringify(context),
963 iso(),
964 id,
965 )
966 .run();
967 // A root's spend counts its tree for the cap: children add theirs to it too.
968 if (row.root_id !== row.id) {
969 await db.prepare("UPDATE agent_sessions SET charged_micros = charged_micros + ? WHERE id = ?").bind(outcome.charged, row.root_id).run();
970 }
971 row = (await sessionRow(db, id))!;
972 if (row.status === "stopped" || answer.stopped) return finished(env, id);
973
974 // A call waiting on an Ask-first card: the session waits with it, and goes on when the card is answered (cards.ts).
975 const asked = await pendingRequests(db, id).catch(() => []);
976 if (asked.length) {
977 if (answer.text) await db.batch([eventStatement(db, id, "text", agent.handle, answer.text)]);
978 await setStatus(env, row, "needs_approval", `Waiting for an OK: ${asked.map((r) => r.summary).join("; ").slice(0, 200)}.`);
979 return;
980 }
981 const children = await db
982 .prepare("SELECT COUNT(*) AS n FROM agent_sessions WHERE parent_id = ? AND status IN ('queued','working','waiting','needs_approval')")
983 .bind(id)
984 .first<{ n: number }>();
985 if ((children?.n ?? 0) > 0) {
986 if (answer.text) await db.batch([eventStatement(db, id, "text", agent.handle, answer.text)]);
987 await setStatus(env, row, "waiting", null);
988 return;
989 }
990 // Something arrived during the step: another step reads it.
991 if (json<Inbound[]>(row.inbox, []).length && row.steps < plan.steps + 2) {
992 if (answer.text) await db.batch([eventStatement(db, id, "text", agent.handle, answer.text)]);
993 await wake(env, row.agent_id, id);
994 return;
995 }
996 const report = answer.text.trim() || "I finished without anything to report.";
997 await db.batch([eventStatement(db, id, "result", agent.handle, report)]);
998 row = await setStatus(env, row, "done", null, { summary: report.slice(0, MAX_REPORT) });
999 await finished(env, id);
1000}
1001
1002/**
1003 * After a session is over: a root reports in its conversation; a child
1004 * hands its result to its parent and wakes it once its siblings are done.
1005 */
1006async function finished(env: SessionEnv, id: string): Promise<void> {
1007 const db = env.DB;
1008 const row = await sessionRow(db, id);
1009 if (!row) return;
1010 // Everything under a stopped or failed session stops too.
1011 if (row.status === "stopped" || row.status === "failed") await stopChildren(env, row.id, "Its parent session ended.");
1012 const agent = await agentRow(db, row.agent_id);
1013 if (row.parent_id) {
1014 const parent = await sessionRow(db, row.parent_id);
1015 if (!parent || OVER.includes(parent.status as AgentSessionStatus)) return;
1016 const who = row.subagent ? `your subagent ${row.subagent}` : `@${agent?.handle ?? "a colleague"}`;
1017 const body = row.status === "done" ? (row.summary ?? "(no result)") : `They couldn't finish (${row.status}): ${row.status_note ?? "no reason given"}.`;
1018 await pushInbox(db, parent.id, { kind: "child", by: who, body: body.slice(0, 8000) });
1019 await db.batch([eventStatement(db, parent.id, "child", agent?.handle ?? null, `${who} ${row.status === "done" ? "finished" : row.status}: ${row.title}`)]);
1020 await wake(env, parent.agent_id, parent.id);
1021 return;
1022 }
1023 // A root's report, where it was asked: the request's thread, or the conversation.
1024 if (row.status === "done" && row.summary && agent) {
1025 const mention = row.kind === "chat" && row.asked_by_username ? `@${row.asked_by_username} ` : "";
1026 await chatClient(env.CHAT)
1027 .postAsAgent(row.workspace, row.channel_id, row.agent_id, {
1028 body: `${mention}${row.summary}`.slice(0, MAX_REPORT),
1029 thread_root: row.thread_root,
1030 hops: row.hops,
1031 asked_by: row.asked_by,
1032 asker: json<AskerAccess | null>(row.asker, null),
1033 chain: json<string[]>(row.chain, []),
1034 })
1035 .catch((error: unknown) => console.error("agents: a session's report was not posted", row.id, String(error)));
1036 } else if ((row.status === "stopped" || row.status === "failed") && agent && row.status_note) {
1037 await postInThread(env, row, agent, `${row.status === "failed" ? "This session failed" : "This session stopped"}: ${row.status_note}`);
1038 }
1039 await refreshCard(env, row);
1040}
1041
1042export async function pushInbox(db: D1Database, id: string, item: Inbound): Promise<void> {
1043 const row = await db.prepare("SELECT inbox FROM agent_sessions WHERE id = ?").bind(id).first<{ inbox: string }>();
1044 const list = json<Inbound[]>(row?.inbox, []);
1045 list.push(item);
1046 await db.prepare("UPDATE agent_sessions SET inbox = ?, updated_at = ? WHERE id = ?").bind(JSON.stringify(list.slice(-20)), iso(), id).run();
1047}
1048
1049/** Stops every live session under `id`. */
1050async function stopChildren(env: SessionEnv, id: string, note: string): Promise<void> {
1051 const db = env.DB;
1052 const children = await db
1053 .prepare("SELECT * FROM agent_sessions WHERE parent_id = ? AND status IN ('queued','working','waiting','needs_approval')")
1054 .bind(id)
1055 .all<SessionRow>();
1056 for (const child of children.results) {
1057 await db.prepare("UPDATE agent_sessions SET status = 'stopped', status_note = ?, finished_at = ?, updated_at = ? WHERE id = ?").bind(note, iso(), iso(), child.id).run();
1058 await stopChildren(env, child.id, note);
1059 }
1060}
1061
1062/** Stops a session and everything under it, by a person. */
1063export async function stop(env: SessionEnv, row: SessionRow, by: string): Promise<SessionRow> {
1064 const fresh = await setStatus(env, row, "stopped", `Stopped by @${by}.`);
1065 await env.DB.batch([eventStatement(env.DB, row.id, "note", null, `Stopped by @${by}.`)]);
1066 await stopChildren(env, row.id, `Stopped by @${by}.`);
1067 if (row.parent_id) await finished(env, row.id);
1068 else await refreshCard(env, fresh);
1069 return fresh;
1070}
1071
1072/** A person's message to a session: read at its next step; a finished root goes on again. */
1073export async function steer(env: SessionEnv, row: SessionRow, by: string, body: string): Promise<SessionRow> {
1074 const db = env.DB;
1075 await pushInbox(db, row.id, { kind: "steer", by, body: body.slice(0, 4000) });
1076 await db.batch([eventStatement(db, row.id, "steer", by, body)]);
1077 if (row.status === "working" || row.status === "waiting" || row.status === "queued") {
1078 if (row.status !== "working") await wake(env, row.agent_id, row.id);
1079 return (await sessionRow(db, row.id))!;
1080 }
1081 // Over (or at its cap): it picks up again with its context, a fresh set of steps.
1082 if (row.status === "needs_approval") return (await sessionRow(db, row.id))!;
1083 await db.prepare("UPDATE agent_sessions SET status = 'queued', status_note = NULL, steps = MIN(steps, ?), finished_at = NULL, updated_at = ? WHERE id = ?").bind(Math.max(0, MAX_STEPS - 3), iso(), row.id).run();
1084 const fresh = (await sessionRow(db, row.id))!;
1085 await refreshCard(env, fresh);
1086 await wake(env, row.agent_id, row.id);
1087 return fresh;
1088}
1089
1090/** Raises a session's cap past what it has spent and lets it go on. */
1091export async function approve(env: SessionEnv, row: SessionRow, by: string, capMicros: number): Promise<SessionRow> {
1092 const db = env.DB;
1093 await db.prepare("UPDATE agent_sessions SET cap_micros = ?, status = 'queued', status_note = NULL, updated_at = ? WHERE id = ?").bind(Math.floor(capMicros), iso(), row.id).run();
1094 await db.batch([eventStatement(db, row.id, "note", null, `@${by} raised its cap to ${dollars(capMicros)}.`)]);
1095 const fresh = (await sessionRow(db, row.id))!;
1096 await refreshCard(env, fresh);
1097 await wake(env, row.agent_id, row.id);
1098 return fresh;
1099}
1100
1101/**
1102 * After an Ask-first card was answered (cards.ts): the session that asked
1103 * reads the answer at its next step and goes on, if it was waiting for it.
1104 * A reply (no session) hears nothing more: the card says what happened.
1105 */
1106export async function resumeAfterDecision(env: SessionEnv, request: { session_id: string | null; decided_by?: string | null }, body: string): Promise<void> {
1107 if (!request.session_id) return;
1108 const db = env.DB;
1109 const row = await sessionRow(db, request.session_id);
1110 if (!row || OVER.includes(row.status as AgentSessionStatus)) return;
1111 await pushInbox(db, row.id, { kind: "child", by: "the Ask-first card", body: body.slice(0, 8000) });
1112 await db.batch([eventStatement(db, row.id, "note", null, body.slice(0, 2000))]);
1113 const waiting = await db.prepare("SELECT COUNT(*) AS n FROM agent_ability_requests WHERE session_id = ? AND status = 'pending'").bind(row.id).first<{ n: number }>();
1114 if (row.status === "needs_approval" && (waiting?.n ?? 0) === 0) {
1115 await db.prepare("UPDATE agent_sessions SET status = 'queued', status_note = NULL, updated_at = ? WHERE id = ?").bind(iso(), row.id).run();
1116 const fresh = (await sessionRow(db, row.id))!;
1117 await refreshCard(env, fresh);
1118 }
1119 await wake(env, row.agent_id, row.id);
1120}
1121
1122/** Tells whoever asked, and the agent's maker, that a session waits for more budget. */
1123async function notifyApproval(env: SessionEnv, row: SessionRow, agent: Row): Promise<void> {
1124 if (!env.NOTIFY) return;
1125 // The card's own buttons ride along, so it can be approved from the notification.
1126 const { root } = await speaker(env.DB, row);
1127 const href = `/${row.workspace}/-/agents/${agent.handle}/sessions/${row.id}`;
1128 const card = root.card_message_id
1129 ? { channel_id: root.channel_id, message_id: root.card_message_id, actions: sessionActions("needs_approval", row.cap_micros, row.charged_micros, href) }
1130 : null;
1131 const targets = new Set<string>();
1132 if (row.asked_by_username) targets.add(row.asked_by_username);
1133 if (agent.created_by) targets.add(agent.created_by);
1134 for (const username of targets) {
1135 await env.NOTIFY.fetch("https://service/rpc/notify", {
1136 method: "POST",
1137 headers: { "content-type": "application/json" },
1138 body: JSON.stringify({
1139 target: { username },
1140 notification: {
1141 id: `approval:${row.id}:${row.cap_micros ?? 0}`,
1142 kind: "approval",
1143 workspace: row.workspace,
1144 title: `${agent.display_name} needs more budget`,
1145 body: `"${row.title}" reached its cap of ${dollars(row.cap_micros ?? 0)}.`,
1146 href,
1147 actor: { kind: "agent", id: agent.id, name: agent.display_name, avatar_seed: agent.avatar_seed || agent.handle, look: readLook(agent.look ?? null) },
1148 // While that conversation is open the card is there already: no toast.
1149 channel_id: root.channel_id,
1150 card,
1151 created_at: iso(),
1152 },
1153 }),
1154 }).catch(() => undefined);
1155 }
1156}
1157
1158/** Sessions stuck mid-step (their desk died): picked up again. */
1159export async function sweep(env: SessionEnv): Promise<number> {
1160 const before = new Date(Date.now() - 20 * 60_000).toISOString();
1161 const stuck = await env.DB.prepare(
1162 "SELECT id, agent_id FROM agent_sessions WHERE (status = 'working' AND step_started_at < ?) OR (status = 'queued' AND updated_at < ?) LIMIT 50",
1163 )
1164 .bind(before, before)
1165 .all<{ id: string; agent_id: string }>();
1166 for (const s of stuck.results) await wake(env, s.agent_id, s.id).catch(() => undefined);
1167 return stuck.results.length;
1168}
1169
1170type ToolPorts = import("./tools.ts").ToolPorts;