From 1e74dfa5d36f26fb5effec3cd71975157d58a604 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 18:31:47 -0400 Subject: [PATCH] Simplification pass: respond is a Debounced; held folds its predicate into caughtUp; once() takes the tier it needs; Voice.begin opens; tools resolve once; Roster loads itself; soul and reply try/catch removed Co-Authored-By: Claude Fable 5.1 --- src/codex.ts | 15 +++++++-------- src/ledger-service.ts | 19 ++++++------------- src/main.ts | 2 -- src/policy.ts | 1 + src/prompt-renderer.ts | 8 ++++---- src/roster.ts | 10 +++++++--- src/scheduler.ts | 38 +++++++++++--------------------------- src/soul.ts | 35 ++++++++++++++--------------------- src/tools.ts | 36 ++++++++++++++++++------------------ src/voice.ts | 34 ++++++++-------------------------- 10 files changed, 76 insertions(+), 122 deletions(-) diff --git a/src/codex.ts b/src/codex.ts index 34ce633..95effdd 100644 --- a/src/codex.ts +++ b/src/codex.ts @@ -2,12 +2,11 @@ import { codexThread, maybeRotateGateway, type Tools } from "@bevyl-ai/agent-too import { inject, singleton } from "tsyringe"; import type { Task } from "./ledger/schema"; import { log } from "./log"; -import { POLICY, type Policy } from "./policy"; +import { POLICY, type Policy, type Tier } from "./policy"; import { Soul } from "./soul"; import { earTools, residentTools, workerTools } from "./tools"; import { Workspaces, type Role } from "./workspaces"; -type Tier = Policy["models"]["low"]; type Thread = Awaited>["thread"]; @singleton() @@ -19,11 +18,11 @@ export class Codex { ) {} respond(prompt: string): Promise { - return this.once("resident", residentTools, {}, this.policy.turns.timeout_ms, prompt); + return this.once("resident", residentTools, prompt); } - judge(prompt: string): Promise { - return this.once("ear", earTools, this.policy.models.low, this.policy.turns.timeout_ms, prompt); + triage(prompt: string): Promise { + return this.once("ear", earTools, prompt, this.policy.models.low); } async runWorker(taskId: string, tier: Task["tier"], next: () => string | null): Promise { @@ -37,16 +36,16 @@ export class Codex { } } - private async once(role: Role, tools: Tools, tier: Tier, timeoutMs: number, prompt: string) { + private async once(role: Role, tools: Tools, prompt: string, tier?: Tier) { const { thread, close } = await this.thread(role, tools, tier); try { - await this.turn(thread, role, prompt, timeoutMs); + await this.turn(thread, role, prompt, this.policy.turns.timeout_ms); } finally { close(); } } - private thread(role: Role, tools: Tools, tier: Tier) { + private thread(role: Role, tools: Tools, tier: Tier = {}) { this.soul.refresh(); return codexThread({ tools, diff --git a/src/ledger-service.ts b/src/ledger-service.ts index 8027e73..df492cd 100644 --- a/src/ledger-service.ts +++ b/src/ledger-service.ts @@ -1,4 +1,4 @@ -import { and, asc, count, eq, like, lte, min, or, sql } from "drizzle-orm"; +import { and, asc, count, eq, like, lte, min, not, sql, type 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,7 +18,7 @@ export function openDb(path: string): Db { return db; } -export const WANTED = or(eq(conversations.direct, true), eq(conversations.woken, true)); +export const WANTED = sql`(${conversations.direct} OR ${conversations.woken})`; export function thread( table: typeof conversations | typeof mutedThreads, @@ -138,7 +138,7 @@ ${text}`, } rendered(convos: Conversation[], settled: Task[]): void { - for (const convo of convos) this.settle(convo); + for (const convo of convos) this.caughtUp(convo); for (const task of settled) this.db .update(tasks) @@ -211,18 +211,11 @@ ${text}`, } held(convos: Conversation[]): void { - for (const convo of convos) if (!this.wanted(convo)) this.settle(convo); + for (const convo of convos) this.caughtUp(convo, not(WANTED)); } - 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); + private caughtUp(convo: Conversation, only?: SQL): void { + const where = and(thread(conversations, convo.channel, convo.threadTs), only); const gone = this.db .delete(conversations) .where(and(where, eq(conversations.last, convo.last))) diff --git a/src/main.ts b/src/main.ts index acafd9d..1611921 100644 --- a/src/main.ts +++ b/src/main.ts @@ -4,12 +4,10 @@ import { container } from "tsyringe"; import { watchFile } from "node:fs"; import { SocketModeClient } from "@slack/socket-mode"; import type { MessageEvent } from "@slack/types"; -import { Roster } from "./roster"; import { Scheduler } from "./scheduler"; import { log } from "./log"; import { POLICY, POLICY_PATH, loadPolicy } from "./policy"; -await container.resolve(Roster).load(); const scheduler = container.resolve(Scheduler); log.info("service started"); diff --git a/src/policy.ts b/src/policy.ts index 086c329..a977b9c 100644 --- a/src/policy.ts +++ b/src/policy.ts @@ -27,6 +27,7 @@ const PolicySchema = z.object({ }); export type Policy = z.infer; +export type Tier = z.infer; export const POLICY: InjectionToken = Symbol("policy"); export const POLICY_PATH: InjectionToken = Symbol("policyPath"); diff --git a/src/prompt-renderer.ts b/src/prompt-renderer.ts index 9583644..8e16ea9 100644 --- a/src/prompt-renderer.ts +++ b/src/prompt-renderer.ts @@ -38,7 +38,7 @@ export class PromptRenderer { return parts.join("\n\n"); } - async noise(convos: Conversation[]): Promise { + async overheard(convos: Conversation[]): Promise { return LEGEND + (await this.batch(convos)); } @@ -95,11 +95,11 @@ export class PromptRenderer { const saved = await Promise.all((line.files ?? []).map((file) => this.attachments.save(file))); const files = saved.length > 0 ? ` [attached: ${saved.join(", ")}]` : ""; const text = (line.text ?? "").slice(0, limit); - return ` [${channel} ${line.ts}] ${this.speaker(line.user ?? line.bot_id)}: ${text}${files}`; + return ` [${channel} ${line.ts}] ${await this.speaker(line.user ?? line.bot_id)}: ${text}${files}`; } - private speaker(user: string | undefined): string { - const name = user ? this.roster.nameOf(user) : null; + private async speaker(user: string | undefined): Promise { + const name = user ? await this.roster.nameOf(user) : null; return `<@${user ?? "?"}>${name ? ` (${name})` : ""}`; } } diff --git a/src/roster.ts b/src/roster.ts index 74721c3..5f7a8d7 100644 --- a/src/roster.ts +++ b/src/roster.ts @@ -4,14 +4,18 @@ import { WebClient, type UsersListResponse } from "@slack/web-api"; @singleton() export class Roster { private readonly names = new Map(); + private readonly loaded: Promise; - constructor(private readonly web: WebClient) {} + constructor(private readonly web: WebClient) { + this.loaded = this.load(); + } - nameOf(principalId: string): string | null { + async nameOf(principalId: string): Promise { + await this.loaded; return this.names.get(principalId) ?? null; } - async load(): Promise { + private async load(): Promise { for await (const page of this.web.paginate("users.list", { limit: 200 })) { for (const member of (page as UsersListResponse).members ?? []) { const name = [member.profile?.display_name, member.profile?.real_name, member.name].find( diff --git a/src/scheduler.ts b/src/scheduler.ts index 7aeec85..49ed19f 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -50,9 +50,8 @@ const HEARD_SUBTYPES = new Set([ ]) @singleton() export class Scheduler { - private responding: Promise | null = null; - private respondAgain = false; - private readonly noise = new Debounced(() => this.listenToNoise()); + private readonly overheard = new Debounced(() => this.triage()); + private readonly replies = new Debounced(() => this.respond()); constructor( @inject(DB) private readonly db: Db, @@ -65,7 +64,7 @@ export class Scheduler { ) { this.tick(); this.beat(); - this.noise.schedule(0); + this.overheard.schedule(0); } heard(event: MessageEvent): void { @@ -78,21 +77,7 @@ export class Scheduler { const threadTs = message.thread_ts ?? event.ts; if (!direct && this.ledger.muted(event.channel, threadTs)) return; this.ledger.heard(event.channel, threadTs, event.ts, direct); - this.noise.schedule(direct ? 0 : this.policy.ear_debounce_ms); - } - - private respondSoon(): void { - if (this.responding) { - this.respondAgain = true; - return; - } - this.responding = this.respond().finally(() => { - this.responding = null; - if (this.respondAgain) { - this.respondAgain = false; - this.respondSoon(); - } - }); + this.overheard.schedule(direct ? 0 : this.policy.ear_debounce_ms); } private async respond(): Promise { @@ -115,25 +100,24 @@ export class Scheduler { if (convos.length === 0 && settled.length === 0) return; const prompt = await this.prompts.response(convos, settled); const direct = convos.filter((convo) => convo.direct); - for (const convo of direct) this.voice.open(convo); this.ledger.rendered(convos, settled); - this.voice.begin(); + this.voice.begin(direct); await this.codex.respond(prompt).finally(() => { this.voice.close(direct); }); this.tick(); - if (this.ledger.wantsResponse()) this.respondSoon(); + if (this.ledger.wantsResponse()) this.replies.schedule(0); } - private async listenToNoise(): Promise { + private async triage(): Promise { 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)); + await this.codex.triage(await this.prompts.overheard(unjudged)); this.ledger.held(unjudged); } - if (this.ledger.wantsResponse()) this.respondSoon(); + if (this.ledger.wantsResponse()) this.replies.schedule(0); } private async runWorker(taskId: string): Promise { @@ -164,7 +148,7 @@ export class Scheduler { outcome: after?.outcome, turns, }); - if (after?.status === "done" || after?.waitingOn === "human") this.respondSoon(); + if (after?.status === "done" || after?.waitingOn === "human") this.replies.schedule(0); this.tick(); } @@ -176,7 +160,7 @@ export class Scheduler { } private tick(): void { - if (this.ledger.wakeDueTasks()) this.respondSoon(); + if (this.ledger.wakeDueTasks()) this.replies.schedule(0); for (const taskId of this.ledger.dispatchRunnable(this.policy.executions.max_concurrent)) void this.runWorker(taskId); } diff --git a/src/soul.ts b/src/soul.ts index b17784f..a058dfa 100644 --- a/src/soul.ts +++ b/src/soul.ts @@ -1,7 +1,6 @@ import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { inject, singleton } from "tsyringe"; -import { log } from "./log"; import { POLICY, type Policy } from "./policy"; import { BOT_USER_ID } from "./tokens"; import ear from "./soul/ear.md" with { type: "text" }; @@ -26,25 +25,19 @@ export class Soul { ) {} refresh(): void { - try { - const memoryPath = join(this.workspaces.resident, "MEMORY.md"); - const holes = { - botUserId: this.botUserId, - persona: orElse(this.policy.persona, "(none)"), - memory: orElse(existsSync(memoryPath) ? readFileSync(memoryPath, "utf8") : "", "(empty)"), - venues: orElse( - Object.entries(this.policy.venue_instructions) - .map(([venueId, instruction]) => `- <#${venueId}>: ${instruction}`) - .join("\n"), - "(none)", - ), - }; - writeFileSync(join(this.workspaces.resident, "AGENTS.md"), fill(resident, holes)); - writeFileSync(join(this.workspaces.ear, "AGENTS.md"), fill(ear, holes)); - } catch (error) { - log.warn("could not write soul (AGENTS.md) — using codex default voice", { - error: String(error), - }); - } + const memoryPath = join(this.workspaces.resident, "MEMORY.md"); + const holes = { + botUserId: this.botUserId, + persona: orElse(this.policy.persona, "(none)"), + memory: orElse(existsSync(memoryPath) ? readFileSync(memoryPath, "utf8") : "", "(empty)"), + venues: orElse( + Object.entries(this.policy.venue_instructions) + .map(([venueId, instruction]) => `- <#${venueId}>: ${instruction}`) + .join("\n"), + "(none)", + ), + }; + writeFileSync(join(this.workspaces.resident, "AGENTS.md"), fill(resident, holes)); + writeFileSync(join(this.workspaces.ear, "AGENTS.md"), fill(ear, holes)); } } diff --git a/src/tools.ts b/src/tools.ts index 811afff..86451a5 100644 --- a/src/tools.ts +++ b/src/tools.ts @@ -17,16 +17,16 @@ import { POLICY } from "./policy"; import { requireEnv } from "./tokens"; import { Voice } from "./voice"; -const ledger = () => container.resolve(LedgerService); - const shared: Tools = (server) => { + const ledger = container.resolve(LedgerService); + const { query } = container.resolve(DB); tool( server, "task_create", "Delegate to a worker; the spec is its whole briefing.", TaskCreate.shape, async (args) => { - const task = ledger().createTask(args); + const task = ledger.createTask(args); return { id: task.id, status: task.status }; }, ); @@ -36,7 +36,7 @@ const shared: Tools = (server) => { "Append to a task's spec.", { taskId: z.string(), text: z.string() }, async ({ taskId, text: more }) => { - const task = ledger().appendGuidance(taskId, more); + const task = ledger.appendGuidance(taskId, more); return { id: task.id, status: task.status }; }, ); @@ -46,8 +46,8 @@ const shared: Tools = (server) => { "Cancel a task.", { taskId: z.string(), report: z.string().optional() }, async ({ taskId, report }) => { - const task = ledger().requireTask(taskId); - ledger().transition(taskId, { + const task = ledger.requireTask(taskId); + ledger.transition(taskId, { type: "finish", outcome: "cancelled", report: report ?? `Cancelled "${task.title}".`, @@ -56,7 +56,6 @@ const shared: Tools = (server) => { }, ); tool(server, "task_query", "Your open and recently finished tasks.", {}, async () => { - const { query } = container.resolve(DB); return { open: query.tasks .findMany({ where: ne(tasks.status, "done"), orderBy: asc(tasks.openedAt) }) @@ -72,7 +71,7 @@ const shared: Tools = (server) => { "Mute a thread until mentioned there again.", { why: z.string(), channel: z.string(), thread_ts: z.string() }, async ({ why, channel, thread_ts }) => { - ledger().mute(channel, thread_ts, why); + ledger.mute(channel, thread_ts, why); return "muted; a mention brings you back"; }, ); @@ -93,13 +92,13 @@ const shared: Tools = (server) => { export const residentTools: Tools = (server) => { shared(server); + const voice = container.resolve(Voice); tool( server, "reply", "Post a message; omit thread_ts for channel level.", { text: z.string(), channel: z.string(), thread_ts: z.string().optional() }, - ({ text: body, channel, thread_ts }) => - container.resolve(Voice).reply(channel, thread_ts ?? null, body), + ({ text: body, channel, thread_ts }) => voice.reply(channel, thread_ts ?? null, body), ); tool( server, @@ -111,7 +110,7 @@ export const residentTools: Tools = (server) => { ts: z.string(), }, async ({ emoji, channel, ts }) => { - await container.resolve(Voice).react(channel, ts, emoji); + await voice.react(channel, ts, emoji); return `reacted :${emoji}:`; }, ); @@ -128,13 +127,15 @@ export const workerTools = (taskId: string): Tools => (server) => { shared(server); + const ledger = container.resolve(LedgerService); + const { park_after_ms } = container.resolve(POLICY).tasks; tool( server, "task_complete", "Finish this task with a report.", { outcome: z.enum(["done", "failed"]), report: z.string() }, async ({ outcome, report }) => { - ledger().transition(taskId, { type: "finish", outcome, report }); + ledger.transition(taskId, { type: "finish", outcome, report }); return `task ${taskId} ${outcome}`; }, ); @@ -144,13 +145,11 @@ export const workerTools = "Ask a human a question; pauses the task.", { question: z.string() }, async ({ question }) => { - ledger().transition(taskId, { + ledger.transition(taskId, { type: "wait", waitingOn: "human", why: question, - wakeAt: new Date( - Date.now() + container.resolve(POLICY).tasks.park_after_ms, - ).toISOString(), + wakeAt: new Date(Date.now() + park_after_ms).toISOString(), }); return `task ${taskId} waiting on a human`; }, @@ -161,13 +160,14 @@ export const workerTools = "Pause this task until an ISO-8601 time.", { wakeAt: FutureTime }, async ({ wakeAt }) => { - ledger().transition(taskId, { type: "wait", waitingOn: "timer", wakeAt }); + ledger.transition(taskId, { type: "wait", waitingOn: "timer", wakeAt }); return `paused until ${wakeAt}; the task picks up again then`; }, ); }; export const earTools: Tools = (server) => { + const ledger = container.resolve(LedgerService); tool( server, "verdict", @@ -179,7 +179,7 @@ export const earTools: Tools = (server) => { thread_ts: z.string(), }, async ({ decision, channel, thread_ts }) => { - if (decision === "wake") ledger().wake(channel, thread_ts); + if (decision === "wake") ledger.wake(channel, thread_ts); return "noted"; }, ); diff --git a/src/voice.ts b/src/voice.ts index 96726dc..55a00c3 100644 --- a/src/voice.ts +++ b/src/voice.ts @@ -2,7 +2,6 @@ import { WebAPIPlatformError, WebClient } from "@slack/web-api"; import { inject, singleton } from "tsyringe"; import { conversations, type Conversation } from "./ledger/schema"; import { DB, LedgerService, thread, type Db } from "./ledger-service"; -import { log } from "./log"; type Thread = Pick; @@ -19,17 +18,15 @@ export class Voice { private readonly ledger: LedgerService, ) {} - begin(): void { + begin(convos: Thread[]): void { this.replied = new Set(); this.bounced = new Set(); - } - - open(convo: Thread): void { - void this.web.agents.sessions.setStatus({ - channel_id: convo.channel, - thread_ts: convo.threadTs, - status: "processing", - }); + for (const convo of convos) + void this.web.agents.sessions.setStatus({ + channel_id: convo.channel, + thread_ts: convo.threadTs, + status: "processing", + }); } close(convos: Thread[]): void { @@ -54,22 +51,7 @@ export class Voice { ); } } - let posted: string | undefined; - try { - posted = ( - await this.web.chat.postMessage({ channel, text, ...(thread_ts ? { thread_ts } : {}) }) - ).ts; - } catch (error) { - log.error("OUTBOUND DELIVERY FAILED — operator must convey this manually", { - channel, - thread_ts, - text, - error: String(error), - }); - } - if (!posted) { - throw new Error("that didn't send — the surface rejected it. try again, or let it go"); - } + await this.web.chat.postMessage({ channel, text, ...(thread_ts ? { thread_ts } : {}) }); if (thread_ts) { this.ledger.unmute(channel, thread_ts); this.replied.add(key({ channel, threadTs: thread_ts }));