From 1e33cae9a7adc16d861b705a6fef4de2f5f0733d Mon Sep 17 00:00:00 2001 From: Dhruv Raajeev Date: Wed, 22 Jul 2026 17:21:14 -0500 Subject: [PATCH] fix(gateway,tools,agent): close leaked SQLite connections in delivery, delegation, and verification ledgers Three durable ledgers used `with _connect() as conn:` where the sqlite3 connection context manager commits/rolls back but never closes, leaking the db/-wal/-shm file descriptors on every call. On a long-running gateway this exhausts RLIMIT_NOFILE and fails unrelated components with `[Errno 24] Too many open files`. Same bug class as the cron execution ledger (#69567 / PR #69594), which the connection helpers here are modeled on. Fix: route every ledger operation through a `_transaction()` context manager that guarantees `conn.close()` on exit. `_connect()` keeps its schema-on-connect contract (several tests call it directly) and now self-closes if schema init fails. Adds per-module regression tests asserting every opened connection is closed, including the no-op-update and exception-mid-transaction paths. Co-Authored-By: Claude Opus 4.8 --- agent/verification_evidence.py | 40 ++++- gateway/delivery_ledger.py | 45 +++++- .../test_verification_evidence_fd_leak.py | 126 ++++++++++++++++ tests/gateway/test_delivery_ledger_fd_leak.py | 137 ++++++++++++++++++ tests/tools/test_async_delegation_fd_leak.py | 133 +++++++++++++++++ tools/async_delegation.py | 60 ++++++-- 6 files changed, 512 insertions(+), 29 deletions(-) create mode 100644 tests/agent/test_verification_evidence_fd_leak.py create mode 100644 tests/gateway/test_delivery_ledger_fd_leak.py create mode 100644 tests/tools/test_async_delegation_fd_leak.py diff --git a/agent/verification_evidence.py b/agent/verification_evidence.py index d66a1534045ce..5c99ba62fef4b 100644 --- a/agent/verification_evidence.py +++ b/agent/verification_evidence.py @@ -13,10 +13,11 @@ import sqlite3 import tempfile import threading +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Optional +from typing import Any, Iterator, Optional from hermes_constants import get_hermes_home @@ -63,13 +64,38 @@ def _connect() -> sqlite3.Connection: path = _db_path() path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path) - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA busy_timeout=5000") conn.row_factory = sqlite3.Row - _ensure_schema(conn) + try: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + _ensure_schema(conn) + except Exception: + # A PRAGMA/DDL failure after a successful connect() must not leak the + # just-opened connection back to the caller. + conn.close() + raise return conn +@contextmanager +def _transaction() -> Iterator[sqlite3.Connection]: + """Open a connection, commit/rollback on exit, and ALWAYS close it. + + ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the + transaction; they do not close the connection. Using ``with _connect()`` + alone therefore leaks a connection — and its WAL/SHM file descriptors — on + every call, deferring the close to the garbage collector, which over a + long-running process can exhaust ``RLIMIT_NOFILE`` (the cron-ledger sibling + of this bug was #69567 / PR #69594). + """ + conn = _connect() + try: + with conn: + yield conn + finally: + conn.close() + + def _ensure_schema(conn: sqlite3.Connection) -> None: conn.execute( """ @@ -452,7 +478,7 @@ def record_terminal_result( created_at = _utc_now() with _DB_LOCK: - with _connect() as conn: + with _transaction() as conn: cur = conn.execute( """ INSERT INTO verification_events( @@ -518,7 +544,7 @@ def mark_workspace_edited( edited_at = _utc_now() with _DB_LOCK: - with _connect() as conn: + with _transaction() as conn: row = conn.execute( """ SELECT changed_paths_json FROM verification_state @@ -568,7 +594,7 @@ def verification_status( sid = str(session_id or "default") root = str(facts.get("root") or Path(cwd or ".").resolve()) with _DB_LOCK: - with _connect() as conn: + with _transaction() as conn: state = conn.execute( """ SELECT last_event_id, last_edit_at, changed_paths_json diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index 955e1d1d3e95e..2c4e3d8de639f 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -46,7 +46,8 @@ import sqlite3 import threading import time -from typing import Any, Dict, List, Optional +from contextlib import contextmanager +from typing import Any, Dict, Iterator, List, Optional from hermes_constants import get_hermes_home @@ -78,6 +79,17 @@ def _connect() -> sqlite3.Connection: path = _db_path() path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path, timeout=10) + try: + _initialize_schema(conn) + except Exception: + # A PRAGMA/DDL failure after a successful connect() must not leak the + # just-opened connection back to the caller. + conn.close() + raise + return conn + + +def _initialize_schema(conn: sqlite3.Connection) -> None: conn.execute("PRAGMA journal_mode=WAL") conn.execute( """CREATE TABLE IF NOT EXISTS delivery_obligations ( @@ -96,7 +108,26 @@ def _connect() -> sqlite3.Connection: last_error TEXT )""" ) - return conn + + +@contextmanager +def _transaction() -> Iterator[sqlite3.Connection]: + """Open a connection, commit/rollback on exit, and ALWAYS close it. + + ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the + transaction; they do not close the connection. Using ``with _connect()`` + alone therefore leaks a connection — and its WAL/SHM file descriptors — on + every call, deferring the close to the garbage collector. On a long-running + gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of this + bug was #69567 / PR #69594). ``record_obligation`` runs on every outbound + final response, so this ledger is the highest-frequency leaker. + """ + conn = _connect() + try: + with conn: + yield conn + finally: + conn.close() def _owner_stamp() -> tuple[int, Optional[int]]: @@ -164,7 +195,7 @@ def record_obligation( """Record a final response as owed to the platform (state='pending').""" now = time.time() pid, started = _owner_stamp() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( """INSERT OR REPLACE INTO delivery_obligations (obligation_id, session_key, platform, chat_id, thread_id, @@ -191,7 +222,7 @@ def mark_failed(obligation_id: str, error: str = "") -> None: def _update_state(obligation_id: str, state: str, error: str = "") -> None: - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( """UPDATE delivery_obligations SET state=?, updated_at=?, last_error=? @@ -224,7 +255,7 @@ def sweep_recoverable( now = now if now is not None else time.time() pid, started = _owner_stamp() claimed: List[Dict[str, Any]] = [] - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: rows = conn.execute( """SELECT obligation_id, session_key, platform, chat_id, thread_id, content, state, attempts, created_at, @@ -277,7 +308,7 @@ def _prune(now: Optional[float] = None) -> None: now = now if now is not None else time.time() cutoff = now - _RETENTION_SECONDS try: - with _connect() as conn: + with _transaction() as conn: conn.execute( """DELETE FROM delivery_obligations WHERE state IN ('delivered', 'abandoned') AND updated_at < ?""", @@ -321,7 +352,7 @@ def ledger_enabled(config: Optional[Dict[str, Any]] = None) -> bool: def debug_rows(limit: int = 20) -> str: """Human-readable dump for ad-hoc inspection (sqlite3-free path).""" - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: rows = conn.execute( """SELECT obligation_id, session_key, state, attempts, created_at, updated_at, last_error diff --git a/tests/agent/test_verification_evidence_fd_leak.py b/tests/agent/test_verification_evidence_fd_leak.py new file mode 100644 index 0000000000000..e973aabb9e089 --- /dev/null +++ b/tests/agent/test_verification_evidence_fd_leak.py @@ -0,0 +1,126 @@ +"""Regression: the verification-evidence ledger must close every connection. + +Sibling of the cron execution-ledger leak (#69567 / PR #69594). The evidence +ledger used ``with _connect() as conn:`` where the connection context manager +commits/rolls back but never closes, leaking the db/-wal/-shm file descriptors +on every recorded terminal result, workspace edit, and status read. These tests +fail if the deterministic ``close()`` is ever removed again. +""" + +import sqlite3 + +import pytest + +from agent import verification_evidence as ve + + +class _TrackingConnection: + """Delegates to a real sqlite3.Connection while recording close() calls. + + sqlite3.Connection is a static C type: it has no per-instance __dict__ and + its methods can't be monkeypatched, so open/close tracking is done via a + delegating wrapper returned in place of the real connection. + """ + + def __init__(self, real, closed_ids): + object.__setattr__(self, "_real", real) + object.__setattr__(self, "_closed_ids", closed_ids) + + def close(self): + self._closed_ids.append(id(self._real)) + self._real.close() + + def __enter__(self): + self._real.__enter__() + return self + + def __exit__(self, exc_type, exc, tb): + return self._real.__exit__(exc_type, exc, tb) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __setattr__(self, name, value): + setattr(self._real, name, value) + + +def _point_ledger(monkeypatch, tmp_path): + monkeypatch.setattr(ve, "_db_path", lambda: tmp_path / "verification_evidence.db") + return ve + + +def _track_connections(monkeypatch): + opened, closed = [], [] + real_connect = sqlite3.connect + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _TrackingConnection(conn, closed) + + monkeypatch.setattr(ve.sqlite3, "connect", tracking_connect) + return opened, closed + + +def _python_project(root): + (root / "pyproject.toml").write_text("[tool.pytest.ini_options]\n") + + +def test_ledger_operations_close_every_connection(monkeypatch, tmp_path): + """Recording, editing, and status reads must close every connection opened.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + _point_ledger(monkeypatch, tmp_path) + _python_project(tmp_path) + opened, closed = _track_connections(monkeypatch) + + ve.record_terminal_result( + command="python -m pytest tests/test_calc.py::test_even -q", + cwd=tmp_path, session_id="s1", exit_code=0, output="1 passed", + ) + ve.verification_status(session_id="s1", cwd=tmp_path) + ve.mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=["mod.py"]) + + assert opened, "expected at least one connection to be opened" + assert len(opened) == len(closed) + assert set(opened) == set(closed) + + +def test_exception_during_operation_still_closes_connection(monkeypatch, tmp_path): + """A failing statement inside the transaction must roll back and close.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + with pytest.raises(sqlite3.IntegrityError): + with ve._transaction() as conn: + # Missing NOT NULL columns -> constraint failure inside the block. + conn.execute("INSERT INTO verification_events (id) VALUES (1)") + + assert len(opened) == 1 + assert len(closed) == 1 + + +def test_schema_init_failure_still_closes_connection(monkeypatch, tmp_path): + """A PRAGMA/DDL failure after connect() must still close the connection.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = [], [] + real_connect = sqlite3.connect + + class _FailingSchemaConnection(_TrackingConnection): + def execute(self, sql, *args, **kwargs): + if "CREATE TABLE" in sql: + raise sqlite3.OperationalError("simulated schema init failure") + return self._real.execute(sql, *args, **kwargs) + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _FailingSchemaConnection(conn, closed) + + monkeypatch.setattr(ve.sqlite3, "connect", tracking_connect) + + with pytest.raises(sqlite3.OperationalError): + with ve._transaction(): + pass + + assert len(opened) == 1 + assert len(closed) == 1 diff --git a/tests/gateway/test_delivery_ledger_fd_leak.py b/tests/gateway/test_delivery_ledger_fd_leak.py new file mode 100644 index 0000000000000..02cde134f4c9d --- /dev/null +++ b/tests/gateway/test_delivery_ledger_fd_leak.py @@ -0,0 +1,137 @@ +"""Regression: the gateway delivery ledger must close every SQLite connection. + +Sibling of the cron execution-ledger leak (#69567 / PR #69594). The ledger used +``with _connect() as conn:`` where ``sqlite3.Connection.__exit__`` commits or +rolls back but never closes, leaking the db/-wal/-shm file descriptors on every +call until a long-running gateway exhausts ``RLIMIT_NOFILE``. ``record_obligation`` +runs on every outbound final response, so this is the highest-frequency leaker of +the set. These tests fail if the deterministic ``close()`` is ever removed again. +""" + +import sqlite3 + +import pytest + +from gateway import delivery_ledger as dl + + +class _TrackingConnection: + """Delegates to a real sqlite3.Connection while recording close() calls. + + sqlite3.Connection is a static C type: it has no per-instance __dict__ and + its methods can't be monkeypatched, so open/close tracking is done via a + delegating wrapper returned in place of the real connection. + """ + + def __init__(self, real, closed_ids): + object.__setattr__(self, "_real", real) + object.__setattr__(self, "_closed_ids", closed_ids) + + def close(self): + self._closed_ids.append(id(self._real)) + self._real.close() + + def __enter__(self): + self._real.__enter__() + return self + + def __exit__(self, exc_type, exc, tb): + return self._real.__exit__(exc_type, exc, tb) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __setattr__(self, name, value): + setattr(self._real, name, value) + + +def _point_ledger(monkeypatch, tmp_path): + monkeypatch.setattr(dl, "_db_path", lambda: tmp_path / "state.db") + return dl + + +def _track_connections(monkeypatch): + opened, closed = [], [] + real_connect = sqlite3.connect + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _TrackingConnection(conn, closed) + + monkeypatch.setattr(dl.sqlite3, "connect", tracking_connect) + return opened, closed + + +def test_ledger_operations_close_every_connection(monkeypatch, tmp_path): + """Every public ledger operation must close the connection it opened.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + oid = dl.compute_obligation_id("sess", "msg", "content") + dl.record_obligation( + obligation_id=oid, session_key="sess", platform="telegram", + chat_id="123", thread_id=None, content="hello", + ) + dl.mark_attempting(oid) + dl.mark_delivered(oid) + dl.sweep_recoverable() + dl.debug_rows() + + assert opened, "expected at least one connection to be opened" + assert len(opened) == len(closed) + assert set(opened) == set(closed) + + +def test_early_return_still_closes_connection(monkeypatch, tmp_path): + """A no-op update (no matching row) must still open and close exactly once.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + dl.mark_delivered("does-not-exist") + + assert len(opened) == 1 + assert len(closed) == 1 + + +def test_exception_during_operation_still_closes_connection(monkeypatch, tmp_path): + """A failing statement inside the transaction must roll back and close.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + with pytest.raises(sqlite3.IntegrityError): + with dl._transaction() as conn: + # Missing NOT NULL columns -> constraint failure inside the block. + conn.execute( + "INSERT INTO delivery_obligations (obligation_id) VALUES ('x')" + ) + + assert len(opened) == 1 + assert len(closed) == 1 + + +def test_schema_init_failure_still_closes_connection(monkeypatch, tmp_path): + """A PRAGMA/DDL failure after connect() must still close the connection.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = [], [] + real_connect = sqlite3.connect + + class _FailingSchemaConnection(_TrackingConnection): + def execute(self, sql, *args, **kwargs): + if "CREATE TABLE" in sql: + raise sqlite3.OperationalError("simulated schema init failure") + return self._real.execute(sql, *args, **kwargs) + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _FailingSchemaConnection(conn, closed) + + monkeypatch.setattr(dl.sqlite3, "connect", tracking_connect) + + with pytest.raises(sqlite3.OperationalError): + with dl._transaction(): + pass + + assert len(opened) == 1 + assert len(closed) == 1 diff --git a/tests/tools/test_async_delegation_fd_leak.py b/tests/tools/test_async_delegation_fd_leak.py new file mode 100644 index 0000000000000..ddc51f8984a6c --- /dev/null +++ b/tests/tools/test_async_delegation_fd_leak.py @@ -0,0 +1,133 @@ +"""Regression: the async-delegation ledger must close every SQLite connection. + +Sibling of the cron execution-ledger leak (#69567 / PR #69594). The durable +delegation ledger used ``with _connect() as conn:`` where the connection +context manager commits/rolls back but never closes, leaking the db/-wal/-shm +file descriptors on every dispatch, completion, and delivery-claim. These tests +fail if the deterministic ``close()`` is ever removed again. +""" + +import queue +import sqlite3 + +import pytest + +from tools import async_delegation as ad + + +class _TrackingConnection: + """Delegates to a real sqlite3.Connection while recording close() calls. + + sqlite3.Connection is a static C type: it has no per-instance __dict__ and + its methods can't be monkeypatched, so open/close tracking is done via a + delegating wrapper returned in place of the real connection. + """ + + def __init__(self, real, closed_ids): + object.__setattr__(self, "_real", real) + object.__setattr__(self, "_closed_ids", closed_ids) + + def close(self): + self._closed_ids.append(id(self._real)) + self._real.close() + + def __enter__(self): + self._real.__enter__() + return self + + def __exit__(self, exc_type, exc, tb): + return self._real.__exit__(exc_type, exc, tb) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __setattr__(self, name, value): + setattr(self._real, name, value) + + +def _point_ledger(monkeypatch, tmp_path): + monkeypatch.setattr(ad, "_db_path", lambda: tmp_path / "state.db") + return ad + + +def _track_connections(monkeypatch): + opened, closed = [], [] + real_connect = sqlite3.connect + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _TrackingConnection(conn, closed) + + monkeypatch.setattr(ad.sqlite3, "connect", tracking_connect) + return opened, closed + + +def test_ledger_operations_close_every_connection(monkeypatch, tmp_path): + """Public durable-ledger reads/writes must close every connection opened.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + ad.get_durable_delegation("nope") + ad.recover_abandoned_delegations() + ad.restore_undelivered_completions(queue.Queue()) + ad.mark_completion_delivered("nope") + ad.claim_completion_delivery("nope", "claim-1") + + assert opened, "expected at least one connection to be opened" + assert len(opened) == len(closed) + assert set(opened) == set(closed) + + +def test_early_return_still_closes_connection(monkeypatch, tmp_path): + """A no-op update (no matching row) must still open and close exactly once.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + assert ad.mark_completion_delivered("does-not-exist") is False + + assert len(opened) == 1 + assert len(closed) == 1 + + +def test_exception_during_operation_still_closes_connection(monkeypatch, tmp_path): + """A failing statement inside the transaction must roll back and close.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = _track_connections(monkeypatch) + + with pytest.raises(sqlite3.IntegrityError): + with ad._transaction() as conn: + # Missing NOT NULL columns -> constraint failure inside the block. + conn.execute( + "INSERT INTO async_delegations (delegation_id) VALUES ('x')" + ) + + assert len(opened) == 1 + assert len(closed) == 1 + + +def test_schema_init_failure_still_closes_connection(monkeypatch, tmp_path): + """A PRAGMA/DDL failure after connect() must still close the connection.""" + _point_ledger(monkeypatch, tmp_path) + opened, closed = [], [] + real_connect = sqlite3.connect + + class _FailingSchemaConnection(_TrackingConnection): + def execute(self, sql, *args, **kwargs): + if "CREATE TABLE" in sql: + raise sqlite3.OperationalError("simulated schema init failure") + return self._real.execute(sql, *args, **kwargs) + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(id(conn)) + return _FailingSchemaConnection(conn, closed) + + monkeypatch.setattr(ad.sqlite3, "connect", tracking_connect) + + with pytest.raises(sqlite3.OperationalError): + with ad._transaction(): + pass + + assert len(opened) == 1 + assert len(closed) == 1 diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 5181149c4d3ac..f670678b351ad 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -43,7 +43,8 @@ import time import uuid from concurrent.futures import ThreadPoolExecutor -from typing import Any, Callable, Dict, List, Optional +from contextlib import contextmanager +from typing import Any, Callable, Dict, Iterator, List, Optional from hermes_constants import get_hermes_home from tools.daemon_pool import DaemonThreadPoolExecutor @@ -93,6 +94,17 @@ def _connect() -> sqlite3.Connection: path = _db_path() path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path, timeout=10) + try: + _initialize_schema(conn) + except Exception: + # A PRAGMA/DDL failure after a successful connect() must not leak the + # just-opened connection back to the caller. + conn.close() + raise + return conn + + +def _initialize_schema(conn: sqlite3.Connection) -> None: conn.execute("PRAGMA journal_mode=WAL") conn.execute( """CREATE TABLE IF NOT EXISTS async_delegations ( @@ -126,7 +138,25 @@ def _connect() -> sqlite3.Connection: ): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") - return conn + + +@contextmanager +def _transaction() -> Iterator[sqlite3.Connection]: + """Open a connection, commit/rollback on exit, and ALWAYS close it. + + ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the + transaction; they do not close the connection. Using ``with _connect()`` + alone therefore leaks a connection — and its WAL/SHM file descriptors — on + every durable dispatch, completion, and delivery-claim, deferring the close + to the garbage collector. On a long-running gateway that exhausts + ``RLIMIT_NOFILE`` (the cron-ledger sibling of this bug was #69567 / PR #69594). + """ + conn = _connect() + try: + with conn: + yield conn + finally: + conn.close() def _persist_dispatch(record: Dict[str, Any]) -> None: @@ -141,7 +171,7 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch") if key in record } - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( """INSERT OR REPLACE INTO async_delegations (delegation_id, origin_session, origin_ui_session_id, @@ -158,7 +188,7 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: def _delete_durable_delegation(delegation_id: str) -> None: - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute("DELETE FROM async_delegations WHERE delegation_id=?", (delegation_id,)) @@ -166,7 +196,7 @@ def _prune_durable_records() -> None: """Bound terminal history, preferring delivered records for deletion.""" now = time.time() cutoff = now - _DURABLE_RETENTION_SECONDS - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( "DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?", (cutoff,), @@ -203,7 +233,7 @@ def _prune_durable_records() -> None: def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None: now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( """UPDATE async_delegations SET state=?, completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' @@ -214,7 +244,7 @@ def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None: def _note_delivery_attempt(delegation_id: str) -> None: - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: conn.execute( "UPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?", (time.time(), delegation_id), @@ -229,7 +259,7 @@ def recover_abandoned_delegations() -> int: return 0 now = time.time() recovered = 0 - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: rows = conn.execute( """SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, @@ -281,7 +311,7 @@ def restore_undelivered_completions(target_queue) -> int: results seconds after boot (#64484). """ recover_abandoned_delegations() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: rows = conn.execute( """SELECT delegation_id, event_json FROM async_delegations WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL @@ -298,7 +328,7 @@ def restore_undelivered_completions(target_queue) -> int: def mark_completion_delivered(delegation_id: str) -> bool: """Atomically acknowledge successful injection of a durable completion.""" now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: cur = conn.execute( """UPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=? WHERE delegation_id=? AND delivery_state!='delivered'""", @@ -310,7 +340,7 @@ def mark_completion_delivered(delegation_id: str) -> bool: def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool: """Claim one pending completion across competing consumers/processes.""" now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: row = conn.execute( "SELECT delivery_state FROM async_delegations WHERE delegation_id=?", (delegation_id,), @@ -349,7 +379,7 @@ def release_completion_delivery(delegation_id: str, claim_id: str) -> bool: pending rows). """ now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: capped = conn.execute( """UPDATE async_delegations SET delivery_state='dropped', delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? @@ -384,7 +414,7 @@ def drop_completion_delivery(delegation_id: str, claim_id: str) -> bool: completion that will be fail-closed dropped again every time. """ now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: cur = conn.execute( """UPDATE async_delegations SET delivery_state='dropped', updated_at=?, delivery_claim=NULL, @@ -399,7 +429,7 @@ def drop_completion_delivery(delegation_id: str, claim_id: str) -> bool: def complete_completion_delivery(delegation_id: str, claim_id: str) -> bool: """Acknowledge acceptance for the consumer holding this claim.""" now = time.time() - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: cur = conn.execute( """UPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=?, delivery_claim=NULL, @@ -422,7 +452,7 @@ def release_event_delivery(evt: Dict[str, Any], claim_id: str) -> None: def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: - with _DB_LOCK, _connect() as conn: + with _DB_LOCK, _transaction() as conn: row = conn.execute( """SELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts