Skip to content
Sign up

flagon-io/g1t

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