From c91498c01ab73fd68061cd79542722691b1e5922 Mon Sep 17 00:00:00 2001 From: TerrifiedBug <35064668+TerrifiedBug@users.noreply.github.com> Date: Sun, 23 Aug 2026 08:48:59 +0000 Subject: [PATCH] feat: durable inbound receipts, and a /cleanup that can clean (#61, #67) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **#61 — a durable record that an inbound message arrived.** The spooled payload is unlinked before the handoff, so after a successful delivery the only surviving copy lived inside the receiving agent's transcript, in that agent's private format. A supervisor asking "did the operator's reply arrive?" had to parse another program's log — and when it did, it broke on a host whose session directory and cwd disagreed. One bounded receipt per route now, written BEFORE `onMsg`: message id, date, sender, chat, thread, and a sha256 of the text. Before the handoff because the case a receipt exists for is a consumer that died; a receipt written afterwards records only the deliveries that already succeeded. A hash rather than the text because the payload already lives in the consuming agent's transcript, a supervisor verifying a challenge knows what it sent, and a hash cannot leak a message to anything that did not already know it. Bounded by construction — one file per route, replaced in place — so the existing route purge is the only reaper needed. Documented in docs/guide.md as a stable contract, since the whole point is that another process may rely on it. The receipt lives in the watched directory and ends in `.json`, so the watcher now excludes it by name: evidence, never a payload. **#67 — /cleanup could not clean the mess it exists for.** `staleThreads` was `!alive(pid)` and nothing else, and in a DM host cleanup *deletes*. So the only offered remedy for 83 topics minutes old would also have permanently destroyed an unrelated project topic from eight days earlier. There was no way to accept part of it, which is why the incident was resolved by scripting deletions by id outside the tool. Stale topics are now classified. The discriminator is evidence, not a guess: an entry recording a session file that does not exist belonged to a process that claimed a topic and died without writing one line of transcript. A claim with no recorded session file at all is an older-format claim, not a crash — absence of evidence must not become evidence of a crash. The preview says which is which and how old. And selection is possible: `/cleanup go …` for exact topics, `/cleanup go never-ran` for the crash-loop set. Unparseable arguments print usage instead of falling through to "all", because falling through deletes irreversibly. Selection always applies to a freshly derived set, so naming an id that has since resumed selects nothing. Replayed against the real registry from the incident: `never-ran` takes exactly the 11 spam topics and preserves both legitimate ones. The old predicate took all 13. Each test was checked against the old behaviour: moving the receipt after the handoff turns "survives a consumer that dies mid-handoff" red (a subprocess that hard-exits inside onMsg — an in-process throw is caught and cannot distinguish the two orderings). bun run check: 314 pass, 0 fail. --- docs/guide.md | 40 ++++++++++++++ src/bridge.test.ts | 82 ++++++++++++++++++++++++++-- src/bridge.ts | 106 +++++++++++++++++++++++++++++++++--- src/topics.test.ts | 95 +++++++++++++++++++++++++++++++- src/topics.ts | 131 +++++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 439 insertions(+), 15 deletions(-) diff --git a/docs/guide.md b/docs/guide.md index 6705bca..f24a0dd 100644 --- a/docs/guide.md +++ b/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//` / `route/dm/` | Cross-process routed-message spools for topics and untopiced private DMs | +| `route//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 diff --git a/src/bridge.test.ts b/src/bridge.test.ts index d987690..c148524 100644 --- a/src/bridge.test.ts +++ b/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"); + }); +}); diff --git a/src/bridge.ts b/src/bridge.ts index 15f9db1..e955c50 100644 --- a/src/bridge.ts +++ b/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 [… | 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 […|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)); } } diff --git a/src/topics.test.ts b/src/topics.test.ts index b665e3d..d85dbc7 100644 --- a/src/topics.test.ts +++ b/src/topics.test.ts @@ -6,6 +6,7 @@ import { statePath } from "./access"; import type { TgMessage } from "./api"; import { DM_ROUTE_KEY, + INBOUND_RECEIPT, ROUTED_TTL_MS, type ThreadEntry, type ThreadRegistry, @@ -15,6 +16,7 @@ import { isResumedOwner, loadRegistry, purgeRouteDir, + readInboundReceipt, releaseThread, sessionTopicTitle, staleThreads, @@ -208,7 +210,11 @@ describe("writeRouted / watchRoute", () => { expect(got).toHaveLength(1); expect(got[0].message_id).toBe(42); expect(got[0].text).toBe("hi"); - expect(readdirSync(statePath("route", "7"))).toHaveLength(0); + // The payload is consumed; the receipt stays (#61). Asserted as "no payload + // left" rather than "empty dir", because the dir now deliberately retains + // exactly one bounded file as evidence the message arrived. + expect(readdirSync(statePath("route", "7")).filter((f) => f !== INBOUND_RECEIPT)).toHaveLength(0); + expect(readInboundReceipt(7)?.messageId).toBe(42); }); test("a watcher that no longer owns a mutable route leaves its payload for the owner", () => { @@ -338,3 +344,90 @@ describe("registry writes survive concurrency (#68)", () => { expect(loadRegistry().threads["8002"]?.pid).toBe(8002); }); }); + +describe("durable inbound receipt (#61)", () => { + const msg = (id: number, text: string, thread?: number): TgMessage => ({ + message_id: id, + date: 1_700_000_000, + from: { id: 555, is_bot: false, first_name: "op" }, + chat: { id: 100, type: "supergroup" }, + text, + ...(thread === undefined ? {} : { is_topic_message: true, message_thread_id: thread }), + }); + + test("a delivered message leaves a receipt a supervisor can verify", () => { + writeRouted(7, msg(42, "arm-code-abc", 7)); + watchRoute(7, () => {})(); + const r = readInboundReceipt(7); + expect(r?.messageId).toBe(42); + expect(r?.chatId).toBe(100); + expect(r?.fromId).toBe(555); + expect(r?.messageThreadId).toBe(7); + expect(r?.receivedAt).toBeGreaterThan(0); + // A hash, not the text: the payload already lives in the consuming agent's + // transcript, and a supervisor verifying a challenge knows what it sent. + expect(r?.textSha256).toBe(new Bun.CryptoHasher("sha256").update("arm-code-abc").digest("hex")); + expect(JSON.stringify(r)).not.toContain("arm-code-abc"); + }); + + test("the receipt survives a consumer that dies mid-handoff — the case it exists for", async () => { + // A thrown error is caught and execution continues, so an in-process throw + // cannot distinguish before-handoff from after. Process death can: the child + // hard-exits inside `onMsg`, so anything sequenced after the handoff never + // runs. This is the assertion that fails when the receipt is written last. + writeRouted(8, msg(43, "boom", 8)); + const runner = join(dir, "die.ts"); + writeFileSync( + runner, + `import { watchRoute } from ${JSON.stringify(join(import.meta.dirname, "topics.ts"))};\n` + + `watchRoute(8, () => process.exit(9));\n`, + ); + const { exitCode } = await Bun.spawn([process.execPath, runner], { + env: { ...process.env, OMP_TELEGRAM_STATE_DIR: dir }, + stdout: "ignore", + stderr: "ignore", + }).exited.then((code) => ({ exitCode: code })); + expect(exitCode).toBe(9); // the consumer really did die inside onMsg + expect(readInboundReceipt(8)?.messageId).toBe(43); + }); + + test("it is bounded: a second delivery replaces the first, never accumulates", () => { + writeRouted(9, msg(44, "one", 9)); + watchRoute(9, () => {})(); + writeRouted(9, msg(45, "two", 9)); + watchRoute(9, () => {})(); + expect(readInboundReceipt(9)?.messageId).toBe(45); + expect(readdirSync(statePath("route", "9"))).toEqual([INBOUND_RECEIPT]); + }); + + test("the watcher never consumes its own receipt as a payload", () => { + // It lives in the watched dir and ends in `.json`, so this is a real hazard. + writeRouted(10, msg(46, "hi", 10)); + watchRoute(10, () => {})(); + const delivered: TgMessage[] = []; + watchRoute(10, (m) => delivered.push(m))(); + expect(delivered).toEqual([]); + expect(readInboundReceipt(10)?.messageId).toBe(46); + }); + + test("the DM route gets one too", () => { + writeRouted(DM_ROUTE_KEY, msg(47, "dm")); + watchRoute(DM_ROUTE_KEY, () => {})(); + expect(readInboundReceipt(DM_ROUTE_KEY)?.messageId).toBe(47); + }); + + test("purgeRouteDir removes it with the rest of the route state", () => { + writeRouted(11, msg(48, "hi", 11)); + watchRoute(11, () => {})(); + expect(readInboundReceipt(11)).toBeDefined(); + purgeRouteDir(11); + expect(readInboundReceipt(11)).toBeUndefined(); + }); + + test("a corrupt or absent receipt reads as absent, never throws", () => { + expect(readInboundReceipt(9999)).toBeUndefined(); + mkdirSync(statePath("route", "12"), { recursive: true }); + writeFileSync(join(statePath("route", "12"), INBOUND_RECEIPT), "{ not json"); + expect(readInboundReceipt(12)).toBeUndefined(); + }); +}); diff --git a/src/topics.ts b/src/topics.ts index 3b7ee38..d5fb3a1 100644 --- a/src/topics.ts +++ b/src/topics.ts @@ -6,9 +6,10 @@ // + policy, so it is fully unit-testable. Telegram I/O stays in api.ts / // outbound.ts. -import { randomBytes } from "node:crypto"; +import { createHash, randomBytes } from "node:crypto"; import { type FSWatcher, + existsSync, linkSync, mkdirSync, readFileSync, @@ -219,6 +220,48 @@ export function staleThreads( } +/** + * Why a stale topic is stale (#67). + * + * `!alive(pid)` alone cannot tell a crash loop from a fortnight of history, and + * in a DM host `/cleanup` deletes irreversibly — so its only remedy for 83 + * topics minutes old also destroyed an unrelated project topic from eight days + * earlier. An operator needs to see which is which before tapping Delete. + * + * `never-ran` is the high-confidence signal: the entry records a session file + * that does not exist, so that process claimed a topic and died without writing + * one line of transcript. A session that did any work leaves a file behind. + */ +export type StaleReason = "never-ran" | "ended"; + +export interface StaleTopic { + threadId: number; + entry: ThreadEntry; + reason: StaleReason; + /** How long ago the claim was made, in ms. */ + ageMs: number; +} + +/** + * Annotate stale topics with why they are stale and how old the claim is. + * + * `exists` is injected so this stays pure and testable, like `alive` above. + */ +export function classifyStale( + stale: Array<[number, ThreadEntry]>, + now: number, + exists: (path: string) => boolean = existsSync, +): StaleTopic[] { + return stale.map(([threadId, entry]) => ({ + threadId, + entry, + // No recorded session file at all is an older-format claim, not evidence of + // a crash: only a recorded-but-absent file proves nothing ever ran. + reason: entry.sessionFile !== undefined && !exists(entry.sessionFile) ? "never-ran" : "ended", + ageMs: Math.max(0, now - entry.claimedAt), + })); +} + type SessionIdentity = Pick; /** Session files survive `omp --resume`; runtime session IDs may change. */ @@ -384,6 +427,71 @@ export function writeRouted(threadId: number | typeof DM_ROUTE_KEY, msg: TgMessa renameSync(tmp, join(dir, base)); } +/** File name of the per-route inbound receipt (#61). Stable external contract. */ +export const INBOUND_RECEIPT = "last-inbound.json"; + +/** + * What a supervising process can rely on after an inbound message is delivered. + * + * `textSha256` rather than the text: the payload already exists in the + * receiving agent's transcript, and a receipt is for *proving arrival*, not for + * holding a second copy of what a user wrote. A supervisor verifying a + * challenge code knows the code it sent, so hashing its own copy is enough — + * and a hash cannot leak a message to anything that did not already know it. + */ +export interface InboundReceipt { + messageId: number; + date: number; + fromId?: number; + chatId: number; + messageThreadId?: number; + textSha256?: string; + /** When this receipt was written, which is when the payload was consumed. */ + receivedAt: number; +} + +/** + * Record that a message arrived, before anything consumes it (#61). + * + * Written before the handoff on purpose: a receipt whose whole value is + * surviving a consumer that died is worthless if the consumer writes it. One + * file per route, replaced in place, so it is bounded by construction and needs + * no reaper beyond {@link purgeRouteDir}. + */ +export function writeInboundReceipt(threadId: number | typeof DM_ROUTE_KEY, msg: TgMessage): void { + const dir = routeDir(threadId); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + const receipt: InboundReceipt = { + messageId: msg.message_id, + date: msg.date, + ...(msg.from?.id === undefined ? {} : { fromId: msg.from.id }), + chatId: msg.chat.id, + ...(msg.message_thread_id === undefined ? {} : { messageThreadId: msg.message_thread_id }), + ...(msg.text === undefined ? {} : { textSha256: createHash("sha256").update(msg.text).digest("hex") }), + receivedAt: Date.now(), + }; + const tmp = join(dir, `tmp-${process.pid}-${INBOUND_RECEIPT}`); + try { + writeFileSync(tmp, JSON.stringify(receipt, null, 2) + "\n", { mode: 0o600 }); + renameSync(tmp, join(dir, INBOUND_RECEIPT)); + } catch { + rmSync(tmp, { force: true }); + throw new Error("could not write inbound receipt"); + } +} + +/** Read a route's inbound receipt, or `undefined` when none has been written. */ +export function readInboundReceipt(threadId: number | typeof DM_ROUTE_KEY): InboundReceipt | undefined { + try { + const parsed: unknown = JSON.parse(readFileSync(join(routeDir(threadId), INBOUND_RECEIPT), "utf8")); + if (!parsed || typeof parsed !== "object") return undefined; + const r = parsed as InboundReceipt; + return typeof r.messageId === "number" && typeof r.chatId === "number" ? r : undefined; + } catch { + return undefined; + } +} + /** * Watch a route's spool dir and hand each spooled message to `onMsg`. Uses an * initial scan + fs.watch + a 5s rescan (fs.watch alone is not reliable enough). @@ -403,7 +511,9 @@ export function watchRoute( const processed = new Set(); const handle = (name: string): void => { - if (!name || name.startsWith("tmp-") || !name.endsWith(".json") || processed.has(name)) return; + // The receipt lives in this same dir and ends in `.json` (#61): it is + // evidence, never a payload, so it must never be consumed or unlinked. + if (!name || name.startsWith("tmp-") || name === INBOUND_RECEIPT || !name.endsWith(".json") || processed.has(name)) return; if (accept && !accept()) return; const full = join(dir, name); let mtimeMs: number; @@ -432,10 +542,25 @@ export function watchRoute( } catch { /* ignore */ } + let msg: TgMessage; try { - onMsg(JSON.parse(raw) as TgMessage); + msg = JSON.parse(raw) as TgMessage; } catch (err) { log?.warn(`[telegram] routed payload parse failed (${name}): ${String(err)}`); + return; + } + // Receipt BEFORE the handoff (#61): its whole purpose is to outlive a + // consumer that dies, so a receipt written after `onMsg` records only the + // deliveries that already succeeded. + try { + writeInboundReceipt(threadId, msg); + } catch (err) { + log?.warn(`[telegram] inbound receipt write failed: ${String(err)}`); + } + try { + onMsg(msg); + } catch (err) { + log?.warn(`[telegram] routed delivery failed (${name}): ${String(err)}`); } };