repro/single_poller_harness.py

1#!/usr/bin/env python3

2"""Off-device harness for the SINGLE-POLLER claim and the HONEST-SEND return.

4Everything here runs against repro/mock_telegram.py. NOTHING can reach real

5Telegram: every courier instance is started with COURIER_TELEGRAM_API_URL

6pointed at the local mock, which is the only endpoint the send AND poll paths

7use once that variable is set.

9The cases are NUMBERED BY TOPIC, and run in the order 1, 3, 4, 7, 10, 9, 8, 5, 6, 2 --

10the ordering is dictated by shared state (cases 1, 3, 4 and 7 all operate on the

11same pair of live instances, and the pre-fix control in case 2 runs last so its

12409s can never be confused with the fixed binary's). What each proves:

14 1 ONE POLLER two leader-mode couriers on one token -> exactly one

15 poller child process, and zero 409s at the mock.

16 2 409 DETECTOR WORKS the same two-instance run with the PRE-FIX binary

17 (COURIER_CONTROL_BIN) -> 409s appear. This is the

18 positive control: without it, case 1's "zero 409s"

19 is a claim of absence made by an instrument never

20 shown capable of detecting presence.

21 3 NON-POLLER IS FINE the instance that lost the claim still serves its

22 MCP tools -- create-relay, a real worker connecting

23 over that relay, send both ways, close-relay.

24 4 SIGKILL RECOVERY SIGKILL (not a graceful stop) the claim holder, then

25 (a) the OTHER already-running instance re-acquires

26 and starts polling on its own, and (b) a fresh

27 instance can also break a stale claim. (a) is the

28 one that matters: without the periodic re-acquire,

29 a healthy instance would sit there forever and the

30 estate would silently stop receiving Telegram.

31 5 DRY-RUN IS HONEST with COURIER_DISABLE_TELEGRAM_SEND=1 the tool returns

32 a string that cannot be read as success, does not

33 raise, and delivers nothing at the mock.

34 6 GAG DOESN'T POISON a dry-run must not leave a dedup key behind that

35 would suppress a LATER real send of the same text.

36 Dry-run, then flag off, then the same message must

37 actually deliver.

38 7 LOG HYGIENE two concurrent leader instances write two distinct

39 log files, and grep works on both (the shared-log

40 interleaving produced NUL bytes, which made grep

41 silently report zero matches during an incident).

42 8 LOG PRUNING instance logs are capped, a LIVE instance's log is

43 never pruned however old it looks, and nothing that

44 is not an instance log is touched.

45 10 PRODUCTION SHAPE two instances launched from DIFFERENT binary paths,

46 as claude-ops actually does, still yield one poller.

47 Every other case runs both from one path and so

48 cannot see a name-dependent identity bug.

49 9 FORCED TAKEOVER COURIER_TELEGRAM_POLL_FORCE takes the claim from a

50 LIVE holder; the displaced instance stops polling,

51 stays healthy, and does not delete the new owner's

52 claim -- with zero 409s across the handover.

54Usage:

55 single_poller_harness.py --bin <courier> [--control-bin <pre-fix courier>]

56"""

57import argparse

58import json

59import os

60import re

61import shutil

62import signal

63import subprocess

64import sys

65import threading

66import time

67import urllib.error

68import urllib.request

70CHAT_ID = "331005009"

71PORT = int(os.environ.get("PORT", "8613"))

72BASE = f"http://127.0.0.1:{PORT}"

74# Longer than courier's 2s poll interval, so two independent pollers overlap

75# deterministically rather than by luck. See mock_telegram.py's docstring.

76DWELL = 2.5

78# How long to watch a two-instance run. Must be long enough that the pre-fix

79# control reliably produces 409s inside it, or "zero 409s" from the fixed

80# binary would only mean "we did not watch for long enough".

81OBSERVE_WINDOW = 20

83# Must match *claim-recheck-interval* in src/courier/poller.sgl.

84CLAIM_RECHECK = 60

86FAILED = []

87PASSED = []

90def check(label, ok, detail=""):

91 (PASSED if ok else FAILED).append(label)

92 print(f" {'PASS' if ok else 'FAIL'} {label}" + (f": {detail}" if detail else ""))

93 return ok

96# ----------------------------------------------------------------- mock ----

98def post(path, obj):

99 data = json.dumps(obj).encode()

100 req = urllib.request.Request(BASE + path, data=data,

101 headers={"Content-Type": "application/json"})

102 with urllib.request.urlopen(req, timeout=10) as r:

103 return json.load(r)

104

105

106def stats():

107 with urllib.request.urlopen(BASE + "/_stats", timeout=10) as r:

108 return json.load(r)

109

110

111class Mock:

112 def __init__(self, logdir):

113 self.logdir = logdir

114 here = os.path.dirname(os.path.abspath(__file__))

115 os.makedirs(logdir, exist_ok=True)

