diff --git a/scripts/buzz-codex-bridge/README.md b/scripts/buzz-codex-bridge/README.md index 1407d42a30..90f860b653 100644 --- a/scripts/buzz-codex-bridge/README.md +++ b/scripts/buzz-codex-bridge/README.md @@ -42,7 +42,8 @@ python3 scripts/buzz-codex-bridge/bridge.py --config /private/routes.json --stat ``` The bridge remains in the foreground. Stop with Ctrl-C. It cleans up its watch -children. No login daemon or scheduled task is installed automatically. Use the +process groups, including descendants of authentication wrappers. Codex queue +commands use the same containment on exit, timeout, or cancellation. No login daemon or scheduled task is installed automatically. Use the same state DB after restart; a different routing configuration requires a new DB. The DB and lock are created owner-only. Treat the DB as private conversation data. diff --git a/scripts/buzz-codex-bridge/bridge.py b/scripts/buzz-codex-bridge/bridge.py index cb7d71c742..2842f7a123 100644 --- a/scripts/buzz-codex-bridge/bridge.py +++ b/scripts/buzz-codex-bridge/bridge.py @@ -8,6 +8,7 @@ import os from pathlib import Path import re import selectors +import signal import sqlite3 import subprocess import sys @@ -112,24 +113,42 @@ def deliver(db, cfg): # An interruption after queue acceptance is ambiguous. Do not blindly replay it. with db: db.execute("UPDATE deliveries SET state='sending' WHERE id=?", (eid,)) - result = subprocess.run(cfg['codex_command'] + ['queue', '--thread', route['thread'], '--message', message], - stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, timeout=30, check=False) - if result.returncode: - raise RuntimeError(f'Codex queue exited {result.returncode}; delivery {eid} requires reconciliation') + returncode = run_queue(cfg['codex_command'] + ['queue', '--thread', route['thread'], '--message', message]) + if returncode: + raise RuntimeError(f'Codex queue exited {returncode}; delivery {eid} requires reconciliation') with db: db.execute("UPDATE deliveries SET state='sent' WHERE id=?", (eid,)) print(json.dumps({'delivered': eid, 'thread': route['thread']}), flush=True) def stop(process): - if process.poll() is None: - process.terminate() + # Every owned command starts a new session. Its group survives wrapper exit. + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + finally: try: - process.wait(timeout=5) - except subprocess.TimeoutExpired: - process.kill() - process.wait() - process.stdout.close() + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait() + if process.stdout is not None: + process.stdout.close() + + +def run_queue(argv, timeout=30): + process = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + start_new_session=True) + try: + return process.wait(timeout=timeout) + finally: + # Includes timeout, cancellation, and wrappers exiting before descendants. + stop(process) def run(cfg, db): @@ -152,7 +171,7 @@ def run(cfg, db): cmd = cfg['buzz_command'] + ['--relay', cfg['relay'], 'messages', 'watch', '--channel', route['channel'], '--since', since] for author in route['authors']: cmd += ['--author', author] - p = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE) + p = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, start_new_session=True) worker.update(process=p, buffer=b'', started=time.monotonic()) sel.register(p.stdout, selectors.EVENT_READ, worker) for key, _ in sel.select(timeout=1): diff --git a/scripts/buzz-codex-bridge/test_bridge.py b/scripts/buzz-codex-bridge/test_bridge.py index 30ed9ba7d2..f5613e2324 100644 --- a/scripts/buzz-codex-bridge/test_bridge.py +++ b/scripts/buzz-codex-bridge/test_bridge.py @@ -1,5 +1,6 @@ import copy import json +import time from pathlib import Path import sys import selectors @@ -31,8 +32,8 @@ class BridgeTest(unittest.TestCase): def test_replay_after_restart_delivers_only_once(self): bridge.ingest(self.db, self.route, self.event) - with patch('bridge.subprocess.run') as run: - run.return_value.returncode = 0 + with patch('bridge.run_queue') as run: + run.return_value = 0 bridge.deliver(self.db, self.cfg) self.assertEqual(run.call_count, 1) args = run.call_args.args[0] @@ -48,8 +49,8 @@ class BridgeTest(unittest.TestCase): reopened.close() def test_oversized_report_is_journaled_and_does_not_block_replay_or_summary(self): - with patch('bridge.subprocess.run') as run: - run.return_value.returncode = 0 + with patch('bridge.run_queue') as run: + run.return_value = 0 oversized = dict(self.event, content='x' * 60001) bridge.ingest(self.db, self.route, oversized) bridge.deliver(self.db, self.cfg) @@ -69,7 +70,7 @@ class BridgeTest(unittest.TestCase): def test_ambiguous_submission_is_not_retried(self): bridge.ingest(self.db, self.route, self.event) - with patch('bridge.subprocess.run', side_effect=TimeoutError('queue timed out')): + with patch('bridge.run_queue', side_effect=TimeoutError('queue timed out')): with self.assertRaises(TimeoutError): bridge.deliver(self.db, self.cfg) with self.assertRaisesRegex(RuntimeError, 'ambiguous'): @@ -114,6 +115,46 @@ class BridgeTest(unittest.TestCase): self.assertIn(self.event['content'], args[-1]) self.assertEqual(self.db.execute('SELECT state FROM deliveries').fetchone()[0], 'sent') + def process_tree_command(self): + child = self.root / 'child.py' + child.write_text('import os,signal,time\nfrom pathlib import Path\n' + 'signal.signal(signal.SIGTERM, signal.SIG_IGN)\n' + 'Path(' + repr(str(self.root / 'child.pid')) + ').write_text(str(os.getpid()))\n' + 'time.sleep(60)\n') + wrapper = self.root / 'wrapper.py' + wrapper.write_text('import subprocess,sys,time\n' + 'subprocess.Popen([sys.executable, ' + repr(str(child)) + '])\n' + 'time.sleep(60)\n') + return [sys.executable, str(wrapper)] + + def assert_child_stopped(self): + pid = int((self.root / 'child.pid').read_text()) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + status = subprocess.run(['ps', '-o', 'stat=', '-p', str(pid)], capture_output=True, text=True) + if status.returncode != 0 or status.stdout.strip().startswith('Z'): + return + time.sleep(0.02) + self.fail('wrapper descendant is still running') + + def test_stop_kills_wrapper_descendant_that_ignores_termination(self): + process = subprocess.Popen(self.process_tree_command(), stdout=subprocess.PIPE, start_new_session=True) + try: + deadline = time.monotonic() + 5 + while not (self.root / 'child.pid').exists() and time.monotonic() < deadline: + time.sleep(0.02) + self.assertTrue((self.root / 'child.pid').exists()) + bridge.stop(process) + self.assert_child_stopped() + finally: + if process.poll() is None: + bridge.stop(process) + + def test_queue_timeout_kills_descendants(self): + with self.assertRaises(subprocess.TimeoutExpired): + bridge.run_queue(self.process_tree_command(), timeout=2) + self.assert_child_stopped() + def test_foreground_bridge_streams_and_stops_its_watcher(self): watcher = self.root / 'watcher.py' watcher.write_text('import json,time\n' +