Skip to content
Sign up

flagon-io/g1t

public
0
1,061 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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

This file's history is long; its oldest lines are credited to the oldest commit read.