fix: recover contended session activity writes

docs/persistence.md

@@ -29,13 +29,18 @@ 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. 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
 

src/github_agent_bridge/dispatch.py

@@ -2,10 +2,13 @@
 
 import json
 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
@@ -781,38 +784,96 @@ 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] = []
+        stop_stream_readers = threading.Event()
+
+        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:
                 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_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:
+            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()
+            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:

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,

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, "<gisce/erp/pull/27315/ecarreras@github.com>", 1)