From 626f1075abc0104bee2087b965a1bbd69d5974f7 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Tue, 8 Sep 2026 18:40:42 -0400 Subject: [PATCH] Bug hunt: tick() is the one place the ledger is read for wanted work; failed turns re-arm on the beat; since only moves forward; triage batched; handlers before boot - tick(): wakeDueTasks, then respond if any wanted conversation or unseen settled task, then dispatch; respond/triage/runWorker all end in tick() - a failed respond no longer strands wanted rows: the beat re-arms within 60s instead of a hot loop - Debounced logs its run's rejection under a label instead of leaking an unhandled rejection - caughtUp's watermark update is guarded by since <= last so a recreated row is never rewound - triage takes at most BATCH oldest unjudged rows - signal and unhandledRejection handlers register before the first resolve; Roster loads before the scheduler - park_after_ms read at call time; persona defaults to empty so a reload can clear it Co-Authored-By: Claude Fable 5.1 --- src/debounce.ts | 25 +++++++++++++++++-------- src/ledger-service.ts | 16 ++++++++++++---- src/main.ts | 20 +++++++++++--------- src/policy.ts | 2 +- src/prompt-renderer.ts | 6 +++--- src/roster.ts | 10 +++------- src/scheduler.ts | 35 +++++++++++++++++------------------ src/tools.ts | 4 ++-- 8 files changed, 66 insertions(+), 52 deletions(-) diff --git a/src/debounce.ts b/src/debounce.ts index 6c33361..a3ef5b4 100644 --- a/src/debounce.ts +++ b/src/debounce.ts @@ -1,9 +1,14 @@ +import { log } from "./log"; + export class Debounced { private timer: ReturnType | null = null; private running = false; private rerun = false; - constructor(private readonly run: () => Promise) {} + constructor( + private readonly label: string, + private readonly run: () => Promise, + ) {} schedule(delayMs: number): void { if (delayMs <= 0) { @@ -25,12 +30,16 @@ export class Debounced { return; } this.running = true; - void this.run().finally(() => { - this.running = false; - if (this.rerun) { - this.rerun = false; - this.start(); - } - }); + void this.run() + .catch((error: unknown) => { + log.error(this.label, { error: String(error) }); + }) + .finally(() => { + this.running = false; + if (this.rerun) { + this.rerun = false; + this.start(); + } + }); } } diff --git a/src/ledger-service.ts b/src/ledger-service.ts index df492cd..7ff40ba 100644 --- a/src/ledger-service.ts +++ b/src/ledger-service.ts @@ -19,6 +19,7 @@ export function openDb(path: string): Db { } export const WANTED = sql`(${conversations.direct} OR ${conversations.woken})`; +export const UNSEEN = sql`((${tasks.status} = 'done' OR ${tasks.waitingOn} = 'human') AND (${tasks.seenAt} IS NULL OR ${tasks.updatedAt} > ${tasks.seenAt}))`; export function thread( table: typeof conversations | typeof mutedThreads, @@ -147,7 +148,7 @@ ${text}`, .run(); } - wakeDueTasks(): boolean { + wakeDueTasks(): void { const due = this.db .select({ id: tasks.id, waitingOn: tasks.waitingOn }) .from(tasks) @@ -162,7 +163,6 @@ ${text}`, report: "No answer arrived before the deadline; the task was closed without acting.", }); } - return due.some((task) => task.waitingOn === "human"); } msUntilNextWake(maxMs: number): number { @@ -221,7 +221,12 @@ ${text}`, .where(and(where, eq(conversations.last, convo.last))) .returning() .get(); - if (!gone) this.db.update(conversations).set({ since: convo.last }).where(where).run(); + if (!gone) + this.db + .update(conversations) + .set({ since: convo.last }) + .where(and(where, lte(conversations.since, convo.last))) + .run(); } wake(channel: string, threadTs: string): void { @@ -233,7 +238,10 @@ ${text}`, } wantsResponse(): boolean { - return this.db.query.conversations.findFirst({ where: WANTED }).sync() !== undefined; + return ( + this.db.query.conversations.findFirst({ where: WANTED }).sync() !== undefined || + this.db.query.tasks.findFirst({ where: UNSEEN }).sync() !== undefined + ); } muted(channel: string, threadTs: string): string | null { diff --git a/src/main.ts b/src/main.ts index 1611921..f6a2721 100644 --- a/src/main.ts +++ b/src/main.ts @@ -4,10 +4,21 @@ 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"; +for (const signal of ["SIGTERM", "SIGINT"] as const) + process.on(signal, () => { + log.info("service stopped", { signal }); + process.exit(0); + }); +process.on("unhandledRejection", (error) => { + log.error("unhandled rejection", { error: String(error) }); +}); + +await container.resolve(Roster).load(); const scheduler = container.resolve(Scheduler); log.info("service started"); @@ -31,12 +42,3 @@ watchFile(policyPath, { interval: 2000, persistent: false }, (curr, prev) => { log.error("policy reload rejected — keeping last-known-good", { error: String(error) }); } }); - -for (const signal of ["SIGTERM", "SIGINT"] as const) - process.on(signal, () => { - log.info("service stopped", { signal }); - process.exit(0); - }); -process.on("unhandledRejection", (error) => { - log.error("unhandled rejection", { error: String(error) }); -}); diff --git a/src/policy.ts b/src/policy.ts index a977b9c..2a4fdbd 100644 --- a/src/policy.ts +++ b/src/policy.ts @@ -10,7 +10,7 @@ const ModelTier = z .prefault({}); const PolicySchema = z.object({ - persona: z.string().optional(), + persona: z.string().default(""), venue_instructions: z.record(z.string(), z.string()).default({}), ear_debounce_ms: z.number().default(45_000), turns: z.object({ timeout_ms: z.number().default(600_000) }).prefault({}), diff --git a/src/prompt-renderer.ts b/src/prompt-renderer.ts index 8e16ea9..61259da 100644 --- a/src/prompt-renderer.ts +++ b/src/prompt-renderer.ts @@ -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}] ${await this.speaker(line.user ?? line.bot_id)}: ${text}${files}`; + return ` [${channel} ${line.ts}] ${this.speaker(line.user ?? line.bot_id)}: ${text}${files}`; } - private async speaker(user: string | undefined): Promise { - const name = user ? await this.roster.nameOf(user) : null; + private speaker(user: string | undefined): string { + const name = user ? this.roster.nameOf(user) : null; return `<@${user ?? "?"}>${name ? ` (${name})` : ""}`; } } diff --git a/src/roster.ts b/src/roster.ts index 5f7a8d7..74721c3 100644 --- a/src/roster.ts +++ b/src/roster.ts @@ -4,18 +4,14 @@ 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) { - this.loaded = this.load(); - } + constructor(private readonly web: WebClient) {} - async nameOf(principalId: string): Promise { - await this.loaded; + nameOf(principalId: string): string | null { return this.names.get(principalId) ?? null; } - private async load(): Promise { + 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 49ed19f..b26ec02 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 } from "drizzle-orm"; import { inject, instanceCachingFactory, registry, singleton } from "tsyringe"; import { Codex } from "./codex"; import { Debounced } from "./debounce"; -import { DB, LedgerService, openDb, WANTED, type Db } from "./ledger-service"; +import { DB, LedgerService, openDb, UNSEEN, 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"; @@ -50,8 +50,8 @@ const HEARD_SUBTYPES = new Set([ ]) @singleton() export class Scheduler { - private readonly overheard = new Debounced(() => this.triage()); - private readonly replies = new Debounced(() => this.respond()); + private readonly overheard = new Debounced("triage", () => this.triage()); + private readonly replies = new Debounced("respond", () => this.respond()); constructor( @inject(DB) private readonly db: Db, @@ -89,35 +89,34 @@ export class Scheduler { }) .sync(); const settled = this.db.query.tasks - .findMany({ - where: and( - or(eq(tasks.status, "done"), eq(tasks.waitingOn, "human")), - or(isNull(tasks.seenAt), gt(tasks.updatedAt, tasks.seenAt)), - ), - orderBy: asc(tasks.updatedAt), - }) + .findMany({ where: UNSEEN, orderBy: asc(tasks.updatedAt) }) .sync(); if (convos.length === 0 && settled.length === 0) return; const prompt = await this.prompts.response(convos, settled); const direct = convos.filter((convo) => convo.direct); this.ledger.rendered(convos, settled); this.voice.begin(direct); - await this.codex.respond(prompt).finally(() => { + try { + await this.codex.respond(prompt); + } finally { this.voice.close(direct); - }); + } this.tick(); - if (this.ledger.wantsResponse()) this.replies.schedule(0); } private async triage(): Promise { const unjudged = this.db.query.conversations - .findMany({ where: and(eq(conversations.direct, false), eq(conversations.woken, false)) }) + .findMany({ + where: and(eq(conversations.direct, false), eq(conversations.woken, false)), + orderBy: asc(conversations.since), + limit: BATCH, + }) .sync(); if (unjudged.length > 0) { await this.codex.triage(await this.prompts.overheard(unjudged)); this.ledger.held(unjudged); } - if (this.ledger.wantsResponse()) this.replies.schedule(0); + this.tick(); } private async runWorker(taskId: string): Promise { @@ -148,7 +147,6 @@ export class Scheduler { outcome: after?.outcome, turns, }); - if (after?.status === "done" || after?.waitingOn === "human") this.replies.schedule(0); this.tick(); } @@ -160,7 +158,8 @@ export class Scheduler { } private tick(): void { - if (this.ledger.wakeDueTasks()) this.replies.schedule(0); + this.ledger.wakeDueTasks(); + if (this.ledger.wantsResponse()) this.replies.schedule(0); for (const taskId of this.ledger.dispatchRunnable(this.policy.executions.max_concurrent)) void this.runWorker(taskId); } diff --git a/src/tools.ts b/src/tools.ts index 86451a5..1fa34b0 100644 --- a/src/tools.ts +++ b/src/tools.ts @@ -128,7 +128,7 @@ export const workerTools = (server) => { shared(server); const ledger = container.resolve(LedgerService); - const { park_after_ms } = container.resolve(POLICY).tasks; + const policy = container.resolve(POLICY); tool( server, "task_complete", @@ -149,7 +149,7 @@ export const workerTools = type: "wait", waitingOn: "human", why: question, - wakeAt: new Date(Date.now() + park_after_ms).toISOString(), + wakeAt: new Date(Date.now() + policy.tasks.park_after_ms).toISOString(), }); return `task ${taskId} waiting on a human`; },