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

@@ -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
 

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
 

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
 

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
 

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

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)

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,