feat: durable inbound receipts, and a /cleanup that can clean (#61, #67)

docs/guide.md

@@ -309,6 +309,46 @@ deliver from. `telegram_ask` responds only to the exact user who originated the
 | `daemon.log` | Rotating daemon output (5 MiB, one previous generation) |
 | `threads.json` | Topic registry — which session (pid/cwd) owns which forum topic |
 | `route/<thread_id>/` / `route/dm/` | Cross-process routed-message spools for topics and untopiced private DMs |
+| `route/<thread_id>/last-inbound.json` / `route/dm/last-inbound.json` | Receipt for the most recent delivered inbound message (see below) |
+
+### Inbound receipts — a contract for external supervisors
+
+A routed payload is deleted the moment it is consumed, so after a successful
+delivery the only surviving copy used to be inside the receiving agent's own
+transcript, in that agent's private format. A supervising process asking the
+reasonable question *"did the operator's reply actually arrive?"* had to parse
+another program's log to find out.
+
+So one bounded receipt is written per route, **before** the payload is handed to
+the session:
+
+```json
+{
+  "messageId": 12345,
+  "date": 1700000000,
+  "fromId": 555,
+  "chatId": 100,
+  "messageThreadId": 7,
+  "textSha256": "9f86d081…",
+  "receivedAt": 1700000000123
+}
+```
+
+- **Written before the handoff**, because the case a receipt exists for is a
+  consumer that died. One written afterwards records only the deliveries that
+  already succeeded.
+- **A hash, not the text.** The message already lives in the consuming agent's
+  transcript, and a supervisor verifying a challenge code knows the code it
+  sent — hashing its own copy is enough. A hash also cannot leak a message to
+  anything that did not already know it.
+- **Bounded by construction**: exactly one file per route, replaced in place. No
+  reaper is needed beyond the existing route purge, which removes it with the
+  rest of the route state.
+- Written `0600` inside the `0700` route directory, tmp+rename, so a reader
+  never sees a partial file.
+
+This file is a stable contract: read it rather than scraping a session
+transcript.
 
 ## Streaming behavior
 

src/bridge.test.ts

@@ -1,13 +1,13 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
-import { existsSync, mkdtempSync, readdirSync, rmSync } from "node:fs";
+import { existsSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { defaultAccess, saveAccess, statePath } from "./access";
 import { type Logger, type TgMessage, TgError } from "./api";
-import { type BridgeHost, BOT_COMMANDS, PUBLIC_BOT_COMMANDS, handleUpdate, syncBotCommands } from "./bridge";
+import { type BridgeHost, BOT_COMMANDS, PUBLIC_BOT_COMMANDS, cleanupPreviewLine, handleUpdate, parseCleanupArgs, selectCleanupTargets, syncBotCommands } from "./bridge";
 import { type TelegramCall, SpawnController } from "./control";
 import { TelegramPromptController } from "./prompts";
-import { DM_ROUTE_KEY, claimDmOwner, loadRegistry, saveRegistry, watchRoute } from "./topics";
+import { DM_ROUTE_KEY, type ThreadEntry, claimDmOwner, classifyStale, loadRegistry, saveRegistry, watchRoute } from "./topics";
 
 const previousStateDir = process.env.OMP_TELEGRAM_STATE_DIR;
 let dir: string;
@@ -590,3 +590,79 @@ describe("status surfaces the DM user-topic-creation setting", () => {
     expect(String(reply?.payload.text)).not.toContain("Users can create DM topics");
   });
 });
