fix(memory,prompts): rollover repair + retire-for-all + seedgo #37 path cleanup

Batch of S243/S244 work held for PR640. Only devpulse has git write, so all
branches' changes (@memory, @prax, @hooks, @aipass, @skills, @flow, @backup,
@seedgo) land through this single commit.

Memory rollover (correctness):
- @hooks rollover hook now delegates to drone @memory rollover check/run — it
  had been reading stale .trinity limits (moved to memory.config.json by
  DPLAN-0210), falling back to a 600-line check that never fired, so rollover was
  silently dead for weeks. compact.py reads the current list schema. Both fail loud.
- @memory removed the v1 line-count/600 fallback entirely — v2-only, fail-loud
  (959 tests).

Retire-for-all (DPLAN-0215):
- Legacy global prompt fully removed across every runtime: global_loader.py +
  tests deleted, hooks.json/project_hooks.json blocks stripped, bootstrap global
  seeding removed, cadence default + bypass cleaned, global .md files archived.
  Codex SessionStart + Claude cadence read the same tier files.

Hardcoded-path cleanup (seedgo #37):
- New HARDCODED_PATH checker (#37). @memory symbolic.py + @prax branch_detector.py
  home paths -> dynamic/generic. Both 100% Hardcoded_Path.
- seedgo provider_hooks_snapshot.json fixture refreshed to the tiered baseline.

Genericization: patrick -> user across tracked source, docs, templates.
CHANGELOG: 2026-06-19 + 2026-06-23 sections added.
This commit is contained in:
AIOSAI
2026-06-23 16:17:10 -07:00
parent 4cd68a78b6
commit 6db606eb01
56 changed files with 1739 additions and 1182 deletions
@@ -165,7 +165,7 @@ def _is_private_branch_email(email: str) -> bool:
email address is registered to a private (isolated) branch.
Args:
email: Email address to check (e.g., "@patrick_private")
email: Email address to check (e.g., "@private_branch")
Returns:
True if email belongs to a private branch, False otherwise
@@ -11,7 +11,8 @@ Init Bootstrap Handler - PRIVATE implementation
Business logic for `aipass init`. Creates the project scaffold:
1. {NAME}_REGISTRY.json — project registry with UUID
2. .aipass/aipass_global_prompt.md — global prompt (injected every turn)
2. .aipass/tier0_kernel.md — tier 0 kernel prompt (every turn)
2b..aipass/tier1_navmap.md — tier 1 navigation map (periodic)
3. CLAUDE.md — project prompt (Claude Code reads this)
4. AGENTS.md — Codex equivalent of CLAUDE.md
5. README.md — getting started guide
@@ -69,14 +70,6 @@ def _detect_aipass_home() -> str | None:
return None
def _resolve_global_prompt(name: str, aipass_home: str | None, dest: Path) -> str:
"""Resolve global prompt content from source template or fallback generator."""
source = Path(aipass_home) / ".aipass" / "project_global_prompt.md" if aipass_home else None
if source and source.is_file():
return source.read_text(encoding="utf-8").replace("{name}", name)
return sc.with_source(sc.global_prompt_md(name), dest)
def _hook_fingerprint(hook_entry: dict) -> str:
"""Extract a comparable fingerprint from a hook entry."""
commands = []
@@ -323,10 +316,14 @@ def init_project(target: Path, project_name: str | None = None) -> dict:
aipass_dir = target / ".aipass"
aipass_dir.mkdir(exist_ok=True)
global_prompt_path = aipass_dir / "aipass_global_prompt.md"
if not global_prompt_path.exists():
global_prompt_path.write_text(_resolve_global_prompt(name, aipass_home, global_prompt_path), encoding="utf-8")
created.append(str(global_prompt_path))
# 2. .aipass/tier0_kernel.md + tier1_navmap.md — tiered prompt injection
for tier_file in ("tier0_kernel.md", "tier1_navmap.md"):
tier_dest = aipass_dir / tier_file
if not tier_dest.exists() and aipass_home:
tier_src = Path(aipass_home) / ".aipass" / tier_file
if tier_src.is_file():
shutil.copy2(str(tier_src), str(tier_dest))
created.append(str(tier_dest))
# 2b. .aipass/hooks.json — project hook config from template
hooks_json_path = aipass_dir / "hooks.json"
@@ -488,14 +485,21 @@ def update_project(target: Path) -> dict:
# --- Managed files: write only when content has changed ---
global_prompt_path = aipass_dir / "aipass_global_prompt.md"
aipass_home = aipass_home or _detect_aipass_home()
generated = _resolve_global_prompt(name, aipass_home, global_prompt_path)
if not global_prompt_path.exists() or global_prompt_path.read_text(encoding="utf-8") != generated:
global_prompt_path.write_text(generated, encoding="utf-8")
updated.append(str(global_prompt_path))
else:
already_current.append(str(global_prompt_path))
# tier0_kernel.md + tier1_navmap.md — tiered prompt injection
for tier_file in ("tier0_kernel.md", "tier1_navmap.md"):
tier_dest = aipass_dir / tier_file
tier_src = Path(aipass_home) / ".aipass" / tier_file if aipass_home else None
if tier_src and tier_src.is_file():
canonical = tier_src.read_text(encoding="utf-8")
if not tier_dest.exists() or tier_dest.read_text(encoding="utf-8") != canonical:
tier_dest.write_text(canonical, encoding="utf-8")
updated.append(str(tier_dest))
else:
already_current.append(str(tier_dest))
elif tier_dest.exists():
already_current.append(str(tier_dest))
# settings.json — smart merge: preserve user hooks + env, update AIPass hooks
settings_path = claude_dir / "settings.json"
+2 -2
View File
@@ -16,7 +16,7 @@ Usage:
aipass init # show progress / introspection
aipass init run # interactive
aipass init run --non-interactive # CI/headless, all defaults
aipass init run --name Patrick --cli claude
aipass init run --name YourName --cli claude
aipass init run --dry-run # walk all 12 stages, no destructive ops
# - skips drone @spawn create (stage 8)
# - skips tmux/wt handoff (stage 11)
@@ -876,7 +876,7 @@ def print_help() -> None:
console.print("[yellow]USAGE:[/yellow]")
console.print(" [green]aipass init run[/green] [dim]# interactive[/dim]")
console.print(" [green]aipass init run --non-interactive[/green] [dim]# CI/headless[/dim]")
console.print(" [green]aipass init run --name Patrick[/green] [dim]# pre-fill name[/dim]")
console.print(" [green]aipass init run --name YourName[/green] [dim]# pre-fill name[/dim]")
console.print(" [green]aipass init run --cli claude[/green] [dim]# pre-fill CLI[/dim]")
console.print(" [green]aipass init run --no-docker[/green] [dim]# skip docker offer[/dim]")
console.print(" [green]aipass init run --dry-run[/green] [dim]# walk all stages, no writes[/dim]")
+160 -20
View File
@@ -98,7 +98,6 @@ def test_init_project_creates_all_expected_files(tmp_path):
expected_files = [
target / "DEMO_REGISTRY.json",
target / ".aipass" / "aipass_global_prompt.md",
target / "CLAUDE.md",
target / "AGENTS.md",
target / "README.md",
@@ -107,8 +106,13 @@ def test_init_project_creates_all_expected_files(tmp_path):
target / ".claude" / "commands" / "prep.md",
target / "src" / "demo" / "__init__.py",
]
# Tier files are env-dependent (need AIPASS_HOME)
if result["aipass_home"]:
expected_files.append(target / ".aipass" / "tier0_kernel.md")
expected_files.append(target / ".aipass" / "tier1_navmap.md")
for f in expected_files:
assert f.exists(), f"Expected file not created: {f}"
assert not (target / ".aipass" / "aipass_global_prompt.md").exists(), "Retired global prompt should NOT be seeded"
# src/<package>/ is a directory with __init__.py
assert (target / "src" / "demo").is_dir(), "Expected src/demo/ package directory"
@@ -126,7 +130,7 @@ def test_init_project_creates_all_expected_files(tmp_path):
created_basenames = [Path(f).name for f in result["created_files"]]
for f in expected_files:
assert f.name in created_basenames or f.exists(), f"Expected {f.name} in created_files"
assert len(result["created_files"]) >= 11
assert len(result["created_files"]) >= 10
def test_init_project_return_dict_structure(tmp_path):
@@ -288,19 +292,6 @@ def test_init_project_settings_no_hooks(tmp_path):
assert "permissions" in data
def test_init_project_global_prompt_content(tmp_path):
"""Global prompt contains project name and AIPass terminology."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="alpha")
content = (target / ".aipass" / "aipass_global_prompt.md").read_text(encoding="utf-8")
assert "# ALPHA" in content
assert "ALPHA_REGISTRY.json" in content
assert "# Commands" in content
def test_init_project_readme_md_content(tmp_path):
"""README.md contains getting started guide with project name."""
target = tmp_path / "proj"
@@ -324,7 +315,7 @@ def test_init_project_auto_creates_target_dir(tmp_path):
assert target.is_dir()
assert result["project_name"] == "NESTED"
assert len(result["created_files"]) >= 11
assert len(result["created_files"]) >= 10
def test_init_project_defaults_name_from_directory(tmp_path):
@@ -359,7 +350,6 @@ def test_init_project_skips_existing_optional_files(tmp_path):
# Pre-create optional files
aipass_dir = target / ".aipass"
aipass_dir.mkdir()
(aipass_dir / "aipass_global_prompt.md").write_text("# Custom global\n", encoding="utf-8")
(target / "CLAUDE.md").write_text("# Custom CLAUDE\n", encoding="utf-8")
(target / "AGENTS.md").write_text("# Custom AGENTS\n", encoding="utf-8")
(target / "README.md").write_text("# Custom README\n", encoding="utf-8")
@@ -542,10 +532,12 @@ def test_update_project_creates_missing_managed_dirs(tmp_path):
result = update_project(target)
assert (target / ".aipass" / "aipass_global_prompt.md").exists()
assert (target / ".claude" / "settings.json").exists()
# Managed files in deleted dirs re-written (global_prompt, hooks.json, settings, prep)
assert len(result["updated_files"]) == 4
# Managed files in deleted dirs re-written (tier0_kernel, tier1_navmap, hooks.json, settings, prep)
if result["aipass_home"]:
assert len(result["updated_files"]) == 5
else:
assert len(result["updated_files"]) == 2
assert len(result["already_current"]) >= 2
@@ -850,6 +842,154 @@ def test_update_project_hooks_json_already_current(tmp_path):
assert any("hooks.json" in f for f in result["already_current"])
# ---------------------------------------------------------------------------
# Tiered prompt injection tests (FPLAN-0284)
# ---------------------------------------------------------------------------
def test_init_project_creates_tier_files(tmp_path):
"""init_project seeds tier0_kernel.md and tier1_navmap.md when AIPASS_HOME available."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="tiers")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
assert (target / ".aipass" / "tier0_kernel.md").exists()
assert (target / ".aipass" / "tier1_navmap.md").exists()
def test_init_project_tier_files_match_canonical(tmp_path):
"""Tier files in new project match the canonical source exactly."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="canon")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
for tier_file in ("tier0_kernel.md", "tier1_navmap.md"):
canonical = Path(result["aipass_home"]) / ".aipass" / tier_file
if not canonical.exists():
pytest.skip(f"{tier_file} not found in canonical .aipass/")
assert (target / ".aipass" / tier_file).read_bytes() == canonical.read_bytes()
def test_init_project_tier_files_in_created_list(tmp_path):
"""Tier files appear in created_files list."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="listed")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
assert any("tier0_kernel.md" in f for f in result["created_files"])
assert any("tier1_navmap.md" in f for f in result["created_files"])
def test_init_project_no_tier_files_without_aipass_home(tmp_path, monkeypatch):
"""Without AIPASS_HOME, tier files are not created."""
target = tmp_path / "proj"
target.mkdir()
monkeypatch.setattr(
"aipass.aipass.apps.handlers.init.bootstrap._detect_aipass_home",
lambda: None,
)
init_project(target, project_name="notiers")
assert not (target / ".aipass" / "tier0_kernel.md").exists()
assert not (target / ".aipass" / "tier1_navmap.md").exists()
def test_init_project_hooks_json_has_tiers_enabled(tmp_path):
"""hooks.json from template has tier0_kernel and navmap enabled, no global_prompt."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="hookstier")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
hooks_json = target / ".aipass" / "hooks.json"
data = json.loads(hooks_json.read_text(encoding="utf-8"))
ups = data["UserPromptSubmit"]
assert ups["tier0_kernel"]["enabled"] is True
assert ups["navmap"]["enabled"] is True
assert "global_prompt" not in ups
def test_update_project_adds_tier_files_to_existing(tmp_path):
"""update_project adds tier files to a project that lacks them."""
target = tmp_path / "proj"
target.mkdir()
registry_data = {
"metadata": {
"id": "test-id",
"name": "OLD",
"version": "1.0.0",
"created": "2026-01-01",
"last_updated": "2026-01-01",
"total_branches": 0,
},
"branches": [],
}
(target / "OLD_REGISTRY.json").write_text(json.dumps(registry_data), encoding="utf-8")
(target / ".aipass").mkdir()
result = update_project(target)
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
assert (target / ".aipass" / "tier0_kernel.md").exists()
assert (target / ".aipass" / "tier1_navmap.md").exists()
assert any("tier0_kernel.md" in f for f in result["updated_files"])
assert any("tier1_navmap.md" in f for f in result["updated_files"])
def test_update_project_tier_files_already_current(tmp_path):
"""update reports tier files as already_current when unchanged."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="tiercurr")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
result = update_project(target)
assert any("tier0_kernel.md" in f for f in result["already_current"])
assert any("tier1_navmap.md" in f for f in result["already_current"])
def test_update_project_refreshes_stale_tier_files(tmp_path):
"""update overwrites tier files when they differ from canonical source."""
target = tmp_path / "proj"
target.mkdir()
result = init_project(target, project_name="stale")
if result["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
(target / ".aipass" / "tier0_kernel.md").write_text("# stale\n", encoding="utf-8")
result = update_project(target)
assert any("tier0_kernel.md" in f for f in result["updated_files"])
content = (target / ".aipass" / "tier0_kernel.md").read_text(encoding="utf-8")
assert "AIPass" in content
# ---------------------------------------------------------------------------
# scaffold_content — global_prompt_md tests
# ---------------------------------------------------------------------------
+1 -1
View File
@@ -56,7 +56,7 @@ def handle_command(command: str, args: list) -> bool:
show_panels = "--quiet" not in args
logger.info(f"[backup] Running full backup cycle for {project_root}")
# ONE scan shared between both modes (Patrick's Law #1)
# ONE scan shared between both modes (single-scan rule)
config = load_project_config(project_root)
spec = load_spec(project_root)
whitelist_entries = load_whitelist(project_root)
+28
View File
@@ -0,0 +1,28 @@
# Backup System ignore patterns (gitignore-style)
# Lines starting with # are comments. Blank lines are ignored.
# Edit this file to customize. Source defaults: handlers/ignore/patterns.py
.backup/
.git/
.svn/
.hg/
__pycache__/
.pytest_cache/
*.pyc
*.pyo
*.egg-info/
.venv/
venv/
.tox/
node_modules/
.vscode/
.idea/
*.swp
*.swo
.DS_Store
Thumbs.db
build/
dist/
*.log
.ruff_cache/
.coverage
@@ -4,7 +4,7 @@ Breadcrumbs only — details in README, `--help`, `.trinity/`, `DASHBOARD.local.
# Identity
DEVPULSE — Patrick's primary collaborator, orchestration hub. Design, plan, debug, dispatch, track. Build own modules (watchdog, feedback), DPLANs, FPLANs, memories. Venture into other branches to investigate, debug, fix small bugs. Delegate heavy multi-file builds to sub-agents. CWD is identity grounding.
DEVPULSE — the user's primary collaborator, orchestration hub. Design, plan, debug, dispatch, track. Build own modules (watchdog, feedback), DPLANs, FPLANs, memories. Venture into other branches to investigate, debug, fix small bugs. Delegate heavy multi-file builds to sub-agents. CWD is identity grounding.
# How you work
@@ -82,7 +82,7 @@ drone @devpulse watchdog agent @target # Monitor tool, never run_in_b
# Interactive wake — tmux
Gives Patrick an interactive session, distinct from autonomous dispatch. Find the agent via `.trinity/passport.json`; use `dangerouslyDisableSandbox: true`.
Gives User an interactive session, distinct from autonomous dispatch. Find the agent via `.trinity/passport.json`; use `dangerouslyDisableSandbox: true`.
```
tmux new-session -d -s "name" -c "/path/to/branch"
@@ -96,4 +96,4 @@ Compass is the curated truth-store of rated decisions (`good/bad/impressive/inte
- Recall what happened / did we do X → `drone @memory search`.
- At a fork, setting a pattern, or unsure of a convention → `drone @devpulse compass query "topic"` (rating shows per hit).
- A good or bad decision made, or a convention confirmed → `drone @devpulse compass add "context" "decision" --rating good`. Add freely, no asking.
- Patrick fires `/compass <rating> <note>` when he notices a decision — you write the entry from context.
- User fires `/compass <rating> <note>` when he notices a decision — you write the entry from context.
@@ -53,7 +53,7 @@ _BRANCH_ROOT = Path(__file__).resolve().parents[3]
DEFAULT_DB_PATH = _BRANCH_ROOT / "devpulse_json" / "compass" / "compass.db"
VALID_RATINGS = ("good", "bad", "impressive", "interesting")
VALID_SOURCES = ("devpulse", "patrick")
VALID_SOURCES = ("devpulse", "user")
VALID_STATUSES = ("active", "archived")
# Columns we return / surface from the decisions table (everything useful).
@@ -177,7 +177,7 @@ def add_decision(
rating: One of ``good | bad | impressive | interesting``.
note: Optional human observation.
tags: Optional comma-separated tags.
source: ``devpulse`` or ``patrick`` (default ``devpulse``).
source: ``devpulse`` or ``user`` (default ``devpulse``).
db_path: Optional DB path override (tests pass a temp path).
created: Optional ISO date override; defaults to today. This is the
ONLY place a "today" date is stamped.
+2 -2
View File
@@ -63,13 +63,13 @@ HELP_TEXT = """\
compass --help Show this help
[bold]Ratings:[/bold] good | bad | impressive | interesting
[bold]Sources:[/bold] devpulse | patrick
[bold]Sources:[/bold] devpulse | user
[bold]Options (add):[/bold]
--rating R Required. One of the ratings above.
--note "..." Optional human observation.
--tags a,b,c Optional comma-separated tags.
--source S Optional. devpulse (default) or patrick.
--source S Optional. devpulse (default) or user.
[bold]Options (all subcommands):[/bold]
--db PATH Use an alternate SQLite store (testing / power use).
+9 -26
View File
@@ -1,10 +1,8 @@
# Flow — Plan Lifecycle Management
Flow is AIPass's unified plan lifecycle system. Creates, tracks, closes, archives numbered work plans across multiple plan types (FPLAN, DPLAN) via data-driven plugin architecture.
## Commands
```bash
# Commands
`bash
drone @flow create . "Subject" # FPLAN (default)
drone @flow create . "Subject" master # FPLAN master template
drone @flow create . "Design topic" dplan # DPLAN
@@ -14,17 +12,13 @@ drone @flow list open # List open plans (all types)
drone @flow list all # List all plans
drone @flow restore FPLAN-0042 # Reopen closed plan
```
## Architecture
# Architecture
- `apps/flow.py` -- Entry point. Auto-discovers modules `apps/modules/` via `handle_command()` convention.
- `apps/modules/` -- Thin orchestrators. No business logic. Route handlers, display results.
- `apps/handlers/` -- Implementation. Grouped domain: `plan/`, `registry/`, `template/`, `dashboard/`, `mbank/`, `summary/`.
- `templates/` -- Plan type directories. Each subdirectory contains Markdown templates. Registered via `drone @flow register`.
- `flow_json/` -- Registries: per-type plan registries + `template_registry.json` (plan type definitions).
## Plan Type System
- `flow_json/` -- Registries: per-type plan registries + `template_registry.json` (plan type definitions).]
# Plan Type System
Plan types filesystem-driven. Drop directory `.md` templates into `templates/`, register, done:
```bash
drone @flow register testing TPLAN # Register new type
@@ -32,35 +26,24 @@ drone @flow unregister testing # Remove type
drone @flow templates # List registered types
drone @flow scan # Find unregistered directories
```
Discovered runtime `plan_type_loader.py` + `registry_ops.py`. No per-directory JSON config needed.
| Type | Prefix | Registry File | Templates |
|------|--------|---------------|-----------|
| flow_plans | FPLAN | fplan_registry.json | default, master |
| dev_plans | DPLAN | dplan_registry.json | default |
## Critical Files
- `apps/flow.py` -- CLI entry point, module discovery, command routing
- `apps/modules/create_plan.py` -- Plan creation orchestrator
- `apps/modules/close_plan.py` -- Plan closure orchestrator (async post-processing, archival)
- `apps/modules/list_plans.py` -- Multi-registry plan listing
# Critical Files
- `apps/modules/create/close/list_plan.py` -- Plan creation orchestrator
- `apps/handlers/plan/list_ops.py` -- Merges plans all registries display
- `apps/handlers/plan/display.py` -- All formatting functions (prefix-aware)
- `apps/handlers/plan/close_ops.py` -- Close implementation (file ops, registry update, vector intake)
- `apps/handlers/template/plan_type_loader.py` -- Plugin discovery + config resolution
- `apps/handlers/registry/load_registry.py` -- Registry loader (supports per-type registry files)
## Integration Points
#Integration Points
- **aipass.cli** -- Rich console output (`console`, `header`, `success`, `error`, `warning`)
- **aipass.prax** -- System logger
- **aipass.memory** -- Vector intake pipeline plan close
- **aipass.trigger** -- Startup events + branch dashboard updates
## Conventions
#Conventions
- Modules return `True` `handle_command()` when command recognized (even on failure), `False` only "not my command".
- Plan IDs follow `{PREFIX}-{NNNN}_topic_slug_YYYY-MM-DD.md`.
- All file I/O uses `pathlib.Path` + `encoding='utf-8'`.
@@ -9,12 +9,12 @@
## What Are Flow Plans?
Flow Plans (FPLANs) are **building** - autonomous construction: systems, features, modules.
Default Flow Plans (FPLANs) are **building** - autonomous construction: systems, features, modules.
**FPLANs are disposable.** Exist exactly one task. When task complete, close this plan immediately -- do not leave open. Open FPLANs mean unfinished work. Work done = plan done: `drone @flow close {plan_number}`
**FPLANs are disposable.** Exist exactly one task. When task complete, close this plan immediately -- do not leave open. Open FPLANs mean unfinished work. Work done = plan done: `drone @flow close {plan_number}` The plan is never lost. It is strored into our locl and global ,chroma vector db.
**This is NOT for:**
- Research or exploration (use agents directly)
- Research or exploration (use DPLANs and APLANs directly)
- Quick fixes (just do it)
- Discussion or planning (happens before creating FPLAN)
@@ -22,6 +22,7 @@ Flow Plans (FPLANs) are **building** - autonomous construction: systems, feature
- Building features or modules
- Single focused construction tasks
- Sub-plans within master plan
- Mostly use for sub-agents, in some cases may be issued to Agents citizens. Judjmemts calls.
---
@@ -49,6 +50,7 @@ Use dedicated directories - don't scatter files:
| `tools/` | Utility scripts |
| `artifacts/` | Agent outputs |
| `docs/` | Documentation |
| `Sub-agents/`|Projects/AIPass/src/aipass/flow/docs.local/sub_agent_drops|
---
@@ -56,7 +58,7 @@ Use dedicated directories - don't scatter files:
**You are orchestrator, not builder.**
Your 200k context is precious. Burning it on file reads + code writing risks compaction during autonomous work. Agents have clean context - use them for * building.
Your 200k context is precious. Burning it on file reads + code writing risks compaction tool early during autonomous work. Agents have clean context - use them for
| You Do (Orchestrator) | Agents Do (Builders) |
|-----------------------|----------------------|
@@ -46,10 +46,11 @@ Master Plan (roadmap)
```
**How to start:**
1. User provides planning doc or instructions (coordinate @devpulse)
1. User provides planning doc DPLAN or instructions (coordinate @devpulse)
2. Branch manager reads + understands scope
3. Branch manager creates master plan: `drone @flow create . "Build X" master`
3. Branch manager creates master plan: `drone @flow create "Build X" master`
4. Branch manager fills phases, then executes autonomously
5. Devepulse mayprovide a complete plan to you. Always confirm, Alwayd confirm the plan is sound, acucurate
---
@@ -57,7 +58,7 @@ Master Plan (roadmap)
**You are ORCHESTRATOR, not builder.**
Your 200k context is precious. Burning it on file reads + code writing risks compaction during autonomous work. Agents have clean context - use them for ALL building.
Your 200k context is precious. Burning it on file reads + code writing risks auto compaction during autonomous work. Agents have clean context - use them for ALL building. Only devpulse is this accempion, user decideds when to compact. no auto compct for devpulse.
| You Do (Orchestrator) | Agents Do (Builders) |
|-----------------------|----------------------|
@@ -17,7 +17,7 @@ Run by **devpulse** (only branch with git write). Tick each step as you go; fill
> `.trinity/passport.json` in the cwd; running from the repo root fails with "No
> passport found"). Read git (`status`, `log`, `diff`, `rev-parse`) is allowed raw.
> ⚠️ `drone @git` has **no `tag` verb** — pushing the release tag is a MANUAL step
> (Patrick, or raw `git tag`/`push` via `!`). All other writes go through drone.
> (the user, or raw `git tag`/`push` via `!`). All other writes go through drone.
---
@@ -0,0 +1,103 @@
# {plan_number} - {subject} (PROMPT CHANGE)
**Created**: {today}
**Branch**: {location}
**Status**: Active
**Type**: Playbook — Prompt Change SOP
---
## Purpose
Any change to an injected prompt — the kernel, the navmap, a branch-local prompt, identity, a brand-new tier, a cadence tweak, or retiring one — must propagate to **every place that injects it AND every place that seeds it into a fresh install**. This SOP is the checklist that makes sure nothing is half-done. Run by **devpulse** (orchestrator); tick as you go; dispatch the owning branch where a step lives in its core. Born from FPLAN-0284, where the live wiring was fixed but `aipass init` kept handing new projects the retired global prompt for days because the *seed* template was never updated.
---
## The Law (read first — this is the trap)
1. **LIVE ≠ SEEDED.** Editing the repo's own `.aipass/hooks.json` makes the change work *for us* — it does **nothing** for a fresh clone, a new `aipass init` project, or a freshly spawned branch. Those are built from **separate seed sources**. A prompt change is not done until every seed path carries it.
2. **Three wiring layers, each with its own seed.** A prompt only injects if all three agree:
| Layer | Live file | Seeded for fresh installs by |
|---|---|---|
| Handler registration | `.aipass/hooks.json` | git-tracked (travels with clone) **+** `.aipass/project_hooks.json` (what `aipass init` copies) |
| Cadence (period/offset) | `hooks_json/custom_config/cadence_config.json` (machine-local, gitignored) | `cadence.py` `DEFAULTS` (the committed backstop) |
| Event→handler bridge | `~/.claude/settings.json` (machine-local) | `setup.sh` **+** `.claude/provider_manifest.json` |
3. **Never delete a retired prompt.** Mark `(superseded)` in its header / disable it in `hooks.json`, or move it to `.archive/`. If a disabled handler still reads it **by path** for rollback, leave the file exactly where the handler looks.
4. **The `.md` files travel by git; the wiring does not.** Prompt/tier `.md` files are tracked → a clone gets them free. The machine-local wiring (cadence_config, settings bridge) must be regenerated from the committed seed sources, so those seeds are what you must update.
5. **One prompt source, every runtime — retire for one = retire for all.** There is ONE source of prompt truth (the tier `.md` files); both runtimes inject *the same content*, only delivery differs: **Claude Code** tiers by cadence (`UserPromptSubmit`, per turn — tier0 every turn, navmap periodically); **Codex CLI** injects the combined tiers once at **SessionStart** (`.codex/hooks/session_start_identity.py`). A change or retirement is NOT done until *both* runtimes point at the new content. (FPLAN-0284 retired the 8k global for Claude via cadence, but Codex's SessionStart kept reading the old global for days until separately rewired — the classic "fixed one runtime, forgot the other.")
---
## 1. Identify the change
- [ ] Which prompt? (`.aipass/tier0_kernel.md` / `.aipass/tier1_navmap.md` / a branch's `.aipass/aipass_local_prompt.md` / identity / a NEW prompt / retiring one)
- [ ] Kind? edit content · change cadence · add a new injected prompt · retire one
- [ ] Owners of the files you'll touch — devpulse owns the tier files + project prompts; **@hooks** owns the engine, handlers, cadence, **and the Codex SessionStart hook** (`.codex/hooks/`); **@aipass** owns init/bootstrap; **@spawn** owns branch templates
## 2. Content
- [ ] Edit the `.md`. Follow `.aipass/PROMPT_STYLE.md`.
- [ ] ⚠️ Size caps: tier0 kernel target **< 2,000 chars**; navmap **< 8,000** (the hook truncates near 10k). To shrink, cut CONTENT, not whitespace — newlines are nearly free, so trimming spaces saves almost nothing.
- [ ] ⚠️ An unclosed `<!-- comment` swallows the whole prompt downstream — confirm every comment is closed.
## 3. Wiring — make it inject LIVE (3 layers)
- [ ] **Registration** — `.aipass/hooks.json`: handler entry present + `"enabled": true` (or `false` + a `_retired` note if retiring). New prompt → add the handler entry.
- [ ] **Cadence** — `cadence_config.json` loaders: correct `period`/`offset` (tier0 `period:1` = every turn; navmap `period:5`). New loader → add it.
- [ ] **Bridge (Claude Code)** — `~/.claude/settings.json`: a `UserPromptSubmit:<handler>` bridge line exists (machine-local — edit directly for the live machine). Retiring → remove the stale line.
- [ ] **Codex runtime** — `.codex/hooks/session_start_identity.py` reads the tier `.md` files at SessionStart (tier0_kernel + tier1_navmap, each size-capped). New tier → add it to what this hook reads; retiring → stop reading the old prompt. Same content as Claude, different delivery. **@hooks owns — dispatch.**
- [ ] New handler file lives at `src/aipass/hooks/apps/handlers/prompt/<name>.py` — **@hooks owns it; dispatch them to build it.**
## 4. Seed propagation — so FRESH installs get it (the step that's always forgotten)
- [ ] **Fresh clone** — `setup.sh`: writes the settings bridge. Grep it for the handler name; add the `UserPromptSubmit:<handler>` line; remove any retired-handler bridge.
- [ ] **Doctor source-of-truth** — `.claude/provider_manifest.json`: mirrors setup.sh's bridge list. Add/retire to match.
- [ ] **Cadence backstop** — `cadence.py` `DEFAULTS`: contains the loader (so a clone with no `cadence_config.json` still fires it). Remove retired loaders' crumbs.
- [ ] **`aipass init` (new projects)** — ⚠️ TWO files, both easy to miss:
- `.aipass/project_hooks.json` (the template init copies into new projects) — mirror the live `hooks.json` (right handlers enabled, retired ones off/removed).
- `src/aipass/aipass/apps/handlers/init/bootstrap.py` — seeds the prompt `.md` files into the new project's `.aipass/`. New tier → it must copy that `.md`; retiring → stop seeding the old one. **@aipass owns — dispatch.**
- [ ] **`@spawn` (new branches)** — branch local-prompt template `src/aipass/spawn/templates/*/.aipass/aipass_local_prompt.md` + any stale doc-prose referencing old prompts. **@spawn owns — dispatch if more than cosmetic.**
- [ ] **Codex for new installs** — if a fresh project/clone runs Codex, confirm its SessionStart hook reads the seeded tier files, not a retired prompt (the Codex entry point `AGENTS.md` is generated by `bootstrap.py`). **@hooks/@aipass own — dispatch.**
## 5. Verify LIVE (don't trust the edit)
- [ ] Watch it actually inject — right content, right turns. Tier0 every turn; periodic tiers on their cadence.
- [ ] After a `/compact`, confirm the prompt you expect reloads (post-compaction is exactly when the map matters most).
- [ ] **seedgo** the touched branches to 100%. ⚠️ importlib-dispatched prompt handlers trip dead-code / unused-function / json-structure false-positives — add **3 `bypass.json` entries per new handler**, mirroring an existing prompt handler (DPLAN-0191).
## 6. Tidy superseded prompts
- [ ] Disable in `hooks.json` (`enabled:false` + `_retired` note pointing at the replacement).
- [ ] Mark the file `(superseded-<plan>)` in its header, or move to `.archive/` — **but** if a disabled handler reads it by path for rollback, leave it in place.
- [ ] Archive pure snapshots/backups + stale design docs. Never delete.
## 7. CHANGELOG + wrap
- [ ] `CHANGELOG.md`: entry under the dated section (Added / Changed / Fixed).
- [ ] Update `.trinity/` memories.
- [ ] Fill the Run Summary below; close (vectorizes the run to @memory).
---
## Run Summary
Fill as you go — this becomes the vectorized trail.
- **Date:** {today}
- **Prompt(s) changed:**
- **Layers touched:** (hooks.json · cadence · bridge · setup.sh · manifest · DEFAULTS · project_hooks · bootstrap · spawn)
- **Dispatched to:** (@hooks / @aipass / @spawn …)
- **Verified live:** (how — which turns, post-compaction reload)
- **Seedgo:** (touched branches @ 100%?)
- **Issues hit:**
- **Notes for next run:** (refine this SOP — what was missing or wrong?)
---
## Listen (TTS-friendly summary)
This playbook is the checklist for changing any injected prompt in AIPass. The big lesson it captures is that live is not the same as seeded. When you change a prompt, editing the repository's own hook configuration makes it work for us right now, but it does nothing for a fresh clone, a newly initialised project, or a newly spawned branch, because those are all built from separate seed files. So the checklist walks you through three things. First, the content of the prompt itself, watching the size limits. Second, the three wiring layers that make it inject live, which are the handler registration, the cadence timing, and the event bridge. Third, and most important, every seed path that a fresh install is built from, including the main setup script, the doctor manifest, the cadence defaults, the two files behind the aipass init flow, and the spawn templates. There are also two runtimes — Claude Code and Codex — both injecting the same prompt content, so retiring or changing a prompt for one means doing it for the other; Claude reads it on a per-turn cadence, Codex reads it once at session start. Then you verify it actually injects, including right after a compaction, tidy away any retired prompt without deleting it, and update the changelog. Run it any time a prompt changes so nothing is left half wired.
---
## Close Command
When all steps are ticked and the Run Summary is filled:
```bash
drone @flow close {plan_number}
```
@@ -42,7 +42,8 @@ apps/
claude.py # Claude Code bridge (provider settings entry point)
prompt/ # Prompt injection hooks
branch_loader.py # Injects aipass_local_prompt.md
global_loader.py # Injects global prompt
tier0_kernel.py # Injects tier0 kernel prompt (every turn)
navmap.py # Injects tier1 navmap prompt (periodic)
identity.py # Injects passport identity block
security/ # Enforcement hooks
edit_gate.py # Blocks edits while type errors exist
@@ -70,7 +71,7 @@ tests/ # 15 test files, 244 tests
| Category | Count | Handlers |
|----------|-------|----------|
| prompt | 3 | branch_loader, global_loader, identity |
| prompt | 4 | branch_loader, tier0_kernel, navmap, identity |
| security | 3 | edit_gate, git_gate, subagent_gate |
| lifecycle | 4 | auto_fix, auto_watchdog, compact, rollover |
| notification | 4 | announce, email, stop_sound, tool_sound |
+4 -39
View File
@@ -71,25 +71,10 @@
"standard": "json_structure",
"reason": "No JSON operations \u2014 reads markdown files and outputs text."
},
{
"file": "apps/handlers/prompt/global_loader.py",
"standard": "dead_code",
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.prompt.global_loader.handle' \u2014 not statically imported by design. Verified wired in UserPromptSubmit.global_prompt + fires in engine.jsonl."
},
{
"file": "apps/handlers/prompt/global_loader.py",
"standard": "unused_function",
"reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Verified fires in engine.jsonl (UserPromptSubmit.global_prompt)."
},
{
"file": "apps/handlers/prompt/global_loader.py",
"standard": "json_structure",
"reason": "No JSON operations \u2014 reads markdown file and outputs text."
},
{
"file": "apps/handlers/prompt/tier0_kernel.py",
"standard": "dead_code",
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.prompt.tier0_kernel.handle' - not statically imported by design. Wired in UserPromptSubmit.tier0_kernel (FPLAN-0284). Same pattern as global_loader.py."
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.prompt.tier0_kernel.handle' - not statically imported by design. Wired in UserPromptSubmit.tier0_kernel (FPLAN-0284)."
},
{
"file": "apps/handlers/prompt/tier0_kernel.py",
@@ -104,7 +89,7 @@
{
"file": "apps/handlers/prompt/navmap.py",
"standard": "dead_code",
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.prompt.navmap.handle' - not statically imported by design. Wired in UserPromptSubmit.navmap (FPLAN-0284). Same pattern as global_loader.py."
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.prompt.navmap.handle' - not statically imported by design. Wired in UserPromptSubmit.navmap (FPLAN-0284)."
},
{
"file": "apps/handlers/prompt/navmap.py",
@@ -259,12 +244,12 @@
{
"file": "apps/modules/cadence.py",
"standard": "dead_code",
"reason": "Cadence module \u2014 should_fire() called from global_loader.py and branch_loader.py guard lines; reset_counter() called from compact.py PreCompact handler. Not statically discoverable because callers are themselves dynamically dispatched."
"reason": "Cadence module \u2014 should_fire() called from tier0_kernel.py, navmap.py, and branch_loader.py guard lines; reset_counter() called from compact.py PreCompact handler. Not statically discoverable because callers are themselves dynamically dispatched."
},
{
"file": "apps/modules/cadence.py",
"standard": "unused_function",
"reason": "should_fire() called from global_loader.py + branch_loader.py; reset_counter() called from compact.py. All callers are dynamically dispatched handlers."
"reason": "should_fire() called from tier0_kernel.py, navmap.py, branch_loader.py; reset_counter() called from compact.py. All callers are dynamically dispatched handlers."
},
{
"file": "apps/modules/cadence.py",
@@ -604,26 +589,6 @@
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_global_loader.py",
"standard": "architecture",
"reason": "Test files live in tests/, not in the 3-layer apps structure."
},
{
"file": "tests/test_global_loader.py",
"standard": "documentation",
"reason": "Test methods use descriptive names as documentation per pytest convention."
},
{
"file": "tests/test_global_loader.py",
"standard": "encapsulation",
"reason": "Tests import handlers directly to test implementation details."
},
{
"file": "tests/test_global_loader.py",
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_compact.py",
"standard": "architecture",
+3 -2
View File
@@ -58,7 +58,8 @@ src/aipass/hooks/
│ │ │ └── claude.py # Claude Code bridge
│ │ ├── prompt/ # Prompt injection hooks
│ │ │ ├── branch_loader.py # Injects aipass_local_prompt.md
│ │ │ ├── global_loader.py # Injects global prompt
│ │ │ ├── tier0_kernel.py # Injects tier0 kernel prompt (every turn)
│ │ │ ├── navmap.py # Injects tier1 navmap prompt (periodic)
│ │ │ └── identity.py # Injects passport identity block
│ │ ├── security/ # Enforcement hooks
│ │ │ ├── edit_gate.py # Blocks unsafe edits (cross-branch, inbox, diagnostics)
@@ -102,7 +103,7 @@ Handlers are called **dynamically at runtime** — the engine uses `importlib.im
| Event | Hooks | Description |
|---|---|---|
| UserPromptSubmit | identity, email, branch_loader, global_loader | Prompt injection + inbox check |
| UserPromptSubmit | identity, email, branch_loader, tier0_kernel, navmap | Prompt injection + inbox check |
| PreToolUse | tool_sound, edit_gate, git_gate, rm_gate | Security gates + guardrails + sound |
| PostToolUse | auto_fix, auto_watchdog | Diagnostics + watchdog |
| SubagentStop | subagent_gate | Seedgo validation |
@@ -40,13 +40,15 @@ def _read_last_session(branch_dir: Path) -> str | None:
sessions = data.get("sessions", [])
if sessions:
last = sessions[0]
result.append(
f"Last session (#{last.get('id', '?')}, {last.get('d', '?')}): {last.get('sum', 'no summary')}"
)
learnings = data.get("key_learnings", {})
if learnings:
keys = list(learnings.keys())[-10:]
result.append(f"Key learnings available: {', '.join(keys)}")
sid = last.get("number", "?")
sdate = last.get("date", "?")
ssum = last.get("summary", "no summary")
result.append(f"Last session (#{sid}, {sdate}): {ssum}")
learnings = data.get("key_learnings", [])
if isinstance(learnings, list) and learnings:
keys = [entry.get("key", "?") for entry in learnings[-10:] if isinstance(entry, dict)]
if keys:
result.append(f"Key learnings available: {', '.join(keys)}")
return "\n".join(result) if result else None
except Exception as exc:
logger.info("[HOOKS] compact: read session failed: %s", exc)
@@ -85,7 +87,7 @@ def handle(hook_data: dict) -> dict:
import importlib
cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
cadence.reset_counter()
cadence.reset_counter(hook_data=hook_data)
except Exception as exc:
logger.info("[HOOKS] compact: cadence reset failed: %s", exc)
@@ -1,16 +1,15 @@
# =================== AIPass ====================
# Name: rollover.py
# Version: 1.0.0
# Description: Checks branch memory files and runs rollover if overdue (PreCompact)
# Version: 2.0.0
# Description: Triggers memory rollover via @memory when files are overdue (PreCompact)
# Branch: hooks
# Layer: apps/handlers/lifecycle
# Created: 2026-05-22
# Modified: 2026-05-22
# Modified: 2026-06-19
# =============================================
"""Scans all branches for over-limit memory files and triggers rollover via drone."""
"""Delegates rollover detection to @memory and triggers rollover if overdue."""
import json
import os
import subprocess
from pathlib import Path
@@ -31,83 +30,24 @@ def _find_repo_root() -> Path | None:
return None
def _read_registry(repo_root: Path) -> list[dict]:
registry_path = repo_root / "AIPASS_REGISTRY.json"
if not registry_path.exists():
return []
def _run_check(repo_root: Path) -> tuple[bool, str]:
try:
data = json.loads(registry_path.read_text(encoding="utf-8"))
branches = data.get("branches", [])
for branch in branches:
raw_path = branch.get("path", "")
resolved = Path(raw_path)
if not resolved.is_absolute():
resolved = repo_root / raw_path
branch["_resolved_path"] = resolved
return branches
result = subprocess.run(
["drone", "@memory", "rollover", "check"],
capture_output=True,
text=True,
timeout=30,
cwd=str(repo_root),
)
stdout = result.stdout.strip()
has_overdue = "ready for rollover" in stdout.lower()
return has_overdue, stdout
except subprocess.TimeoutExpired:
logger.warning("[HOOKS] rollover: check timed out (30s)")
return False, "check timed out"
except Exception as exc:
logger.info("[HOOKS] rollover: registry read failed: %s", exc)
return []
def _check_file(file_path: Path) -> tuple[bool, str]:
if not file_path.is_file():
return False, ""
try:
raw = file_path.read_text(encoding="utf-8")
data = json.loads(raw)
except Exception as exc:
logger.info("[HOOKS] rollover: file parse failed %s: %s", file_path, exc)
return False, ""
limits = data.get("document_metadata", {}).get("limits", {})
has_v2_limits = any(k in limits for k in ("max_sessions", "max_key_learnings", "max_observations"))
if has_v2_limits:
reasons: list[str] = []
max_sessions = limits.get("max_sessions")
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")
if max_key_learnings is not None:
key_learnings = data.get("key_learnings", {})
if isinstance(key_learnings, dict) and len(key_learnings) >= max_key_learnings:
reasons.append(f"{len(key_learnings)}/{max_key_learnings} learnings")
max_observations = limits.get("max_observations")
if max_observations is not None:
observations = data.get("observations", [])
if isinstance(observations, list) and len(observations) >= max_observations:
reasons.append(f"{len(observations)}/{max_observations} observations")
if reasons:
return True, ", ".join(reasons)
return False, ""
max_lines = limits.get("max_lines", 600)
current_lines = raw.count("\n") + 1
if current_lines >= max_lines:
return True, f"{current_lines}/{max_lines} lines"
return False, ""
def _find_overdue(repo_root: Path) -> list[tuple[str, str, str]]:
branches = _read_registry(repo_root)
overdue: list[tuple[str, str, str]] = []
for branch in branches:
name = branch.get("name", "unknown")
branch_path = branch.get("_resolved_path")
if not branch_path or not branch_path.is_dir():
continue
for memory_type in ("local", "observations"):
file_path = branch_path / ".trinity" / f"{memory_type}.json"
is_overdue, reason = _check_file(file_path)
if is_overdue:
overdue.append((name, memory_type, reason))
return overdue
logger.warning("[HOOKS] rollover: check failed: %s", exc)
return False, str(exc)
def _run_rollover(repo_root: Path) -> tuple[bool, str]:
@@ -121,35 +61,30 @@ def _run_rollover(repo_root: Path) -> tuple[bool, str]:
)
return result.returncode == 0, result.stdout + result.stderr
except subprocess.TimeoutExpired:
logger.info("[HOOKS] rollover: drone rollover timed out (110s)")
logger.warning("[HOOKS] rollover: drone rollover timed out (110s)")
return False, "Rollover timed out (110s)"
except Exception as exc:
logger.info("[HOOKS] rollover: drone rollover failed: %s", exc)
logger.warning("[HOOKS] rollover: drone rollover failed: %s", exc)
return False, str(exc)
def handle(hook_data: dict) -> dict: # noqa: ARG001
"""Check memory files for overflow and trigger rollover if needed."""
try:
repo_root = _find_repo_root()
if not repo_root:
return {"stdout": "", "exit_code": 0}
overdue = _find_overdue(repo_root)
if not overdue:
return {"stdout": "", "exit_code": 0}
summary = "; ".join(f"{name}.{mtype} ({reason})" for name, mtype, reason in overdue)
logger.info("[HOOKS] rollover: %d overdue — %s", len(overdue), summary)
success, output = _run_rollover(repo_root)
if success:
logger.info("[HOOKS] rollover: complete (%d files processed)", len(overdue))
else:
logger.info("[HOOKS] rollover: failed — %s", output[:200])
return {"stdout": "", "exit_code": 0, "sound": "pre compact rollover"}
except Exception as exc:
logger.info("[HOOKS] rollover: unexpected error: %s", exc)
repo_root = _find_repo_root()
if not repo_root:
logger.warning("[HOOKS] rollover: no repo root found — cannot check")
return {"stdout": "", "exit_code": 0}
has_overdue, check_output = _run_check(repo_root)
if not has_overdue:
return {"stdout": "", "exit_code": 0}
logger.info("[HOOKS] rollover: overdue files detected — %s", check_output.replace("\n", " | "))
success, output = _run_rollover(repo_root)
if success:
logger.info("[HOOKS] rollover: complete")
else:
logger.warning("[HOOKS] rollover: FAILED — %s", output[:300])
return {"stdout": "", "exit_code": 0, "sound": "pre compact rollover"}
@@ -1,59 +0,0 @@
# =================== AIPass ====================
# Name: global_loader.py
# Version: 1.0.0
# Description: Loads AIPass global prompt for injection (UserPromptSubmit)
# Branch: hooks
# Layer: apps/handlers/prompt
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Loads .aipass/aipass_global_prompt.md from AIPASS_HOME for prompt injection."""
import os
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
def _find_project_prompt() -> Path | None:
"""Walk up from CWD to find the nearest .aipass/aipass_global_prompt.md."""
cwd = Path.cwd()
for parent in [cwd, *cwd.parents]:
candidate = parent / ".aipass" / "aipass_global_prompt.md"
if candidate.is_file():
return candidate
if parent == parent.parent:
break
return None
def handle(hook_data: dict) -> dict:
"""Load global prompt — project-local if outside AIPass, AIPass-internal if inside."""
try:
import importlib
cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
if not cadence.should_fire("global", hook_data):
return {"stdout": "", "exit_code": 0}
except Exception as exc:
logger.info("[HOOKS] global_loader: cadence check failed, firing anyway: %s", exc)
try:
aipass_home = os.environ.get("AIPASS_HOME", "")
cwd = str(Path.cwd())
if aipass_home and cwd.startswith(aipass_home):
prompt_file = Path(aipass_home) / ".aipass" / "aipass_global_prompt.md"
else:
prompt_file = _find_project_prompt()
if not prompt_file or not prompt_file.exists():
return {"stdout": "", "exit_code": 0}
content = prompt_file.read_text(encoding="utf-8")
return {"stdout": content, "exit_code": 0, "sound": "global prompt"}
except Exception as exc:
logger.info("[HOOKS] global_loader: unexpected error: %s", exc)
return {"stdout": "", "exit_code": 0}
+26 -4
View File
@@ -46,7 +46,6 @@ DEFAULTS = {
"loaders": {
"tier0": {"period": 1},
"navmap": {"period": 5, "offset": 0},
"global": {"offset": 0},
"branch": {"offset": 0},
},
}
@@ -227,10 +226,21 @@ def should_fire(loader_name: str, hook_data: dict | None = None) -> bool:
return fired
def reset_counter() -> None:
def reset_counter(hook_data: dict | None = None) -> None:
"""Reset counter to -1 so next turn reads 0 (all loaders fire). Called from PreCompact."""
path = _state_path()
session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "")
if path is None and hook_data:
fallback_id = hook_data.get("session_id", "")
if fallback_id:
path = _GUARD_DIR / f"aipass-cadence-{fallback_id}.json"
session_id = fallback_id
session_short = session_id[:8] if session_id else "none"
if path is None:
logger.info("[HOOKS] cadence: reset_counter SKIPPED — no session ID (env CLAUDE_CODE_SESSION_ID not set)")
return
fd = None
@@ -238,14 +248,26 @@ def reset_counter() -> None:
fd = open(path, "a+") # noqa: SIM115
_lock(fd)
fd.seek(0)
content = fd.read()
old_turn = -1
if content.strip():
try:
old_turn = json.loads(content).get("turn", -1)
except json.JSONDecodeError as exc:
logger.info("[HOOKS] cadence: reset read old state failed: %s", exc)
fd.seek(0)
fd.truncate()
fd.write(json.dumps({"turn": -1, "token": -1}))
fd.flush()
_close_fd(fd)
fd = None
logger.info("[HOOKS] cadence: counter reset for post-compact re-injection")
logger.info(
"[HOOKS] cadence: counter reset for post-compact re-injection session=%s prev_turn=%d",
session_short,
old_turn,
)
except OSError as exc:
logger.info("[HOOKS] cadence: reset write failed: %s", exc)
logger.info("[HOOKS] cadence: reset write FAILED session=%s: %s", session_short, exc)
if fd is not None:
_close_fd(fd)
+244 -8
View File
@@ -322,8 +322,8 @@ class TestConfig:
assert config["enabled"] is True
assert config["period"] == 5
assert config["loaders"]["global"]["offset"] == 0
assert config["loaders"]["branch"]["offset"] == 0
assert "global" not in config["loaders"]
def test_defaults_include_tiered_loaders(self, tmp_path):
"""Fresh clone with no cadence_config.json gets tiered cadence out of the box."""
@@ -467,22 +467,31 @@ class TestLoaderCadenceGuard:
def setup_method(self):
_reset_module_globals()
def test_global_loader_skips_on_non_fire_turn(self, tmp_path):
"""Skip = empty stdout AND no sound key — a skipped loader is SILENT."""
from aipass.hooks.apps.handlers.prompt.global_loader import handle
def test_tier0_kernel_fires_every_turn(self, tmp_path):
"""tier0 has period:1 — fires on every turn including non-fire turns for others."""
from aipass.hooks.apps.handlers.prompt.tier0_kernel import handle
_write_state(tmp_path, turn=0) # next turn = 1 = skip
config = tmp_path / "cadence.json"
config.write_text(
json.dumps(
{
"enabled": True,
"period": 5,
"loaders": {"tier0": {"period": 1}},
}
)
)
_write_state(tmp_path, turn=2)
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
patch(f"{MODULE}._CONFIG_PATH", config),
):
result = handle({})
assert result["stdout"] == ""
assert result["exit_code"] == 0
assert "sound" not in result
def test_branch_loader_skips_on_non_fire_turn(self, tmp_path):
"""Skip = empty stdout AND no sound key — a skipped loader is SILENT."""
@@ -502,6 +511,233 @@ class TestLoaderCadenceGuard:
assert "sound" not in result
class TestResetCounterObservability:
"""Tests for reset_counter fail-loud logging and session ID tracking."""
def setup_method(self):
_reset_module_globals()
def test_reset_logs_session_id_and_prev_turn(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
_write_state(tmp_path, turn=11)
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}.logger") as mock_logger,
):
reset_counter()
reset_calls = [c for c in mock_logger.info.call_args_list if "post-compact re-injection" in str(c)]
assert len(reset_calls) == 1
fmt_str = reset_calls[0][0][0]
fmt_args = reset_calls[0][0][1:]
log_line = fmt_str % fmt_args
assert "session=test-ses" in log_line
assert "prev_turn=11" in log_line
def test_reset_no_session_logs_warning(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch(f"{MODULE}.logger") as mock_logger,
):
env = dict(os.environ)
env.pop("CLAUDE_CODE_SESSION_ID", None)
with patch.dict("os.environ", env, clear=True):
reset_counter()
calls = [str(c) for c in mock_logger.info.call_args_list]
warning_calls = [c for c in calls if "SKIPPED" in c]
assert len(warning_calls) == 1
def test_reset_fallback_to_hook_data_session_id(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
state_file = tmp_path / "aipass-cadence-fallback-id.json"
state_file.write_text(json.dumps({"turn": 5}))
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
):
env = dict(os.environ)
env.pop("CLAUDE_CODE_SESSION_ID", None)
with patch.dict("os.environ", env, clear=True):
reset_counter(hook_data={"session_id": "fallback-id"})
data = json.loads(state_file.read_text())
assert data["turn"] == -1
def test_reset_fallback_creates_file_if_missing(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
):
env = dict(os.environ)
env.pop("CLAUDE_CODE_SESSION_ID", None)
with patch.dict("os.environ", env, clear=True):
reset_counter(hook_data={"session_id": "new-fallback"})
state_file = tmp_path / "aipass-cadence-new-fallback.json"
assert state_file.exists()
assert json.loads(state_file.read_text())["turn"] == -1
def test_reset_env_takes_priority_over_hook_data(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
env_file = tmp_path / "aipass-cadence-env-session.json"
env_file.write_text(json.dumps({"turn": 9}))
hook_file = tmp_path / "aipass-cadence-hook-session.json"
hook_file.write_text(json.dumps({"turn": 3}))
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "env-session"}),
):
reset_counter(hook_data={"session_id": "hook-session"})
assert json.loads(env_file.read_text())["turn"] == -1
assert json.loads(hook_file.read_text())["turn"] == 3
def test_reset_hook_data_empty_session_id_logs_skip(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch(f"{MODULE}.logger") as mock_logger,
):
env = dict(os.environ)
env.pop("CLAUDE_CODE_SESSION_ID", None)
with patch.dict("os.environ", env, clear=True):
reset_counter(hook_data={"session_id": ""})
calls = [str(c) for c in mock_logger.info.call_args_list]
assert any("SKIPPED" in c for c in calls)
def test_reset_prev_turn_from_corrupt_file(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
state_file = tmp_path / "aipass-cadence-test-session.json"
state_file.write_text("not valid json{{{")
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
):
reset_counter()
data = json.loads(state_file.read_text())
assert data["turn"] == -1
class TestPostCompactDeterminism:
"""Prove that post-compaction reload fires ALL loaders deterministically."""
def setup_method(self):
_reset_module_globals()
def test_all_tiered_loaders_fire_after_reset(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter, should_fire
config = tmp_path / "cadence.json"
config.write_text(
json.dumps(
{
"enabled": True,
"period": 5,
"loaders": {
"tier0": {"period": 1},
"navmap": {"period": 5, "offset": 0},
"branch": {"offset": 0},
},
}
)
)
_write_state(tmp_path, turn=11)
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}._CONFIG_PATH", config),
):
reset_counter()
_reset_module_globals()
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}._CONFIG_PATH", config),
):
assert should_fire("tier0") is True
_reset_module_globals()
assert should_fire("navmap") is True
_reset_module_globals()
assert should_fire("branch") is True
def test_reset_at_any_turn_produces_turn_zero(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter, should_fire
for prev_turn in [0, 1, 4, 5, 10, 11, 99]:
_reset_module_globals()
_write_state(tmp_path, turn=prev_turn)
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
):
reset_counter()
_reset_module_globals()
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
):
result = should_fire("navmap")
state = json.loads((tmp_path / "aipass-cadence-test-session.json").read_text())
assert state["turn"] == 0, f"Expected turn 0 after reset from {prev_turn}"
assert result is True, f"navmap should fire after reset from turn {prev_turn}"
def test_compact_handler_calls_reset_with_hook_data(self):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
hook_data = {"cwd": "/tmp/fake", "session_id": "test-123"}
with (
patch("importlib.import_module") as mock_import,
):
mock_cadence = mock_import.return_value
result = handle(hook_data)
mock_cadence.reset_counter.assert_called_once_with(hook_data=hook_data)
assert result["exit_code"] == 0
def test_double_reset_is_idempotent(self, tmp_path):
from aipass.hooks.apps.modules.cadence import reset_counter
_write_state(tmp_path, turn=11)
with (
patch(f"{MODULE}._GUARD_DIR", tmp_path),
patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
):
reset_counter()
_reset_module_globals()
reset_counter()
state_file = tmp_path / "aipass-cadence-test-session.json"
data = json.loads(state_file.read_text())
assert data["turn"] == -1
class TestPerLoaderPeriod:
def setup_method(self):
_reset_module_globals()
+2 -2
View File
@@ -23,8 +23,8 @@ class TestCompactHandler:
local.write_text(
json.dumps(
{
"sessions": [{"id": "S10", "d": "2026-05-22", "sum": "did stuff"}],
"key_learnings": {"learn1": "value1"},
"sessions": [{"number": 10, "date": "2026-05-22", "summary": "did stuff", "status": "completed"}],
"key_learnings": [{"number": 1, "date": "2026-05-22", "key": "learn1", "value": "value1"}],
}
),
encoding="utf-8",
@@ -1,85 +0,0 @@
# =================== AIPass ====================
# Name: test_global_loader.py
# Version: 1.0.0
# Description: Tests for global_loader prompt handler
# Branch: hooks
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Tests for handlers/prompt/global_loader.py."""
from unittest.mock import patch, MagicMock
def _mock_cadence_fires():
"""Return a mock cadence module where should_fire always returns True."""
mock = MagicMock()
mock.should_fire.return_value = True
return mock
class TestGlobalLoaderHandler:
def test_loads_global_prompt(self, tmp_path, monkeypatch):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
aipass_dir = tmp_path / ".aipass"
aipass_dir.mkdir()
prompt = aipass_dir / "aipass_global_prompt.md"
prompt.write_text("# AIPass Global\nContext here", encoding="utf-8")
monkeypatch.chdir(tmp_path)
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert "AIPass Global" in result["stdout"]
assert "Context here" in result["stdout"]
assert result["sound"] == "global prompt"
def test_returns_empty_when_file_missing(self, tmp_path, monkeypatch):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
monkeypatch.chdir(tmp_path)
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
assert "sound" not in result
def test_empty_hook_data(self, tmp_path, monkeypatch):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
aipass_dir = tmp_path / ".aipass"
aipass_dir.mkdir()
(aipass_dir / "aipass_global_prompt.md").write_text("content", encoding="utf-8")
monkeypatch.chdir(tmp_path)
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == "content"
def test_external_project_gets_own_prompt(self, tmp_path, monkeypatch):
"""CWD outside AIPASS_HOME → loads project-local prompt, not AIPass."""
from aipass.hooks.apps.handlers.prompt.global_loader import handle
project = tmp_path / "my-project"
project.mkdir()
aipass_dir = project / ".aipass"
aipass_dir.mkdir()
(aipass_dir / "aipass_global_prompt.md").write_text("# Project Prompt", encoding="utf-8")
monkeypatch.chdir(project)
with patch.dict("os.environ", {"AIPASS_HOME": "/some/other/path"}):
with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert "Project Prompt" in result["stdout"]
assert result["sound"] == "global prompt"
+83 -56
View File
@@ -1,23 +1,43 @@
# =================== AIPass ====================
# Name: test_rollover.py
# Version: 1.0.0
# Version: 2.0.0
# Description: Tests for rollover lifecycle handler
# Branch: hooks
# Created: 2026-05-22
# Modified: 2026-05-22
# Modified: 2026-06-19
# =============================================
"""Tests for handlers/lifecycle/rollover.py."""
import json
from unittest.mock import patch, MagicMock
import subprocess
MOD = "aipass.hooks.apps.handlers.lifecycle.rollover"
CHECK_OVERDUE_OUTPUT = (
"Found 3 files ready for rollover:\n"
" * HOOKS.local (15/15 sessions)\n"
" * aipass.local (15/15 key_learnings)\n"
" * devpulse.local (15/15 key_learnings)\n"
)
CHECK_CLEAN_OUTPUT = "No files need rollover.\n"
def _mock_run(stdout="", returncode=0):
m = MagicMock()
m.stdout = stdout
m.stderr = ""
m.returncode = returncode
return m
class TestRolloverHandler:
def test_no_repo_root_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=None):
with patch(f"{MOD}._find_repo_root", return_value=None):
result = handle({})
assert result["exit_code"] == 0
@@ -27,8 +47,8 @@ class TestRolloverHandler:
def test_no_overdue_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue", return_value=[]):
with patch(f"{MOD}._find_repo_root", return_value=MagicMock()):
with patch(f"{MOD}._run_check", return_value=(False, CHECK_CLEAN_OUTPUT)):
result = handle({})
assert result["exit_code"] == 0
@@ -38,68 +58,75 @@ class TestRolloverHandler:
def test_overdue_triggers_rollover(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
with patch(
"aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue",
return_value=[("devpulse", "local", "21/20 sessions")],
):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._run_rollover", return_value=(True, "ok")):
with patch(f"{MOD}._find_repo_root", return_value=MagicMock()):
with patch(f"{MOD}._run_check", return_value=(True, CHECK_OVERDUE_OUTPUT)):
with patch(f"{MOD}._run_rollover", return_value=(True, "ok")):
result = handle({})
assert result["exit_code"] == 0
assert result["sound"] == "pre compact rollover"
def test_check_file_v2_sessions_overdue(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.rollover import _check_file
def test_overdue_rollover_failure_still_returns_sound(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
f = tmp_path / "local.json"
f.write_text(
json.dumps(
{
"document_metadata": {"limits": {"max_sessions": 5}},
"sessions": [{"id": i} for i in range(6)],
}
),
encoding="utf-8",
)
with patch(f"{MOD}._find_repo_root", return_value=MagicMock()):
with patch(f"{MOD}._run_check", return_value=(True, CHECK_OVERDUE_OUTPUT)):
with patch(f"{MOD}._run_rollover", return_value=(False, "error")):
result = handle({})
overdue, reason = _check_file(f)
assert overdue
assert "6/5" in reason
assert result["exit_code"] == 0
assert result["sound"] == "pre compact rollover"
def test_check_file_v2_not_overdue(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.rollover import _check_file
def test_run_check_parses_overdue(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_check
f = tmp_path / "local.json"
f.write_text(
json.dumps(
{
"document_metadata": {"limits": {"max_sessions": 20}},
"sessions": [{"id": i} for i in range(5)],
}
),
encoding="utf-8",
)
mock_result = _mock_run(stdout=CHECK_OVERDUE_OUTPUT)
with patch("subprocess.run", return_value=mock_result):
has_overdue, summary = _run_check(MagicMock())
overdue, _ = _check_file(f)
assert not overdue
assert has_overdue
assert "ready for rollover" in summary.lower()
def test_check_file_v1_line_count(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.rollover import _check_file
def test_run_check_parses_clean(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_check
f = tmp_path / "obs.json"
content = {"document_metadata": {"limits": {"max_lines": 10}}}
text = json.dumps(content, indent=2)
lines_needed = 10 - text.count("\n")
text += "\n" * lines_needed
f.write_text(text, encoding="utf-8")
mock_result = _mock_run(stdout=CHECK_CLEAN_OUTPUT)
with patch("subprocess.run", return_value=mock_result):
has_overdue, _ = _run_check(MagicMock())
overdue, reason = _check_file(f)
assert overdue
assert "lines" in reason
assert not has_overdue
def test_check_file_missing(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.rollover import _check_file
def test_run_check_timeout_returns_false(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_check
overdue, _ = _check_file(tmp_path / "nonexistent.json")
assert not overdue
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 30)):
has_overdue, _ = _run_check(MagicMock())
assert not has_overdue
def test_run_rollover_success(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_rollover
mock_result = _mock_run(stdout="done", returncode=0)
with patch("subprocess.run", return_value=mock_result):
success, _ = _run_rollover(MagicMock())
assert success
def test_run_rollover_failure(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_rollover
mock_result = _mock_run(stdout="error", returncode=1)
with patch("subprocess.run", return_value=mock_result):
success, _ = _run_rollover(MagicMock())
assert not success
def test_run_rollover_timeout(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import _run_rollover
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 110)):
success, msg = _run_rollover(MagicMock())
assert not success
assert "timed out" in msg
@@ -10,11 +10,13 @@
Rollover Trigger Detection Handler
Monitors branch memory files via AIPASS_REGISTRY.json and detects when
files exceed their max_lines threshold (typically 600 lines).
entry counts exceed v2 limits (sessions, key_learnings, observations).
Purpose:
Detect rollover conditions without active monitoring. Called by
rollover module to check all branches for files needing rollover.
All branches use v2 entry-count limits from memory.config.json
(per_branch with defaults fallback). No line-count fallbacks.
Independence:
No module imports - pure handler, transportable
@@ -74,20 +76,17 @@ def _find_caller_registries() -> List[Path]:
@dataclass
class RolloverTrigger:
"""Represents a file that needs rollover"""
"""Represents a file that needs rollover (v2 entry-count based)"""
branch: str
memory_type: str # 'observations' or 'local'
file_path: Path
current_lines: int
max_lines: int
schema_version: str = "1.0.0"
schema_version: str = "3.0.0"
v2_reason: str = ""
def __str__(self):
if self.schema_version.startswith("2") and self.v2_reason:
return f"{self.branch}.{self.memory_type} ({self.v2_reason})"
return f"{self.branch}.{self.memory_type} ({self.current_lines}/{self.max_lines} lines)"
return f"{self.branch}.{self.memory_type} ({self.v2_reason})"
# =============================================================================
@@ -166,16 +165,6 @@ def _get_memory_file_path(branch: Dict, memory_type: str) -> Path | None:
return file_path if file_path.exists() else None
# =============================================================================
# CONFIG LOADING
# =============================================================================
def _load_config() -> Dict[str, Any]:
"""Load memory.config.json via config_loader."""
return config_loader.load()
# =============================================================================
# LINE COUNTING
# =============================================================================
@@ -253,86 +242,35 @@ def _recreate_trinity_file(branch_path: Path, branch_name: str, memory_type: str
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
Args:
file_path: Path to JSON file
branch_name: Optional branch name for config lookup
Returns:
Max lines limit (default 600)
"""
# 1. Try file-level metadata first (highest priority)
try:
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
metadata = data.get("document_metadata", {})
limits = metadata.get("limits", {})
file_limit = limits.get("max_lines")
if file_limit is not None:
return file_limit
except Exception as e:
logger.warning(f"[detector] Failed to read file-level max_lines from {file_path}: {e}")
# 2. Try branch-level config (if branch_name provided or can be extracted)
if branch_name is 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()
branch_limits = config.get("rollover", {}).get("per_branch", {}).get(branch_name, {})
if "max_lines" in branch_limits:
return branch_limits["max_lines"]
# 3. Fall back to global default from config
config = _load_config()
default_limit = config.get("rollover", {}).get("defaults", {}).get("max_lines")
if default_limit is not None:
return default_limit
# 4. Final fallback to hardcoded 600
return 600
# =============================================================================
# ROLLOVER DETECTION
# =============================================================================
def _should_rollover(file_path: Path) -> tuple[bool, int, int, str, str]:
def _should_rollover(file_path: Path) -> tuple[bool, int, str, str]:
"""
Check if file should rollover (supports v1 line-based and v2 entry-count based).
Check if file should rollover (v2 entry-count based only).
All branches use v2 entry-count limits from config (per_branch with
defaults fallback). No line-count fallbacks — errors over silent fallbacks.
Args:
file_path: Path to memory JSON file
Returns:
Tuple of (should_rollover, current_lines, max_lines, schema_version, v2_reason)
For v2 files, max_lines is 0 and v2_reason describes which limits are exceeded.
Tuple of (should_rollover, current_lines, schema_version, v2_reason)
"""
current_lines = _count_file_lines(file_path)
# Read file data once for schema detection + limit checks
# Read file data once for limit checks
try:
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
except Exception as e:
# Can't parse — fall back to line-based with hardcoded default
logger.warning(f"[detector] Failed to parse {file_path} for rollover check: {e}")
return (current_lines >= 600, current_lines, 600, "1.0.0", "")
# Can't parse — do NOT fall back to 600. Fail honestly.
logger.warning(f"[detector] PARSE FAILURE for {file_path}: {e} — skipping rollover check")
return (False, current_lines, "3.0.0", "parse failure — skipped")
metadata = data.get("document_metadata", {})
limits = metadata.get("limits", {})
# 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()
@@ -344,39 +282,46 @@ def _should_rollover(file_path: Path) -> tuple[bool, int, int, str, str]:
# .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, {})
per_branch = cfg.get("per_branch", {})
defaults = cfg.get("defaults", {})
if file_limits:
reasons = []
# v2 lookup: per_branch[branch][file_type], fallback to defaults[file_type]
file_limits = per_branch.get(branch_name, {}).get(file_type, {})
if not file_limits:
file_limits = defaults.get(file_type, {})
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")
if not file_limits:
# Neither per_branch NOR defaults have limits for this branch/file_type
logger.warning(
f"[detector] CONFIG GAP: no v2 limits for branch={branch_name} file_type={file_type} "
f"in per_branch or defaults — skipping rollover"
)
return (False, current_lines, "3.0.0", f"config gap for {branch_name}/{file_type}")
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")
reasons = []
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:
reasons.append(f"{len(observations)}/{max_observations} observations")
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")
if reasons:
return (True, current_lines, 0, "2.0.0", ", ".join(reasons))
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")
# v1: line-count based (fallback when no v2 limits triggered)
max_lines = limits.get("max_lines")
if max_lines is None:
max_lines = _get_max_lines(file_path)
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:
reasons.append(f"{len(observations)}/{max_observations} observations")
return (current_lines >= max_lines, current_lines, max_lines, "1.0.0", "")
if reasons:
return (True, current_lines, "3.0.0", ", ".join(reasons))
return (False, current_lines, "3.0.0", "")
def check_all_branches() -> Dict[str, Any]:
@@ -411,7 +356,7 @@ def check_all_branches() -> Dict[str, Any]:
if file_path is None:
continue
should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)
should_trigger, current_lines, schema_ver, v2_reason = _should_rollover(file_path)
if should_trigger:
trigger = RolloverTrigger(
@@ -419,7 +364,6 @@ def check_all_branches() -> Dict[str, Any]:
memory_type=memory_type,
file_path=file_path,
current_lines=current_lines,
max_lines=max_lines,
schema_version=schema_ver,
v2_reason=v2_reason,
)
@@ -450,7 +394,7 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
if not file_path.exists():
return {"success": False, "error": f"File not found: {file_path}"}
should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)
should_trigger, current_lines, schema_ver, v2_reason = _should_rollover(file_path)
if should_trigger:
# Extract branch and type from file path
@@ -468,21 +412,18 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
memory_type=memory_type,
file_path=file_path,
current_lines=current_lines,
max_lines=max_lines,
schema_version=schema_ver,
v2_reason=v2_reason,
)
return {"success": True, "trigger": trigger, "should_rollover": True}
else:
remaining = max_lines - current_lines if max_lines > 0 else 0
return {
"success": True,
"should_rollover": False,
"current_lines": current_lines,
"max_lines": max_lines,
"schema_version": schema_ver,
"remaining": remaining,
"v2_reason": v2_reason,
}
@@ -514,13 +455,11 @@ def get_rollover_stats() -> Dict[str, Any]:
continue
stats["files_checked"] += 1
should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)
should_trigger, current_lines, schema_ver, v2_reason = _should_rollover(file_path)
stat_entry = {
stat_entry: Dict[str, Any] = {
"current": current_lines,
"max": max_lines,
"ready": should_trigger,
"remaining": max_lines - current_lines if max_lines > 0 else 0,
"schema_version": schema_ver,
}
if v2_reason:
@@ -106,7 +106,7 @@ def check_and_rollover() -> Dict[str, Any]:
Check all memory files and trigger rollover if any exceed their threshold.
Also processes any new files in memory_pool.
Threshold is determined per-branch from config (defaults to 600).
Threshold is determined per-branch from v2 entry-count limits in config.
This is a startup check - runs once per command, synchronous.
No daemon or file watcher needed.
@@ -157,10 +157,10 @@ def check_and_rollover() -> Dict[str, Any]:
normalize_memory_file(memory_file)
# Use detector for trigger decision (handles both v1 line-based and v2 entry-count)
# Use detector for trigger decision (v2 entry-count based)
from aipass.memory.apps.handlers.monitor.detector import _should_rollover
triggered, current_lines, _, _, _ = _should_rollover(memory_file)
triggered, current_lines, _, _ = _should_rollover(memory_file)
if triggered:
results["files_over_limit"].append(
{"file": str(memory_file), "lines": current_lines, "threshold": 0}
@@ -13,17 +13,15 @@ Surgically extracts oldest items from memory files during rollover.
Understands real JSON structure (sessions, observations, key_learnings arrays).
Purpose:
v1 (schema <2.0.0): When file exceeds max_lines, extract oldest items from
growing arrays to get under line limit.
v2 (schema >=2.0.0): When entry counts exceed limits (max_sessions,
max_key_learnings), extract oldest entries by count.
v2 entry-count based extraction: When entry counts exceed limits
(sessions, key_learnings, observations), extract oldest entries by count.
Limits are read from config per_branch with defaults fallback.
Strategy:
- Detect schema version from document_metadata
- v1: line-count based extraction (legacy)
- v2: entry-count based extraction (sessions + key_learnings + observations arrays)
- Extract oldest items (FIFO)
- Update document_metadata.status
- No line-count fallbacks — errors over silent fallbacks
"""
import shutil
@@ -163,95 +161,11 @@ def _derive_branch_and_type(file_path: Path) -> tuple[str, str]:
return branch_name, memory_type
# =============================================================================
# STRUCTURE DETECTION
# =============================================================================
def _detect_growing_array(data: Dict[str, Any]) -> str | None:
"""
Detect which array field is growing in memory file
Memory files have different structures:
- .local.json → 'sessions' array
- .observations.json → 'observations' array
- Future types → other array fields
Args:
data: Parsed JSON data
Returns:
Array field name (e.g., 'sessions'), or None if not found
"""
# Known array fields that grow over time
candidates = ["sessions", "observations", "recent_work", "entries", "items", "records"]
for field in candidates:
if field in data and isinstance(data[field], list) and len(data[field]) > 0:
return field
return None
# =============================================================================
# EXTRACTION CALCULATION
# =============================================================================
def _calculate_items_to_extract_by_lines(
data: Dict[str, Any], array_field: str, file_path: Path, max_lines: int, target_buffer: int = 100
) -> int:
"""
Calculate items to extract by SIMULATING line count (accurate)
Removes items one by one, counting actual lines after each removal,
until we reach target line count (max_lines - buffer).
Args:
data: Full memory file data
array_field: Name of array field to extract from
file_path: Path (for line counting)
max_lines: Maximum allowed lines
target_buffer: Lines of buffer to leave (default 100)
Returns:
Number of items to extract
Example:
File is 645 lines, limit is 600, buffer is 100
Target: 500 lines (600 - 100)
Simulate removing items until file is ~500 lines
"""
import tempfile
import json
target_lines = max_lines - target_buffer
total_items = len(data[array_field])
# Binary search for optimal item count
for items_to_remove in range(1, total_items + 1):
# Simulate removal (remove from END - oldest items)
test_data = data.copy()
test_data[array_field] = data[array_field][:-items_to_remove] # Keep newest
# Count lines in simulated result
with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as tmp:
json.dump(test_data, tmp, indent=2, ensure_ascii=False)
tmp_path = Path(tmp.name)
with open(tmp_path, "r") as f:
line_count = len(f.readlines())
tmp_path.unlink()
# Check if we've reached target
if line_count <= target_lines:
return items_to_remove
# Fallback: remove 50% if simulation fails
return max(1, total_items // 2)
# =============================================================================
# V2 EXTRACTION (ENTRY-COUNT BASED)
# =============================================================================
@@ -375,7 +289,6 @@ def extract_items(file_path: Path, percentage: int | None = None) -> Dict[str, A
data = _read_memory_file(file_path)
if data is None:
return {"success": False, "error": f"Failed to parse memory file: {file_path.name}"}
current_lines = _count_file_lines(file_path)
except Exception as e:
logger.warning(f"[extractor] Failed to read file {file_path}: {e}")
return {"success": False, "error": f"Failed to read file: {e}"}
@@ -389,71 +302,25 @@ def extract_items(file_path: Path, percentage: int | None = None) -> Dict[str, A
_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 not _ext_file_limits:
# Also check defaults fallback
_ext_file_limits = _ext_cfg.get("defaults", {}).get(_ext_ftype, {})
if _ext_file_limits:
return _extract_items_v2(file_path, data)
# v1: line-count based extraction
# Detect structure
array_field = _detect_growing_array(data)
if not array_field:
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", 600)
max_lines = data.get("document_metadata", {}).get("limits", {}).get("max_lines", _cfg_max)
# Check if under limit
if current_lines < max_lines:
return {"success": True, "skipped": True, "message": f"File under limit ({current_lines}/{max_lines} lines)"}
# Calculate extraction amount (simulate actual line reduction)
total_items = len(data[array_field])
if percentage is None:
# Use line-based calculation for accuracy
items_to_extract = _calculate_items_to_extract_by_lines(
data, array_field, file_path, max_lines, target_buffer=100
)
else:
# Manual percentage override (for testing)
items_to_extract = max(1, int(total_items * percentage / 100))
# Extract oldest items (LAST N in array - newest first, oldest last)
extracted = data[array_field][-items_to_extract:] # Take from end (oldest)
remaining = data[array_field][:-items_to_extract] # Keep from start (newest)
# Update array
data[array_field] = remaining
# Update metadata
_update_metadata_after_extraction(data)
# Write back
try:
_write_memory_file(file_path, data)
new_line_count = _count_file_lines(file_path)
except Exception as e:
logger.error(f"[extractor] Failed to write file after v1 extraction: {e}")
return {"success": False, "error": f"Failed to write file: {e}"}
# Derive branch and type from path
branch_name, memory_type = _derive_branch_and_type(file_path)
# No v2 limits found — fail loud, never fall back to v1 line-count
logger.warning(
f"[extractor] NO V2 LIMITS for branch={_ext_branch} file_type={_ext_ftype} "
f"— cannot extract without config. Check per_branch and defaults in memory.config.json"
)
json_handler.log_operation(
"extract_items", {"branch": branch_name, "type": memory_type, "extracted_count": items_to_extract}
"extract_items_no_limits",
{"branch": _ext_branch, "file_type": _ext_ftype, "error": "no v2 limits configured"},
)
return {
"success": True,
"file": str(file_path),
"branch": branch_name,
"type": memory_type,
"array_field": array_field,
"extracted": extracted,
"extracted_count": items_to_extract,
"remaining_count": len(remaining),
"old_lines": current_lines,
"new_lines": new_line_count,
"success": False,
"error": f"No v2 extraction limits configured for {_ext_branch}/{_ext_ftype}",
}
+2 -2
View File
@@ -257,8 +257,8 @@ def start_watch() -> None:
"""
Start memory watcher - monitors branch memory files for auto-rollover
Watches all branches from AIPASS_REGISTRY.json. When a memory file
exceeds 600 lines, automatically triggers rollover.
Watches all branches from AIPASS_REGISTRY.json. When entry counts
exceed v2 limits, automatically triggers rollover.
Press Ctrl+C to stop.
"""
+4 -14
View File
@@ -167,11 +167,10 @@ def print_help() -> None:
console.print(" [cyan]help[/cyan] Show this help message")
console.print()
console.print("[bold]LIMITS:[/bold]")
console.print(" v1 (schema <2.0): Line-count based (max_lines, default 600)")
console.print(" v2 (schema 2.0+): Entry-count based (max_sessions, max_key_learnings)")
console.print(" v2 entry-count based (sessions, key_learnings, observations) from config")
console.print()
console.print("[bold]WORKFLOW:[/bold]")
console.print(" 1. Detect files exceeding limits (line count or entry count)")
console.print(" 1. Detect files exceeding v2 entry-count limits")
console.print(" 2. Extract oldest entries")
console.print(" 3. Generate embeddings via fastembed")
console.print(" 4. Store vectors in local + global ChromaDB")
@@ -400,21 +399,12 @@ def show_status() -> None:
console.print(f" [bold]{branch_name}[/bold]")
for memory_type, file_stats in branch_stats.items():
current = file_stats["current"]
max_val = file_stats["max"]
ready = file_stats["ready"]
remaining = file_stats["remaining"]
schema_ver = file_stats.get("schema_version", "1.0.0")
v2_reason = file_stats.get("v2_reason", "")
status_marker = "[red]![/red]" if ready else "[green]OK[/green]"
if schema_ver.startswith("2"):
status_text = f"READY ({v2_reason})" if ready else "OK (v2)"
console.print(f" {status_marker} {memory_type}: {status_text}")
else:
status_text = "READY" if ready else f"{remaining} remaining"
console.print(f" {status_marker} {memory_type}: {current}/{max_val} lines ({status_text})")
status_text = f"READY ({v2_reason})" if ready else "OK"
console.print(f" {status_marker} {memory_type}: {status_text}")
console.print()
+7 -10
View File
@@ -869,7 +869,8 @@ def run_demo() -> None:
mock_fragments = [
{
"summary": "User was stuck on a code error for a while, then solved it with assistant's step-by-step debugging guidance",
"summary": "User was stuck on a code error for a while, then solved it with "
"assistant's step-by-step debugging guidance",
"insight": "Step-by-step debugging with explanation is more effective than just providing the fix",
"type": "episodic",
"triggers": ["error", "debug", "stuck", "breakthrough"],
@@ -877,7 +878,8 @@ def run_demo() -> None:
"technical_domain": "debugging",
},
{
"summary": "Collaborative pattern where assistant explains reasoning before giving solutions leads to better understanding",
"summary": "Collaborative pattern where assistant explains reasoning before giving "
"solutions leads to better understanding",
"insight": "Teaching approach builds deeper knowledge than direct answers",
"type": "procedural",
"triggers": ["explain", "step by step", "understanding"],
@@ -1408,15 +1410,10 @@ def _find_bootstrap_sessions(max_sessions: int = 8) -> List[Path]:
return []
# Priority branch directories (diverse content sources)
aipass_src = Path(__file__).resolve().parents[3]
priority_dirs = [
"-home-patrick-Projects-AIPass-src-aipass-memory",
"-home-patrick-Projects-AIPass-src-aipass-devpulse",
"-home-patrick-Projects-AIPass-src-aipass-seedgo",
"-home-patrick-Projects-AIPass-src-aipass-drone",
"-home-patrick-Projects-AIPass-src-aipass-flow",
"-home-patrick-Projects-AIPass-src-aipass-prax",
"-home-patrick-Projects-AIPass-src-aipass-ai-mail",
"-home-patrick-Projects-AIPass-src-aipass-api",
str(aipass_src / name).replace("/", "-").replace("_", "-")
for name in ["memory", "devpulse", "seedgo", "drone", "flow", "prax", "ai_mail", "api"]
]
selected = []
+89 -90
View File
@@ -139,81 +139,6 @@ class TestGetMemoryFilePath:
assert result.name == "local.json"
# ===========================================================================
# _load_config
# ===========================================================================
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"
config_data = {
"rollover": {
"defaults": {"max_lines": 500},
"per_branch": {"SEEDGO": {"max_lines": 800}},
}
}
config_file.write_text(json.dumps(config_data), encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
# Patch the config path resolution to point at our tmp_path
monkeypatch.setattr(
detector,
"_load_config",
lambda: json.loads(config_file.read_text(encoding="utf-8")),
)
result = detector._load_config()
assert result == config_data
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
monkeypatch.setattr(
detector,
"_load_config",
lambda: {},
)
result = detector._load_config()
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"
config_file.write_text("NOT VALID JSON {{", encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
# Simulate the real _load_config behavior on bad JSON
def _broken_load() -> dict:
try:
with open(config_file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as exc:
logger.warning("Failed to parse config file %s: %s", config_file, exc)
return {}
monkeypatch.setattr(detector, "_load_config", _broken_load)
result = detector._load_config()
assert result == {}
# ===========================================================================
# check_single_file
# ===========================================================================
@@ -222,14 +147,11 @@ class TestLoadConfig:
class TestCheckSingleFile:
"""Tests for check_single_file(file_path)."""
def test_file_under_threshold_no_rollover(self, tmp_path: Path):
"""A small file should not trigger rollover."""
def test_file_under_threshold_no_rollover(self, tmp_path: Path, monkeypatch):
"""A small file with entries under v2 limits should not trigger rollover."""
mem_file = tmp_path / "SEEDGO.observations.json"
data = {
"document_metadata": {
"schema_version": "1.0.0",
"limits": {"max_lines": 600},
},
"document_metadata": {"schema_version": "3.0.0"},
"observations": [],
}
content = json.dumps(data, indent=2)
@@ -237,28 +159,35 @@ class TestCheckSingleFile:
from aipass.memory.apps.handlers.monitor import detector
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"seedgo": {"observations": {"observations": {"count": 10}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
assert result["current_lines"] < 600
def test_file_over_threshold_triggers_rollover(self, tmp_path: Path):
"""A file exceeding max_lines should trigger rollover."""
def test_file_over_threshold_triggers_rollover(self, tmp_path: Path, monkeypatch):
"""A file exceeding v2 entry-count limits should trigger rollover."""
mem_file = tmp_path / "SEEDGO.local.json"
# Build a file with many lines so it exceeds threshold of 10
data = {
"document_metadata": {
"schema_version": "1.0.0",
"limits": {"max_lines": 10},
},
"sessions": [{"id": f"s{i}", "notes": "padding " * 20} for i in range(50)],
"document_metadata": {"schema_version": "3.0.0"},
"sessions": [{"id": f"s{i}", "notes": "padding"} for i in range(50)],
}
content = json.dumps(data, indent=2)
mem_file.write_text(content, encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"seedgo": {"local": {"sessions": {"count": 10}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
@@ -371,6 +300,76 @@ class TestCheckSingleFile:
assert result["success"] is True
assert result["should_rollover"] is False
def test_parse_failure_returns_no_rollover(self, tmp_path: Path, monkeypatch):
"""JSON parse failure should return should_rollover=False, not 600-line fallback."""
mem_file = tmp_path / "BROKEN.local.json"
mem_file.write_text("NOT VALID JSON {{{", encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {"broken": {"local": {"sessions": {"count": 5}}}}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
assert "parse failure" in result.get("v2_reason", "")
def test_defaults_fallback_when_branch_missing_from_per_branch(self, tmp_path: Path, monkeypatch):
"""Branch not in per_branch should fall back to defaults and still trigger on entry count."""
mem_file = tmp_path / "NEWBRANCH.local.json"
data = {
"document_metadata": {"schema_version": "3.0.0"},
"sessions": [{"id": f"s{i}"} for i in range(15)],
}
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
# per_branch has NO entry for "newbranch", but defaults has local limits
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {
"per_branch": {},
"defaults": {"local": {"sessions": {"count": 10}}},
},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is True
assert "15/10 sessions" in result["trigger"].v2_reason
def test_no_limits_in_per_branch_or_defaults_skips_rollover(self, tmp_path: Path, monkeypatch):
"""Branch missing from BOTH per_branch AND defaults should not roll and log config gap."""
mem_file = tmp_path / "ORPHAN.local.json"
data = {
"document_metadata": {"schema_version": "3.0.0"},
"sessions": [{"id": f"s{i}"} for i in range(100)],
}
mem_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
from aipass.memory.apps.handlers.monitor import detector
# Neither per_branch nor defaults have anything for "orphan"/"local"
monkeypatch.setattr(
detector.config_loader,
"section",
lambda name: {"per_branch": {}, "defaults": {}},
)
result = detector.check_single_file(mem_file)
assert result["success"] is True
assert result["should_rollover"] is False
assert "config gap" in result.get("v2_reason", "")
# ===========================================================================
# _read_registry
+5 -54
View File
@@ -9,7 +9,7 @@
"""Targeted handler-layer tests for critical untested handlers.
Covers:
- rollover/extractor.py (_extract_items_v2, _detect_growing_array, helpers)
- rollover/extractor.py (_extract_items_v2, helpers)
- tracking/line_counter.py (_count_physical_lines, update_line_count)
- schema/normalize.py (normalize_memory_file)
- todos[] operational schema (rollover ignores, caps enforced)
@@ -117,48 +117,19 @@ def _import_normalize(monkeypatch):
# ===========================================================================
class TestDetectGrowingArray:
"""Test _detect_growing_array helper."""
def test_detects_sessions_array(self, monkeypatch):
ext, _ = _import_extractor(monkeypatch)
data = {"sessions": [{"id": 1}, {"id": 2}], "metadata": {}}
assert ext._detect_growing_array(data) == "sessions"
def test_detects_observations_array(self, monkeypatch):
ext, _ = _import_extractor(monkeypatch)
data = {"observations": [{"note": "x"}]}
assert ext._detect_growing_array(data) == "observations"
def test_returns_none_for_empty_arrays(self, monkeypatch):
ext, _ = _import_extractor(monkeypatch)
data = {"sessions": [], "observations": []}
assert ext._detect_growing_array(data) is None
def test_returns_none_when_no_array_fields(self, monkeypatch):
ext, _ = _import_extractor(monkeypatch)
data = {"document_metadata": {}, "key_learnings": {"a": "b"}}
assert ext._detect_growing_array(data) is None
def test_prefers_sessions_over_later_candidates(self, monkeypatch):
ext, _ = _import_extractor(monkeypatch)
data = {"sessions": [{"id": 1}], "entries": [{"id": 2}]}
assert ext._detect_growing_array(data) == "sessions"
class TestDerivebranchAndType:
"""Test _derive_branch_and_type path helper."""
def test_trinity_path_local(self, monkeypatch):
def test_trinity_path_local(self, monkeypatch, tmp_path):
ext, _ = _import_extractor(monkeypatch)
p = Path("/home/user/src/aipass/devpulse/.trinity/local.json")
p = tmp_path / "devpulse" / ".trinity" / "local.json"
branch, mtype = ext._derive_branch_and_type(p)
assert branch == "DEVPULSE"
assert mtype == "local"
def test_trinity_path_observations(self, monkeypatch):
def test_trinity_path_observations(self, monkeypatch, tmp_path):
ext, _ = _import_extractor(monkeypatch)
p = Path("/home/user/src/aipass/memory/.trinity/observations.json")
p = tmp_path / "memory" / ".trinity" / "observations.json"
branch, mtype = ext._derive_branch_and_type(p)
assert branch == "MEMORY"
assert mtype == "observations"
@@ -632,26 +603,6 @@ class TestTodosOperational:
assert data["todos"][0]["id"] == "t1"
assert data["todos"][4]["id"] == "t5"
def test_v1_detect_growing_array_ignores_todos(self, monkeypatch):
"""v1 _detect_growing_array does NOT consider todos as a growing array."""
ext, _ = _import_extractor(monkeypatch)
data = {
"sessions": [{"session_number": 1}],
"todos": [{"id": "t1", "text": "Something"}],
}
result = ext._detect_growing_array(data)
assert result == "sessions"
def test_v1_detect_growing_array_skips_todos_only(self, monkeypatch):
"""If only todos[] exists (no memory arrays), _detect_growing_array returns None."""
ext, _ = _import_extractor(monkeypatch)
data = {
"todos": [{"id": "t1", "text": "Something"}],
"key_learnings": {"k1": "v1"},
}
result = ext._detect_growing_array(data)
assert result is None
def test_todos_schema_shape(self):
"""Validate the expected todos[] item schema: id, text, created, optional priority."""
todo_item = {"id": "t1", "text": "Fix the bug", "created": "2026-06-07"}
@@ -634,8 +634,8 @@ class TestExtractWithMetadata:
assert "branch" in entry["_metadata"]
assert "extracted_at" in entry["_metadata"]
def test_v1_extracts_at_exactly_max_lines(self, monkeypatch, tmp_path):
"""v1 file at exactly max_lines should extract, not skip."""
def test_v2_extracts_when_observations_at_limit(self, monkeypatch, tmp_path):
"""v2 file at entry-count limit should extract, not skip."""
ext, mocks = _import_extractor(monkeypatch)
observations = [
@@ -643,20 +643,20 @@ class TestExtractWithMetadata:
]
data = {
"document_metadata": {
"schema_version": "1.0.0",
"limits": {"max_lines": 50},
"schema_version": "3.0.0",
"status": {},
},
"observations": observations,
}
file_path = tmp_path / "DEVPULSE.observations.json"
content = json.dumps(data, indent=2)
file_path.write_text(content, encoding="utf-8")
actual_lines = len(content.splitlines())
data["document_metadata"]["limits"]["max_lines"] = actual_lines
file_path = tmp_path / ".trinity" / "observations.json"
file_path.parent.mkdir(parents=True)
file_path.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: {"observations": {"observations": {"count": 5}}}},
}
mocks["memory_files"].read_memory_file_data.return_value = data
def fake_write(fp, d):
@@ -676,8 +676,7 @@ class TestExtractWithMetadata:
# v2 file under limits (no extraction needed)
data = {
"document_metadata": {
"schema_version": "2.0.0",
"limits": {"max_sessions": 10},
"schema_version": "3.0.0",
"status": {},
},
"sessions": [{"session_number": 1, "summary": "only one"}],
@@ -686,6 +685,11 @@ class TestExtractWithMetadata:
file_path.parent.mkdir(parents=True)
file_path.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": 10}}}},
}
mocks["memory_files"].read_memory_file_data.return_value = data
result = ext.extract_with_metadata(file_path)
@@ -787,10 +791,8 @@ class TestShowStatus:
"TEST": {
"local": {
"current": 500,
"max": 600,
"ready": False,
"remaining": 100,
"schema_version": "1.0.0",
"schema_version": "3.0.0",
}
}
},
@@ -818,10 +820,8 @@ class TestShowStatus:
"V2BRANCH": {
"local": {
"current": 25,
"max": 20,
"ready": True,
"remaining": 0,
"schema_version": "2.0.0",
"schema_version": "3.0.0",
"v2_reason": "sessions: 25/20",
}
}
+2 -2
View File
@@ -404,7 +404,7 @@ class TestMemoryFileWatcherOnModified:
mocks["check_single_file"].return_value = {
"success": True,
"should_rollover": True,
"trigger": "lines exceeded 600",
"trigger": "5/3 sessions",
}
watcher = mod.MemoryFileWatcher()
@@ -498,7 +498,7 @@ class TestMemoryFileWatcherOnModified:
mocks["check_single_file"].return_value = {
"success": True,
"should_rollover": True,
"trigger": "lines exceeded",
"trigger": "10/5 sessions",
}
mocks["execute_rollover"].side_effect = RuntimeError("Rollover crashed")
@@ -137,9 +137,9 @@ class BranchDetector:
Handles hyphens in project names by splitting on -Projects- and -src-.
Examples:
-home-patrick-Projects-Vera-Studio -> ('VERA-STUDIO', None)
-home-patrick-Projects-AIPL-src-polyglot -> ('AIPL', 'POLYGLOT')
-home-patrick-Projects-Vera-Studio-src-vera -> ('VERA-STUDIO', 'VERA')
-home-user-Projects-Vera-Studio -> ('VERA-STUDIO', None)
-home-user-Projects-AIPL-src-polyglot -> ('AIPL', 'POLYGLOT')
-home-user-Projects-Vera-Studio-src-vera -> ('VERA-STUDIO', 'VERA')
Returns:
(project_name, agent_name) -- agent_name is None if no src subdir
+4
View File
@@ -271,6 +271,10 @@
"file": "apps/handlers/aipass_proof/",
"standard": "handlers",
"reason": "Proof handlers import SOURCE_SKIP_DIRS from aipass_standards/skip_dirs.py — shared constant, same-branch cross-handler import, intentional (FPLAN-0261)."
},
{
"file": "tests/test_checkers_batch10.py",
"reason": "Test file: outside 3-layer structure by convention, imports handlers directly for unit testing, and contains intentional hardcoded-path strings as test input data."
}
],
"notes": {
@@ -0,0 +1,187 @@
# =================== AIPass ====================
# Name: hardcoded_path_check.py
# Description: Hardcoded Absolute Path Standards Checker Handler
# Version: 1.0.0
# Created: 2026-06-18
# Modified: 2026-06-18
# =============================================
"""Hardcoded Absolute Path Standards Checker Handler."""
import re
from pathlib import Path
from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
from aipass.seedgo.apps.handlers.json import json_handler
AUDIT_SCOPE = "all_files"
_POSIX_HOME = re.compile(r"/home/[a-zA-Z][a-zA-Z0-9_.-]+/")
_MACOS_HOME = re.compile(r"/Users/[a-zA-Z][a-zA-Z0-9_.-]+/")
_WINDOWS_HOME = re.compile(r"[A-Z]:\\\\?Users\\\\?[a-zA-Z]")
_DASH_POSIX = re.compile(r"-home-[a-zA-Z][a-zA-Z0-9_.]+-")
_DASH_MACOS = re.compile(r"-Users-[a-zA-Z][a-zA-Z0-9_.]+-")
_ALL_PATTERNS = [
(_POSIX_HOME, "POSIX home path"),
(_MACOS_HOME, "macOS home path"),
(_WINDOWS_HOME, "Windows home path"),
(_DASH_POSIX, "dash-encoded POSIX home"),
(_DASH_MACOS, "dash-encoded macOS home"),
]
_COMMENT_RE = re.compile(r"^\s*#")
_DOCSTRING_DELIMITERS = ('"""', "'''")
def _in_docstring(lines: list[str], line_idx: int) -> bool:
"""Return True if line_idx falls inside a docstring."""
in_ds = False
for i, line in enumerate(lines):
stripped = line.strip()
for delim in _DOCSTRING_DELIMITERS:
count = stripped.count(delim)
if count >= 2:
continue
if count == 1:
in_ds = not in_ds
if i == line_idx:
return in_ds
return False
def _scan_file(content: str) -> list[tuple[int, str, str]]:
"""Scan content for hardcoded home paths.
Returns list of (line_number, description, matched_text).
"""
violations: list[tuple[int, str, str]] = []
lines = content.split("\n")
for idx, line in enumerate(lines):
lineno = idx + 1
if _COMMENT_RE.match(line):
continue
if _in_docstring(lines, idx):
continue
for pattern, desc in _ALL_PATTERNS:
match = pattern.search(line)
if match:
violations.append((lineno, desc, match.group()))
break
return violations
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""Check a Python file for hardcoded absolute home-directory paths."""
path = Path(module_path)
module_path = Path(module_path).as_posix()
if is_bypassed(module_path, "hardcoded_path", bypass_rules=bypass_rules):
return {
"passed": True,
"checks": [
{
"name": "Bypassed",
"passed": True,
"message": "Standard bypassed via .seedgo/bypass.json",
}
],
"score": 100,
"standard": "HARDCODED_PATH",
}
if path.suffix != ".py" or path.name == "__init__.py":
return {
"passed": True,
"checks": [
{
"name": "Hardcoded path",
"passed": True,
"message": "File skipped (non-target)",
}
],
"score": 100,
"standard": "HARDCODED_PATH",
}
if not path.exists():
return {
"passed": False,
"checks": [
{
"name": "File exists",
"passed": False,
"message": f"File not found: {module_path}",
}
],
"score": 0,
"standard": "HARDCODED_PATH",
}
try:
source = path.read_text(encoding="utf-8")
except Exception as e:
logger.info("Cannot read %s: %s", path, e)
return {
"passed": False,
"checks": [
{
"name": "File readable",
"passed": False,
"message": f"Error reading file: {e}",
}
],
"score": 0,
"standard": "HARDCODED_PATH",
}
all_violations = _scan_file(source)
non_bypassed = [
(ln, desc, txt)
for ln, desc, txt in all_violations
if not is_bypassed(module_path, "hardcoded_path", ln, bypass_rules)
]
non_bypassed.sort(key=lambda x: x[0])
checks = []
violation_count = len(non_bypassed)
if violation_count == 0:
checks.append(
{
"name": "Hardcoded path",
"passed": True,
"message": "No hardcoded absolute home paths found",
}
)
else:
previews = [f"L{ln}: {desc} ({txt})" for ln, desc, txt in non_bypassed[:3]]
preview_str = "; ".join(previews)
suffix = f" (and {violation_count - 3} more)" if violation_count > 3 else ""
checks.append(
{
"name": "Hardcoded path",
"passed": False,
"message": f"{violation_count} hardcoded path(s): {preview_str}{suffix}",
}
)
passed_checks = sum(1 for c in checks if c["passed"])
total_checks = len(checks)
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
overall_passed = score >= 75
json_handler.log_operation(
"check_completed",
{"file": str(module_path), "score": score, "standard": "hardcoded_path"},
)
return {
"passed": overall_passed,
"checks": checks,
"score": score,
"standard": "HARDCODED_PATH",
}
@@ -0,0 +1,93 @@
# =================== AIPass ====================
# Name: hardcoded_path_content.py
# Description: Hardcoded Absolute Path Standards Content Handler
# Version: 1.0.0
# Created: 2026-06-18
# Modified: 2026-06-18
# =============================================
"""
Hardcoded Absolute Path Standards Content Handler
Provides formatted hardcoded path standards content.
Module orchestrates, handler implements.
"""
from aipass.seedgo.apps.handlers.json import json_handler
def get_hardcoded_path_standards() -> str:
"""Return formatted hardcoded_path standards content with Rich markup.
Returns:
str: Formatted standards text with Rich styling
"""
lines = [
"[bold cyan]CORE PRINCIPLE:[/bold cyan]",
" AIPass is a cross-platform PUBLIC repo. Source code must NEVER contain",
" hardcoded absolute home-directory paths. They leak usernames, break on",
" other machines, and violate cross-platform portability.",
"",
"[bold cyan]WHAT IT CHECKS:[/bold cyan]",
" Scans Python source lines (excluding comments and docstrings) for:",
"",
" 1. [red]POSIX home paths[/red]: /home/<user>/...",
" 2. [red]macOS home paths[/red]: /Users/<user>/...",
" 3. [red]Windows home paths[/red]: C:\\Users\\<user>\\...",
" 4. [red]Dash-encoded POSIX[/red]: -home-<user>-... (Claude Code project-dir encoding)",
" 5. [red]Dash-encoded macOS[/red]: -Users-<user>-... (Claude Code project-dir encoding)",
"",
"[bold cyan]FALSE-POSITIVE GUARDS:[/bold cyan]",
" - [dim]# comments[/dim] are skipped (line-level)",
" - [dim]Docstrings[/dim] are skipped (triple-quote blocks)",
" - [dim]__init__.py[/dim] files are skipped",
" - [dim]Test files[/dim] are NOT skipped by default -- use line-level bypass",
" for test fixtures that must assert on literal path strings",
"",
"[bold cyan]VIOLATIONS:[/bold cyan]",
"",
" [red]Bad -- hardcoded POSIX home:[/red]",
' [dim]ROOT = "/home/patrick/Projects/AIPass"[/dim]',
"",
" [red]Bad -- dash-encoded home (leaks username):[/red]",
' [dim]dirs = ["-home-patrick-Projects-AIPass-src-aipass-memory"][/dim]',
"",
"[bold cyan]HOW TO FIX:[/bold cyan]",
"",
" [green]Good -- use Path(__file__) or env:[/green]",
" [dim]ROOT = Path(__file__).resolve().parents[3][/dim]",
" [dim]ROOT = Path(os.environ['AIPASS_HOME'])[/dim]",
"",
" [green]Good -- use pathlib for home:[/green]",
" [dim]home = Path.home()[/dim]",
"",
" [green]Good -- test fixtures with generic paths:[/green]",
' [dim]path = "/home/user/Projects/AIPass/..." # generic, no real username[/dim]',
"",
"[yellow]SCOPE:[/yellow]",
" AUDIT_SCOPE = [bold]all_files[/bold]",
" Checks every .py file in the branch",
"",
"[bold cyan]SCORING:[/bold cyan]",
" One check per file (Hardcoded path)",
" [green]100[/green] = no hardcoded absolute home paths found",
" [red]0[/red] = one or more hardcoded paths found",
" Reports up to 3 offending line numbers with descriptions",
" Overall pass threshold: [yellow]75%[/yellow]",
"",
"[bold cyan]BYPASS:[/bold cyan]",
" Via [dim].seedgo/bypass.json[/dim] -- supports standard, file-level,",
" and line-level bypass rules",
"",
" [dim]Example bypass entry (test fixture):[/dim]",
' [dim]{{"file": "tests/test_paths.py", "standard": "hardcoded_path",[/dim]',
' [dim] "lines": [42, 43], "reason": "Test asserts on literal path"}}[/dim]',
"",
"[bold cyan]REFERENCE:[/bold cyan]",
" [dim]See: seedgo standards pack (hardcoded_path)[/dim]",
" [dim]Checker: hardcoded_path_check.py[/dim]",
" [dim]House rule: cross-platform, no hardcoded paths, use pathlib[/dim]",
]
json_handler.log_operation("standard_content_queried", {"standard": "hardcoded_path"})
return "\n".join(lines)
@@ -4,7 +4,15 @@
"hooks": [
{
"type": "command",
"command": "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py UserPromptSubmit:global_prompt"
"command": "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py UserPromptSubmit:tier0_kernel"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py UserPromptSubmit:navmap"
}
]
},
@@ -0,0 +1,283 @@
"""Tests for seedgo checker handlers -- batch 10 (hardcoded_path_check)."""
# =================== META ====================
# Name: test_checkers_batch10.py
# Description: Unit tests for hardcoded_path_check
# Version: 1.0.0
# Created: 2026-06-18
# Modified: 2026-06-18
# =============================================
import pytest
from unittest.mock import MagicMock
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_infrastructure(monkeypatch):
"""Mock heavy infrastructure imports for standards checkers."""
import sys
mock_logger = MagicMock()
mock_json_handler = MagicMock()
mock_json_handler.log_operation = MagicMock(return_value=True)
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed as real_is_bypassed
bypass_pkg = MagicMock()
bypass_utils = MagicMock()
bypass_utils.is_bypassed = real_is_bypassed
bypass_pkg.utils = bypass_utils
bypass_ignore = MagicMock()
bypass_ignore.get_template_ignore_patterns = MagicMock(return_value=[])
bypass_pkg.ignore_handler = bypass_ignore
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass", bypass_pkg)
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass.utils", bypass_utils)
monkeypatch.setitem(
sys.modules,
"aipass.seedgo.apps.handlers.bypass.ignore_handler",
bypass_ignore,
)
for mod_name in [
"aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check",
]:
monkeypatch.delitem(sys.modules, mod_name, raising=False)
# ===========================================================================
# 1. _scan_file — core scanning logic
# ===========================================================================
class TestScanFile:
"""Tests for the _scan_file helper."""
def test_posix_home_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'ROOT = "/home/patrick/Projects/AIPass"\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "POSIX home path"
def test_macos_home_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'ROOT = "/Users/patrick/Projects/AIPass"\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "macOS home path"
def test_windows_home_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'ROOT = "C:\\\\Users\\\\patrick\\\\Projects"\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "Windows home path"
def test_dash_encoded_posix_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'dirs = ["-home-patrick-Projects-AIPass"]\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "dash-encoded POSIX home"
def test_dash_encoded_macos_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'dirs = ["-Users-patrick-Projects-AIPass"]\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "dash-encoded macOS home"
def test_comment_skipped(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = '# ROOT = "/home/patrick/Projects/AIPass"\n'
result = _scan_file(content)
assert len(result) == 0
def test_indented_comment_skipped(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = ' # path = "/home/patrick/test"\n'
result = _scan_file(content)
assert len(result) == 0
def test_docstring_skipped(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = '"""\nExample: /home/patrick/Projects\n"""\nx = 1\n'
result = _scan_file(content)
assert len(result) == 0
def test_clean_file(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = "from pathlib import Path\nROOT = Path(__file__).parent\n"
result = _scan_file(content)
assert len(result) == 0
def test_generic_user_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'path = "/home/user/Projects/AIPass"\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][1] == "POSIX home path"
def test_multiple_violations_same_file(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'A = "/home/alice/foo"\nB = "/Users/bob/bar"\nC = "-home-charlie-baz"\n'
result = _scan_file(content)
assert len(result) == 3
def test_line_numbers_correct(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _scan_file
content = 'clean = 1\nbad = "/home/patrick/x"\nalso_clean = 2\n'
result = _scan_file(content)
assert len(result) == 1
assert result[0][0] == 2
# ===========================================================================
# 2. check_module — full integration via tmp files
# ===========================================================================
class TestCheckModule:
"""Tests for check_module entry point."""
def test_clean_file_passes(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "clean.py"
f.write_text("from pathlib import Path\nROOT = Path(__file__).parent\n")
result = check_module(str(f))
assert result["passed"] is True
assert result["score"] == 100
assert result["standard"] == "HARDCODED_PATH"
def test_violation_fails(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "bad.py"
f.write_text('ROOT = "/home/patrick/Projects/AIPass"\n')
result = check_module(str(f))
assert result["passed"] is False
assert result["score"] == 0
def test_bypass_whole_standard(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "bypassed.py"
f.write_text('ROOT = "/home/patrick/Projects/AIPass"\n')
rules = [{"standard": "hardcoded_path", "file": "bypassed.py"}]
result = check_module(str(f), bypass_rules=rules)
assert result["passed"] is True
assert result["score"] == 100
def test_bypass_specific_line(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "partial.py"
f.write_text('A = "/home/alice/ok"\nB = "/home/bob/also_ok"\n')
rules = [
{"standard": "hardcoded_path", "file": "partial.py", "lines": [1, 2]},
]
result = check_module(str(f), bypass_rules=rules)
assert result["passed"] is True
def test_init_py_skipped(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "__init__.py"
f.write_text('X = "/home/patrick/nope"\n')
result = check_module(str(f))
assert result["passed"] is True
assert "skipped" in result["checks"][0]["message"].lower()
def test_nonexistent_file(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
result = check_module("/no/such/file.py")
assert result["passed"] is False
assert result["score"] == 0
def test_non_python_skipped(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "readme.md"
f.write_text("/home/patrick/whatever\n")
result = check_module(str(f))
assert result["passed"] is True
def test_violation_message_includes_line_info(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "info.py"
f.write_text('x = "/home/alice/stuff"\n')
result = check_module(str(f))
msg = result["checks"][0]["message"]
assert "L1" in msg
assert "POSIX home path" in msg
def test_more_than_three_violations_truncates(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import check_module
f = tmp_path / "many.py"
lines = [f'v{i} = "/home/u{i}/x"\n' for i in range(5)]
f.write_text("".join(lines))
result = check_module(str(f))
msg = result["checks"][0]["message"]
assert "and 2 more" in msg
# ===========================================================================
# 3. _in_docstring — edge cases
# ===========================================================================
class TestInDocstring:
"""Tests for docstring detection."""
def test_single_line_docstring(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _in_docstring
lines = ['"""This is a docstring."""', 'x = "/home/pat/y"']
assert _in_docstring(lines, 0) is False
assert _in_docstring(lines, 1) is False
def test_multiline_docstring(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _in_docstring
lines = ['"""', "/home/patrick/inside", '"""', "/home/patrick/outside"]
assert _in_docstring(lines, 1) is True
assert _in_docstring(lines, 3) is False
def test_single_quote_docstring(self):
from aipass.seedgo.apps.handlers.aipass_standards.hardcoded_path_check import _in_docstring
lines = ["'''", "/home/patrick/inside", "'''", "/home/patrick/outside"]
assert _in_docstring(lines, 1) is True
assert _in_docstring(lines, 3) is False
@@ -13,7 +13,7 @@ has_handler: true
# Telegram Bridge
Multi-bot personal-assistant bridge: long-polling listener routes Patrick's Telegram messages into Claude tmux sessions; Claude's Stop hook writes a pending file and the bot sends the response back to Telegram.
Multi-bot personal-assistant bridge: long-polling listener routes user Telegram messages into Claude tmux sessions; Claude's Stop hook writes a pending file and the bot sends the response back to Telegram.
## Architecture
@@ -16,7 +16,7 @@ Stdlib-only implementation using urllib for Telegram API. No python-telegram-bot
dependency. Follows the same polling/tmux injection pattern as direct_chat.py.
Flow:
Patrick sends Telegram message
User sends Telegram message
-> BaseBot receives it via getUpdates long-polling
-> If /command -> handle via telegram_standards, reply, return
-> If /new -> kill tmux session, reply, return
@@ -169,8 +169,8 @@ class BaseBot:
custom_commands: Dict of bot-specific commands in telegram_standards format
branch_name: Branch name for log streaming (None = no streaming, e.g. base bot)
shared_session: tmux session name to inject into instead of creating own session.
When set, the bot attaches to an existing session (e.g., Patrick's
running Claude Code on PC). Falls back to own session if not found.
When set, the bot attaches to an existing session (e.g., the user's
running Claude Code session). Falls back to own session if not found.
"""
self.bot_id = bot_id
self.bot_token = bot_token
@@ -182,6 +182,8 @@ class BaseBot:
if not hasattr(self, "branch_name"):
self.branch_name = branch_name
self._current_sender_name: str = "User"
self.session_name = f"telegram-{bot_id}"
self.pending_file = PENDING_DIR / f"bot-{bot_id}.json"
@@ -470,6 +472,7 @@ class BaseBot:
from_user = message.get("from", {})
user_id = from_user.get("id", 0)
username = from_user.get("username", "unknown")
self._current_sender_name = from_user.get("first_name", "User")
_ = message.get("message_id", 0) # Available for future use
# Start log streamer on first valid message (if branch has a name)
@@ -711,7 +714,7 @@ class BaseBot:
# Detect type and build prompt
file_type = detect_file_type(file_path)
prompt = build_file_prompt(file_path, file_type, caption=caption or None, sender_name="Patrick")
prompt = build_file_prompt(file_path, file_type, caption=caption or None, sender_name=self._current_sender_name)
# Hook: pre-process
prompt = self.on_message(prompt)
@@ -1205,7 +1208,7 @@ class BaseBot:
Ensure a tmux session is available for message injection.
In shared-session mode: attaches to an existing tmux session (e.g.,
Patrick's running Claude Code on PC). Falls back to own session if
the user's running Claude Code session). Falls back to own session if
the shared session is not found.
In normal mode: creates telegram-{bot_id} session with Claude Code.
@@ -47,9 +47,10 @@ class BranchPlugin(BaseBot):
text: Raw message text from Telegram
Returns:
Prefixed text for Claude: "Patrick via Telegram: {text}"
Prefixed text for Claude: "{sender_name} via Telegram: {text}"
"""
return f"Patrick via Telegram: {text}"
sender = getattr(self, "_current_sender_name", "User")
return f"{sender} via Telegram: {text}"
def on_response(self, text: str) -> str:
"""
@@ -162,7 +162,7 @@ def detect_file_type(file_path: Path) -> str:
return "binary"
def build_file_prompt(file_path: Path, file_type: str, caption: str | None = None, sender_name: str = "Patrick") -> str:
def build_file_prompt(file_path: Path, file_type: str, caption: str | None = None, sender_name: str = "User") -> str:
"""
Build a Claude prompt that includes file content.
@@ -203,7 +203,7 @@ class TestBotFatherClientConnect:
"""connect() returns True when session is authorized."""
mock_client_instance = AsyncMock()
mock_client_instance.is_user_authorized.return_value = True
mock_client_instance.get_me.return_value = MagicMock(first_name="Patrick", id=123)
mock_client_instance.get_me.return_value = MagicMock(first_name="TestUser", id=123)
with patch(
"apps.handlers.botfather_client._telethon_check",
@@ -281,11 +281,11 @@ class TestBranchPluginHooks:
def test_on_message_prefixes_text(self, branch_bot):
result = branch_bot.on_message("deploy the fix")
assert result == "Patrick via Telegram: deploy the fix"
assert result == "User via Telegram: deploy the fix"
def test_on_message_empty_text(self, branch_bot):
result = branch_bot.on_message("")
assert result == "Patrick via Telegram: "
assert result == "User via Telegram: "
def test_on_response_tags_with_branch(self, branch_bot):
result = branch_bot.on_response("Done. Everything is deployed.")
@@ -850,7 +850,7 @@ class TestProcessUpdate:
"message": {
"text": "do something",
"chat": {"id": 1},
"from": {"id": 111, "username": "patrick"},
"from": {"id": 111, "username": "testuser"},
"message_id": 1,
},
}
@@ -868,7 +868,7 @@ class TestProcessUpdate:
"message": {
"text": "hello",
"chat": {"id": 1},
"from": {"id": 111, "username": "patrick"},
"from": {"id": 111, "username": "testuser"},
"message_id": 1,
},
}
@@ -1786,7 +1786,7 @@ class TestLockPidReuse:
)
# Mock /proc read to return different bot's cmdline
with patch("pathlib.Path.read_bytes") as mock_read:
mock_read.return_value = b"python3\x00base_bot.py\x00--bot-id\x00patrick_private"
mock_read.return_value = b"python3\x00base_bot.py\x00--bot-id\x00other_bot"
assert self.bot._check_lock() is False
assert not self.bot._lock_file.exists() # Stale lock cleaned