feat(memory,spawn): self-documenting .trinity state-tabs + todo delete-on-done discipline

- .trinity sections carry config-sourced rollover/keep/char-cap tabs; @memory owns
  values (render_all_meta_tabs), @spawn resolves placeholders at create via spawn_pusher
- todos confirmed rollover-exempt; vestigial rollover-config entry removed
- prep/memo/startup (Claude+Codex): delete finished todos, don't leave status:done
- @memory README documents the system
- FPLAN-0285, FPLAN-0286
This commit is contained in:
AIOSAI
2026-06-25 03:00:36 -07:00
parent be8f87d6fb
commit fbf102c636
26 changed files with 1090 additions and 39 deletions
+21
View File
@@ -141,6 +141,11 @@
"standard": "handlers",
"reason": "Architectural: imports json.memory_files for metadata updates and monitor.detector for registry reading."
},
{
"file": "apps/handlers/tracking/tab_renderer.py",
"standard": "handlers",
"reason": "Architectural: imports json.memory_files, json.config_loader, and monitor.detector for config-driven tab rendering across all branches."
},
{
"file": "apps/handlers/search/vector_search.py",
"standard": "handlers",
@@ -451,6 +456,12 @@
"standard": "unused_function",
"reason": "is_plan_vectorized is a public API function for external callers."
},
{
"file": "apps/handlers/tracking/tab_renderer.py",
"standard": "unused_function",
"functions": ["render_all_meta_tabs"],
"reason": "Cross-branch public API — called by @spawn's build_replacements_dict to resolve {{*_META}} placeholders at branch creation."
},
{
"file": "apps/modules/templates.py",
"standard": "deep_nesting",
@@ -693,6 +704,16 @@
"standard": "meta",
"reason": "Test file — META block present at lines 1-7; hook false-positive on test file format."
},
{
"file": "tests/test_tab_renderer.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by design, not in 3-layer apps/ structure."
},
{
"file": "tests/test_tab_renderer.py",
"standard": "documentation",
"reason": "Test file — test functions don't require docstrings."
},
{
"file": "tests/test_handlers.py",
"standard": "architecture",
+37 -6
View File
@@ -64,11 +64,11 @@ memory/
│ ├── storage/ # chroma.py, chroma_subprocess.py
│ ├── symbolic/ # chroma_client, deduplicator, extractor, hook, retriever, storage
│ ├── templates/ # pusher.py, differ.py, spawn_pusher.py
│ ├── tracking/ # line_counter.py
│ ├── tracking/ # line_counter.py, tab_renderer.py
│ ├── vector/ # embedder.py, embed_subprocess.py
│ └── central_writer.py
├── templates/ # LOCAL.template.json, OBSERVATIONS.template.json
├── tests/ # 949 tests (31 test files)
├── tests/ # 978 tests
├── .chroma/ # ChromaDB vector store
└── memory_json/ # Operation logs + custom_config/memory.config.json
```
@@ -95,6 +95,38 @@ All ML operations (fastembed, chromadb) run via subprocess. The main process nev
---
## State-Tabs (`*_meta` keys)
Every `.trinity/local.json` and `.trinity/observations.json` carries inline `*_meta` banner strings that tell the editing agent what rollover rules apply to each section. Example:
```
"sessions_meta": "⟦ rollover ON → oldest archived to @memory · keep 15 · summary ≤300 chars ⟧"
```
**Source of truth:** `memory.config.json` — rollover counts (defaults + per-branch overrides) and entry char limits. Tab strings are *generated*, never hand-written.
**Sections:** `todos_meta` (rollover OFF — operational, never trimmed), `key_learnings_meta`, `sessions_meta`, `observations_meta` (all rollover ON).
### Two value flows
| Scenario | How tabs arrive |
|---|---|
| **Live branches** | `refresh_all_tabs()` walks the registry, renders tabs from config with per-branch overrides, writes them into `.trinity/` files. Wired after rollover, sync-lines, and push-templates. |
| **New branches** | Templates carry `{{TODOS_META}}`, `{{KEY_LEARNINGS_META}}`, `{{SESSIONS_META}}`, `{{OBSERVATIONS_META}}` placeholders. `spawn_pusher` propagates these (unresolved) from memory templates → spawn template sets. At branch creation, @spawn calls `render_all_meta_tabs()` to get rendered defaults and resolves the placeholders. |
### Public API
```python
from aipass.memory.apps.handlers.tracking.tab_renderer import render_all_meta_tabs
tabs = render_all_meta_tabs()
# → {"TODOS_META": "⟦ rollover OFF ...", "KEY_LEARNINGS_META": "⟦ rollover ON ...", ...}
```
Returns defaults (not per-branch overrides) — appropriate for template resolution at branch creation.
---
## Integration Points
**Depends on:**
@@ -117,9 +149,8 @@ All ML operations (fastembed, chromadb) run via subprocess. The main process nev
## Quality
- **Tests:** 949 passed, 0 failures, 0 skips
- **Test files:** 31
- **Seedgo:** 100% — maintained since s12
- **Tests:** 978 passed, 0 failures, 0 skips
- **Seedgo:** 100%
---
@@ -131,7 +162,7 @@ All ML operations (fastembed, chromadb) run via subprocess. The main process nev
---
*Last Updated: 2026-05-16*
*Last Updated: 2026-06-25*
---
[← Back to AIPass](../../../README.md)
@@ -114,7 +114,6 @@ DEFAULT_CONFIG: dict[str, Any] = {
"local": {
"sessions": {"count": 20},
"key_learnings": {"count": 25},
"todos": {"count": 10},
},
"observations": {
"observations": {"count": 25},
@@ -160,6 +160,12 @@ def _merge_metadata(curr_meta: dict, tmpl_meta: dict) -> List[str]:
curr_meta[key] = tmpl_val
changes.append(f"document_metadata.{key}: {old} -> {tmpl_val}")
# _usage
tmpl_usage = tmpl_meta.get("_usage")
if tmpl_usage and curr_meta.get("_usage") != tmpl_usage:
curr_meta["_usage"] = tmpl_usage
changes.append("document_metadata._usage: updated from template")
# Tags
tmpl_tags = tmpl_meta.get("tags", [])
if tmpl_tags and set(curr_meta.get("tags", [])) != set(tmpl_tags):
@@ -0,0 +1,332 @@
# =================== AIPass ====================
# Name: tab_renderer.py
# Description: Config-generated state-tabs for .trinity memory files
# Version: 1.0.0
# Created: 2026-06-25
# Modified: 2026-06-25
# =============================================
"""
Tab Renderer Handler
Generates per-section state-tab strings (e.g. ``⟦ rollover ON ... ⟧``) from
``memory.config.json`` and writes them as ``*_meta`` keys into every branch's
``.trinity/local.json`` and ``.trinity/observations.json``.
Purpose:
Make memory files self-documenting. Each section carries a single-line
banner that tells the editing agent whether rollover is ON/OFF, the keep
count, and the char cap — all derived from config so they never drift.
Independence:
Uses config_loader for config, detector helpers for branch discovery,
and memory_files for safe I/O. No service or module dependencies.
"""
from typing import Any, Dict
from aipass.prax.apps.modules.logger import get_system_logger
from aipass.memory.apps.handlers.json import json_handler
logger = get_system_logger()
_CORRECTED_USAGE_LOCAL = (
"Automated file — add entries within your sections, newest on top. "
"Rollover auto-archives sessions/key_learnings (+ observations.json) to @memory; "
"todos[] are OPERATIONAL and NEVER rolled — prune done ones by hand at /prep. "
"Limits live in @memory’s memory.config.json."
)
_CORRECTED_USAGE_OBS = (
"Automated file — add entries within your sections, newest on top. "
"Rollover auto-archives the oldest observations to @memory. "
"Limits live in @memory’s memory.config.json."
)
# =============================================================================
# KEY ORDERING
# =============================================================================
# Canonical key order for local.json
_LOCAL_KEY_ORDER = [
"document_metadata",
"todos_meta",
"todos",
"key_learnings_meta",
"key_learnings",
"sessions_meta",
"sessions",
]
# Canonical key order for observations.json
_OBSERVATIONS_KEY_ORDER = [
"document_metadata",
"guidelines",
"observations_meta",
"observations",
]
def _reorder_keys(data: Dict[str, Any], key_order: list[str]) -> Dict[str, Any]:
"""Rebuild *data* with keys in *key_order* first, then any remaining keys."""
ordered: Dict[str, Any] = {}
for key in key_order:
if key in data:
ordered[key] = data[key]
# Append any keys not in the canonical order
for key in data:
if key not in ordered:
ordered[key] = data[key]
return ordered
# =============================================================================
# TAB RENDERING
# =============================================================================
def render_tab(
section_name: str,
rollover_cfg: dict,
entry_limits_cfg: dict,
branch_name: str,
) -> str:
"""Generate the state-tab string for a section.
Args:
section_name: One of 'key_learnings', 'sessions', 'observations', 'todos'.
rollover_cfg: The ``rollover`` section from memory.config.json.
entry_limits_cfg: The ``entry_limits`` section from memory.config.json.
branch_name: Branch name (lowercase) for per-branch overrides.
Returns:
The rendered tab string (e.g. ``⟦ rollover ON ... ⟧``).
"""
# --- Resolve entry-limits for this section --------------------------------
entry_types = entry_limits_cfg.get("entry_types", {})
section_limits = entry_types.get(section_name, {})
max_chars = section_limits.get("max_chars", 300)
field = section_limits.get("field", "value")
# --- Todos are special: rollover OFF, static shape ------------------------
if section_name == "todos":
return (
f"⟦ rollover OFF — operational, never trimmed · "
f"cap ~10 entries · task ≤{max_chars} chars ⟧ "
f"RULE: DELETE each todo when done (never leave status:done) "
f"+ reconcile on load; proof goes in the session entry, "
f"not the todo. Add freely — BAU."
)
# --- Rollover sections: look up count -------------------------------------
per_branch = rollover_cfg.get("per_branch", {})
defaults = rollover_cfg.get("defaults", {})
branch_cfg = per_branch.get(branch_name, defaults)
# Determine which file-level block to read
if section_name == "observations":
file_block = branch_cfg.get("observations", {})
else:
file_block = branch_cfg.get("local", {})
section_cfg = file_block.get(section_name, {})
count = section_cfg.get("count", 15)
return f"⟦ rollover ON → oldest archived to @memory · keep {count} · {field} ≤{max_chars} chars ⟧"
# =============================================================================
# PER-FILE TAB WRITERS
# =============================================================================
def _refresh_local(branch_name, local_path, rollover_cfg, entry_limits_cfg):
"""Inject *_meta tabs into a branch's local.json. Returns (ok, error_msg)."""
from aipass.memory.apps.handlers.json.memory_files import (
read_memory_file_data,
write_memory_file_simple,
)
data = read_memory_file_data(local_path)
if data is None:
return False, None # file unreadable, skip silently
meta = data.get("document_metadata", {})
if meta.get("_usage") != _CORRECTED_USAGE_LOCAL:
meta["_usage"] = _CORRECTED_USAGE_LOCAL
data["todos_meta"] = render_tab(
"todos",
rollover_cfg,
entry_limits_cfg,
branch_name,
)
data["key_learnings_meta"] = render_tab(
"key_learnings",
rollover_cfg,
entry_limits_cfg,
branch_name,
)
data["sessions_meta"] = render_tab(
"sessions",
rollover_cfg,
entry_limits_cfg,
branch_name,
)
data = _reorder_keys(data, _LOCAL_KEY_ORDER)
if write_memory_file_simple(local_path, data):
return True, None
return False, f"{branch_name}/local.json: write failed"
def _refresh_observations(branch_name, obs_path, rollover_cfg, entry_limits_cfg):
"""Inject observations_meta tab into a branch's observations.json. Returns (ok, error_msg)."""
from aipass.memory.apps.handlers.json.memory_files import (
read_memory_file_data,
write_memory_file_simple,
)
data = read_memory_file_data(obs_path)
if data is None:
return False, None # file unreadable, skip silently
meta = data.get("document_metadata", {})
if meta.get("_usage") != _CORRECTED_USAGE_OBS:
meta["_usage"] = _CORRECTED_USAGE_OBS
data["observations_meta"] = render_tab(
"observations",
rollover_cfg,
entry_limits_cfg,
branch_name,
)
data = _reorder_keys(data, _OBSERVATIONS_KEY_ORDER)
if write_memory_file_simple(obs_path, data):
return True, None
return False, f"{branch_name}/observations.json: write failed"
# =============================================================================
# REFRESH ALL BRANCHES
# =============================================================================
def refresh_all_tabs() -> dict:
"""Render and write state-tabs to all branch .trinity files.
Walks the registry, reads each branch's memory files, computes tab
strings from config, injects them as ``*_meta`` keys, and writes back
with correct key ordering.
Returns:
Dict with success status and counts.
"""
from aipass.memory.apps.handlers.json.config_loader import (
load as load_config,
)
from aipass.memory.apps.handlers.monitor.detector import (
_read_registry,
_get_memory_file_path,
)
config = load_config()
rollover_cfg = config.get("rollover", {})
entry_limits_cfg = config.get("entry_limits", {})
branches = _read_registry()
if not branches:
return {
"success": True,
"updated": 0,
"skipped": 0,
"message": "No branches in registry",
}
updated = 0
skipped = 0
errors: list[str] = []
for branch in branches:
branch_name = branch.get("name", "UNKNOWN").lower()
for mem_type in ("local", "observations"):
u, s, e = _refresh_one_file(
branch,
branch_name,
mem_type,
rollover_cfg,
entry_limits_cfg,
_get_memory_file_path,
)
updated += u
skipped += s
errors.extend(e)
json_handler.log_operation(
"refresh_all_tabs",
{"updated": updated, "skipped": skipped, "errors": len(errors)},
module_name="tab_renderer",
)
logger.info(
f"[tab_renderer] Refreshed tabs: {updated} updated, {skipped} skipped, {len(errors)} errors",
)
return {
"success": True,
"updated": updated,
"skipped": skipped,
"errors": errors,
}
def render_all_meta_tabs() -> dict[str, str]:
"""Render all four *_meta tab strings from memory.config.json defaults.
Public API for @spawn (and any other consumer) to resolve ``{{*_META}}``
placeholders at branch-creation time.
Returns:
Dict with keys TODOS_META, KEY_LEARNINGS_META, SESSIONS_META,
OBSERVATIONS_META — each a rendered state-tab string.
"""
from aipass.memory.apps.handlers.json.config_loader import (
load as load_config,
)
config = load_config()
rollover_cfg = config.get("rollover", {})
entry_limits_cfg = config.get("entry_limits", {})
_default = "__template_default__"
return {
"TODOS_META": render_tab("todos", rollover_cfg, entry_limits_cfg, _default),
"KEY_LEARNINGS_META": render_tab("key_learnings", rollover_cfg, entry_limits_cfg, _default),
"SESSIONS_META": render_tab("sessions", rollover_cfg, entry_limits_cfg, _default),
"OBSERVATIONS_META": render_tab("observations", rollover_cfg, entry_limits_cfg, _default),
}
def _refresh_one_file(branch, branch_name, mem_type, rollover_cfg, entry_limits_cfg, get_path_fn):
"""Refresh tabs for a single memory file. Returns (updated, skipped, errors)."""
file_path = get_path_fn(branch, mem_type)
if file_path is None:
return 0, 1, []
refresher = _refresh_local if mem_type == "local" else _refresh_observations
try:
ok, err = refresher(
branch_name,
file_path,
rollover_cfg,
entry_limits_cfg,
)
except Exception as e:
logger.warning(f"[tab_renderer] {branch_name}/{mem_type}.json: {e}")
return 0, 0, [f"{branch_name}/{mem_type}.json: {e}"]
if ok:
return 1, 0, []
if err:
return 0, 0, [err]
return 0, 1, []
@@ -236,6 +236,15 @@ def run_rollover() -> bool:
error(f"{fail['trigger']} - {fail['stage']}: {fail['error']}")
json_handler.log_operation("rollover_execute", {"triggers": triggers_count, "success_count": success_count})
# Refresh state-tabs after rollover (counts may have changed)
try:
from aipass.memory.apps.handlers.tracking.tab_renderer import refresh_all_tabs
refresh_all_tabs()
except Exception as e:
logger.warning(f"[rollover] Tab refresh failed: {e}")
return success_count > 0
@@ -325,6 +334,14 @@ def sync_line_counts() -> None:
else:
error("Failed to sync line counts")
# Refresh state-tabs after line count sync
try:
from aipass.memory.apps.handlers.tracking.tab_renderer import refresh_all_tabs
refresh_all_tabs()
except Exception as e:
logger.warning(f"[rollover] Tab refresh failed: {e}")
console.print()
@@ -119,6 +119,14 @@ def handle_command(command: str, args: List[str]) -> bool:
except Exception as e:
error(f"Spawn template push crashed: {e}")
logger.error(f"[templates] spawn push crashed: {e}")
# Refresh state-tabs after template push
if not dry_run:
try:
from aipass.memory.apps.handlers.tracking.tab_renderer import refresh_all_tabs
refresh_all_tabs()
except Exception as e:
logger.warning(f"[templates] Tab refresh failed: {e}")
return True
if sub == "diff-templates":
@@ -162,6 +170,14 @@ def handle_command(command: str, args: List[str]) -> bool:
except Exception as e:
error(f"Spawn template push crashed: {e}")
logger.error(f"[templates] spawn push crashed: {e}")
# Refresh state-tabs after template push
if not dry_run:
try:
from aipass.memory.apps.handlers.tracking.tab_renderer import refresh_all_tabs
refresh_all_tabs()
except Exception as e:
logger.warning(f"[templates] Tab refresh failed: {e}")
return True
elif command == "diff-templates":
@@ -1,7 +1,11 @@
{
"last_push": "2026-06-07 21:40:48",
"last_push": "2026-06-25 01:55:01",
"last_push_branches": [
"COMMONS",
"DAEMON",
"HOOKS",
"SKILLS",
"ai_mail",
"aipass",
"api",
"cli",
@@ -12,14 +12,17 @@
"work_log",
"{{BRANCHNAME}}"
],
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"_usage": "Automated file — add entries within your sections, newest on top. Rollover auto-archives sessions/key_learnings (+ observations.json) to @memory; todos[] are OPERATIONAL and NEVER rolled — prune done ones by hand at /prep. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}"
}
},
"key_learnings": [],
"todos_meta": "{{TODOS_META}}",
"todos": [],
"key_learnings_meta": "{{KEY_LEARNINGS_META}}",
"key_learnings": [],
"sessions_meta": "{{SESSIONS_META}}",
"sessions": [
{
"number": 1,
@@ -12,7 +12,7 @@
"patterns",
"{{BRANCHNAME}}"
],
"_usage": "Automated file — add entries within your sections; rollover trims automatically. Limits live in @memory's memory.config.json.",
"_usage": "Automated file — add entries within your sections, newest on top. Rollover auto-archives sessions/key_learnings (+ observations.json) to @memory; todos[] are OPERATIONAL and NEVER rolled — prune done ones by hand at /prep. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}"
@@ -22,6 +22,7 @@
"purpose": "Capture collaboration patterns and experiential insights over time",
"chronological_order": "Newest entries at TOP, oldest at BOTTOM - NEVER reorder"
},
"observations_meta": "{{OBSERVATIONS_META}}",
"observations": [
{
"number": 1,
@@ -0,0 +1,571 @@
# =================== AIPass ====================
# Name: test_tab_renderer.py
# Description: Tests for tab_renderer handler (FPLAN-0285)
# Version: 1.0.0
# Created: 2026-06-25
# Modified: 2026-06-25
# =============================================
"""
Tests for the tab_renderer handler.
Covers:
1. render_tab() — correct strings for each section type.
2. render_tab() — per-branch overrides from config.
3. render_tab() — fallback to defaults when branch not in per_branch.
4. _reorder_keys() — canonical key ordering.
5. refresh_all_tabs() — reads config and writes tabs (mocked I/O).
6. Key ordering verification after tab insertion.
"""
import importlib
import json
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _fresh_tab_renderer(monkeypatch):
"""Drop cached module so each test gets a fresh import."""
for mod_name in list(sys.modules):
if "tab_renderer" in mod_name:
sys.modules.pop(mod_name, None)
# Also clear json sub-modules that conftest replaces with MagicMock
sys.modules.pop("aipass.memory.apps.handlers.json", None)
sys.modules.pop("aipass.memory.apps.handlers.json.json_handler", None)
yield
def _get_module():
"""Import and return the tab_renderer module."""
return importlib.import_module(
"aipass.memory.apps.handlers.tracking.tab_renderer",
)
# ---------------------------------------------------------------------------
# Shared config fixtures
# ---------------------------------------------------------------------------
SAMPLE_ROLLOVER_CFG = {
"defaults": {
"local": {
"sessions": {"count": 15},
"key_learnings": {"count": 15},
},
"observations": {
"observations": {"count": 15},
},
},
"per_branch": {
"devpulse": {
"local": {
"sessions": {"count": 20},
"key_learnings": {"count": 25},
},
"observations": {
"observations": {"count": 30},
},
},
},
}
SAMPLE_ENTRY_LIMITS_CFG = {
"entry_types": {
"key_learnings": {"field": "value", "max_chars": 200},
"sessions": {"field": "summary", "max_chars": 300},
"todos": {"field": "task", "max_chars": 150},
"observations": {"field": "note", "max_chars": 300},
},
}
# ===========================================================================
# 1. render_tab — key_learnings (rollover ON)
# ===========================================================================
class TestRenderTabKeyLearnings:
def test_default_branch(self):
mod = _get_module()
tab = mod.render_tab(
"key_learnings",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"memory",
)
assert tab.startswith("⟦")
assert tab.endswith("⟧")
assert "rollover ON" in tab
assert "keep 15" in tab
assert "value ≤20" in tab # ≤200
def test_per_branch_override(self):
mod = _get_module()
tab = mod.render_tab(
"key_learnings",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"devpulse",
)
assert "keep 25" in tab
# ===========================================================================
# 2. render_tab — sessions (rollover ON)
# ===========================================================================
class TestRenderTabSessions:
def test_sessions_default(self):
mod = _get_module()
tab = mod.render_tab(
"sessions",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"memory",
)
assert "rollover ON" in tab
assert "keep 15" in tab
assert "summary" in tab
assert "≤30" in tab # ≤300
def test_sessions_per_branch(self):
mod = _get_module()
tab = mod.render_tab(
"sessions",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"devpulse",
)
assert "keep 20" in tab
# ===========================================================================
# 3. render_tab — observations (rollover ON)
# ===========================================================================
class TestRenderTabObservations:
def test_observations_default(self):
mod = _get_module()
tab = mod.render_tab(
"observations",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"memory",
)
assert "rollover ON" in tab
assert "keep 15" in tab
assert "note" in tab
def test_observations_per_branch(self):
mod = _get_module()
tab = mod.render_tab(
"observations",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"devpulse",
)
assert "keep 30" in tab
# ===========================================================================
# 4. render_tab — todos (rollover OFF, static shape)
# ===========================================================================
class TestRenderTabTodos:
def test_todos_static(self):
mod = _get_module()
tab = mod.render_tab(
"todos",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"memory",
)
assert "rollover OFF" in tab
assert "cap ~10 entries" in tab
assert "task ≤15" in tab # ≤150
assert "RULE: DELETE" in tab
assert "BAU" in tab
def test_todos_ignores_per_branch_rollover(self):
"""Todos are always rollover OFF regardless of per_branch config."""
mod = _get_module()
tab = mod.render_tab(
"todos",
SAMPLE_ROLLOVER_CFG,
SAMPLE_ENTRY_LIMITS_CFG,
"devpulse",
)
assert "rollover OFF" in tab
# ===========================================================================
# 5. _reorder_keys — canonical key ordering
# ===========================================================================
class TestReorderKeys:
def test_local_key_order(self):
mod = _get_module()
data = {
"sessions": [],
"document_metadata": {},
"todos": [],
"key_learnings": [],
"extra_field": "preserved",
}
ordered = mod._reorder_keys(data, mod._LOCAL_KEY_ORDER)
keys = list(ordered.keys())
assert keys[0] == "document_metadata"
# todos before key_learnings before sessions
assert keys.index("todos") < keys.index("key_learnings")
assert keys.index("key_learnings") < keys.index("sessions")
# extra_field at the end
assert keys[-1] == "extra_field"
def test_observations_key_order(self):
mod = _get_module()
data = {
"observations": [],
"document_metadata": {},
"guidelines": {},
"observations_meta": "tab",
}
ordered = mod._reorder_keys(data, mod._OBSERVATIONS_KEY_ORDER)
keys = list(ordered.keys())
assert keys == [
"document_metadata",
"guidelines",
"observations_meta",
"observations",
]
def test_meta_before_array(self):
"""Meta key must appear immediately before its corresponding array."""
mod = _get_module()
data = {
"document_metadata": {},
"todos": [],
"todos_meta": "tab-todos",
"key_learnings": [],
"key_learnings_meta": "tab-kl",
"sessions": [],
"sessions_meta": "tab-sessions",
}
ordered = mod._reorder_keys(data, mod._LOCAL_KEY_ORDER)
keys = list(ordered.keys())
# Each *_meta must be immediately before its array
assert keys.index("todos_meta") + 1 == keys.index("todos")
assert keys.index("key_learnings_meta") + 1 == keys.index(
"key_learnings",
)
assert keys.index("sessions_meta") + 1 == keys.index("sessions")
# ===========================================================================
# 6. refresh_all_tabs — integration with mocked I/O
# ===========================================================================
class TestRefreshAllTabs:
def _make_local_data(self):
return {
"document_metadata": {"document_type": "session_history"},
"todos": [{"task": "test"}],
"key_learnings": [],
"sessions": [],
}
def _make_obs_data(self):
return {
"document_metadata": {
"document_type": "collaboration_patterns",
},
"guidelines": {},
"observations": [],
}
def test_writes_tabs_to_files(self, tmp_path):
"""refresh_all_tabs reads config, walks branches, writes tabs."""
mod = _get_module()
# Set up branch dir with .trinity files
branch_dir = tmp_path / "src" / "aipass" / "test_branch"
trinity = branch_dir / ".trinity"
trinity.mkdir(parents=True)
local_path = trinity / "local.json"
obs_path = trinity / "observations.json"
local_path.write_text(
json.dumps(self._make_local_data(), indent=2),
encoding="utf-8",
)
obs_path.write_text(
json.dumps(self._make_obs_data(), indent=2),
encoding="utf-8",
)
# Mock registry to return our test branch
mock_branches = [
{"name": "test_branch", "path": str(branch_dir)},
]
def mock_get_path(branch, mem_type):
p = Path(branch["path"]) / ".trinity" / f"{mem_type}.json"
return p if p.exists() else None
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with (
patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._read_registry",
return_value=mock_branches,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._get_memory_file_path",
side_effect=mock_get_path,
),
):
result = mod.refresh_all_tabs()
assert result["success"] is True
assert result["updated"] == 2 # local + observations
# Verify local.json has tabs
local_data = json.loads(local_path.read_text(encoding="utf-8"))
assert "todos_meta" in local_data
assert "key_learnings_meta" in local_data
assert "sessions_meta" in local_data
assert "rollover OFF" in local_data["todos_meta"]
assert "rollover ON" in local_data["key_learnings_meta"]
assert "rollover ON" in local_data["sessions_meta"]
# Verify observations.json has tab
obs_data = json.loads(obs_path.read_text(encoding="utf-8"))
assert "observations_meta" in obs_data
assert "rollover ON" in obs_data["observations_meta"]
def test_key_order_after_refresh(self, tmp_path):
"""After refresh, keys are in canonical order."""
mod = _get_module()
branch_dir = tmp_path / "src" / "aipass" / "ordered_branch"
trinity = branch_dir / ".trinity"
trinity.mkdir(parents=True)
local_path = trinity / "local.json"
local_path.write_text(
json.dumps(self._make_local_data(), indent=2),
encoding="utf-8",
)
obs_path = trinity / "observations.json"
obs_path.write_text(
json.dumps(self._make_obs_data(), indent=2),
encoding="utf-8",
)
mock_branches = [
{"name": "ordered_branch", "path": str(branch_dir)},
]
def mock_get_path(branch, mem_type):
p = Path(branch["path"]) / ".trinity" / f"{mem_type}.json"
return p if p.exists() else None
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with (
patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._read_registry",
return_value=mock_branches,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._get_memory_file_path",
side_effect=mock_get_path,
),
):
mod.refresh_all_tabs()
local_data = json.loads(local_path.read_text(encoding="utf-8"))
keys = list(local_data.keys())
expected_prefix = [
"document_metadata",
"todos_meta",
"todos",
"key_learnings_meta",
"key_learnings",
"sessions_meta",
"sessions",
]
assert keys[: len(expected_prefix)] == expected_prefix
def test_empty_registry(self):
"""refresh_all_tabs returns early if no branches in registry."""
mod = _get_module()
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with (
patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._read_registry",
return_value=[],
),
):
result = mod.refresh_all_tabs()
assert result["success"] is True
assert result["updated"] == 0
assert "No branches" in result.get("message", "")
def test_no_templates_updated_key(self, tmp_path):
"""refresh_all_tabs result dict has no templates_updated key (literal-baking removed)."""
mod = _get_module()
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with (
patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._read_registry",
return_value=[],
),
):
result = mod.refresh_all_tabs()
assert "templates_updated" not in result
def test_missing_file_skipped(self, tmp_path):
"""Branch with missing .trinity files is skipped, not errored."""
mod = _get_module()
branch_dir = tmp_path / "src" / "aipass" / "empty_branch"
branch_dir.mkdir(parents=True)
# No .trinity directory at all
mock_branches = [
{"name": "empty_branch", "path": str(branch_dir)},
]
def mock_get_path(branch, mem_type):
p = Path(branch["path"]) / ".trinity" / f"{mem_type}.json"
return p if p.exists() else None
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with (
patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._read_registry",
return_value=mock_branches,
),
patch(
"aipass.memory.apps.handlers.monitor.detector._get_memory_file_path",
side_effect=mock_get_path,
),
):
result = mod.refresh_all_tabs()
assert result["success"] is True
assert result["skipped"] == 2 # local + observations
assert result["updated"] == 0
# ===========================================================================
# 8. render_all_meta_tabs — public API for spawn
# ===========================================================================
class TestRenderAllMetaTabs:
def test_returns_four_keys(self):
mod = _get_module()
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
):
tabs = mod.render_all_meta_tabs()
assert set(tabs.keys()) == {
"TODOS_META",
"KEY_LEARNINGS_META",
"SESSIONS_META",
"OBSERVATIONS_META",
}
def test_values_are_rendered_strings(self):
mod = _get_module()
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
):
tabs = mod.render_all_meta_tabs()
assert "rollover OFF" in tabs["TODOS_META"]
assert "rollover ON" in tabs["KEY_LEARNINGS_META"]
assert "rollover ON" in tabs["SESSIONS_META"]
assert "rollover ON" in tabs["OBSERVATIONS_META"]
assert "{{" not in tabs["TODOS_META"]
def test_uses_defaults_not_per_branch(self):
mod = _get_module()
mock_config = {
"rollover": SAMPLE_ROLLOVER_CFG,
"entry_limits": SAMPLE_ENTRY_LIMITS_CFG,
}
with patch(
"aipass.memory.apps.handlers.json.config_loader.load",
return_value=mock_config,
):
tabs = mod.render_all_meta_tabs()
assert "keep 15" in tabs["KEY_LEARNINGS_META"]
assert "keep 15" in tabs["SESSIONS_META"]