#683 hooks: rollover _find_repo_root fails loud + edit_gate soft entry-count guard (#664 follow-up).

Two hardening items surfaced during #664 that @memory could not touch (cross-branch edit gate blocked it).

ITEM 1 — _find_repo_root fail-loud (lifecycle/rollover.py). The PreCompact rollover hook's _find_repo_root() returned None SILENTLY when AIPASS_HOME/cwd was wrong -> rollover no-ops invisibly (the exact silent-skip that hid #664 for months). Now logs a logger.error with the AIPASS_HOME value + cwd before returning None (still degrades, just visibly).

ITEM 2 — edit_gate soft entry-count guard (security/edit_gate.py). edit_gate enforced per-entry CHARACTER caps but not entry COUNTS, so a branch could drift past its count cap between rollovers. New _check_section_counts warns (NEVER blocks) when a rolling section exceeds its cap, reading the SAME memory.config.json rollover caps @memory uses (config_loader.section('rollover') -> per_branch/defaults -> count); wrapped so a config-import failure degrades silently.

Built by @hooks, verified by devpulse: 70 tests green (+14 incl never-blocks guarantee, boundary cases, per-branch override, import-failure resilience); LIVE repro proves item1 logs the error on a bad root and item2 warns over-cap (20/15) without blocking; config structure confirmed to match memory's real caps (not inert).

