services/agents/src/reply.ts · flagon-io/g1t

.prepare(`${selectAgents("a.workspace_id = ?3 AND a.archived_at IS NULL AND a.id <> ?4 AND a.scope = 'workspace'")} ORDER BY a.builtin DESC, a.handle LIMIT 50`)

95

.bind(...periods(now), workspaceId, selfId)

96

.all<Row>(),

97

teamsOf,

98

]);

99

return rows.results.map((row) => {

100

const agent = toAgent(row, now);

101

return {

102

handle: agent.handle,

103

display_name: agent.display_name,

104

role: agent.role,

105

title: agent.title,

106

teams: onTeams.get(agent.id) ?? [],

107

responsibilities: agent.responsibilities,

108

status: agent.status,

109

spent_month_micros: agent.spent_month_micros,

110

monthly_micros: agent.budget.monthly_micros,

};

});

}

/** The agent's latest sessions in this conversation, one line each. */

116

async function sessionsHere(db: D1Database, agentId: string, channelId: string): Promise<string | null> {

117

const rows = await db

118

.prepare(

119

"SELECT id, title, status, summary, created_at FROM agent_sessions WHERE agent_id = ? AND channel_id = ? AND parent_id IS NULL ORDER BY created_at DESC LIMIT 5",

120

)

121

.bind(agentId, channelId)

122

.all<{ id: string; title: string; status: string; summary: string | null; created_at: string }>();

123

if (!rows.results.length) return null;

124

return rows.results

125

.map((s) => `- "${s.title}" (${s.status}, ${s.created_at.slice(0, 10)})${s.summary ? `: ${s.summary.replace(/\s+/g, " ").slice(0, 300)}` : ""}`)

.join("\n");

}

/**

* Consulting a colleague:

131

* the colleague answers in a nested turn that posts nothing, with the same

132

* audience (so it can read no more than the conversation may), the same

133

* asker, one hop further, its own routing limits, on the same model

134

* session, billed to this reply. A compact card in the thread says who

135

* asked whom. Consults don't nest: the colleague can't consult in turn.

136

*/

137

function consulting(input: {

db: D1Database;

row: Row;

delivery: DeskWork;

send: Send;

policy: Policy;

ownModel: boolean;

sessionModel: string | null;

145

own: string | null;

146

ports: (consult: ToolPorts["consult"]) => ToolPorts;

hops: number;

now: Date;

surface: Surface;

asker: { name: string; display_name: string | null; access: NonNullable<AgentDelivery["asker"]> | null };

151

spent: { tokens: Tokens; cost: number };

152

}): { ask: ToolPorts["consult"]; attach(box: ToolBox): void } {

153

let parent: ToolBox | null = null;

154

const noNesting: ToolPorts["consult"] = async () => ({ ok: false, message: "Consults don't nest: answer with what you have." });

155

const ask: ToolPorts["consult"] = async (handle, question) => {

156

const { row, delivery } = input;

157

const colleague = await input.db

158

.prepare("SELECT * FROM agents WHERE workspace_id = ? AND handle = ? AND archived_at IS NULL AND scope = 'workspace'")

159

.bind(row.workspace_id, handle)

160

.first<Row>();

161

if (!colleague || colleague.id === row.id) return { ok: false, message: `There is no other agent called @${handle} here.` };

162

// No ping-pong: never back to the agent that sent this work.

163

if ((delivery.chain ?? []).at(-1) === colleague.id) return { ok: false, message: `@${handle} sent you this work; answer with what you have.` };

164

if (input.hops + 1 > CHAT_MAX_HOPS) return { ok: false, message: "This request has been passed along too many times; answer with what you have." };

165

const definition = definitionOf(colleague);

166

const allowed = allowedProviders(definition.routing, input.own);

167

if (input.ownModel ? !allowed.own : !allowed.hosted) return { ok: false, message: `@${handle} can't use the model this conversation runs on.` };

168

const model = replyModel(

169

input.policy,

170

{ ...definition.routing, pinned: input.ownModel ? definition.routing.pinned : null },

171

{ named: input.ownModel ? input.sessionModel : null },

172

);

173

const tools = parent

174

? parent.forColleague(input.ports(noNesting), { agentId: colleague.id, notConsult: [colleague.handle, row.handle], hops: input.hops + 1, maxHops: input.hops + 1 })

175

: null;

176

const system = systemPrompt({

177

agent: { ...definition, id: colleague.id },

178

workspace: delivery.workspace,

179

channel: { kind: delivery.channel_kind, name: delivery.channel_name },

180

asker: input.asker,

181

today: input.now,

182

tools: tools ? { code: tools.definitions().some((tool) => tool.name === "read_file") } : null,

183

consultedBy: row.handle,

184

});

185

const result = await runTurn(input.send, {

186

model: model.model,

187

system,

188

messages: [{ role: "user", content: `@${row.handle} (agent) asks you: ${question}` }],

189

tools,

190

price: input.ownModel ? null : model.price,

191

});

192

input.spent.tokens = addTokens(input.spent.tokens, result.tokens);

193

input.spent.cost += input.ownModel ? 0 : result.cost;

194

const answer = result.text || "(no answer)";

195

const exchange = `Q: ${question}\nA: ${answer}`;

196

// The exchange, collapsed, in the thread. No @ in its text: it wakes nobody.

197

await input.surface

198

.post(`${row.display_name} asked ${colleague.display_name}`, {

199

kind: "consult",

200

title: `${row.display_name} asked ${colleague.display_name}`,

201

detail: exchange.length > 1000 ? `${exchange.slice(0, 1000)}…` : exchange,

state: null,

href: null,

})

.catch((error: unknown) => console.error("agents: a consult card was not posted", String(error)));

206

return { ok: true, colleague: colleague.handle, answer };

};

return {

ask,

attach(box) {

parent = box;

},

};

}

