From 579dbe0c713b167b7c67e751d4e2d883d6260a45 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sat, 12 Sep 2026 23:43:39 -0700 Subject: [PATCH] fix(cron): sqlite_util imports are late so a running scheduler survives an on-disk upgrade cron/ledger.py (e24c8499) existed so a long-running scheduler that lazily imports notepad/incidents AFTER `hermes update` never needs new names from a module it already has cached. The dedup deleted it and imported open_db/transaction from hermes_cli.sqlite_util at module level; a pre-upgrade daemon has the OLD sqlite_util cached (executions imported add_column_if_missing from it), so the first job tick after an upgrade would ImportError in scheduler_prompt._build_job_prompt until restart. - cron/{notepad,incidents,executions,delivery_queue}: import open_db/transaction/ add_column_if_missing and cron.jobs._ensure_cron_dir inside _connect/_transaction/ _initialize_schema. This also stops the 3.8k-line cron.jobs being pulled eagerly by importing a store (it was lazy in cron/ledger.open_ledger). - gateway/hosted_rooms_common, hosted_room_policy_checkpoint: same treatment; the gateway imports hosted_rooms lazily from request handlers, so it has the same skew exposure. - tests/cron/test_upgrade_module_skew.py: simulate the real skew (delete open_db/transaction from the cached sqlite_util, then import each store). The previous repoint deleted names from cron.executions, which notepad/incidents do not import from, so it passed regardless. Sabotage: a module-level `from hermes_cli.sqlite_util import open_db` in notepad fails it with "cannot import name 'open_db'". --- cron/delivery_queue.py | 10 ++++++- cron/executions.py | 12 ++++++-- cron/incidents.py | 10 +++++-- cron/notepad.py | 10 +++++-- gateway/hosted_room_policy_checkpoint.py | 6 +++- gateway/hosted_rooms_common.py | 9 ++++-- tests/cron/test_upgrade_module_skew.py | 35 +++++++++++++++--------- 7 files changed, 69 insertions(+), 23 deletions(-) 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,