+
+describe("cleanup selection (#67)", () => {
+  const entry = (name: string, claimedAt: number, sessionFile?: string): ThreadEntry => ({
+    pid: 4242,
+    cwd: `/w/${name}`,
+    name,
+    claimedAt,
+    ...(sessionFile === undefined ? {} : { sessionFile }),
+  });
+
+  test("bad input prints usage rather than cleaning everything", () => {
+    // In a DM host `/cleanup go` deletes irreversibly, so an argument the
+    // parser does not understand must never fall through to "all".
+    expect(parseCleanupArgs("nonsense")).toBeUndefined();
+    expect(parseCleanupArgs("go 10071 banana")).toBeUndefined();
+    expect(parseCleanupArgs("go -5")).toBeUndefined();
+    expect(parseCleanupArgs("go never-ran extra")).toBeUndefined();
+  });
+
+  test("the grammar", () => {
+    expect(parseCleanupArgs("")).toEqual({ kind: "preview" });
+    expect(parseCleanupArgs("  ")).toEqual({ kind: "preview" });
+    expect(parseCleanupArgs("go")).toEqual({ kind: "all" });
+    expect(parseCleanupArgs("go never-ran")).toEqual({ kind: "never-ran" });
+    expect(parseCleanupArgs("go 10071 10073")).toEqual({ kind: "ids", ids: [10071, 10073] });
+  });
+
+  test("naming ids cleans only those — the case that forced a hand-written script", () => {
+    // The incident: 83 topics minutes old alongside one project topic from eight
+    // days earlier. `/cleanup go` would have deleted all of them, irreversibly,
+    // so the only safe remedy was scripting the deletions by id outside the tool.
+    const stale: Array<[number, ThreadEntry]> = [
+      [9549, entry("veltrosecurity", 1_000)],
+      [10071, entry("conductor", 2_000)],
+      [10073, entry("conductor", 3_000)],
+    ];
+    const chosen = selectCleanupTargets(stale, { kind: "ids", ids: [10071, 10073] }, 10_000);
+    expect(chosen.map(([id]) => id)).toEqual([10071, 10073]);
+  });
+
+  test("an id that is no longer stale selects nothing, never something else", () => {
+    const stale: Array<[number, ThreadEntry]> = [[10071, entry("conductor", 2_000)]];
+    expect(selectCleanupTargets(stale, { kind: "ids", ids: [99999] }, 10_000)).toEqual([]);
+  });
+
+  test("never-ran selects exactly the topics whose session wrote no transcript", () => {
+    // The crash-loop signature: a recorded session file that does not exist.
+    const stale: Array<[number, ThreadEntry]> = [
+      [9549, entry("veltrosecurity", 1_000, join(dir, "real.jsonl"))],
+      [10071, entry("conductor", 2_000, "/does/not/exist-1.jsonl")],
+      [10073, entry("conductor", 3_000, "/does/not/exist-2.jsonl")],
+      [10075, entry("legacy-no-sessionfile", 4_000)],
+    ];
+    writeFileSync(join(dir, "real.jsonl"), "{}\n");
+    const chosen = selectCleanupTargets(stale, { kind: "never-ran" }, 10_000);
+    expect(chosen.map(([id]) => id)).toEqual([10071, 10073]);
+  });
+
+  test("a claim with no recorded session file is history, not a crash", () => {
+    // Older-format claims record no session file. Absence of evidence must not
+    // become evidence of a crash, or a legacy topic gets swept up.
+    const classified = classifyStale([[10075, entry("legacy", 1_000)]], 10_000);
+    expect(classified[0]?.reason).toBe("ended");
+  });
+
+  test("the preview says which is which, and how old", () => {
+    writeFileSync(join(dir, "real.jsonl"), "{}\n");
+    const [ranTopic] = classifyStale([[9549, entry("veltrosecurity", 0, join(dir, "real.jsonl"))]], 8 * 24 * 3_600_000);
+    const [neverRan] = classifyStale([[10071, entry("conductor", 0, "/nope.jsonl")]], 120_000);
+    expect(cleanupPreviewLine(neverRan!)).toContain("session never ran");
+    expect(cleanupPreviewLine(neverRan!)).toContain("recent");
+    expect(cleanupPreviewLine(neverRan!)).toContain("2m ago");
+    expect(cleanupPreviewLine(ranTopic!)).not.toContain("session never ran");
+    expect(cleanupPreviewLine(ranTopic!)).toContain("h ago");
+  });
+});

src/bridge.ts

