diff --git a/deploy/policy.yaml b/deploy/policy.yaml index f365564..b532192 100644 --- a/deploy/policy.yaml +++ b/deploy/policy.yaml @@ -30,13 +30,8 @@ venue_instructions: # #app-alerts C099YDUFQ9Z: >- prod alert feed; noah follows it religiously. you are the only triager here, and the - ticket is where triage lands (real receipts from ops_read/github_read, not - channel-history guesses). suspected user-facing or revenue impact belongs to - <@U096GG0F354>. Every alert is worth independent investigation — do not assume - another person's ownership, a prior false-positive, or similarity to an older - alert covers a new one; inspect each on its own and respond with the finding. - Narrow exception: nightly VRT alert/workflow posts are dead signal unless Noah - explicitly says they are revived. + ticket is where triage lands (real receipts, not channel-history guesses). suspected + user-facing or revenue impact belongs to <@U096GG0F354>. nightly VRT posts are dead signal. # #releases-eng C0BAQK9PECA: >- noah and anya run the train; julia runs qa. when the release train posts a release ready diff --git a/drizzle/0004_conversations_last_woken.sql b/drizzle/0004_conversations_last_woken.sql new file mode 100644 index 0000000..0f24bab --- /dev/null +++ b/drizzle/0004_conversations_last_woken.sql @@ -0,0 +1,4 @@ +ALTER TABLE `conversations` ADD `last` text NOT NULL;--> statement-breakpoint +ALTER TABLE `conversations` ADD `woken` integer NOT NULL;--> statement-breakpoint +ALTER TABLE `conversations` DROP COLUMN `judged`;--> statement-breakpoint +ALTER TABLE `conversations` DROP COLUMN `wake_why`; \ No newline at end of file diff --git a/drizzle/meta/0004_snapshot.json b/drizzle/meta/0004_snapshot.json new file mode 100644 index 0000000..423ed10 --- /dev/null +++ b/drizzle/meta/0004_snapshot.json @@ -0,0 +1,274 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "0be1bb01-172d-4215-9da5-e36b2411ccd1", + "prevId": "2698e3b3-5c25-42db-8694-53e2d539cddc", + "tables": { + "conversations": { + "name": "conversations", + "columns": { + "channel": { + "name": "channel", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "thread_ts": { + "name": "thread_ts", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "since": { + "name": "since", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "last": { + "name": "last", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "direct": { + "name": "direct", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "woken": { + "name": "woken", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "conversations_channel_thread_ts_pk": { + "columns": ["channel", "thread_ts"], + "name": "conversations_channel_thread_ts_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "muted_threads": { + "name": "muted_threads", + "columns": { + "channel": { + "name": "channel", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "thread_ts": { + "name": "thread_ts", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "why": { + "name": "why", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "at": { + "name": "at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "muted_threads_channel_thread_ts_pk": { + "columns": ["channel", "thread_ts"], + "name": "muted_threads_channel_thread_ts_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "tasks": { + "name": "tasks", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "spec": { + "name": "spec", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "waiting_on": { + "name": "waiting_on", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "waiting_why": { + "name": "waiting_why", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "wake_at": { + "name": "wake_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "outcome": { + "name": "outcome", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "report": { + "name": "report", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "seen_at": { + "name": "seen_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "channel": { + "name": "channel", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "thread_ts": { + "name": "thread_ts", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tier": { + "name": "tier", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'high'" + }, + "interruptions": { + "name": "interruptions", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "opened_at": { + "name": "opened_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "tasks_dispatch": { + "name": "tasks_dispatch", + "columns": ["status", "opened_at"], + "isUnique": false + }, + "tasks_due": { + "name": "tasks_due", + "columns": ["status", "wake_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "tasks_waiting_on": { + "name": "tasks_waiting_on", + "value": "(\"tasks\".\"status\" = 'waiting') = (\"tasks\".\"waiting_on\" IS NOT NULL)" + }, + "tasks_wake_at": { + "name": "tasks_wake_at", + "value": "\"tasks\".\"wake_at\" IS NULL OR \"tasks\".\"status\" = 'waiting'" + }, + "tasks_waiting_why": { + "name": "tasks_waiting_why", + "value": "\"tasks\".\"waiting_on\" IS NOT 'human' OR (\"tasks\".\"waiting_why\" IS NOT NULL AND trim(\"tasks\".\"waiting_why\") <> '')" + }, + "tasks_outcome": { + "name": "tasks_outcome", + "value": "(\"tasks\".\"status\" = 'done') = (\"tasks\".\"outcome\" IS NOT NULL)" + }, + "tasks_report": { + "name": "tasks_report", + "value": "\"tasks\".\"status\" <> 'done' OR (\"tasks\".\"report\" IS NOT NULL AND trim(\"tasks\".\"report\") <> '')" + } + } + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 78581eb..dbc19ae 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -29,6 +29,13 @@ "when": 1788812956600, "tag": "0003_conversations", "breakpoints": true + }, + { + "idx": 4, + "version": "6", + "when": 1788904893450, + "tag": "0004_conversations_last_woken", + "breakpoints": true } ] } diff --git a/src/codex.ts b/src/codex.ts index 1ae9cd9..34ce633 100644 --- a/src/codex.ts +++ b/src/codex.ts @@ -1,6 +1,5 @@ import { codexThread, maybeRotateGateway, type Tools } from "@bevyl-ai/agent-tools"; import { inject, singleton } from "tsyringe"; -import { LedgerService } from "./ledger-service"; import type { Task } from "./ledger/schema"; import { log } from "./log"; import { POLICY, type Policy } from "./policy"; @@ -16,7 +15,6 @@ export class Codex { constructor( @inject(POLICY) private readonly policy: Policy, private readonly soul: Soul, - private readonly ledger: LedgerService, private readonly workspaces: Workspaces, ) {} @@ -24,9 +22,8 @@ export class Codex { return this.once("resident", residentTools, {}, this.policy.turns.timeout_ms, prompt); } - async shouldAgentRespond(prompt: string): Promise { - await this.once("ear", earTools, this.policy.models.low, this.policy.turns.timeout_ms, prompt); - return this.ledger.wantsResponse(); + judge(prompt: string): Promise { + return this.once("ear", earTools, this.policy.models.low, this.policy.turns.timeout_ms, prompt); } async runWorker(taskId: string, tier: Task["tier"], next: () => string | null): Promise { @@ -34,12 +31,7 @@ export class Codex { const { thread, close } = await this.thread("worker", workerTools(taskId), models[tier]); try { for (let prompt = next(); prompt !== null; prompt = next()) - await this.turn(thread, taskId, prompt, executions.turn_timeout_ms).catch( - (error: unknown) => { - this.ledger.interrupt(taskId); - throw error; - }, - ); + await this.turn(thread, taskId, prompt, executions.turn_timeout_ms); } finally { close(); } @@ -72,7 +64,10 @@ export class Codex { if (event.type === "item.completed") { const { item } = event; if (item.type === "command_execution") log.info(label, { line: `$ ${item.command}` }); - else if (item.type === "mcp_tool_call") log.info(label, { line: `⚙ ${item.tool}` }); + else if (item.type === "mcp_tool_call") + log.info(label, { + line: `⚙ ${item.tool} ${JSON.stringify(item.arguments)}${item.error ? ` ✗ ${item.error.message}` : ""}`, + }); else if (item.type === "agent_message") log.info(label, { line: `● ${item.text}` }); } else if (event.type === "turn.failed") { maybeRotateGateway({ reason: event.error.message }); diff --git a/src/ledger-service.ts b/src/ledger-service.ts index 023ad04..8027e73 100644 --- a/src/ledger-service.ts +++ b/src/ledger-service.ts @@ -1,4 +1,4 @@ -import { and, asc, count, eq, isNotNull, like, lte, min, or, sql } from "drizzle-orm"; +import { and, asc, count, eq, like, lte, min, or, sql } from "drizzle-orm"; import { drizzle, type BunSQLiteDatabase } from "drizzle-orm/bun-sqlite"; import { migrate } from "drizzle-orm/bun-sqlite/migrator"; import { inject, singleton, type InjectionToken } from "tsyringe"; @@ -18,6 +18,8 @@ export function openDb(path: string): Db { return db; } +export const WANTED = or(eq(conversations.direct, true), eq(conversations.woken, true)); + export function thread( table: typeof conversations | typeof mutedThreads, channel: string, @@ -136,11 +138,7 @@ ${text}`, } rendered(convos: Conversation[], settled: Task[]): void { - for (const convo of convos) - this.db - .delete(conversations) - .where(thread(conversations, convo.channel, convo.threadTs)) - .run(); + for (const convo of convos) this.settle(convo); for (const task of settled) this.db .update(tasks) @@ -204,41 +202,45 @@ ${text}`, heard(channel: string, threadTs: string, ts: string, direct: boolean): void { this.db .insert(conversations) - .values({ channel, threadTs, since: ts, direct, judged: direct, wakeWhy: null }) + .values({ channel, threadTs, since: ts, last: ts, direct, woken: false }) .onConflictDoUpdate({ target: [conversations.channel, conversations.threadTs], - set: { - direct: sql`${conversations.direct} OR ${direct}`, - judged: sql`${conversations.judged} AND ${direct}`, - }, + set: { last: ts, direct: sql`${conversations.direct} OR ${direct}` }, }) .run(); } - wakeFor(channel: string, threadTs: string, why: string): boolean { - const hit = this.db - .update(conversations) - .set({ wakeWhy: why }) - .where(thread(conversations, channel, threadTs)) - .returning({ channel: conversations.channel }) - .get(); - return hit !== undefined; + held(convos: Conversation[]): void { + for (const convo of convos) if (!this.wanted(convo)) this.settle(convo); } - wantsResponse(): boolean { - const wanted = this.db.query.conversations - .findFirst({ where: or(eq(conversations.direct, true), isNotNull(conversations.wakeWhy)) }) + private wanted(convo: Conversation): boolean { + const row = this.db.query.conversations + .findFirst({ where: and(thread(conversations, convo.channel, convo.threadTs), WANTED) }) .sync(); - return wanted !== undefined; + return row !== undefined; } - judged(convos: Conversation[]): void { - for (const convo of convos) - this.db - .update(conversations) - .set({ judged: true }) - .where(thread(conversations, convo.channel, convo.threadTs)) - .run(); + private settle(convo: Conversation): void { + const where = thread(conversations, convo.channel, convo.threadTs); + const gone = this.db + .delete(conversations) + .where(and(where, eq(conversations.last, convo.last))) + .returning() + .get(); + if (!gone) this.db.update(conversations).set({ since: convo.last }).where(where).run(); + } + + wake(channel: string, threadTs: string): void { + this.db + .update(conversations) + .set({ woken: true }) + .where(thread(conversations, channel, threadTs)) + .run(); + } + + wantsResponse(): boolean { + return this.db.query.conversations.findFirst({ where: WANTED }).sync() !== undefined; } muted(channel: string, threadTs: string): string | null { diff --git a/src/ledger/schema.ts b/src/ledger/schema.ts index 2884ca7..3b1ba8c 100644 --- a/src/ledger/schema.ts +++ b/src/ledger/schema.ts @@ -57,9 +57,9 @@ export const conversations = sqliteTable( channel: text("channel").notNull(), threadTs: text("thread_ts").notNull(), since: text("since").notNull(), + last: text("last").notNull(), direct: integer("direct", { mode: "boolean" }).notNull(), - judged: integer("judged", { mode: "boolean" }).notNull(), - wakeWhy: text("wake_why"), + woken: integer("woken", { mode: "boolean" }).notNull(), }, (t) => [primaryKey({ columns: [t.channel, t.threadTs] })], ); diff --git a/src/main.ts b/src/main.ts index 08a0988..acafd9d 100644 --- a/src/main.ts +++ b/src/main.ts @@ -9,8 +9,8 @@ import { Scheduler } from "./scheduler"; import { log } from "./log"; import { POLICY, POLICY_PATH, loadPolicy } from "./policy"; -const scheduler = container.resolve(Scheduler); await container.resolve(Roster).load(); +const scheduler = container.resolve(Scheduler); log.info("service started"); const socket = container.resolve(SocketModeClient); diff --git a/src/policy.ts b/src/policy.ts index 8a407ef..086c329 100644 --- a/src/policy.ts +++ b/src/policy.ts @@ -20,7 +20,6 @@ const PolicySchema = z.object({ max_turns: z.number().default(40), turn_timeout_ms: z.number().default(30 * 60 * 1000), max_attempts: z.number().default(3), - backoff_ms: z.number().default(30_000), }) .prefault({}), tasks: z.object({ park_after_ms: z.number().default(48 * 60 * 60 * 1000) }).prefault({}), diff --git a/src/prompt-renderer.ts b/src/prompt-renderer.ts index 239028a..9583644 100644 --- a/src/prompt-renderer.ts +++ b/src/prompt-renderer.ts @@ -38,8 +38,8 @@ export class PromptRenderer { return parts.join("\n\n"); } - noise(convos: Conversation[]): Promise { - return this.batch(convos); + async noise(convos: Conversation[]): Promise { + return LEGEND + (await this.batch(convos)); } private async batch(convos: Conversation[]): Promise { @@ -61,20 +61,25 @@ export class PromptRenderer { private header(convo: Conversation): string { const muted = this.ledger.muted(convo.channel, convo.threadTs); - const notes = [muted ? `Muted: ${muted}` : "", convo.wakeWhy ?? ""].filter(Boolean); const head = `## <#${convo.channel}> thread=${convo.threadTs}`; - return notes.length > 0 ? `${head}\n${notes.join(" · ")}` : head; + return muted ? `${head}\nMuted: ${muted}` : head; } private async messages( convo: Conversation, ): Promise<{ earlier: MessageElement[]; fresh: MessageElement[] }> { - const { messages } = await this.web.conversations.replies({ - channel: convo.channel, - ts: convo.threadTs, - limit: 200, - }); - const all = messages ?? []; + const all: MessageElement[] = []; + let cursor: string | undefined; + do { + const page = await this.web.conversations.replies({ + channel: convo.channel, + ts: convo.threadTs, + limit: 200, + ...(cursor ? { cursor } : {}), + }); + all.push(...(page.messages ?? [])); + cursor = page.response_metadata?.next_cursor; + } while (cursor); return { earlier: all.filter((m) => m.ts && m.ts < convo.since).slice(-TAIL_LIMIT), fresh: all.filter((m) => m.ts && m.ts >= convo.since), diff --git a/src/scheduler.ts b/src/scheduler.ts index e19f811..7aeec85 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -8,7 +8,7 @@ import { and, asc, desc, eq, gt, isNull, or } from "drizzle-orm"; import { inject, instanceCachingFactory, registry, singleton } from "tsyringe"; import { Codex } from "./codex"; import { Debounced } from "./debounce"; -import { DB, LedgerService, openDb, type Db } from "./ledger-service"; +import { DB, LedgerService, openDb, WANTED, type Db } from "./ledger-service"; import { conversations, tasks } from "./ledger/schema"; import { log } from "./log"; import { loadPolicy, POLICY, POLICY_PATH, type Policy } from "./policy"; @@ -97,7 +97,11 @@ export class Scheduler { private async respond(): Promise { const convos = this.db.query.conversations - .findMany({ orderBy: [desc(conversations.direct), asc(conversations.since)], limit: BATCH }) + .findMany({ + where: WANTED, + orderBy: [desc(conversations.direct), asc(conversations.since)], + limit: BATCH, + }) .sync(); const settled = this.db.query.tasks .findMany({ @@ -114,24 +118,22 @@ export class Scheduler { for (const convo of direct) this.voice.open(convo); this.ledger.rendered(convos, settled); this.voice.begin(); - await this.codex.respond(prompt); - this.voice.close(direct); + await this.codex.respond(prompt).finally(() => { + this.voice.close(direct); + }); this.tick(); if (this.ledger.wantsResponse()) this.respondSoon(); } private async listenToNoise(): Promise { const unjudged = this.db.query.conversations - .findMany({ where: eq(conversations.judged, false) }) + .findMany({ where: and(eq(conversations.direct, false), eq(conversations.woken, false)) }) .sync(); - if (unjudged.length === 0) { - if (this.ledger.wantsResponse()) this.respondSoon(); - return; + if (unjudged.length > 0) { + await this.codex.judge(await this.prompts.noise(unjudged)); + this.ledger.held(unjudged); } - const prompt = await this.prompts.noise(unjudged); - const wanted = await this.codex.shouldAgentRespond(prompt); - this.ledger.judged(unjudged); - if (wanted) this.respondSoon(); + if (this.ledger.wantsResponse()) this.respondSoon(); } private async runWorker(taskId: string): Promise { @@ -140,15 +142,20 @@ export class Scheduler { const first = task(); if (first?.status !== "active") return; let turns = 0; - await this.codex.runWorker(taskId, first.tier, () => { - const t = task(); - return t?.status === "active" && turns++ < executions.max_turns ? t.spec : null; - }); + await this.codex + .runWorker(taskId, first.tier, () => { + const t = task(); + return t?.status === "active" && turns++ < executions.max_turns ? t.spec : null; + }) + .catch((error: unknown) => { + log.warn("worker turn failed", { taskId, error: String(error) }); + if (task()?.status === "active") this.ledger.interrupt(taskId); + }); if (task()?.status === "active") this.ledger.transition(taskId, { - type: "wait", - waitingOn: "timer", - wakeAt: new Date(Date.now() + executions.backoff_ms).toISOString(), + type: "finish", + outcome: "failed", + report: `The worker used all ${executions.max_turns} turns without finishing.`, }); const after = task(); log.info("worker finished", { diff --git a/src/soul/ear.md b/src/soul/ear.md index a0f2b0e..cb63710 100644 --- a/src/soul/ear.md +++ b/src/soul/ear.md @@ -1,6 +1,6 @@ Decide whether each conversation needs the agent's attention, given its role and context. Default to hold; wake when something needs its response or action. Do not overlook unanswered requests directed at it. An ask aimed at a named person, or at the room, is not for the agent even if it could answer. -Submit one verdict per conversation with a brief reason, through the tool only. Do not write messages; when the verdicts are in, end the turn. +Decide from the transcript alone. Submit one verdict per conversation with a brief reason, through the tool only. Do not write messages; when the verdicts are in, end the turn. ## The agent @@ -8,8 +8,12 @@ The agent is <@{{botUserId}}>. ## Its role +The agent's own instructions, addressed to it: + {{persona}} ## Its context +The agent's memory, in its own voice: + {{memory}} diff --git a/src/soul/resident.md b/src/soul/resident.md index 8824904..c471418 100644 --- a/src/soul/resident.md +++ b/src/soul/resident.md @@ -2,7 +2,7 @@ Be a useful coworker, not another source of noise. ## Speak With Purpose -Answer direct questions and assigned work. Otherwise, join only when you have something useful to add. You are not a human. you can't review code. most requests are not targetted at you. When told to stop, stop silently. +Answer direct questions and assigned work. You are not a human. you can't review code. most requests are not targetted at you. When told to stop, stop silently. The room sees only tool calls: `reply` for words, `react` for reactions. Do not write messages; act through tools, and when there is nothing to do, end the turn. Reply in-thread and address people directly. diff --git a/src/tools.ts b/src/tools.ts index 29e74c0..811afff 100644 --- a/src/tools.ts +++ b/src/tools.ts @@ -178,9 +178,8 @@ export const earTools: Tools = (server) => { channel: z.string(), thread_ts: z.string(), }, - async ({ decision, why, channel, thread_ts }) => { - if (decision === "wake" && !ledger().wakeFor(channel, thread_ts, why)) - throw new Error(`no conversation at ${channel} thread=${thread_ts} in this batch`); + async ({ decision, channel, thread_ts }) => { + if (decision === "wake") ledger().wake(channel, thread_ts); return "noted"; }, ); diff --git a/src/voice.ts b/src/voice.ts index 2cacb04..96726dc 100644 --- a/src/voice.ts +++ b/src/voice.ts @@ -37,14 +37,10 @@ export class Voice { void this.web.agents.sessions.setStatus({ channel_id: convo.channel, thread_ts: convo.threadTs, - status: this.answered(convo) ? "active" : "closed", + status: this.replied.has(key(convo)) ? "active" : "closed", }); } - answered(convo: Thread): boolean { - return this.replied.has(key(convo)); - } - async reply(channel: string, thread_ts: string | null, text: string): Promise { if (thread_ts) { const convo = { channel, threadTs: thread_ts }; @@ -74,8 +70,10 @@ export class Voice { if (!posted) { throw new Error("that didn't send — the surface rejected it. try again, or let it go"); } - this.ledger.unmute(channel, thread_ts ?? posted); - this.replied.add(key({ channel, threadTs: thread_ts ?? posted })); + if (thread_ts) { + this.ledger.unmute(channel, thread_ts); + this.replied.add(key({ channel, threadTs: thread_ts })); + } return "posted"; }