Rides PR#659 (issue-clearing, no main-merge). Source: #664 verify (S292).
This commit is contained in:
AIOSAI
2026-07-10 10:27:03 -07:00
parent a3a476f59d
commit e302df6ec0
6 changed files with 249 additions and 1 deletions
+11
View File
@@ -47,6 +47,17 @@ PyPI version — not the changelog header.
relays. Stall logic extracted into a `StallTracker` for clarity; +9 tests
(142 green), devpulse audit 100%. (devpulse)
- **Rollover `_find_repo_root` now fails loud, and `edit_gate` warns on over-count
memory sections (issue #683, #664 follow-up).** The PreCompact rollover hook's
`_find_repo_root` returned `None` silently when `AIPASS_HOME`/cwd was wrong — the
exact silent-skip that hid #664 for months; it now logs a `logger.error` with the
`AIPASS_HOME` value and cwd before returning. And `edit_gate` enforced per-entry
*character* caps but not entry *counts*, so a branch could drift past its count
cap between rollovers; a soft `_check_section_counts` now warns (never blocks),
reading the same `memory.config.json` rollover caps @memory uses. +14 tests
(70 green); both live-proven (bad root → error logged; over-cap → warn, no block).
(@hooks, verified devpulse)
- **`is_owner()` now case-folds — `is_owner('DEVPULSE')` matches `is_owner('devpulse')`
(issue #679).** The spawn-registry resolver (`registry.py:382`) `@`-normalized the
email but never lowercased, so a mixed-case branch name (registry names are
+25
View File
@@ -784,6 +784,31 @@
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_edit_gate_trinity.py",
"standard": "architecture",
"reason": "Test files live in tests/, not in the 3-layer apps structure."
},
{
"file": "tests/test_edit_gate_trinity.py",
"standard": "documentation",
"reason": "Test methods use descriptive names as documentation per pytest convention."
},
{
"file": "tests/test_edit_gate_trinity.py",
"standard": "encapsulation",
"reason": "Tests import handlers directly to test implementation details."
},
{
"file": "tests/test_edit_gate_trinity.py",
"standard": "hardcoded_path",
"reason": "Test fixture strings contain paths as test input data, not real filesystem ops."
},
{
"file": "tests/test_edit_gate_trinity.py",
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_git_gate.py",
"standard": "architecture",
@@ -27,6 +27,11 @@ def _find_repo_root() -> Path | None:
for parent in [cwd, *list(cwd.parents)]:
if (parent / "AIPASS_REGISTRY.json").exists():
return parent
logger.error(
"[HOOKS] rollover: _find_repo_root failed — no AIPASS_REGISTRY.json found. AIPASS_HOME=%r, cwd=%s",
aipass_home,
cwd,
)
return None
@@ -117,6 +117,35 @@ def _todos_count_advisory(after: dict, branch: str) -> str:
return ""
def _check_section_counts(after: dict, branch: str, file_stem: str) -> None:
"""Warn (never block) when rolling sections exceed their configured entry-count cap."""
try:
cl = importlib.import_module("aipass.memory.apps.handlers.json.config_loader")
roll = cl.section("rollover")
branch_cfg = roll.get("per_branch", {}).get(branch) or roll.get("defaults", {})
file_cfg = branch_cfg.get(file_stem, {})
for section_name, section_cfg in file_cfg.items():
if not isinstance(section_cfg, dict):
continue
cap = section_cfg.get("count")
if cap is None:
continue
entries = after.get(section_name)
if not isinstance(entries, list):
continue
count = len(entries)
if count > cap:
logger.warning(
"[HOOKS] edit_gate: %s.%s count over limit (%d/%d) — rollover will trim at next PreCompact",
file_stem,
section_name,
count,
cap,
)
except Exception as exc:
logger.warning("[HOOKS] edit_gate: section count check failed (skipping): %s", exc)
def _check_trinity_change(fp: Path, tool_name: str, tool_input: dict, branch: str) -> dict | None:
"""Check .trinity Write/Edit/MultiEdit for over-limit entries. Returns block dict or None."""
try:
@@ -147,6 +176,8 @@ def _check_trinity_change(fp: Path, tool_name: str, tool_input: dict, branch: st
if block:
return block
_check_section_counts(after, branch, fp.stem)
if fp.name == "local.json":
advisory = _todos_count_advisory(after, branch)
if advisory:
@@ -94,6 +94,9 @@ _ROLLOVER_CONFIG_10 = {
"key_learnings": {"count": 25},
"todos": {"count": 10},
},
"observations": {
"observations": {"count": 15},
},
},
"per_branch": {},
},
@@ -109,7 +112,9 @@ def _mock_importlib_modules(limits, rollover_cfg=None):
entry_limits_mock.changed_entries = el_real.changed_entries
config_loader_mock = MagicMock()
config_loader_mock.load.return_value = rollover_cfg if rollover_cfg is not None else _ROLLOVER_CONFIG_10
cfg = rollover_cfg if rollover_cfg is not None else _ROLLOVER_CONFIG_10
config_loader_mock.load.return_value = cfg
config_loader_mock.section.side_effect = lambda name: cfg.get(name, {})
def side_effect(name):
if "entry_limits" in name:
@@ -1170,3 +1175,137 @@ class TestTrinityTodosCountAdvisory:
assert result["exit_code"] == 0
assert result["stdout"] == ""
class TestSectionCountGuard:
"""Soft count guard: warn (never block) when rolling sections exceed count cap."""
def test_sessions_over_count_warns(self, tmp_path, caplog):
"""21 sessions vs 20 cap -> warning logged, write still allowed."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(21)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "local.sessions count over limit (21/20)" in caplog.text
def test_key_learnings_over_count_warns(self, tmp_path, caplog):
"""26 key_learnings vs 25 cap -> warning logged."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"key_learnings": [{"value": "v"} for _ in range(26)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "local.key_learnings count over limit (26/25)" in caplog.text
def test_observations_over_count_warns(self, tmp_path, caplog):
"""16 observations vs 15 cap -> warning logged."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "observations.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"observations": [{"note": "n"} for _ in range(16)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "observations.observations count over limit (16/15)" in caplog.text
def test_under_count_no_warning(self, tmp_path, caplog):
"""10 sessions vs 20 cap -> no warning."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(10)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "count over limit" not in caplog.text
def test_at_count_no_warning(self, tmp_path, caplog):
"""Exactly 20 sessions vs 20 cap -> no warning (only > triggers)."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(20)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "count over limit" not in caplog.text
def test_count_guard_never_blocks(self, tmp_path):
"""Even with enforce=True char limits, count guard only warns — exit_code always 0."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(30)]})
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
def test_per_branch_count_override(self, tmp_path, caplog):
"""per_branch overrides default count -> 6 sessions vs 5 cap warns."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(6)]})
rollover_cfg = {
"rollover": {
"defaults": {"local": {"sessions": {"count": 20}}},
"per_branch": {"hooks": {"local": {"sessions": {"count": 5}}}},
},
}
with patch("importlib.import_module", side_effect=_mock_importlib_modules(_TEST_LIMITS_WARN, rollover_cfg)):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "local.sessions count over limit (6/5)" in caplog.text
def test_config_loader_import_failure_silent(self, tmp_path, caplog):
"""config_loader import always fails -> no count warning, no crash, write allowed."""
from aipass.hooks.apps.handlers.security.edit_gate import handle
file_path = _make_trinity_path(tmp_path, "hooks", "local.json")
cwd = str(tmp_path / "src" / "aipass" / "hooks")
content = json.dumps({"sessions": [{"summary": "s"} for _ in range(30)]})
el_real = importlib.import_module("aipass.memory.apps.handlers.json.entry_limits")
entry_limits_mock = MagicMock()
entry_limits_mock.load_entry_limits.return_value = _TEST_LIMITS_WARN
entry_limits_mock.changed_entries = el_real.changed_entries
def _side_effect(name):
if "entry_limits" in name:
return entry_limits_mock
if "config_loader" in name:
raise ImportError("no config_loader")
return importlib.import_module(name)
with patch("importlib.import_module", side_effect=_side_effect):
result = handle(_hook_data(file_path, content, cwd=cwd))
assert result["exit_code"] == 0
assert "count over limit" not in caplog.text
+37
View File
@@ -130,3 +130,40 @@ class TestRolloverHandler:
assert not success
assert "timed out" in msg
class TestFindRepoRootFailLoud:
"""_find_repo_root logs error (not silent skip) when no AIPASS_REGISTRY.json found."""
def test_bad_root_logs_error(self, tmp_path, caplog):
"""No AIPASS_REGISTRY.json anywhere -> logger.error with AIPASS_HOME + cwd."""
import logging
from aipass.hooks.apps.handlers.lifecycle.rollover import _find_repo_root
with caplog.at_level(logging.ERROR):
with patch.dict("os.environ", {"AIPASS_HOME": ""}):
with patch(f"{MOD}.Path") as mock_path_cls:
mock_path_cls.cwd.return_value = tmp_path
result = _find_repo_root()
assert result is None
assert "_find_repo_root failed" in caplog.text
assert "AIPASS_REGISTRY.json" in caplog.text
def test_bad_aipass_home_falls_through_to_cwd(self, tmp_path, caplog):
"""AIPASS_HOME set but no registry there -> falls through, still logs error if cwd also fails."""
import logging
from aipass.hooks.apps.handlers.lifecycle.rollover import _find_repo_root
bad_home = str(tmp_path / "nonexistent")
with caplog.at_level(logging.ERROR):
with patch.dict("os.environ", {"AIPASS_HOME": bad_home}):
with patch(f"{MOD}.Path") as mock_path_cls:
mock_path_cls.return_value = tmp_path / "nonexistent"
mock_path_cls.cwd.return_value = tmp_path
result = _find_repo_root()
assert result is None
assert "_find_repo_root failed" in caplog.text
assert bad_home in caplog.text