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

feat(memory): FPLAN-0276 — unify memory limits into config single-source + cleanup

Browse Source 
- memory.config.json is the single source of truth: char caps (entry_limits, global) + rollover counts (per_branch, materialized from registry); .trinity files stripped to a one-line _usage header
- changed_entries gate now matches by content identity, not array position — prepending a new entry never re-flags unchanged legacy entries (old=old, new=new; no trimming required on a cap change)
- rollover push command + top-level 'drone @memory push' alias; corrected config _note + --help (rollover push surfaced, labeled destructive system-wide reset)
- removed 3 dead functions: seed_per_branch, its orphaned write_config, add_learning + its tests
- spawn birthright/builder + LOCAL/OBSERVATIONS templates aligned to the stripped shape
- 966 tests, seedgo 100%

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-14 20:54:34 -07:00
co-authored by Claude Opus 4.8
parent 6c675e9b78
commit 72659eccbd
21 changed files with 743 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/memory/apps/handlers/json/config_loader.py
+96 -22
View File
@@ -40,6 +40,14 @@ DEFAULT_CONFIG: dict[str, Any] = {
"consumers": ["intake/pool_processor.py", "intake/auto_process.py", "monitor/memory_watcher.py"],
"purpose": "Vectorize files dropped in memory_pool/, archive beyond keep_recent",
},
"entry_limits": {
"consumers": ["json/entry_limits.py", "modules/lint.py"],
"purpose": "Per-entry char caps on .trinity writes (warn-first baseline)",
},
"plans": {
"consumers": ["intake/plans_processor.py", "monitor/memory_watcher.py"],
"purpose": "Vectorize closed plan .md files into ChromaDB",
},
"rollover": {
"consumers": [
"monitor/detector.py",
@@ -47,15 +55,7 @@ DEFAULT_CONFIG: dict[str, Any] = {
"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)",
"purpose": "Entry-count thresholds that trigger .trinity rollover",
},
},
"memory_pool": {
@@ -68,19 +68,6 @@ DEFAULT_CONFIG: dict[str, Any] = {
"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,
@@ -116,6 +103,27 @@ DEFAULT_CONFIG: dict[str, Any] = {
},
"per_branch": {},
},
"plans": {
"enabled": True,
"path": ".backup/processed_plans",
"collection_name": "plans",
"supported_extensions": [".md"],
},
"rollover": {
"defaults": {
"local": {
"sessions": {"count": 20},
"key_learnings": {"count": 25},
"todos": {"count": 10},
},
"observations": {
"observations": {"count": 25},
},
"_note": "DEFAULTS — edit then `drone @memory rollover push` to apply system-wide."
" Char caps live in entry_limits.",
},
"per_branch": {},
},
}
@@ -184,3 +192,69 @@ def load(self_heal: bool = True) -> dict[str, Any]:
def section(name: str) -> dict[str, Any]:
"""Return a single top-level section from the config, or empty dict."""
return load().get(name, {})
def _find_repo_root() -> Path:
"""Walk up from this file to find repo root (contains AIPASS_REGISTRY.json)."""
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
if (parent / "AIPASS_REGISTRY.json").exists():
return parent
return Path.cwd()
def materialize_per_branch() -> dict[str, Any]:
"""Build per_branch from AIPASS_REGISTRY.json, seeded from rollover.defaults."""
repo_root = _find_repo_root()
registry_path = repo_root / "AIPASS_REGISTRY.json"
if not registry_path.exists():
logger.warning("[config_loader] AIPASS_REGISTRY.json not found")
return {}
try:
registry = json.loads(registry_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
logger.warning(f"[config_loader] Failed to load registry: {e}")
return {}
cfg = load()
defaults = cfg.get("rollover", {}).get("defaults", {})
limits_only = {k: v for k, v in defaults.items() if k != "_note"}
branches = registry.get("branches", [])
active = [b for b in branches if b.get("status") == "active"]
per_branch: dict[str, Any] = {}
for branch in active:
name = branch.get("name", "").lower()
if not name:
continue
entry = copy.deepcopy(limits_only)
entry["_note"] = f"Limits for @{name}. Manual edits persist until next push."
per_branch[name] = entry
return per_branch
def push_defaults_to_per_branch() -> dict[str, Any]:
"""Overwrite every per_branch entry with defaults (full replacement, not merge).
Returns:
Dict with branch count and the new per_branch data.
"""
per_branch = materialize_per_branch()
if not per_branch:
return {"success": False, "error": "No branches found in registry"}
current: dict = {}
if _CONFIG_PATH.exists():
try:
current = json.loads(_CONFIG_PATH.read_text(encoding="utf-8"))
except json.JSONDecodeError:
logger.warning("[config_loader] Malformed config on disk, starting fresh")
current.setdefault("rollover", {})["per_branch"] = per_branch
_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
_CONFIG_PATH.write_text(json.dumps(current, indent=2) + "\n", encoding="utf-8")
return {"success": True, "branches": len(per_branch), "per_branch": per_branch}
src/aipass/memory/apps/handlers/json/entry_limits.py
+8 -5
View File
@@ -87,13 +87,15 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
Returns:
Dict with keys: enabled, enforce, entry_types.
"""
branch_key = branch.lower()
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(
"load_entry_limits",
{"branch": branch, "fallback": "missing_section"},
{"branch": branch_key, "fallback": "missing_section"},
module_name="entry_limits",
)
section = config_loader.DEFAULT_CONFIG["entry_limits"]
@@ -103,7 +105,7 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
base_types = section.get("entry_types", {})
per_branch = section.get("per_branch", {})
branch_overrides = per_branch.get(branch, {})
branch_overrides = per_branch.get(branch_key, {})
if branch_overrides:
effective_types = _deep_merge_entry_types(base_types, branch_overrides)
@@ -118,7 +120,7 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
json_handler.log_operation(
"load_entry_limits",
{"branch": branch, "types_count": len(effective_types)},
{"branch": branch_key, "types_count": len(effective_types)},
module_name="entry_limits",
)
@@ -275,12 +277,13 @@ def _check_list_container(
if not isinstance(after_container, list):
return []
before_list = before_container if isinstance(before_container, list) else []
before_texts = {_extract_text(item, field) for item in before_list}
hits: list[dict[str, Any]] = []
for idx, after_item in enumerate(after_container):
after_text = _extract_text(after_item, field)
if idx < len(before_list) and after_text == _extract_text(before_list[idx], field):
continue # Unchanged — skip even if over-limit
if after_text in before_texts:
continue # Already on disk — not new/changed; skip even if over cap
verdict = check_entry(type_name, after_text, limits)
if not verdict["ok"]:
hits.append(
src/aipass/memory/apps/handlers/learnings/manager.py
-79
View File
@@ -753,85 +753,6 @@ def enforce_limit_completed(file_path: Path) -> Dict[str, Any]:
}
def add_learning(file_path: Path, key: str, value: str) -> Dict[str, Any]:
"""
Add or update a key_learning entry.
Automatically adds timestamp and enforces limit after adding.
Args:
file_path: Path to .local.json file
key: Learning key (snake_case recommended)
value: Learning value (without timestamp - will be added)
Returns:
Dict with add status
"""
if not file_path.exists():
return {"success": False, "error": f"File not found: {file_path}"}
try:
data = read_memory_file_data(file_path)
if data is None:
return {"success": False, "error": f"Failed to parse file: {file_path.name}"}
except Exception as e:
logger.warning(f"[learnings_manager] Failed to read file: {e}")
return {"success": False, "error": f"Failed to read file: {e}"}
learnings = _get_learnings(data)
if isinstance(learnings, list):
# v3 list: find existing entry by key, or insert new at front
is_update = False
for entry in learnings:
if isinstance(entry, dict) and entry.get("key") == key:
entry["value"] = value
entry["date"] = datetime.now().strftime("%Y-%m-%d")
is_update = True
break
if not is_update:
max_num = max((e.get("number", 0) for e in learnings if isinstance(e, dict)), default=0)
learnings.insert(
0,
{
"number": max_num + 1,
"date": datetime.now().strftime("%Y-%m-%d"),
"key": key,
"value": value,
},
)
timestamped_value = value
_set_learnings(data, learnings)
else:
if not learnings:
learnings = {}
timestamped_value = add_timestamp(value)
is_update = key in learnings
learnings[key] = timestamped_value
_set_learnings(data, learnings)
try:
write_memory_file_simple(file_path, data)
except Exception as e:
logger.warning(f"[learnings_manager] Failed to write file: {e}")
return {"success": False, "error": f"Failed to write file: {e}"}
# Enforce limit after adding
enforce_result = enforce_limit(file_path)
json_handler.log_operation(
"add_learning", {"key": key, "action": "updated" if is_update else "added", "success": True}
)
return {
"success": True,
"action": "updated" if is_update else "added",
"key": key,
"value": timestamped_value,
"limit_enforced": enforce_result.get("removed", 0) > 0,
}
# =============================================================================
# STATUS COUNT UPDATES
# =============================================================================
src/aipass/memory/apps/handlers/monitor/detector.py
+95 -14
View File
@@ -21,6 +21,7 @@ Independence:
"""
import json
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any
from dataclasses import dataclass
@@ -198,6 +199,60 @@ def _count_file_lines(file_path: Path) -> int:
return 0
_TEMPLATES_DIR = Path(__file__).resolve().parents[3] / "templates"
_TEMPLATE_MAP = {
"local": _TEMPLATES_DIR / "LOCAL.template.json",
"observations": _TEMPLATES_DIR / "OBSERVATIONS.template.json",
}
def _recreate_trinity_file(branch_path: Path, branch_name: str, memory_type: str) -> Path | None:
"""Recreate a missing .trinity file from canonical template."""
template_path = _TEMPLATE_MAP.get(memory_type)
if not template_path or not template_path.exists():
logger.warning(f"[detector] No template for {memory_type}")
return None
try:
template = json.loads(template_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
logger.warning(f"[detector] Failed to read template {template_path}: {e}")
return None
today = datetime.now().strftime("%Y-%m-%d")
upper_name = branch_name.upper()
def _walk(val):
if isinstance(val, str):
return val.replace("{{BRANCHNAME}}", upper_name).replace("{{DATE}}", today)
if isinstance(val, list):
return [_walk(item) for item in val]
if isinstance(val, dict):
return {k: _walk(v) for k, v in val.items()}
return val
data = _walk(template)
trinity_dir = branch_path / ".trinity"
trinity_dir.mkdir(parents=True, exist_ok=True)
file_path = trinity_dir / f"{memory_type}.json"
try:
file_path.write_text(
json.dumps(data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
logger.info(f"[detector] Recreated {file_path}")
json_handler.log_operation(
"recreate_trinity_file",
{"branch": branch_name, "type": memory_type, "path": str(file_path)},
)
return file_path
except OSError as e:
logger.warning(f"[detector] Failed to write {file_path}: {e}")
return None
def _get_max_lines(file_path: Path, branch_name: str | None = None) -> int:
"""
Get max_lines limit with priority: file metadata > branch config > default
@@ -223,9 +278,13 @@ def _get_max_lines(file_path: Path, branch_name: str | None = None) -> int:
# 2. Try branch-level config (if branch_name provided or can be extracted)
if branch_name is None:
# Extract from filename (e.g., SEEDGO.local.json -> SEEDGO)
parts = file_path.stem.split(".")
branch_name = parts[0] if parts else None
# Derive from directory: .trinity/local.json → parent of .trinity is the branch
if file_path.parent.name == ".trinity":
branch_name = file_path.parents[1].name.lower()
else:
# Legacy flat files: SEEDGO.local.json → "SEEDGO"
parts = file_path.stem.split(".")
branch_name = parts[0].lower() if parts else None
if branch_name:
config = _load_config()
@@ -273,24 +332,37 @@ def _should_rollover(file_path: Path) -> tuple[bool, int, int, str, str]:
metadata = data.get("document_metadata", {})
limits = metadata.get("limits", {})
# v2: entry-count based limits (checked when v2 limit keys are present, regardless of schema_version)
v2_limit_keys = {"max_sessions", "max_key_learnings", "max_observations"}
if v2_limit_keys & set(limits.keys()):
# v2: entry-count based limits — read from config per_branch, not file metadata
# Derive branch name from file path: .trinity/local.json → parent of .trinity
if file_path.parent.name == ".trinity":
branch_name = file_path.parents[1].name.lower()
else:
branch_name = file_path.stem.split(".")[0].lower()
# Determine file type from filename
file_type = file_path.stem.split(".")[0] if file_path.parent.name == ".trinity" else file_path.stem.split(".")[-1]
# .trinity/local.json → "local"; .trinity/observations.json → "observations"
cfg = config_loader.section("rollover")
branch_limits = cfg.get("per_branch", {}).get(branch_name, {})
file_limits = branch_limits.get(file_type, {})
if file_limits:
reasons = []
max_sessions = limits.get("max_sessions")
max_sessions = file_limits.get("sessions", {}).get("count")
if max_sessions is not None:
sessions = data.get("sessions", [])
if isinstance(sessions, list) and len(sessions) >= max_sessions:
reasons.append(f"{len(sessions)}/{max_sessions} sessions")
max_key_learnings = limits.get("max_key_learnings")
max_key_learnings = file_limits.get("key_learnings", {}).get("count")
if max_key_learnings is not None:
key_learnings = data.get("key_learnings", [])
if isinstance(key_learnings, (list, dict)) and len(key_learnings) >= max_key_learnings:
reasons.append(f"{len(key_learnings)}/{max_key_learnings} key_learnings")
max_observations = limits.get("max_observations")
max_observations = file_limits.get("observations", {}).get("count")
if max_observations is not None:
observations = data.get("observations", [])
if isinstance(observations, list) and len(observations) >= max_observations:
@@ -333,7 +405,11 @@ def check_all_branches() -> Dict[str, Any]:
file_path = _get_memory_file_path(branch, memory_type)
if file_path is None:
continue # File doesn't exist, skip
branch_path = Path(branch.get("path", ""))
if branch_path.exists():
file_path = _recreate_trinity_file(branch_path, branch_name, memory_type)
if file_path is None:
continue
should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)
@@ -377,10 +453,15 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)
if should_trigger:
# Extract branch and type from filename (e.g., SEEDGO.observations.json)
parts = file_path.stem.split(".")
branch_name = parts[0] if len(parts) > 0 else "UNKNOWN"
memory_type = parts[1] if len(parts) > 1 else "unknown"
# Extract branch and type from file path
if file_path.parent.name == ".trinity":
branch_name = file_path.parents[1].name
memory_type = file_path.stem # "local" or "observations"
else:
# Legacy flat files: SEEDGO.observations.json
parts = file_path.stem.split(".")
branch_name = parts[0] if len(parts) > 0 else "UNKNOWN"
memory_type = parts[1] if len(parts) > 1 else "unknown"
trigger = RolloverTrigger(
branch=branch_name,
src/aipass/memory/apps/handlers/monitor/memory_watcher.py
-48
View File
@@ -66,54 +66,6 @@ _observer: Any = None
_startup_check_done = False
def _get_rollover_threshold(branch_name: str, file_path: Path | None = None) -> int:
"""
Get rollover threshold for a memory file (line-based, v1 only).
For v2 files (schema_version >= 2.0.0), returns a very large number so
line-based checks never trigger. v2 rollover is handled by the detector
using entry-count limits.
Priority: file metadata > per_branch config > defaults > hardcoded 600
Args:
branch_name: Branch name (uppercase, e.g., 'DEVPULSE')
file_path: Optional path to memory file (checks file-level limits first)
Returns:
Max lines threshold for rollover
"""
import json
# 1. Check file-level metadata first (highest priority)
if file_path is not None:
try:
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
metadata = data.get("document_metadata", {})
# v2 files use entry-count limits — return -1 so caller uses detector
schema_version = metadata.get("schema_version", "1.0.0")
if schema_version.startswith("2"):
return -1
file_limit = metadata.get("limits", {}).get("max_lines")
if file_limit is not None:
return file_limit
except Exception as e:
logger.warning(f"[memory_watcher] Failed to read file-level threshold from {file_path}: {e}")
# 2. Check per-branch config override
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:
"""
Check whether the memory venv has chromadb and numpy available.
src/aipass/memory/apps/handlers/rollover/extractor.py
+25 -9
View File
@@ -271,13 +271,23 @@ def _extract_items_v2(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]:
Returns:
Dict with extracted items and metadata
"""
limits = data.get("document_metadata", {}).get("limits", {})
old_lines = _count_file_lines(file_path)
# Read limits from config per_branch instead of file metadata
if file_path.parent.name == ".trinity":
branch_key = file_path.parents[1].name.lower()
file_type = file_path.stem # "local" or "observations"
else:
branch_key = file_path.parent.name.lower()
file_type = file_path.stem.split(".")[-1]
cfg = config_loader.section("rollover")
file_limits = cfg.get("per_branch", {}).get(branch_key, {}).get(file_type, {})
all_extracted = []
# Extract from sessions array (newest first, oldest at end)
max_sessions = limits.get("max_sessions")
max_sessions = file_limits.get("sessions", {}).get("count")
if max_sessions is not None:
sessions = data.get("sessions", [])
if isinstance(sessions, list) and len(sessions) >= max_sessions:
@@ -287,7 +297,7 @@ def _extract_items_v2(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]:
all_extracted.extend(extracted_sessions)
# Extract from key_learnings list (sorted newest-first; oldest at end)
max_key_learnings = limits.get("max_key_learnings")
max_key_learnings = file_limits.get("key_learnings", {}).get("count")
if max_key_learnings is not None:
key_learnings = data.get("key_learnings", [])
if isinstance(key_learnings, list) and len(key_learnings) >= max_key_learnings:
@@ -297,7 +307,7 @@ def _extract_items_v2(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]:
all_extracted.extend(extracted_kl)
# Extract from observations array (if v2 observations file)
max_observations = limits.get("max_observations")
max_observations = file_limits.get("observations", {}).get("count")
if max_observations is not None:
observations = data.get("observations", [])
if isinstance(observations, list) and len(observations) >= max_observations:
@@ -370,10 +380,16 @@ def extract_items(file_path: Path, percentage: int | None = None) -> Dict[str, A
logger.warning(f"[extractor] Failed to read file {file_path}: {e}")
return {"success": False, "error": f"Failed to read file: {e}"}
# v2: entry-count based extraction (when v2 limit keys are present, regardless of schema_version)
ext_limits = data.get("document_metadata", {}).get("limits", {})
v2_limit_keys = {"max_sessions", "max_key_learnings", "max_observations"}
if v2_limit_keys & set(ext_limits.keys()):
# v2: entry-count based extraction — triggered when config has per_branch counts
if file_path.parent.name == ".trinity":
_ext_branch = file_path.parents[1].name.lower()
_ext_ftype = file_path.stem
else:
_ext_branch = file_path.parent.name.lower()
_ext_ftype = file_path.stem.split(".")[-1]
_ext_cfg = config_loader.section("rollover")
_ext_file_limits = _ext_cfg.get("per_branch", {}).get(_ext_branch, {}).get(_ext_ftype, {})
if _ext_file_limits:
return _extract_items_v2(file_path, data)
# v1: line-count based extraction
@@ -383,7 +399,7 @@ 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
_cfg_max = config_loader.section("rollover").get("defaults", {}).get("max_lines", 500)
_cfg_max = config_loader.section("rollover").get("defaults", {}).get("max_lines", 600)
max_lines = data.get("document_metadata", {}).get("limits", {}).get("max_lines", _cfg_max)
# Check if under limit
src/aipass/memory/apps/handlers/templates/pusher.py
+6 -14
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, config_loader
from aipass.memory.apps.handlers.json import json_handler
# 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
@@ -67,7 +67,7 @@ VERSION_FILE_PATH = TEMPLATES_DIR / ".template_version.json"
# Deprecated sections to REMOVE during push
DEPRECATED_METADATA_KEYS = ["allowed_emojis"]
DEPRECATED_LIMIT_KEYS = ["max_word_count", "max_token_count"]
DEPRECATED_LIMIT_KEYS = ["max_word_count", "max_token_count", "max_lines", "archive_oldest"]
DEPRECATED_STATUS_KEYS = ["auto_compress_at"]
DEPRECATED_NOTES_KEYS = ["formatting_reference", "slash_command_tracking"]
DEPRECATED_GUIDELINES_KEYS = ["emoji_usage", "high_value_patterns", "low_value_patterns"]
@@ -166,18 +166,10 @@ def _merge_metadata(curr_meta: dict, tmpl_meta: dict) -> List[str]:
curr_meta["tags"] = tmpl_tags
changes.append("document_metadata.tags: updated to template tags")
# Limits (preserve per-branch max_lines override)
tmpl_limits = tmpl_meta.get("limits", {})
curr_limits = curr_meta.get("limits", {})
branch_max_lines = curr_limits.get("max_lines")
_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
if curr_limits != new_limits:
curr_meta["limits"] = new_limits
changes.append("document_metadata.limits: updated from template")
# Limits live in memory.config.json now — strip from files if still present
if "limits" in curr_meta:
del curr_meta["limits"]
changes.append("document_metadata.limits: removed (lives in memory.config.json)")
# Status (add missing fields, preserve current values)
curr_status = curr_meta.setdefault("status", {})
src/aipass/memory/apps/memory.py
+5 -1
View File
@@ -115,6 +115,7 @@ def print_help():
table.add_row("rollover status", "Show rollover statistics")
table.add_row("rollover check", "Dry run — check what needs rollover")
table.add_row("rollover sync-lines", "Update line count metadata")
table.add_row("rollover push", "⚠ Reset ALL per_branch limits to defaults (system-wide)")
table.add_row("search <query>", "Semantic search across all branch memories")
table.add_row("symbolic <subcommand>", "Symbolic/fragmented memory extraction and search")
table.add_row("templates <subcommand>", "Living template push, diff, and status")
@@ -168,7 +169,7 @@ def print_help():
console.print()
console.print(
"Commands: search, rollover [run|status|check|sync-lines], lint,"
"Commands: search, rollover [run|status|check|sync-lines|push], lint,"
" pool [process|status], symbolic, templates, verify, watch"
)
console.print()
@@ -234,6 +235,9 @@ def route_command(command: str, args: List[str], modules: List[Any]) -> bool:
start_watch()
return True
if command == "push":
return route_command("rollover", ["push"], modules)
for module in modules:
try:
if module.handle_command(command, args):
src/aipass/memory/apps/modules/rollover.py
+36 -1
View File
@@ -52,6 +52,7 @@ _SUBCOMMANDS = {
"status": "Show rollover statistics for all branches",
"check": "Check which files need rollover (dry run)",
"sync-lines": "Update line count metadata for all branches",
"push": "Overwrite all per_branch limits to defaults (system-wide reset)",
}
@@ -112,6 +113,10 @@ def handle_command(command: str, args: List[str]) -> bool:
sync_line_counts()
return True
if sub == "push":
push_defaults()
return True
# Unknown subcommand
error(
f"Unknown subcommand: '{sub}'",
@@ -158,6 +163,7 @@ def print_help() -> None:
console.print(" [cyan]status[/cyan] Show rollover statistics for all branches")
console.print(" [cyan]check[/cyan] Check which files need rollover (dry run)")
console.print(" [cyan]sync-lines[/cyan] Update line count metadata for all branches")
console.print(" [cyan]push[/cyan] ⚠ Reset ALL per_branch limits to defaults (system-wide)")
console.print(" [cyan]help[/cyan] Show this help message")
console.print()
console.print("[bold]LIMITS:[/bold]")
@@ -213,7 +219,8 @@ def run_rollover() -> bool:
local_status = "> local" if item.get("local_stored") else "x local"
console.print(
f" [green]>[/green] Rolled over {item['memories_count']} items -> {item['global_collection']} "
f"({item['old_lines']} -> {item['new_lines']} lines, global: {item['global_total']} vectors, {local_status})"
f"({item['old_lines']} -> {item['new_lines']} lines, "
f"global: {item['global_total']} vectors, {local_status})"
)
# Report results
@@ -322,6 +329,34 @@ def sync_line_counts() -> None:
console.print()
# =============================================================================
# PUSH DEFAULTS
# =============================================================================
def push_defaults() -> None:
"""Overwrite every per_branch entry in memory.config.json with defaults."""
from ..handlers.json import config_loader
console.print()
console.print(Panel.fit("[bold cyan]Memory - Push Defaults[/bold cyan]", border_style="cyan", box=box.ROUNDED))
console.print()
console.print("[cyan]Overwriting all per_branch limits with defaults...[/cyan]")
console.print()
result = config_loader.push_defaults_to_per_branch()
if not result.get("success"):
error(result.get("error", "Unknown error"))
return
count = result.get("branches", 0)
console.print(f"[green]>[/green] Pushed defaults to {count} branches")
console.print()
json_handler.log_operation("push_defaults", {"branches": count})
# =============================================================================
# STATUS & CHECKING
# =============================================================================
src/aipass/memory/templates/LOCAL.template.json
+1 -9
View File
@@ -12,15 +12,7 @@
"work_log",
"{{BRANCHNAME}}"
],
"limits": {
"max_sessions": 20,
"max_key_learnings": 25,
"max_todos": 10,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"todo_text_max_chars": 200,
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries. todos[] is OPERATIONAL — completing/removing items is normal, not data destruction."
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}"
src/aipass/memory/templates/OBSERVATIONS.template.json
+1 -4
View File
@@ -12,10 +12,7 @@
"patterns",
"{{BRANCHNAME}}"
],
"limits": {
"max_observations": 25,
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_observations exceeded."
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}"
src/aipass/memory/tests/test_changed_entries.py
+46 -1
View File
@@ -269,7 +269,52 @@ class TestListContainer:
# ===========================================================================
# 7. changed_entries: empty before (new file) — all entries treated as new
# 7. changed_entries: list prepend identity-match (Fix 1 — FPLAN-0276 cleanup)
# ===========================================================================
class TestListPrependIdentityMatch:
"""Prepending a new entry must NOT re-flag shifted legacy over-cap entries."""
def test_prepend_with_legacy_overcap_entries_allowed(self) -> None:
"""Full container of over-cap legacy entries + one new in-cap prepend → no violations."""
mod = _get_entry_limits()
legacy = [{"session_number": i, "summary": "s" * 400} for i in range(5, 0, -1)]
before = {"sessions": legacy}
new_entry = {"session_number": 6, "summary": "short new"}
after = {"sessions": [new_entry] + legacy}
result = mod.changed_entries(before, after, _SESSIONS_ONLY)
assert result == []
def test_edited_existing_entry_text_still_caught(self) -> None:
"""Changing an existing entry's text to over-cap is still flagged."""
mod = _get_entry_limits()
before = {"sessions": [{"session_number": 1, "summary": "short"}]}
after = {"sessions": [{"session_number": 1, "summary": "s" * 400}]}
result = mod.changed_entries(before, after, _SESSIONS_ONLY)
assert len(result) == 1
assert result[0]["over_by"] == 100
def test_genuinely_new_overcap_entry_still_caught(self) -> None:
"""A brand-new over-cap entry is still flagged even alongside legacy."""
mod = _get_entry_limits()
legacy = [{"session_number": 1, "summary": "ok"}]
before = {"sessions": legacy}
new_fat = {"session_number": 2, "summary": "s" * 400}
after = {"sessions": [new_fat] + legacy}
result = mod.changed_entries(before, after, _SESSIONS_ONLY)
assert len(result) == 1
assert result[0]["over_by"] == 100
# ===========================================================================
# 8. changed_entries: empty before (new file) — all entries treated as new
# ===========================================================================
src/aipass/memory/tests/test_detector.py
+226 -68
View File
@@ -1,24 +1,28 @@
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/test_detector.py
# Date: 2026-03-24
# Version: 1.0.0
# Category: memory/tests
# =================== AIPass ====================
# Name: test_detector.py
# Description: Tests for rollover trigger detection handler
# Version: 1.1.0
# Created: 2026-03-24
# Modified: 2026-06-14
# =============================================
"""Tests for the rollover trigger detection handler (apps/handlers/monitor/detector.py).
"""Tests for the rollover trigger detection module (apps/handlers/monitor/detector).
Uses tmp_path for all file operations. Creates real temp files with JSON content
rather than mocking open(). The detector functions are imported inside each test
rather than mocking open(). The detector module is imported inside each test
to ensure the autouse conftest fixture for json_handler is already applied.
"""
import json
import logging
import sys
import pytest
from pathlib import Path
from unittest.mock import MagicMock
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Autouse fixture -- mock heavy infrastructure before detector is imported
@@ -44,7 +48,7 @@ def _mock_detector_infrastructure(monkeypatch):
# -- 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": {}},
"rollover": {"defaults": {}, "per_branch": {}},
}
mock_config_loader.section.side_effect = lambda name: mock_config_loader.load.return_value.get(name, {})
@@ -55,8 +59,13 @@ def _mock_detector_infrastructure(monkeypatch):
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
# Force fresh import every test — must also clean the parent package's
# cached attribute, otherwise Python reuses a stale detector module
# that holds an unconfigured config_loader reference.
monkeypatch.delitem(sys.modules, "aipass.memory.apps.handlers.monitor.detector", raising=False)
parent = sys.modules.get("aipass.memory.apps.handlers.monitor")
if parent is not None and hasattr(parent, "detector"):
monkeypatch.delattr(parent, "detector", raising=False)
# ===========================================================================
@@ -68,58 +77,63 @@ class TestGetMemoryFilePath:
"""Tests for _get_memory_file_path(branch, memory_type)."""
def test_returns_path_when_file_exists(self, tmp_path: Path):
"""Existing .trinity file should resolve to a valid Path."""
trinity_dir = tmp_path / ".trinity"
trinity_dir.mkdir()
obs_file = trinity_dir / "observations.json"
obs_file.write_text("{}", encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path
from aipass.memory.apps.handlers.monitor import detector
branch = {"path": str(tmp_path)}
result = _get_memory_file_path(branch, "observations")
result = detector._get_memory_file_path(branch, "observations")
assert result is not None
assert result == obs_file
def test_returns_none_when_file_missing(self, tmp_path: Path):
"""Missing memory file should return None."""
trinity_dir = tmp_path / ".trinity"
trinity_dir.mkdir()
# No observations.json created
from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path
from aipass.memory.apps.handlers.monitor import detector
branch = {"path": str(tmp_path)}
result = _get_memory_file_path(branch, "observations")
result = detector._get_memory_file_path(branch, "observations")
assert result is None
def test_returns_none_when_branch_path_missing(self, tmp_path: Path):
from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path
"""Nonexistent branch path should return None."""
from aipass.memory.apps.handlers.monitor import detector
nonexistent = tmp_path / "does_not_exist"
branch = {"path": str(nonexistent)}
result = _get_memory_file_path(branch, "local")
result = detector._get_memory_file_path(branch, "local")
assert result is None
def test_returns_none_when_path_key_empty(self, tmp_path: Path):
from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path
"""Empty path key in branch dict should return None."""
from aipass.memory.apps.handlers.monitor import detector
branch: dict[str, str] = {"path": ""}
result = _get_memory_file_path(branch, "local")
result = detector._get_memory_file_path(branch, "local")
assert result is None
def test_local_memory_type(self, tmp_path: Path):
"""Local memory type should resolve to local.json in .trinity dir."""
trinity_dir = tmp_path / ".trinity"
trinity_dir.mkdir()
local_file = trinity_dir / "local.json"
local_file.write_text("{}", encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path
from aipass.memory.apps.handlers.monitor import detector
branch = {"path": str(tmp_path)}
result = _get_memory_file_path(branch, "local")
result = detector._get_memory_file_path(branch, "local")
assert result is not None
assert result.name == "local.json"
@@ -134,6 +148,7 @@ class TestLoadConfig:
"""Tests for _load_config()."""
def test_returns_config_dict_when_file_exists(self, tmp_path: Path, monkeypatch):
"""Valid config file should return parsed dict with rollover settings."""
config_dir = tmp_path / "config"
config_dir.mkdir()
config_file = config_dir / "memory.config.json"
@@ -160,6 +175,7 @@ class TestLoadConfig:
assert result["rollover"]["defaults"]["max_lines"] == 500
def test_returns_empty_dict_when_file_missing(self, monkeypatch):
"""Missing config file should return empty dict."""
from aipass.memory.apps.handlers.monitor import detector
# Point config resolution at a path that does not exist
@@ -174,6 +190,7 @@ class TestLoadConfig:
assert result == {}
def test_returns_empty_dict_on_invalid_json(self, tmp_path: Path, monkeypatch):
"""Malformed JSON config should return empty dict and log the error."""
config_dir = tmp_path / "config"
config_dir.mkdir()
config_file = config_dir / "memory.config.json"
@@ -186,7 +203,8 @@ class TestLoadConfig:
try:
with open(config_file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
except Exception as exc:
logger.warning("Failed to parse config file %s: %s", config_file, exc)
return {}
monkeypatch.setattr(detector, "_load_config", _broken_load)
@@ -217,9 +235,9 @@ class TestCheckSingleFile:
content = json.dumps(data, indent=2)
mem_file.write_text(content, encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
@@ -239,9 +257,9 @@ class TestCheckSingleFile:
content = json.dumps(data, indent=2)
mem_file.write_text(content, encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is True
@@ -251,66 +269,60 @@ class TestCheckSingleFile:
"""check_single_file on a nonexistent path returns success=False."""
missing = tmp_path / "ghost.json"
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(missing)
result = detector.check_single_file(missing)
assert result["success"] is False
assert "error" in result
def test_v2_schema_entry_count_trigger(self, tmp_path: Path):
"""v2 schema triggers on entry counts, not line counts."""
def test_v2_schema_entry_count_trigger(self, tmp_path: Path, monkeypatch):
"""v2 schema triggers on entry counts from config per_branch."""
mem_file = tmp_path / "DRONE.local.json"
data = {
"document_metadata": {
"schema_version": "2.0.0",
"limits": {"max_sessions": 3},
},
"sessions": [
{"id": "s1"},
{"id": "s2"},
{"id": "s3"},
{"id": "s4"},
],
"document_metadata": {"schema_version": "2.0.0"},
"sessions": [{"id": "s1"}, {"id": "s2"}, {"id": "s3"}, {"id": "s4"}],
}
content = json.dumps(data, indent=2)
mem_file.write_text(content, encoding="utf-8")
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"drone": {"local": {"sessions": {"count": 3}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is True
def test_v2_schema_under_limit_no_trigger(self, tmp_path: Path):
def test_v2_schema_under_limit_no_trigger(self, tmp_path: Path, monkeypatch):
"""v2 schema with entries under the limit should not trigger."""
mem_file = tmp_path / "FLOW.local.json"
data = {
"document_metadata": {
"schema_version": "2.0.0",
"limits": {"max_sessions": 10},
},
"document_metadata": {"schema_version": "2.0.0"},
"sessions": [{"id": "s1"}, {"id": "s2"}],
}
content = json.dumps(data, indent=2)
mem_file.write_text(content, encoding="utf-8")
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"flow": {"local": {"sessions": {"count": 10}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
def test_v2_list_key_learnings_triggers_rollover(self, tmp_path: Path):
"""List-shaped key_learnings at/over max_key_learnings triggers v2 rollover."""
def test_v2_list_key_learnings_triggers_rollover(self, tmp_path: Path, monkeypatch):
"""List-shaped key_learnings at/over count triggers v2 rollover."""
mem_file = tmp_path / "DEVPULSE.local.json"
data = {
"document_metadata": {
"schema_version": "3.0.0",
"limits": {"max_key_learnings": 3},
},
"document_metadata": {"schema_version": "3.0.0"},
"key_learnings": [
{"number": 3, "date": "2026-06-13", "key": "c", "value": "vc"},
{"number": 2, "date": "2026-06-12", "key": "b", "value": "vb"},
@@ -319,22 +331,24 @@ class TestCheckSingleFile:
}
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"devpulse": {"local": {"key_learnings": {"count": 3}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is True
assert "3/3 key_learnings" in result["trigger"].v2_reason
def test_v2_list_key_learnings_under_limit_no_trigger(self, tmp_path: Path):
"""List-shaped key_learnings under limit does not trigger."""
def test_v2_list_key_learnings_under_limit_no_trigger(self, tmp_path: Path, monkeypatch):
"""List-shaped key_learnings under count does not trigger."""
mem_file = tmp_path / "DRONE.local.json"
data = {
"document_metadata": {
"schema_version": "3.0.0",
"limits": {"max_key_learnings": 10},
},
"document_metadata": {"schema_version": "3.0.0"},
"key_learnings": [
{"number": 2, "date": "2026-06-13", "key": "b", "value": "vb"},
{"number": 1, "date": "2026-06-12", "key": "a", "value": "va"},
@@ -342,9 +356,17 @@ class TestCheckSingleFile:
}
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor.detector import check_single_file
from aipass.memory.apps.handlers.monitor import detector
result = check_single_file(mem_file)
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {
"per_branch": {"drone": {"local": {"key_learnings": {"count": 10}}}},
"defaults": {},
},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
@@ -359,6 +381,7 @@ class TestReadRegistry:
"""Tests for _read_registry()."""
def test_valid_registry_returns_branches(self, tmp_path: Path, monkeypatch):
"""Valid registry JSON should return all branches with absolute paths."""
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_data = {
"branches": [
@@ -382,6 +405,7 @@ class TestReadRegistry:
assert Path(result[0]["path"]).is_absolute()
def test_missing_registry_returns_empty(self, tmp_path: Path, monkeypatch):
"""Missing registry file should return empty list."""
from aipass.memory.apps.handlers.monitor import detector
# Point _REPO_ROOT at a directory with no registry file
@@ -393,6 +417,7 @@ class TestReadRegistry:
assert result == []
def test_invalid_json_returns_empty(self, tmp_path: Path, monkeypatch):
"""Malformed registry JSON should return empty list."""
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_file.write_text("NOT JSON {{{", encoding="utf-8")
@@ -406,6 +431,7 @@ class TestReadRegistry:
assert result == []
def test_registry_resolves_relative_paths(self, tmp_path: Path, monkeypatch):
"""Relative paths in registry should be resolved to absolute."""
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_data = {
"branches": [
@@ -426,6 +452,7 @@ class TestReadRegistry:
assert str(resolved_path) == str(tmp_path / "src/aipass/cli")
def test_empty_branches_list(self, tmp_path: Path, monkeypatch):
"""Registry with empty branches list should return empty list."""
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_data = {"branches": []}
registry_file.write_text(json.dumps(registry_data), encoding="utf-8")
@@ -438,3 +465,134 @@ class TestReadRegistry:
result = detector._read_registry()
assert result == []
# ===========================================================================
# Self-healing recreation (_recreate_trinity_file)
# ===========================================================================
class TestRecreateTrinityFile:
"""Tests for _recreate_trinity_file — P4 self-healing."""
def test_recreates_missing_local_file(self, tmp_path: Path, monkeypatch):
"""Missing local.json should be recreated from template with _usage and no limits."""
from aipass.memory.apps.handlers.monitor import detector
templates_dir = tmp_path / "templates"
templates_dir.mkdir()
template = {
"document_metadata": {
"document_type": "session_history",
"document_name": "{{BRANCHNAME}}.LOCAL",
"_usage": "Automated file.",
"status": {"health": "healthy"},
},
"sessions": [],
"key_learnings": [],
}
(templates_dir / "LOCAL.template.json").write_text(json.dumps(template), encoding="utf-8")
monkeypatch.setattr(detector, "_TEMPLATES_DIR", templates_dir)
monkeypatch.setattr(
detector,
"_TEMPLATE_MAP",
{"local": templates_dir / "LOCAL.template.json"},
)
branch_dir = tmp_path / "testbranch"
branch_dir.mkdir()
result = detector._recreate_trinity_file(branch_dir, "testbranch", "local")
assert result is not None
assert result.exists()
data = json.loads(result.read_text(encoding="utf-8"))
assert data["document_metadata"]["document_name"] == "TESTBRANCH.LOCAL"
assert "_usage" in data["document_metadata"]
assert "limits" not in data["document_metadata"]
def test_recreates_missing_observations_file(self, tmp_path: Path, monkeypatch):
"""Missing observations.json should be recreated from template."""
from aipass.memory.apps.handlers.monitor import detector
templates_dir = tmp_path / "templates"
templates_dir.mkdir()
template = {
"document_metadata": {
"document_type": "collaboration_patterns",
"document_name": "{{BRANCHNAME}}.OBSERVATIONS",
"_usage": "Automated file.",
},
"observations": [],
}
(templates_dir / "OBSERVATIONS.template.json").write_text(json.dumps(template), encoding="utf-8")
monkeypatch.setattr(detector, "_TEMPLATES_DIR", templates_dir)
monkeypatch.setattr(
detector,
"_TEMPLATE_MAP",
{"observations": templates_dir / "OBSERVATIONS.template.json"},
)
branch_dir = tmp_path / "api"
branch_dir.mkdir()
result = detector._recreate_trinity_file(branch_dir, "api", "observations")
assert result is not None
data = json.loads(result.read_text(encoding="utf-8"))
assert data["document_metadata"]["document_name"] == "API.OBSERVATIONS"
assert "limits" not in data["document_metadata"]
def test_check_all_branches_recreates_missing(self, tmp_path: Path, monkeypatch):
"""check_all_branches should auto-recreate missing .trinity files."""
from aipass.memory.apps.handlers.monitor import detector
branch_dir = tmp_path / "mybranch"
trinity_dir = branch_dir / ".trinity"
trinity_dir.mkdir(parents=True)
# Only create observations, NOT local — local should be recreated
obs_data = {
"document_metadata": {"schema_version": "1.0.0", "limits": {"max_lines": 600}},
"observations": [],
}
(trinity_dir / "observations.json").write_text(json.dumps(obs_data, indent=2), encoding="utf-8")
templates_dir = tmp_path / "templates"
templates_dir.mkdir()
local_template = {
"document_metadata": {
"document_type": "session_history",
"document_name": "{{BRANCHNAME}}.LOCAL",
"_usage": "Automated file.",
"status": {"health": "healthy"},
},
"sessions": [],
}
(templates_dir / "LOCAL.template.json").write_text(json.dumps(local_template), encoding="utf-8")
monkeypatch.setattr(detector, "_TEMPLATES_DIR", templates_dir)
monkeypatch.setattr(
detector,
"_TEMPLATE_MAP",
{
"local": templates_dir / "LOCAL.template.json",
"observations": templates_dir / "OBSERVATIONS.template.json",
},
)
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_data = {
"branches": [
{"name": "mybranch", "path": str(branch_dir), "status": "active"},
]
}
registry_file.write_text(json.dumps(registry_data), encoding="utf-8")
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
detector.check_all_branches()
recreated = trinity_dir / "local.json"
assert recreated.exists()
data = json.loads(recreated.read_text(encoding="utf-8"))
assert data["document_metadata"]["document_name"] == "MYBRANCH.LOCAL"
assert "limits" not in data["document_metadata"]
src/aipass/memory/tests/test_handlers.py
+69 -29
View File
@@ -36,12 +36,17 @@ def _import_extractor(monkeypatch):
mock_memory_files.read_memory_file_data = MagicMock(return_value=None)
mock_memory_files.write_memory_file_simple = MagicMock()
mock_config_loader = MagicMock()
mock_config_loader.section.return_value = {"defaults": {}, "per_branch": {}}
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.memory_files", mock_memory_files)
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.config_loader", mock_config_loader)
sys.modules.pop("aipass.memory.apps.handlers.rollover.extractor", None)
parent = sys.modules.get("aipass.memory.apps.handlers.rollover")
@@ -53,6 +58,7 @@ def _import_extractor(monkeypatch):
return extractor, {
"json_handler": mock_json_handler,
"memory_files": mock_memory_files,
"config_loader": mock_config_loader,
}
@@ -194,14 +200,19 @@ class TestExtractItemsV2:
}
def test_trims_sessions_to_limit(self, monkeypatch, tmp_path):
ext = _import_extractor(monkeypatch)[0]
ext, mocks = _import_extractor(monkeypatch)
data = self._make_v2_data(num_sessions=6, max_sessions=3)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
# Patch _write_memory_file to write actual JSON so _count_file_lines works
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 3}, "key_learnings": {"count": 3}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
@@ -209,20 +220,23 @@ class TestExtractItemsV2:
result = ext._extract_items_v2(mem_file, data)
assert result["success"] is True
assert result["extracted_count"] == 5 # 3 sessions + 2 learnings trimmed... let me check
# 6 sessions - 3 max = 3 excess sessions extracted
# 5 learnings - 3 max = 2 excess learnings extracted
# total = 5
assert result["extracted_count"] == 5
assert len(data["sessions"]) == 3
def test_trims_key_learnings_to_limit(self, monkeypatch, tmp_path):
ext = _import_extractor(monkeypatch)[0]
ext, mocks = _import_extractor(monkeypatch)
data = self._make_v2_data(num_sessions=2, num_learnings=7, max_sessions=3, max_learnings=4)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 3}, "key_learnings": {"count": 4}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
@@ -230,32 +244,42 @@ class TestExtractItemsV2:
result = ext._extract_items_v2(mem_file, data)
assert result["success"] is True
# sessions: 2 <= 3, no trim
# learnings: 7 - 4 = 3 extracted
assert len(data["key_learnings"]) == 4
assert result["extracted_count"] == 3
def test_skips_when_under_limits(self, monkeypatch, tmp_path):
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_v2_data(num_sessions=2, num_learnings=2, max_sessions=5, max_learnings=5)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 5}, "key_learnings": {"count": 5}}}},
}
result = ext._extract_items_v2(mem_file, data)
assert result["success"] is True
assert result.get("skipped") is True
def test_extracts_oldest_sessions_from_end(self, monkeypatch, tmp_path):
"""Sessions are stored newest-first, oldest at end. Extraction takes from end."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_v2_data(num_sessions=5, num_learnings=0, max_sessions=3, max_learnings=100)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 3}, "key_learnings": {"count": 100}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
@@ -271,13 +295,19 @@ class TestExtractItemsV2:
def test_extracts_oldest_key_learnings_from_end(self, monkeypatch, tmp_path):
"""Lowest-numbered entries (oldest, at end) should be extracted."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_v2_data(num_sessions=0, num_learnings=5, max_sessions=100, max_learnings=3)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 100}, "key_learnings": {"count": 3}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
@@ -416,7 +446,8 @@ class TestNormalizeMemoryFile:
result = norm.normalize_memory_file(tmp_path / "nope.json")
assert result["success"] is False
def test_moves_root_limits_into_metadata(self, monkeypatch, tmp_path):
def test_moves_root_limits_then_strips(self, monkeypatch, tmp_path):
"""Root limits merged into metadata, then stripped (limits live in config now)."""
norm, _ = _import_normalize(monkeypatch)
f = tmp_path / "test.local.json"
self._write_json(
@@ -431,10 +462,11 @@ class TestNormalizeMemoryFile:
assert result["success"] is True
data = json.loads(f.read_text(encoding="utf-8"))
assert "limits" not in {k for k in data if k != "document_metadata"}
assert data["document_metadata"]["limits"]["max_sessions"] == 20
assert "limits" not in data
assert "limits" not in data["document_metadata"]
def test_merges_root_limits_preserving_metadata_values(self, monkeypatch, tmp_path):
def test_merges_root_limits_then_strips(self, monkeypatch, tmp_path):
"""Root + metadata limits both get stripped (limits live in config now)."""
norm, _ = _import_normalize(monkeypatch)
f = tmp_path / "test.local.json"
self._write_json(
@@ -452,10 +484,8 @@ class TestNormalizeMemoryFile:
assert result["success"] is True
data = json.loads(f.read_text(encoding="utf-8"))
# metadata value (20) wins over root value (30)
assert data["document_metadata"]["limits"]["max_sessions"] == 20
# valid key from root gets merged in
assert data["document_metadata"]["limits"]["max_key_learnings"] == 25
assert "limits" not in data
assert "limits" not in data["document_metadata"]
def test_removes_root_status(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
@@ -518,7 +548,7 @@ class TestNormalizeMemoryFile:
f,
{
"document_metadata": {
"limits": {"max_sessions": 20},
"_usage": "Automated file.",
"status": {"last_health_check": "2026-03-31"},
},
"sessions": [],
@@ -528,14 +558,15 @@ class TestNormalizeMemoryFile:
assert result["success"] is True
assert result["changes"] == []
def test_removes_unused_limit_fields(self, monkeypatch, tmp_path):
def test_strips_entire_limits_block(self, monkeypatch, tmp_path):
"""Limits block in metadata is fully stripped (lives in config now)."""
norm, _ = _import_normalize(monkeypatch)
f = tmp_path / "test.local.json"
self._write_json(
f,
{
"document_metadata": {
"limits": {"max_sessions": 20, "max_lines": 600, "max_word_count": 9999, "max_token_count": 5000},
"limits": {"max_sessions": 20, "max_lines": 600, "max_word_count": 9999},
"status": {"last_health_check": "2026-03-31"},
},
"sessions": [],
@@ -545,10 +576,7 @@ class TestNormalizeMemoryFile:
assert result["success"] is True
data = json.loads(f.read_text(encoding="utf-8"))
assert "max_word_count" not in data["document_metadata"]["limits"]
assert "max_token_count" not in data["document_metadata"]["limits"]
assert "max_lines" not in data["document_metadata"]["limits"]
assert data["document_metadata"]["limits"]["max_sessions"] == 20
assert "limits" not in data["document_metadata"]
class TestTodosOperational:
@@ -579,13 +607,19 @@ class TestTodosOperational:
def test_v2_extraction_leaves_todos_untouched(self, monkeypatch, tmp_path):
"""v2 rollover trims sessions but never touches todos[]."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_data_with_todos(num_sessions=6, max_sessions=3, num_todos=5)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 3}, "key_learnings": {"count": 25}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
@@ -643,13 +677,19 @@ class TestTodosOperational:
def test_todos_survives_full_extraction_cycle(self, monkeypatch, tmp_path):
"""End-to-end: extract_items on a file with todos[] preserves them completely."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_data_with_todos(num_sessions=25, max_sessions=20, num_todos=8)
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
branch_key = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {branch_key: {"local": {"sessions": {"count": 20}, "key_learnings": {"count": 25}}}},
}
def fake_write(fp, d):
fp.write_text(json.dumps(d, indent=2), encoding="utf-8")
src/aipass/memory/tests/test_learnings.py
+1 -65
View File
@@ -13,7 +13,7 @@ Covers:
get_max_learnings, get_max_recently_completed,
ensure_timestamps, enforce_limit,
ensure_timestamps_completed, enforce_limit_completed,
add_learning, update_status_counts, process_file
update_status_counts, process_file
All tests use mocks or tmp_path -- no live filesystem or infrastructure access.
"""
@@ -445,70 +445,6 @@ class TestEnforceLimitCompleted:
assert result["max"] == 2
# ===========================================================================
# ADD LEARNING
# ===========================================================================
class TestAddLearning:
"""Tests for add_learning()."""
def test_file_not_found(self, mgr, tmp_path):
manager, _ = mgr
missing = tmp_path / "missing.json"
result = manager.add_learning(missing, "key", "value")
assert result["success"] is False
def test_adds_new_entry(self, mgr, tmp_path, monkeypatch):
manager, mock_mf = mgr
fp = tmp_path / "TEST.local.json"
fp.write_text("{}", encoding="utf-8")
data = {
"key_learnings": {},
"document_metadata": {"limits": {"max_learnings": 100}},
}
mock_mf.read_memory_file_data.return_value = data
mock_mf.write_memory_file_simple.return_value = None
# Mock enforce_limit inside add_learning
monkeypatch.setattr(
manager,
"enforce_limit",
lambda fp: {"success": True, "removed": 0},
)
result = manager.add_learning(fp, "test_key", "test value")
assert result["success"] is True
assert result["action"] == "added"
assert result["key"] == "test_key"
assert "[" in result["value"] # has timestamp
def test_updates_existing_entry(self, mgr, tmp_path, monkeypatch):
manager, mock_mf = mgr
fp = tmp_path / "TEST.local.json"
fp.write_text("{}", encoding="utf-8")
data = {
"key_learnings": {"existing_key": "old value [2025-01-01]"},
"document_metadata": {"limits": {"max_learnings": 100}},
}
mock_mf.read_memory_file_data.return_value = data
mock_mf.write_memory_file_simple.return_value = None
monkeypatch.setattr(
manager,
"enforce_limit",
lambda fp: {"success": True, "removed": 0},
)
result = manager.add_learning(fp, "existing_key", "updated value")
assert result["success"] is True
assert result["action"] == "updated"
# ===========================================================================
# UPDATE STATUS COUNTS
# ===========================================================================
src/aipass/memory/tests/test_rollover_pipeline.py
+19 -2
View File
@@ -82,12 +82,17 @@ def _import_extractor(monkeypatch):
mock_memory_files.read_memory_file_data = MagicMock(return_value=None)
mock_memory_files.write_memory_file_simple = MagicMock()
mock_config_loader = MagicMock()
mock_config_loader.section.return_value = {"defaults": {}, "per_branch": {}}
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.memory_files", mock_memory_files)
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.config_loader", mock_config_loader)
sys.modules.pop("aipass.memory.apps.handlers.rollover.extractor", None)
parent = sys.modules.get("aipass.memory.apps.handlers.rollover")
@@ -99,6 +104,7 @@ def _import_extractor(monkeypatch):
return extractor, {
"json_handler": mock_json_handler,
"memory_files": mock_memory_files,
"config_loader": mock_config_loader,
}
@@ -585,11 +591,22 @@ class TestExtractWithMetadata:
"""v2 schema extraction adds _metadata to each extracted entry."""
ext, mocks = _import_extractor(monkeypatch)
# Create a v2 file with sessions exceeding limits
# Branch name derived from parent of .trinity: tmp_path name (lowercase)
branch_name = tmp_path.name.lower()
# Provision limits via config per_branch
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {
branch_name: {
"local": {"sessions": {"count": 2}},
},
},
}
data = {
"document_metadata": {
"schema_version": "2.0.0",
"limits": {"max_sessions": 2},
"status": {},
},
"sessions": [
src/aipass/memory/tests/test_unified_schema.py
+100 -5
View File
@@ -61,12 +61,17 @@ def _import_extractor(monkeypatch):
mock_memory_files.read_memory_file_data = MagicMock(return_value=None)
mock_memory_files.write_memory_file_simple = MagicMock()
mock_config_loader = MagicMock()
mock_config_loader.section.return_value = {"defaults": {}, "per_branch": {}}
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.memory_files", mock_memory_files)
monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.config_loader", mock_config_loader)
sys.modules.pop("aipass.memory.apps.handlers.rollover.extractor", None)
parent = sys.modules.get("aipass.memory.apps.handlers.rollover")
@@ -78,6 +83,7 @@ def _import_extractor(monkeypatch):
return extractor, {
"json_handler": mock_json_handler,
"memory_files": mock_memory_files,
"config_loader": mock_config_loader,
}
@@ -121,7 +127,6 @@ class TestNormalizerNumberSort:
f,
{
"document_metadata": {
"limits": {"max_sessions": 20},
"status": {"last_health_check": "2026-06-13"},
},
"sessions": [
@@ -155,7 +160,6 @@ class TestNormalizerNumberSort:
f,
{
"document_metadata": {
"limits": {"max_sessions": 20},
"status": {"last_health_check": "2026-06-13"},
},
"sessions": original_sessions,
@@ -179,7 +183,6 @@ class TestNormalizerNumberSort:
f,
{
"document_metadata": {
"limits": {"max_sessions": 20},
"status": {"last_health_check": "2026-06-13"},
},
"sessions": [
@@ -232,9 +235,19 @@ class TestExtractorKeyLearningsList:
def test_kl_list_trims_oldest_from_end(self, monkeypatch, tmp_path):
"""List with 5 key_learnings, max 3 -> extracts 2 oldest (lowest numbers at end), keeps 3 newest."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_kl_data(num_kl=5, max_kl=3)
branch_name = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {
branch_name: {
"local": {"sessions": {"count": 100}, "key_learnings": {"count": 3}},
},
},
}
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
@@ -259,9 +272,19 @@ class TestExtractorKeyLearningsList:
def test_kl_list_under_limit_no_trim(self, monkeypatch, tmp_path):
"""List with 2 key_learnings, max 5 -> skipped, no extraction."""
ext, _ = _import_extractor(monkeypatch)
ext, mocks = _import_extractor(monkeypatch)
data = self._make_kl_data(num_kl=2, max_kl=5)
branch_name = tmp_path.name.lower()
mocks["config_loader"].section.return_value = {
"defaults": {},
"per_branch": {
branch_name: {
"local": {"sessions": {"count": 100}, "key_learnings": {"count": 5}},
},
},
}
mem_file = tmp_path / ".trinity" / "local.json"
mem_file.parent.mkdir(parents=True)
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
@@ -318,3 +341,75 @@ class TestListKeyLearningCharLimit:
assert result[0]["length"] == 250
assert result[0]["cap"] == 200
assert result[0]["over_by"] == 50
# ===========================================================================
# 4. Entry limits: casing normalization + char-cap is single source
# ===========================================================================
class TestEntryLimitsCasingAndCaps:
"""P6 — verify entry_limits normalizes branch casing and is the single cap home."""
def test_uppercase_branch_resolves_per_branch_overrides(self):
"""load_entry_limits('DEVPULSE') should find per_branch['devpulse'] overrides."""
mod = _get_entry_limits()
limits: dict[str, Any] = {
"enabled": True,
"enforce": True,
"entry_types": {
"sessions": {
"file": "local.json",
"container": "sessions",
"kind": "list",
"field": "summary",
"max_chars": 300,
},
},
"per_branch": {
"devpulse": {"sessions": {"max_chars": 500}},
},
}
with patch.object(mod.config_loader, "load", return_value={"entry_limits": limits}):
result = mod.load_entry_limits("DEVPULSE")
assert result["entry_types"]["sessions"]["max_chars"] == 500
def test_mixed_case_branch_resolves(self):
"""load_entry_limits('DevPulse') should normalize to lowercase."""
mod = _get_entry_limits()
limits: dict[str, Any] = {
"enabled": True,
"enforce": False,
"entry_types": {
"key_learnings": {
"file": "local.json",
"container": "key_learnings",
"kind": "list",
"field": "value",
"max_chars": 200,
},
},
"per_branch": {
"devpulse": {"key_learnings": {"max_chars": 150}},
},
}
with patch.object(mod.config_loader, "load", return_value={"entry_limits": limits}):
result = mod.load_entry_limits("DevPulse")
assert result["entry_types"]["key_learnings"]["max_chars"] == 150
def test_rollover_defaults_have_no_max_chars(self):
"""rollover.defaults should only carry counts, not max_chars (P6 unification)."""
mod = _get_entry_limits()
default_rollover = mod.config_loader.DEFAULT_CONFIG["rollover"]["defaults"]
for file_type, sections in default_rollover.items():
if file_type.startswith("_"):
continue
for section_name, section_val in sections.items():
assert "max_chars" not in section_val, (
f"rollover.defaults.{file_type}.{section_name} still has max_chars"
)
src/aipass/spawn/templates/birthright/.trinity/local.json
+3 -9
View File
@@ -12,20 +12,14 @@
"work_log",
"{{BRANCHNAME}}"
],
"limits": {
"max_sessions": 20,
"max_key_learnings": 25,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries."
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}",
"current_lines": 0
"last_health_check": "{{DATE}}"
}
},
"key_learnings": [],
"todos": [],
"sessions": [
{
"number": 1,
src/aipass/spawn/templates/birthright/.trinity/observations.json
+1 -5
View File
@@ -12,13 +12,9 @@
"patterns",
"{{BRANCHNAME}}"
],
"limits": {
"max_lines": 600,
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"current_lines": 0,
"last_health_check": "{{DATE}}"
}
},
src/aipass/spawn/templates/builder/.trinity/local.json
+4 -13
View File
@@ -12,23 +12,14 @@
"work_log",
"{{BRANCHNAME}}"
],
"limits": {
"max_sessions": 20,
"max_key_learnings": 25,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries. todos[] is OPERATIONAL — never vectorized/trimmed by rollover; completing/removing a todo is normal.",
"max_todos": 10,
"todo_text_max_chars": 200
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}",
"current_lines": 0
"last_health_check": "{{DATE}}"
}
},
"todos": [],
"key_learnings": [],
"todos": [],
"sessions": [
{
"number": 1,
@@ -38,4 +29,4 @@
"tags": []
}
]
}
}
src/aipass/spawn/templates/builder/.trinity/observations.json
+1 -5
View File
@@ -12,13 +12,9 @@
"patterns",
"{{BRANCHNAME}}"
],
"limits": {
"max_lines": 600,
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
},
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"current_lines": 0,
"last_health_check": "{{DATE}}"
}
},