From 70643755896a495457afefaed28ca9652a048c84 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sat, 13 Jun 2026 01:55:58 -0700 Subject: [PATCH] =?UTF-8?q?feat(memory):=20FPLAN-0270=20Phase=203=20?= =?UTF-8?q?=E2=80=94=20changed=5Fentries=20diff=20helper=20+=20write=5Fmem?= =?UTF-8?q?ory=5Ffile=20enforcement=20(warn-only,=20rollover-safe)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit changed_entries(before,after,limits): pure diff that flags only NEW/CHANGED over-limit entries, ignoring unchanged legacy fat — so rollover (trims by count, writes back recent fat entries) is never rejected. Wired into write_memory_file via _validate_entry_limits (gates only .trinity/{local,observations}.json): warn mode logs+writes, enforce mode rejects new/changed over-limit only. Validation wrapped in try/except → a validator bug can never abort a write. +15 tests (917 total), seedgo 100%, enforce stays false. Verified by artifact incl. live proof of rollover-safety + the defensive guarantee. @memory side (P1-3) complete. The changed_entries() helper is what @hooks imports next. Part of DPLAN-0205. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/aipass/memory/.seedgo/bypass.json | 15 + .../memory/apps/handlers/json/entry_limits.py | 192 +++++++- .../memory/apps/handlers/json/memory_files.py | 102 +++- .../memory/tests/test_changed_entries.py | 444 ++++++++++++++++++ 4 files changed, 747 insertions(+), 6 deletions(-) create mode 100644 src/aipass/memory/tests/test_changed_entries.py diff --git a/src/aipass/memory/.seedgo/bypass.json b/src/aipass/memory/.seedgo/bypass.json index 1dc3a82a..96fc0208 100644 --- a/src/aipass/memory/.seedgo/bypass.json +++ b/src/aipass/memory/.seedgo/bypass.json @@ -660,6 +660,21 @@ "file": "tests/test_entry_limits.py", "standard": "meta", "reason": "Test file — META block present at lines 1-7; hook false-positive on test file format." + }, + { + "file": "tests/test_changed_entries.py", + "standard": "architecture", + "reason": "Test file — lives in tests/ by design, not in 3-layer apps/ structure." + }, + { + "file": "tests/test_changed_entries.py", + "standard": "documentation", + "reason": "Test file — test functions don't require docstrings." + }, + { + "file": "tests/test_changed_entries.py", + "standard": "meta", + "reason": "Test file — META block present at lines 1-7; hook false-positive on test file format." } ], "notes": { diff --git a/src/aipass/memory/apps/handlers/json/entry_limits.py b/src/aipass/memory/apps/handlers/json/entry_limits.py index d5a21dd5..aa6888e7 100644 --- a/src/aipass/memory/apps/handlers/json/entry_limits.py +++ b/src/aipass/memory/apps/handlers/json/entry_limits.py @@ -1,13 +1,13 @@ # =================== AIPass ==================== # Name: entry_limits.py -# Description: Entry limits config reader and validator for memory files -# Version: 1.1.0 +# Description: Entry limits config reader, validator, and diff helper for memory files +# Version: 1.2.0 # Created: 2026-06-13 # Modified: 2026-06-13 # ============================================= """ -Entry Limits Config Reader & Validator +Entry Limits Config Reader, Validator & Diff Helper Reads the entry_limits section from memory.config.json and returns the effective limits for a given branch, with per_branch overrides @@ -16,12 +16,22 @@ deep-merged over the default entry_types. Provides ``check_entry()`` — a pure validator that checks whether a single entry text exceeds its character cap. +Provides ``changed_entries()`` — a pure diff helper that compares +before/after file dicts and returns only NEW or CHANGED entries that +exceed their character cap. Unchanged legacy over-limit entries pass +untouched (rollover-safe). + Usage: - from aipass.memory.apps.handlers.json.entry_limits import load_entry_limits, check_entry + from aipass.memory.apps.handlers.json.entry_limits import ( + load_entry_limits, check_entry, changed_entries, + ) limits = load_entry_limits("devpulse") verdict = check_entry("key_learnings", some_text, limits) # => {"ok": True/False, "length": int, "cap": int, "over_by": int, "entry_type": str} + + violations = changed_entries(before_dict, after_dict, limits) + # => [{"entry_type", "container", "key", "length", "cap", "over_by"}, ...] """ import copy @@ -243,3 +253,177 @@ def check_entry(entry_type: str, text: str, limits: dict[str, Any]) -> dict[str, "over_by": over_by, "entry_type": entry_type, } + + +# --------------------------------------------------------------------------- +# Phase 3: changed-entries diff helper (rollover-safe) +# --------------------------------------------------------------------------- + + +def _extract_text(value: Any, field: str) -> str: + """Extract the text payload from a container entry. + + For dict containers the value may be a plain string or a dict + with a *field* key (e.g. ``{"value": "some text", ...}``). + For list containers the entry is always a dict with a *field* key. + + Args: + value: The entry value (string or dict). + field: The field name to extract from a dict value. + + Returns: + The text string, or ``""`` if extraction fails. + """ + if isinstance(value, str): + return value + if isinstance(value, dict): + text = value.get(field, "") + return text if isinstance(text, str) else "" + return "" + + +def _check_dict_container( + type_name: str, + container: str, + field: str, + before_container: Any, + after_container: Any, + limits: dict[str, Any], +) -> list[dict[str, Any]]: + """Check dict-shaped container for new/changed over-limit entries. + + Args: + type_name: Entry type name (e.g. ``"key_learnings"``). + container: Container key in the file dict. + field: Field to extract text from dict-valued entries. + before_container: The container value from the on-disk file. + after_container: The container value from the proposed file. + limits: The dict returned by :func:`load_entry_limits`. + + Returns: + List of violation dicts for new/changed entries that exceed cap. + """ + if not isinstance(after_container, dict): + return [] + before_dict = before_container if isinstance(before_container, dict) else {} + hits: list[dict[str, Any]] = [] + + for key, after_value in after_container.items(): + after_text = _extract_text(after_value, field) + if key in before_dict and after_text == _extract_text(before_dict[key], field): + continue # Unchanged — skip even if over-limit + verdict = check_entry(type_name, after_text, limits) + if not verdict["ok"]: + hits.append( + { + "entry_type": type_name, + "container": container, + "key": key, + "length": verdict["length"], + "cap": verdict["cap"], + "over_by": verdict["over_by"], + } + ) + return hits + + +def _check_list_container( + type_name: str, + container: str, + field: str, + before_container: Any, + after_container: Any, + limits: dict[str, Any], +) -> list[dict[str, Any]]: + """Check list-shaped container for new/changed over-limit entries. + + Args: + type_name: Entry type name (e.g. ``"sessions"``). + container: Container key in the file dict. + field: Field to extract text from list-item dicts. + before_container: The container value from the on-disk file. + after_container: The container value from the proposed file. + limits: The dict returned by :func:`load_entry_limits`. + + Returns: + List of violation dicts for new/changed entries that exceed cap. + """ + if not isinstance(after_container, list): + return [] + before_list = before_container if isinstance(before_container, list) else [] + hits: list[dict[str, Any]] = [] + + for idx, after_item in enumerate(after_container): + after_text = _extract_text(after_item, field) + if idx < len(before_list) and after_text == _extract_text(before_list[idx], field): + continue # Unchanged — skip even if over-limit + verdict = check_entry(type_name, after_text, limits) + if not verdict["ok"]: + hits.append( + { + "entry_type": type_name, + "container": container, + "key": str(idx), + "length": verdict["length"], + "cap": verdict["cap"], + "over_by": verdict["over_by"], + } + ) + return hits + + +def changed_entries( + before: dict[str, Any], + after: dict[str, Any], + limits: dict[str, Any], +) -> list[dict[str, Any]]: + """Return over-limit entries that are NEW or CHANGED between *before* and *after*. + + This is a **pure function** — no I/O, no file reads, no side effects. + Unchanged entries (even if over-limit) are intentionally skipped so + that rollover and other maintenance writes are never blocked by + legacy fat entries. + + Args: + before: Parsed .trinity file dict (current on-disk content). + after: Parsed .trinity file dict (proposed new content). + limits: The dict returned by :func:`load_entry_limits`. + + Returns: + List of violation dicts, each containing:: + + { + "entry_type": str, # e.g. "key_learnings" + "container": str, # e.g. "key_learnings" + "key": str, # dict key or list index (as str) + "length": int, # len(text) + "cap": int, # max_chars + "over_by": int, # length - cap + } + + Empty list when everything is within limits or unchanged. + """ + entry_types = limits.get("entry_types", {}) + violations: list[dict[str, Any]] = [] + + for type_name, type_def in entry_types.items(): + container = type_def.get("container", "") + kind = type_def.get("kind", "dict") + field = type_def.get("field", "value") + + after_container = after.get(container) + if after_container is None: + continue + + before_container = before.get(container) + + if kind == "dict": + violations.extend( + _check_dict_container(type_name, container, field, before_container, after_container, limits) + ) + elif kind == "list": + violations.extend( + _check_list_container(type_name, container, field, before_container, after_container, limits) + ) + + return violations diff --git a/src/aipass/memory/apps/handlers/json/memory_files.py b/src/aipass/memory/apps/handlers/json/memory_files.py index db4751d0..21e4f721 100644 --- a/src/aipass/memory/apps/handlers/json/memory_files.py +++ b/src/aipass/memory/apps/handlers/json/memory_files.py @@ -1,9 +1,9 @@ # =================== AIPass ==================== # Name: memory_files.py # Description: Memory File Safe I/O Handler -# Version: 1.0.0 +# Version: 1.1.0 # Created: 2026-03-17 -# Modified: 2026-03-17 +# Modified: 2026-06-13 # ============================================= """ @@ -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.entry_limits import load_entry_limits, changed_entries logger = get_system_logger() @@ -45,6 +46,93 @@ _TEMPLATES_DIR = _MEMORY_ROOT / "apps" / "json_templates" # No service imports - handlers are pure workers (3-tier architecture) # No module imports (handler independence) +# Files tracked by entry-limits validation +_TRACKED_TRINITY_FILES = {"local.json", "observations.json"} + + +# ============================================================================= +# ENTRY-LIMITS VALIDATION (rollover-safe) +# ============================================================================= + + +def _validate_entry_limits( + file_path: Path, + data: Dict[str, Any], +) -> Optional[Dict[str, Any]]: + """Check entry limits for a .trinity/ write and return a rejection or None. + + Only validates files inside a ``.trinity/`` directory whose name is + in :data:`_TRACKED_TRINITY_FILES`. For all other paths this function + returns ``None`` immediately (no validation). + + Unchanged entries (same text as on disk) are intentionally skipped so + that rollover and other maintenance writes are never blocked by + legacy over-limit entries. + + Args: + file_path: Target path for the write. + data: The dict about to be written. + + Returns: + ``None`` when the write should proceed normally. + A ``{"success": False, "error": ...}`` dict when enforce mode is + on and new/changed entries exceed their caps. + """ + # --- Gate: only tracked .trinity files ------------------------------------ + if file_path.parent.name != ".trinity": + return None + if file_path.name not in _TRACKED_TRINITY_FILES: + return None + + branch = file_path.parent.parent.name + limits = load_entry_limits(branch) + + if not limits.get("enabled", True): + return None + + # Filter limits to entry_types that belong to THIS file + filtered_types = { + name: tdef for name, tdef in limits.get("entry_types", {}).items() if tdef.get("file") == file_path.name + } + if not filtered_types: + return None + + filtered_limits = { + "enabled": limits["enabled"], + "enforce": limits["enforce"], + "entry_types": filtered_types, + } + + # Read current on-disk content (before) + before: Dict[str, Any] = {} + if file_path.exists(): + try: + before = json.loads(file_path.read_text(encoding="utf-8")) + except Exception as exc: + logger.warning(f"[entry_limits] Could not parse {file_path.name} for diff: {exc}") + before = {} # Unparseable — treat as empty (all entries "new") + + over = changed_entries(before, data, filtered_limits) + if not over: + return None + + # --- Violations found ----------------------------------------------------- + enforce = limits.get("enforce", False) + + if not enforce: + for violation in over: + logger.warning( + f"[entry_limits] WARN {branch} {file_path.name} " + f"{violation['container']}[{violation['key']}] " + f"{violation['length']}/{violation['cap']} " + f"(+{violation['over_by']} over)" + ) + return None # Write through in warn mode + + # Enforce mode — block the write + details = "; ".join(f"{v['container']}[{v['key']}] {v['length']}/{v['cap']} (+{v['over_by']} over)" for v in over) + return {"success": False, "error": f"Entry limit exceeded: {details}"} + # ============================================================================= # CORE READ/WRITE OPERATIONS @@ -120,6 +208,16 @@ def write_memory_file(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]: if not isinstance(data, dict): return {"success": False, "error": f"Data must be dict, got {type(data).__name__}"} + # --- Entry-limits validation (rollover-safe) ------------------------------ + # Only validate .trinity/ files that are tracked (local.json, observations.json). + # Unchanged legacy over-limit entries pass untouched so rollover is never blocked. + try: + rejection = _validate_entry_limits(file_path, data) + if rejection is not None: + return rejection + except Exception as exc: + logger.warning(f"[memory_files] Entry-limits validation 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") diff --git a/src/aipass/memory/tests/test_changed_entries.py b/src/aipass/memory/tests/test_changed_entries.py new file mode 100644 index 00000000..8b098627 --- /dev/null +++ b/src/aipass/memory/tests/test_changed_entries.py @@ -0,0 +1,444 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_changed_entries.py +# Date: 2026-06-13 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +""" +Tests for Phase 3 of FPLAN-0270: changed_entries diff helper and +write_memory_file entry-limits wiring. + +Covers: + - changed_entries: new over-limit, changed over-limit, unchanged legacy + fat entries (rollover-safe), shrinking, dict/list containers, empty before. + - write_memory_file wiring: warn mode writes through + logs, enforce mode + rejects new fat entries, enforce mode allows unchanged legacy fat entries, + non-trinity files unaffected, passport.json unaffected. +""" + +import importlib +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + + +# --------------------------------------------------------------------------- +# Per-test fixture: fresh-import modules with mocks in place +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _fresh_modules(monkeypatch): + """Drop cached modules so each test gets fresh imports.""" + sys.modules.pop("aipass.memory.apps.handlers.json", None) + sys.modules.pop("aipass.memory.apps.handlers.json.json_handler", None) + sys.modules.pop("aipass.memory.apps.handlers.json.entry_limits", None) + sys.modules.pop("aipass.memory.apps.handlers.json.memory_files", None) + sys.modules.pop("aipass.memory.apps.handlers.json.lint_handler", None) + yield + + +def _get_entry_limits(): + """Import and return the entry_limits module.""" + return importlib.import_module("aipass.memory.apps.handlers.json.entry_limits") + + +def _get_memory_files(): + """Import and return the memory_files module.""" + return importlib.import_module("aipass.memory.apps.handlers.json.memory_files") + + +# --------------------------------------------------------------------------- +# Helpers: build limits dicts for testing +# --------------------------------------------------------------------------- + +_KEY_LEARNINGS_ONLY: dict[str, Any] = { + "enabled": True, + "enforce": False, + "entry_types": { + "key_learnings": { + "file": "local.json", + "container": "key_learnings", + "kind": "dict", + "field": "value", + "max_chars": 200, + }, + }, +} + +_SESSIONS_ONLY: dict[str, Any] = { + "enabled": True, + "enforce": False, + "entry_types": { + "sessions": { + "file": "local.json", + "container": "sessions", + "kind": "list", + "field": "summary", + "max_chars": 300, + }, + }, +} + + +def _full_limits(**overrides: Any) -> dict[str, Any]: + """Return a complete limits dict with all four default entry types.""" + base: dict[str, Any] = { + "enabled": True, + "enforce": False, + "entry_types": { + "key_learnings": { + "file": "local.json", + "container": "key_learnings", + "kind": "dict", + "field": "value", + "max_chars": 200, + }, + "sessions": { + "file": "local.json", + "container": "sessions", + "kind": "list", + "field": "summary", + "max_chars": 300, + }, + "todos": { + "file": "local.json", + "container": "todos", + "kind": "list", + "field": "task", + "max_chars": 200, + }, + "observations": { + "file": "observations.json", + "container": "observations", + "kind": "list", + "field": "note", + "max_chars": 600, + }, + }, + } + base.update(overrides) + return base + + +# =========================================================================== +# 1. changed_entries: new over-limit entry detected +# =========================================================================== + + +class TestNewOverLimitEntry: + """A new dict entry that exceeds the cap is returned as a violation.""" + + def test_new_overlimit_key_learning(self) -> None: + mod = _get_entry_limits() + before = {"key_learnings": {"a": "short", "b": "also short"}} + fat_text = "x" * 250 + after = {"key_learnings": {"a": "short", "b": "also short", "c": fat_text}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert len(result) == 1 + assert result[0]["entry_type"] == "key_learnings" + assert result[0]["key"] == "c" + assert result[0]["length"] == 250 + assert result[0]["cap"] == 200 + assert result[0]["over_by"] == 50 + + +# =========================================================================== +# 2. changed_entries: changed entry exceeds cap +# =========================================================================== + + +class TestChangedEntryOverCap: + """An existing entry whose text grew past the cap is flagged.""" + + def test_changed_key_learning_over_cap(self) -> None: + mod = _get_entry_limits() + before = {"key_learnings": {"a": "short text"}} + after = {"key_learnings": {"a": "y" * 300}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert len(result) == 1 + assert result[0]["key"] == "a" + assert result[0]["over_by"] == 100 + + +# =========================================================================== +# 3. changed_entries: UNCHANGED legacy over-limit entry NOT returned +# =========================================================================== + + +class TestUnchangedLegacyFatEntry: + """THE KEY TEST: unchanged fat entries must NOT be flagged (rollover-safe).""" + + def test_unchanged_500char_key_learning_not_flagged(self) -> None: + mod = _get_entry_limits() + fat_text = "z" * 500 + before = {"key_learnings": {"legacy": fat_text}} + after = {"key_learnings": {"legacy": fat_text}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert result == [] + + +# =========================================================================== +# 4. changed_entries: shrinking an entry is not flagged +# =========================================================================== + + +class TestShrinkingEntry: + """An entry that went from 500 chars to 100 is not flagged.""" + + def test_shrunk_entry_not_flagged(self) -> None: + mod = _get_entry_limits() + before = {"key_learnings": {"item": "z" * 500}} + after = {"key_learnings": {"item": "z" * 100}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert result == [] + + +# =========================================================================== +# 5. changed_entries: dict container — value-as-string and value-as-dict +# =========================================================================== + + +class TestDictContainerShapes: + """Both plain-string and dict-with-field value shapes are handled.""" + + def test_value_as_string(self) -> None: + mod = _get_entry_limits() + before: dict[str, Any] = {"key_learnings": {}} + after = {"key_learnings": {"new_key": "x" * 250}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert len(result) == 1 + assert result[0]["length"] == 250 + + def test_value_as_dict_with_field(self) -> None: + mod = _get_entry_limits() + before: dict[str, Any] = {"key_learnings": {}} + after = {"key_learnings": {"new_key": {"value": "x" * 250, "source": "test"}}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert len(result) == 1 + assert result[0]["length"] == 250 + + +# =========================================================================== +# 6. changed_entries: list container — appended and unchanged +# =========================================================================== + + +class TestListContainer: + """List containers detect new appended items and skip unchanged ones.""" + + def test_appended_item_over_cap_detected(self) -> None: + mod = _get_entry_limits() + existing = {"session_number": 1, "summary": "short"} + new_fat = {"session_number": 2, "summary": "s" * 400} + before = {"sessions": [existing]} + after = {"sessions": [existing, new_fat]} + + result = mod.changed_entries(before, after, _SESSIONS_ONLY) + + assert len(result) == 1 + assert result[0]["key"] == "1" + assert result[0]["over_by"] == 100 + + def test_existing_unchanged_items_not_flagged(self) -> None: + mod = _get_entry_limits() + fat_item = {"session_number": 1, "summary": "s" * 400} + before = {"sessions": [fat_item]} + after = {"sessions": [fat_item]} + + result = mod.changed_entries(before, after, _SESSIONS_ONLY) + + assert result == [] + + +# =========================================================================== +# 7. changed_entries: empty before (new file) — all entries treated as new +# =========================================================================== + + +class TestEmptyBefore: + """When before is empty, all after entries are treated as new.""" + + def test_all_over_limit_entries_flagged(self) -> None: + mod = _get_entry_limits() + before: dict[str, Any] = {} + after = {"key_learnings": {"a": "x" * 250, "b": "ok"}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert len(result) == 1 + assert result[0]["key"] == "a" + + def test_within_limit_entries_not_flagged(self) -> None: + mod = _get_entry_limits() + before: dict[str, Any] = {} + after = {"key_learnings": {"a": "short", "b": "also short"}} + + result = mod.changed_entries(before, after, _KEY_LEARNINGS_ONLY) + + assert result == [] + + +# =========================================================================== +# 8. write_memory_file: warn mode writes through + logs warning +# =========================================================================== + + +class TestWarnModeWritesThrough: + """In warn mode (enforce=False), over-limit entries log a warning but file is written.""" + + def test_warn_mode_writes_and_logs(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mem_mod = _get_memory_files() + mock_logger = mem_mod.logger + + # Build .trinity/local.json path + trinity = tmp_path / "test_branch" / ".trinity" + trinity.mkdir(parents=True) + local_path = trinity / "local.json" + + before_data = {"key_learnings": {"existing": "short"}} + local_path.write_text(json.dumps(before_data, indent=2), encoding="utf-8") + + fat_text = "x" * 300 + after_data = {"key_learnings": {"existing": "short", "new_fat": fat_text}} + + warn_limits = _full_limits(enforce=False) + monkeypatch.setattr(mem_mod, "load_entry_limits", lambda branch: warn_limits) + + result = mem_mod.write_memory_file(local_path, after_data) + + assert result["success"] is True + written = json.loads(local_path.read_text(encoding="utf-8")) + assert written["key_learnings"]["new_fat"] == fat_text + mock_logger.warning.assert_called() + warning_calls = [str(c) for c in mock_logger.warning.call_args_list] + assert any("entry_limits" in w for w in warning_calls) + + +# =========================================================================== +# 9. write_memory_file: enforce mode rejects new over-limit entry +# =========================================================================== + + +class TestEnforceModeRejects: + """In enforce mode, a new over-limit entry is rejected and file is unchanged.""" + + def test_enforce_rejects_new_fat_entry(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mem_mod = _get_memory_files() + + trinity = tmp_path / "test_branch" / ".trinity" + trinity.mkdir(parents=True) + local_path = trinity / "local.json" + + before_data = {"key_learnings": {"existing": "short"}} + local_path.write_text(json.dumps(before_data, indent=2), encoding="utf-8") + + fat_text = "x" * 300 + after_data = {"key_learnings": {"existing": "short", "new_fat": fat_text}} + + enforce_limits = _full_limits(enforce=True) + monkeypatch.setattr(mem_mod, "load_entry_limits", lambda branch: enforce_limits) + + result = mem_mod.write_memory_file(local_path, after_data) + + assert result["success"] is False + assert "Entry limit exceeded" in result["error"] + # File on disk is UNCHANGED + on_disk = json.loads(local_path.read_text(encoding="utf-8")) + assert "new_fat" not in on_disk["key_learnings"] + + +# =========================================================================== +# 10. write_memory_file: enforce mode ALLOWS unchanged legacy fat entries +# =========================================================================== + + +class TestEnforceAllowsUnchangedLegacy: + """THE CRITICAL ROLLOVER-SAFE TEST: enforce mode allows writing back same fat data.""" + + def test_enforce_allows_same_data_with_fat_entries( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + mem_mod = _get_memory_files() + + trinity = tmp_path / "test_branch" / ".trinity" + trinity.mkdir(parents=True) + local_path = trinity / "local.json" + + fat_data = {"key_learnings": {"legacy": "z" * 500, "also_fat": "y" * 400}} + local_path.write_text(json.dumps(fat_data, indent=2), encoding="utf-8") + + enforce_limits = _full_limits(enforce=True) + monkeypatch.setattr(mem_mod, "load_entry_limits", lambda branch: enforce_limits) + + result = mem_mod.write_memory_file(local_path, fat_data) + + assert result["success"] is True + on_disk = json.loads(local_path.read_text(encoding="utf-8")) + assert on_disk["key_learnings"]["legacy"] == "z" * 500 + + +# =========================================================================== +# 11. write_memory_file: non-trinity file unaffected +# =========================================================================== + + +class TestNonTrinityFileUnaffected: + """Files outside .trinity/ bypass validation entirely.""" + + def test_writes_normally_outside_trinity(self, tmp_path: Path) -> None: + mem_mod = _get_memory_files() + + output_path = tmp_path / "some_output.json" + data = {"key": "value"} + + result = mem_mod.write_memory_file(output_path, data) + + assert result["success"] is True + assert output_path.exists() + written = json.loads(output_path.read_text(encoding="utf-8")) + assert written == data + + +# =========================================================================== +# 12. write_memory_file: passport.json unaffected +# =========================================================================== + + +class TestPassportUnaffected: + """Writes to .trinity/passport.json bypass validation.""" + + def test_passport_writes_normally(self, tmp_path: Path) -> None: + mem_mod = _get_memory_files() + + trinity = tmp_path / "test_branch" / ".trinity" + trinity.mkdir(parents=True) + passport_path = trinity / "passport.json" + + data = {"branch_info": {"branch_name": "test_branch"}, "identity": {"role": "test"}} + + result = mem_mod.write_memory_file(passport_path, data) + + assert result["success"] is True + assert passport_path.exists() + written = json.loads(passport_path.read_text(encoding="utf-8")) + assert written == data