@@ -23,7 +23,9 @@ import {
 import type { TelegramPromptController } from "./prompts";
 import {
   DM_ROUTE_KEY,
+  type StaleTopic,
   type ThreadEntry,
+  classifyStale,
   decideRoute,
   isAlive,
   isResumedOwner,
@@ -66,7 +68,15 @@ const COMMANDS: CommandSpec[] = [
     ],
   },
   { command: "sessions", description: "List active omp sessions", scope: "global", help: ["/sessions — list active omp sessions and topic attachment"] },
-  { command: "cleanup", description: "Tidy topics of exited sessions", scope: "global", help: ["/cleanup — preview exited-session topics, then tap to confirm (or /cleanup go to skip the preview)"] },
+  {
+    command: "cleanup",
+    description: "Tidy topics of exited sessions",
+    scope: "global",
+    help: [
+      "/cleanup — preview exited-session topics, then tap to confirm",
+      "/cleanup go [<id>… | never-ran] — skip the preview, optionally for a subset",
+    ],
+  },
   { command: "stop", description: "Abort this topic's omp task", scope: "session", help: ["/stop — abort this topic’s current task"] },
   { command: "compact", description: "Compact this session's context", scope: "session", help: ["/compact [focus] — compact this session’s context"] },
   { command: "model", description: "Show or change this session's model", scope: "session", help: ["/model [provider/id] — show or change this session’s model"] },
@@ -339,6 +349,71 @@ function cleanupResultText(cleaned: number, failed: number, deletes: boolean): s
   return `🧹 ${verb} ${cleaned} stale topic${cleaned === 1 ? "" : "s"}${suffix}`;
 }
 