type Outcome = {

status: "replied" | "steered" | "blocked" | "skipped" | "failed";

218

error?: string | null;

219

reply_id?: string | null;

220

model?: string | null;

221

tier?: string | null;

tokens?: Tokens;

cost?: number;

charged?: number;

/** The effort level it ran at. */

226

effort?: string | null;

};

/**

* What a desk is handed: a message to answer, or (`hello`) the agent's

231

* first message to the person who made it, in the DM that just opened.

232

*/

233

export type DeskWork = AgentDelivery & { hello?: boolean };

234

235

/** How a limit's refusal reads in chat, in the agent's voice. */

236

function noticeFor(reason: string, message: string): string {

237

if (reason.startsWith("budget_") || reason === "workspace_agent_budget" || message.startsWith("My settings") || message.startsWith("I ")) return message;

238

if (reason === "no_model") return `I can't reply yet: ${message}`;

239

return `I can't reply right now: ${message}`;

240

}

241

242

/** The session a message in this thread is for: a reply under one of this agent's session cards. */

243

async function steeredSession(db: D1Database, agentId: string, threadRoot: string | null): Promise<SessionRow | null> {

244

if (!threadRoot) return null;

245

const found = await db.prepare("SELECT id FROM agent_sessions WHERE agent_id = ? AND card_message_id = ?").bind(agentId, threadRoot).first<{ id: string }>();

246

return found ? sessionRow(db, found.id) : null;

}

/**

* Answers `delivery` as its agent. Never throws: every way it ends is

251

* recorded on the reply's row. A message handed over twice is answered

252

* once.

253

*/

254

