From 422252781f8c8baf9e19f77867a43ccc804f5972 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 17:41:04 -0400 Subject: [PATCH 1/4] Cut the alert-venue investigate-everything clause; log tool-call arguments Co-Authored-By: Claude Fable 5.1 --- deploy/policy.yaml | 9 ++------- src/codex.ts | 3 ++- 2 files changed, 4 insertions(+), 8 deletions(-) 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/src/codex.ts b/src/codex.ts index 1ae9cd9..5293f4c 100644 --- a/src/codex.ts +++ b/src/codex.ts @@ -72,7 +72,8 @@ 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)}` }); else if (item.type === "agent_message") log.info(label, { line: `● ${item.text}` }); } else if (event.type === "turn.failed") { maybeRotateGateway({ reason: event.error.message }); From 3d298634209a5ffb7eec668e866cc55178a59053 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 17:57:38 -0400 Subject: [PATCH 2/4] Review round: resident sees only wanted conversations; ledger rows survive until unchanged; workers fail out instead of parking forever - respond() filters on WANTED (direct or woken); held threads never reach the resident - conversations.judged replaced by last: a row is deleted only when its last heard ts is unchanged, so messages landing mid-turn are not lost - runWorker maps failure to a boolean; interrupt only when still active; max_turns finishes the task as failed - Codex no longer owns ledger logic; judge() replaces shouldAgentRespond - voice.close runs in finally; channel-level replies are not recorded as thread answers - replies paginate; ear prompt gets the legend; wake reason no longer rendered to the resident - roster loads before the scheduler starts - ear soul: decide from the transcript; persona and memory framed as the agent's, not the ear's Co-Authored-By: Claude Fable 5.1 --- drizzle/0004_conversations_last.sql | 2 + drizzle/meta/0004_snapshot.json | 274 ++++++++++++++++++++++++++++ drizzle/meta/_journal.json | 7 + src/codex.ts | 18 +- src/ledger-service.ts | 42 ++--- src/ledger/schema.ts | 2 +- src/main.ts | 2 +- src/prompt-renderer.ts | 25 ++- src/scheduler.ts | 61 ++++--- src/soul/ear.md | 6 +- src/voice.ts | 12 +- 11 files changed, 370 insertions(+), 81 deletions(-) create mode 100644 drizzle/0004_conversations_last.sql create mode 100644 drizzle/meta/0004_snapshot.json diff --git a/drizzle/0004_conversations_last.sql b/drizzle/0004_conversations_last.sql new file mode 100644 index 0000000..12d86ef --- /dev/null +++ b/drizzle/0004_conversations_last.sql @@ -0,0 +1,2 @@ +ALTER TABLE `conversations` ADD `last` text NOT NULL;--> statement-breakpoint +ALTER TABLE `conversations` DROP COLUMN `judged`; \ 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..b75fb09 --- /dev/null +++ b/drizzle/meta/0004_snapshot.json @@ -0,0 +1,274 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "ac0fe91c-6d35-46ca-bdf1-00f7e1f66e05", + "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 + }, + "wake_why": { + "name": "wake_why", + "type": "text", + "primaryKey": false, + "notNull": false, + "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..d3a8bbe 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": 1788904595844, + "tag": "0004_conversations_last", + "breakpoints": true } ] } diff --git a/src/codex.ts b/src/codex.ts index 5293f4c..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(); } @@ -73,7 +65,9 @@ export class Codex { 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} ${JSON.stringify(item.arguments)}` }); + 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..588717f 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, not, 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,11 @@ export function openDb(path: string): Db { return db; } +export const WANTED = sql`(${conversations.direct} OR ${conversations.wakeWhy} IS NOT NULL)`; + +const unchanged = (convo: Conversation) => + and(thread(conversations, convo.channel, convo.threadTs), eq(conversations.last, convo.last)); + export function thread( table: typeof conversations | typeof mutedThreads, channel: string, @@ -136,11 +141,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.db.delete(conversations).where(unchanged(convo)).run(); for (const task of settled) this.db .update(tasks) @@ -204,17 +205,22 @@ ${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, wakeWhy: null }) .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(); } + held(convos: Conversation[]): void { + for (const convo of convos) + this.db + .delete(conversations) + .where(and(unchanged(convo), not(WANTED))) + .run(); + } + wakeFor(channel: string, threadTs: string, why: string): boolean { const hit = this.db .update(conversations) @@ -226,19 +232,7 @@ ${text}`, } wantsResponse(): boolean { - const wanted = this.db.query.conversations - .findFirst({ where: or(eq(conversations.direct, true), isNotNull(conversations.wakeWhy)) }) - .sync(); - return wanted !== 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(); + 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..226cdf6 100644 --- a/src/ledger/schema.ts +++ b/src/ledger/schema.ts @@ -57,8 +57,8 @@ 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"), }, (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/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..da3e6f6 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -4,11 +4,11 @@ import { SocketModeClient } from "@slack/socket-mode"; import type { MessageEvent } from "@slack/types"; import type { MessageElement } from "@slack/web-api/dist/types/response/ConversationsRepliesResponse"; import { WebClient } from "@slack/web-api"; -import { and, asc, desc, eq, gt, isNull, or } from "drizzle-orm"; +import { and, asc, desc, eq, gt, isNull, not, 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,20 @@ 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) }) - .sync(); - if (unjudged.length === 0) { - if (this.ledger.wantsResponse()) this.respondSoon(); - return; + const unjudged = this.db.query.conversations.findMany({ where: not(WANTED) }).sync(); + 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,16 +140,27 @@ 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; - }); - if (task()?.status === "active") - this.ledger.transition(taskId, { - type: "wait", - waitingOn: "timer", - wakeAt: new Date(Date.now() + executions.backoff_ms).toISOString(), - }); + const failed = await this.codex + .runWorker(taskId, first.tier, () => { + const t = task(); + return t?.status === "active" && turns++ < executions.max_turns ? t.spec : null; + }) + .then( + () => false, + (error: unknown) => { + log.warn("worker turn failed", { taskId, error: String(error) }); + return true; + }, + ); + if (task()?.status === "active") { + if (failed) this.ledger.interrupt(taskId); + else + this.ledger.transition(taskId, { + type: "finish", + outcome: "failed", + report: `The worker used all ${executions.max_turns} turns without finishing.`, + }); + } const after = task(); log.info("worker finished", { taskId, 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/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"; } From 2df26b11f0562117e4abad3a1a00ff23973278d6 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 18:01:49 -0400 Subject: [PATCH 3/4] Second review round: woken flag replaces wake_why; since advances when a turn misses; one catch in runWorker; dead backoff_ms removed Co-Authored-By: Claude Fable 5.1 --- drizzle/0004_conversations_last.sql | 2 -- drizzle/0004_conversations_last_woken.sql | 4 +++ drizzle/meta/0004_snapshot.json | 10 +++--- drizzle/meta/_journal.json | 4 +-- src/ledger-service.ts | 44 +++++++++++++---------- src/ledger/schema.ts | 2 +- src/policy.ts | 1 - src/scheduler.ts | 34 ++++++++---------- src/tools.ts | 5 ++- 9 files changed, 55 insertions(+), 51 deletions(-) delete mode 100644 drizzle/0004_conversations_last.sql create mode 100644 drizzle/0004_conversations_last_woken.sql diff --git a/drizzle/0004_conversations_last.sql b/drizzle/0004_conversations_last.sql deleted file mode 100644 index 12d86ef..0000000 --- a/drizzle/0004_conversations_last.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE `conversations` ADD `last` text NOT NULL;--> statement-breakpoint -ALTER TABLE `conversations` DROP COLUMN `judged`; \ No newline at end of file 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 index b75fb09..423ed10 100644 --- a/drizzle/meta/0004_snapshot.json +++ b/drizzle/meta/0004_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "ac0fe91c-6d35-46ca-bdf1-00f7e1f66e05", + "id": "0be1bb01-172d-4215-9da5-e36b2411ccd1", "prevId": "2698e3b3-5c25-42db-8694-53e2d539cddc", "tables": { "conversations": { @@ -42,11 +42,11 @@ "notNull": true, "autoincrement": false }, - "wake_why": { - "name": "wake_why", - "type": "text", + "woken": { + "name": "woken", + "type": "integer", "primaryKey": false, - "notNull": false, + "notNull": true, "autoincrement": false } }, diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index d3a8bbe..dbc19ae 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -33,8 +33,8 @@ { "idx": 4, "version": "6", - "when": 1788904595844, - "tag": "0004_conversations_last", + "when": 1788904893450, + "tag": "0004_conversations_last_woken", "breakpoints": true } ] diff --git a/src/ledger-service.ts b/src/ledger-service.ts index 588717f..8027e73 100644 --- a/src/ledger-service.ts +++ b/src/ledger-service.ts @@ -1,4 +1,4 @@ -import { and, asc, count, eq, like, lte, min, not, 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,10 +18,7 @@ export function openDb(path: string): Db { return db; } -export const WANTED = sql`(${conversations.direct} OR ${conversations.wakeWhy} IS NOT NULL)`; - -const unchanged = (convo: Conversation) => - and(thread(conversations, convo.channel, convo.threadTs), eq(conversations.last, convo.last)); +export const WANTED = or(eq(conversations.direct, true), eq(conversations.woken, true)); export function thread( table: typeof conversations | typeof mutedThreads, @@ -141,7 +138,7 @@ ${text}`, } rendered(convos: Conversation[], settled: Task[]): void { - for (const convo of convos) this.db.delete(conversations).where(unchanged(convo)).run(); + for (const convo of convos) this.settle(convo); for (const task of settled) this.db .update(tasks) @@ -205,7 +202,7 @@ ${text}`, heard(channel: string, threadTs: string, ts: string, direct: boolean): void { this.db .insert(conversations) - .values({ channel, threadTs, since: ts, last: ts, direct, wakeWhy: null }) + .values({ channel, threadTs, since: ts, last: ts, direct, woken: false }) .onConflictDoUpdate({ target: [conversations.channel, conversations.threadTs], set: { last: ts, direct: sql`${conversations.direct} OR ${direct}` }, @@ -214,21 +211,32 @@ ${text}`, } held(convos: Conversation[]): void { - for (const convo of convos) - this.db - .delete(conversations) - .where(and(unchanged(convo), not(WANTED))) - .run(); + for (const convo of convos) if (!this.wanted(convo)) this.settle(convo); } - wakeFor(channel: string, threadTs: string, why: string): boolean { - const hit = this.db + private wanted(convo: Conversation): boolean { + const row = this.db.query.conversations + .findFirst({ where: and(thread(conversations, convo.channel, convo.threadTs), WANTED) }) + .sync(); + return row !== undefined; + } + + 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({ wakeWhy: why }) + .set({ woken: true }) .where(thread(conversations, channel, threadTs)) - .returning({ channel: conversations.channel }) - .get(); - return hit !== undefined; + .run(); } wantsResponse(): boolean { diff --git a/src/ledger/schema.ts b/src/ledger/schema.ts index 226cdf6..3b1ba8c 100644 --- a/src/ledger/schema.ts +++ b/src/ledger/schema.ts @@ -59,7 +59,7 @@ export const conversations = sqliteTable( since: text("since").notNull(), last: text("last").notNull(), direct: integer("direct", { mode: "boolean" }).notNull(), - wakeWhy: text("wake_why"), + woken: integer("woken", { mode: "boolean" }).notNull(), }, (t) => [primaryKey({ columns: [t.channel, t.threadTs] })], ); 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/scheduler.ts b/src/scheduler.ts index da3e6f6..7aeec85 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -4,7 +4,7 @@ import { SocketModeClient } from "@slack/socket-mode"; import type { MessageEvent } from "@slack/types"; import type { MessageElement } from "@slack/web-api/dist/types/response/ConversationsRepliesResponse"; import { WebClient } from "@slack/web-api"; -import { and, asc, desc, eq, gt, isNull, not, or } from "drizzle-orm"; +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"; @@ -126,7 +126,9 @@ export class Scheduler { } private async listenToNoise(): Promise { - const unjudged = this.db.query.conversations.findMany({ where: not(WANTED) }).sync(); + const unjudged = this.db.query.conversations + .findMany({ where: and(eq(conversations.direct, false), eq(conversations.woken, false)) }) + .sync(); if (unjudged.length > 0) { await this.codex.judge(await this.prompts.noise(unjudged)); this.ledger.held(unjudged); @@ -140,27 +142,21 @@ export class Scheduler { const first = task(); if (first?.status !== "active") return; let turns = 0; - const failed = await this.codex + await this.codex .runWorker(taskId, first.tier, () => { const t = task(); return t?.status === "active" && turns++ < executions.max_turns ? t.spec : null; }) - .then( - () => false, - (error: unknown) => { - log.warn("worker turn failed", { taskId, error: String(error) }); - return true; - }, - ); - if (task()?.status === "active") { - if (failed) this.ledger.interrupt(taskId); - else - this.ledger.transition(taskId, { - type: "finish", - outcome: "failed", - report: `The worker used all ${executions.max_turns} turns without finishing.`, - }); - } + .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: "finish", + outcome: "failed", + report: `The worker used all ${executions.max_turns} turns without finishing.`, + }); const after = task(); log.info("worker finished", { taskId, 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"; }, ); From d1a4b01be73c2ba8f504b867d8dafc4fccd59f1f Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 18:08:27 -0400 Subject: [PATCH 4/4] Resident soul: drop the join-when-useful loophole Co-Authored-By: Claude Fable 5.1 --- src/soul/resident.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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.