feat(memory): FPLAN-0270 Phase 1 — entry_limits config (warn-first, enforce:false) + load_entry_limits reader + 14 tests

Config-driven char caps for .trinity memory entries. Phase 1 = foundation only:
adds entry_limits section to memory.config.json (4 caps: learnings 200, sessions
300, todos 200, observations 600) and the load_entry_limits(branch) reader
(deep-merge per_branch overrides, safe-defaults on missing/malformed). Reader has
NO callers yet (Phase 3 wires it) — unused_function bypass is intentional.
Verified by artifact: 14/14 tests, seedgo 100%, scope clean. enforce:false →
zero behavior change. Part of DPLAN-0205.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-13 01:15:54 -07:00
co-authored by Claude Opus 4.8
parent d9a2a48a1e
commit e47d4f0463
4 changed files with 560 additions and 0 deletions
+20
View File
@@ -630,6 +630,26 @@
"file": "tests/test_auto_process.py",
"standard": "documentation",
"reason": "Test file — test functions don't require docstrings."
},
{
"file": "apps/handlers/json/entry_limits.py",
"standard": "unused_function",
"reason": "Phase 1 of FPLAN-0270: load_entry_limits() is deliberately uncalled — Phase 3 wires it into write_memory_file(). Tests cover it."
},
{
"file": "tests/test_entry_limits.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by design, not in 3-layer apps/ structure."
},
{
"file": "tests/test_entry_limits.py",
"standard": "documentation",
"reason": "Test file — test functions don't require docstrings."
},
{
"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."
}
],
"notes": {
@@ -0,0 +1,190 @@
# =================== AIPass ====================
# Name: entry_limits.py
# Description: Entry limits config reader for memory files
# Version: 1.0.0
# Created: 2026-06-13
# Modified: 2026-06-13
# =============================================
"""
Entry Limits Config Reader
Reads the entry_limits section from memory.config.json and returns
the effective limits for a given branch, with per_branch overrides
deep-merged over the default entry_types.
Phase 1 only: reader + safe defaults. No enforcement, no validation,
no write-path integration.
Usage:
from aipass.memory.apps.handlers.json.entry_limits import load_entry_limits
limits = load_entry_limits("devpulse")
# => {"enabled": True, "enforce": False, "entry_types": {...}}
"""
import copy
import json
from pathlib import Path
from typing import Any
from aipass.prax import logger
from aipass.memory.apps.handlers.json import json_handler
# Resolve paths relative to handler location (same pattern as memory_files.py)
_MEMORY_ROOT = Path(__file__).resolve().parents[3]
_CONFIG_PATH = _MEMORY_ROOT / "config" / "memory.config.json"
# Safe defaults — returned when config is missing or malformed.
# These match the canonical values in memory.config.json.
_SAFE_DEFAULTS: 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,
},
},
}
def _deep_merge_entry_types(
base: dict[str, Any],
overrides: dict[str, Any],
) -> dict[str, Any]:
"""Deep-merge per_branch overrides into entry_types.
For each key in *overrides*:
- If the key exists in *base*, shallow-merge the override dict
into a copy of the base dict (override wins per field).
- If the key is new, add it verbatim (new entry type for branch).
Args:
base: Default entry_types dict.
overrides: per_branch[branch] dict (same shape as entry_types).
Returns:
Merged entry_types dict. The originals are not mutated.
"""
merged = copy.deepcopy(base)
for type_name, type_overrides in overrides.items():
if type_name in merged:
merged[type_name].update(type_overrides)
else:
merged[type_name] = copy.deepcopy(type_overrides)
return merged
def load_entry_limits(branch: str) -> dict[str, Any]:
"""Load effective entry limits for *branch*.
Reads memory.config.json, pulls the ``entry_limits`` section, then
deep-merges any ``per_branch[branch]`` overrides on top of the
default ``entry_types``.
Graceful degradation:
- Missing config file -> safe defaults + warning log.
- Malformed JSON -> safe defaults + loud warning log.
- Missing entry_limits section -> safe defaults + warning log.
Args:
branch: Branch name (e.g. "devpulse", "memory").
Returns:
Dict with keys: enabled, enforce, entry_types.
"""
# --- Attempt to read config ------------------------------------------------
try:
raw_text = _CONFIG_PATH.read_text(encoding="utf-8")
except FileNotFoundError:
logger.warning(f"[entry_limits] Config file not found at {_CONFIG_PATH}, returning safe defaults")
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "fallback": "missing_config"},
module_name="entry_limits",
)
return copy.deepcopy(_SAFE_DEFAULTS)
except OSError as exc:
logger.warning(f"[entry_limits] Could not read config at {_CONFIG_PATH}: {exc}, returning safe defaults")
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "fallback": "read_error"},
module_name="entry_limits",
)
return copy.deepcopy(_SAFE_DEFAULTS)
# --- Parse JSON ------------------------------------------------------------
try:
config = json.loads(raw_text)
except json.JSONDecodeError as exc:
logger.warning(f"[entry_limits] Malformed JSON in {_CONFIG_PATH}: {exc}, returning safe defaults")
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "fallback": "malformed_json", "error": str(exc)},
module_name="entry_limits",
)
return copy.deepcopy(_SAFE_DEFAULTS)
# --- Extract entry_limits section ------------------------------------------
section = config.get("entry_limits")
if not isinstance(section, dict):
logger.warning("[entry_limits] No valid 'entry_limits' section in config, returning safe defaults")
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "fallback": "missing_section"},
module_name="entry_limits",
)
return copy.deepcopy(_SAFE_DEFAULTS)
# --- Build effective result ------------------------------------------------
enabled = section.get("enabled", True)
enforce = section.get("enforce", False)
base_types = section.get("entry_types", {})
# Apply per_branch overrides if present
per_branch = section.get("per_branch", {})
branch_overrides = per_branch.get(branch, {})
if branch_overrides:
effective_types = _deep_merge_entry_types(base_types, branch_overrides)
else:
effective_types = copy.deepcopy(base_types)
result: dict[str, Any] = {
"enabled": enabled,
"enforce": enforce,
"entry_types": effective_types,
}
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "types_count": len(effective_types)},
module_name="entry_limits",
)
return result
@@ -21,5 +21,16 @@
"intake": {
"enabled": false,
"pool_dir": "memory_pool"
},
"entry_limits": {
"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}
},
"per_branch": {}
}
}
@@ -0,0 +1,339 @@
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/test_entry_limits.py
# Date: 2026-06-13
# Version: 1.0.0
# Category: memory/tests
# =============================================
"""
Tests for the entry_limits config reader (Phase 1 of FPLAN-0270).
Covers:
- Normal config read returns four default entry types.
- per_branch override changes a cap.
- per_branch adds a new entry type.
- Missing config file returns safe defaults (no crash).
- Malformed JSON returns safe defaults + warning logged (no crash).
"""
import importlib
import json
import sys
from pathlib import Path
import pytest
# ---------------------------------------------------------------------------
# Helpers: fresh-import the module under test with mocks already in place
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _fresh_entry_limits(monkeypatch):
"""Drop cached module so each test gets a fresh import.
The conftest _mock_infrastructure replaces
aipass.memory.apps.handlers.json with a MagicMock, which prevents
sub-module discovery. We pop the json package and its children so
importlib can re-import the real modules with the prax mock still in
place.
"""
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)
yield
def _get_module():
"""Import and return the entry_limits module."""
return importlib.import_module("aipass.memory.apps.handlers.json.entry_limits")
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _write_config(tmp_path: Path, data: dict) -> Path:
"""Write a memory.config.json into tmp_path/config/ and return its path."""
config_dir = tmp_path / "config"
config_dir.mkdir(parents=True, exist_ok=True)
config_path = config_dir / "memory.config.json"
config_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
return config_path
def _full_config(**entry_limits_overrides) -> dict:
"""Return a minimal memory.config.json dict with an entry_limits section.
Any keyword args are merged into the entry_limits section.
"""
section = {
"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,
},
},
"per_branch": {},
}
section.update(entry_limits_overrides)
return {"entry_limits": section}
# ===========================================================================
# 1. Normal config returns four default entry types
# ===========================================================================
class TestNormalConfig:
"""Reader returns the four default caps with a normal config."""
def test_returns_four_entry_types(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_path = _write_config(tmp_path, _full_config())
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("some_branch")
assert "entry_types" in result
assert len(result["entry_types"]) == 4
assert set(result["entry_types"].keys()) == {"key_learnings", "sessions", "todos", "observations"}
def test_enabled_is_true(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_path = _write_config(tmp_path, _full_config())
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
assert result["enabled"] is True
def test_enforce_is_false(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_path = _write_config(tmp_path, _full_config())
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
assert result["enforce"] is False
def test_default_max_chars_values(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_path = _write_config(tmp_path, _full_config())
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
types = result["entry_types"]
assert types["key_learnings"]["max_chars"] == 200
assert types["sessions"]["max_chars"] == 300
assert types["todos"]["max_chars"] == 200
assert types["observations"]["max_chars"] == 600
# ===========================================================================
# 2. per_branch override changes a cap
# ===========================================================================
class TestPerBranchOverride:
"""per_branch override changes a cap for the specified branch."""
def test_override_max_chars(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cfg = _full_config(per_branch={"devpulse": {"sessions": {"max_chars": 400}}})
config_path = _write_config(tmp_path, cfg)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("devpulse")
assert result["entry_types"]["sessions"]["max_chars"] == 400
# Other fields on sessions should be preserved from base
assert result["entry_types"]["sessions"]["file"] == "local.json"
assert result["entry_types"]["sessions"]["container"] == "sessions"
def test_override_does_not_affect_other_branches(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cfg = _full_config(per_branch={"devpulse": {"sessions": {"max_chars": 400}}})
config_path = _write_config(tmp_path, cfg)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("memory")
# memory branch should get the default, not devpulse's override
assert result["entry_types"]["sessions"]["max_chars"] == 300
def test_override_does_not_affect_other_types(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cfg = _full_config(per_branch={"devpulse": {"sessions": {"max_chars": 400}}})
config_path = _write_config(tmp_path, cfg)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("devpulse")
# Other types should be unchanged
assert result["entry_types"]["key_learnings"]["max_chars"] == 200
assert result["entry_types"]["observations"]["max_chars"] == 600
# ===========================================================================
# 3. per_branch adds a NEW entry type
# ===========================================================================
class TestPerBranchNewType:
"""per_branch adds a new entry type and the reader includes it."""
def test_new_type_added(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
new_type = {
"file": "local.json",
"container": "custom_notes",
"kind": "list",
"field": "text",
"max_chars": 500,
}
cfg = _full_config(per_branch={"special": {"custom_notes": new_type}})
config_path = _write_config(tmp_path, cfg)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("special")
assert "custom_notes" in result["entry_types"]
assert result["entry_types"]["custom_notes"]["max_chars"] == 500
assert result["entry_types"]["custom_notes"]["container"] == "custom_notes"
# Original four types still present
assert len(result["entry_types"]) == 5
def test_new_type_not_present_for_other_branch(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
new_type = {
"file": "local.json",
"container": "custom_notes",
"kind": "list",
"field": "text",
"max_chars": 500,
}
cfg = _full_config(per_branch={"special": {"custom_notes": new_type}})
config_path = _write_config(tmp_path, cfg)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("other_branch")
assert "custom_notes" not in result["entry_types"]
assert len(result["entry_types"]) == 4
# ===========================================================================
# 4. Missing config file returns safe defaults (no crash)
# ===========================================================================
class TestMissingConfig:
"""Missing config file returns safe defaults without crashing."""
def test_missing_config_returns_defaults(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nonexistent" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
result = mod.load_entry_limits("any_branch")
assert result["enabled"] is True
assert result["enforce"] is False
assert len(result["entry_types"]) == 4
assert result["entry_types"]["sessions"]["max_chars"] == 300
def test_missing_config_logs_warning(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nonexistent" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mock_logger = mod.logger
mod.load_entry_limits("any_branch")
mock_logger.warning.assert_called()
warning_msg = mock_logger.warning.call_args[0][0]
assert "entry_limits" in warning_msg.lower() or "config" in warning_msg.lower()
# ===========================================================================
# 5. Malformed JSON returns safe defaults + warning logged (no crash)
# ===========================================================================
class TestMalformedJson:
"""Malformed JSON returns safe defaults and logs a warning."""
def test_malformed_json_returns_defaults(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "config"
config_dir.mkdir(parents=True, exist_ok=True)
bad_config = config_dir / "memory.config.json"
bad_config.write_text("{this is not valid json!!!", encoding="utf-8")
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", bad_config)
result = mod.load_entry_limits("any_branch")
assert result["enabled"] is True
assert result["enforce"] is False
assert len(result["entry_types"]) == 4
def test_malformed_json_logs_warning(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "config"
config_dir.mkdir(parents=True, exist_ok=True)
bad_config = config_dir / "memory.config.json"
bad_config.write_text("{broken json", encoding="utf-8")
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", bad_config)
mock_logger = mod.logger
mod.load_entry_limits("any_branch")
mock_logger.warning.assert_called()
warning_msg = mock_logger.warning.call_args[0][0]
assert "malformed" in warning_msg.lower() or "json" in warning_msg.lower()
def test_missing_entry_limits_section_returns_defaults(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Config file exists but has no entry_limits section."""
config_path = _write_config(tmp_path, {"rollover": {"defaults": {"max_lines": 500}}})
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any_branch")
assert result["enabled"] is True
assert result["enforce"] is False
assert len(result["entry_types"]) == 4