export async function reply(env: ReplyEnv, delivery: DeskWork, now = new Date()): Promise<void> {

255

const db = env.DB;

256

const row = await db.prepare("SELECT * FROM agents WHERE id = ?").bind(delivery.agent_id).first<Row>();

257

if (!row || row.archived_at || row.workspace_id !== delivery.workspace_id) return;

258

const id = newId("arp", now.getTime());

259

const claimed = await db

260

.prepare(

261

`INSERT INTO agent_replies (id, agent_id, workspace_id, channel_id, message_id, asked_by, asked_by_username, channel_name, agent_version, status, created_at)

262

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'working', ?)

263

ON CONFLICT (agent_id, message_id) DO NOTHING RETURNING id`,

264

)

265

.bind(id, row.id, row.workspace_id, delivery.channel_id, delivery.message_id, delivery.asked_by, delivery.asker?.username ?? null, delivery.channel_name, row.version, now.toISOString())

266

.first<{ id: string }>();

267

if (!claimed) return;

268

269

const surface = surfaceFor(env.CHAT, delivery);

270

// 👀 as soon as the desk has it, while everything else goes on.

271

const acknowledging = surface.acknowledge();

272

const slug = delivery.workspace.toLowerCase();

273

274

// What the answer used, once there is one: counted however the reply ends.

275

let usage: Partial<Outcome> = {};

276

// What the tools did, for the audit table, and who the audience was.

277

let toolCalls: ToolCall[] = [];

278

let audienceHash: string | null = null;

279

280

const finish = async (outcome: Outcome) => {

const statements = [

db

.prepare(

`UPDATE agent_replies SET status = ?, error = ?, reply_id = ?, model = ?, tier = ?, input_tokens = ?, output_tokens = ?,

285

cost_micros = ?, charged_micros = ?, tool_count = ?, finished_at = ?, effort = ? WHERE id = ?`,

)

.bind(

outcome.status,

outcome.error?.slice(0, 1000) ?? null,

290

outcome.reply_id ?? null,

291

outcome.model ?? null,

292

outcome.tier ?? null,

293

(outcome.tokens?.input ?? 0) + (outcome.tokens?.cacheRead ?? 0) + (outcome.tokens?.cacheWrite ?? 0),

294

outcome.tokens?.output ?? 0,

295

outcome.cost ?? 0,

296

outcome.charged ?? 0,

297

toolCalls.length,

298

new Date().toISOString(),

299

outcome.effort ?? null,

300

id,

301

),

302

// Every tool call: what was asked, for whom, and whether it was read or withheld.

303

...toolCalls.map((call, n) =>

304

db

305

.prepare(

306

`INSERT INTO agent_tool_calls (id, reply_id, agent_id, workspace_id, asked_by, tool, args, audience_hash, outcome, bytes, created_at)

307

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,

308

)

309

.bind(`${id}_${n}`, id, row.id, row.workspace_id, delivery.asked_by, call.tool, call.args, audienceHash ?? "", call.outcome, call.bytes, now.toISOString()),

310

),

311

];

312

await db.batch(statements);

313

// ✅ when it answered; 👀 taken back after a notice, an apology or nothing.

314

await acknowledging;

315

await surface.settle(outcome.status === "replied" || outcome.status === "steered" ? "done" : "withdrawn");

316

};

317

318

/** Says once, in this conversation, why the agent cannot answer; a repeat within hours is kept back. */

319

const notice = async (message: string, reason: string) => {

320

// An agent's first words are never an excuse: without a model, a fixed hello.

321

if (delivery.hello) {

322

const posted = await surface.post(fixedHello(row, delivery.asker?.username ?? null)).catch(() => null);

323

return await finish({ status: "blocked", error: reason, reply_id: posted });

324

}

325

const since = new Date(now.getTime() - NOTICE_QUIET_MS).toISOString();

326

const recent = await db

327

.prepare(

328

`SELECT 1 FROM agent_replies WHERE agent_id = ? AND channel_id = ? AND status = 'blocked' AND error = ?

329

AND reply_id IS NOT NULL AND created_at > ? AND id <> ? LIMIT 1`,

330

)

331

.bind(row.id, delivery.channel_id, reason, since, id)

332

.first();

333

const posted = recent ? null : await surface.post(message).catch(() => null);

334

await finish({ status: "blocked", error: reason, reply_id: posted });

};

try {

// A reply under one of its session cards steers that session; nothing is answered here.

339

const steered = delivery.hello ? null : await steeredSession(db, row.id, delivery.thread_root);

340

if (steered) {

341

const history = await surface.history(HISTORY_LIMIT).catch(() => [] as SurfaceMessage[]);

342

const message = history.find((m) => m.id === delivery.message_id);

343

if (message?.body.trim()) {

344

await steer(env as unknown as SessionEnv, steered, message.author.name, message.body);

345

return await finish({ status: "steered" });

}

}

// Read the conversation while showing that the agent is on it.

350

// A hello has no conversation yet: it is asked to introduce itself.

351

// And who is here: said every turn, so the agent knows who reads it and who doesn't.

352

// And its teams, from their pages: who leads, who owns what, who is around.

353

const [, history, conversationHere, teamsHere] = await Promise.all([

354

surface.typing(),

355

delivery.hello ? Promise.resolve([]) : surface.history(HISTORY_LIMIT),

356

delivery.hello ? Promise.resolve(null) : surface.conversation(),

357

loadTeams(env, slug, row.workspace_id, { id: row.id }).catch(() => null),

358

]);

359

// A personal agent answers only its member, in the DM of the two of them (access.ts).

360

const refused = personalRefusal(row, {

361

asked_by: delivery.asked_by,

362

channel_kind: delivery.channel_kind,

363

members: conversationHere ? conversationHere.people + conversationHere.agents : null,

364

});

365

if (refused) return await notice(refused, "personal_agent");

366

const conversation = delivery.hello ? [{ role: "user" as const, content: helloAsk(delivery.asker?.username ?? null) }] : turns(history, row.id);

367

if (!conversation.length) return await finish({ status: "skipped", error: "nothing to answer" });

368

const author = askerIn(history, delivery);

369

const askerName = delivery.asker?.username ?? author?.name ?? null;

370

// Every agent knows its colleagues; @g1t also steps up a tier to decide

371

// who gets the work in a long thread.

372

const specialists = await team(db, row.workspace_id, row.id, now, teamsOfAgents(env, slug));

373

const definition = definitionOf(row);

374

// Its effort setting decides where the reply starts and how hard the model reasons.

375

const plan = effortPlan(effortOf(definition.routing), row.builtin ? orchestratorTier(history.length, specialists.filter((a) => a.handle !== "g1t").length) : REPLY_TIER);

376

const hops = Math.max(0, Math.floor(delivery.hops || 0));

377

const chain = delivery.chain ?? [];

378

const sender = chain.length ? await db.prepare("SELECT handle FROM agents WHERE id = ?").bind(chain[chain.length - 1]).first<{ handle: string }>() : null;

379

const asker = { name: askerName ?? "someone", display_name: author?.display_name ?? null, access: delivery.asker ?? null };

380

let posted: string | null = null;

381

let spinOffs = 0;

382

383

const done = await metered(env, { row, payer: row, slug, task: "reply", start: plan.start, effort: plan.level, askerName, person: delivery.asker?.username ?? null, teams: teamsHere }, async (model) => {

384

// What colleagues consulted along the way used: billed to this reply.

385

const consulted = { tokens: NO_TOKENS, cost: 0 };

386

let toolbox: ToolBox | null = null;

387

// The "Your abilities outside g1t" section, once the tool box has them.

388

let abilitiesText: string | null = null;

389

let place: RecallPlace = { channel_id: delivery.channel_id, kind: delivery.channel_kind === "dm" ? "dm" : "private", people: [delivery.asked_by] };

390

if (!delivery.hello) {

391

try {

392

const audience = await Audience.build(slug, delivery.asked_by, audiencePorts(env, slug, delivery.channel_id));

393

place = { channel_id: delivery.channel_id, kind: audience.kind, people: audience.shared ? [delivery.asked_by] : audience.members.map((m) => m.id) };

394

const ports = (consult: ToolPorts["consult"]) => toolPorts(env, slug, row.workspace_id, delivery.channel_id, consult, row.id);

395

audienceHash = audience.hash;

396

const consult = consulting({

db,

row,

delivery,

send: model.send,

policy: model.policy,

402

ownModel: model.ownModel,

403

sessionModel: model.sessionModel,

own: model.own,

ports,

hops,

now,

surface,

asker,

spent: consulted,

});

const where = delivery.channel_kind === "dm" ? "a direct message" : `#${delivery.channel_name ?? "a channel"}`;

413

const actions = actionPorts(env as unknown as SessionEnv, {

414

agent: row,

415

place,

416

source: { kind: "message", ref: delivery.message_id, label: askerName ? `@${askerName} in ${where}` : where, channel_id: delivery.channel_id },

417

asker: { id: delivery.asked_by, username: askerName },

418

workspace: slug,

419

postCard: (card) => surface.post("", card).catch(() => null),

420

// Real work becomes a session, with its card in this conversation.

421

spinOff: async (title, goal) => {

422

if (spinOffs >= MAX_SPIN_OFFS) return { ok: false, message: "You've started enough sessions from this message." };

423

spinOffs++;

424

const session = await startSession(env as unknown as SessionEnv, {

agent: row,

kind: "chat",

title,

goal: `${askerName ? `@${askerName}` : "Someone"} asked in ${where}:\n\n${goal}`,

429

workspace: slug,

430

channel_id: delivery.channel_id,

431

channel_kind: delivery.channel_kind,

432

channel_name: delivery.channel_name,

433

thread_root: delivery.thread_root,

434

message_id: delivery.message_id,

435

asked_by: delivery.asked_by,

436

asked_by_username: askerName,

437

asker: delivery.asker ?? null,

chain: [...chain],

hops,

});

const cap = session.cap_micros ? ` with a cap of ${dollars(session.cap_micros)}` : "";

442

return { ok: true, message: `Started the session "${session.title}"${cap}. Its card is in the conversation and it reports back there. Tell them in a sentence; don't do the work here.` };

443

},

444

});

445

// Work handed to a colleague: the only way an agent gets another working (handoff.ts).

446

const viewer = audience.asker;

447

actions.handOff = handOffPort({

448

self: { id: row.id, handle: row.handle },

449

chain,

450

asker: delivery.asker ?? null,

451

agent: async (handle) => {

452

const found = await db

453

.prepare(selectAgents("a.workspace_id = ?3 AND a.handle = ?4 AND a.archived_at IS NULL AND a.scope = 'workspace'"))

454

.bind(...periods(now), row.workspace_id, handle)

455

.first<Row>();

456

if (!found) return null;

457

const agent = toAgent(found, now);

458

return { id: agent.id, handle: agent.handle, display_name: agent.display_name, builtin: !!found.builtin, status: agent.status };

459

},

460

person: async (handle) => {

461

if (!viewer) return false;

462

const members = await identityClient(env.IDENTITY).listMembers(slug, viewer);

463

return members.ok && members.value.some((m) => m.username.toLowerCase() === handle);

464

},

465

handOff: (colleagueId, brief) => surface.handOff(colleagueId, brief),

466

});

467

toolbox = new ToolBox(

468

audience,

469

ports(consult.ask),

470

{ agentId: row.id, notConsult: [row.handle, ...(sender ? [sender.handle] : [])], hops, maxHops: CHAT_MAX_HOPS },

[],

actions,

);

consult.attach(toolbox);

475

// Its abilities outside g1t (abilities.ts): offered and enforced by the tool box, for the person it acts for.

476

if (viewer && !delivery.hello) {

477

const sections = await abilitiesFor(env as unknown as AbilityEnv, { agent: row, definition, workspace: slug, asker: viewer }).catch(() => null);

478

if (sections) {

479

const saidByPeople = saidText([...history].reverse().filter((m) => m.author.kind === "user").slice(0, 3).map((m) => m.body));

480

toolbox.useAbilities(

481

sections,

482

abilityPorts(env as unknown as AbilityEnv, {

483

agent: row,

484

workspace: slug,

485

channel_id: delivery.channel_id,

486

session: null,

487

asker: { id: delivery.asked_by, username: askerName },

488

postCard: (card) => surface.post("", card).catch(() => null),

489

}),

490

saidByPeople,

491

definition.abilities.mcp_servers,

492

);

493

abilitiesText = abilitiesSection(sections);

}

}

} catch (error) {

// Without an audience nothing may be read: the reply goes on with this conversation only.

498

console.error("agents: no audience for a reply, so no tools", row.id, String(error));

499

}

500

}

501

// What people said last, for recalling what the workspace's artifacts say about it.

502

const said = [...history].reverse().filter((m) => m.author.kind === "user").slice(0, 3).map((m) => m.body);

503

const [facts, recent, passages, shelf] = delivery.hello

504

? [[], null, [], []]

505

: await Promise.all([

506

recall(db, row.id, place).catch(() => []),

507

sessionsHere(db, row.id, delivery.channel_id).catch(() => null),

508

toolbox ? toolbox.recall(recallQuery(said), definition.reading ?? []) : Promise.resolve([]),

509

// Its skills: named in the prompt, read with use_skill (skills.ts).

510

toolbox ? loadShelf(db, row.workspace_id, { id: row.id, skills_off: definition.skills_off }, teamSlugs(teamsHere)) : Promise.resolve([]),

511

]);

512

toolbox?.useShelf(shelf, (skillId, version) => readVersion(db, skillId, version));

const system = [

systemPrompt({

agent: {

...definition,

id: row.id,

teams: teamsHere?.teams.map((t) => t.name) ?? [],

519

// @g1t's job is fixed; what the workspace wrote is added to it.

520

instructions: row.builtin ? orchestratorInstructions(specialists.filter((a) => a.handle !== "g1t"), definition.instructions) : definition.instructions,

521

},

522

workspace: delivery.workspace,

523

channel: { kind: delivery.channel_kind, name: delivery.channel_name },

524

asker,

525

today: now,

526

tools: toolbox ? { code: toolbox.definitions().some((tool) => tool.name === "read_file") } : null,

527

// @g1t's team is in its job; everyone else is told who their colleagues are.

528

colleagues: row.builtin ? null : rosterLines(specialists),

529

recentSessions: recent,

530

conversation: conversationHere,

531

teams: teamsSection(row.id, teamsHere, now),

532

canHandOff: !!toolbox?.definitions().some((tool) => tool.name === "hand_off"),

533

abilities: abilitiesText,

534

handedOffBy: sender?.handle ?? null,

535

skills: skillsSection(shelf, toolbox?.definitions().map((tool) => tool.name) ?? []),

536

}),

537

memorySection(facts),

538

recallSection(passages),

]

.filter(Boolean)

.join("\n\n");

toolCalls = toolbox?.calls ?? [];

543

const answer = await runTurn(model.send, { model: model.model.model, system, messages: conversation as ModelMessage[], tools: toolbox, price: model.ownModel ? null : model.model.price, effort: model.effort });

544

// Post as soon as there is an answer; the bill is settled after.

545

// It wakes nobody, @mentions or not: colleagues get work only through hand_off.

546