From 7ed9ddb4a419d3067def68a0d16e0dcf11fc91c5 Mon Sep 17 00:00:00 2001 From: GISCE Bot Date: Thu, 8 Oct 2026 12:13:56 +0200 Subject: [PATCH 1/2] fix: retry contended session activity writes --- docs/persistence.md | 17 +++--- src/github_agent_bridge/dispatch.py | 54 +++++++++++++++-- src/github_agent_bridge/executor.py | 26 +++++++- tests/test_executor.py | 94 +++++++++++++++++++++++++++++ tests/test_modes_cli.py | 89 +++++++++++++++++++++++++++ 5 files changed, 266 insertions(+), 14 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index c8e133d..878de9b 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -29,13 +29,16 @@ Backup and restore are deliberately excluded. They use SQLite's backup API in `autoupdate.py` and remain specialized database operations rather than repository methods. -Executor heartbeat writes and acknowledgement claims treat `SQLITE_BUSY` and -`SQLITE_LOCKED` as transient contention after the connection timeout: they wait -and retry instead of terminating the background thread. Acknowledgement retries -stop before the external GitHub reaction, so ambiguous post-side-effect failures -still surface. Other `OperationalError` failures also propagate so schema or -storage faults are not hidden. Recovered heartbeat contention increments the -worker's persisted `recent_error_count` on the next successful heartbeat. +Executor heartbeat writes, acknowledgement claims and streamed session-event +writes treat `SQLITE_BUSY` and `SQLITE_LOCKED` as transient contention after the +connection timeout: they wait and retry instead of terminating the background +thread. Stream readers enqueue activity before persistence so a blocked SQLite +writer cannot stop draining the OpenClaw subprocess pipes. Acknowledgement +retries stop before the external GitHub reaction, so ambiguous post-side-effect +failures still surface. Other `OperationalError` failures also propagate so +schema or storage faults are not hidden. Recovered heartbeat or session-event +contention increments the worker's persisted `recent_error_count` on the next +successful heartbeat. ## Queue transaction boundaries diff --git a/src/github_agent_bridge/dispatch.py b/src/github_agent_bridge/dispatch.py index 7798ad7..08cd629 100644 --- a/src/github_agent_bridge/dispatch.py +++ b/src/github_agent_bridge/dispatch.py @@ -2,6 +2,7 @@ import json import os +import queue import re import signal import subprocess @@ -781,6 +782,24 @@ def dispatch( stdout_chunks: list[str] = [] stderr_chunks: list[str] = [] + activity_queue: queue.Queue[tuple[str, str, str] | None] | None = ( + queue.Queue() if activity_callback else None + ) + activity_errors: list[BaseException] = [] + + def deliver_activity() -> None: + assert activity_queue is not None + assert activity_callback is not None + while True: + activity = activity_queue.get() + if activity is None: + return + if activity_errors: + continue + try: + activity_callback(*activity) + except BaseException as exc: + activity_errors.append(exc) def read_stream(stream, chunks: list[str], event_type: str) -> None: if stream is None: @@ -791,28 +810,51 @@ def read_stream(stream, chunks: list[str], event_type: str) -> None: break chunk = data.decode("utf-8", errors="replace") chunks.append(chunk) - if activity_callback: - activity_callback(event_type, "OpenClaw CLI output" if event_type == "openclaw_stdout" else "OpenClaw CLI error output", chunk.rstrip("\n")) + if activity_queue is not None: + activity_queue.put( + ( + event_type, + "OpenClaw CLI output" + if event_type == "openclaw_stdout" + else "OpenClaw CLI error output", + chunk.rstrip("\n"), + ) + ) stdout_thread = threading.Thread(target=read_stream, args=(proc.stdout, stdout_chunks, "openclaw_stdout"), daemon=True) stderr_thread = threading.Thread(target=read_stream, args=(proc.stderr, stderr_chunks, "openclaw_stderr"), daemon=True) + activity_thread = ( + threading.Thread(target=deliver_activity, daemon=True) + if activity_queue is not None + else None + ) + if activity_thread is not None: + activity_thread.start() stdout_thread.start() stderr_thread.start() + + def finish_streaming() -> None: + stdout_thread.join() + stderr_thread.join() + if activity_queue is not None and activity_thread is not None: + activity_queue.put(None) + activity_thread.join() + if activity_errors: + raise activity_errors[0] + try: try: # Let OpenClaw's own --timeout own the agent run deadline. The bridge only # keeps a small grace window so it can capture the CLI result cleanly. proc.wait(timeout=agent_timeout + self.cli_grace_seconds) - stdout_thread.join(timeout=1) - stderr_thread.join(timeout=1) + finish_streaming() out, err = "".join(stdout_chunks), "".join(stderr_chunks) cancelled = self._shutdown_event.is_set() and proc.returncode != 0 return DispatchResult(proc.returncode == 0, proc.returncode, (out or "")[:2000], (err or "")[:4000], False, reaction_ok, cmd, cancelled) except subprocess.TimeoutExpired: self._signal_process_group(proc, signal.SIGKILL) proc.wait() - stdout_thread.join(timeout=1) - stderr_thread.join(timeout=1) + finish_streaming() out, err = "".join(stdout_chunks), "".join(stderr_chunks) return DispatchResult(False, 124, (out or "")[:2000], (err or "")[:4000], True, reaction_ok, cmd) finally: diff --git a/src/github_agent_bridge/executor.py b/src/github_agent_bridge/executor.py index ed1735e..0c9577f 100644 --- a/src/github_agent_bridge/executor.py +++ b/src/github_agent_bridge/executor.py @@ -97,6 +97,28 @@ def _record_worker_storage_error(self, worker_id: str) -> None: ) self._worker_states[worker_id] = (loop_state, active_job_id, errors + 1) + def _record_session_activity( + self, + worker_id: str, + job_id: int, + event_type: str, + summary: str, + detail: str | None, + ) -> bool: + while not self.stop_event.is_set(): + try: + self.queue.add_session_event( + job_id, event_type, summary, redact_event_detail(detail) + ) + return True + except sqlite3.OperationalError as exc: + if not _is_sqlite_contention_error(exc): + raise + self._record_worker_storage_error(worker_id) + if self.stop_event.wait(self.config.idle_sleep_seconds): + return False + return False + def _heartbeat_loop(self, worker_id: str) -> None: while not self.stop_event.is_set(): try: @@ -342,7 +364,9 @@ def work_one(self, worker_id: str | None = None) -> bool: job, self.policy, reaction_ok=reaction_ok, - activity_callback=lambda event_type, summary, detail: self.queue.add_session_event(job.id, event_type, summary, redact_event_detail(detail)), + activity_callback=lambda event_type, summary, detail: self._record_session_activity( + worker_id, job.id, event_type, summary, detail + ), process_callback=lambda identity: self.queue.register_runtime_process( job.id, worker_id, diff --git a/tests/test_executor.py b/tests/test_executor.py index 75fee60..e7c187d 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -351,6 +351,100 @@ def test_executor_records_session_activity_events(tmp_path): assert stderr_event["detail"] == "token=[redacted] [redacted]" +def test_session_activity_recovers_after_transient_database_lock(tmp_path): + class PausingActivityDispatcher(RecordingDispatcher): + def __init__(self): + super().__init__() + self.ready = threading.Event() + self.resume = threading.Event() + + def dispatch( + self, + job, + policy, + reaction_ok=None, + activity_callback=None, + process_callback=None, + ): + self.jobs.append(job) + if process_callback: + process_callback( + { + "pid": 456, + "ppid": 123, + "pgid": 456, + "sid": 456, + "start_time_ticks": 999, + } + ) + self.ready.set() + assert self.resume.wait(timeout=2) + assert activity_callback is not None + activity_callback( + "openclaw_stdout", "OpenClaw CLI output", "persist after contention" + ) + return DispatchResult(True, 0, "ok", "", False, reaction_ok, ["openclaw"]) + + db = tmp_path / "bridge.sqlite3" + queue = JobQueue(db) + job = enqueue_pr_comment(queue) + queue.database.timeout_seconds = 0.01 + queue.database.busy_timeout_ms = 10 + dispatcher = PausingActivityDispatcher() + pool = ExecutorPool( + queue, + Policy(trusted_orgs={"gisce"}), + dispatcher, + github=FakeGitHub(assigned=True), + config=ExecutorConfig(run_once=True, idle_sleep_seconds=0.01), + ) + worker = threading.Thread(target=pool.work_one, args=("worker-test",)) + worker.start() + assert dispatcher.ready.wait(timeout=2) + locker = sqlite3.connect(db, isolation_level=None) + locker.execute("BEGIN IMMEDIATE") + try: + dispatcher.resume.set() + worker.join(timeout=0.1) + assert worker.is_alive() is True + finally: + locker.commit() + locker.close() + worker.join(timeout=2) + + assert worker.is_alive() is False + stored = queue.get(job.id) + assert stored is not None + assert stored.status == "done" + events = job_session_events(db, job.id) + assert any( + event["event_type"] == "openclaw_stdout" + and event["detail"] == "persist after contention" + for event in events + ) + + +def test_session_activity_does_not_hide_non_contention_database_errors( + tmp_path, monkeypatch +): + queue = JobQueue(tmp_path / "bridge.sqlite3") + pool = ExecutorPool(queue, Policy(), RecordingDispatcher()) + + def fail_session_event(*args, **kwargs): + raise sqlite3.OperationalError("no such table: job_session_events") + + monkeypatch.setattr(queue, "add_session_event", fail_session_event) + + with pytest.raises(sqlite3.OperationalError, match="no such table"): + pool._record_session_activity( + "worker-test", + 1, + "openclaw_stdout", + "OpenClaw CLI output", + "detail", + ) + + def test_dispatched_job_completion_pushes_trigger_actor(tmp_path, monkeypatch): queue = JobQueue(tmp_path / "bridge.sqlite3") job, state = enqueue_pr_comment_from(queue, "ecarreras", 1, "", 1) diff --git a/tests/test_modes_cli.py b/tests/test_modes_cli.py index 1a44577..4d927c2 100644 --- a/tests/test_modes_cli.py +++ b/tests/test_modes_cli.py @@ -386,6 +386,95 @@ def test_live_dispatch_streams_openclaw_output_to_activity_callback(tmp_path): assert processes[0]["start_time_ticks"] > 0 +def test_live_dispatch_propagates_activity_callback_failure(tmp_path): + openclaw = tmp_path / "openclaw" + openclaw.write_text( + "#!/bin/sh\n" + "printf 'thinking line\\n'\n" + "printf 'final line\\n'\n", + encoding="utf-8", + ) + openclaw.chmod(0o755) + + def fail_activity_callback(event_type, summary, detail): + raise RuntimeError("activity persistence failed") + + dispatcher = OpenClawDispatcher( + openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=1 + ) + with pytest.raises(RuntimeError, match="activity persistence failed"): + dispatcher.dispatch( + make_job(), + Policy(trusted_orgs={"gisce"}), + reaction_ok=True, + activity_callback=fail_activity_callback, + ) + + +def test_live_dispatch_drains_process_while_activity_callback_is_blocked( + tmp_path, monkeypatch +): + done = tmp_path / "done" + openclaw = tmp_path / "openclaw" + openclaw.write_text( + "#!/usr/bin/env python3\n" + "import os\n" + "import pathlib\n" + "import sys\n" + "sys.stdout.write('first chunk\\n')\n" + "sys.stdout.flush()\n" + "sys.stdout.write('x' * 1048576)\n" + "sys.stdout.flush()\n" + "pathlib.Path(os.environ['DONE_FILE']).write_text('done', encoding='utf-8')\n", + encoding="utf-8", + ) + openclaw.chmod(0o755) + monkeypatch.setenv("DONE_FILE", str(done)) + callback_started = threading.Event() + release_callback = threading.Event() + results = [] + errors = [] + + def block_first_activity(event_type, summary, detail): + if not callback_started.is_set(): + callback_started.set() + assert release_callback.wait(timeout=2) + + dispatcher = OpenClawDispatcher( + openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=1 + ) + + def dispatch(): + try: + results.append( + dispatcher.dispatch( + make_job(), + Policy(trusted_orgs={"gisce"}), + reaction_ok=True, + activity_callback=block_first_activity, + ) + ) + except BaseException as exc: + errors.append(exc) + + thread = threading.Thread(target=dispatch) + thread.start() + try: + assert callback_started.wait(timeout=1) + deadline = time.monotonic() + 1 + while not done.exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert done.exists() is True + finally: + release_callback.set() + thread.join(timeout=2) + + assert thread.is_alive() is False + assert errors == [] + assert len(results) == 1 + assert results[0].ok is True + + def test_live_dispatch_streams_partial_openclaw_output_before_process_exits(tmp_path, monkeypatch): done = tmp_path / "done" openclaw = tmp_path / "openclaw" From 2c090bb889f70f2432e5e46c5bef72cde097188e Mon Sep 17 00:00:00 2001 From: GISCE Bot Date: Thu, 8 Oct 2026 12:46:01 +0200 Subject: [PATCH 2/2] fix: bound stream drain after cli exit Co-authored-by: pilipilisbot --- docs/persistence.md | 14 +++++++------ src/github_agent_bridge/dispatch.py | 25 ++++++++++++++++++++--- tests/test_modes_cli.py | 31 +++++++++++++++++++++++++++++ 3 files changed, 61 insertions(+), 9 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index 878de9b..5d8fffa 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -33,12 +33,14 @@ Executor heartbeat writes, acknowledgement claims and streamed session-event writes treat `SQLITE_BUSY` and `SQLITE_LOCKED` as transient contention after the connection timeout: they wait and retry instead of terminating the background thread. Stream readers enqueue activity before persistence so a blocked SQLite -writer cannot stop draining the OpenClaw subprocess pipes. Acknowledgement -retries stop before the external GitHub reaction, so ambiguous post-side-effect -failures still surface. Other `OperationalError` failures also propagate so -schema or storage faults are not hidden. Recovered heartbeat or session-event -contention increments the worker's persisted `recent_error_count` on the next -successful heartbeat. +writer cannot stop draining the OpenClaw subprocess pipes. Once the main CLI +process exits, readers get a bounded drain window before the bridge stops them +and closes its pipe ends; a descendant that inherited stdout or stderr cannot +hold the worker indefinitely. Acknowledgement retries stop before the external +GitHub reaction, so ambiguous post-side-effect failures still surface. Other +`OperationalError` failures also propagate so schema or storage faults are not +hidden. Recovered heartbeat or session-event contention increments the worker's +persisted `recent_error_count` on the next successful heartbeat. ## Queue transaction boundaries diff --git a/src/github_agent_bridge/dispatch.py b/src/github_agent_bridge/dispatch.py index 08cd629..31eb750 100644 --- a/src/github_agent_bridge/dispatch.py +++ b/src/github_agent_bridge/dispatch.py @@ -4,9 +4,11 @@ import os import queue import re +import select import signal import subprocess import threading +import time from dataclasses import dataclass from importlib import resources from enum import StrEnum @@ -786,6 +788,7 @@ def dispatch( queue.Queue() if activity_callback else None ) activity_errors: list[BaseException] = [] + stop_stream_readers = threading.Event() def deliver_activity() -> None: assert activity_queue is not None @@ -804,10 +807,15 @@ def deliver_activity() -> None: def read_stream(stream, chunks: list[str], event_type: str) -> None: if stream is None: return - while True: + while not stop_stream_readers.is_set(): + ready, _, _ = select.select([stream], [], [], 0.1) + if not ready: + continue data = os.read(stream.fileno(), 4096) if not data: break + if stop_stream_readers.is_set(): + break chunk = data.decode("utf-8", errors="replace") chunks.append(chunk) if activity_queue is not None: @@ -834,8 +842,19 @@ def read_stream(stream, chunks: list[str], event_type: str) -> None: stderr_thread.start() def finish_streaming() -> None: - stdout_thread.join() - stderr_thread.join() + drain_deadline = time.monotonic() + 1 + for stream_thread in (stdout_thread, stderr_thread): + stream_thread.join( + timeout=max(0.0, drain_deadline - time.monotonic()) + ) + stop_stream_readers.set() + for stream, stream_thread in ( + (proc.stdout, stdout_thread), + (proc.stderr, stderr_thread), + ): + stream_thread.join(timeout=0.5) + if not stream_thread.is_alive() and stream is not None: + stream.close() if activity_queue is not None and activity_thread is not None: activity_queue.put(None) activity_thread.join() diff --git a/tests/test_modes_cli.py b/tests/test_modes_cli.py index 4d927c2..d75a6ef 100644 --- a/tests/test_modes_cli.py +++ b/tests/test_modes_cli.py @@ -475,6 +475,37 @@ def dispatch(): assert results[0].ok is True +def test_live_dispatch_bounds_pipe_drain_when_descendant_inherits_streams(tmp_path): + openclaw = tmp_path / "openclaw" + openclaw.write_text( + "#!/usr/bin/env python3\n" + "import subprocess\n" + "import sys\n" + "subprocess.Popen([\n" + " sys.executable,\n" + " '-c',\n" + " 'import time; " + "[(time.sleep(0.1), print(\"child\", flush=True)) for _ in range(50)]',\n" + "])\n" + "print('parent complete', flush=True)\n", + encoding="utf-8", + ) + openclaw.chmod(0o755) + dispatcher = OpenClawDispatcher( + openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=0.1 + ) + + started = time.monotonic() + result = dispatcher.dispatch( + make_job(), Policy(trusted_orgs={"gisce"}), reaction_ok=True + ) + elapsed = time.monotonic() - started + + assert result.ok is True + assert "parent complete" in result.stdout + assert elapsed < 3 + + def test_live_dispatch_streams_partial_openclaw_output_before_process_exits(tmp_path, monkeypatch): done = tmp_path / "done" openclaw = tmp_path / "openclaw"