fix: supervise fatal heartbeat storage errors

src/github_agent_bridge/executor.py

@@ -584,14 +584,27 @@ def _run_worker(
             self._loop(worker_id, retire_event)
         except BaseException as exc:
             self._set_worker_state(worker_id, "error", error=True)
-            self._record_worker_heartbeat(worker_id)
-            with self._worker_failure_lock:
-                self._worker_failures.append((worker_id, exc))
-            self._request_shutdown()
+            self._record_worker_failure(worker_id, exc)
         finally:
             if heartbeat_stop_event is not None:
                 heartbeat_stop_event.set()
 
+    def _run_heartbeat(
+        self,
+        worker_id: str,
+        heartbeat_stop_event: threading.Event,
+    ) -> None:
+        try:
+            self._heartbeat_loop(worker_id, heartbeat_stop_event)
+        except BaseException as exc:
+            self._set_worker_state(worker_id, "error", error=True)
+            self._record_worker_failure(f"{worker_id}/heartbeat", exc)
+
+    def _record_worker_failure(self, worker_id: str, exc: BaseException) -> None:
+        with self._worker_failure_lock:
+            self._worker_failures.append((worker_id, exc))
+        self._request_shutdown()
+
     def _spawn_worker(self) -> _WorkerSlot:
         with self._worker_slots_lock:
             worker_number = self._next_worker_number
@@ -601,7 +614,7 @@ def _spawn_worker(self) -> _WorkerSlot:
         heartbeat_stop_event = threading.Event()
         claim_lock = threading.Lock()
         heartbeat_thread = threading.Thread(
-            target=self._heartbeat_loop,
+            target=self._run_heartbeat,
             args=(worker_id, heartbeat_stop_event),
             daemon=True,
             name=f"gab-heartbeat-{worker_number}",

tests/test_executor.py

@@ -1152,6 +1152,29 @@ def fail_heartbeat(*args, **kwargs):
         pool._heartbeat_loop("worker-test")
 
 
+def test_run_stops_when_heartbeat_hits_fatal_storage_error(tmp_path, monkeypatch):
+    queue = JobQueue(tmp_path / "bridge.sqlite3")
+    pool = ExecutorPool(
+        queue,
+        Policy(),
+        RecordingDispatcher(),
+        config=ExecutorConfig(run_once=True),
+    )
+
+    def fail_heartbeat(*args, **kwargs):
+        raise sqlite3.OperationalError("disk I/O error")
+
+    monkeypatch.setattr(queue, "record_worker_heartbeat", fail_heartbeat)
+
+    with pytest.raises(
+        RuntimeError,
+        match=r"executor worker terminated unexpectedly: .*/heartbeat: OperationalError: disk I/O error",
+    ):
+        pool.run()
+
+    assert pool.stop_event.is_set() is True
+
+
 def test_acknowledgement_loop_does_not_hide_non_contention_database_errors(tmp_path, monkeypatch):
     queue = JobQueue(tmp_path / "bridge.sqlite3")
     pool = ExecutorPool(queue, Policy(), RecordingDispatcher())