#664 memory: external-project branches now auto-roll (rollover discovery no longer cwd-scoped). Branch discovery only saw registries reachable by walking up from the caller cwd, so branches living solely in an external project's *_REGISTRY.json were never reached by rollovers fired from the AIPass tree (PreCompact hook runs cwd=repo root) — their .trinity grew unbounded (one hit 110 key_learnings vs a 15 cap) and vectors went stale. @memory added a persisted known_registries.json (gitignored per-install data) recording every external registry seen via the cwd walk, so discovery reaches them regardless of caller cwd; stale/deleted paths filtered on load. Plus a soft entry-COUNT guard at write time (warns, never blocks) since gates only enforced char caps. Remaining hooks-side harden (_find_repo_root fail-loud + edit_gate count-guard) filed for @hooks. Built by @memory, verified by devpulse: 70 changed-file tests green (+12), memory_json gitignored (data local, code ships), LIVE REPRO proves a rollover fired from AIPass root now reaches an external-registry branch (was invisible before).
This commit is contained in:
@@ -77,6 +77,22 @@ PyPI version — not the changelog header.
|
||||
planted cruft file. (built by @aipass, verified by devpulse — incl. the
|
||||
gitignore ship-gap)
|
||||
|
||||
- **External-project branches now auto-roll — rollover discovery is no longer
|
||||
cwd-scoped (issue #664).** Branch discovery only saw registries reachable by
|
||||
walking up from the caller's cwd, so branches living solely in an external
|
||||
project's `*_REGISTRY.json` were never reached by rollovers fired from the
|
||||
AIPass tree (the PreCompact hook runs with cwd = repo root) — their `.trinity`
|
||||
files grew unbounded (one hit 110 key_learnings against a 15 cap) and vector
|
||||
stores went stale. `@memory` added a persisted `known_registries.json`
|
||||
(gitignored per-install data) that records every external registry seen via the
|
||||
cwd walk, so discovery reaches them regardless of caller cwd; stale/deleted
|
||||
registry paths are filtered on load. Plus a soft entry-**count** guard at write
|
||||
time (warns, never blocks) since the write gates only enforced char caps. The
|
||||
remaining hooks-side harden (`_find_repo_root` fail-loud + the `edit_gate`
|
||||
count-guard) is filed for `@hooks`. +12 tests; live repro confirms a rollover
|
||||
fired from the AIPass root now reaches an external-registry branch.
|
||||
(built by @memory, verified by devpulse)
|
||||
|
||||
---
|
||||
|
||||
## [2026-07-09]
|
||||
|
||||
@@ -34,6 +34,7 @@ from typing import Dict, Any, Optional
|
||||
|
||||
from aipass.prax.apps.modules.logger import get_system_logger
|
||||
from aipass.memory.apps.handlers.json import json_handler
|
||||
from aipass.memory.apps.handlers.json import config_loader
|
||||
from aipass.memory.apps.handlers.json.entry_limits import load_entry_limits, changed_entries
|
||||
|
||||
logger = get_system_logger()
|
||||
@@ -134,6 +135,47 @@ def _validate_entry_limits(
|
||||
return {"success": False, "error": f"Entry limit exceeded: {details}"}
|
||||
|
||||
|
||||
def _check_entry_counts(file_path: Path, data: Dict[str, Any]) -> None:
|
||||
"""Soft guard: warn when list entry counts exceed rollover limits.
|
||||
|
||||
Does NOT block writes — only logs warnings so over-limit growth
|
||||
between rollovers is visible in prax logs.
|
||||
"""
|
||||
if file_path.parent.name != ".trinity":
|
||||
return
|
||||
if file_path.name not in _TRACKED_TRINITY_FILES:
|
||||
return
|
||||
|
||||
branch = file_path.parent.parent.name.lower()
|
||||
file_type = file_path.stem
|
||||
|
||||
cfg = config_loader.section("rollover")
|
||||
per_branch = cfg.get("per_branch", {})
|
||||
defaults = cfg.get("defaults", {})
|
||||
|
||||
file_limits = per_branch.get(branch, {}).get(file_type, {})
|
||||
if not file_limits:
|
||||
file_limits = defaults.get(file_type, {})
|
||||
if not file_limits:
|
||||
return
|
||||
|
||||
for section_name, section_cfg in file_limits.items():
|
||||
if section_name.startswith("_"):
|
||||
continue
|
||||
if not isinstance(section_cfg, dict):
|
||||
continue
|
||||
max_count = section_cfg.get("count")
|
||||
if max_count is None:
|
||||
continue
|
||||
entries = data.get(section_name, [])
|
||||
if isinstance(entries, list) and len(entries) > max_count:
|
||||
logger.warning(
|
||||
f"[memory_files] ENTRY COUNT: {branch} {file_path.name} "
|
||||
f"{section_name} has {len(entries)}/{max_count} entries "
|
||||
f"(+{len(entries) - max_count} over rollover limit)"
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# CORE READ/WRITE OPERATIONS
|
||||
# =============================================================================
|
||||
@@ -218,6 +260,12 @@ def write_memory_file(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
except Exception as exc:
|
||||
logger.warning(f"[memory_files] Entry-limits validation error (writing anyway): {exc}")
|
||||
|
||||
# --- Soft entry-count guard (warn-only, never blocks) --------------------
|
||||
try:
|
||||
_check_entry_counts(file_path, data)
|
||||
except Exception as exc:
|
||||
logger.warning(f"[memory_files] Entry-count check error (writing anyway): {exc}")
|
||||
|
||||
try:
|
||||
# Create temp file in same directory (for atomic rename)
|
||||
temp_fd, temp_path = tempfile.mkstemp(dir=file_path.parent, prefix=f".{file_path.name}.", suffix=".tmp")
|
||||
|
||||
@@ -48,24 +48,78 @@ def _find_repo_root() -> Path:
|
||||
|
||||
|
||||
_REPO_ROOT = _find_repo_root()
|
||||
_MEMORY_ROOT = Path(__file__).resolve().parents[3]
|
||||
_KNOWN_REGISTRIES_PATH = _MEMORY_ROOT / "memory_json" / "known_registries.json"
|
||||
|
||||
|
||||
def load_known_registries() -> List[Path]:
|
||||
"""Load persisted external registry paths from known_registries.json.
|
||||
|
||||
Returns only paths that currently exist on disk.
|
||||
"""
|
||||
if not _KNOWN_REGISTRIES_PATH.exists():
|
||||
return []
|
||||
try:
|
||||
data = json.loads(_KNOWN_REGISTRIES_PATH.read_text(encoding="utf-8"))
|
||||
return [Path(p) for p in data.get("registries", []) if Path(p).exists()]
|
||||
except Exception as e:
|
||||
logger.warning(f"[detector] Failed to read known_registries.json: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def persist_registry(registry_path: Path) -> None:
|
||||
"""Persist a newly discovered external registry so future runs find it."""
|
||||
current: List[str] = []
|
||||
if _KNOWN_REGISTRIES_PATH.exists():
|
||||
try:
|
||||
data = json.loads(_KNOWN_REGISTRIES_PATH.read_text(encoding="utf-8"))
|
||||
current = data.get("registries", [])
|
||||
except Exception as e:
|
||||
logger.warning(f"[detector] Failed to parse known_registries.json, starting fresh: {e}")
|
||||
resolved = str(registry_path.resolve())
|
||||
if resolved not in current:
|
||||
current.append(resolved)
|
||||
_KNOWN_REGISTRIES_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
_KNOWN_REGISTRIES_PATH.write_text(
|
||||
json.dumps({"registries": current}, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
logger.info(f"[detector] Persisted external registry: {resolved}")
|
||||
|
||||
|
||||
def _find_caller_registries() -> List[Path]:
|
||||
"""Find project registries reachable from CWD (for external projects)."""
|
||||
"""Find all external project registries (persisted + cwd-reachable)."""
|
||||
import os
|
||||
|
||||
aipass_registry = (_REPO_ROOT / "AIPASS_REGISTRY.json").resolve()
|
||||
registries: List[Path] = []
|
||||
seen: set[Path] = set()
|
||||
|
||||
for reg in load_known_registries():
|
||||
resolved = reg.resolve()
|
||||
if resolved != aipass_registry and resolved not in seen:
|
||||
registries.append(reg)
|
||||
seen.add(resolved)
|
||||
|
||||
caller_cwd = (
|
||||
Path(os.environ.get("AIPASS_CALLER_CWD", "")).resolve() if os.environ.get("AIPASS_CALLER_CWD") else Path.cwd()
|
||||
)
|
||||
aipass_registry = (_REPO_ROOT / "AIPASS_REGISTRY.json").resolve()
|
||||
|
||||
registries = []
|
||||
cwd_found: List[Path] = []
|
||||
for parent in [caller_cwd] + list(caller_cwd.parents):
|
||||
for reg in parent.glob("*_REGISTRY.json"):
|
||||
if reg.resolve() != aipass_registry:
|
||||
registries.append(reg)
|
||||
if registries:
|
||||
cwd_found.append(reg)
|
||||
if cwd_found:
|
||||
break
|
||||
|
||||
for reg in cwd_found:
|
||||
resolved = reg.resolve()
|
||||
if resolved not in seen:
|
||||
registries.append(reg)
|
||||
seen.add(resolved)
|
||||
persist_registry(reg)
|
||||
|
||||
return registries
|
||||
|
||||
|
||||
|
||||
@@ -369,32 +369,47 @@ def _get_branch_paths() -> list[Path]:
|
||||
"""
|
||||
Get all branch paths from AIPass registry + external project registries.
|
||||
|
||||
Reads persisted known_registries.json AND does the cwd walk so
|
||||
external-project branches are always reachable regardless of caller cwd.
|
||||
|
||||
Returns:
|
||||
List of Path objects for each branch
|
||||
"""
|
||||
import os
|
||||
|
||||
from aipass.memory.apps.handlers.monitor.detector import load_known_registries, persist_registry
|
||||
|
||||
repo_root = _find_repo_root()
|
||||
paths = _paths_from_registry(repo_root / "AIPASS_REGISTRY.json", repo_root)
|
||||
seen = {p.resolve() for p in paths}
|
||||
aipass_registry = (repo_root / "AIPASS_REGISTRY.json").resolve()
|
||||
|
||||
def _add_from_registry(reg: Path) -> None:
|
||||
for p in _paths_from_registry(reg, reg.parent):
|
||||
if p.resolve() not in seen:
|
||||
paths.append(p)
|
||||
seen.add(p.resolve())
|
||||
|
||||
for reg in load_known_registries():
|
||||
if reg.resolve() != aipass_registry:
|
||||
_add_from_registry(reg)
|
||||
|
||||
caller_cwd = (
|
||||
Path(os.environ.get("AIPASS_CALLER_CWD", "")).resolve() if os.environ.get("AIPASS_CALLER_CWD") else Path.cwd()
|
||||
)
|
||||
aipass_registry = (repo_root / "AIPASS_REGISTRY.json").resolve()
|
||||
|
||||
found_external = False
|
||||
cwd_found: list[Path] = []
|
||||
for parent in [caller_cwd] + list(caller_cwd.parents):
|
||||
for reg in parent.glob("*_REGISTRY.json"):
|
||||
if reg.resolve() != aipass_registry:
|
||||
found_external = True
|
||||
for p in _paths_from_registry(reg, reg.parent):
|
||||
if p.resolve() not in seen:
|
||||
paths.append(p)
|
||||
seen.add(p.resolve())
|
||||
if found_external:
|
||||
cwd_found.append(reg)
|
||||
if cwd_found:
|
||||
break
|
||||
|
||||
for reg in cwd_found:
|
||||
_add_from_registry(reg)
|
||||
persist_registry(reg)
|
||||
|
||||
return paths
|
||||
|
||||
|
||||
|
||||
@@ -595,3 +595,173 @@ class TestRecreateTrinityFile:
|
||||
data = json.loads(recreated.read_text(encoding="utf-8"))
|
||||
assert data["document_metadata"]["document_name"] == "MYBRANCH.LOCAL"
|
||||
assert "limits" not in data["document_metadata"]
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Known registries (persist / load / discovery)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestKnownRegistries:
|
||||
"""Tests for load_known_registries() and persist_registry()."""
|
||||
|
||||
def test_load_returns_empty_when_file_missing(self, tmp_path: Path, monkeypatch):
|
||||
"""No known_registries.json → empty list."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", tmp_path / "nope.json")
|
||||
|
||||
result = detector.load_known_registries()
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_persist_creates_file_and_stores_path(self, tmp_path: Path, monkeypatch):
|
||||
"""persist_registry should create the file and store the absolute path."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
|
||||
ext_reg = tmp_path / "EXT_REGISTRY.json"
|
||||
ext_reg.write_text('{"branches":[]}', encoding="utf-8")
|
||||
|
||||
detector.persist_registry(ext_reg)
|
||||
|
||||
assert kr_path.exists()
|
||||
data = json.loads(kr_path.read_text(encoding="utf-8"))
|
||||
assert str(ext_reg.resolve()) in data["registries"]
|
||||
|
||||
def test_persist_deduplicates(self, tmp_path: Path, monkeypatch):
|
||||
"""Persisting the same registry twice should not create duplicates."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
|
||||
ext_reg = tmp_path / "EXT_REGISTRY.json"
|
||||
ext_reg.write_text('{"branches":[]}', encoding="utf-8")
|
||||
|
||||
detector.persist_registry(ext_reg)
|
||||
detector.persist_registry(ext_reg)
|
||||
|
||||
data = json.loads(kr_path.read_text(encoding="utf-8"))
|
||||
assert len(data["registries"]) == 1
|
||||
|
||||
def test_load_filters_nonexistent_paths(self, tmp_path: Path, monkeypatch):
|
||||
"""load_known_registries filters out paths that no longer exist."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
|
||||
existing = tmp_path / "REAL_REGISTRY.json"
|
||||
existing.write_text('{"branches":[]}', encoding="utf-8")
|
||||
|
||||
kr_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"registries": [str(existing), "/nonexistent/GHOST_REGISTRY.json"],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = detector.load_known_registries()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0] == existing
|
||||
|
||||
def test_load_handles_malformed_json(self, tmp_path: Path, monkeypatch):
|
||||
"""Malformed known_registries.json → empty list, not crash."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
kr_path.write_text("NOT JSON", encoding="utf-8")
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
|
||||
result = detector.load_known_registries()
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_find_caller_registries_includes_known(self, tmp_path: Path, monkeypatch):
|
||||
"""_find_caller_registries should include registries from known_registries.json."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
|
||||
ext_project = tmp_path / "ext_project"
|
||||
ext_project.mkdir()
|
||||
ext_reg = ext_project / "MYPROJECT_REGISTRY.json"
|
||||
ext_reg.write_text('{"branches":[]}', encoding="utf-8")
|
||||
|
||||
kr_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"registries": [str(ext_reg)],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
aipass_reg = tmp_path / "AIPASS_REGISTRY.json"
|
||||
aipass_reg.write_text('{"branches":[]}', encoding="utf-8")
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
|
||||
monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
|
||||
|
||||
result = detector._find_caller_registries()
|
||||
|
||||
resolved_paths = [r.resolve() for r in result]
|
||||
assert ext_reg.resolve() in resolved_paths
|
||||
|
||||
def test_read_registry_discovers_external_branches_via_known(self, tmp_path: Path, monkeypatch):
|
||||
"""_read_registry should find external branches via known_registries.json
|
||||
even when cwd is AIPass root (the core bug from #664)."""
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
core_dir = tmp_path / "aipass"
|
||||
core_dir.mkdir()
|
||||
core_reg = core_dir / "AIPASS_REGISTRY.json"
|
||||
core_reg.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"branches": [{"name": "memory", "path": "src/memory"}],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
ext_project = tmp_path / "myproject"
|
||||
ext_project.mkdir()
|
||||
ext_branch = ext_project / "src" / "mybranch"
|
||||
ext_branch.mkdir(parents=True)
|
||||
ext_reg = ext_project / "MYPROJECT_REGISTRY.json"
|
||||
ext_reg.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"branches": [{"name": "mybranch", "path": "src/mybranch"}],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
kr_path = tmp_path / "known_registries.json"
|
||||
kr_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"registries": [str(ext_reg)],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", core_dir)
|
||||
monkeypatch.setattr(detector, "_KNOWN_REGISTRIES_PATH", kr_path)
|
||||
monkeypatch.setenv("AIPASS_CALLER_CWD", str(core_dir))
|
||||
|
||||
branches = detector._read_registry()
|
||||
|
||||
names = [b["name"] for b in branches]
|
||||
assert "memory" in names
|
||||
assert "mybranch" in names
|
||||
|
||||
@@ -51,17 +51,9 @@ def _fresh_memory_files(monkeypatch):
|
||||
# of the import. Otherwise, remove the mock so Python can discover
|
||||
# the real package on disk.
|
||||
# Try to find the real package by importing with the mock removed
|
||||
saved = sys.modules.pop("aipass.memory.apps.handlers.json", None)
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json", None)
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json.memory_files", None)
|
||||
|
||||
try:
|
||||
# Import the real package so memory_files can be found
|
||||
sys.modules.get("aipass.memory.apps.handlers.json")
|
||||
except Exception:
|
||||
# If we can't import the real package, restore the mock
|
||||
if saved is not None:
|
||||
sys.modules["aipass.memory.apps.handlers.json"] = saved
|
||||
|
||||
# Now force-reimport memory_files via importlib.reload or fresh import
|
||||
mem_files_key = "aipass.memory.apps.handlers.json.memory_files"
|
||||
existing = sys.modules.get(mem_files_key)
|
||||
@@ -588,3 +580,135 @@ class TestUpdateMetadata:
|
||||
assert status["health"] == "healthy"
|
||||
assert status["current_lines"] == 200
|
||||
assert status["last_health_check"] == "2026-03-24"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# _check_entry_counts (soft count guard)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class TestCheckEntryCounts:
|
||||
"""Tests for the soft entry-count guard added for #664."""
|
||||
|
||||
@staticmethod
|
||||
def _warn_calls(memory_files_mod): # type: ignore[no-untyped-def]
|
||||
"""Get warning call args from the mocked logger."""
|
||||
return getattr(memory_files_mod.logger, "warning").call_args_list
|
||||
|
||||
@staticmethod
|
||||
def _reset_warns(memory_files_mod): # type: ignore[no-untyped-def]
|
||||
"""Reset warning mock on the mocked logger."""
|
||||
getattr(memory_files_mod.logger, "warning").reset_mock()
|
||||
|
||||
def test_warns_when_over_rollover_limit(self, tmp_path: Path, monkeypatch) -> None:
|
||||
"""Writing more entries than rollover count limit should log a warning."""
|
||||
from aipass.memory.apps.handlers.json import memory_files
|
||||
|
||||
monkeypatch.setattr(
|
||||
memory_files.config_loader,
|
||||
"section",
|
||||
lambda name: {
|
||||
"per_branch": {},
|
||||
"defaults": {"local": {"sessions": {"count": 5}}},
|
||||
},
|
||||
)
|
||||
|
||||
trinity_dir = tmp_path / "testbranch" / ".trinity"
|
||||
trinity_dir.mkdir(parents=True)
|
||||
file_path = trinity_dir / "local.json"
|
||||
|
||||
data = {"sessions": [{"id": f"s{i}"} for i in range(10)]}
|
||||
|
||||
memory_files._check_entry_counts(file_path, data)
|
||||
|
||||
calls = self._warn_calls(memory_files)
|
||||
assert any("ENTRY COUNT" in str(c) and "10/5" in str(c) for c in calls)
|
||||
|
||||
def test_no_warn_when_under_limit(self, tmp_path: Path, monkeypatch) -> None:
|
||||
"""Entries under rollover limit should produce no warning."""
|
||||
from aipass.memory.apps.handlers.json import memory_files
|
||||
|
||||
monkeypatch.setattr(
|
||||
memory_files.config_loader,
|
||||
"section",
|
||||
lambda name: {
|
||||
"per_branch": {},
|
||||
"defaults": {"local": {"sessions": {"count": 20}}},
|
||||
},
|
||||
)
|
||||
|
||||
trinity_dir = tmp_path / "testbranch" / ".trinity"
|
||||
trinity_dir.mkdir(parents=True)
|
||||
file_path = trinity_dir / "local.json"
|
||||
|
||||
data = {"sessions": [{"id": f"s{i}"} for i in range(5)]}
|
||||
|
||||
self._reset_warns(memory_files)
|
||||
memory_files._check_entry_counts(file_path, data)
|
||||
|
||||
calls = self._warn_calls(memory_files)
|
||||
assert not any("ENTRY COUNT" in str(c) for c in calls)
|
||||
|
||||
def test_skips_non_trinity_files(self, tmp_path: Path, monkeypatch) -> None:
|
||||
"""Non-.trinity/ files should be silently skipped."""
|
||||
from aipass.memory.apps.handlers.json import memory_files
|
||||
|
||||
file_path = tmp_path / "random.json"
|
||||
data = {"sessions": [{"id": f"s{i}"} for i in range(100)]}
|
||||
|
||||
self._reset_warns(memory_files)
|
||||
memory_files._check_entry_counts(file_path, data)
|
||||
|
||||
calls = self._warn_calls(memory_files)
|
||||
assert not any("ENTRY COUNT" in str(c) for c in calls)
|
||||
|
||||
def test_uses_per_branch_override(self, tmp_path: Path, monkeypatch) -> None:
|
||||
"""Per-branch rollover config should take precedence over defaults."""
|
||||
from aipass.memory.apps.handlers.json import memory_files
|
||||
|
||||
monkeypatch.setattr(
|
||||
memory_files.config_loader,
|
||||
"section",
|
||||
lambda name: {
|
||||
"per_branch": {"mybranch": {"local": {"sessions": {"count": 3}}}},
|
||||
"defaults": {"local": {"sessions": {"count": 100}}},
|
||||
},
|
||||
)
|
||||
|
||||
trinity_dir = tmp_path / "mybranch" / ".trinity"
|
||||
trinity_dir.mkdir(parents=True)
|
||||
file_path = trinity_dir / "local.json"
|
||||
|
||||
data = {"sessions": [{"id": f"s{i}"} for i in range(5)]}
|
||||
|
||||
memory_files._check_entry_counts(file_path, data)
|
||||
|
||||
calls = self._warn_calls(memory_files)
|
||||
assert any("ENTRY COUNT" in str(c) and "5/3" in str(c) for c in calls)
|
||||
|
||||
def test_does_not_block_write(self, tmp_path: Path, monkeypatch) -> None:
|
||||
"""Entry count guard should never prevent write_memory_file from succeeding."""
|
||||
from aipass.memory.apps.handlers.json import memory_files
|
||||
|
||||
monkeypatch.setattr(
|
||||
memory_files.config_loader,
|
||||
"section",
|
||||
lambda name: {
|
||||
"per_branch": {},
|
||||
"defaults": {"local": {"sessions": {"count": 2}}},
|
||||
},
|
||||
)
|
||||
|
||||
trinity_dir = tmp_path / "testbranch" / ".trinity"
|
||||
trinity_dir.mkdir(parents=True)
|
||||
file_path = trinity_dir / "local.json"
|
||||
|
||||
data = {
|
||||
"document_metadata": {"document_type": "test"},
|
||||
"sessions": [{"id": f"s{i}"} for i in range(20)],
|
||||
}
|
||||
|
||||
result = memory_files.write_memory_file(file_path, data)
|
||||
|
||||
assert result["success"] is True
|
||||
assert file_path.exists()
|
||||
|
||||
Reference in New Issue
Block a user