diff --git a/src/aipass/memory/.seedgo/bypass.json b/src/aipass/memory/.seedgo/bypass.json index 677925fb..64f90f87 100644 --- a/src/aipass/memory/.seedgo/bypass.json +++ b/src/aipass/memory/.seedgo/bypass.json @@ -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": { diff --git a/src/aipass/memory/apps/handlers/json/entry_limits.py b/src/aipass/memory/apps/handlers/json/entry_limits.py new file mode 100644 index 00000000..088b86f5 --- /dev/null +++ b/src/aipass/memory/apps/handlers/json/entry_limits.py @@ -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 diff --git a/src/aipass/memory/config/memory.config.json b/src/aipass/memory/config/memory.config.json index 22026773..5554c22b 100644 --- a/src/aipass/memory/config/memory.config.json +++ b/src/aipass/memory/config/memory.config.json @@ -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": {} } } diff --git a/src/aipass/memory/tests/test_entry_limits.py b/src/aipass/memory/tests/test_entry_limits.py new file mode 100644 index 00000000..2ebeb594 --- /dev/null +++ b/src/aipass/memory/tests/test_entry_limits.py @@ -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