+/**
+ * What `/cleanup [go [ids|never-ran]]` asked for, or `undefined` for bad input.
+ *
+ * Parsed separately from the handler so the grammar is testable without a live
+ * bridge, and so an unparseable argument prints usage instead of quietly
+ * cleaning everything — which, in a DM host, deletes irreversibly.
+ */
+export type CleanupSelection =
+  | { kind: "preview" }
+  | { kind: "all" }
+  | { kind: "ids"; ids: readonly number[] }
+  | { kind: "never-ran" };
+
+export function parseCleanupArgs(args: string): CleanupSelection | undefined {
+  const words = args.trim().split(/\s+/).filter(Boolean);
+  if (words.length === 0) return { kind: "preview" };
+  if (words[0] !== "go") return undefined;
+  const rest = words.slice(1);
+  if (rest.length === 0) return { kind: "all" };
+  if (rest.length === 1 && rest[0] === "never-ran") return { kind: "never-ran" };
+  const ids = rest.map(Number);
+  return ids.every((n) => Number.isSafeInteger(n) && n > 0) ? { kind: "ids", ids } : undefined;
+}
+
+/** How recent a claim has to be for the preview to call it out as a burst. */
+const CLEANUP_RECENT_MS = 15 * 60_000;
+
+/**
+ * Narrow the stale set to what was asked for (#67).
+ *
+ * The set is always re-derived by the caller first, so naming an id that has
+ * since resumed selects nothing rather than acting on a stale snapshot.
+ */
+export function selectCleanupTargets(
+  stale: Array<[number, ThreadEntry]>,
+  selection: CleanupSelection,
+  now: number,
+): Array<[number, ThreadEntry]> {
+  if (selection.kind === "ids") {
+    const wanted = new Set(selection.ids);
+    return stale.filter(([threadId]) => wanted.has(threadId));
+  }
+  if (selection.kind === "never-ran") {
+    const neverRan = new Set(
+      classifyStale(stale, now)
+        .filter((t) => t.reason === "never-ran")
+        .map((t) => t.threadId),
+    );
+    return stale.filter(([threadId]) => neverRan.has(threadId));
+  }
+  return stale;
+}
+
+/** One preview line: what it is, how old, and whether it ever did anything. */
+export function cleanupPreviewLine(topic: StaleTopic): string {
+  const mins = Math.round(topic.ageMs / 60_000);
+  const age = mins < 1 ? "just now" : mins < 90 ? `${mins}m ago` : `${Math.round(mins / 60)}h ago`;
+  // The distinction that matters before an irreversible delete: a topic whose
+  // session never wrote a transcript did nothing, and a burst of them is a
+  // crash loop. A topic that ran is somebody's history.
+  const note = topic.reason === "never-ran" ? " ⚠ session never ran" : "";
+  const burst = topic.reason === "never-ran" && topic.ageMs < CLEANUP_RECENT_MS ? ", recent" : "";
+  return `#${topic.threadId} ${topic.entry.name} — ${topic.entry.cwd} (${age}${burst})${note}`;
+}
+
 /**
  * Preview the stale topics with a confirm/cancel keyboard so the owner can tidy
  * with one tap instead of typing `/cleanup go`. The picker records the exact
@@ -356,7 +431,7 @@ async function sendCleanupPreview(
   if (!ownerId) return;
   const deletes = isDmChat(topicsChat);
   const plural = stale.length === 1 ? "" : "s";
-  const lines = stale.map(([threadId, entry]) => `#${threadId} ${entry.name} — ${entry.cwd}`).join("\n");
+  const lines = classifyStale(stale, Date.now()).map(cleanupPreviewLine).join("\n");
   const prompt = deletes
     ? `Delete these ${stale.length} topic${plural} and their message history?`
     : `Close these ${stale.length} topic${plural}? History is kept and reopened on re-adoption.`;
@@ -516,12 +591,22 @@ export async function handleGlobalCommand(
       }
     }
   } else if (command === "cleanup") {
+    const selection = parseCleanupArgs(args);
     if (!owner) {
       await commandReply(host, access, msg, "Pair this DM locally before using cleanup.");
     } else if (!access.topicsChat) {
       await commandReply(host, access, msg, "Topics mode is off — nothing to clean.");
-    } else if (args !== "" && args !== "go") {
-      await commandReply(host, access, msg, "usage: /cleanup [go]");
+    } else if (selection === undefined) {
+      await commandReply(
+        host,
+        access,
+        msg,
+        "usage: /cleanup [go [<id>…|never-ran]]\n\n" +
+          "/cleanup — preview\n" +
+          "/cleanup go — everything stale\n" +
+          "/cleanup go 10071 10073 — only those topics\n" +
+          "/cleanup go never-ran — only topics whose session never wrote a transcript",
+      );
     } else {
       const registry = loadRegistry(host.warn);
       const topicsChat = registry.chatId || access.topicsChat;
@@ -530,13 +615,18 @@ export async function handleGlobalCommand(
       // host it could numerically collide with a real stale group topic).
       const controlExclude = topicsChat === pairedOwnerId(access) ? access.controlThreadId : undefined;
       const stale = staleThreads(registry, isAlive, controlExclude);
+      // Selection is applied to the freshly derived set, never to a remembered
+      // one: an id the operator names that is no longer stale is simply absent.
+      const chosen = selectCleanupTargets(stale, selection, Date.now());
       if (stale.length === 0) {
         await commandReply(host, access, msg, "Nothing to clean — no stale session topics. Live sessions and omp control remain.");
-      } else if (args === "") {
-        await sendCleanupPreview(host, access, msg, topicsChat, stale);
+      } else if (chosen.length === 0) {
+        await commandReply(host, access, msg, `Nothing matched. ${stale.length} stale topic(s) exist — run /cleanup to see them.`);
+      } else if (selection.kind === "preview") {
+        await sendCleanupPreview(host, access, msg, topicsChat, chosen);
       } else {
-        // args === "go": re-derived above; never act on the preview.
-        const { cleaned, failed, deletes } = await executeCleanup(host, topicsChat, stale);
+        // Re-derived above; never act on the preview.
+        const { cleaned, failed, deletes } = await executeCleanup(host, topicsChat, chosen);
         await commandReply(host, access, msg, cleanupResultText(cleaned, failed, deletes));
       }
     }