diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index 1ddb6e6cad..a69c2db0e8 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -22,7 +22,6 @@ from typing import Any, Callable, Iterator, Optional from agent.redact import redact_sensitive_text from cron.executions import _owner_is_live, _process_start_time -from hermes_cli.sqlite_util import add_column_if_missing, open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -78,6 +77,8 @@ def _path() -> Path: def _initialize_schema(conn: sqlite3.Connection) -> None: + from hermes_cli.sqlite_util import add_column_if_missing + conn.execute( """CREATE TABLE IF NOT EXISTS deliveries ( execution_id TEXT PRIMARY KEY, @@ -109,6 +110,11 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: def _connect() -> sqlite3.Connection: + # Late imports: a scheduler daemon that outlives an on-disk upgrade already has the OLD + # ``hermes_cli.sqlite_util`` / ``cron.jobs`` cached, so new names must be resolved at call time, + # not at import time (the guarantee cron/ledger.py used to carry, see e24c8499). + from hermes_cli.sqlite_util import open_db + path = _path() conn = open_db(path, db_label="cron/deliveries.db", synchronous_full=True, initialize=_initialize_schema) try: @@ -123,6 +129,8 @@ def _transaction() -> Iterator[sqlite3.Connection]: # Pruning is done explicitly by the paths that create terminal # rows (_finish / recover_abandoned / _terminalize_wait_timeout); # read-only polls must not pay for a full-table UPDATE + COUNT. + from hermes_cli.sqlite_util import transaction + with _lock, transaction(_connect()) as conn: yield conn diff --git a/cron/executions.py b/cron/executions.py index 292b8b22df..35dd8b5b60 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -16,8 +16,6 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, Iterator, List, Optional -from cron.jobs import _ensure_cron_dir -from hermes_cli.sqlite_util import add_column_if_missing, open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -35,12 +33,20 @@ _PROCESS_ID = uuid.uuid4().hex # --- executions ledger -------------------------------------------------------------------------- def _connect() -> sqlite3.Connection: + # Late imports: a scheduler daemon that outlives an on-disk upgrade already has the OLD + # ``hermes_cli.sqlite_util`` / ``cron.jobs`` cached, so new names must be resolved at call time, + # not at import time (the guarantee cron/ledger.py used to carry, see e24c8499). + from cron.jobs import _ensure_cron_dir + from hermes_cli.sqlite_util import open_db + path = EXECUTIONS_FILE or (get_hermes_home().resolve() / "cron" / "executions.db") _ensure_cron_dir(path.parent) return open_db(path, db_label="cron/executions.db", synchronous_full=True, initialize=_initialize_schema) def _initialize_schema(conn: sqlite3.Connection) -> None: + from hermes_cli.sqlite_util import add_column_if_missing + conn.execute( """CREATE TABLE IF NOT EXISTS executions ( id TEXT PRIMARY KEY, @@ -84,6 +90,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: + from hermes_cli.sqlite_util import transaction + with _lock, transaction(_connect()) as conn: yield conn diff --git a/cron/incidents.py b/cron/incidents.py index 44f9e0c1dc..fee52bad55 100644 --- a/cron/incidents.py +++ b/cron/incidents.py @@ -19,8 +19,6 @@ from pathlib import Path from typing import Any, Dict, Iterator, List, Optional from cron import executions as _executions -from cron.jobs import _ensure_cron_dir -from hermes_cli.sqlite_util import open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -54,6 +52,12 @@ def _db_path() -> Path: def _connect() -> sqlite3.Connection: + # Late imports: a scheduler daemon that outlives an on-disk upgrade already has the OLD + # ``hermes_cli.sqlite_util`` / ``cron.jobs`` cached, so new names must be resolved at call time, + # not at import time (the guarantee cron/ledger.py used to carry, see e24c8499). + from cron.jobs import _ensure_cron_dir + from hermes_cli.sqlite_util import open_db + path = _db_path() _ensure_cron_dir(path.parent) return open_db(path, db_label="cron/executions.db", synchronous_full=True, initialize=_initialize_schema) @@ -87,6 +91,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: + from hermes_cli.sqlite_util import transaction + with _lock, transaction(_connect()) as conn: yield conn diff --git a/cron/notepad.py b/cron/notepad.py index d7b2092b21..12fc7954c8 100644 --- a/cron/notepad.py +++ b/cron/notepad.py @@ -15,8 +15,6 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, Iterator, List, Optional -from cron.jobs import _ensure_cron_dir -from hermes_cli.sqlite_util import open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -36,6 +34,12 @@ def _current_notepad_file() -> Path: def _connect() -> sqlite3.Connection: + # Late imports: a scheduler daemon that outlives an on-disk upgrade already has the OLD + # ``hermes_cli.sqlite_util`` / ``cron.jobs`` cached, so new names must be resolved at call time, + # not at import time (the guarantee cron/ledger.py used to carry, see e24c8499). + from cron.jobs import _ensure_cron_dir + from hermes_cli.sqlite_util import open_db + path = _current_notepad_file() _ensure_cron_dir(path.parent) return open_db(path, db_label="cron/notepad.db", initialize=_initialize_schema) @@ -55,6 +59,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: + from hermes_cli.sqlite_util import transaction + with _lock, transaction(_connect()) as conn: yield conn diff --git a/gateway/hosted_room_policy_checkpoint.py b/gateway/hosted_room_policy_checkpoint.py index ae3efbe512..8de9187d30 100644 --- a/gateway/hosted_room_policy_checkpoint.py +++ b/gateway/hosted_room_policy_checkpoint.py @@ -15,7 +15,6 @@ from typing import Any, Callable, Mapping from gateway import hosted_rooms from gateway.hosted_rooms_common import DbPath, compact_json, fenced_update -from hermes_cli.sqlite_util import open_db, transaction MAX_ACTIVE_POLICY_EVENTS = 64 @@ -111,9 +110,14 @@ class HostedRoomPolicyCheckpoint: conn.execute(ddl) def _connect(self) -> sqlite3.Connection: + # Late import: a gateway that outlives an on-disk upgrade has the OLD sqlite_util cached. + from hermes_cli.sqlite_util import open_db + return open_db(self.db_path, db_label="shared-state.db (room policy checkpoint)", busy_timeout_ms=10_000) def _transaction(self): + from hermes_cli.sqlite_util import transaction + return transaction(self._connect()) @staticmethod diff --git a/gateway/hosted_rooms_common.py b/gateway/hosted_rooms_common.py index 2ac7feb627..5bd58e5c3f 100644 --- a/gateway/hosted_rooms_common.py +++ b/gateway/hosted_rooms_common.py @@ -15,8 +15,6 @@ import time from pathlib import Path from typing import Any, Callable, Iterator, Mapping -from hermes_cli.sqlite_util import open_db -from hermes_cli.sqlite_util import transaction as _transaction IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") DbPath = Path | str @@ -98,6 +96,8 @@ def clock(now: float | None) -> float: def open_sqlite(path: DbPath, *, timeout: float = 10) -> sqlite3.Connection: """Row-factory connection with foreign keys on; no journal or schema work (steady-state readers).""" + from hermes_cli.sqlite_util import open_db + return open_db(path, db_label="shared-state.db", busy_timeout_ms=int(timeout * 1000), wal=False, foreign_keys=True) @@ -120,6 +120,9 @@ def connect( except Exception: conn.rollback() raise + # Late import: a gateway that outlives an on-disk upgrade has the OLD sqlite_util cached. + from hermes_cli.sqlite_util import open_db + return open_db(db_path, db_label=db_label, busy_timeout_ms=10_000, foreign_keys=True, wal_lock_retries=lock_retries, initialize=_initialize) @@ -143,4 +146,6 @@ def transaction( connect: Callable[[DbPath], sqlite3.Connection], db_path: DbPath, *, immediate: bool ) -> Iterator[sqlite3.Connection]: """Open via ``connect``, optionally ``BEGIN IMMEDIATE``, commit on success, always close.""" + from hermes_cli.sqlite_util import transaction as _transaction + return _transaction(connect(db_path), immediate=immediate) diff --git a/tests/cron/test_upgrade_module_skew.py b/tests/cron/test_upgrade_module_skew.py index 64b144603c..439f92366e 100644 --- a/tests/cron/test_upgrade_module_skew.py +++ b/tests/cron/test_upgrade_module_skew.py @@ -1,4 +1,10 @@ -"""Cron imports remain usable when a daemon spans an on-disk upgrade.""" +"""Cron imports remain usable when a daemon spans an on-disk upgrade. + +A long-running scheduler already has ``hermes_cli.sqlite_util`` and ``cron.jobs`` cached from +BEFORE the upgrade; the first lazy import of a cron store afterwards must not need names those +stale modules lack (``scheduler_prompt._build_job_prompt`` imports ``cron.notepad`` unguarded, so +an ``ImportError`` there fails every job tick until restart). +""" from __future__ import annotations @@ -6,24 +12,27 @@ import subprocess import sys from pathlib import Path +import pytest -def test_lazy_cron_stores_do_not_require_new_symbols_on_cached_executions_module(): - repo_root = Path(__file__).resolve().parents[2] - script = """ -import sys -import cron.executions as executions +_SKEW_SCRIPT = """ +import sys, types +import hermes_cli.sqlite_util as sqlite_util +import cron.jobs as jobs -for name in ("open_db", "transaction", "add_column_if_missing", "_ensure_cron_dir"): - delattr(executions, name) -sys.modules.pop("cron.incidents", None) -sys.modules.pop("cron.notepad", None) +# The pre-upgrade sqlite_util only had add_column_if_missing / write_txn. +for name in ("open_db", "transaction"): + delattr(sqlite_util, name) +sys.modules.pop("cron.{store}", None) -import cron.incidents -import cron.notepad +import cron.{store} """ + +@pytest.mark.parametrize("store", ["notepad", "incidents", "executions", "delivery_queue"]) +def test_lazy_cron_stores_import_against_pre_upgrade_sqlite_util(store): + repo_root = Path(__file__).resolve().parents[2] result = subprocess.run( - [sys.executable, "-c", script], + [sys.executable, "-c", _SKEW_SCRIPT.format(store=store)], cwd=repo_root, capture_output=True, text=True,