Single poller per Telegram token, and an honest return on a dropped send
Three defects in the same neighbourhood, all of the same kind: courier reporting that something happened which did not.
- Every leader-mode courier started its own Telegram poller. Telegram permits exactly ONE getUpdates consumer per bot token, so N concurrent sessions meant N pollers and a continuous 409 stream (three measured concurrently on 2026-08-04, one parented by a session alive since Jul 23).
New (courier claim): a filesystem claim scoped to a hash of the token and
to the relay dir. The claim is a SYMLINK whose target is the owner's pid,
so it is created atomically with its owner already recorded -- a
mkdir-plus-pidfile lock has a window where a SIGKILLed holder leaves an
ownerless claim nobody can ever attribute or safely break. sigil's
make-symlink swallows EEXIST (measured), so the win/lose verdict is read
back from the link rather than taken from the call. Stale claims are breakable: pid-liveness answers 'alive, 'dead or
'undetermined, never conflating "the owner is gone" with "I could not
ask". Each tier carries its own positive control, and a live pid whose
comm does not match ours is treated as a recycled pid, not a live owner. Losing the claim is normal and logged at info. The instance runs on as a
full MCP server: tools, relays, sends, notifications. No retry, no exit.- The dry-run path returned "Message sent." byte-for-byte. Every gagged send from 2026-07-06 to 2026-08-04 reported delivery it had not earned, and mimicked the silent-drop bug it was masking so exactly that production observation of that bug was impossible while the gag was set.
It now returns a string that cannot be read as success, and does not
raise (an MCP error invites the retry loop behind the send storm). It
also RELEASES the dedup key, because a dry-run is a definite
non-delivery: without that, the next real send of the same text hit the
suppress path and reported success while delivering nothing. Measured on
the pre-fix binary: gag on, gag off, same text -> 0 deliveries,
"Message sent.". Fixed: 1 delivery. The suppress path now says which of the two things happened rather than
reusing the plain success string.- Concurrent leader instances shared one courier.log. Append is emulated (read, truncate, write back), so two instances zero-filled each other's file; the resulting NUL bytes made grep report zero matches on a pattern tail showed plainly, mid-incident. Leader mode now writes courier-<pid>.log, keeping the ten newest. The --log path is deliberately untouched: claude-ops passes an already-unique per-task path there and reads it back by name.
Verification is behavioural throughout. repro/repro-single-poller.sh runs two real couriers against the mock; mock_telegram.py now models Telegram's single-consumer 409, because the previous mock could not produce the error the fix removes and "zero 409s" would have passed for free. The pre-fix binary produces 4 conflicts and 2 poller children in the same window where the fixed binary produces 0 and 1. Run against the pre-fix binary the harness turns 17 assertions red.
The storm repro keeps 1 send -> 1 delivery and gains an always-on positive control: the same binary with only the dedup WINDOW neutered delivers 4, proving the counter can see duplicates. A pre-dedup control binary cannot be built on the current toolchain (SGB v10 bundle, v9 runtime) and reported 0 deliveries because it never started -- which would have read as a result.
Suite 105 passed / 0 failed (baseline 82/0).
repro/mock_telegram.py | 106 ++++++++++++++++++++++++---
repro/repro-send-duplication.sh | 23 ++++++
repro/repro-single-poller.sh | 30 ++++++++
repro/single_poller_harness.py | 630 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
src/courier/claim.sgl | 278 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
src/courier/dedup.sgl | 22 ++++--
src/courier/main.sgl | 47 +++++++++++-
src/courier/poller.sgl | 22 +++++-
src/courier/telegram.sgl | 42 +++++++++--
test/test-claim.sgl | 185 +++++++++++++++++++++++++++++++++++++++++++++++
10 files changed, 1361 insertions(+), 24 deletions(-)repro/mock_telegram.pymodified
POST /_seed body: {"updates":[{...},...]} replace the backlog POST /_add body: {"text":"...","sender_id":"..","chat_id":".."} append one update GET /_stats {"delivered":[ids], "sends":N} POST /_dwell body: {"seconds":N} long-poll dwell (see below) GET /_stats {"delivered":[ids], "sends":N, "conflicts":N, ...}The SINGLE-CONSUMER rule (409 Conflict) is modelled too, because the"exactly one poller per token" fix cannot be verified against a mock thatis incapable of producing the error it is supposed to eliminate -- "zero409s" would then pass for free, on a mock, with the bug fully present.Telegram permits ONE getUpdates consumer per bot token: a getUpdates thatarrives while another is already in flight is answered 409 {"ok":false,"error_code":409, "description":"Conflict: terminated by other getUpdates request..."}and that is the error stream observed live on 2026-08-04 with threeconcurrent couriers.`dwell` is what makes this observable off-device. Real Telegram HOLDS anempty getUpdates open for the caller's `timeout` seconds, which is whatmakes two independent pollers overlap almost continuously. courier pollswith timeout=0, so a mock that answers instantly leaves an in-flightwindow of microseconds and two pollers would collide only by luck. Settinga dwell (e.g. 2.5s, longer than courier's 2s poll interval) restores theoverlap deterministically: * TWO pollers -> the second one's call always lands inside the first one's dwell -> conflicts climb. * ONE poller -> its calls are strictly sequential (poll, wait, poll), so it can never overlap ITSELF -> conflicts stay 0.Dwell defaults to 0 (env GETUPDATES_DWELL), so every pre-existing reprokeeps its original timing and behaviour.The sentinel-zero (2026-07-02) storm case is exercised by`repro-sentinel-zero.sh`: `_seed []` gives an EMPTY backlog so a cold start "sends": 0, # count of sendMessage calls "delivered_ids": [], # update_ids the mock RETURNED to a poller (per call) "send_mode": SEND_MODE_DEFAULT, # --- single-consumer (409) modelling; see module docstring --- "inflight": 0, # getUpdates calls currently being served "conflicts": 0, # getUpdates calls rejected with 409 "max_inflight": 0, # high-water mark, i.e. observed concurrent pollers "dwell": float(os.environ.get("GETUPDATES_DWELL", "0")),} self._reply({"delivered": STATE["delivered_ids"], "sends": STATE["sends"], "getupdates": STATE["getupdates"], "conflicts": STATE["conflicts"], "max_inflight": STATE["max_inflight"], "dwell": STATE["dwell"], "pending": [u["update_id"] for u in STATE["updates"]]}) return self._reply({"ok": False, "description": "not found"}, 404) STATE["getupdates"] = 0 STATE["sends"] = 0 STATE["delivered_ids"] = [] STATE["conflicts"] = 0 STATE["max_inflight"] = 0 self._reply({"ok": True}) return if path == "/_dwell": with LOCK: STATE["dwell"] = float(body.get("seconds", 0)) d = STATE["dwell"] logline(REQ_LOG, f"--- getUpdates dwell set to {d}s ---") self._reply({"ok": True, "dwell": d}) return if path == "/_add": with LOCK: uid = STATE["next_id"] offset = body.get("offset", 0) or 0 with LOCK: STATE["getupdates"] += 1 if offset > 0: # Confirm: drop everything below the offset. STATE["updates"] = [u for u in STATE["updates"] if u["update_id"] >= offset] result = list(STATE["updates"]) ids = [u["update_id"] for u in result] STATE["delivered_ids"].append({"offset": offset, "returned": ids}) logline(REQ_LOG, f"getUpdates offset={offset} -> returned={ids}") self._reply({"ok": True, "result": result}) # Single-consumer rule: a getUpdates arriving while another # is in flight is a second poller on this token. Real # Telegram terminates one of them with 409; we reject the # newcomer, which is equivalent for counting purposes. if STATE["inflight"] > 0: STATE["conflicts"] += 1 STATE["max_inflight"] = max(STATE["max_inflight"], STATE["inflight"] + 1) n = STATE["conflicts"] conflict = True else: STATE["inflight"] += 1 STATE["max_inflight"] = max(STATE["max_inflight"], STATE["inflight"]) conflict = False if conflict: logline(REQ_LOG, f"getUpdates offset={offset} -> 409 CONFLICT (#{n})") self._reply({"ok": False, "error_code": 409, "description": "Conflict: terminated by other " "getUpdates request; make sure that " "only one bot instance is running"}, code=409) return try: with LOCK: if offset > 0: # Confirm: drop everything below the offset. STATE["updates"] = [u for u in STATE["updates"] if u["update_id"] >= offset] result = list(STATE["updates"]) ids = [u["update_id"] for u in result] STATE["delivered_ids"].append({"offset": offset, "returned": ids}) dwell = STATE["dwell"] # Long-poll emulation: hold an EMPTY response open, exactly # as Telegram holds the caller's `timeout`. This is what makes # two independent pollers overlap observably; see the module # docstring. A non-empty result returns immediately, as # Telegram does. if dwell > 0 and not result: time.sleep(dwell) logline(REQ_LOG, f"getUpdates offset={offset} -> returned={ids}") self._reply({"ok": True, "result": result}) finally: with LOCK: STATE["inflight"] -= 1 return if method == "sendMessage":repro/repro-send-duplication.shmodified
stop_mock() { kill "$MOCK" 2>/dev/null; wait "$MOCK" 2>/dev/null; sleep 0.2; }# run_case <label> <bin> <expected-deliveries> <driver-args...># Honours DEDUP_WINDOW: when set, exports COURIER_SEND_DEDUP_WINDOW for the# driver's couriers. Used by the positive control below.run_case() { local label="$1" bin="$2" expect="$3"; shift 3 start_mock COURIER_SEND_DEDUP_WINDOW="${DEDUP_WINDOW:-}" \ "$PY" "$HERE/send_driver.py" --bin "$bin" \ --api-url "http://127.0.0.1:$PORT" \ --stats-url "http://127.0.0.1:$PORT/_stats" \run_case "STORM x3 -> 1 (exactly-once)" "$BIN" 1 --mode delay:8 --tool-timeout 3 --retries 3 --text storm3run_case "STORM x5 -> 1 (exactly-once)" "$BIN" 1 --mode delay:8 --tool-timeout 2 --retries 5 --text storm5# ---------------------------------------------------------------------------# POSITIVE CONTROL (always runs).## The four cases above are all assertions that something did NOT happen --# no duplicate delivery -- and an absence proves nothing until the instrument# has been shown to detect the thing when it IS present. A count of 1 is# returned just as cheerfully by a courier that never started as by a working# one. (Measured 2026-08-04: a control binary built from the pre-dedup commit# reported 0 deliveries because it could not run at all -- SGB v10 bundle# against a v9 runtime -- and 0 would have read as a very convincing result.)## So neuter an INPUT and nothing else: keep the same binary, the same driver,# the same scoring, and shrink only the dedup WINDOW so the recorded key# expires between storm attempts. The suppression should disappear and the# same run should deliver once per attempt.# ---------------------------------------------------------------------------echo "=== POSITIVE CONTROL: same binary, dedup window neutered to 1s ==="DEDUP_WINDOW=1 run_case "window=1s STORM x3 -> 4 (counter CAN see duplicates)" \ "$BIN" 4 --mode delay:8 --tool-timeout 3 --retries 3 --text storm-window-controlif [ -n "${COURIER_CONTROL_BIN:-}" ] && [ -x "${COURIER_CONTROL_BIN}" ]; then echo "=== CONTROL (pre-fix) binary: $COURIER_CONTROL_BIN ===" # The pre-fix binary re-delivers once per attempt: 1 logical send + 3repro/repro-single-poller.shadded
#!/usr/bin/env bash# Entry point for the single-poller + honest-send harness.## Everything runs against repro/mock_telegram.py. NOTHING reaches real# Telegram: every courier started here has COURIER_TELEGRAM_API_URL pointed# at the local mock, which both the send and the poll path honour.## The pre-fix CONTROL is not optional in spirit: case 1 asserts ZERO 409s,# and an absence assertion is worth nothing until the instrument has been# shown to detect the thing when it IS present. Build one with:## git worktree add /tmp/courier-control <pre-fix-sha># (cd /tmp/courier-control && sigil deps install && sigil build)# COURIER_CONTROL_BIN=/tmp/courier-control/build/dev/bin/courier \# repro/repro-single-poller.sh## Without it the harness reports the control as FAILED rather than quietly# passing, because "we did not check" must not read as "it is fine".set -uHERE="$(cd "$(dirname "$0")" && pwd)"BIN="${COURIER_BIN:-$HERE/../build/dev/bin/courier}"PY="${PYTHON:-python3}"if [ ! -x "$BIN" ]; then echo "FATAL: courier binary not found at $BIN (run 'sigil build' first)" exit 2fiexec "$PY" "$HERE/single_poller_harness.py" --bin "$BIN" "$@"repro/single_poller_harness.pyadded
#!/usr/bin/env python3"""Off-device harness for the SINGLE-POLLER claim and the HONEST-SEND return.Everything here runs against repro/mock_telegram.py. NOTHING can reach realTelegram: every courier instance is started with COURIER_TELEGRAM_API_URLpointed at the local mock, which is the only endpoint the send AND poll pathsuse once that variable is set.What it proves, in the order the cases run: 1 ONE POLLER two leader-mode couriers on one token -> exactly one poller child process, and zero 409s at the mock. 2 409 DETECTOR WORKS the same two-instance run with the PRE-FIX binary (COURIER_CONTROL_BIN) -> 409s appear. This is the positive control: without it, case 1's "zero 409s" is a claim of absence made by an instrument never shown capable of detecting presence. 3 NON-POLLER IS FINE the instance that lost the claim still serves its MCP tools -- create-relay / list-relays / close-relay. 4 SIGKILL RECOVERY SIGKILL (not a graceful stop) the claim holder, start a fresh instance: it detects the dead owner, breaks the stale claim, and polls. 5 DRY-RUN IS HONEST with COURIER_DISABLE_TELEGRAM_SEND=1 the tool returns a string that cannot be read as success, does not raise, and delivers nothing at the mock. 6 GAG DOESN'T POISON a dry-run must not leave a dedup key behind that would suppress a LATER real send of the same text. Dry-run, then flag off, then the same message must actually deliver. 7 LOG HYGIENE two concurrent leader instances write two distinct log files, and grep works on both (the shared-log interleaving produced NUL bytes, which made grep silently report zero matches during an incident).Usage: single_poller_harness.py --bin <courier> [--control-bin <pre-fix courier>]"""import argparseimport jsonimport osimport reimport shutilimport signalimport subprocessimport sysimport threadingimport timeimport urllib.errorimport urllib.requestCHAT_ID = "331005009"PORT = int(os.environ.get("PORT", "8613"))BASE = f"http://127.0.0.1:{PORT}"# Longer than courier's 2s poll interval, so two independent pollers overlap# deterministically rather than by luck. See mock_telegram.py's docstring.DWELL = 2.5# How long to watch a two-instance run. Must be long enough that the pre-fix# control reliably produces 409s inside it, or "zero 409s" from the fixed# binary would only mean "we did not watch for long enough".OBSERVE_WINDOW = 20FAILED = []PASSED = []def check(label, ok, detail=""): (PASSED if ok else FAILED).append(label) print(f" {'PASS' if ok else 'FAIL'} {label}" + (f": {detail}" if detail else "")) return ok# ----------------------------------------------------------------- mock ----def post(path, obj): data = json.dumps(obj).encode() req = urllib.request.Request(BASE + path, data=data, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=10) as r: return json.load(r)def stats(): with urllib.request.urlopen(BASE + "/_stats", timeout=10) as r: return json.load(r)class Mock: def __init__(self, logdir): self.logdir = logdir here = os.path.dirname(os.path.abspath(__file__)) os.makedirs(logdir, exist_ok=True) self.proc = subprocess.Popen( [sys.executable, os.path.join(here, "mock_telegram.py"), str(PORT), os.path.join(logdir, "mocklog")], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(50): try: stats() return except Exception: time.sleep(0.1) raise RuntimeError("mock did not come up") def stop(self): self.proc.terminate() try: self.proc.wait(timeout=5) except Exception: self.proc.kill()# -------------------------------------------------------------- courier ----class Courier: """One courier subprocess speaking MCP over stdio.""" def __init__(self, bin_path, run_dir, name, disable_send=False, extra_env=None, argv=("serve",)): self.name = name self.run_dir = run_dir self.notifications = [] os.makedirs(run_dir, exist_ok=True) env = dict(os.environ) env["COURIER_TELEGRAM_TOKEN"] = "TESTTOKEN-single-poller" env["COURIER_TELEGRAM_CHAT_ID"] = CHAT_ID env["COURIER_TELEGRAM_API_URL"] = BASE # Shared state dir: this is what makes the instances contend, exactly # as concurrent leader sessions share ~/.courier. env["COURIER_RELAY_DIR"] = os.path.join(run_dir, "relays") if disable_send: env["COURIER_DISABLE_TELEGRAM_SEND"] = "1" else: env.pop("COURIER_DISABLE_TELEGRAM_SEND", None) if extra_env: env.update(extra_env) self.errf = open(os.path.join(run_dir, f"{name}.stderr"), "wb") # NOTE: no --log. That is deliberate -- it exercises the leader-mode # persistent-log path (configure-persistent-log!), which is the one # that used to collide on a single shared courier.log. self.proc = subprocess.Popen( [bin_path, *argv], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.errf, env=env, bufsize=0) self._id = 0 self._pending = {} self._lock = threading.Lock() self._reader = threading.Thread(target=self._read_loop, daemon=True) self._reader.start() # --- MCP plumbing --- def _read_loop(self): for raw in self.proc.stdout: try: msg = json.loads(raw) except Exception: continue rid = msg.get("id") if rid is not None: with self._lock: ev = self._pending.get(rid) if ev: ev[1] = msg ev[0].set() else: # Server-initiated notification (channel events: relay and # Telegram messages arrive this way). with self._lock: self.notifications.append(msg) def _send(self, obj): self.proc.stdin.write((json.dumps(obj) + "\n").encode()) self.proc.stdin.flush() def rpc(self, method, params, timeout=20): with self._lock: self._id += 1 rid = self._id ev = [threading.Event(), None] self._pending[rid] = ev self._send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}) if not ev[0].wait(timeout): raise TimeoutError(f"{self.name}: no response to {method}") return ev[1] def initialize(self): res = self.rpc("initialize", { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "single-poller-harness", "version": "0"}}) self._send({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}) return res def call(self, tool, arguments, timeout=30): res = self.rpc("tools/call", {"name": tool, "arguments": arguments}, timeout=timeout) if "result" in res: try: return res["result"]["content"][0]["text"] except Exception: return json.dumps(res["result"]) return "ERROR:" + json.dumps(res.get("error", res)) # --- lifecycle --- @property def pid(self): return self.proc.pid def alive(self): return self.proc.poll() is None def sigkill(self): """SIGKILL, as a /mcp teardown does. No chance to release a claim.""" os.kill(self.proc.pid, signal.SIGKILL) self.proc.wait(timeout=10) def close_stdin(self): """Graceful shutdown: stdin EOF is how a session ends.""" try: self.proc.stdin.close() except Exception: pass def kill(self): try: if self.alive(): self.proc.kill() self.proc.wait(timeout=5) except Exception: pass try: self.errf.close() except Exception: pass def stderr_text(self): try: self.errf.flush() except Exception: pass with open(os.path.join(self.run_dir, f"{self.name}.stderr"), "rb") as f: return f.read().decode("utf-8", "replace") def log_path(self): """This instance's own log file. Leader mode with no --log logs to <state-dir>/courier-<pid>.log, NOT to stderr, so assertions about log lines must read this. """ return os.path.join(self.run_dir, f"courier-{self.pid}.log") def log_text(self): try: with open(self.log_path(), "rb") as f: return f.read().decode("utf-8", "replace") except FileNotFoundError: return ""# ------------------------------------------------------- process helpers ----def child_pids(pid): """Direct children of `pid`, read straight from /proc. Deliberately not pgrep/pkill: a name-matching pattern here could match this harness or an unrelated courier on the host, and a missing tool would return an empty list that reads exactly like 'no children'. """ out = [] for entry in os.listdir("/proc"): if not entry.isdigit(): continue try: with open(f"/proc/{entry}/status") as f: ppid = None for line in f: if line.startswith("PPid:"): ppid = int(line.split()[1]) break if ppid == pid: out.append(int(entry)) except (FileNotFoundError, ProcessLookupError, PermissionError): continue return outdef poller_children(pid): """Children of `pid` that are courier poller processes.""" found = [] for c in child_pids(pid): try: with open(f"/proc/{c}/cmdline", "rb") as f: cmd = f.read().decode("utf-8", "replace") except (FileNotFoundError, ProcessLookupError, PermissionError): continue if "--telegram-poller" in cmd: found.append(c) return founddef claim_files(state_dir): if not os.path.isdir(state_dir): return [] return sorted(n for n in os.listdir(state_dir) if n.startswith("telegram-poller-") and n.endswith(".claim"))def claim_owner(state_dir): files = claim_files(state_dir) if not files: return None return os.readlink(os.path.join(state_dir, files[0]))def wait_for(predicate, timeout=30, interval=0.25): deadline = time.time() + timeout while time.time() < deadline: if predicate(): return True time.sleep(interval) return False# ----------------------------------------------------------------- cases ----def case_one_poller(binp, run_dir, label="1 ONE POLLER"): """Two leader-mode couriers, one token. Returns (a, b, state_dir).""" print(f"=== {label} ===") post("/_seed", {"updates": []}) post("/_dwell", {"seconds": DWELL}) state_dir = run_dir a = Courier(binp, run_dir, "inst-a") a.initialize() got_a = wait_for(lambda: len(poller_children(a.pid)) == 1, timeout=40) # POSITIVE CONTROL for the child-detector: if this is False, every later # "0 pollers" reading below is meaningless. check("instance A starts a poller child (detector positive control)", got_a, f"children={poller_children(a.pid)}") b = Courier(binp, run_dir, "inst-b") b.initialize() # Give B every chance to (wrongly) start one, and give the mock a window # comfortably longer than the one in which the pre-fix control produces # its first 409. time.sleep(OBSERVE_WINDOW) pa, pb = poller_children(a.pid), poller_children(b.pid) check("exactly one poller child across both instances", len(pa) + len(pb) == 1, f"A={pa} B={pb}") check("instance B did NOT start a poller", len(pb) == 0, f"B={pb}") owner = claim_owner(state_dir) check("claim file records the polling instance's pid", owner == str(a.pid), f"claim owner={owner} A={a.pid} B={b.pid}") log = b.log_text() skip_lines = [l for l in log.splitlines() if "poller claim" in l] check("instance B logs the skip, at INFO and not WARN/ERROR", (len(skip_lines) == 1 and "[INFO]" in skip_lines[0] and "another courier instance holds" in skip_lines[0]), repr(skip_lines)) s = stats() # An absence is only meaningful if the thing had a chance to happen: # zero 409s across zero polls would pass for free. Assert the window # actually contained polling before believing the zero. check("the observation window contained real polling (denominator)", s["getupdates"] >= 3, f"getupdates={s['getupdates']} over {OBSERVE_WINDOW}s") check("zero 409 conflicts at the mock", s["conflicts"] == 0, f"conflicts={s['conflicts']} getupdates={s['getupdates']} " f"max_inflight={s['max_inflight']}") check("the mock never saw two concurrent pollers", s["max_inflight"] <= 1, f"max_inflight={s['max_inflight']}") return a, b, state_dirdef case_control(control_bin, run_dir): """Positive control: the pre-fix binary MUST produce 409s.""" print("=== 2 409 DETECTOR WORKS (pre-fix control) ===") if not control_bin: print(" SKIP no COURIER_CONTROL_BIN given -- case 1's 'zero 409s' " "is UNCONTROLLED and proves nothing on its own") FAILED.append("409 positive control not run") return post("/_seed", {"updates": []}) post("/_dwell", {"seconds": DWELL}) # Confirm the control really IS the pre-fix code before drawing any # conclusion from it. A control that silently pointed at the fixed binary # would report "no 409s" and look like a disproof of the whole bug. probe = Courier(control_bin, os.path.join(run_dir, "probe"), "ctl-probe", disable_send=True) probe.initialize() old = probe.call("send-message", {"text": "control identity probe", "to": CHAT_ID}) check("control binary is genuinely PRE-FIX (dry-run still lies)", old == "Message sent.", repr(old)) probe.kill() a = Courier(control_bin, run_dir, "ctl-a") a.initialize() b = Courier(control_bin, run_dir, "ctl-b") b.initialize() t0 = time.time() ok = wait_for(lambda: stats()["conflicts"] > 0, timeout=60) first = time.time() - t0 # Keep watching for the SAME window the fixed run got, so the two numbers # are comparable rather than one being a shorter look than the other. time.sleep(OBSERVE_WINDOW) s = stats() check("pre-fix binary: two instances DO produce 409s", ok, f"conflicts={s['conflicts']} in {OBSERVE_WINDOW}s " f"(first after {first:.1f}s) getupdates={s['getupdates']} " f"max_inflight={s['max_inflight']}") check("pre-fix binary: two poller children exist", len(poller_children(a.pid)) + len(poller_children(b.pid)) == 2, f"A={poller_children(a.pid)} B={poller_children(b.pid)}") for c in (a, b): c.kill() time.sleep(1)def case_relay_tools(binp, b, run_dir): """The instance that LOST the claim must be a fully working leader. Create a relay, connect a real worker-mode courier to it, send through it, and close it. This also re-checks that worker mode is untouched: three live worker relays depend on it. """ print("=== 3 NON-POLLER IS FINE (relay tools on the skipped instance) ===") r = b.call("create-relay", {"name": "harness-relay"}) check("non-polling instance: create-relay works", r == "Relay 'harness-relay' created and listening.", repr(r)) w = Courier(binp, run_dir, "worker", argv=("--relay", "harness-relay")) try: w.initialize() text = "leader-to-worker over the relay" connected = wait_for( lambda: "harness-relay" in b.call("list-relays", {}) and "connected" in b.call("list-relays", {}).lower(), timeout=20) check("non-polling instance: list-relays shows the connected worker", connected, repr(b.call("list-relays", {}))) r = b.call("send-message", {"text": text, "to": "harness-relay"}) check("non-polling instance: send-message over the relay works", r == "Message sent.", repr(r)) got = wait_for(lambda: any(text in json.dumps(n) for n in list(w.notifications)), timeout=20) check("the worker actually RECEIVED the relayed message", got, repr(w.notifications[-1] if w.notifications else None)) back = w.call("send-message", {"text": "worker-to-leader ack"}) check("worker mode: send-message back to the leader works", back == "Message sent to leader.", repr(back)) finally: w.kill() r = b.call("close-relay", {"name": "harness-relay"}) check("non-polling instance: close-relay works", r == "Relay 'harness-relay' closed.", repr(r))def case_sigkill(binp, run_dir, a, state_dir): print("=== 4 SIGKILL RECOVERY ===") dead_pid = a.pid a.sigkill() time.sleep(1) owner = claim_owner(state_dir) check("claim SURVIVES the SIGKILL (nothing released it)", owner == str(dead_pid), f"owner={owner} killed={dead_pid}") c = Courier(binp, run_dir, "inst-c") c.initialize() got = wait_for(lambda: len(poller_children(c.pid)) == 1, timeout=45) check("a fresh instance breaks the stale claim and polls", got, f"children={poller_children(c.pid)}") check("claim now records the new instance", claim_owner(state_dir) == str(c.pid), f"owner={claim_owner(state_dir)} C={c.pid}") log = c.log_text() check("the break is logged", "Breaking stale" in log, repr([l for l in log.splitlines() if "stale" in l][:2])) return cEXPECTED_DRY_RUN = ("Message NOT sent: COURIER_DISABLE_TELEGRAM_SEND is set " "(dry-run, nothing delivered).")def case_dry_run(binp, run_dir): print("=== 5 DRY-RUN IS HONEST ===")Showing the first 500 of 631 diff lines for this file. This diff is INCOMPLETE; read the file or clone the repository for the rest.
src/courier/claim.sgladded
;;; (courier claim) - Single-poller claim for the Telegram getUpdates stream.;;;;;; Telegram allows exactly ONE getUpdates consumer per bot token. Every;;; leader-mode courier used to start its own poller child, so N concurrent;;; Claude sessions meant N pollers fighting over one token and a continuous;;; stream of `Conflict: terminated by other getUpdates request (409)`.;;; Measured 2026-08-04: three pollers at once, one parented by a session;;; alive since Jul 23.;;;;;; This module hands out a cooperative, filesystem-backed claim so exactly;;; one instance polls. Everything else about a non-polling instance keeps;;; working: MCP tools, relays, sends, notifications. LOSING THE CLAIM IS THE;;; NORMAL CASE for the 2nd..Nth session and is logged at info, never warn.;;;;;; Why a lockfile rather than configuration: config drifts, gets copied;;; between machines, and its failure mode is silent -- two machines both;;; believing they are the designated poller looks exactly like no config at;;; all. The filesystem claim is derived from the running state instead.;;;;;; ---------------------------------------------------------------------;;; Why a SYMLINK is the claim;;; ---------------------------------------------------------------------;;; symlink(2) is atomic against a concurrent creator AND carries a payload;;; chosen at creation time. Putting the owner's pid in the link TARGET means;;; there is never a moment where the claim exists but its owner is unknown.;;; The mkdir-plus-pidfile alternative has exactly that window: a holder;;; SIGKILLed between the mkdir and the pid write leaves an ownerless claim;;; that no later instance can ever attribute or safely break -- a permanent;;; estate-wide deadlock, which is the specific outcome this design is;;; required to avoid.;;;;;; MEASURED QUIRK, load-bearing: sigil's `make-symlink` SWALLOWS EEXIST. It;;; neither raises nor returns a failure indicator when the path is already;;; taken (probed 2026-08-04: second call over an existing link returns;;; normally, and the link still points at the first owner). So the win/lose;;; decision must NEVER be read from the call itself -- it comes from reading;;; the link back and checking whose pid is in it. A `guard` around;;; make-symlink would look like a check and would be inert.;;;;;; ---------------------------------------------------------------------;;; Stale claims;;; ---------------------------------------------------------------------;;; A courier killed with SIGKILL cannot release its claim, and SIGKILL is;;; routine here (an MCP reconnect tears the server down that way). So a claim;;; whose owner is gone must be breakable, or the first crash would silence;;; Telegram for the whole estate forever. `pid-liveness` answers that, and it;;; is careful to distinguish three states, not two:;;;;;; 'alive -- the owner is running; do not touch the claim;;; 'dead -- the owner is gone; the claim may be broken;;; 'undetermined -- WE COULD NOT ASK. Not the same fact as 'dead, and;;; conflating them is how a probe silently degrades.;;;;;; Each tier carries its own positive control, so a broken instrument;;; reports 'undetermined instead of a confident wrong answer.(define-library (courier claim) (import (sigil core) (sigil string) (sigil io) (sigil fs) (sigil path) (sigil time) (sigil process) (sigil log) (courier dedup) (courier relay)) (export acquire-poller-claim! release-poller-claim! poller-claim-path claim-token-hash read-claim-owner pid-liveness pid-liveness/comm proc-comm) (begin ;; After breaking a stale claim, two instances can both have decided the ;; same dead pid was stale and both re-created the link. The link ends up ;; holding exactly one pid; re-reading after a short settle lets the loser ;; notice. Small enough to be invisible at session start, long enough to ;; cover the unlink/symlink pair on any real filesystem. (define *claim-settle* 0.5) ;; ============================================================ ;; Claim path ;; ============================================================ ;;; Short, non-reversible tag for a bot token. The token itself is NEVER ;;; written anywhere -- not to the claim, not to a log. 48 bits of a ;;; non-cryptographic FNV-1a hash is deliberately lossy: it is enough to ;;; keep two different bots on one machine from colliding, and far too ;;; little to reconstruct a token from. (define (claim-token-hash token) (: string? -> string?) (let* ((h (fnv1a-hex token)) (n (string-length h)) (padded (if (< n 16) (string-append (make-string (- 16 n) #\0) h) h))) (substring padded 0 12))) ;;; Where the claim for `token` lives. Scoped to the token (two bots on ;;; one machine do not collide) AND to the relay dir (so a test courier ;;; with its own COURIER_RELAY_DIR never contends with the live one). (define (poller-claim-path token) (: string? -> string?) (path-join (path-dirname (default-relay-dir)) (string-append "telegram-poller-" (claim-token-hash token) ".claim"))) ;; ============================================================ ;; Liveness ;; ============================================================ ;;; First line of /proc/<pid>/comm, or #f if it cannot be read. ;;; ;;; MEASURED: procfs files report st_size 0, so `read-file-string` (which ;;; reads stat's size) returns "" for every one of them -- a confident, ;;; wrong empty answer. A PORT read is size-independent and returns the ;;; real content. (define (proc-comm pid-str) (: string? -> any?) (guard (e (else #f)) (let ((line (call-with-input-file (string-append "/proc/" pid-str "/comm") (lambda (port) (read-line port))))) (and (string? line) (string-trim line))))) ;;; Is the process holding a claim still running? ;;; ;;; Returns 'alive, 'dead, or 'undetermined. See the header: ;;; 'undetermined means the instrument could not answer, which is a ;;; different fact from "the process is gone" and must not be treated as ;;; permission to break the claim. (define (pid-liveness pid-str) (: string? -> symbol?) (pid-liveness/comm pid-str (proc-comm (number->string (process-id))))) ;;; pid-liveness with the expected command name passed in explicitly. ;;; ;;; Split out so the recycled-pid discriminator can be tested from both ;;; sides: a live process whose comm MATCHES must read 'alive, and the ;;; same live process with a non-matching comm must read 'dead. Testing ;;; only the second would be satisfied by an implementation that always ;;; answers 'dead -- which is the pre-fix behaviour wearing a lock's ;;; clothes, and would break every claim it ever saw. (define (pid-liveness/comm pid-str expected-comm) (: string? any? -> symbol?) (let ((self (number->string (process-id)))) (cond ;; Tier 1: /proc. POSITIVE CONTROL -- our own pid must be visible ;; through the instrument before its answer about another pid means ;; anything. Without this, a host with no /proc reports every pid ;; dead and every claim breakable. ((file-exists? (string-append "/proc/" self)) (if (not (file-exists? (string-append "/proc/" pid-str))) 'dead ;; The pid exists -- but pids are recycled, and a recycled pid ;; would make a dead courier's claim look alive forever. Compare ;; the command name against OUR OWN (passed in, read at runtime ;; rather than hardcoded, so it cannot drift if the binary is ;; renamed). If either name is unreadable we cannot rule out a ;; live courier, so we say 'alive and leave the claim alone -- ;; the conservative direction. (let ((theirs (proc-comm pid-str))) (cond ((or (not expected-comm) (not theirs)) 'alive) ((string=? expected-comm theirs) 'alive) (else 'dead))))) ;; Tier 2: kill -0, for a host with no /proc. Same shape: prove the ;; instrument works on a pid known to be alive (ours) before ;; believing what it says about another. MEASURED exit codes: ;; 0 = alive, 1 = no such process, 127 = `kill` not found. Reading ;; 127 as "dead" is precisely the always-degrade failure this guards. ((and (command-exists? "kill") (eqv? 0 (guard (e (else #f)) (process-run "kill" "-0" self)))) (let ((rc (guard (e (else #f)) (process-run "kill" "-0" pid-str)))) (cond ((eqv? rc 0) 'alive) ((eqv? rc 1) 'dead) (else 'undetermined)))) (else 'undetermined)))) ;; ============================================================ ;; Acquire / release ;; ============================================================ ;; Path of the claim this process currently owns, or #f. (define *held-claim* (list #f)) ;;; Owner pid recorded in the claim, as a string, or #f when the claim is ;;; absent or unreadable. Uses `symlink?` (lstat) rather than ;;; `file-exists?` (stat): a claim link never has a target that exists as ;;; a file, so `file-exists?` reports #f for a perfectly valid claim. (define (read-claim-owner path) (: string? -> any?) (guard (e (else #f)) (and (symlink? path) (let ((target (read-symlink path))) (and (string? target) (string-trim target)))))) ;; Attempt the atomic create. Returns 'acquired or 'lost. The verdict ;; comes from reading the link back -- see the header note on ;; make-symlink swallowing EEXIST. (define (try-claim! path me) (guard (e (else 'lost)) (make-symlink me path) (if (equal? (read-claim-owner path) me) 'acquired 'lost))) ;;; Try to become the single poller for `token`. ;;; ;;; Returns: ;;; 'acquired -- this process polls, and must call ;;; release-poller-claim! on clean shutdown ;;; 'held-by-other -- another live courier polls. NORMAL. Log at info. ;;; 'undetermined -- liveness could not be established; do not poll, ;;; and do not break the existing claim. (define (acquire-poller-claim! token) (: string? -> symbol?) (let* ((path (poller-claim-path token)) (me (number->string (process-id)))) (guard (e (else (log-warn "Poller claim failed" error: (format "~a" e)) 'undetermined)) (ensure-directory (path-dirname path)) (if (eq? (try-claim! path me) 'acquired) (begin (set-car! *held-claim* path) 'acquired) (let ((owner (read-claim-owner path))) (cond ;; Present but unreadable: leave it alone rather than break ;; something we cannot attribute. ((not owner) 'held-by-other) ;; Our own pid already in the link (a re-entry): keep it. ((string=? owner me) (set-car! *held-claim* path) 'acquired) (else (case (pid-liveness owner) ((alive) 'held-by-other) ((undetermined) (log-warn "Poller claim owner liveness undetermined" owner-pid: owner) 'undetermined) (else (break-stale-claim! path owner me)))))))))) ;; Owner is gone: remove the claim and try once to take it. Deliberately ;; ONE retry -- a loop here would be the retry-against-a-dead-peer shape ;; that caused the send storm. (define (break-stale-claim! path owner me) (log-info "Breaking stale Telegram poller claim (owner gone)" owner-pid: owner) (guard (e (else #f)) (delete-file path)) (if (not (eq? (try-claim! path me) 'acquired)) 'held-by-other (begin ;; Two instances can break the SAME stale claim at the same ;; instant and both create the link; the file then holds exactly ;; one pid. Re-read after a settle so the loser finds out. (sleep *claim-settle*) (if (equal? (read-claim-owner path) me) (begin (set-car! *held-claim* path) 'acquired) 'held-by-other)))) ;;; Release the claim held by this process, if any. Safe to call when no ;;; claim is held. Only removes a claim still recording OUR pid, so a ;;; late shutdown cannot delete a claim another instance has since taken. (define (release-poller-claim!) (: -> void?) (let ((path (car *held-claim*))) (when path (set-car! *held-claim* #f) (guard (e (else #f)) (when (equal? (read-claim-owner path) (number->string (process-id))) (delete-file path) (log-info "Released Telegram poller claim"))))))))src/courier/dedup.sglmodified
send-dedup-window send-dedup-path send-dedup-key fnv1a-hex dedup-check-and-record! dedup-unrecord! ;; exported for tests (define *fnv-prime* 1099511628211) (define *fnv-mod* 18446744073709551616) ;; 2^64 (define (send-dedup-key chat-id text) (: (any-of integer? string?) string? -> string?) (let* ((s (string-append (if (string? chat-id) chat-id (number->string chat-id)) ":" text)) (n (string-length s))) ;;; FNV-1a 64-bit hash of a string, as lowercase hex (no leading-zero ;;; padding). Shared with (courier claim), which tags the poller claim ;;; file with a hash of the bot token so the token itself is never ;;; written to disk. (define (fnv1a-hex s) (: string? -> string?) (let ((n (string-length s))) (let loop ((i 0) (h *fnv-offset*)) (if (>= i n) (number->string h 16) *fnv-prime*) *fnv-mod*)))))) (define (send-dedup-key chat-id text) (: (any-of integer? string?) string? -> string?) (fnv1a-hex (string-append (if (string? chat-id) chat-id (number->string chat-id)) ":" text))) ;; ---- File-backed store: list of (key-string . epoch-seconds) ---- ;; Parse one "epoch hash" line into (hash . epoch), or #f if malformed.src/courier/main.sglmodified
(flush-output-port port)) port))) ;; How many courier-<pid>.log files to keep in the state dir. (define *log-keep-generations* 10) ;; Remove all but the newest *log-keep-generations* courier-<pid>.log ;; files. Best-effort and never fatal: a failure here costs disk, not ;; correctness. A live instance's file has a recent mtime and is ;; therefore kept; even if one were removed, Unix keeps the open fd ;; valid, so the writer cannot be broken by this. (define (prune-instance-logs! log-dir) (guard (e (else #f)) (let* ((entries (or (directory-list log-dir) '())) (logs (filter (lambda (n) (and (string-prefix? "courier-" n) (string-suffix? ".log" n))) entries)) (paths (map (lambda (n) (path-join log-dir n)) logs)) (sorted (sort paths (lambda (a b) (> (or (file-mtime a) 0) (or (file-mtime b) 0)))))) (let loop ((rest sorted) (i 0)) (unless (null? rest) (when (>= i *log-keep-generations*) (guard (e (else #f)) (delete-file (car rest)))) (loop (cdr rest) (+ i 1))))))) ;; Configure persistent logging for leader mode in APPEND mode, so ;; a hung instance's evidence survives across leader restarts. A ;; /mcp reconnect SIGKILLs the leader and spawns a fresh one; the ;; generation, so a hang -> /mcp -> still-hung -> /mcp loop lost ;; the original (load-bearing) evidence. Appending into one file ;; keeps the whole history (capped by *log-carryover-max*). ;; ;; PER-INSTANCE FILE (courier-<pid>.log, was the shared courier.log): ;; append here is EMULATED -- read the old contents, reopen (which ;; truncates), write them back, then log from the end. With two leader ;; couriers running that emulation is mutually destructive: one truncates ;; while the other's port sits at a high offset, so the next write lands ;; past end-of-file and the kernel zero-fills the gap. That is where the ;; NUL bytes in the 2026-08-04 log came from, and NUL bytes make grep ;; treat the file as binary and report ZERO matches for a pattern `tail` ;; shows plainly -- an instrument failing silently, mid-incident. ;; sigil has no O_APPEND open, so the emulation cannot be made atomic; ;; giving each instance its own file removes the sharing instead. ;; The `--log <path>` form is deliberately NOT rewritten: claude-ops ;; passes an already-unique per-task path there and reads it back by ;; name, so inserting a pid would break that contract for no gain. ;; Part 2 (sync logging): the target is a plain file port ;; (open-output-file → async-port? = #f), so log writes never go ;; through the scheduler's await-port-writable / abort-to-prompt path. ;; was given), so production never logs to an async port. (define (configure-persistent-log!) (let* ((log-dir (path-dirname (default-relay-dir))) (log-path (path-join log-dir "courier.log"))) (log-path (path-join log-dir (string-append "courier-" (number->string (process-id)) ".log")))) (ensure-directory log-dir) (prune-instance-logs! log-dir) (log-configure! target: (open-log-append! log-path)))) ;; Configure logging from CLI args, opening any --log <path> targetsrc/courier/poller.sglmodified
(sigil mcp channel) (sigil log) (courier config) (courier claim) (courier relay)) (export telegram-poller-supervisor telegram-poller-shutdown! (let ((st (car *active-poller*))) (when st (set-poller-state-running?! st #f) (poller-kill! st)))) (poller-kill! st))) ;; Give up the single-poller claim so the next session can poll ;; without waiting for a staleness check. Called unconditionally: an ;; instance that claimed and then failed to spawn still owns the claim. (release-poller-claim!)) ;; ============================================================ ;; Supervisor: Child Lifecycle ;;; detect wedges, respawn. Runs as a goroutine in the leader. ;;; Never returns; never raises past its own guards -- a Telegram ;;; failure of any kind cannot wedge MCP request servicing. ;;; ;;; Telegram allows ONE getUpdates consumer per bot token, so this first ;;; takes the single-poller claim (see (courier claim)). Not getting it is ;;; the NORMAL outcome for the 2nd..Nth concurrent session: this returns ;;; quietly and the instance runs on as a full MCP server -- tools, ;;; relays, sends and notifications all unaffected. There is deliberately ;;; no retry loop and no exit. (define (telegram-poller-supervisor server config) (: mcp-server? courier-config? -> void?) (channel-wait-initialized! server) (case (acquire-poller-claim! (courier-config-telegram-token config)) ((acquired) (poller-supervise-loop server)) ((held-by-other) (log-info "Telegram polling skipped -- another courier instance holds the poller claim (normal; this instance is fully functional)")) (else (log-warn "Telegram polling skipped -- could not establish poller claim ownership")))) (define (poller-supervise-loop server) (log-info "Telegram polling started (isolated poller process)") (let respawn ((delay *restart-delay-initial*)) (let ((started (current-second))src/courier/telegram.sglmodified
register-media-upload-handler! path-media-type resolve-media-type format-size) format-size ;; exported so tests assert on the exact returned strings *dry-run-not-sent-message* *duplicate-suppressed-message*) (begin ;; HTTP read timeout (seconds) for leader-side Telegram sends. A ;; read, so the request timeout alone can't bound it). (define *send-connect-timeout* 10) ;; Returned when COURIER_DISABLE_TELEGRAM_SEND gagged the send. It must ;; be impossible to read as delivery: it does not begin with, and does not ;; contain, a bare "Message sent". Named rather than inlined so the test ;; suite asserts on the exact string the tool returns. (define *dry-run-not-sent-message* "Message NOT sent: COURIER_DISABLE_TELEGRAM_SEND is set (dry-run, nothing delivered).") ;; Returned when the persistent dedup window suppressed a repeat. Under ;; record-before-send this means "an identical message to this chat was ;; already accepted for delivery", NOT "this call delivered something" -- ;; so it says which happened rather than reusing the plain success string. (define *duplicate-suppressed-message* "Message already sent (duplicate suppressed: an identical message to this chat was accepted for delivery within the dedup window; nothing was sent again).") ;; Idempotency/dedup backstop for Telegram sends lives in ;; (courier dedup): a PERSISTENT, restart-surviving (chat-id,text) ;; window. It is the load-bearing defence against the send-path ((suppress) (log-info "Duplicate send-message suppressed" chat-id: chat-id) "Message sent.") *duplicate-suppressed-message*) (else ;; Test hook: latency simulation (inert by default). (when (and (number? send-delay) (> send-delay 0)) (sleep send-delay)) (if send-disabled ;; Test hook: dry-run — no real delivery. ;; This used to return the real success ;; string, so every gagged send from ;; 2026-07-06 to 2026-08-04 reported ;; delivery it had not earned — and, worse, ;; was a perfect mimic of the silent-drop ;; bug it was masking, making production ;; observation of that bug impossible. ;; A dry-run is a DEFINITE non-delivery, so ;; it also releases the dedup key: without ;; that, the very next identical send hits ;; the 'suppress path and reports "Message ;; sent." — reintroducing the same lie one ;; call later. Returning (not raising) is ;; deliberate: an MCP error invites a client ;; retry, and retry-against-a-killed-courier ;; is the mechanism behind the send storm. (begin (log-info "Telegram send disabled (dry-run)" (dedup-unrecord! dedup-path key) (log-info "Telegram send disabled (dry-run) -- NOT delivered" chat-id: chat-id) "Message sent.") *dry-run-not-sent-message*) ;; Deliver exactly once. The key is already ;; recorded. A transport error must NEVER ;; escape this handler (an MCP error invitestest/test-claim.sgladded
;; Unit tests for (courier claim) -- the single-poller claim.;;;; Telegram allows ONE getUpdates consumer per bot token. Every leader-mode;; courier used to start its own poller, so N concurrent sessions produced N;; pollers and a continuous 409 stream (measured 2026-08-04: three at once).;; These tests pin the claim's contract; the end-to-end behaviour -- two real;; couriers, one poller child, zero 409s at a mock that CAN produce them -- is;; proved by repro/repro-single-poller.sh.;;;; Note on liveness: under the test runner the "courier" process is the sigil;; runtime, so pid-liveness compares /proc/<pid>/comm against OUR OWN comm;; rather than a hardcoded name. That is deliberate in the implementation and;; is what makes these tests meaningful here at all.(import (sigil test) (sigil core) (sigil string) (sigil fs) (sigil path) (sigil process) (courier dedup) (courier claim))(define (fresh-dir) (let ((d (make-temp-directory))) (setenv! "COURIER_RELAY_DIR" (path-join d "relays")) d));; ============================================================;; Token hashing;; ============================================================(test-group "claim-token-hash" (test "is deterministic" (assert-equal (claim-token-hash "123456:AAbbCC") (claim-token-hash "123456:AAbbCC"))) (test "differs for different tokens" (assert-true (not (string=? (claim-token-hash "123456:AAbbCC") (claim-token-hash "123456:AAbbCD"))))) (test "is a fixed 12 hex characters regardless of the token" (assert-equal 12 (string-length (claim-token-hash "a"))) (assert-equal 12 (string-length (claim-token-hash ""))) (assert-equal 12 (string-length (claim-token-hash "9999999999:AAEEFFbbccddeeff00112233445566778899")))) ;; The token must never be recoverable from anything written to disk. (test "never embeds the token itself" (let ((token "123456:SECRETSECRETSECRET")) (assert-true (not (string-contains? (claim-token-hash token) "SECRET"))) (assert-true (not (string-contains? (poller-claim-path token) "SECRET"))))))(test-group "poller-claim-path" (test "is scoped to the token" (let ((d (fresh-dir))) (assert-true (not (string=? (poller-claim-path "bot-one") (poller-claim-path "bot-two")))))) (test "is scoped to the relay dir, so tests never contend with the live one" (let ((a (begin (fresh-dir) (poller-claim-path "same-token"))) (b (begin (fresh-dir) (poller-claim-path "same-token")))) (assert-true (not (string=? a b))))));; ============================================================;; Liveness -- three states, not two;; ============================================================(test-group "pid-liveness" ;; POSITIVE CONTROL. If this fails, every 'dead below is meaningless: ;; an instrument that answers 'dead for everything would pass those tests ;; and would silently break every claim it ever saw. (test "reports our own pid alive" (assert-equal 'alive (pid-liveness (number->string (process-id))))) (test "reports an impossible pid dead" (assert-equal 'dead (pid-liveness "999999"))) (test "reports garbage dead rather than raising" (assert-equal 'dead (pid-liveness "not-a-pid"))) (test "can read our own comm (the recycled-pid discriminator works)" (assert-true (string? (proc-comm (number->string (process-id))))) (assert-true (> (string-length (proc-comm (number->string (process-id)))) 0))) (test "reads nothing for a pid that does not exist" (assert-equal #f (proc-comm "999999"))) ;; The recycled-pid discriminator, tested from BOTH sides against the same ;; live process (pid 1 always exists). Only testing the 'dead side would be ;; satisfied by an implementation that always answers 'dead, which would ;; break every claim it ever saw. (test "a live pid with a MATCHING comm is alive" (assert-equal 'alive (pid-liveness/comm "1" (proc-comm "1")))) (test "a live pid with a DIFFERENT comm is treated as dead (recycled pid)" (assert-equal 'dead (pid-liveness/comm "1" "not-the-same-comm"))) (test "an unreadable comm falls back to 'alive, never to 'dead" (assert-equal 'alive (pid-liveness/comm "1" #f))));; ============================================================;; Acquire / release;; ============================================================(test-group "acquire-poller-claim!" (test "a first caller acquires" (fresh-dir) (assert-equal 'acquired (acquire-poller-claim! "tok-a")) (release-poller-claim!)) (test "records OUR pid in the claim" (fresh-dir) (acquire-poller-claim! "tok-b") (assert-equal (number->string (process-id)) (read-claim-owner (poller-claim-path "tok-b"))) (release-poller-claim!)) (test "release removes the claim" (fresh-dir) (acquire-poller-claim! "tok-c") (release-poller-claim!) (assert-equal #f (read-claim-owner (poller-claim-path "tok-c")))) (test "release is safe when nothing is held" (fresh-dir) (release-poller-claim!) (release-poller-claim!) (assert-true #t)) ;; A claim left by a process that is GONE must not block the estate: that ;; is the SIGKILL case, and SIGKILL is routine (an /mcp reconnect is one). (test "breaks a stale claim whose owner is gone" (let ((d (fresh-dir))) (let ((path (poller-claim-path "tok-stale"))) (ensure-directory (path-dirname path)) (make-symlink "999999" path) (assert-equal "999999" (read-claim-owner path)) (assert-equal 'acquired (acquire-poller-claim! "tok-stale")) (assert-equal (number->string (process-id)) (read-claim-owner path)) (release-poller-claim!)))) ;; A claim whose owner pid is ALIVE but is not a courier can only be a ;; recycled pid, so it is stale and must be broken -- otherwise one ;; unlucky pid reuse would silence Telegram until that unrelated process ;; exited. (The complementary case, a claim held by a genuinely live ;; courier being left strictly alone, needs a second courier process and ;; is proved end-to-end by repro/repro-single-poller.sh case 1, where ;; instance B neither steals nor rewrites instance A's claim.) (test "treats a live-but-unrelated owner pid as a recycled, stale claim" (let ((d (fresh-dir))) (let ((path (poller-claim-path "tok-recycled"))) (ensure-directory (path-dirname path)) (make-symlink "1" path) ;; pid 1 is alive and is definitely not us (assert-equal 'acquired (acquire-poller-claim! "tok-recycled")) (assert-equal (number->string (process-id)) (read-claim-owner path)) (release-poller-claim!)))) (test "does not release a claim that now belongs to someone else" (let ((d (fresh-dir))) (let ((path (poller-claim-path "tok-steal"))) (acquire-poller-claim! "tok-steal") ;; Someone else takes over (as would happen if we were SIGKILLed and ;; a later instance broke our stale claim). (delete-file path) (make-symlink "1" path) (release-poller-claim!) (assert-equal "1" (read-claim-owner path))))));; ============================================================;; Shared hash core;; ============================================================(test-group "fnv1a-hex" (test "send-dedup-key is unchanged by sharing the hash core" (assert-equal (fnv1a-hex "331005009:hello") (send-dedup-key 331005009 "hello"))) (test "is deterministic and differs per input" (assert-equal (fnv1a-hex "abc") (fnv1a-hex "abc")) (assert-true (not (string=? (fnv1a-hex "abc") (fnv1a-hex "abd"))))))