Bug hunt: one ledger read in tick(), failed turns re-arm, watermark monotonic

src/debounce.ts

@@ -1,9 +1,14 @@
+import { log } from "./log";
+
 export class Debounced {
   private timer: ReturnType<typeof setTimeout> | null = null;
   private running = false;
   private rerun = false;
 
-  constructor(private readonly run: () => Promise<void>) {}
+  constructor(
+    private readonly label: string,
+    private readonly run: () => Promise<void>,
+  ) {}
 
   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();
+        }
+      });
   }
 }

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 {

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) });
-});

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({}),

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<string> {
-    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})` : ""}`;
   }
 }

src/roster.ts

@@ -4,18 +4,14 @@ import { WebClient, type UsersListResponse } from "@slack/web-api";
 @singleton()
 export class Roster {
   private readonly names = new Map<string, string>();
-  private readonly loaded: Promise<void>;
 
-  constructor(private readonly web: WebClient) {
-    this.loaded = this.load();
-  }
+  constructor(private readonly web: WebClient) {}
 
-  async nameOf(principalId: string): Promise<string | null> {
-    await this.loaded;
+  nameOf(principalId: string): string | null {
     return this.names.get(principalId) ?? null;
   }
 
-  private async load(): Promise<void> {
+  async load(): Promise<void> {
     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(

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<string | undefined>([
 ])
 @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<void> {
     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<void> {
@@ -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);
   }

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`;
       },