fix: a daemon spawn either becomes a daemon or says why (#68)

src/api.ts

@@ -309,7 +309,14 @@ function lockFresh(lockPath: string, freshMs: number): boolean {
  * not a racy read-then-write — decides the single winner, and the target is
  * populated the instant it appears (no empty mid-write window).
  */
-function linkClaim(target: string, pid: number, content: string = String(pid)): boolean {
+/**
+ * Create `target` atomically, or report that somebody else already has.
+ *
+ * Exported so the thread registry can serialise its read-modify-write on the
+ * same primitive the poll lock uses (#68): two writers racing a shared
+ * `threads.json.tmp` published each other's file and silently lost claims.
+ */
+export function linkClaim(target: string, pid: number, content: string = String(pid)): boolean {
   const temp = `${target}.${pid}.${randomBytes(6).toString("hex")}`;
   writeFileSync(temp, content, { mode: 0o600 });
   try {

src/daemon.test.ts

@@ -3,7 +3,7 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { defaultAccess, saveAccess, statePath } from "./access";
-import { daemonDisableReason, ensureDaemon, readDaemonState } from "./daemon";
+import { daemonDisableReason, ensureDaemon, type EnsureDaemonOptions, readDaemonState, resolveRuntime } from "./daemon";
 
 const previousStateDir = process.env.OMP_TELEGRAM_STATE_DIR;
 const previousToken = process.env.TELEGRAM_BOT_TOKEN;
@@ -82,3 +82,138 @@ describe("daemon upgrades", () => {
     expect(readDaemonState()).toEqual({ pid: 9876, version: "0.2.0", startedAt: 1 });
   });
 });
+
+describe("daemon spawn preconditions (#68)", () => {
+  const enable = (): void => {
+    saveAccess({ ...defaultAccess(), enabled: true, topicsChat: "42" });
+    process.env.TELEGRAM_BOT_TOKEN = "token";
+  };
+  const spy = () => {
+    const calls: Array<{ executable: string; env?: NodeJS.ProcessEnv }> = [];
+    return {
+      calls,
+      spawn: ((executable, _args, options) => {
+        calls.push({ executable, env: options.env });
+        return { once: () => undefined, unref: () => {} };
+      }) as NonNullable<EnsureDaemonOptions["spawn"]>,
+    };
+  };
+
+  test("declines instead of spawning when another live process owns the poll lock", () => {
+    // The whole defect. Before the fix this spawned a child to discover the
+    // lock was taken, and because the child was `omp <script>` it became a
+    // session that claimed a permanent Telegram topic before exiting. Measured
+    // cost on one host: 83 topics.
+    enable();
+    const s = spy();
+    const result = ensureDaemon(() => {}, {
+      spawn: s.spawn,
+      runtime: () => "/usr/bin/bun",
+      lockOwner: () => ({ pid: 4242, startedAt: 1, name: "conductor" }),
+      alive: (pid) => pid === 4242,
+    });
+    expect(result).toBe("declined");
+    expect(s.calls).toEqual([]);
+  });
+
+  test("spawns when the lock holder is dead — a stale lock must not wedge it shut", () => {
+    enable();
+    const s = spy();
+    const result = ensureDaemon(() => {}, {
+      spawn: s.spawn,
+      runtime: () => "/usr/bin/bun",
+      lockOwner: () => ({ pid: 4242, startedAt: 1 }),
+      alive: () => false,
+    });
+    expect(result).toBe("spawned");
+    expect(s.calls).toHaveLength(1);
+  });
+
+  test("our own lock is not foreign", () => {
+    enable();
+    const s = spy();
+    expect(
+      ensureDaemon(() => {}, {
+        spawn: s.spawn,
+        runtime: () => "/usr/bin/bun",
+        lockOwner: () => ({ pid: process.pid, startedAt: 1 }),
+        alive: () => true,
+      }),
+    ).toBe("spawned");
+  });
+
+  test("declines rather than launching something that cannot run daemon.ts", () => {
+    // `process.execPath` inside the omp binary is omp itself, which ignores a
+    // script argument and boots an agent session. A spawn that cannot become a
+    // daemon must not be reported as one.
+    enable();
+    const s = spy();
+    const warnings: string[] = [];
+    const result = ensureDaemon((m) => warnings.push(m), {
+      spawn: s.spawn,
+      runtime: () => undefined,
+      lockOwner: () => undefined,
+    });
+    expect(result).toBe("declined");
+    expect(s.calls).toEqual([]);
+    expect(warnings.join(" ")).toContain("no bun/node runtime");
+  });
+
+  test("launches the resolved runtime, never process.execPath, and marks the child", () => {
+    enable();
+    const s = spy();
+    expect(
+      ensureDaemon(() => {}, { spawn: s.spawn, runtime: () => "/opt/bun/bin/bun", lockOwner: () => undefined }),
+    ).toBe("spawned");
+    expect(s.calls[0]?.executable).toBe("/opt/bun/bin/bun");
+    // The marker is the blast-radius cap: a child that somehow boots as a
+    // session still refuses to claim a topic.
+    expect(s.calls[0]?.env?.OMP_TELEGRAM_DAEMON_CHILD).toBe("1");
+  });
+
+  test("a stale-version daemon is still stopped before the lock is consulted", () => {
+    // Ordering matters: if the lock were checked first, a stale-version daemon
+    // holding its own lock would be immortal.
+    enable();
+    writeFileSync(statePath("daemon.json"), JSON.stringify({ pid: 9876, version: "0.1.1", startedAt: 1 }));
+    const s = spy();
+    let running = true;
+    const killed: number[] = [];
+    const result = ensureDaemon(() => {}, {
+      version: "0.2.0",
+      spawn: s.spawn,
+      runtime: () => "/usr/bin/bun",
+      lockOwner: () => undefined,
+      alive: () => running,
+      kill: (pid) => {
+        killed.push(pid);
+        running = false;
+      },
+    });
+    expect(killed).toEqual([9876]);
+    expect(result).toBe("spawned");
+  });
+});
+
+describe("resolveRuntime (#68)", () => {
+  test("rejects a host binary that only looks like a launcher", () => {
+    // The bug in one assertion: omp is not a runtime, so it must never be
+    // returned even though it is `process.execPath`.
+    expect(resolveRuntime({ PATH: "" }, "/root/.local/bin/omp")).toBeUndefined();
+  });
+
+  test("uses the host runtime when the host IS one", () => {
+    expect(resolveRuntime({ PATH: "" }, "/usr/local/bin/bun")).toBe("/usr/local/bin/bun");
+    expect(resolveRuntime({ PATH: "" }, "/usr/bin/node")).toBe("/usr/bin/node");
+  });
+
+  test("finds bun on PATH when the host is not a runtime", () => {
+    const bin = mkdtempSync(join(tmpdir(), "omp-tg-runtime-"));
+    try {
+      writeFileSync(join(bin, "bun"), "#!/bin/sh\n", { mode: 0o755 });
+      expect(resolveRuntime({ PATH: bin }, "/root/.local/bin/omp")).toBe(join(bin, "bun"));
+    } finally {
+      rmSync(bin, { recursive: true, force: true });
+    }
+  });
+});

src/daemon.ts

@@ -8,7 +8,7 @@ import {
   statSync,
   writeFileSync,
 } from "node:fs";
-import { join } from "node:path";
+import { delimiter, join } from "node:path";
 import {
   type Access,
   canAnswerPrompt,
@@ -18,7 +18,7 @@ import {
   resolveToken,
   statePath,
 } from "./access";
-import { acquireLock, type Logger, Poller, releaseLock, startLockHeartbeat, tg, webhookConflictHint } from "./api";
+import { acquireLock, type LockOwner, type Logger, Poller, readLockOwner, releaseLock, startLockHeartbeat, tg, webhookConflictHint } from "./api";
 import { type BridgeHost, ensureControlTopic, handleUpdate, syncBotCommands } from "./bridge";
 import { SpawnController } from "./control";
 import { TelegramPromptController } from "./prompts";
@@ -90,7 +90,7 @@ interface SpawnedDaemon {
 type SpawnDaemon = (
   executable: string,
   args: string[],
-  options: { detached: true; stdio: ["ignore", number, number] },
+  options: { detached: true; stdio: ["ignore", number, number]; env: NodeJS.ProcessEnv },
 ) => SpawnedDaemon;
 
 export interface EnsureDaemonOptions {
@@ -100,13 +100,63 @@ export interface EnsureDaemonOptions {
   sleep?: (ms: number) => void;
   now?: () => number;
   version?: string;
+  /** Resolves the JS runtime to launch the daemon with. Injected for tests. */
+  runtime?: () => string | undefined;
+  /** Reads the poll lock's owner without acquiring it. Injected for tests. */
+  lockOwner?: (lockPath: string) => LockOwner | undefined;
 }
 
-/** Ensure one current-version daemon is alive when topics-only routing permits it. */
+/** A path whose basename is a JS runtime that can execute a script argument. */
+const RUNTIME_NAME = /(?:^|\/)(?:bun|node)(?:-[\d.]+)?$/;
+
+/**
+ * The runtime that can actually execute `daemon.ts`.
+ *
+ * NOT `process.execPath` (#68). When this plugin is hosted inside the omp
+ * binary, `execPath` is *omp* — a compiled Bun executable that ignores a script
+ * argument and boots an interactive agent session instead. `omp daemon.ts` and
+ * `omp /nonexistent.ts` produce byte-identical output, so the daemon never ran;
+ * every "spawn" was a fresh session that claimed a Telegram topic and exited.
+ *
+ * So the runtime is resolved by name, and when there is none we refuse to spawn
+ * rather than launch something that cannot become a daemon.
+ */
+export function resolveRuntime(env: NodeJS.ProcessEnv = process.env, self: string = process.execPath): string | undefined {
+  // `self` counts only when it IS a runtime. Inside omp it is the agent, which
+  // is the whole bug.
+  if (RUNTIME_NAME.test(self)) return self;
+  // Otherwise resolve `bun` by name: this package ships unbuilt TypeScript, so
+  // Bun is the runtime that can execute it. `node` cannot, and is not a
+  // fallback — a spawn that cannot parse the entrypoint is the same silent
+  // no-op in a different costume.
+  for (const dir of (env.PATH ?? "").split(delimiter)) {
+    if (dir === "") continue;
+    const candidate = join(dir, "bun");
+    try {
+      if (statSync(candidate).isFile()) return candidate;
+    } catch {
+      // absent or unreadable: keep looking
+    }
+  }
+  return undefined;
+}
+
+/**
+ * Ensure one current-version daemon is alive when topics-only routing permits it.
+ *
+ * Returns `"declined"` when a daemon is neither possible nor needed: another
+ * process already owns the poll lock (an interactive session polls for itself),
+ * or no JS runtime can execute `daemon.ts`. Both are steady states, not
+ * failures, and both must be decided *here* — before spawning (#68). The old
+ * code spawned a child to find out, and since the child was `omp <script>` it
+ * became a session that claimed a Telegram topic before discovering it should
+ * exit. Each caller then re-ran the same experiment, one permanent topic at a
+ * time, forever.
+ */
 export function ensureDaemon(
   warn: (message: string) => void,
   options: EnsureDaemonOptions = {},
-): "alive" | "spawned" | "disabled" | "failed" {
+): "alive" | "spawned" | "disabled" | "declined" | "failed" {
   const reason = daemonDisableReason(loadAccess(warn), resolveToken());
   if (reason) return "disabled";
 
@@ -115,6 +165,8 @@ export function ensureDaemon(
   const sleep = options.sleep ?? ((ms) => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms));
   const now = options.now ?? Date.now;
   const version = options.version ?? packageVersion();
+  const runtime = options.runtime ?? resolveRuntime;
+  const lockOwner = options.lockOwner ?? readLockOwner;
   const spawnDaemon: SpawnDaemon = options.spawn ?? ((executable, args, spawnOptions) => spawn(executable, args, spawnOptions));
   const current = readDaemonState();
   if (current && daemonAlive(current, alive)) {
@@ -133,13 +185,34 @@ export function ensureDaemon(
     }
   }
 
+  // Read the lock; never spawn to discover it. A live foreign owner means
+  // something is already polling, and `runDaemon` would exit immediately —
+  // which is exactly the no-op that used to cost a topic per call. Checked
+  // AFTER the upgrade path above, so stopping a stale-version daemon still
+  // happens and its released lock is observed on the next call.
+  const owner = lockOwner(statePath("bot.lock"));
+  if (owner !== undefined && owner.pid !== process.pid && alive(owner.pid)) {
+    return "declined";
+  }
+
+  // No runtime, no spawn. Launching the host binary here is what produced the
+  // loop: it cannot run `daemon.ts` and silently becomes a session instead.
+  const executable = runtime();
+  if (executable === undefined) {
+    warn("no bun/node runtime on PATH to run the telegram daemon; inbound relies on a polling session");
+    return "declined";
+  }
+
   ensureStateDir();
   rotateDaemonLog();
   const logFd = openSync(statePath("daemon.log"), "a", 0o600);
   try {
-    const child = spawnDaemon(process.execPath, [join(import.meta.dirname, "daemon.ts")], {
+    const child = spawnDaemon(executable, [join(import.meta.dirname, "daemon.ts")], {
       detached: true,
       stdio: ["ignore", logFd, logFd],
+      // Declares what this child is, so a process that somehow boots as a
+      // session instead of a daemon still cannot claim a topic (#68).
+      env: { ...process.env, OMP_TELEGRAM_DAEMON_CHILD: "1" },
     });
     child.once("error", (err) => warn(`daemon process failed: ${String(err)}`));
     child.unref();

src/index.ts

@@ -966,6 +966,13 @@ export default function telegramExtension(pi: ExtensionAPI): void {
    */
   async function ensureTopic(ctx?: ExtensionContext): Promise<void> {
     if (!access.topicsChat || !token || ownTopic) return;
+    // Blast-radius cap (#68): a process launched to BE the daemon is not a
+    // conversation and must never own a human-visible topic. `ensureDaemon` now
+    // refuses to launch anything that cannot run `daemon.ts`, so this should be
+    // unreachable — it is here because the cost of being wrong was 83 permanent
+    // topics, and a marker the launcher sets is cheaper than trusting that the
+    // launcher is always right.
+    if (process.env.OMP_TELEGRAM_DAEMON_CHILD === "1") return;
     // DM host with forum-topic mode provably off: skip creation (createForumTopic
     // would just fail) and run untopiced with an actionable hint. Only an explicit
     // false blocks — undefined (older server / field absent) still attempts create.

src/topics.test.ts

@@ -288,3 +288,53 @@ describe("purgeRouteDir", () => {
     expect(() => purgeRouteDir(999)).not.toThrow();
   });
 });
+
+describe("registry writes survive concurrency (#68)", () => {
+  const entry = (pid: number): ThreadEntry => ({ pid, cwd: `/w/${pid}`, name: "conductor", claimedAt: 1_000 + pid });
+
+  test("concurrent claims from separate processes all persist", async () => {
+    // The measured failure: a burst that created 16 topics recorded 15 rows.
+    // Whole-file writes make every claim a read-modify-write, so unserialised
+    // the last writer wins and the rows in between are lost — and a topic whose
+    // row is lost is invisible to /cleanup forever, because the registry is the
+    // only index that exists. Losing the index is worse than losing the topic.
+    const runner = join(dir, "claim.ts");
+    writeFileSync(
+      runner,
+      `import { claimThread } from ${JSON.stringify(join(import.meta.dirname, "topics.ts"))};\n` +
+        `claimThread("42", Number(process.argv[2]), { pid: Number(process.argv[2]), cwd: "/w", name: "conductor", claimedAt: 1 });\n`,
+    );
+    const ids = [7001, 7002, 7003, 7004, 7005, 7006, 7007, 7008];
+    await Promise.all(
+      ids.map((id) =>
+        Bun.spawn([process.execPath, runner, String(id)], {
+          env: { ...process.env, OMP_TELEGRAM_STATE_DIR: dir },
+          stdout: "ignore",
+          stderr: "ignore",
+        }).exited,
+      ),
+    );
+    const got = Object.keys(loadRegistry().threads).map(Number).sort((a, b) => a - b);
+    expect(got).toEqual(ids);
+  });
+
+  test("a lock left by a dead process does not wedge the registry shut", () => {
+    // Self-healing by age. A mutation lock is held for microseconds, so one
+    // that is seconds old belonged to a process that died holding it.
+    const lock = `${statePath("threads.json")}.lock`;
+    writeFileSync(lock, "999999");
+    const old = new Date(Date.now() - 60_000);
+    utimesSync(lock, old, old);
+    claimThread("42", 8001, entry(8001));
+    expect(Object.keys(loadRegistry().threads)).toEqual(["8001"]);
+  });
+
+  test("the temp file is per-process, so two writers cannot publish each other's", () => {
+    // It was a shared `threads.json.tmp`: both writers wrote that one path and
+    // both renamed it, so one could publish the other's half-written file.
+    claimThread("42", 8002, entry(8002));
+    const leftovers = readdirSync(dir).filter((f) => f.startsWith("threads.json.tmp"));
+    expect(leftovers).toEqual([]);
+    expect(loadRegistry().threads["8002"]?.pid).toBe(8002);
+  });
+});