review-fix(seams): late-bind hermes_state._connect_repair_durable and tools.approval._command_detection_variants

Follow-up to b8f99bfc43: 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 63279301bc.
This commit is contained in:
Teknium
2026-09-03 09:46:30 -07:00
parent 3b963ce253
commit cad9e90732
2 changed files with 12 additions and 5 deletions
+6 -3
View File
@@ -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.<name>
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.<name>
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.<name>
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;
+6 -2
View File
@@ -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):