feat(memory): FPLAN-0270 Phase 3 — changed_entries diff helper + write_memory_file enforcement (warn-only, rollover-safe)

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) <noreply@anthropic.com>

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": {

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

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")

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