slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(memory): FPLAN-0271 — relocate config to json-home + unify 9 loaders behind one self-healing config_loader

Browse Source 
Move memory.config.json to memory_json/custom_config/ and .plans_processed.json
to memory_json/ root; delete the loose config/ dir. Replace 9 disagreeing
per-loader config readers with one DEFAULT_CONFIG + non-mutating deep-merge +
self-heal (missing file -> write defaults; malformed JSON -> fail loud, never
overwrite). Resolves 8 default divergences incl. the silent enforce-off bug and
rollover 600-vs-500. Static _meta documents consumer files; dead intake removed.
949 tests green, seedgo @memory 100%. Design: DPLAN-0206.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-13 13:50:42 -07:00
co-authored by Claude Opus 4.8
parent 5336d8f61f
commit a8d05e2602
21 changed files with 1012 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+24
View File
@@ -9,6 +9,30 @@ PyPI version — not the changelog header.
---
## [2026-06-13]
### Changed
- **Memory config relocated to the json-home and unified behind one
self-healing loader (FPLAN-0271).** `memory.config.json` moved from the loose
tracked `config/` dir into the gitignored `memory_json/custom_config/`
(operator-tunable, fast-access) and `.plans_processed.json` into
`memory_json/` root; the empty `config/` dir was removed. The config was
previously read by **9 separate loaders**, each carrying its own *disagreeing*
defaults (8 divergence classes — incl. the headline bug where a missing config
silently flipped `entry_limits.enforce` off, plus rollover defaulting to 600
vs the configured 500). All 9 now read through one
`apps/handlers/json/config_loader.py` with a single `DEFAULT_CONFIG` +
non-mutating deep-merge + self-heal: a missing file is rewritten from code
defaults (warn-first `enforce: false`), while malformed JSON fails loud and is
never overwritten. Dead `intake` section deleted; a static `_meta` block in
`DEFAULT_CONFIG` documents each section's consumer files. Code-as-Template:
the on-disk file is local tuning, code carries the committed defaults — same
model as hooks `cadence_config.json`. Verified: 949 memory tests green, seedgo
@memory 100%, live self-heal / malformed-no-clobber / edit_gate checks pass.
Design: DPLAN-0206. Follow-up parked: issue #643 (codify `custom_config/` as a
seedgo standard).
## [2026-06-12]
### Changed
src/aipass/memory/.seedgo/bypass.json
+30
View File
@@ -101,6 +101,21 @@
"standard": "handlers",
"reason": "Architectural: watcher coordinates tracking, rollover, intake, and archive handlers for auto-rollover pipeline."
},
{
"file": "apps/handlers/intake/plans_processor.py",
"standard": "handlers",
"reason": "Architectural: imports json.config_loader and json_handler for centralized config access and operation logging."
},
{
"file": "apps/handlers/intake/pool_processor.py",
"standard": "handlers",
"reason": "Architectural: imports json.config_loader and json_handler for centralized config access and operation logging."
},
{
"file": "apps/handlers/monitor/detector.py",
"standard": "handlers",
"reason": "Architectural: imports json.config_loader and json_handler for centralized config access and operation logging."
},
{
"file": "apps/handlers/symbolic/retriever.py",
"standard": "handlers",
@@ -675,6 +690,21 @@
"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."
},
{
"file": "tests/test_config_loader.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by design, not in 3-layer apps/ structure."
},
{
"file": "tests/test_config_loader.py",
"standard": "documentation",
"reason": "Test file — test functions don't require docstrings."
},
{
"file": "tests/test_config_loader.py",
"standard": "meta",
"reason": "Test file — META block present at lines 1-7; hook false-positive on test file format."
}
],
"notes": {
src/aipass/memory/README.md
+5 -6
View File
@@ -55,7 +55,7 @@ memory/
│ └── handlers/ # 14 handler groups
│ ├── archive/ # indexer.py
│ ├── intake/ # plans_processor.py, pool_processor.py
│ ├── json/ # json_handler.py, memory_files.py, entry_limits.py, lint_handler.py
│ ├── json/ # json_handler.py, memory_files.py, entry_limits.py, lint_handler.py, config_loader.py
│ ├── learnings/ # manager.py
│ ├── monitor/ # detector.py, memory_watcher.py
│ ├── rollover/ # extractor.py, orchestrator.py
@@ -67,11 +67,10 @@ memory/
│ ├── tracking/ # line_counter.py
│ ├── vector/ # embedder.py, embed_subprocess.py
│ └── central_writer.py
├── config/ # memory.config.json — per-branch rollover limits
├── templates/ # LOCAL.template.json, OBSERVATIONS.template.json
├── tests/ # 839 tests (28 test files)
├── tests/ # 949 tests (31 test files)
├── .chroma/ # ChromaDB vector store
└── memory_json/ # Operation log files (auto-created)
└── memory_json/ # Operation logs + custom_config/memory.config.json
```
### Rollover Pipeline
@@ -116,8 +115,8 @@ All ML operations (fastembed, chromadb) run via subprocess. The main process nev
## Quality
- **Tests:** 839 passed, 0 failures, 0 skips
- **Test files:** 28
- **Tests:** 949 passed, 0 failures, 0 skips
- **Test files:** 31
- **Seedgo:** 100% — maintained since s12
---
src/aipass/memory/apps/handlers/intake/auto_process.py
+2 -10
View File
@@ -23,25 +23,17 @@ HOOK ENGINE CONTRACT:
Returns: dict with success, pool, and rollover results
"""
import json
from pathlib import Path
from typing import Any, Dict
from aipass.prax import logger
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import json_handler, config_loader
_MEMORY_ROOT = Path(__file__).resolve().parent.parent.parent.parent
CONFIG_PATH = _MEMORY_ROOT / "config" / "memory.config.json"
def _load_pool_enabled() -> bool:
try:
with open(CONFIG_PATH, encoding="utf-8") as f:
config = json.load(f)
return config.get("memory_pool", {}).get("enabled", False)
except Exception as e:
logger.warning(f"[auto_process] Failed to load config: {e}")
return False
return config_loader.section("memory_pool").get("enabled", False)
def run_pool_processing() -> Dict[str, Any]:
src/aipass/memory/apps/handlers/intake/plans_processor.py
+4 -9
View File
@@ -28,6 +28,7 @@ from typing import Dict, Any, List
from aipass.prax import logger
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import config_loader
# Subprocess scripts
_HANDLERS_DIR = Path(__file__).resolve().parent.parent
@@ -60,7 +61,7 @@ def _get_memory_python() -> str:
MEMORY_PYTHON = _get_memory_python()
# Track which files have been processed
_PROCESSED_MANIFEST = _MEMORY_ROOT / "config" / ".plans_processed.json"
_PROCESSED_MANIFEST = _MEMORY_ROOT / "memory_json" / ".plans_processed.json"
# Chunk settings
MAX_CHUNK_CHARS = 1500 # ~375 tokens, fits well with all-MiniLM-L6-v2
@@ -230,13 +231,7 @@ def process_plans() -> Dict[str, Any]:
Dict with success, files_processed, total_chunks
"""
# Load config
config_path = _MEMORY_ROOT / "config" / "memory.config.json"
try:
config = json.loads(config_path.read_text(encoding="utf-8"))
plans_config = config.get("plans", {})
except Exception as e:
logger.warning(f"[plans_processor] Config load failed: {e}")
return {"success": False, "error": f"Config load failed: {e}"}
plans_config = config_loader.section("plans")
if not plans_config.get("enabled", False):
return {"success": True, "skipped": True, "reason": "plans disabled"}
@@ -246,7 +241,7 @@ def process_plans() -> Dict[str, Any]:
repo_root = _find_repo_root()
plans_path = Path(plans_dir) if Path(plans_dir).is_absolute() else repo_root / plans_dir
extensions = plans_config.get("supported_extensions", [".md"])
collection_name = plans_config.get("collection_name", "flow_plans")
collection_name = plans_config.get("collection_name", "plans")
if not plans_path.exists():
return {"success": True, "files_processed": 0, "total_chunks": 0, "reason": "plans dir not found"}
src/aipass/memory/apps/handlers/intake/pool_processor.py
+3 -9
View File
@@ -27,10 +27,10 @@ from typing import List, Dict, Any
from aipass.prax import logger
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import config_loader
# Paths
_MEMORY_ROOT = Path(__file__).resolve().parent.parent.parent.parent # handlers/intake/ → handlers/ → apps/ → memory/
CONFIG_PATH = _MEMORY_ROOT / "config" / "memory.config.json"
MEMORY_POOL_PATH = _MEMORY_ROOT / "memory_pool"
CHROMA_PATH = _MEMORY_ROOT / ".chroma"
@@ -103,14 +103,8 @@ def find_source_file(filename: str) -> Path | None:
def load_config() -> dict:
"""Load memory_pool config from memory.config.json"""
try:
with open(CONFIG_PATH) as f:
config = json.load(f)
return config.get("memory_pool", {})
except Exception as e:
logger.warning(f"[pool_processor] Failed to load config: {e}")
return {"enabled": False, "error": str(e)}
"""Load memory_pool config from memory.config.json via config_loader."""
return config_loader.section("memory_pool")
def get_pool_files(extensions: List[str] | None = None) -> List[Path]:
src/aipass/memory/apps/handlers/json/__init__.py
+8 -3
View File
@@ -1,9 +1,10 @@
"""
Memory JSON Handler Package
Provides two sub-modules:
json_handler -- Standard three-JSON logging (read_json, write_json, log_operation)
memory_files -- Memory file safe I/O (read_memory_file, write_memory_file, etc.)
Provides three sub-modules:
json_handler -- Standard three-JSON logging (read_json, write_json, log_operation)
memory_files -- Memory file safe I/O (read_memory_file, write_memory_file, etc.)
config_loader -- Unified config reader for memory.config.json
"""
from .json_handler import (
@@ -21,6 +22,8 @@ from .memory_files import (
validate_memory_file_structure,
)
from . import config_loader
__all__ = [
# json_handler (three-JSON standard)
"log_operation",
@@ -33,4 +36,6 @@ __all__ = [
"read_memory_file_data",
"write_memory_file_simple",
"validate_memory_file_structure",
# config_loader (unified config reader)
"config_loader",
]
src/aipass/memory/apps/handlers/json/config_loader.py
+186
View File
@@ -0,0 +1,186 @@
# =================== AIPass ====================
# Name: config_loader.py
# Description: Unified config loader for memory.config.json
# Version: 1.0.0
# Created: 2026-06-13
# Modified: 2026-06-13
# =============================================
"""
Unified Config Loader
Single entry point for reading memory.config.json. Replaces the 9
ad-hoc readers that previously loaded the file independently, each
with subtly different defaults and error handling.
Provides a canonical DEFAULT_CONFIG, a non-mutating deep_merge, and a
self-healing load() that guarantees callers always receive a usable dict.
Usage:
from aipass.memory.apps.handlers.json.config_loader import load, section
cfg = load()
rollover = section("rollover")
"""
import copy
import json
from pathlib import Path
from typing import Any
from aipass.memory.apps.handlers.json import json_handler
from aipass.prax import logger
_MEMORY_ROOT = Path(__file__).resolve().parents[3]
_CONFIG_PATH = _MEMORY_ROOT / "memory_json" / "custom_config" / "memory.config.json"
DEFAULT_CONFIG: dict[str, Any] = {
"_meta": {
"memory_pool": {
"consumers": ["intake/pool_processor.py", "intake/auto_process.py", "monitor/memory_watcher.py"],
"purpose": "Vectorize files dropped in memory_pool/, archive beyond keep_recent",
},
"rollover": {
"consumers": [
"monitor/detector.py",
"monitor/memory_watcher.py",
"rollover/extractor.py",
"templates/pusher.py",
],
"purpose": "Line/entry thresholds that trigger .trinity rollover",
},
"plans": {
"consumers": ["intake/plans_processor.py", "monitor/memory_watcher.py"],
"purpose": "Vectorize closed plan .md files into ChromaDB",
},
"entry_limits": {
"consumers": ["json/entry_limits.py", "modules/lint.py"],
"purpose": "Per-entry char caps on .trinity writes (warn-first baseline)",
},
},
"memory_pool": {
"enabled": True,
"process_on_startup": False,
"keep_recent": 0,
"supported_extensions": [".md", ".txt"],
"collection_name": "memory_pool_docs",
"chunk_size": 1000,
"chunk_overlap": 100,
"archive_path": "memory_pool_archive",
},
"rollover": {
"defaults": {
"max_lines": 500,
"archive_oldest": 100,
},
"per_branch": {},
},
"plans": {
"enabled": True,
"path": ".backup/processed_plans",
"collection_name": "plans",
"supported_extensions": [".md"],
},
"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": {},
},
}
def deep_merge(base: dict, overrides: dict) -> dict:
"""Recursively merge *overrides* into *base* without mutating either."""
result = copy.deepcopy(base)
for key, val in overrides.items():
if key in result and isinstance(result[key], dict) and isinstance(val, dict):
result[key] = deep_merge(result[key], val)
else:
result[key] = copy.deepcopy(val)
return result
def load(self_heal: bool = True) -> dict[str, Any]:
"""Load memory.config.json, deep-merged over DEFAULT_CONFIG.
Args:
self_heal: If True and the file is missing, create it from defaults.
Returns:
The effective config dict (always safe to use).
"""
if not _CONFIG_PATH.exists():
if self_heal:
_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
_CONFIG_PATH.write_text(json.dumps(DEFAULT_CONFIG, indent=2) + "\n", encoding="utf-8")
logger.info(f"[config_loader] Created default config at {_CONFIG_PATH}")
json_handler.log_operation(
"config_load_self_heal",
{"path": str(_CONFIG_PATH), "action": "created_default"},
module_name="config_loader",
)
return copy.deepcopy(DEFAULT_CONFIG)
logger.warning(f"[config_loader] Config not found at {_CONFIG_PATH}, using defaults")
json_handler.log_operation(
"config_load_missing",
{"path": str(_CONFIG_PATH)},
module_name="config_loader",
)
return copy.deepcopy(DEFAULT_CONFIG)
raw = _CONFIG_PATH.read_text(encoding="utf-8")
try:
file_config = json.loads(raw)
except json.JSONDecodeError as exc:
# Malformed JSON is a red flag — log as error, don't overwrite
logger.error(f"[config_loader] Malformed JSON in {_CONFIG_PATH}: {exc}")
json_handler.log_operation(
"config_load_malformed",
{"path": str(_CONFIG_PATH), "error": str(exc)},
module_name="config_loader",
)
return copy.deepcopy(DEFAULT_CONFIG)
merged = deep_merge(DEFAULT_CONFIG, file_config)
json_handler.log_operation(
"config_load",
{"path": str(_CONFIG_PATH)},
module_name="config_loader",
)
return merged
def section(name: str) -> dict[str, Any]:
"""Return a single top-level section from the config, or empty dict."""
return load().get(name, {})
src/aipass/memory/apps/handlers/json/entry_limits.py
+12 -88
View File
@@ -7,11 +7,11 @@
# =============================================
"""
Entry Limits Config Reader, Validator & Diff Helper
Entry Limits 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
deep-merged over the default entry_types.
Delegates config reading to ``config_loader`` and returns the effective
limits for a given branch, with per_branch overrides deep-merged over
the default entry_types.
Provides ``check_entry()`` — a pure validator that checks whether a
single entry text exceeds its character cap.
@@ -35,53 +35,15 @@ Usage:
"""
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
from aipass.memory.apps.handlers.json import config_loader
# 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(
@@ -114,14 +76,10 @@ def _deep_merge_entry_types(
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.
Delegates config reading to ``config_loader``, pulls the
``entry_limits`` section, then deep-merges any
``per_branch[branch]`` overrides on top of the default
``entry_types``.
Args:
branch: Branch name (e.g. "devpulse", "memory").
@@ -129,40 +87,8 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
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")
cfg = config_loader.load()
section = cfg.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(
@@ -170,14 +96,12 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
{"branch": branch, "fallback": "missing_section"},
module_name="entry_limits",
)
return copy.deepcopy(_SAFE_DEFAULTS)
section = config_loader.DEFAULT_CONFIG["entry_limits"]
# --- 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, {})
src/aipass/memory/apps/handlers/monitor/detector.py
+3 -18
View File
@@ -27,6 +27,7 @@ from dataclasses import dataclass
from aipass.prax.apps.modules.logger import get_system_logger
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import config_loader
logger = get_system_logger()
@@ -170,24 +171,8 @@ def _get_memory_file_path(branch: Dict, memory_type: str) -> Path | None:
def _load_config() -> Dict[str, Any]:
"""
Load memory.config.json
Returns:
Config dict, or empty dict on error
"""
# Look for config relative to this handler's location
config_path = Path(__file__).resolve().parents[3] / "config" / "memory.config.json"
if not config_path.exists():
return {}
try:
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
logger.warning(f"[detector] Failed to load config: {e}")
return {}
"""Load memory.config.json via config_loader."""
return config_loader.load()
# =============================================================================
src/aipass/memory/apps/handlers/monitor/memory_watcher.py
+14 -45
View File
@@ -48,6 +48,7 @@ from aipass.memory.apps.handlers.tracking.line_counter import update_line_count
from aipass.memory.apps.handlers.monitor.detector import check_single_file # noqa: E402
from aipass.prax.apps.modules.logger import get_system_logger # noqa: E402
from aipass.memory.apps.handlers.json import json_handler # noqa: E402
from aipass.memory.apps.handlers.json import config_loader # noqa: E402
logger = get_system_logger()
@@ -103,26 +104,14 @@ def _get_rollover_threshold(branch_name: str, file_path: Path | None = None) ->
logger.warning(f"[memory_watcher] Failed to read file-level threshold from {file_path}: {e}")
# 2. Check per-branch config override
config_path = _MEMORY_ROOT / "config" / "memory.config.json"
try:
with open(config_path) as f:
config = json.load(f)
branch_limits = config.get("rollover", {}).get("per_branch", {}).get(branch_name, {})
if "max_lines" in branch_limits:
return branch_limits["max_lines"]
# 3. Fall back to defaults
default_limit = config.get("rollover", {}).get("defaults", {}).get("max_lines")
if default_limit is not None:
return default_limit
except Exception as e:
logger.warning(f"[memory_watcher] Failed to read rollover config: {e}")
# 4. Final fallback
return 600
cfg = config_loader.load()
branch_limits = cfg.get("rollover", {}).get("per_branch", {}).get(branch_name, {})
if "max_lines" in branch_limits:
return branch_limits["max_lines"]
default_limit = cfg.get("rollover", {}).get("defaults", {}).get("max_lines")
if default_limit is not None:
return default_limit
return 500
def _check_vector_deps() -> bool:
@@ -274,27 +263,16 @@ def _check_memory_pool() -> Dict[str, Any]:
Returns:
Dict with processing status
"""
import json
config_path = _MEMORY_ROOT / "config" / "memory.config.json"
pool_config = config_loader.section("memory_pool")
pool_path = _MEMORY_ROOT / "memory_pool"
# Load config
try:
with open(config_path) as f:
config = json.load(f)
pool_config = config.get("memory_pool", {})
except Exception as exc:
logger.warning(f"[memory_watcher] Could not load memory pool config: {exc}")
return {"success": False, "error": "Could not load config"}
# Check if enabled
if not pool_config.get("enabled", False):
return {"success": True, "skipped": True, "reason": "memory_pool disabled"}
# Count files in pool (excluding .archive)
extensions = pool_config.get("supported_extensions", [".md", ".txt"])
keep_recent = pool_config.get("keep_recent", 10)
keep_recent = pool_config.get("keep_recent", 0)
files = []
for ext in extensions:
@@ -336,23 +314,14 @@ def _check_plans() -> Dict[str, Any]:
"""
import json
config_path = _MEMORY_ROOT / "config" / "memory.config.json"
# Load config
try:
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
plans_config = config.get("plans", {})
except Exception as exc:
logger.warning(f"[memory_watcher] Could not load plans config: {exc}")
return {"success": False, "error": "Could not load config"}
plans_config = config_loader.section("plans")
# Check if enabled
if not plans_config.get("enabled", False):
return {"success": True, "skipped": True, "reason": "plans disabled"}
# Get plans path and count files (supports absolute paths)
plans_dir = plans_config.get("path", "plans")
plans_dir = plans_config.get("path", ".backup/processed_plans")
repo_root = _find_repo_root()
plans_path = Path(plans_dir) if Path(plans_dir).is_absolute() else repo_root / plans_dir
extensions = plans_config.get("supported_extensions", [".md"])
@@ -370,7 +339,7 @@ def _check_plans() -> Dict[str, Any]:
return {"success": True, "pending_files": 0, "action": "count_only"}
# Load manifest to count unprocessed files
manifest_path = _MEMORY_ROOT / "config" / ".plans_processed.json"
manifest_path = _MEMORY_ROOT / "memory_json" / ".plans_processed.json"
manifest: Dict[str, str] = {}
if manifest_path.exists():
try:
src/aipass/memory/apps/handlers/rollover/extractor.py
+3 -2
View File
@@ -32,7 +32,7 @@ from typing import Dict, Any
from datetime import datetime
# Handler imports (relative within package)
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import json_handler, config_loader
from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data, write_memory_file_simple
from aipass.prax.apps.modules.logger import get_system_logger
@@ -385,7 +385,8 @@ def extract_items(file_path: Path, percentage: int | None = None) -> Dict[str, A
return {"success": False, "error": f"No growing array found in {file_path.name}"}
# Get metadata
max_lines = data.get("document_metadata", {}).get("limits", {}).get("max_lines", 600)
_cfg_max = config_loader.section("rollover").get("defaults", {}).get("max_lines", 500)
max_lines = data.get("document_metadata", {}).get("limits", {}).get("max_lines", _cfg_max)
# Check if under limit
if current_lines < max_lines:
src/aipass/memory/apps/handlers/templates/pusher.py
+3 -2
View File
@@ -30,7 +30,7 @@ from typing import Dict, Any, List, Tuple, Optional
from datetime import datetime
from aipass.prax import logger
from aipass.memory.apps.handlers.json import json_handler
from aipass.memory.apps.handlers.json import json_handler, config_loader
# Handler imports (same-branch allowed per handler boundaries)
from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data, write_memory_file_simple
@@ -170,7 +170,8 @@ def _merge_metadata(curr_meta: dict, tmpl_meta: dict) -> List[str]:
tmpl_limits = tmpl_meta.get("limits", {})
curr_limits = curr_meta.get("limits", {})
branch_max_lines = curr_limits.get("max_lines")
tmpl_max_lines = tmpl_limits.get("max_lines", 600)
_cfg_max = config_loader.section("rollover").get("defaults", {}).get("max_lines", 500)
tmpl_max_lines = tmpl_limits.get("max_lines", _cfg_max)
new_limits = copy.deepcopy(tmpl_limits)
if branch_max_lines is not None and branch_max_lines != tmpl_max_lines:
new_limits["max_lines"] = branch_max_lines
src/aipass/memory/config/memory.config.json
-36
View File
@@ -1,36 +0,0 @@
{
"memory_pool": {
"enabled": true,
"process_on_startup": false,
"keep_recent": 0,
"supported_extensions": [".md", ".txt"]
},
"rollover": {
"defaults": {
"max_lines": 500,
"archive_oldest": 100
},
"per_branch": {}
},
"plans": {
"enabled": true,
"path": ".backup/processed_plans",
"collection_name": "plans",
"supported_extensions": [".md"]
},
"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": {}
}
}
src/aipass/memory/config/memory_bank.config.example.json
-18
View File
@@ -1,18 +0,0 @@
{
"memory_pool": {
"enabled": false,
"process_on_startup": false,
"extensions": [".md", ".txt"]
},
"rollover": {
"defaults": {
"max_lines": 500,
"archive_oldest": 100
},
"per_branch": {}
},
"intake": {
"enabled": false,
"pool_dir": "memory_pool"
}
}
src/aipass/memory/tests/test_auto_process.py
+46 -8
View File
@@ -21,8 +21,11 @@ enabled=false respected, rollover-trigger path.
All tests use mocks/tmp_path — no live filesystem or infrastructure access.
"""
import importlib
import importlib.util
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -30,9 +33,36 @@ from unittest.mock import MagicMock, patch
# Import helpers
# ---------------------------------------------------------------------------
_CONFIG_LOADER_PATH = Path(__file__).resolve().parent.parent / "apps" / "handlers" / "json" / "config_loader.py"
def _load_real_config_loader():
"""Load the real config_loader module from disk (bypassing mocked sys.modules)."""
spec = importlib.util.spec_from_file_location(
"aipass.memory.apps.handlers.json.config_loader",
_CONFIG_LOADER_PATH,
)
assert spec is not None, f"Could not find config_loader at {_CONFIG_LOADER_PATH}"
assert spec.loader is not None, "config_loader spec has no loader"
mod = importlib.util.module_from_spec(spec)
sys.modules["aipass.memory.apps.handlers.json.config_loader"] = mod
spec.loader.exec_module(mod)
# Also attach to the (mocked) parent package so `from ... import config_loader` works
parent = sys.modules.get("aipass.memory.apps.handlers.json")
if parent is not None:
setattr(parent, "config_loader", mod)
return mod
def _import_auto_process(monkeypatch):
"""Import auto_process with mocked dependencies."""
"""Import auto_process with mocked dependencies.
Loads the real config_loader (bypassing the conftest MagicMock for the
json package) so that _CONFIG_PATH can be patched per-test.
"""
# Load real config_loader into sys.modules before auto_process imports it
_load_real_config_loader()
sys.modules.pop("aipass.memory.apps.handlers.intake.auto_process", None)
parent = sys.modules.get("aipass.memory.apps.handlers.intake")
if parent is not None and hasattr(parent, "auto_process"):
@@ -65,39 +95,47 @@ class TestLoadPoolEnabled:
def test_returns_true_when_enabled(self, monkeypatch, tmp_path):
mod = _import_auto_process(monkeypatch)
cl = mod.config_loader
config_file = tmp_path / "memory.config.json"
config_file.write_text(
json.dumps({"memory_pool": {"enabled": True}}),
encoding="utf-8",
)
monkeypatch.setattr(mod, "CONFIG_PATH", config_file)
monkeypatch.setattr(cl, "_CONFIG_PATH", config_file)
assert mod._load_pool_enabled() is True
def test_returns_false_when_disabled(self, monkeypatch, tmp_path):
mod = _import_auto_process(monkeypatch)
cl = mod.config_loader
config_file = tmp_path / "memory.config.json"
config_file.write_text(
json.dumps({"memory_pool": {"enabled": False}}),
encoding="utf-8",
)
monkeypatch.setattr(mod, "CONFIG_PATH", config_file)
monkeypatch.setattr(cl, "_CONFIG_PATH", config_file)
assert mod._load_pool_enabled() is False
def test_returns_false_when_config_missing(self, monkeypatch, tmp_path):
def test_returns_true_when_config_missing_self_heals(self, monkeypatch, tmp_path):
"""Missing config triggers self-heal which writes DEFAULT_CONFIG (enabled=True)."""
mod = _import_auto_process(monkeypatch)
monkeypatch.setattr(mod, "CONFIG_PATH", tmp_path / "missing.json")
cl = mod.config_loader
monkeypatch.setattr(cl, "_CONFIG_PATH", tmp_path / "missing.json")
assert mod._load_pool_enabled() is False
# Self-heal writes DEFAULT_CONFIG which has memory_pool.enabled = True
assert mod._load_pool_enabled() is True
def test_returns_false_when_key_missing(self, monkeypatch, tmp_path):
mod = _import_auto_process(monkeypatch)
cl = mod.config_loader
config_file = tmp_path / "memory.config.json"
config_file.write_text(json.dumps({"rollover": {}}), encoding="utf-8")
monkeypatch.setattr(mod, "CONFIG_PATH", config_file)
monkeypatch.setattr(cl, "_CONFIG_PATH", config_file)
assert mod._load_pool_enabled() is False
# Config exists but has no memory_pool key; deep_merge with DEFAULT_CONFIG
# fills it in, so enabled comes from DEFAULT_CONFIG (True)
assert mod._load_pool_enabled() is True
# ===========================================================================
src/aipass/memory/tests/test_config_loader.py
+489
View File
@@ -0,0 +1,489 @@
# =================== AIPass ====================
# Name: test_config_loader.py
# Description: Tests for config_loader handler (FPLAN-0271 Phase 1)
# Version: 1.0.0
# Created: 2026-06-13
# Modified: 2026-06-13
# =============================================
"""
Tests for the config_loader handler (Phase 1 of FPLAN-0271).
Covers:
1. Missing file + self_heal=True -- creates dirs, writes defaults, returns defaults.
2. Missing file + self_heal=False -- no disk write, returns defaults, logs warning.
3. Malformed JSON -- does NOT overwrite, logs ERROR, returns defaults.
4. Partial config -- deep_merge fills missing defaults, preserves file values.
5. Full config -- passthrough of file values.
6. section() -- returns named section or empty dict for unknown.
7. deep_merge() -- nested merge, non-mutation, override precedence.
"""
import copy
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_config_loader(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.config_loader", None)
yield
def _get_module():
"""Import and return the config_loader module."""
return importlib.import_module("aipass.memory.apps.handlers.json.config_loader")
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _write_config(tmp_path: Path, data: dict) -> Path:
"""Write a memory.config.json into tmp_path/custom_config/ and return its path."""
config_dir = tmp_path / "custom_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
# ===========================================================================
# 1. Missing file + self_heal=True -- creates dirs, writes defaults, returns defaults
# ===========================================================================
class TestMissingFileSelfHealTrue:
"""When the config file is missing and self_heal=True, load() should
create parent directories, write DEFAULT_CONFIG to disk, and return defaults.
"""
def test_creates_parent_dirs(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nonexistent" / "deep" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mod.load(self_heal=True)
assert missing_path.parent.exists()
def test_writes_file_to_disk(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)
mod.load(self_heal=True)
assert missing_path.exists()
def test_written_file_matches_default_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "auto_created" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mod.load(self_heal=True)
written = json.loads(missing_path.read_text(encoding="utf-8"))
assert written == mod.DEFAULT_CONFIG
def test_returns_default_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "auto_created" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
result = mod.load(self_heal=True)
assert result == mod.DEFAULT_CONFIG
def test_returned_dict_is_not_same_object_as_default(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "auto_created" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
result = mod.load(self_heal=True)
assert result is not mod.DEFAULT_CONFIG
# ===========================================================================
# 2. Missing file + self_heal=False -- no disk write, returns defaults, logs warning
# ===========================================================================
class TestMissingFileSelfHealFalse:
"""When the config file is missing and self_heal=False, load() should
NOT write to disk, should return defaults, and should log a warning.
"""
def test_does_not_create_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nope" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mod.load(self_heal=False)
assert not missing_path.exists()
def test_does_not_create_parent_dirs(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nope" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mod.load(self_heal=False)
assert not missing_path.parent.exists()
def test_returns_default_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nope" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
result = mod.load(self_heal=False)
assert result == mod.DEFAULT_CONFIG
def test_logs_warning(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing_path = tmp_path / "nope" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mock_logger = mod.logger
mod.load(self_heal=False)
mock_logger.warning.assert_called()
# ===========================================================================
# 3. Malformed JSON -- does NOT overwrite, logs ERROR, returns defaults
# ===========================================================================
class TestMalformedJson:
"""When the config file exists but contains invalid JSON, load() must
NOT overwrite it, must log an ERROR, and must return defaults.
"""
def test_returns_defaults(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "custom_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(self_heal=True)
assert result == mod.DEFAULT_CONFIG
def test_does_not_overwrite_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "custom_config"
config_dir.mkdir(parents=True, exist_ok=True)
bad_config = config_dir / "memory.config.json"
garbage = "{broken json 12345"
bad_config.write_text(garbage, encoding="utf-8")
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", bad_config)
mod.load(self_heal=True)
# File content must be UNCHANGED -- self_heal must NOT overwrite existing files
assert bad_config.read_text(encoding="utf-8") == garbage
def test_logs_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "custom_config"
config_dir.mkdir(parents=True, exist_ok=True)
bad_config = config_dir / "memory.config.json"
bad_config.write_text("not json", encoding="utf-8")
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", bad_config)
mock_logger = mod.logger
mod.load(self_heal=True)
mock_logger.error.assert_called()
# ===========================================================================
# 4. Partial config -- deep_merge fills missing defaults, preserves file values
# ===========================================================================
class TestPartialConfig:
"""When the config file exists with only some sections, deep_merge
fills in missing defaults while preserving file values.
"""
def test_fills_missing_sections(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""File with only entry_limits should get all other sections from defaults."""
partial = {"entry_limits": {"enforce": True}}
config_path = _write_config(tmp_path, partial)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
# memory_pool, rollover, plans should be filled in from defaults
assert "memory_pool" in result
assert "rollover" in result
assert "plans" in result
def test_preserves_file_value_over_default(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""File has enforce: true (default is false) -- merged result must be true."""
partial = {"entry_limits": {"enforce": True}}
config_path = _write_config(tmp_path, partial)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
assert result["entry_limits"]["enforce"] is True
def test_fills_missing_keys_within_section(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Partial entry_limits section should get enabled, entry_types, etc. from defaults."""
partial = {"entry_limits": {"enforce": True}}
config_path = _write_config(tmp_path, partial)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
el = result["entry_limits"]
# enabled should come from default
assert el["enabled"] is True
# entry_types should be filled from default
assert "entry_types" in el
assert "key_learnings" in el["entry_types"]
def test_partial_memory_pool_preserves_file_values(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Partial memory_pool with only enabled=false should preserve that override."""
partial = {"memory_pool": {"enabled": False}}
config_path = _write_config(tmp_path, partial)
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
assert result["memory_pool"]["enabled"] is False
# Other memory_pool keys should be filled from defaults
assert "supported_extensions" in result["memory_pool"]
def test_partial_does_not_mutate_default_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Loading a partial config must not change DEFAULT_CONFIG in-place."""
mod = _get_module()
original_default = copy.deepcopy(mod.DEFAULT_CONFIG)
partial = {"entry_limits": {"enforce": True}}
config_path = _write_config(tmp_path, partial)
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
mod.load()
assert mod.DEFAULT_CONFIG == original_default
# ===========================================================================
# 5. Full config -- passthrough of file values
# ===========================================================================
class TestFullConfig:
"""When the config file contains a complete config, load() should
return the file values as-is (deep_merge should be a no-op).
"""
def test_returns_file_values(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
full = copy.deepcopy(mod.DEFAULT_CONFIG)
# Customize some values to differentiate from defaults
full["memory_pool"]["chunk_size"] = 2000
full["entry_limits"]["enforce"] = True
config_path = _write_config(tmp_path, full)
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
assert result["memory_pool"]["chunk_size"] == 2000
assert result["entry_limits"]["enforce"] is True
def test_full_config_matches_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
full = copy.deepcopy(mod.DEFAULT_CONFIG)
config_path = _write_config(tmp_path, full)
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.load()
assert result == full
# ===========================================================================
# 6. section() -- returns named section or empty dict for unknown
# ===========================================================================
class TestSection:
"""section(name) returns the named section from the loaded config,
or an empty dict for unknown section names.
"""
def test_returns_known_section(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
config_path = _write_config(tmp_path, copy.deepcopy(mod.DEFAULT_CONFIG))
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.section("memory_pool")
assert isinstance(result, dict)
assert "enabled" in result
def test_returns_entry_limits_section(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
config_path = _write_config(tmp_path, copy.deepcopy(mod.DEFAULT_CONFIG))
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.section("entry_limits")
assert "enforce" in result
assert "entry_types" in result
def test_returns_empty_dict_for_unknown_section(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
config_path = _write_config(tmp_path, copy.deepcopy(mod.DEFAULT_CONFIG))
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.section("totally_nonexistent_section")
assert result == {}
def test_section_values_match_loaded_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
mod = _get_module()
full = copy.deepcopy(mod.DEFAULT_CONFIG)
full["rollover"]["defaults"]["max_lines"] = 999
config_path = _write_config(tmp_path, full)
monkeypatch.setattr(mod, "_CONFIG_PATH", config_path)
result = mod.section("rollover")
assert result["defaults"]["max_lines"] == 999
# ===========================================================================
# 7. deep_merge() -- nested merge, non-mutation, override precedence
# ===========================================================================
class TestDeepMerge:
"""deep_merge(base, overrides) performs a recursive non-mutating dict merge."""
def test_overrides_take_precedence(self) -> None:
mod = _get_module()
base = {"a": 1, "b": 2}
overrides = {"b": 99}
result = mod.deep_merge(base, overrides)
assert result["b"] == 99
assert result["a"] == 1
def test_nested_override(self) -> None:
mod = _get_module()
base = {"outer": {"inner": 1, "keep": True}}
overrides = {"outer": {"inner": 42}}
result = mod.deep_merge(base, overrides)
assert result["outer"]["inner"] == 42
assert result["outer"]["keep"] is True
def test_adds_new_keys(self) -> None:
mod = _get_module()
base = {"a": 1}
overrides = {"b": 2}
result = mod.deep_merge(base, overrides)
assert result == {"a": 1, "b": 2}
def test_does_not_mutate_base(self) -> None:
mod = _get_module()
base = {"outer": {"inner": 1}}
base_copy = copy.deepcopy(base)
overrides = {"outer": {"inner": 99}}
mod.deep_merge(base, overrides)
assert base == base_copy
def test_does_not_mutate_overrides(self) -> None:
mod = _get_module()
base = {"a": 1}
overrides = {"a": 2, "b": {"c": 3}}
overrides_copy = copy.deepcopy(overrides)
mod.deep_merge(base, overrides)
assert overrides == overrides_copy
def test_deeply_nested_merge(self) -> None:
mod = _get_module()
base = {"l1": {"l2": {"l3": {"val": "original", "other": True}}}}
overrides = {"l1": {"l2": {"l3": {"val": "changed"}}}}
result = mod.deep_merge(base, overrides)
assert result["l1"]["l2"]["l3"]["val"] == "changed"
assert result["l1"]["l2"]["l3"]["other"] is True
def test_empty_overrides_returns_copy_of_base(self) -> None:
mod = _get_module()
base = {"a": 1, "b": {"c": 2}}
result = mod.deep_merge(base, {})
assert result == base
assert result is not base
def test_empty_base_returns_copy_of_overrides(self) -> None:
mod = _get_module()
overrides = {"a": 1, "b": {"c": 2}}
result = mod.deep_merge({}, overrides)
assert result == overrides
assert result is not overrides
def test_non_dict_override_replaces_dict(self) -> None:
"""When an override value is a non-dict (e.g., list or scalar),
it should replace the base value even if base has a dict there.
"""
mod = _get_module()
base = {"a": {"nested": True}}
overrides = {"a": "flat_string"}
result = mod.deep_merge(base, overrides)
assert result["a"] == "flat_string"
src/aipass/memory/tests/test_detector.py
+10
View File
@@ -40,10 +40,20 @@ def _mock_detector_infrastructure(monkeypatch):
# -- memory json handler ------------------------------------------------
mock_json_handler = MagicMock()
mock_json_handler.log_operation = MagicMock(return_value=True)
# -- config_loader (must return real dicts, not MagicMocks) -------------
mock_config_loader = MagicMock()
mock_config_loader.load.return_value = {
"rollover": {"defaults": {"max_lines": 500}, "per_branch": {}},
}
mock_config_loader.section.side_effect = lambda name: mock_config_loader.load.return_value.get(name, {})
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
json_pkg.config_loader = mock_config_loader
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json", json_pkg)
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.json_handler", mock_json_handler)
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.config_loader", mock_config_loader)
# Force fresh import every test
monkeypatch.delitem(sys.modules, "aipass.memory.apps.handlers.monitor.detector", raising=False)
src/aipass/memory/tests/test_entry_limits.py
+56 -45
View File
@@ -2,7 +2,7 @@
# META DATA HEADER
# Name: tests/test_entry_limits.py
# Date: 2026-06-13
# Version: 1.0.0
# Version: 1.1.0
# Category: memory/tests
# =============================================
@@ -14,7 +14,10 @@ Covers:
- 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).
- Malformed JSON returns safe defaults + error logged (no crash).
Note: entry_limits delegates config reading to config_loader, so tests
patch config_loader._CONFIG_PATH rather than a removed entry_limits attr.
"""
import importlib
@@ -38,16 +41,22 @@ def _fresh_entry_limits(monkeypatch):
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.
config_loader must also be popped so its _CONFIG_PATH can be
re-patched per test.
"""
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.config_loader", 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")
def _get_modules():
"""Import and return (entry_limits, config_loader) modules."""
config_loader = importlib.import_module("aipass.memory.apps.handlers.json.config_loader")
entry_limits = importlib.import_module("aipass.memory.apps.handlers.json.entry_limits")
return entry_limits, config_loader
# ---------------------------------------------------------------------------
@@ -118,8 +127,8 @@ class TestNormalConfig:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("some_branch")
@@ -129,8 +138,8 @@ class TestNormalConfig:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
@@ -138,8 +147,8 @@ class TestNormalConfig:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
@@ -147,8 +156,8 @@ class TestNormalConfig:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any")
types = result["entry_types"]
@@ -170,8 +179,8 @@ class TestPerBranchOverride:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("devpulse")
@@ -183,8 +192,8 @@ class TestPerBranchOverride:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("memory")
@@ -194,8 +203,8 @@ class TestPerBranchOverride:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("devpulse")
@@ -222,8 +231,8 @@ class TestPerBranchNewType:
}
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("special")
@@ -243,8 +252,8 @@ class TestPerBranchNewType:
}
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("other_branch")
@@ -262,8 +271,8 @@ class TestMissingConfig:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", missing_path)
result = mod.load_entry_limits("any_branch")
@@ -272,26 +281,27 @@ class TestMissingConfig:
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:
def test_missing_config_logs_info_on_self_heal(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""config_loader self-heals (creates defaults) when file is missing, logging at INFO level."""
missing_path = tmp_path / "nonexistent" / "memory.config.json"
mod = _get_module()
monkeypatch.setattr(mod, "_CONFIG_PATH", missing_path)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", missing_path)
mock_logger = mod.logger
mock_logger = loader.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()
mock_logger.info.assert_called()
info_msg = mock_logger.info.call_args[0][0]
assert "config" in info_msg.lower()
# ===========================================================================
# 5. Malformed JSON returns safe defaults + warning logged (no crash)
# 5. Malformed JSON returns safe defaults + error logged (no crash)
# ===========================================================================
class TestMalformedJson:
"""Malformed JSON returns safe defaults and logs a warning."""
"""Malformed JSON returns safe defaults and logs an error."""
def test_malformed_json_returns_defaults(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_dir = tmp_path / "config"
@@ -299,8 +309,8 @@ class TestMalformedJson:
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", bad_config)
result = mod.load_entry_limits("any_branch")
@@ -308,29 +318,30 @@ class TestMalformedJson:
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:
def test_malformed_json_logs_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""config_loader logs malformed JSON at ERROR level (not warning)."""
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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", bad_config)
mock_logger = mod.logger
mock_logger = loader.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()
mock_logger.error.assert_called()
error_msg = mock_logger.error.call_args[0][0]
assert "malformed" in error_msg.lower() or "json" in error_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)
mod, loader = _get_modules()
monkeypatch.setattr(loader, "_CONFIG_PATH", config_path)
result = mod.load_entry_limits("any_branch")
src/aipass/memory/tests/test_intake.py
+63 -17
View File
@@ -1,9 +1,9 @@
# ===================AIPASS====================
# META DATA HEADER
# =================== AIPass ====================
# Name: tests/test_intake.py
# Date: 2026-04-03
# Description: Tests for the intake/pool_processor handler
# Version: 1.0.0
# Category: memory/tests
# Created: 2026-04-03
# Modified: 2026-06-13
# =============================================
"""Tests for the intake/pool_processor handler.
@@ -33,7 +33,15 @@ from unittest.mock import MagicMock
def _import_pool_processor(monkeypatch):
"""Import pool_processor with mocked dependencies."""
"""Import pool_processor with mocked dependencies.
Pops the json handler package and its sub-modules from sys.modules so
that the real modules (json_handler, config_loader) can be re-imported
fresh, bypassing the conftest MagicMock replacement.
"""
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.config_loader", None)
sys.modules.pop("aipass.memory.apps.handlers.intake.pool_processor", None)
parent = sys.modules.get("aipass.memory.apps.handlers.intake")
if parent is not None and hasattr(parent, "pool_processor"):
@@ -53,6 +61,7 @@ class TestFindSourceFile:
"""Test find_source_file function."""
def test_found_in_active_pool(self, monkeypatch, tmp_path):
"""Test finding a file in the active memory pool directory."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -65,6 +74,7 @@ class TestFindSourceFile:
assert result == target
def test_found_in_archive(self, monkeypatch, tmp_path):
"""Test finding a file in the archive subdirectory of the pool."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
archive = pool / ".archive"
@@ -78,6 +88,7 @@ class TestFindSourceFile:
assert result == target
def test_not_found_returns_none(self, monkeypatch, tmp_path):
"""Test that nonexistent files return None."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -88,6 +99,7 @@ class TestFindSourceFile:
assert result is None
def test_prefers_active_over_archive(self, monkeypatch, tmp_path):
"""Test that active pool files are preferred over archive copies."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
archive = pool / ".archive"
@@ -112,13 +124,15 @@ class TestLoadConfig:
"""Test load_config function."""
def test_loads_valid_config(self, monkeypatch, tmp_path):
"""Test loading and parsing a valid memory.config.json file."""
mod = _import_pool_processor(monkeypatch)
cl = mod.config_loader
config_file = tmp_path / "memory.config.json"
config_file.write_text(
json.dumps({"memory_pool": {"enabled": True, "keep_recent": 5, "collection_name": "test_pool"}}),
encoding="utf-8",
)
monkeypatch.setattr(mod, "CONFIG_PATH", config_file)
monkeypatch.setattr(cl, "_CONFIG_PATH", config_file)
result = mod.load_config()
@@ -126,24 +140,29 @@ class TestLoadConfig:
assert result["keep_recent"] == 5
assert result["collection_name"] == "test_pool"
def test_returns_disabled_when_file_missing(self, monkeypatch, tmp_path):
def test_returns_defaults_when_file_missing(self, monkeypatch, tmp_path):
"""Missing config triggers self-heal; returns DEFAULT_CONFIG memory_pool."""
mod = _import_pool_processor(monkeypatch)
monkeypatch.setattr(mod, "CONFIG_PATH", tmp_path / "missing.json")
cl = mod.config_loader
monkeypatch.setattr(cl, "_CONFIG_PATH", tmp_path / "missing.json")
result = mod.load_config()
assert result["enabled"] is False
assert "error" in result
# Self-heal writes DEFAULT_CONFIG which has memory_pool.enabled = True
assert result["enabled"] is True
def test_returns_empty_when_no_memory_pool_key(self, monkeypatch, tmp_path):
def test_returns_defaults_when_no_memory_pool_key(self, monkeypatch, tmp_path):
"""Config without memory_pool key still returns defaults via deep_merge."""
mod = _import_pool_processor(monkeypatch)
cl = mod.config_loader
config_file = tmp_path / "memory.config.json"
config_file.write_text(json.dumps({"rollover": {}}), encoding="utf-8")
monkeypatch.setattr(mod, "CONFIG_PATH", config_file)
monkeypatch.setattr(cl, "_CONFIG_PATH", config_file)
result = mod.load_config()
assert result == {}
# deep_merge fills in memory_pool from DEFAULT_CONFIG
assert result["enabled"] is True
# ===========================================================================
@@ -155,6 +174,7 @@ class TestGetPoolFiles:
"""Test get_pool_files function."""
def test_returns_empty_when_no_directory(self, monkeypatch, tmp_path):
"""Test that missing pool directory returns empty list."""
mod = _import_pool_processor(monkeypatch)
monkeypatch.setattr(mod, "MEMORY_POOL_PATH", tmp_path / "nonexistent")
@@ -163,6 +183,7 @@ class TestGetPoolFiles:
assert result == []
def test_returns_empty_when_no_matching_files(self, monkeypatch, tmp_path):
"""Test that directory with no matching extensions returns empty list."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -174,6 +195,7 @@ class TestGetPoolFiles:
assert result == []
def test_returns_sorted_by_mtime_newest_first(self, monkeypatch, tmp_path):
"""Test that files are sorted by modification time, newest first."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -197,6 +219,7 @@ class TestGetPoolFiles:
assert result[1].name == "old.md"
def test_filters_by_custom_extensions(self, monkeypatch, tmp_path):
"""Test filtering files by custom extension list."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -220,6 +243,7 @@ class TestReadFileContent:
"""Test read_file_content function."""
def test_reads_successfully(self, monkeypatch, tmp_path):
"""Test successfully reading file content with metadata."""
mod = _import_pool_processor(monkeypatch)
test_file = tmp_path / "test.md"
test_file.write_text("Hello, world!", encoding="utf-8")
@@ -233,6 +257,7 @@ class TestReadFileContent:
assert result["metadata"]["size"] > 0
def test_returns_failure_for_missing_file(self, monkeypatch, tmp_path):
"""Test that reading a missing file returns failure status."""
mod = _import_pool_processor(monkeypatch)
missing = tmp_path / "nonexistent.md"
@@ -251,6 +276,7 @@ class TestChunkContent:
"""Test chunk_content function."""
def test_short_text_single_chunk(self, monkeypatch):
"""Test that text shorter than chunk_size produces a single chunk."""
mod = _import_pool_processor(monkeypatch)
result = mod.chunk_content("Short text.", chunk_size=1000)
@@ -260,6 +286,7 @@ class TestChunkContent:
assert result[0]["chunk_index"] == 0
def test_long_text_multiple_chunks(self, monkeypatch):
"""Test that long text is split into multiple chunks."""
mod = _import_pool_processor(monkeypatch)
# Create text longer than chunk_size
content = "word " * 300 # ~1500 chars
@@ -272,6 +299,7 @@ class TestChunkContent:
assert indices == list(range(len(result)))
def test_chunk_indices_are_sequential(self, monkeypatch):
"""Test that chunk indices are sequential starting from zero."""
mod = _import_pool_processor(monkeypatch)
content = "A" * 2500
@@ -281,6 +309,7 @@ class TestChunkContent:
assert chunk["chunk_index"] == i
def test_paragraph_break_splitting(self, monkeypatch):
"""Test that paragraph breaks (double newlines) trigger chunk splits."""
mod = _import_pool_processor(monkeypatch)
# Build content with a paragraph break in the right spot
# chunk_size=100, so we need content > 100 chars
@@ -295,6 +324,7 @@ class TestChunkContent:
assert len(result) >= 2
def test_empty_content_returns_single_chunk(self, monkeypatch):
"""Test that empty content returns a single empty chunk."""
mod = _import_pool_processor(monkeypatch)
result = mod.chunk_content("", chunk_size=1000)
@@ -304,6 +334,7 @@ class TestChunkContent:
assert result[0]["text"] == ""
def test_exact_chunk_size_single_chunk(self, monkeypatch):
"""Test that content exactly matching chunk_size produces one chunk."""
mod = _import_pool_processor(monkeypatch)
content = "X" * 100
@@ -322,6 +353,7 @@ class TestProcessFileToVectors:
"""Test process_file_to_vectors with mocked chromadb."""
def test_processes_file_with_mocked_chromadb(self, monkeypatch, tmp_path):
"""Test processing a file into vectors with mocked chromadb."""
mod = _import_pool_processor(monkeypatch)
monkeypatch.setattr(mod, "CHROMA_PATH", tmp_path / ".chroma")
@@ -353,6 +385,7 @@ class TestProcessFileToVectors:
mock_collection.upsert.assert_called_once()
def test_returns_failure_when_file_unreadable(self, monkeypatch, tmp_path):
"""Test that unreadable files return failure status."""
mod = _import_pool_processor(monkeypatch)
missing = tmp_path / "nonexistent.md"
@@ -361,6 +394,7 @@ class TestProcessFileToVectors:
assert result["success"] is False
def test_returns_failure_when_chromadb_import_fails(self, monkeypatch, tmp_path):
"""Test that chromadb import failures return failure status."""
mod = _import_pool_processor(monkeypatch)
test_file = tmp_path / "test.md"
@@ -373,12 +407,13 @@ class TestProcessFileToVectors:
# Patch the builtins __import__ to raise for chromadb
original_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__
def fake_import(name, *args, **kwargs):
def _fake_import(name, *args, **kwargs):
"""Intercept imports to simulate missing chromadb."""
if name == "chromadb":
raise ImportError("chromadb not installed")
return original_import(name, *args, **kwargs)
monkeypatch.setattr("builtins.__import__", fake_import)
monkeypatch.setattr("builtins.__import__", _fake_import)
result = mod.process_file_to_vectors(test_file, "test_collection")
@@ -395,6 +430,7 @@ class TestArchiveOldFiles:
"""Test archive_old_files function."""
def test_no_archiving_when_under_limit(self, monkeypatch, tmp_path):
"""Test that files under keep_recent limit are not archived."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -413,6 +449,7 @@ class TestArchiveOldFiles:
assert result["kept_count"] == 2
def test_moves_old_files_to_archive(self, monkeypatch, tmp_path):
"""Test that old files beyond keep_recent are moved to archive."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -442,6 +479,7 @@ class TestArchiveOldFiles:
assert len(archived_files) == 2
def test_handles_duplicate_names_in_archive(self, monkeypatch, tmp_path):
"""Test that duplicate filenames in archive are handled with timestamps."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -482,6 +520,7 @@ class TestProcessMemoryPool:
"""Test process_memory_pool main entry point."""
def test_returns_error_when_disabled(self, monkeypatch, tmp_path):
"""Test that disabled memory pool returns error."""
mod = _import_pool_processor(monkeypatch)
monkeypatch.setattr(mod, "MEMORY_POOL_PATH", tmp_path / "pool")
monkeypatch.setattr(mod, "load_config", lambda: {"enabled": False})
@@ -492,6 +531,7 @@ class TestProcessMemoryPool:
assert "disabled" in result["error"]
def test_returns_success_with_no_files(self, monkeypatch, tmp_path):
"""Test that empty memory pool returns success with zero files processed."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "pool"
monkeypatch.setattr(mod, "MEMORY_POOL_PATH", pool)
@@ -516,6 +556,7 @@ class TestProcessMemoryPool:
assert result["files_processed"] == 0
def test_processes_files_and_archives(self, monkeypatch, tmp_path):
"""Test processing files and archiving with full workflow."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "pool"
pool.mkdir(parents=True)
@@ -561,6 +602,7 @@ class TestProcessMemoryPool:
mock_jh.log_operation.assert_called_once()
def test_reports_errors_and_notifies(self, monkeypatch, tmp_path):
"""Test that errors in processing are reported and notification sent."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "pool"
pool.mkdir(parents=True)
@@ -615,6 +657,7 @@ class TestGetPoolStatus:
"""Test get_pool_status function."""
def test_returns_status_with_mocked_chromadb(self, monkeypatch, tmp_path):
"""Test returning pool status with mocked chromadb backend."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -654,6 +697,7 @@ class TestGetPoolStatus:
assert result["oldest_file"] == "recent.md"
def test_returns_zero_vectors_when_chromadb_fails(self, monkeypatch, tmp_path):
"""Test that chromadb import failures return zero vectors gracefully."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
@@ -665,12 +709,13 @@ class TestGetPoolStatus:
# Make chromadb import raise
original_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__
def fake_import(name, *args, **kwargs):
def _fake_import(name, *args, **kwargs):
"""Intercept imports to simulate missing chromadb."""
if name == "chromadb":
raise ImportError("no chromadb")
return original_import(name, *args, **kwargs)
monkeypatch.setattr("builtins.__import__", fake_import)
monkeypatch.setattr("builtins.__import__", _fake_import)
result = mod.get_pool_status()
@@ -680,6 +725,7 @@ class TestGetPoolStatus:
assert result["oldest_file"] is None
def test_returns_zero_vectors_when_collection_not_found(self, monkeypatch, tmp_path):
"""Test that missing collection returns zero vectors."""
mod = _import_pool_processor(monkeypatch)
pool = tmp_path / "memory_pool"
pool.mkdir()
src/aipass/memory/tests/test_plans_processor.py
+51 -72
View File
@@ -393,28 +393,27 @@ class TestGetMemoryPython:
class TestProcessPlans:
"""Test process_plans main entry point."""
def _setup_config(self, tmp_path, config_data):
"""Write a memory.config.json 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(config_data), encoding="utf-8")
return config_path
def _mock_config(self, monkeypatch, mod, plans_config):
"""Mock config_loader.section on the plans_processor module."""
mock_cl = MagicMock()
mock_cl.section.return_value = plans_config
monkeypatch.setattr(mod, "config_loader", mock_cl)
def test_process_plans_config_load_fails(self, monkeypatch, tmp_path):
def test_process_plans_defaults_no_plans_dir(self, monkeypatch, tmp_path):
"""With default config (self-healed), plans dir absent → success + 0 files."""
mod = _import_plans_processor(monkeypatch)
# Point _MEMORY_ROOT to tmp_path -- no config file exists
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
self._mock_config(monkeypatch, mod, {"enabled": True, "path": ".backup/processed_plans"})
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
result = mod.process_plans()
assert result["success"] is False
assert "Config load failed" in result["error"]
assert result["success"] is True
assert result["files_processed"] == 0
assert "not found" in result.get("reason", "")
def test_process_plans_disabled(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
self._setup_config(tmp_path, {"plans": {"enabled": False}})
self._mock_config(monkeypatch, mod, {"enabled": False})
result = mod.process_plans()
@@ -424,12 +423,7 @@ class TestProcessPlans:
def test_process_plans_dir_not_found(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": "nonexistent/plans"}},
)
# _find_repo_root will return tmp_path
self._mock_config(monkeypatch, mod, {"enabled": True, "path": "nonexistent/plans"})
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
result = mod.process_plans()
@@ -440,12 +434,12 @@ class TestProcessPlans:
def test_process_plans_no_files(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
@@ -456,18 +450,17 @@ class TestProcessPlans:
def test_process_plans_all_already_processed(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
plan_file = plans_dir / "done.md"
plan_file.write_text("Already processed plan content that is long enough.", encoding="utf-8")
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
# Pre-populate the manifest
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text(json.dumps({"done.md": "2026-01-01T00:00:00"}), encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
@@ -479,9 +472,7 @@ class TestProcessPlans:
def test_process_plans_success(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
# Create plans directory with a file
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
plan_file = plans_dir / "new_plan.md"
@@ -491,31 +482,27 @@ class TestProcessPlans:
encoding="utf-8",
)
self._setup_config(
tmp_path,
self._mock_config(
monkeypatch,
mod,
{
"plans": {
"enabled": True,
"path": str(plans_dir),
"supported_extensions": [".md"],
"collection_name": "test_plans",
}
"enabled": True,
"path": str(plans_dir),
"supported_extensions": [".md"],
"collection_name": "test_plans",
},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
# Empty manifest
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
# Mock _embed_texts to return success
monkeypatch.setattr(
mod,
"_embed_texts",
lambda texts: {"success": True, "embeddings": [[0.1, 0.2]] * len(texts)},
)
# Mock _store_vectors to return success
monkeypatch.setattr(
mod,
"_store_vectors",
@@ -532,13 +519,11 @@ class TestProcessPlans:
assert result["total_chunks"] >= 2
mock_jh.log_operation.assert_called_once()
# Manifest should be updated
updated_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
assert "new_plan.md" in updated_manifest
def test_process_plans_embed_fails(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
@@ -548,17 +533,17 @@ class TestProcessPlans:
encoding="utf-8",
)
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
# Mock _embed_texts to return failure
monkeypatch.setattr(
mod,
"_embed_texts",
@@ -576,7 +561,6 @@ class TestProcessPlans:
def test_process_plans_no_embeddings(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
@@ -586,17 +570,17 @@ class TestProcessPlans:
encoding="utf-8",
)
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
# Embed succeeds but returns empty embeddings
monkeypatch.setattr(
mod,
"_embed_texts",
@@ -613,7 +597,6 @@ class TestProcessPlans:
def test_process_plans_store_fails(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
@@ -623,23 +606,22 @@ class TestProcessPlans:
encoding="utf-8",
)
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
# Embed succeeds
monkeypatch.setattr(
mod,
"_embed_texts",
lambda texts: {"success": True, "embeddings": [[0.1, 0.2]] * len(texts)},
)
# Store fails
monkeypatch.setattr(
mod,
"_store_vectors",
@@ -657,21 +639,20 @@ class TestProcessPlans:
def test_process_plans_empty_chunks(self, monkeypatch, tmp_path):
mod = _import_plans_processor(monkeypatch)
monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path)
plans_dir = tmp_path / "plans"
plans_dir.mkdir()
plan_file = plans_dir / "tiny.md"
# Content that will produce zero chunks (under 30 chars, no headers)
plan_file.write_text("Hi.", encoding="utf-8")
self._setup_config(
tmp_path,
{"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}},
self._mock_config(
monkeypatch,
mod,
{"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]},
)
monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path)
manifest_path = tmp_path / "config" / ".plans_processed.json"
manifest_path = tmp_path / ".plans_processed.json"
manifest_path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path)
@@ -680,10 +661,8 @@ class TestProcessPlans:
result = mod.process_plans()
# No chunks produced, but no errors either -- files_without_chunks path
assert result["success"] is True
assert result["files_processed"] == 0
# File should still be marked in manifest (files_without_chunks)
updated_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
assert "tiny.md" in updated_manifest