From cad9e90732a49b3fc2a5cacb3611d435fb924a91 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 09:46:30 -0700 Subject: [PATCH] review-fix(seams): late-bind hermes_state._connect_repair_durable and tools.approval._command_detection_variants Follow-up to b8f99bfc439: the seam edits for hermes_state_repair.py and tools/approval_detection.py were overwritten by a concurrent squad's write before that commit landed (only their docstring restores got in). Re-apply: _repair_conn/_open_exclusive/_db_opens_cleanly look up _connect_repair_durable via hermes_state at call time; detect_dangerous_command/ detect_hardline_command look up _command_detection_variants via tools.approval, so patching the facade (as tests/state/test_state_db_wal_unlink_race.py and tests/hermes_cli/test_approvals_test.py do) reaches the call again, as on BASE 63279301bcb. --- hermes_state_repair.py | 9 ++++++--- tools/approval_detection.py | 8 ++++++-- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/hermes_state_repair.py b/hermes_state_repair.py index 9a9957de67..f84a1db981 100644 --- a/hermes_state_repair.py +++ b/hermes_state_repair.py @@ -584,7 +584,8 @@ def _connect_repair_durable(db_path: Path, *, timeout: float = 5.0) -> sqlite3.C def _repair_conn(db_path: Path, *, timeout: float = 5.0): """A :func:`_connect_repair_durable` connection as a context manager, closed on exit.""" - return contextlib.closing(_connect_repair_durable(db_path, timeout=timeout)) + from hermes_state import _connect_repair_durable as _connect # call-time lookup: tests patch hermes_state. + return contextlib.closing(_connect(db_path, timeout=timeout)) def _reapply_durability_barriers(conn: sqlite3.Connection) -> bool: @@ -623,7 +624,8 @@ def _close_unpinned(conn: sqlite3.Connection) -> None: def _open_exclusive(db_path: Path, begin: str) -> sqlite3.Connection: """Zero-timeout connection holding ``locking_mode=EXCLUSIVE`` after a rolled-back *begin*; closed (unpinned) and re-raised when exclusion cannot be taken.""" - conn = _connect_repair_durable(db_path, timeout=0.0) + from hermes_state import _connect_repair_durable as _connect # call-time lookup: tests patch hermes_state. + conn = _connect(db_path, timeout=0.0) try: for statement in ("PRAGMA locking_mode=EXCLUSIVE", begin, "ROLLBACK"): conn.execute(statement) @@ -700,7 +702,8 @@ def _db_opens_cleanly(db_path: Path) -> Optional[str]: # of entries in index" when a B-tree index (e.g. idx_sessions_handoff_state) falls out of sync with its # base table. REINDEX rewrites the index b-tree from the canonical table rows using the existing index # definition, fixing the mismatch without touching data or FTS schema. - conn = _connect_repair_durable(db_path) + from hermes_state import _connect_repair_durable as _connect # call-time lookup: tests patch hermes_state. + conn = _connect(db_path) try: with contextlib.closing(conn): # Best-effort tokenizer load: messages_fts_cjk needs cjk_unicode61 before any statement can touch it; diff --git a/tools/approval_detection.py b/tools/approval_detection.py index 3ff8d918bb..2649da6ee8 100644 --- a/tools/approval_detection.py +++ b/tools/approval_detection.py @@ -178,7 +178,9 @@ def detect_hardline_command(command: str) -> tuple: _, malformed_grep = _grep_safe_detection_variant(normalized) if malformed_grep: return (True, _MALFORMED_EXEC_DESCRIPTION) - for command_variant in _command_detection_variants(command): + # Call-time lookup through the facade: tests/plugins patch tools.approval._command_detection_variants. + from tools.approval import _command_detection_variants as _variants + for command_variant in _variants(command): variant_lower = command_variant.lower() masked_lower: str | None = None for pattern_re, description, quote_masked in HARDLINE_PATTERNS_COMPILED: @@ -1153,7 +1155,9 @@ def detect_dangerous_command(command: str) -> tuple: return (True, _PARSER_LIMIT_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION) if _is_verification_artifact_cleanup(command): return (False, None, None) - for command_variant in _command_detection_variants(command): + # Call-time lookup through the facade: tests/plugins patch tools.approval._command_detection_variants. + from tools.approval import _command_detection_variants as _variants + for command_variant in _variants(command): command_lower = command_variant.lower() for pattern_re, description in DANGEROUS_PATTERNS_COMPILED: if pattern_re.search(command_lower):