116 self.proc = subprocess.Popen(

117 [sys.executable, os.path.join(here, "mock_telegram.py"),

118 str(PORT), os.path.join(logdir, "mocklog")],

119 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

120 for _ in range(50):

121 try:

122 stats()

123 return

124 except Exception:

125 time.sleep(0.1)

126 raise RuntimeError("mock did not come up")

127

128 def stop(self):

129 self.proc.terminate()

130 try:

131 self.proc.wait(timeout=5)

132 except Exception:

133 self.proc.kill()

134

135

136# -------------------------------------------------------------- courier ----

137

138class Courier:

139 """One courier subprocess speaking MCP over stdio."""

140

141 def __init__(self, bin_path, run_dir, name, disable_send=False,

142 extra_env=None, argv=("serve",)):

143 self.name = name

144 self.run_dir = run_dir

145 self.notifications = []

146 os.makedirs(run_dir, exist_ok=True)

147 env = dict(os.environ)

148 env["COURIER_TELEGRAM_TOKEN"] = "TESTTOKEN-single-poller"

149 env["COURIER_TELEGRAM_CHAT_ID"] = CHAT_ID

150 env["COURIER_TELEGRAM_API_URL"] = BASE

151 # Shared state dir: this is what makes the instances contend, exactly

152 # as concurrent leader sessions share ~/.courier.

153 env["COURIER_RELAY_DIR"] = os.path.join(run_dir, "relays")

154 if disable_send:

155 env["COURIER_DISABLE_TELEGRAM_SEND"] = "1"

156 else:

157 env.pop("COURIER_DISABLE_TELEGRAM_SEND", None)

158 if extra_env:

159 env.update(extra_env)

160 self.errf = open(os.path.join(run_dir, f"{name}.stderr"), "wb")

161 # NOTE: no --log. That is deliberate -- it exercises the leader-mode

162 # persistent-log path (configure-persistent-log!), which is the one

163 # that used to collide on a single shared courier.log.

164 self.proc = subprocess.Popen(

165 [bin_path, *argv],

166 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.errf,

167 env=env, bufsize=0)

168 self._id = 0

169 self._pending = {}

170 self._lock = threading.Lock()

171 self._reader = threading.Thread(target=self._read_loop, daemon=True)

172 self._reader.start()

173

174 # --- MCP plumbing ---

175 def _read_loop(self):

176 for raw in self.proc.stdout:

177 try:

178 msg = json.loads(raw)

179 except Exception:

180 continue

181 rid = msg.get("id")

182 if rid is not None:

183 with self._lock:

184 ev = self._pending.get(rid)

185 if ev:

186 ev[1] = msg

187 ev[0].set()

188 else:

189 # Server-initiated notification (channel events: relay and

190 # Telegram messages arrive this way).

191 with self._lock:

192 self.notifications.append(msg)

193

194 def _send(self, obj):

195 self.proc.stdin.write((json.dumps(obj) + "\n").encode())

196 self.proc.stdin.flush()

197

198 def rpc(self, method, params, timeout=20):

199 with self._lock:

200 self._id += 1

201 rid = self._id

202 ev = [threading.Event(), None]

203 self._pending[rid] = ev

204 self._send({"jsonrpc": "2.0", "id": rid, "method": method,

205 "params": params})

206 if not ev[0].wait(timeout):

207 raise TimeoutError(f"{self.name}: no response to {method}")

208 return ev[1]

209

210 def initialize(self):

211 res = self.rpc("initialize", {

212 "protocolVersion": "2024-11-05", "capabilities": {},

213 "clientInfo": {"name": "single-poller-harness", "version": "0"}})

214 self._send({"jsonrpc": "2.0", "method": "notifications/initialized",

215 "params": {}})

216 return res

217

218 def call(self, tool, arguments, timeout=30):

219 res = self.rpc("tools/call", {"name": tool, "arguments": arguments},

220 timeout=timeout)

221 if "result" in res:

222 try:

223 return res["result"]["content"][0]["text"]

224 except Exception:

225 return json.dumps(res["result"])

226 return "ERROR:" + json.dumps(res.get("error", res))

227

228 # --- lifecycle ---

229 @property

230 def pid(self):

231 return self.proc.pid

232

233 def alive(self):

234 return self.proc.poll() is None

235

236 def sigkill(self):

237 """SIGKILL, as a /mcp teardown does. No chance to release a claim."""

238 os.kill(self.proc.pid, signal.SIGKILL)

239 self.proc.wait(timeout=10)

240

241 def close_stdin(self):

242 """Graceful shutdown: stdin EOF is how a session ends."""

243 try:

244 self.proc.stdin.close()

245 except Exception:

246 pass

247

248 def kill(self):

249 try:

250 if self.alive():

251 self.proc.kill()

252 self.proc.wait(timeout=5)

253 except Exception:

254 pass

255 try:

256 self.errf.close()

257 except Exception:

258 pass

259

260 def stderr_text(self):

261 try:

262 self.errf.flush()

263 except Exception:

264 pass

265 with open(os.path.join(self.run_dir, f"{self.name}.stderr"),

266 "rb") as f:

267 return f.read().decode("utf-8", "replace")

268

269 def log_path(self):

270 """This instance's own log file.

271

272 Leader mode with no --log logs to <state-dir>/courier-<pid>.log, NOT

273 to stderr, so assertions about log lines must read this.

274 """

275 return os.path.join(self.run_dir, f"courier-{self.pid}.log")

276

277 def log_text(self):

278 try:

279 with open(self.log_path(), "rb") as f:

280 return f.read().decode("utf-8", "replace")

281 except FileNotFoundError:

282 return ""

283

284

285# ------------------------------------------------------- process helpers ----

286

287def child_pids(pid):

288 """Direct children of `pid`, read straight from /proc.

289

290 Deliberately not pgrep/pkill: a name-matching pattern here could match

291 this harness or an unrelated courier on the host, and a missing tool

292 would return an empty list that reads exactly like 'no children'.

293 """

294 out = []

295 for entry in os.listdir("/proc"):

296 if not entry.isdigit():

297 continue

298 try:

299 with open(f"/proc/{entry}/status") as f:

300 ppid = None

301 for line in f:

302 if line.startswith("PPid:"):

303 ppid = int(line.split()[1])

304 break

305 if ppid == pid:

306 out.append(int(entry))

307 except (FileNotFoundError, ProcessLookupError, PermissionError):

308 continue

309 return out

310

311

312def poller_children(pid):

313 """Children of `pid` that are courier poller processes."""

314 found = []

315 for c in child_pids(pid):

316 try:

317 with open(f"/proc/{c}/cmdline", "rb") as f:

318 argv = f.read().decode("utf-8", "replace").rstrip("\0").split("\0")

319 exe = os.readlink(f"/proc/{c}/exe")

320 except (FileNotFoundError, ProcessLookupError, PermissionError, OSError):

321 continue

322 # Match the LAST argv element exactly, and require the executable to

323 # really be a courier. A substring search over cmdline matches any

324 # process whose command line merely CONTAINS the flag -- including the

325 # very script doing the search. That is the `pgrep -f` self-match

326 # hazard, and it produced two phantom "extra poller" readings on

327 # 2026-08-04 before it was caught.

328 if argv and argv[-1] == "--telegram-poller" and \

329 os.path.basename(exe).startswith("courier"):

330 found.append(c)

331 return found

332

333

334def claim_files(state_dir):

335 if not os.path.isdir(state_dir):

336 return []

337 return sorted(n for n in os.listdir(state_dir)

338 if n.startswith("telegram-poller-") and n.endswith(".claim"))

339

340

341def claim_owner(state_dir):

342 """Raw claim target: "<pid>:<starttime>" (or a bare pid on older builds)."""

343 files = claim_files(state_dir)

344 if not files:

345 return None

346 return os.readlink(os.path.join(state_dir, files[0]))

347

348

349def claim_owner_pid(state_dir):

350 """Just the pid. The claim records pid:starttime -- start time is what

351 makes a recycled pid detectable without depending on the process NAME,

352 which differs per instance in production."""

353 o = claim_owner(state_dir)

354 return o.split(":")[0] if o else None

355

356

357def wait_for(predicate, timeout=30, interval=0.25):

358 deadline = time.time() + timeout

359 while time.time() < deadline:

360 if predicate():

361 return True

362 time.sleep(interval)

363 return False

364

365

366# ----------------------------------------------------------------- cases ----

367

368def case_one_poller(binp, run_dir, label="1 ONE POLLER"):

369 """Two leader-mode couriers, one token. Returns (a, b, state_dir)."""

370 print(f"=== {label} ===")

371 post("/_seed", {"updates": []})

372 post("/_dwell", {"seconds": DWELL})

373 state_dir = run_dir

374

375 a = Courier(binp, run_dir, "inst-a")

376 a.initialize()

377 got_a = wait_for(lambda: len(poller_children(a.pid)) == 1, timeout=40)

378 # POSITIVE CONTROL for the child-detector: if this is False, every later

379 # "0 pollers" reading below is meaningless.

380 check("instance A starts a poller child (detector positive control)",

381 got_a, f"children={poller_children(a.pid)}")

382

383 b = Courier(binp, run_dir, "inst-b")

384 b.initialize()

385 # Give B every chance to (wrongly) start one, and give the mock a window

386 # comfortably longer than the one in which the pre-fix control produces

387 # its first 409.

388 time.sleep(OBSERVE_WINDOW)

389

390 pa, pb = poller_children(a.pid), poller_children(b.pid)

391 check("exactly one poller child across both instances",

392 len(pa) + len(pb) == 1, f"A={pa} B={pb}")

393 check("instance B did NOT start a poller", len(pb) == 0, f"B={pb}")

394

395 owner = claim_owner(state_dir)

396 check("claim file records the polling instance's pid",

397 claim_owner_pid(state_dir) == str(a.pid),

398 f"claim owner={owner} A={a.pid} B={b.pid}")

399

400 log = b.log_text()

401 skip_lines = [l for l in log.splitlines() if "poller claim" in l]

402 check("instance B logs the skip, at INFO and not WARN/ERROR",

403 (len(skip_lines) == 1

404 and "[INFO]" in skip_lines[0]

405 and "another courier instance holds" in skip_lines[0]),

406 repr(skip_lines))

407

408 s = stats()

409 # An absence is only meaningful if the thing had a chance to happen:

410 # zero 409s across zero polls would pass for free. Assert the window

411 # actually contained polling before believing the zero.

412 check("the observation window contained real polling (denominator)",

413 s["getupdates"] >= 3,

414 f"getupdates={s['getupdates']} over {OBSERVE_WINDOW}s")

415 check("zero 409 conflicts at the mock", s["conflicts"] == 0,

416 f"conflicts={s['conflicts']} getupdates={s['getupdates']} "

417 f"max_inflight={s['max_inflight']}")

418 check("the mock never saw two concurrent pollers", s["max_inflight"] <= 1,

419 f"max_inflight={s['max_inflight']}")

420 return a, b, state_dir

421

422

423def case_control(control_bin, run_dir):

424 """Positive control: the pre-fix binary MUST produce 409s."""

425 print("=== 2 409 DETECTOR WORKS (pre-fix control) ===")

426 if not control_bin:

427 print(" SKIP no COURIER_CONTROL_BIN given -- case 1's 'zero 409s' "

428 "is UNCONTROLLED and proves nothing on its own")

429 FAILED.append("409 positive control not run")

430 return

431 post("/_seed", {"updates": []})

432 post("/_dwell", {"seconds": DWELL})

433

434 # Confirm the control really IS the pre-fix code before drawing any

435 # conclusion from it. A control that silently pointed at the fixed binary

436 # would report "no 409s" and look like a disproof of the whole bug.

437 probe = Courier(control_bin, os.path.join(run_dir, "probe"), "ctl-probe",

438 disable_send=True)

439 probe.initialize()

440 old = probe.call("send-message", {"text": "control identity probe",

441 "to": CHAT_ID})

442 check("control binary is genuinely PRE-FIX (dry-run still lies)",

443 old == "Message sent.", repr(old))

444 probe.kill()

445

446 a = Courier(control_bin, run_dir, "ctl-a")

447 a.initialize()

448 b = Courier(control_bin, run_dir, "ctl-b")

449 b.initialize()

450 t0 = time.time()

451 ok = wait_for(lambda: stats()["conflicts"] > 0, timeout=60)

452 first = time.time() - t0

453 # Keep watching for the SAME window the fixed run got, so the two numbers

454 # are comparable rather than one being a shorter look than the other.

455 time.sleep(OBSERVE_WINDOW)

456 s = stats()

457 check("pre-fix binary: two instances DO produce 409s", ok,

458 f"conflicts={s['conflicts']} in {OBSERVE_WINDOW}s "

459 f"(first after {first:.1f}s) getupdates={s['getupdates']} "

460 f"max_inflight={s['max_inflight']}")

461 check("pre-fix binary: two poller children exist",

462 len(poller_children(a.pid)) + len(poller_children(b.pid)) == 2,

463 f"A={poller_children(a.pid)} B={poller_children(b.pid)}")

464 for c in (a, b):

465 c.kill()

466 time.sleep(1)

467

468

469def case_relay_tools(binp, b, run_dir):

470 """The instance that LOST the claim must be a fully working leader.

471

472 Create a relay, connect a real worker-mode courier to it, send through it,

473 and close it. This also re-checks that worker mode is untouched: three

474 live worker relays depend on it.

475 """

476 print("=== 3 NON-POLLER IS FINE (relay tools on the skipped instance) ===")

477 r = b.call("create-relay", {"name": "harness-relay"})

478 check("non-polling instance: create-relay works",

479 r == "Relay 'harness-relay' created and listening.", repr(r))

480

481 w = Courier(binp, run_dir, "worker", argv=("--relay", "harness-relay"))

482 try:

483 w.initialize()

484 text = "leader-to-worker over the relay"

485 connected = wait_for(

486 lambda: "harness-relay" in b.call("list-relays", {})

487 and "connected" in b.call("list-relays", {}).lower(), timeout=20)

488 check("non-polling instance: list-relays shows the connected worker",

489 connected, repr(b.call("list-relays", {})))

490

491 r = b.call("send-message", {"text": text, "to": "harness-relay"})

492 check("non-polling instance: send-message over the relay works",

493 r == "Message sent.", repr(r))

494 got = wait_for(lambda: any(text in json.dumps(n)

495 for n in list(w.notifications)), timeout=20)

496 check("the worker actually RECEIVED the relayed message", got,

497 repr(w.notifications[-1] if w.notifications else None))

498

499 back = w.call("send-message", {"text": "worker-to-leader ack"})

500 check("worker mode: send-message back to the leader works",

501 back == "Message sent to leader.", repr(back))

502 finally:

503 w.kill()

504

505 r = b.call("close-relay", {"name": "harness-relay"})

506 check("non-polling instance: close-relay works",

507 r == "Relay 'harness-relay' closed.", repr(r))

508

509

510def case_sigkill(binp, run_dir, a, b, state_dir):

511 print("=== 4 SIGKILL RECOVERY ===")

512 dead_pid = a.pid

513 a.sigkill()

514 time.sleep(1)

515 owner = claim_owner(state_dir)

516 check("claim SURVIVES the SIGKILL (nothing released it)",

517 claim_owner_pid(state_dir) == str(dead_pid),

518 f"owner={owner} killed={dead_pid}")

519

520 # (a) The instance that was ALREADY RUNNING and had been told to skip must

521 # take over by itself. This is the case that would silently not happen if

522 # the supervisor acquired once and gave up: B is healthy, holds the token,

523 # and would never poll again.

524 took_over = wait_for(lambda: len(poller_children(b.pid)) == 1,

525 timeout=CLAIM_RECHECK + 45)

526 check("the already-running non-poller re-acquires and starts polling",

527 took_over, f"B children={poller_children(b.pid)} "

528 f"claim owner={claim_owner(state_dir)}")

529 check("the claim now records the instance that took over",

530 claim_owner_pid(state_dir) == str(b.pid),

531 f"owner={claim_owner(state_dir)} B={b.pid}")

532 log = b.log_text()

533 check("the takeover is logged, once, at INFO",

534 len([l for l in log.splitlines()

535 if "claim acquired -- the previous holder is gone" in l

536 and "[INFO]" in l]) == 1,

537 repr([l for l in log.splitlines() if "claim" in l]))

538 # And the skip line was NOT repeated every recheck interval.

539 check("the periodic re-check does not spam the log",

540 len([l for l in log.splitlines()

541 if "another courier instance holds" in l]) == 1,

542 f"{len([l for l in log.splitlines() if 'another courier instance holds' in l])} skip lines")

543

544 # (b) A fresh instance can break a stale claim too. Kill the new holder

545 # hard so the claim is stale again.

546 b.sigkill()

547 time.sleep(1)

548 c = Courier(binp, run_dir, "inst-c")

549 c.initialize()

550 got = wait_for(lambda: len(poller_children(c.pid)) == 1, timeout=45)

551 check("a fresh instance breaks the stale claim and polls", got,

552 f"children={poller_children(c.pid)}")

553 check("claim now records the new instance",

554 claim_owner_pid(state_dir) == str(c.pid),

555 f"owner={claim_owner(state_dir)} C={c.pid}")

556 log = c.log_text()

557 check("the break is logged",

558 any("Taking the Telegram poller claim" in l

559 and "previous holder is gone" in l

560 for l in log.splitlines()),

561 repr([l for l in log.splitlines() if "Taking" in l][:2]))

562 return c

563

564

565def case_forced_takeover(binp, run_dir):

566 """COURIER_TELEGRAM_POLL_FORCE: a specific session takes the claim from a

567 LIVE holder, and the displaced one stands down.

568

569 This is the case the unit tests cannot reach: the force branch only runs

570 when the owner reads as 'alive, which needs a second real courier.

571

572 The property that matters is not just "B ends up polling" -- it is that

573 there is never a moment with TWO pollers on the token. A takeover that

574 produced even ten seconds of overlap would be the original 409 bug

575 arriving through the escape hatch, so conflicts are asserted at zero

576 across the whole handover.

577 """

578 print("=== 9 FORCED TAKEOVER (COURIER_TELEGRAM_POLL_FORCE) ===")

579 d = os.path.join(run_dir, "force")

580 os.makedirs(d, exist_ok=True)

581 post("/_seed", {"updates": []})

582 post("/_dwell", {"seconds": DWELL})

583

584 a = Courier(binp, d, "force-a")

585 a.initialize()

586 got_a = wait_for(lambda: len(poller_children(a.pid)) == 1, timeout=40)

587 check("holder A is polling before the takeover", got_a,

588 f"A children={poller_children(a.pid)}")

589

590 b = Courier(binp, d, "force-b",

591 extra_env={"COURIER_TELEGRAM_POLL_FORCE": "1"})

592 b.initialize()

593

594 # B waits out the stand-down window before it starts, so allow for it.

595 took = wait_for(lambda: len(poller_children(b.pid)) == 1, timeout=90)

596 check("the forcing instance takes over and polls", took,

597 f"B children={poller_children(b.pid)} owner={claim_owner(d)}")

598 check("the claim records the forcing instance",

599 claim_owner_pid(d) == str(b.pid),

600 f"owner={claim_owner(d)} B={b.pid}")

601

602 stood_down = wait_for(lambda: len(poller_children(a.pid)) == 0, timeout=45)

603 check("the displaced holder STOPS polling", stood_down,

604 f"A children={poller_children(a.pid)}")

605

606 alog = a.log_text()

607 check("the displaced holder logs that it lost the claim",

608 "claim taken by another instance" in alog,

609 repr([l for l in alog.splitlines() if "claim" in l][-2:]))

610 check("the displaced holder is still alive and serving",

611 a.alive() and a.call("list-relays", {}) is not None)

612

613 blog = b.log_text()

614 check("the forced takeover is logged at INFO",

615 len([l for l in blog.splitlines()

616 if "COURIER_TELEGRAM_POLL_FORCE" in l and "[INFO]" in l]) == 1,

617 repr([l for l in blog.splitlines() if "claim" in l][:3]))

618

619 # Let both run a while so any overlap would show up at the mock.

620 time.sleep(OBSERVE_WINDOW)

621 s = stats()

622 check("the handover produced NO 409s at any point", s["conflicts"] == 0,

623 f"conflicts={s['conflicts']} getupdates={s['getupdates']} "

624 f"max_inflight={s['max_inflight']}")

625 check("the mock never saw two concurrent pollers during the handover",

626 s["max_inflight"] <= 1, f"max_inflight={s['max_inflight']}")

627

628 # And the displaced instance must NOT have deleted the new owner's claim.

629 check("the displaced holder left the new owner's claim intact",

630 claim_owner_pid(d) == str(b.pid), f"owner={claim_owner(d)}")

631

632 for inst in (a, b):

633 inst.close_stdin()

634 time.sleep(1)

635 for inst in (a, b):

636 inst.kill()

637

638

639def case_log_pruning(binp, run_dir):

640 """Per-instance logs must not accumulate forever, and must not take

641 anything else with them.

642

643 Worth a behavioural case rather than trusting the source: the pruning

644 code is wrapped in a guard (a failure there costs disk, not

645 correctness), and a guard is exactly what hides a defect. The first

646 version of this function called two procedures that do not exist in

647 this stdlib; it compiled, it ran, the guard swallowed the unbound

648 variable, and pruning silently never happened. Nothing about the log

649 files would have looked wrong.

650 """

651 print("=== 8 LOG PRUNING (bounded, and only OUR files) ===")

652 d = os.path.join(run_dir, "prune")

653 os.makedirs(os.path.join(d, "relays"), exist_ok=True)

654 now = time.time()

655 fakes = []

656 for i in range(15):

657 p = os.path.join(d, f"courier-9{i:04d}.log")

658 with open(p, "w") as f:

659 f.write(f"fake generation {i}\n")

660 # Staggered mtimes, oldest first, so "keep the newest N" is testable.

661 os.utime(p, (now - (100 - i) * 60, now - (100 - i) * 60))

662 fakes.append(p)

663 # A log belonging to a LIVE COURIER, made to look ancient. This is the

664 # Jul-23 case: a long-lived session whose log has an old mtime because it

665 # has been quiet, with ten newer generations behind it. Pruning it by age

666 # would destroy exactly the evidence per-instance logs exist to preserve

667 # -- and it is the same long-lived process that caused the original bug.

668 #

669 # It has to be a real courier, not this harness: liveness is "the pid is

670 # running AND its comm matches courier's", so a live PYTHON pid is

671 # correctly judged a recycled pid and pruned. (Measured: the first

672 # version of this fixture used os.getpid() and failed for exactly that

673 # reason -- the code was right and the fixture was wrong.)

674 live = Courier(binp, d, "prune-live")

675 live.initialize()

676 live_old = live.log_path()

677 os.utime(live_old, (now - 90 * 86400, now - 90 * 86400))

678

679 # A worker-style name (no pid) must also be left alone.

680 worker_log = os.path.join(d, "courier-some-task-name.log")

681 with open(worker_log, "w") as f:

682 f.write("worker log\n")

683 os.utime(worker_log, (now - 91 * 86400, now - 91 * 86400))

684

685 # Files that are NOT instance logs must survive untouched.

686 bystanders = {

687 os.path.join(d, "send-dedup.state"): "1 abc\n",

688 os.path.join(d, "telegram-offset"): "12345",

689 os.path.join(d, "courier.log"): "the old shared log\n",

690 os.path.join(d, "courier-notes.txt"): "not a log\n",

691 }

692 for p, content in bystanders.items():

693 with open(p, "w") as f:

694 f.write(content)

695

696 c = Courier(binp, d, "prune-a")

697 c.initialize()

698 time.sleep(2)

699 remaining = sorted(n for n in os.listdir(d)

700 if n.startswith("courier-") and n.endswith(".log"))

701 dead_remaining = [n for n in remaining

702 if n not in (f"courier-{c.pid}.log",

703 os.path.basename(live_old),

704 os.path.basename(worker_log))]

705 check("dead instance logs are capped at 10", len(dead_remaining) == 10,

706 f"{len(dead_remaining)} dead remain: {dead_remaining}")

707 check("the instance's own log survived pruning",

708 f"courier-{c.pid}.log" in remaining, f"{remaining}")

709 check("the OLDEST dead fakes are the ones removed",

710 os.path.basename(fakes[0]) not in remaining

711 and os.path.basename(fakes[14]) in remaining,

712 f"{remaining}")

713 # The rule that matters: liveness beats age.

714 check("a LIVE instance's log survives even when 90 days old",

715 os.path.basename(live_old) in remaining, f"{remaining}")

716 check("a log whose name carries no pid is left alone",

717 os.path.basename(worker_log) in remaining, f"{remaining}")

718 survived = [os.path.basename(p) for p in bystanders if os.path.exists(p)]

719 check("non-log files (dedup state, offset) are untouched",

720 len(survived) == len(bystanders),

721 f"survived={survived} of {[os.path.basename(p) for p in bystanders]}")

722 for inst in (c, live):

723 inst.close_stdin()

724 time.sleep(1)

725 for inst in (c, live):

726 inst.kill()

727

728

729EXPECTED_DRY_RUN = ("Message NOT sent: COURIER_DISABLE_TELEGRAM_SEND is set "

730 "(dry-run, nothing delivered).")

731

732

733def case_dry_run(binp, run_dir):

734 print("=== 5 DRY-RUN IS HONEST ===")

735 post("/_seed", {"updates": []})

736 post("/_dwell", {"seconds": 0})

737 d = os.path.join(run_dir, "dry")

738 c = Courier(binp, d, "dry-a", disable_send=True)

739 c.initialize()

740 before = stats()["sends"]

741 text = "harness dry-run probe"

742 r = c.call("send-message", {"text": text, "to": CHAT_ID})

743 after = stats()["sends"]

744 print(f" returned: {r!r}")

745 check("dry-run returns the exact honest string", r == EXPECTED_DRY_RUN,

746 repr(r))

747 check("dry-run return contains no bare 'Message sent'",

748 "Message sent" not in r, repr(r))

749 check("dry-run does not raise (no MCP error)", not r.startswith("ERROR:"),

750 repr(r))

751 check("dry-run delivered nothing at the mock", after == before,

752 f"{before} -> {after}")

753 c.close_stdin()

754 time.sleep(1)

755 c.kill()

756 return d, text

757

758

759def case_gag_does_not_poison(binp, run_dir, dry_dir, text):

760 """A dry-run must not leave a dedup key that suppresses a LATER real send."""

761 print("=== 6 GAG DOESN'T POISON A LATER REAL SEND ===")

762 before = stats()["sends"]

763 # Same state dir (so the same send-dedup.state), gag OFF, same text.

764 c = Courier(binp, dry_dir, "dry-b", disable_send=False)

765 c.initialize()

766 r = c.call("send-message", {"text": text, "to": CHAT_ID})

767 after = stats()["sends"]

768 print(f" returned: {r!r}")

769 check("after the gag comes off, the SAME text really delivers",

770 after == before + 1, f"sends {before} -> {after}, returned {r!r}")

771 check("and it reports plain success", r == "Message sent.", repr(r))

772 c.close_stdin()

773 time.sleep(1)

774 c.kill()

775

776

777def case_log_hygiene(run_dir, pids):

778 print("=== 7 LOG HYGIENE (per-instance files, grep works) ===")

779 logs = sorted(n for n in os.listdir(run_dir)

780 if n.startswith("courier-") and n.endswith(".log"))

781 check("each leader instance wrote its own log file", len(logs) >= 2,

782 f"logs={logs}")

783 for pid in pids:

784 check(f"log file exists for pid {pid}", f"courier-{pid}.log" in logs,

785 f"logs={logs}")

786 nul = []

787 grep_ok = []

788 for n in logs:

789 p = os.path.join(run_dir, n)

790 with open(p, "rb") as f:

791 data = f.read()

792 if b"\x00" in data:

793 nul.append(n)

794 # grep must SEE a pattern that is present -- the incident was grep

795 # returning zero on a pattern `tail` showed plainly, because NUL

796 # bytes made grep treat the file as binary.

797 if shutil.which("grep"):

798 rc = subprocess.run(["grep", "-c", "Courier starting", p],

799 capture_output=True, text=True)

800 grep_ok.append((n, rc.stdout.strip(), rc.returncode))

801 check("no NUL bytes in any instance log", not nul, f"nul in {nul}")

802 check("grep tool is available (instrument check)",

803 shutil.which("grep") is not None)

804 check("grep finds the startup line in every instance log",

805 all(rc == 0 and int(cnt or 0) >= 1 for _, cnt, rc in grep_ok),

806 f"{grep_ok}")

807

808

809# ------------------------------------------------------------------ main ----

810

811def case_production_launch_shape(binp, run_dir):

812 """The claim must survive claude-ops' actual launch shape.

813

814 claude-ops copies each courier to /tmp/mcp-bins/courier-<launcher-pid>, so

815 every instance has a DIFFERENT /proc/<pid>/comm. Every other case in this

816 file runs both instances from ONE binary path, so their names match --

817 which is exactly why none of them could see the defect measured on the

818 live host on 2026-08-04, where a name-based identity check made each live

819 instance look like a recycled pid to the other.

820

821 Distinct names come from symlinks: a copied dev bundle cannot find its own

822 library dir, while a symlink still resolves /proc/self/exe back to the real

823 bundle and still yields the symlink's basename as comm. Both verified.

824 """

825 print("=== 10 PRODUCTION LAUNCH SHAPE (per-instance binary names) ===")

826 d = os.path.join(run_dir, "twopath")

827 os.makedirs(os.path.join(d, "relays"), exist_ok=True)

828 post("/_seed", {"updates": []})

829 post("/_dwell", {"seconds": DWELL})

830

831 a_path = os.path.join(d, "courier-1111")

832 b_path = os.path.join(d, "courier-2222")

833 for p in (a_path, b_path):

834 if not os.path.exists(p):

835 os.symlink(os.path.abspath(binp), p)

836

837 a = Courier(a_path, d, "twopath-a")

838 a.initialize()

839 wait_for(lambda: len(poller_children(a.pid)) == 1, timeout=40)

840 b = Courier(b_path, d, "twopath-b")

841 b.initialize()

842 time.sleep(OBSERVE_WINDOW)

843

844 ca = open(f"/proc/{a.pid}/comm").read().strip()

845 cb = open(f"/proc/{b.pid}/comm").read().strip()

846 # Setup control: if the names are NOT distinct this case degenerates into

847 # a duplicate of case 1 and proves nothing about the production shape.

848 check("the two instances really do have different process names",

849 ca != cb, f"A={ca!r} B={cb!r}")

850

851 pa, pb = poller_children(a.pid), poller_children(b.pid)

852 check("exactly one poller despite per-instance binary names",

853 len(pa) + len(pb) == 1, f"A={pa} B={pb}")

854 s = stats()

855 check("no 409s under the production launch shape", s["conflicts"] == 0,

856 f"conflicts={s['conflicts']} getupdates={s['getupdates']}")

857 # The end-state poller count alone would have PASSED the broken build --

858 # the loser stands down within 10s, so only the mock shows the overlap.

859 check("the mock never saw two concurrent pollers",

860 s["max_inflight"] <= 1, f"max_inflight={s['max_inflight']}")

861 check("the claim records a pid:starttime identity, not a bare pid",

862 (claim_owner(d) or "").count(":") == 1, f"owner={claim_owner(d)!r}")

863

864 for inst in (a, b):

865 inst.close_stdin()

866 time.sleep(1)

867 for inst in (a, b):

868 inst.kill()

869

870

871def main():

872 ap = argparse.ArgumentParser()

873 ap.add_argument("--bin", required=True)

874 ap.add_argument("--control-bin", default=os.environ.get("COURIER_CONTROL_BIN"))

875 ap.add_argument("--run-dir", default=None)

876 args = ap.parse_args()

877

878 # Default to a SHORT path, not repro/run-*, because the relay sockets live

879 # under it and AF_UNIX caps sun_path at ~108 bytes. Inside a task worktree

880 # the repo-relative path is already 126 bytes, and the only symptom is

881 # create-relay failing with a bare "Failed to listen on ..." -- which reads

882 # like a courier bug and is not one.

883 run_dir = args.run_dir or "/tmp/courier-single-poller-run"

884 shutil.rmtree(run_dir, ignore_errors=True)

885 os.makedirs(run_dir, exist_ok=True)

886

887 mock = Mock(run_dir)

888 instances = []

889 try:

890 a, b, state_dir = case_one_poller(args.bin, run_dir)

891 instances += [a, b]

892 case_relay_tools(args.bin, b, run_dir)

893 c = case_sigkill(args.bin, run_dir, a, b, state_dir)

894 instances.append(c)

895 pids = [a.pid, b.pid, c.pid]

896 c.close_stdin()

897 time.sleep(1.5)

898 case_log_hygiene(run_dir, pids)

899 for inst in instances:

900 inst.kill()

901 instances = []

902

903 case_production_launch_shape(args.bin, run_dir)

904 case_forced_takeover(args.bin, run_dir)

905 case_log_pruning(args.bin, run_dir)

906

907 dry_dir, text = case_dry_run(args.bin, run_dir)

908 case_gag_does_not_poison(args.bin, run_dir, dry_dir, text)

909

910 case_control(args.control_bin, os.path.join(run_dir, "control"))

911 finally:

912 for inst in instances:

913 inst.kill()

914 mock.stop()

915

916 print()

917 print(f"=== {len(PASSED)} passed, {len(FAILED)} failed ===")

918 if FAILED:

919 for f in FAILED:

920 print(f" FAILED: {f}")

921 print("SOME CHECKS FAILED")

922 return 1

923 print("ALL CHECKS PASSED")

924 return 0

925

926

927if __name__ == "__main__":

928 sys.exit(main())