Merge pull request #590 from AIOSAI/dev

Structure repair system, init hardening, .venv convention, scaffold source-of-truth, registry auto-heal
This commit is contained in:
AIPass
2026-05-15 23:33:26 -07:00
committed by GitHub
48 changed files with 2775 additions and 1469 deletions
+35 -15
View File
@@ -45,28 +45,33 @@ def _get_cwd_branch() -> str | None:
def get_modified_py_files() -> list[str]:
"""Get Python files modified in the working tree, scoped to the CWD branch.
Only returns files inside the current branch's directory (or repo-root files).
This prevents dispatched agents' changes from triggering violations on the
orchestrator or other agents sharing the worktree.
Uses drone @git status (branch-scoped) instead of raw git to comply with
git_gate enforcement. Only returns .py files inside the current branch.
"""
if AIPASS_ROOT is None:
return []
cwd_branch = _get_cwd_branch()
branch_dir = AIPASS_ROOT / "src" / "aipass" / cwd_branch if cwd_branch else None
if not branch_dir or not branch_dir.exists():
return []
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
["drone", "@git", "status"], capture_output=True, text=True, timeout=10, cwd=str(branch_dir)
)
cwd_branch = _get_cwd_branch()
files = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if line.endswith(".py") and not line.startswith(".claude/"):
if cwd_branch and line.startswith("src/aipass/"):
file_branch = line.split("/")[2] if len(line.split("/")) > 2 else None
if file_branch and file_branch != cwd_branch:
continue
full = AIPASS_ROOT / line
if full.exists():
files.append(str(full))
if not line or "file(s) changed" in line:
continue
parts = line.split(None, 1)
if len(parts) != 2:
continue
_, filepath = parts
if not filepath.endswith(".py") or filepath.startswith(".claude/"):
continue
full = AIPASS_ROOT / filepath
if full.exists():
files.append(str(full))
return files
except Exception:
return []
@@ -104,11 +109,26 @@ def check_hook_readme_accountability() -> str | None:
"""Check if hook files changed but README wasn't updated. Returns reminder or None."""
if AIPASS_ROOT is None:
return None
cwd_branch = _get_cwd_branch()
branch_dir = AIPASS_ROOT / "src" / "aipass" / cwd_branch if cwd_branch else None
if not branch_dir or not branch_dir.exists():
return None
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
["drone", "@git", "status", "--all"],
capture_output=True,
text=True,
timeout=10,
cwd=str(branch_dir),
)
changed = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
changed = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if not line or "file(s) changed" in line:
continue
parts = line.split(None, 1)
if len(parts) == 2:
changed.append(parts[1])
hook_files_changed = any(f.startswith(".claude/hooks/") and f.endswith(".py") for f in changed)
readme_changed = ".claude/hooks/README.md" in changed
+10 -8
View File
@@ -1,17 +1,19 @@
# AIPass
A multi-agent framework where autonomous Agents(AIPass citizens) live in branches and deploy disposable sub-agents to do work.
Multi-agent framework. Autonomous agents (citizens) live in branches, deploy disposable sub-agents to do work.
**User:** Name
User: user
# AIPass — Startup protocol
# Startup protocol
On any greeting, silently read these files from CWD and run the commands — no narration, no announcing steps. Just do it and respond with the status.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
**Check:** If `.ai_mail.local/inbox.json` exists, read it. Process any mail — don't ask,
**Run:** `drone @git status`
- Read: `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
- Check: `drone @ai_mail inbox` — process any mail, don't ask.
- Run: `drone @git status`
## Memories
Use drone commands for all operations. Never raw git, gh, file access, or python -m when drone provides it.
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
# Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
+10 -8
View File
@@ -1,17 +1,19 @@
# AIPass
A multi-agent framework where autonomous Agents(AIPass citizens) live in branches and deploy disposable sub-agents to do work.
Multi-agent framework. Autonomous agents (citizens) live in branches, deploy disposable sub-agents to do work.
**User:** Name
User: user
# AIPass — Startup protocol
# Startup protocol
On any greeting, silently read these files from CWD and run the commands — no narration, no announcing steps. Just do it and respond with the status.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
**Check:** If `.ai_mail.local/inbox.json` exists, read it. Process any mail — don't ask,
**Run:** `drone @git status`
- Read: `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
- Check: `drone @ai_mail inbox` — process any mail, don't ask.
- Run: `drone @git status`
## Memories
Use drone commands for all operations. Never raw git, gh, file access, or python -m when drone provides it.
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
# Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
+10 -8
View File
@@ -1,17 +1,19 @@
# AIPass
A multi-agent framework where autonomous Agents(AIPass citizens) live in branches and deploy disposable sub-agents to do work.
Multi-agent framework. Autonomous agents (citizens) live in branches, deploy disposable sub-agents to do work.
**User:** Name
User: user
# AIPass — Startup protocol
# Startup protocol
On any greeting, silently read these files from CWD and run the commands — no narration, no announcing steps. Just do it and respond with the status.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
**Check:** If `.ai_mail.local/inbox.json` exists, read it. Process any mail — don't ask,
**Run:** `drone @git status`
- Read: `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
- Check: `drone @ai_mail inbox` — process any mail, don't ask.
- Run: `drone @git status`
## Memories
Use drone commands for all operations. Never raw git, gh, file access, or python -m when drone provides it.
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
# Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
-56
View File
@@ -1,56 +0,0 @@
# Project-Level Hooks
These hooks are provisioned by `aipass init` and live in the project's
`.claude/settings.json`. They fire when CWD is inside this project.
## What fires and what doesn't
**UserPromptSubmit** hooks fire from project settings. These work:
- `branch_prompt_loader.py` — injects branch-specific prompt
- `email_notification.py` — shows unread email count
- `identity_injector.py` — injects branch identity from passport
**PreToolUse / PostToolUse** hooks are provisioned but **DO NOT FIRE** from
project-level settings. This is a Claude Code limitation (confirmed S122,
GitHub issue #36071). These scripts exist but are dead weight:
- `pre_edit_gate.py` — intended to block cross-branch writes (never runs)
- `auto_fix_diagnostics.py` — intended to run pyright+ruff (never runs)
- `subagent_stop_gate.py` — intended to check subagent files (never runs)
These same hooks DO fire from provider settings (`~/.claude/settings.json`)
where they are also wired. The provider copies handle all enforcement.
**PreCompact** hooks fire from project settings:
- `pre_compact.py` — injects recovery context after compaction
## CWD guard interaction
When this project has UserPromptSubmit hooks (it does), the provider-level
UserPromptSubmit hooks detect this and exit silently. This prevents the AIPass
global prompt from being injected into projects that manage their own context.
The provider-level PreToolUse/PostToolUse hooks still fire (they can only run
at provider level) — so enforcement (git_gate, pre_edit_gate, auto_fix) is
always active regardless of CWD.
## Testing
Provider-level test harness covers project-level behavior:
```bash
python3 $AIPASS_HOME/.claude/hooks/hook_test.py --direct
```
Tests include:
- `direct_provider_guards_for_init_project` — verifies provider hooks are
CWD-guarded when run from an aipass init project
- `direct_project_settings_schema` — validates project settings.json has
expected hooks and all referenced scripts exist
## Updating hooks
```bash
drone @cli aipass init update # Refresh managed project files to latest templates
```
## Related
See `$AIPASS_HOME/.claude/hooks/README.md` for the full hook system documentation.
@@ -1,366 +0,0 @@
#!/usr/bin/env python3
"""
PostToolUse Auto-fix Hook — Detects errors and surfaces them for fixing.
Two-hook system:
PostToolUse (this file) → runs pyright + ruff on edited file, saves errors to state
PreToolUse (pre_edit_gate.py) → blocks edits to OTHER files until errors fixed
Key behaviors:
- Runs py_compile (syntax), ruff lint+format, pyright (type errors) on edited file
- Runs seedgo checklist for AIPass standards
- Saves ruff lint AND pyright errors to state file for PreToolUse gate (hard block)
- Surfaces ALL errors in additionalContext so Claude sees them
Version: 5.2.0
CHANGELOG:
- v5.2.0 (2026-04-20): Save ruff lint errors to state file for hard-block enforcement.
Pre-edit gate now blocks on F401/lint just like type errors.
- v5.1.0 (2026-04-19): Added ruff format --check to surface format drift.
- v5.0.0 (2026-03-17): Replaced mcp__ide__getDiagnostics with direct pyright.
Added state file for PreToolUse gate integration.
Single-file pyright (not whole project).
- v4.3.0 (2026-03-17): Added seedgo checklist integration
- v4.0.0 (2025-11-27): Complete rewrite - actual validation, silent operation
"""
import json
import sys
import subprocess
from pathlib import Path
EDIT_TOOLS = ["Edit", "Write", "MultiEdit", "NotebookEdit"]
LAST_FILE_PATH = Path(__file__).parent / ".last_diagnostics_file"
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
SKIP_EXTENSIONS = {".md", ".txt", ".log", ".csv", ".html"}
# AIPass-specific Python patterns to check
PYTHON_PATTERNS = {
"bad_optional": {"pattern": ": str = None", "message": "Optional param should use 'str | None = None' pattern"},
"logger_debug": {
"pattern": "logger.debug(",
"message": "Use logger.info for SystemLogger (logger.debug not supported)",
},
"return_error_msg": {
"pattern": "return error_msg",
"message": "Return None for error states, not error_msg string",
},
"open_no_encoding": {
"pattern": "open(",
"requires_missing": "encoding=",
"message": "open() without encoding='utf-8'",
},
"log_not_log_operation": {
"pattern": ".log(",
"message": "Use log_operation() with success/error params, not .log()",
},
"dict_none_no_check": {
"pattern": "Dict | None",
"message": "Dict | None return: Add None check before using (if result is None: return)",
},
}
# JSON-specific patterns for emoji corruption
JSON_CORRUPTION_CHARS = ["\ufffd", "\x00"]
def run_python_checks(file_path: str) -> list[str]:
"""Run actual Python validation - returns list of errors."""
errors = []
# 1. Syntax check with py_compile
try:
result = subprocess.run(
[sys.executable, "-m", "py_compile", file_path], capture_output=True, text=True, timeout=5
)
if result.returncode != 0:
errors.append(f"SYNTAX: {result.stderr.strip()}")
except Exception:
pass
# 2. Ruff check (if available) - fast linter
try:
result = subprocess.run(
["ruff", "check", "--select=E,F,W", "--output-format=text", file_path],
capture_output=True,
text=True,
timeout=10,
)
if result.stdout.strip():
for line in result.stdout.strip().split("\n")[:5]:
errors.append(f"LINT: {line}")
except FileNotFoundError:
pass
except Exception:
pass
# 3. Ruff format check — detect format drift
try:
result = subprocess.run(["ruff", "format", "--check", file_path], capture_output=True, text=True, timeout=10)
if result.returncode != 0:
errors.append(f"FORMAT: {Path(file_path).name} needs ruff format (run: ruff format {Path(file_path).name})")
except FileNotFoundError:
pass
except Exception:
pass
# 4. AIPass-specific pattern checks
try:
content = Path(file_path).read_text(encoding="utf-8")
lines = content.split("\n")
for check in PYTHON_PATTERNS.values():
pattern = check["pattern"]
message = check["message"]
requires_missing = check.get("requires_missing")
if requires_missing:
if pattern in content and requires_missing not in content:
errors.append(f"PATTERN: {message}")
continue
for line in lines:
stripped = line.strip()
if stripped.startswith(("#", '"', "'")):
continue
if f'"{pattern}' in line or f"'{pattern}" in line:
continue
if pattern in line:
errors.append(f"PATTERN: {message}")
break
except Exception:
pass
return errors
def run_ruff_lint_structured(file_path: str) -> list[dict]:
"""Run ruff check and return structured violations for the state file.
Returns list of {line, message} dicts — same format as pyright errors.
Only non-empty when ruff finds real violations (not format drift).
"""
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
["ruff", "check", "--select=E,F,W", "--output-format=json", file_path],
capture_output=True,
text=True,
timeout=10,
)
if not result.stdout.strip():
return []
violations = json.loads(result.stdout)
if not isinstance(violations, list):
return []
errors = []
for v in violations[:10]:
line = v.get("location", {}).get("row", 0)
code = v.get("code", "?")
message = v.get("message", "unknown")[:100]
errors.append({"line": line, "message": f"{code}: {message}"})
return errors
except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired, Exception):
return []
def run_pyright_check(file_path: str) -> list[dict]:
"""Run pyright on a single file. Returns list of error dicts."""
# Skip hook files - they don't follow project standards
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
[sys.executable, "-m", "pyright", "--outputjson", file_path], capture_output=True, text=True, timeout=15
)
try:
data = json.loads(result.stdout)
except (json.JSONDecodeError, ValueError):
return []
errors = []
for diag in data.get("generalDiagnostics", []):
severity = diag.get("severity", "")
if severity == "error":
line = diag.get("range", {}).get("start", {}).get("line", 0)
message = diag.get("message", "Unknown error")
errors.append({"line": line, "message": message[:100]})
return errors[:10] # Max 10 errors
except FileNotFoundError:
return [] # pyright not installed
except subprocess.TimeoutExpired:
return [] # Timeout — don't block
except Exception:
return []
def save_diagnostics_state(file_path: str, errors: list[dict]):
"""Save type errors to state file for PreToolUse gate."""
try:
if errors:
state = {"file": str(Path(file_path).resolve()), "errors": errors}
STATE_FILE.write_text(json.dumps(state), encoding="utf-8")
else:
# No errors — clear the state
if STATE_FILE.exists():
STATE_FILE.unlink()
except Exception:
pass
def run_json_checks(file_path: str) -> list[str]:
"""Run actual JSON validation - returns list of errors."""
errors = []
try:
content = Path(file_path).read_text(encoding="utf-8")
for char in JSON_CORRUPTION_CHARS:
if char in content:
errors.append(f"EMOJI CORRUPTION: Found corrupted character '{repr(char)}'")
break
try:
data = json.loads(content)
if isinstance(data, dict):
for key in ["allowed_emojis", "emojis", "emoji_list"]:
if key in data and isinstance(data[key], list):
for item in data[key]:
if isinstance(item, str) and len(item) == 1:
if ord(item) < 128 and item not in "\u2713\u2717":
errors.append(f"EMOJI CORRUPTION: Suspicious char '{item}' in {key}")
break
except json.JSONDecodeError as e:
errors.append(f"JSON SYNTAX: {e.msg} at line {e.lineno}")
except Exception as e:
errors.append(f"READ ERROR: {e!s}")
return errors
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo standards checklist — returns violations only."""
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True,
text=True,
timeout=15,
cwd=str(Path.home() / "Projects" / "AIPass"),
)
if result.returncode != 0:
return []
violations = []
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("\u2717"):
violation = line[1:].strip()
if violation:
violations.append(violation)
return violations[:5]
except FileNotFoundError:
return []
except Exception:
return []
def should_skip_file(file_path: str) -> bool:
"""Check if file should be skipped."""
if not file_path:
return True
ext = Path(file_path).suffix.lower()
return ext in SKIP_EXTENSIONS
def is_same_file_as_last(file_path: str) -> bool:
"""Smart batching DISABLED — always recheck.
Previously skipped rechecks on the same file, but this caused
errors introduced on second edit to be missed (state file didn't
exist from first clean edit, so skip triggered). The 1.7s pyright
cost per edit is acceptable for correctness.
"""
return False
def main():
"""Main hook entry point."""
try:
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
if tool_name not in EDIT_TOOLS:
return
if should_skip_file(file_path):
return
if is_same_file_as_last(file_path):
return
# Collect all errors
errors = []
if file_path.endswith(".py"):
errors = run_python_checks(file_path)
# Seedgo standards checklist
seedgo_violations = run_seedgo_checklist(file_path)
for v in seedgo_violations:
errors.append(f"SEEDGO: {v}")
# Pyright type errors (single file)
type_errors = run_pyright_check(file_path)
for te in type_errors:
errors.append(f"TYPE: L{te['line']}: {te['message']}")
# Save ruff lint + type errors to state file for PreToolUse gate (hard block)
ruff_lint_errors = run_ruff_lint_structured(file_path)
save_diagnostics_state(file_path, ruff_lint_errors + type_errors)
elif file_path.endswith(".json"):
errors = run_json_checks(file_path)
else:
return
# Build output
if errors:
error_text = "\n".join(f" - {e}" for e in errors)
context = f"""[AUTO-FIX] {len(errors)} error(s) in {Path(file_path).name}:
{error_text}
Fix these errors in {Path(file_path).name} now. Do not skip or defer."""
output = {
"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": context},
"systemMessage": f"[AUTO-FIX] {len(errors)} error(s) — fix before continuing",
}
print(json.dumps(output))
else:
output = {"systemMessage": "[diagnostics] ok"}
print(json.dumps(output))
except Exception:
pass # Silent fail
if __name__ == "__main__":
main()
@@ -1,53 +0,0 @@
#!/usr/bin/env python3
"""
Branch Prompt Loader — AIPass Public Repo
Injects branch-specific prompts based on CWD. When working in a branch
directory, loads .aipass/aipass_local_prompt.md and outputs it so the
AI sees branch-specific context.
Version: 1.0.0
"""
from pathlib import Path
def find_branch_root() -> Path | None:
"""
Find the branch root directory.
Looks for .trinity/ or .aipass/ as branch indicators.
Stops at the repo root (has pyproject.toml or .git).
"""
cwd = Path.cwd()
search_path = cwd
while search_path.parent != search_path:
# Branch indicators: has .trinity/ (memory files) or apps/ (code)
has_trinity = (search_path / ".trinity").is_dir()
has_apps = (search_path / "apps").is_dir()
if has_trinity or has_apps:
return search_path
# Stop at repo root
if (search_path / "pyproject.toml").exists() or (search_path / ".git").is_dir():
return None
search_path = search_path.parent
return None
def main():
branch_root = find_branch_root()
if branch_root:
prompt_file = branch_root / ".aipass" / "aipass_local_prompt.md"
if prompt_file.exists():
content = prompt_file.read_text().strip()
branch_name = branch_root.name.upper()
print(f"\n# Branch Context: {branch_name}\n<!-- Source: {prompt_file} -->\n{content}")
if __name__ == "__main__":
main()
@@ -1,96 +0,0 @@
#!/usr/bin/env python3
"""
Email Notification Hook - Notifies of new emails on prompt submit.
Checks the current branch's inbox for unread emails and displays
a notification if any exist.
Version: 1.0.0
"""
import json
from pathlib import Path
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
while search.parent != search:
if (search / "pyproject.toml").exists() or (search / ".git").is_dir():
return search
search = search.parent
return None
def find_branch_root() -> Path | None:
"""Find the branch root directory by walking up from CWD."""
cwd = Path.cwd()
repo_root = find_repo_root()
if not repo_root:
return None
search_path = cwd
for _ in range(10):
has_trinity = (search_path / ".trinity").is_dir()
has_id = list(search_path.glob("*.id.json"))
has_apps = (search_path / "apps").is_dir()
has_mail = (search_path / ".ai_mail.local").is_dir() or (search_path / "ai_mail.local").is_dir()
if (has_trinity or has_id or has_apps or has_mail) and search_path != repo_root:
return search_path
if search_path == repo_root:
break
parent = search_path.parent
if parent == search_path:
break
search_path = parent
return None
def count_new_emails(branch_root: Path) -> int:
"""Count new (unread) emails in the branch's inbox."""
# Check both patterns: .ai_mail.local (canonical) and ai_mail.local (legacy)
inbox_path = branch_root / ".ai_mail.local" / "inbox.json"
if not inbox_path.exists():
inbox_path = branch_root / "ai_mail.local" / "inbox.json"
if not inbox_path.exists():
return 0
try:
with open(inbox_path, "r", encoding="utf-8") as f:
data = json.load(f)
# Handle both formats: {"messages": [...]} and bare [...]
messages = data if isinstance(data, list) else data.get("messages", [])
count = 0
for msg in messages:
if msg.get("status") == "new":
count += 1
elif msg.get("status") is None and not msg.get("read", False):
count += 1
return count
except (json.JSONDecodeError, OSError):
return 0
def main():
branch_root = find_branch_root()
if not branch_root:
return
new_count = count_new_emails(branch_root)
if new_count > 0:
plural = "s" if new_count != 1 else ""
print(
f"You have {new_count} new email{plural} - check with: drone @ai_mail inbox | then: drone @ai_mail view <id> | close with: drone @ai_mail close <id>"
)
if __name__ == "__main__":
main()
@@ -1,118 +0,0 @@
#!/usr/bin/env python3
"""
Identity Injector - Injects branch identity on every prompt.
Reads from [BRANCH].id.json and outputs core identity fields.
Finds the branch root by walking up from CWD looking for apps/ or *.id.json.
Version: 1.0.0
"""
import json
from pathlib import Path
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
while search.parent != search:
if (search / "pyproject.toml").exists() or (search / ".git").is_dir():
return search
search = search.parent
return None
def find_branch_root() -> Path | None:
"""Find the branch root directory by walking up from CWD."""
cwd = Path.cwd()
repo_root = find_repo_root()
if not repo_root:
return None
search_path = cwd
while search_path >= repo_root:
has_trinity = (search_path / ".trinity").is_dir()
has_id = list(search_path.glob("*.id.json"))
if has_trinity or has_id:
return search_path
if search_path == repo_root:
break
search_path = search_path.parent
return None
def find_id_file(branch_root: Path) -> Path | None:
"""Find the identity file for a branch (.trinity/passport.json or *.id.json)."""
# AIPass pattern: .trinity/passport.json
passport = branch_root / ".trinity" / "passport.json"
if passport.exists():
return passport
# Dev-Pass fallback: *.id.json
id_files = list(branch_root.glob("*.id.json"))
if id_files:
return id_files[0]
return None
def format_identity(data: dict) -> str:
"""Format branch_info + identity for injection."""
lines = []
# Try branch_info first (enriched passports), fall back to identity block (setup.sh passports)
branch = data.get("branch_info", {})
identity = data.get("identity", {})
name = branch.get("branch_name") or identity.get("name", "UNKNOWN")
lines.append(f"# {name} Identity")
lines.append(f"Path: {branch.get('path', 'unknown')}")
lines.append(f"Email: {branch.get('email', 'unknown')}")
identity = data.get("identity", {})
if identity.get("role"):
lines.append(f"Role: {identity['role']}")
traits = identity.get("traits") or data.get("traits")
if traits:
if isinstance(traits, list):
lines.append("Traits: " + " | ".join(traits))
else:
lines.append(f"Traits: {traits}")
if identity.get("purpose"):
lines.append(f"Purpose: {identity['purpose']}")
what_i_do = identity.get("what_i_do", [])
if what_i_do:
lines.append("Do: " + " | ".join(what_i_do[:4]))
what_i_dont_do = identity.get("what_i_dont_do", [])
if what_i_dont_do:
lines.append("Don't: " + " | ".join(what_i_dont_do[:3]))
principles = data.get("principles", [])
if principles:
lines.append("Principles: " + " * ".join(principles))
return "\n".join(lines)
def main():
branch_root = find_branch_root()
if not branch_root:
return
id_file = find_id_file(branch_root)
if not id_file or not id_file.exists():
return
try:
data = json.loads(id_file.read_text(encoding="utf-8"))
output = format_identity(data)
if output:
print(f"\n{output}")
except (json.JSONDecodeError, KeyError):
pass
if __name__ == "__main__":
main()
@@ -1,168 +0,0 @@
#!/usr/bin/env python3
"""
Pre-Compact Hook - Inject live state for post-compact recovery.
Reads STATUS.local.md, last session from local.json, and git branch
to give the model real context after compaction — not generic advice.
Version: 3.0.0
"""
import json
import subprocess
import sys
from pathlib import Path
def _find_branch_dir():
"""Find the current branch directory from CWD."""
cwd = Path.cwd()
# Check if we're in a branch dir or subdirectory of one
# Pattern: .../src/aipass/{branch}/...
parts = cwd.parts
for i, part in enumerate(parts):
if part == "aipass" and i > 0 and parts[i - 1] == "src":
branch_dir = Path(*parts[: i + 2])
if branch_dir.is_dir():
return branch_dir
# Check if CWD itself has .trinity/
if (cwd / ".trinity").is_dir():
return cwd
return None
def _read_status_local(branch_dir):
"""Read STATUS.local.md if it exists."""
for name in ["STATUS.local.md", "dev.local.md"]:
path = branch_dir / name
if path.is_file():
try:
return path.read_text(encoding="utf-8")[:3000]
except Exception:
pass
return None
def _read_last_session(branch_dir):
"""Read the most recent session and key_learnings from local.json."""
local_path = branch_dir / ".trinity" / "local.json"
if not local_path.is_file():
return None
try:
data = json.loads(local_path.read_text(encoding="utf-8"))
result = []
# Last session
sessions = data.get("sessions", [])
if sessions:
last = sessions[0]
result.append(
f"Last session (#{last.get('session_number', '?')}, "
f"{last.get('date', '?')}): {last.get('summary', 'no summary')}"
)
# Key learnings (just the keys, not full values — breadcrumbs)
learnings = data.get("key_learnings", {})
if learnings:
keys = list(learnings.keys())[-10:] # last 10
result.append(f"Key learnings available: {', '.join(keys)}")
return "\n".join(result) if result else None
except Exception:
return None
def _get_git_info():
"""Get current git branch and short status."""
try:
branch = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True,
text=True,
timeout=5,
)
subprocess.run(
["git", "diff", "--stat", "--cached", "HEAD"],
capture_output=True,
text=True,
timeout=5,
)
dirty = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True,
text=True,
timeout=5,
)
result = []
if branch.returncode == 0:
result.append(f"Git branch: {branch.stdout.strip()}")
if dirty.returncode == 0 and dirty.stdout.strip():
lines = dirty.stdout.strip().split("\n")
result.append(f"Uncommitted changes: {len(lines)} files")
return "\n".join(result) if result else None
except Exception:
return None
def _get_branch_name(branch_dir):
"""Extract branch name from directory."""
return branch_dir.name if branch_dir else "unknown"
def main():
"""Main hook entry point."""
try:
json.load(sys.stdin)
branch_dir = _find_branch_dir()
branch_name = _get_branch_name(branch_dir)
sections = []
sections.append(f"""POST-COMPACT RECOVERY — @{branch_name}
Context just compacted. Below is your live state. Use it to continue seamlessly.""")
# Git info
git_info = _get_git_info()
if git_info:
sections.append(f"## Git\n{git_info}")
# Last session from local.json
if branch_dir:
session_info = _read_last_session(branch_dir)
if session_info:
sections.append(f"## Last Session\n{session_info}")
# STATUS.local.md — the main context
if branch_dir:
status = _read_status_local(branch_dir)
if status:
sections.append(f"## STATUS.local.md\n{status}")
# Recovery instructions (lean)
sections.append("""## Recovery Protocol
- Continue where the summary left off — don't restart or ask generic questions
- .trinity/local.json has full session history and key_learnings — read it if you need more context
- STATUS.local.md has current work, known issues, and todos
- Save memories proactively — compaction just proved you need to
- Match the conversation tone from before compaction""")
print("\n\n".join(sections), file=sys.stdout)
print("Pre-compact: live state injected", file=sys.stderr)
except Exception as e:
# Fail silently — never block compaction
print(f"Pre-compact hook error: {e}", file=sys.stderr)
sys.exit(0)
if __name__ == "__main__":
main()
@@ -1,149 +0,0 @@
#!/usr/bin/env python3
"""
PreToolUse Gate — Blocks unsafe edits at the hook layer.
Rules (checked in order):
1. Inbox lock — any write targeting *.ai_mail.local/inbox.json is BLOCKED.
Use `drone @ai_mail email` instead.
2. Cross-branch — writes to src/aipass/X/** from a CWD inside src/aipass/Y/**
are BLOCKED unless the calling branch is in TRUSTED_CROSS_WRITERS.
3. State-file — edits to OTHER .py files while the current branch has unresolved
type errors are BLOCKED. (original v1.2.0 logic)
Track E additions: rules 1 + 2 (DPLAN-0139).
Version: 1.3.0
"""
import json
import os
import sys
from pathlib import Path
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
# Single source of truth lives in permissions.py — inline here as fallback
# so the hook works even when aipass package is not on sys.path.
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
def _get_branch(file_path: str) -> str:
"""Extract AIPass branch name from a file path (src/aipass/{branch}/ pattern)."""
parts = Path(file_path).parts
for i, part in enumerate(parts):
if part == "aipass" and i > 0 and parts[i - 1] == "src" and i + 1 < len(parts):
return parts[i + 1]
return ""
def _block(reason: str) -> None:
# codeql[py/clear-text-logging-sensitive-data]
print(json.dumps({"decision": "block", "reason": reason}))
sys.exit(2)
def main():
try:
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
if tool_name not in EDIT_TOOLS:
return
if not file_path:
return
# ------------------------------------------------------------------
# Rule 1: Inbox lock — block all writes to *.ai_mail.local/inbox.json
# ------------------------------------------------------------------
fp = Path(file_path)
if fp.name == "inbox.json" and ".ai_mail.local" in fp.parts:
_block('Direct writes to inbox.json are blocked.\nUse: drone @ai_mail email @<branch> "Subject" "Body"')
# ------------------------------------------------------------------
# Rule 1.5: Dispatched-agent path confinement (DPLAN-0155 M3)
# Daemon-spawned agents can only write inside their own branch dir.
# Breaks the prompt-injection amplifier chain — even if injected,
# a dispatched agent cannot write to other agents' inboxes or code.
# ------------------------------------------------------------------
cwd = input_data.get("cwd", "") or os.getcwd()
cwd_branch = _get_branch(cwd)
session_type = os.environ.get("AIPASS_SESSION_TYPE", "interactive")
if session_type == "daemon" and cwd_branch:
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if target_branch and target_branch != cwd_branch:
_block(
f"Dispatched agent confined to own branch: '{cwd_branch}' "
f"cannot write to '{target_branch}' in daemon mode."
)
repo_root = None
for parent in Path(cwd).parents:
if (parent / ".git").exists():
repo_root = parent
break
if repo_root and not target_branch:
allowed_prefix = str(repo_root / "src" / "aipass" / cwd_branch)
resolved = str(fp.resolve()) if not fp.is_absolute() else str(fp)
if not resolved.startswith(allowed_prefix):
_block(f"Dispatched agent restricted to {allowed_prefix}. Cannot write to: {file_path}")
# ------------------------------------------------------------------
# Rule 2: Cross-branch write enforcement
# ------------------------------------------------------------------
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if cwd_branch and target_branch and cwd_branch != target_branch:
if cwd_branch not in TRUSTED_CROSS_WRITERS:
_block(
f"Cross-branch write blocked: '{cwd_branch}' cannot write to '{target_branch}'.\n"
f"Trusted cross-writers: {', '.join(TRUSTED_CROSS_WRITERS)}"
)
# ------------------------------------------------------------------
# Rule 3: State-file (original v1.2.0) — .py files only
# ------------------------------------------------------------------
if not file_path.endswith(".py"):
return
if not STATE_FILE.exists():
return
try:
state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, IOError):
return
errored_file = state.get("file", "")
errors = state.get("errors", [])
if not errors:
return
try:
current = str(Path(file_path).resolve())
errored = str(Path(errored_file).resolve())
except (OSError, ValueError):
return
if current == errored:
return
current_branch = _get_branch(current)
errored_branch = _get_branch(errored)
if not errored_branch:
return
if current_branch and errored_branch and current_branch != errored_branch:
return
error_summary = "\n".join(f" L{e['line']}: {e['message']}" for e in errors[:5])
_block(f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n{error_summary}")
except Exception:
pass # Silent fail → allow
if __name__ == "__main__":
main()
@@ -1,114 +0,0 @@
#!/usr/bin/env python3
"""
SubagentStop Gate — Checks files modified by subagents before allowing them to finish.
Runs seedgo checklist + basic validation on any .py files the subagent touched.
If violations found, blocks the stop and tells the subagent to fix them.
Version: 1.0.0
"""
import json
import os
import sys
import subprocess
from pathlib import Path
def _find_repo_root() -> Path | None:
"""Walk up from CWD or AIPASS_HOME to find the git repo root."""
for start in (os.environ.get("AIPASS_HOME", ""), os.getcwd()):
p = Path(start)
while p != p.parent:
if (p / ".git").exists():
return p
p = p.parent
return None
AIPASS_ROOT = _find_repo_root()
def get_modified_py_files() -> list[str]:
"""Get Python files modified in the working tree (unstaged + staged)."""
if AIPASS_ROOT is None:
return []
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
)
files = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if line.endswith(".py") and not line.startswith(".claude/"):
full = AIPASS_ROOT / line
if full.exists():
files.append(str(full))
return files
except Exception:
return []
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo checklist on a single file."""
if AIPASS_ROOT is None:
return []
if "/.claude/" in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True,
text=True,
timeout=15,
cwd=str(AIPASS_ROOT),
)
if result.returncode != 0:
return []
violations = []
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("\u2717"):
v = line[1:].strip()
if v:
violations.append(v)
return violations[:5]
except Exception:
return []
def main():
try:
json.load(sys.stdin)
modified = get_modified_py_files()
if not modified:
return # Nothing to check
all_violations = {}
for f in modified:
vs = run_seedgo_checklist(f)
if vs:
name = Path(f).name
all_violations[name] = vs
if not all_violations:
return # All clear
# Build the block reason
lines = ["Standards violations found in files you modified:\n"]
for fname, vs in all_violations.items():
lines.append(f" {fname}:")
for v in vs:
lines.append(f" - {v}")
lines.append("\nFix these violations before finishing.")
output = {"decision": "block", "reason": "\n".join(lines)}
print(json.dumps(output))
except Exception:
pass # Silent fail — don't block on errors
if __name__ == "__main__":
main()
-43
View File
@@ -1,43 +0,0 @@
{
"hooks": {
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/branch_prompt_loader.py"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/email_notification.py"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/identity_injector.py"
}
]
}
],
"PreCompact": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/pre_compact.py"
}
]
}
]
},
"env": {
"AIPASS_HOME": "/home/patrick/Projects/AIPass"
}
}
-1
View File
@@ -18,5 +18,4 @@ build/
logs/
DASHBOARD.local.json
docs.local/
tools/aipass-dev
stress_test_s117.md
+15
View File
@@ -226,6 +226,16 @@
"standard": "encapsulation",
"reason": "Unit tests must import handlers directly to test them in isolation. Entry-point imports would defeat the purpose of unit testing."
},
{
"file": "tests/test_doctor_fix.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — not in apps/. Standard 3-layer structure applies to production code only."
},
{
"file": "tests/test_doctor_fix.py",
"standard": "encapsulation",
"reason": "Unit tests must import modules directly to test them in isolation. Entry-point imports would defeat the purpose of unit testing."
},
{
"file": "apps/handlers/init/bootstrap.py",
"standard": "debug_print",
@@ -245,6 +255,11 @@
"file": "apps/handlers/init/bootstrap.py",
"standard": "log_visibility",
"reason": "bootstrap.py is Pure Python only (no module/prax/cli imports) by design — stdlib getLogger is correct here. prax system_logger requires AIPass to be installed, which hasn't happened at bootstrap time."
},
{
"file": "apps/handlers/init/scaffold_content.py",
"standard": "json_structure",
"reason": "scaffold_content.py is Pure Python only (no module/prax/cli imports) by design — pure string-returning template generators extracted from bootstrap.py. Same constraint as bootstrap.py."
}
]
}
-30
View File
@@ -1,30 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/AGENTS.md -->
# AIPASS — Agent Instructions
This project uses AIPass, a multi-agent framework.
## Key Concepts
- **Project** — this directory. Contains a registry and one or more agents.
- **Agent** — a citizen that lives inside the project with its own identity, memory, and code.
- **Registry** — `AIPASS_REGISTRY.json` tracks all agents.
## Getting Started
Create your first agent:
```
aipass init agent <name>
```
## Available Commands
```
aipass init agent <name> # Create a new agent
drone @spawn create <name> # Create agent (alternative)
drone @seedgo audit <project> # Run standards audit
drone systems # List all infrastructure
```
## Startup
On startup, read: `AIPASS_REGISTRY.json`, `README.md`, `STATUS.local.md`
-42
View File
@@ -1,42 +0,0 @@
# AIPASS
## Startup
On any greeting, silently read these files and run the commands — no narration, no announcing steps. Just do it and respond with the status.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
**Check:** If `.ai_mail.local/inbox.json` exists, read it. Process any mail.
**Run:** `git status`
## Identity
You are **AIPASS** — an AIPass citizen.
- **Module:** `aipass.aipass`
- **Role:**
- **Purpose:** New agent - purpose TBD
## Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
- `local.json` — Session history, key learnings, active tasks
- `observations.json` — Collaboration patterns, insights
- `passport.json` — Identity (rarely changes)
## AIPass Context
This branch is part of the AIPass multi-agent framework. Key concepts:
- **Branch** — your directory (`src/aipass/aipass/`). Your home.
- **Citizen** — the identity that lives in a branch. Has a passport, memories, mailbox.
- **Agent** — a disposable worker spawned for a task. No passport, no memory.
## Commands
```
drone systems # List available infrastructure
drone @ai_mail inbox # Check mailbox
drone @ai_mail send @branch "Subject" "Body" # Send mail
drone @seedgo audit @aipass # Run standards audit
```
-27
View File
@@ -1,27 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/GEMINI.md -->
# AIPASS — Project Instructions
This project uses AIPass, a multi-agent framework.
## Key Concepts
- **Project** — this directory. Contains a registry and one or more agents.
- **Agent** — a citizen that lives inside the project with its own identity, memory, and code.
- **Registry** — `AIPASS_REGISTRY.json` tracks all agents.
## Getting Started
Create your first agent: `aipass init agent <name>`
## Available Commands
```
aipass init agent <name> # Create a new agent
drone @spawn create <name> # Create agent (alternative)
drone @seedgo audit <project> # Run standards audit
drone systems # List all infrastructure
```
## Startup
On startup, read: `AIPASS_REGISTRY.json`, `README.md`, `STATUS.local.md`
@@ -135,6 +135,14 @@ 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 _claude_settings(aipass_home: str | None = None) -> str:
"""Generate .claude/settings.json — hooks for prompt injection at project level.
@@ -196,6 +204,48 @@ def _claude_settings(aipass_home: str | None = None) -> str:
return json.dumps(data, indent=2, ensure_ascii=False) + "\n"
def _guard_init(target: Path) -> None:
"""Block init if target is inside an agent branch or existing project.
Raises RuntimeError with explanation if init should not proceed.
"""
target = target.resolve()
# Block: target IS an agent branch (has passport)
if (target / ".trinity" / "passport.json").is_file():
raise RuntimeError(
f"BLOCKED: '{target}' is an agent branch (has .trinity/passport.json). "
"Agents are managed by 'drone @spawn', not 'aipass init'."
)
# Block: target is INSIDE an agent branch (passport above us)
for parent in target.parents:
if (parent / ".trinity" / "passport.json").is_file():
raise RuntimeError(
f"BLOCKED: '{target}' is inside agent branch '{parent.name}'. "
"Cannot run aipass init inside an agent directory."
)
if parent == parent.parent:
break
# Block: target already has a registry (is already a project)
for f in target.iterdir() if target.is_dir() else []:
if f.is_file() and f.name.endswith("_REGISTRY.json"):
raise RuntimeError(
f"BLOCKED: '{target}' is already an AIPass project (has {f.name}). "
"Use 'aipass init update' to upgrade an existing project."
)
# Block: target is inside an existing project
for parent in target.parents:
if not parent.is_dir():
continue
for f in parent.iterdir():
if f.is_file() and f.name.endswith("_REGISTRY.json"):
raise RuntimeError(
f"BLOCKED: '{target}' is inside AIPass project at '{parent}' (has {f.name}). "
"Cannot create a nested project."
)
if parent == parent.parent:
break
def init_project(target: Path, project_name: str | None = None) -> dict:
"""Initialize an AIPass project in the target directory.
@@ -208,8 +258,10 @@ def init_project(target: Path, project_name: str | None = None) -> dict:
Raises:
ValueError: If project name is empty after sanitization
RuntimeError: If target is inside an agent branch or existing project
"""
target = target.resolve()
_guard_init(target)
if not target.exists():
target.mkdir(parents=True)
@@ -254,38 +306,20 @@ def init_project(target: Path, project_name: str | None = None) -> dict:
global_prompt_path = aipass_dir / "aipass_global_prompt.md"
if not global_prompt_path.exists():
global_prompt_path.write_text(
sc.with_source(sc.global_prompt_md(name), global_prompt_path),
encoding="utf-8",
)
global_prompt_path.write_text(_resolve_global_prompt(name, aipass_home, global_prompt_path), encoding="utf-8")
created.append(str(global_prompt_path))
# 3. CLAUDE.md
claude_md_path = target / "CLAUDE.md"
if not claude_md_path.exists():
claude_md_path.write_text(
sc.with_source(sc.claude_md(name), claude_md_path),
encoding="utf-8",
)
created.append(str(claude_md_path))
# 4. AGENTS.md (Codex)
agents_md_path = target / "AGENTS.md"
if not agents_md_path.exists():
agents_md_path.write_text(
sc.with_source(sc.agents_md(name), agents_md_path),
encoding="utf-8",
)
created.append(str(agents_md_path))
# 5. GEMINI.md
gemini_md_path = target / "GEMINI.md"
if not gemini_md_path.exists():
gemini_md_path.write_text(
sc.with_source(sc.gemini_md(name), gemini_md_path),
encoding="utf-8",
)
created.append(str(gemini_md_path))
# 3-5. CLAUDE.md, AGENTS.md, GEMINI.md — copy from AIPass source of truth
for md_name in ("CLAUDE.md", "AGENTS.md", "GEMINI.md"):
dest = target / md_name
if dest.exists():
continue
source = Path(aipass_home) / md_name if aipass_home else None
if source and source.is_file():
shutil.copy2(str(source), str(dest))
created.append(str(dest))
else:
logging.getLogger(__name__).warning("Source %s not found at AIPASS_HOME, skipping", md_name)
# 6. README.md
readme_md_path = target / "README.md"
@@ -345,6 +379,14 @@ def init_project(target: Path, project_name: str | None = None) -> dict:
init_py.write_text(f'"""{raw_name} — created with aipass init."""\n', encoding="utf-8")
created.append(str(init_py))
# 11. .venv symlink → AIPass shared runtime
venv_link = target / ".venv"
if not venv_link.exists() and aipass_home:
aipass_venv = Path(aipass_home) / ".venv"
if aipass_venv.is_dir():
venv_link.symlink_to(aipass_venv)
created.append(f".venv (symlink to AIPass runtime: {aipass_venv})")
return {
"registry_id": registry_id,
"registry_file": registry_filename,
@@ -405,7 +447,8 @@ def update_project(target: Path) -> dict:
# --- Managed files: write only when content has changed ---
global_prompt_path = aipass_dir / "aipass_global_prompt.md"
generated = sc.with_source(sc.global_prompt_md(name), global_prompt_path)
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))
@@ -434,29 +477,19 @@ def update_project(target: Path) -> dict:
else:
already_current.append(str(settings_path))
claude_md_path = target / "CLAUDE.md"
generated = sc.with_source(sc.claude_md(name), claude_md_path)
if not claude_md_path.exists() or claude_md_path.read_text(encoding="utf-8") != generated:
claude_md_path.write_text(generated, encoding="utf-8")
updated.append(str(claude_md_path))
else:
already_current.append(str(claude_md_path))
agents_md_path = target / "AGENTS.md"
generated = sc.with_source(sc.agents_md(name), agents_md_path)
if not agents_md_path.exists() or agents_md_path.read_text(encoding="utf-8") != generated:
agents_md_path.write_text(generated, encoding="utf-8")
updated.append(str(agents_md_path))
else:
already_current.append(str(agents_md_path))
gemini_md_path = target / "GEMINI.md"
generated = sc.with_source(sc.gemini_md(name), gemini_md_path)
if not gemini_md_path.exists() or gemini_md_path.read_text(encoding="utf-8") != generated:
gemini_md_path.write_text(generated, encoding="utf-8")
updated.append(str(gemini_md_path))
else:
already_current.append(str(gemini_md_path))
# CLAUDE.md, AGENTS.md, GEMINI.md — sync from AIPass source of truth
for md_name in ("CLAUDE.md", "AGENTS.md", "GEMINI.md"):
dest = target / md_name
source = Path(aipass_home) / md_name if aipass_home else None
if not source or not source.is_file():
already_current.append(str(dest))
continue
source_content = source.read_text(encoding="utf-8")
if not dest.exists() or dest.read_text(encoding="utf-8") != source_content:
shutil.copy2(str(source), str(dest))
updated.append(str(dest))
else:
already_current.append(str(dest))
# .claude/commands/prep.md — managed slash command, refresh to latest
# Only prep.md — memo.md belongs at provider level (~/.claude/commands/)
@@ -485,6 +518,14 @@ def update_project(target: Path) -> dict:
):
skipped.append(skip_name)
# .venv symlink → AIPass shared runtime (create if missing)
venv_link = target / ".venv"
if not venv_link.exists() and aipass_home:
aipass_venv = Path(aipass_home) / ".venv"
if aipass_venv.is_dir():
venv_link.symlink_to(aipass_venv)
updated.append(f".venv (symlink to AIPass runtime: {aipass_venv})")
return {
"project_name": name,
"target": str(target),
@@ -401,7 +401,7 @@ def prep_md() -> str:
"this session.\n"
"- **`.trinity/local.json`** — YOUR MEMORY. Add/update session entry "
"with a summary of work done. Add key_learnings for anything learned. "
"Trim oldest sessions if over 20.\n"
"DO NOT trim or delete entries — rollover to @memory handles overflow automatically.\n"
"- **`.trinity/observations.json`** — YOUR MEMORY OF THE USER. "
"Collaboration insights, preferences, friction points. Skip if nothing "
"new about the user this session.\n"
@@ -480,8 +480,8 @@ def memo_md() -> str:
"this session. Don't touch it just to touch it.\n"
"- **`.trinity/local.json`** — YOUR MEMORY. Session history and "
"key_learnings. Add a session entry for significant work. Add "
"key_learnings for facts you'd need next time. Trim oldest sessions "
"if over 20.\n"
"key_learnings for facts you'd need next time. "
"DO NOT trim or delete entries — rollover to @memory handles overflow automatically.\n"
"- **`.trinity/observations.json`** — YOUR MEMORY OF THE USER. "
"Collaboration insights, preferences, friction points, flow states. "
"Skip entirely if nothing new about the user this session.\n"
@@ -49,6 +49,15 @@ class PollutionHit(NamedTuple):
locations: List[str]
class RootArtifact(NamedTuple):
"""Branch-level file or directory found at project root."""
name: str
artifact_type: str
severity: str
description: str
class RegistryIssue(NamedTuple):
"""Registry path mismatch."""
@@ -84,6 +93,9 @@ def find_project_root(start: Path) -> Optional[Path]:
# =============================================================================
_SCAN_SKIP_DIRS = {".archive", ".venv", ".git", "__pycache__", "node_modules", ".chroma"}
def scan_agents(project_root: Path) -> List[AgentInfo]:
"""Find all agents by scanning for .trinity/passport.json under project_root.
@@ -92,6 +104,11 @@ def scan_agents(project_root: Path) -> List[AgentInfo]:
"""
agents: List[AgentInfo] = []
for passport_path in sorted(project_root.rglob(".trinity/passport.json")):
if any(skip in passport_path.parts for skip in _SCAN_SKIP_DIRS):
continue
agent_dir = passport_path.parent.parent
if agent_dir == project_root:
continue
try:
data = json.loads(passport_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as exc:
@@ -118,6 +135,57 @@ def scan_agents(project_root: Path) -> List[AgentInfo]:
return agents
# =============================================================================
# PACKAGE DETECTION
# =============================================================================
def _detect_package_names(project_root: Path) -> set:
"""Read pyproject.toml to find declared package directory names.
Returns:
Set of package names (e.g. {'aipass', 'aipl'}), empty if none found.
"""
pyproject = project_root / "pyproject.toml"
if not pyproject.exists():
return set()
try:
import tomllib # noqa: F811 — stdlib 3.11+
except ImportError:
try:
import tomli as tomllib # type: ignore[no-redef]
except ImportError:
logger.info("[structure_scan] no TOML parser available — skipping package detection")
return set()
try:
with open(pyproject, "rb") as f:
data = tomllib.load(f)
except Exception as exc:
logger.warning("[structure_scan] pyproject.toml parse error: %s", exc)
return set()
names: set = set()
hatch_pkgs = (
data.get("tool", {}).get("hatch", {}).get("build", {}).get("targets", {}).get("wheel", {}).get("packages", [])
)
for pkg in hatch_pkgs:
name = Path(pkg).name
if name:
names.add(name)
setup_pkgs = data.get("tool", {}).get("setuptools", {}).get("packages", [])
for pkg in setup_pkgs:
if pkg:
names.add(pkg)
if names:
logger.info("[structure_scan] detected packages: %s", names)
return names
# =============================================================================
# PLACEMENT VALIDATION
# =============================================================================
@@ -126,11 +194,15 @@ def scan_agents(project_root: Path) -> List[AgentInfo]:
def check_placement(agents: List[AgentInfo], project_root: Path) -> List[PlacementIssue]:
"""Check whether each agent is in src/<package>/<agent>/ or src/<agent>/.
When pyproject.toml defines packages, agents at src/<name>/ where name
is not a declared package are flagged as misplaced siblings.
Returns:
List of PlacementIssue for agents in unexpected locations.
"""
src_dir = project_root / "src"
issues: List[PlacementIssue] = []
package_names = _detect_package_names(project_root)
for agent in agents:
rel = None
@@ -150,7 +222,15 @@ def check_placement(agents: List[AgentInfo], project_root: Path) -> List[Placeme
parts = rel.parts
if len(parts) == 1:
# src/<agent>/ — valid single-agent layout
if package_names and parts[0] not in package_names:
issues.append(
PlacementIssue(
agent_name=agent.name,
actual_path=str(agent.path),
expected_pattern="src/<package>/<agent>/ — agent outside package framework",
severity="warn",
)
)
continue
elif len(parts) == 2:
# src/<package>/<agent>/ — valid multi-agent package layout
@@ -246,6 +326,52 @@ def check_registry_consistency(
return issues
# =============================================================================
# ROOT ARTIFACT DETECTION
# =============================================================================
_ROOT_ARTIFACTS = {
".chroma": ("chroma", "warn", "Memory vector store belongs inside branch directory"),
".ai_mail.local": ("mailbox", "warn", "Branch mailbox belongs inside branch directory"),
"logs": ("logs", "warn", "Branch logs directory belongs inside branch directory"),
".venv": (
"venv",
"info",
"Redundant venv — AIPass provides runtime via drone, use requirements.project.txt for extras",
),
}
def check_root_artifacts(project_root: Path) -> List[RootArtifact]:
"""Detect branch-level files/dirs sitting at project root instead of inside branches.
Returns:
List of RootArtifact for each misplaced item found.
"""
hits: List[RootArtifact] = []
for dirname, (artifact_type, severity, description) in _ROOT_ARTIFACTS.items():
candidate = project_root / dirname
if not candidate.exists():
continue
if dirname == ".venv" and candidate.is_symlink():
hits.append(
RootArtifact(
name=dirname,
artifact_type=artifact_type,
severity="pass",
description=f"Linked to AIPass runtime ({candidate.resolve()})",
)
)
continue
hits.append(RootArtifact(name=dirname, artifact_type=artifact_type, severity=severity, description=description))
if hits:
logger.info("[structure_scan] found %d root artifacts at %s", len(hits), project_root)
json_handler.log_operation("check_root_artifacts", {"count": len(hits), "root": str(project_root)})
return hits
# =============================================================================
# PYPROJECT CHECK
# =============================================================================
+33 -33
View File
@@ -9,18 +9,7 @@
"""
aipass doctor — system health aggregation
Flutter-doctor-style health check across four groups:
System — Python, git, shell, OS, RAM, CPU, install method
Identity — AIPASS_HOME, registry, passport integrity
Services — drone routing, pytest collect, hooks wired
Community — ai_mail, dropbox
Three-tier glyph output: ✓ green / ! yellow / ✗ red
Remediation shown inline under failing checks.
Exit 0 on pass+warn, non-zero only on errors.
Pure reads — never mutates (unless --fix or interactive auto-wire accepted).
Run: aipass doctor [--verbose] [--fix]
Run: aipass doctor [--verbose] [--fix] [--fix --json]
"""
from __future__ import annotations
@@ -40,11 +29,16 @@ from aipass.aipass.apps.handlers.structure_scan.structure_scanner import (
check_placement,
check_pyproject,
check_registry_consistency,
check_root_artifacts,
detect_pollution,
find_project_root,
find_registry,
scan_agents,
)
from aipass.aipass.apps.modules.doctor_fix import (
print_json_report,
print_remediation_report,
)
from aipass.aipass.apps.modules.doctor_wire import (
_auto_wire_provider,
prompt_auto_wire,
@@ -532,6 +526,15 @@ def _check_structure() -> List[CheckResult]:
else:
results.append(CheckResult("registry", GLYPH_WARN, "not found", "Expected *_REGISTRY.json in project root"))
# Root artifacts
root_hits = check_root_artifacts(project_root)
if root_hits:
for hit in root_hits:
glyph = GLYPH_WARN if hit.severity == "warn" else GLYPH_PASS
results.append(CheckResult(f"root: {hit.name}", glyph, hit.description, ""))
else:
results.append(CheckResult("root artifacts", GLYPH_PASS, "none misplaced", ""))
# Pyproject
pyproject = check_pyproject(project_root)
if pyproject["found"]:
@@ -615,12 +618,8 @@ def print_introspection() -> None:
console.print("[bold cyan]doctor Module[/bold cyan]")
console.print("System health aggregation — flutter-doctor-style output")
console.print()
console.print("[yellow]Handlers:[/yellow] system_detect, ui/progress, json, structure_scan")
console.print("[yellow]Groups:[/yellow] System, Identity, Services, Community, Structure")
console.print()
console.print("[yellow]Next:[/yellow]")
console.print(" [green]aipass doctor[/green] [dim]# Run all checks[/dim]")
console.print(" [green]aipass doctor --verbose[/green] [dim]# Full check detail[/dim]")
console.print("[yellow]Next:[/yellow] [green]aipass doctor[/green] / [green]aipass doctor --fix[/green]")
console.print()
@@ -628,17 +627,15 @@ def print_help() -> None:
"""Print help information."""
console.print()
console.print("[bold cyan]aipass doctor[/bold cyan] — System health aggregation")
console.print("Flutter-doctor-style check across System / Identity / Services / Community / Structure")
console.print()
console.print("[yellow]USAGE:[/yellow]")
console.print(" [green]aipass doctor[/green] [dim]# Run all checks (interactive)[/dim]")
console.print(" [green]aipass doctor[/green] [dim]# Run all checks[/dim]")
console.print(" [green]aipass doctor --verbose[/green] [dim]# Show sub-check detail[/dim]")
console.print(" [green]aipass doctor --fix[/green] [dim]# Auto-wire missing provider settings[/dim]")
console.print(" [green]aipass doctor --fix[/green] [dim]# Auto-wire + remediation report[/dim]")
console.print(" [green]aipass doctor --fix --json[/green][dim]# Remediation as JSON (for spawn)[/dim]")
console.print()
console.print(
"[yellow]OUTPUT:[/yellow] [green]✓[/green] pass [yellow]![/yellow] warning [red]✗[/red] error (remediation shown)"
)
console.print("[yellow]EXIT:[/yellow] 0 = pass/warn only | 1 = errors found")
console.print("[yellow]OUTPUT:[/yellow] [green]✓[/green] pass [yellow]![/yellow] warn [red]✗[/red] error")
console.print("[yellow]EXIT:[/yellow] 0 = pass/warn | 1 = errors found")
console.print()
@@ -670,7 +667,19 @@ def handle_command(command: str, args: list[str]) -> bool:
verbose = "--verbose" in args or "-v" in args
fix_mode = "--fix" in args
json_mode = "--json" in args
if json_mode and fix_mode:
project_root = find_project_root(Path.cwd())
if project_root:
print_json_report(project_root)
return True
error_count = run_doctor(verbose=verbose, interactive=True, fix=fix_mode)
if fix_mode:
project_root = find_project_root(Path.cwd())
if project_root:
print_remediation_report(project_root)
json_handler.log_operation("doctor_run", {"error_count": error_count, "fix": fix_mode})
if error_count > 0:
raise SystemExit(1)
@@ -683,13 +692,4 @@ def handle_command(command: str, args: list[str]) -> bool:
if __name__ == "__main__":
logger.info("Prax logger connected to doctor")
if len(sys.argv) > 1 and sys.argv[1] in ("--help", "-h", "help"):
print_help()
sys.exit(0)
if len(sys.argv) > 1 and sys.argv[1] == "--info":
print_introspection()
sys.exit(0)
handle_command("doctor", sys.argv[1:])
@@ -0,0 +1,322 @@
# =================== AIPass ====================
# Name: doctor_fix.py
# Description: Structure remediation report for aipass doctor --fix
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""
doctor_fix — structure remediation report for aipass doctor --fix
Generates remediation items from structure scan results and formats
them as human-readable text or machine-readable JSON with exact
`drone @spawn repair` commands.
Run: aipass doctor --fix [--json]
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import List, NamedTuple
from aipass.cli.apps.modules import console
from aipass.prax import logger
from aipass.aipass.apps.handlers.json import json_handler
from aipass.aipass.apps.handlers.structure_scan.structure_scanner import (
check_placement,
check_pyproject,
check_registry_consistency,
check_root_artifacts,
detect_pollution,
find_registry,
scan_agents,
)
# =============================================================================
# TYPES
# =============================================================================
class RemediationItem(NamedTuple):
"""Single remediation suggestion with severity and spawn command."""
severity: str
category: str
description: str
fix_command: str
# =============================================================================
# PROJECT NAME DETECTION
# =============================================================================
def detect_project_name(project_root: Path) -> str:
"""Derive project name from registry filename or directory name."""
reg = find_registry(project_root)
if reg:
name = reg.stem.replace("_REGISTRY", "").lower()
if name:
return name
return project_root.name.lower()
# =============================================================================
# REMEDIATION GENERATION
# =============================================================================
def _build_pollution_items(agents: list, project: str) -> List[RemediationItem]:
"""Build remediation items for pollution issues."""
items: List[RemediationItem] = []
for hit in detect_pollution(agents):
items.append(
RemediationItem(
severity="critical",
category="pollution",
description=(
f"Registry pollution: {len(hit.locations)} copies of "
f"{hit.agent_name} share registry_id {hit.registry_id}"
),
fix_command=f"drone @spawn repair @{project} --clean-pollution",
)
)
return items
def _build_placement_items(agents: list, project_root: Path, project: str) -> List[RemediationItem]:
"""Build remediation items for placement issues."""
items: List[RemediationItem] = []
for issue in check_placement(agents, project_root):
try:
rel_path = str(Path(issue.actual_path).relative_to(project_root))
except ValueError:
logger.info("[doctor_fix] agent %s path not relative to root: %s", issue.agent_name, issue.actual_path)
rel_path = issue.actual_path
suggested = f"src/{project}/{issue.agent_name}/"
items.append(
RemediationItem(
severity="warning",
category="placement",
description=f"Misplaced agent: {issue.agent_name} at {rel_path}",
fix_command=f"drone @spawn repair @{project} --relocate {rel_path} {suggested}",
)
)
return items
def _build_registry_items(project_root: Path, agents: list, project: str) -> List[RemediationItem]:
"""Build remediation items for registry consistency issues."""
items: List[RemediationItem] = []
reg_path = find_registry(project_root)
if not reg_path:
return items
for issue in check_registry_consistency(reg_path, agents):
items.append(
RemediationItem(
severity="warning",
category="registry",
description=f"Registry {issue.problem}: {issue.branch_name} at {issue.registered_path}",
fix_command=f"drone @spawn repair @{project} --dedup-registry",
)
)
return items
def generate_remediation(project_root: Path) -> List[RemediationItem]:
"""Scan project structure and build remediation items with spawn commands."""
project = detect_project_name(project_root)
agents = scan_agents(project_root)
items: List[RemediationItem] = []
items.extend(_build_pollution_items(agents, project))
items.extend(_build_placement_items(agents, project_root, project))
items.extend(_build_registry_items(project_root, agents, project))
pyproject = check_pyproject(project_root)
if not pyproject["found"]:
items.append(
RemediationItem(
severity="info",
category="pyproject",
description="Missing pyproject.toml",
fix_command=f"drone @spawn repair @{project} --add-pyproject",
)
)
for hit in check_root_artifacts(project_root):
severity = "info" if hit.severity == "info" else "warning"
items.append(
RemediationItem(
severity=severity,
category="root_artifact",
description=f"{hit.description}: {hit.name}/",
fix_command=f"drone @spawn repair @{project} --relocate-root {hit.name}",
)
)
logger.info("[doctor_fix] generated %d remediation items for %s", len(items), project)
json_handler.log_operation("generate_remediation", {"count": len(items), "project": project})
return items
# =============================================================================
# TEXT FORMATTING
# =============================================================================
def format_text_report(items: List[RemediationItem], project_name: str) -> str:
"""Format remediation items as plain text."""
if not items:
return f"No structure issues found in @{project_name}."
severity_order = {"critical": 0, "warning": 1, "info": 2}
sorted_items = sorted(items, key=lambda i: severity_order.get(i.severity, 99))
critical_count = sum(1 for i in items if i.severity == "critical")
lines = [f"STRUCTURE ISSUES ({len(items)} found, {critical_count} critical)", ""]
for item in sorted_items:
lines.append(f"[{item.severity.upper()}] {item.description}")
lines.append(f" Fix: {item.fix_command}")
lines.append("")
lines.append(f"Preview all fixes: drone @spawn repair @{project_name} --dry-run")
return "\n".join(lines)
# =============================================================================
# JSON FORMATTING
# =============================================================================
def format_json_report(items: List[RemediationItem], project_name: str) -> str:
"""Format remediation items as JSON for spawn consumption."""
report = {
"project": project_name,
"total_issues": len(items),
"critical_count": sum(1 for i in items if i.severity == "critical"),
"warning_count": sum(1 for i in items if i.severity == "warning"),
"info_count": sum(1 for i in items if i.severity == "info"),
"issues": [
{
"severity": item.severity,
"category": item.category,
"description": item.description,
"fix_command": item.fix_command,
}
for item in items
],
}
return json.dumps(report, indent=2)
# =============================================================================
# RICH OUTPUT
# =============================================================================
_LINE_STYLES = {
"[CRITICAL]": "bold red",
"[WARNING]": "yellow",
"[INFO]": "blue",
" Fix:": "green",
"Preview": "dim",
"STRUCTURE": "bold",
}
def _style_line(line: str) -> str:
"""Apply Rich markup to a remediation report line."""
for prefix, style in _LINE_STYLES.items():
if line.startswith(prefix):
content = line.strip() if prefix == " Fix:" else line
indent = " " if prefix == " Fix:" else " "
return f"{indent}[{style}]{content}[/{style}]"
return f" {line}"
def print_remediation_report(project_root: Path) -> int:
"""Print Rich-formatted remediation report. Returns issue count."""
project_name = detect_project_name(project_root)
items = generate_remediation(project_root)
if not items:
console.print()
console.print("[green]No structure issues requiring repair.[/green]")
return 0
console.print()
console.print("[bold cyan]Remediation Report[/bold cyan]")
console.print()
report = format_text_report(items, project_name)
for line in report.split("\n"):
console.print(_style_line(line))
return len(items)
def print_json_report(project_root: Path) -> int:
"""Print JSON remediation report to stdout. Returns issue count."""
project_name = detect_project_name(project_root)
items = generate_remediation(project_root)
console.print(format_json_report(items, project_name))
return len(items)
# =============================================================================
# OUTPUT FORMATTING
# =============================================================================
def print_introspection() -> None:
"""Display module info for doctor_fix."""
console.print()
console.print("[bold cyan]doctor_fix Module[/bold cyan]")
console.print("Structure remediation report — doctor --fix / --fix --json")
console.print()
console.print("[yellow]Provides:[/yellow]")
console.print(" [dim]- generate_remediation() — scan + build fix items[/dim]")
console.print(" [dim]- format_text_report() — human-readable output[/dim]")
console.print(" [dim]- format_json_report() — machine-readable for spawn[/dim]")
console.print()
# =============================================================================
# COMMAND HANDLER
# =============================================================================
def handle_command(command: str, args: list[str]) -> bool:
"""Handle command routing. Helper module — no standalone commands.
Args:
command: Command name.
args: Additional arguments.
Returns:
True if handled, False otherwise.
"""
if command != "doctor_fix":
return False
if not args:
print_introspection()
json_handler.log_operation("doctor_fix_info", {"command": command})
return True
if args[0] in ("--help", "-h", "help"):
print_introspection()
json_handler.log_operation("doctor_fix_help", {"command": command})
return True
if args[0] in ("--info", "info"):
print_introspection()
json_handler.log_operation("doctor_fix_info", {"command": command})
return True
json_handler.log_operation("doctor_fix_noop", {"command": command})
return False
+21 -1
View File
@@ -845,7 +845,27 @@ def _handle_init_update(args: list[str]) -> int:
target = Path(args[0]) if args else Path.cwd()
try:
result = update_project(target)
console.print(f"[green]✓[/green] Project updated at {target}")
updated = result.get("updated_files", [])
current = result.get("already_current", [])
if updated:
console.print(f"[green]✓[/green] Updated {len(updated)} file(s):")
for f in updated:
console.print(f" [green]+[/green] {f}")
else:
console.print("[green]✓[/green] All files already current.")
if current:
console.print(f" ({len(current)} already up to date)")
# Heal registry: prune stale entries (e.g. cross-project ../paths)
try:
from aipass.spawn.apps.modules.sync_registry import sync_registry
sync_result = sync_registry(fix=True)
pruned = sync_result.get("stale", [])
if pruned:
console.print(f" [green]Registry healed:[/green] removed {len(pruned)} stale entry(ies)")
except Exception as sync_exc:
logger.warning("[init_flow] registry sync during update skipped: %s", sync_exc)
json_handler.log_operation("aipass_init_update", {"target": str(target), "result": result})
return 0
except Exception as exc:
+41 -48
View File
@@ -123,8 +123,11 @@ def test_init_project_creates_all_expected_files(tmp_path):
# No project-level mailbox (agents have their own)
assert not (target / ".ai_mail.local").exists(), ".ai_mail.local/ should NOT be at project level"
# 10 items + 1 command (prep.md) + 7 shipped hooks + package_dir + __init__.py = 19
assert len(result["created_files"]) == 19
# Every expected file must appear in created_files (extras like .venv are env-dependent)
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"]) >= 19
def test_init_project_return_dict_structure(tmp_path):
@@ -199,7 +202,7 @@ def test_init_project_no_local_prompt(tmp_path):
def test_init_project_claude_md_content(tmp_path):
"""CLAUDE.md contains real AIPass project content."""
"""CLAUDE.md is copied from AIPass source of truth."""
target = tmp_path / "proj"
target.mkdir()
@@ -207,31 +210,19 @@ def test_init_project_claude_md_content(tmp_path):
md_path = target / "CLAUDE.md"
content = md_path.read_text(encoding="utf-8")
assert "# ZETA" in content
assert "## What is AIPass" in content
assert "## Getting Started" in content
assert "## Available Commands" in content
assert "## Startup Protocol" in content
startup_idx = content.index("## Startup Protocol")
startup_section = content[startup_idx:]
assert ".trinity/passport.json" in startup_section
assert "init_report.json" in startup_section
assert "aipass init agent" in content
assert "ZETA_REGISTRY.json" in content
assert "# AIPass" in content
assert "Multi-agent framework" in content
def test_init_project_rerunnable_skips_existing_registry(tmp_path):
"""Running init twice skips the existing registry and reuses its ID."""
def test_init_project_rerunnable_blocked_by_guard(tmp_path):
"""Running init twice raises RuntimeError due to _guard_init."""
target = tmp_path / "proj"
target.mkdir()
result1 = init_project(target, project_name="dup")
result2 = init_project(target, project_name="dup")
init_project(target, project_name="dup")
# Same registry ID reused
assert result1["registry_id"] == result2["registry_id"]
# Second run creates no new files (all already exist)
assert len(result2["created_files"]) == 0
with pytest.raises(RuntimeError, match="already an AIPass project"):
init_project(target, project_name="dup")
def test_init_project_raises_on_empty_name(tmp_path):
@@ -244,28 +235,27 @@ def test_init_project_raises_on_empty_name(tmp_path):
def test_init_project_agents_md_content(tmp_path):
"""AGENTS.md contains Codex-equivalent content."""
"""AGENTS.md is copied from AIPass source of truth."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="alpha")
content = (target / "AGENTS.md").read_text(encoding="utf-8")
assert "# ALPHA" in content
assert "ALPHA_REGISTRY.json" in content
assert "aipass init agent" in content
assert "# AIPass" in content
assert "Multi-agent framework" in content
def test_init_project_gemini_md_content(tmp_path):
"""GEMINI.md contains Gemini-equivalent content."""
"""GEMINI.md is copied from AIPass source of truth."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="alpha")
content = (target / "GEMINI.md").read_text(encoding="utf-8")
assert "# ALPHA" in content
assert "ALPHA_REGISTRY.json" in content
assert "# AIPass" in content
assert "Multi-agent framework" in content
def test_init_project_gitignore_content(tmp_path):
@@ -332,7 +322,7 @@ def test_init_project_global_prompt_content(tmp_path):
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
assert "# Commands" in content
def test_init_project_readme_md_content(tmp_path):
@@ -358,7 +348,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"]) == 19
assert len(result["created_files"]) >= 19
def test_init_project_defaults_name_from_directory(tmp_path):
@@ -410,8 +400,8 @@ def test_init_project_skips_existing_optional_files(tmp_path):
result = init_project(target, project_name="eta")
# Registry + prep.md + 7 shipped hooks + package_dir + __init__.py = 11
assert len(result["created_files"]) == 11
# Only non-pre-existing files should be created (registry, hooks, package dir, etc.)
assert len(result["created_files"]) >= 11
# Verify pre-existing files were NOT overwritten
md_content = (target / "CLAUDE.md").read_text(encoding="utf-8")
@@ -419,7 +409,7 @@ def test_init_project_skips_existing_optional_files(tmp_path):
def test_init_project_no_overwrite(tmp_path):
"""Init does not overwrite existing files — re-runnable safety."""
"""Init guard blocks re-init; update preserves content."""
target = tmp_path / "proj"
target.mkdir()
@@ -427,13 +417,13 @@ def test_init_project_no_overwrite(tmp_path):
result1 = init_project(target, project_name="safe")
assert len(result1["created_files"]) > 0
# Second run creates nothing — all files skipped
result2 = init_project(target, project_name="safe")
assert len(result2["created_files"]) == 0
# Second run is blocked by _guard_init
with pytest.raises(RuntimeError, match="already an AIPass project"):
init_project(target, project_name="safe")
# Content from first run is preserved
# Content from first run is preserved (copied from AIPass source)
claude_md = (target / "CLAUDE.md").read_text(encoding="utf-8")
assert "SAFE" in claude_md
assert "# AIPass" in claude_md
def test_init_project_returns_dict(tmp_path):
@@ -447,15 +437,16 @@ def test_init_project_returns_dict(tmp_path):
def test_init_project_agents_md_no_trinity(tmp_path):
"""AGENTS.md startup protocol references registry, not .trinity/."""
"""AGENTS.md is copied from AIPass source (may reference .trinity/ as part of agent docs)."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="keep")
content = (target / "AGENTS.md").read_text(encoding="utf-8")
assert ".trinity/" not in content
assert "KEEP_REGISTRY.json" in content
# Source file legitimately references .trinity/ as part of startup protocol docs
assert "# AIPass" in content
assert "Multi-agent framework" in content
# ---------------------------------------------------------------------------
@@ -537,10 +528,10 @@ def test_update_project_updates_modified_managed_file(tmp_path):
assert str(claude_md.resolve()) in result["updated_files"]
assert str(claude_md.resolve()) not in result["already_current"]
# Content is restored
# Content is restored from AIPass source of truth
restored = claude_md.read_text(encoding="utf-8")
assert "MOD" in restored
assert "## What is AIPass" in restored
assert "# AIPass" in restored
assert "Multi-agent framework" in restored
def test_update_project_never_touches_user_owned_files(tmp_path):
@@ -752,12 +743,11 @@ def test_update_project_resyncs_hooks(tmp_path):
def test_init_project_hooks_idempotent_on_rerun(tmp_path):
"""Re-running init does not re-ship hooks when content is identical."""
"""Re-running update does not re-ship hooks when content is identical."""
target = tmp_path / "proj"
target.mkdir()
result1 = init_project(target, project_name="idem")
result2 = init_project(target, project_name="idem")
if result1["aipass_home"] is None:
pytest.skip("AIPASS_HOME not detectable in this environment")
@@ -766,7 +756,10 @@ def test_init_project_hooks_idempotent_on_rerun(tmp_path):
hooks_marker = str(Path(".claude") / "hooks")
hook_paths = [f for f in result1["created_files"] if hooks_marker in f]
assert len(hook_paths) == 7
hook_paths_rerun = [f for f in result2["created_files"] if hooks_marker in f]
# Update should not re-ship hooks (content identical)
result2 = update_project(target)
hook_paths_rerun = [f for f in result2["updated_files"] if hooks_marker in f]
assert len(hook_paths_rerun) == 0
+383
View File
@@ -0,0 +1,383 @@
# =================== AIPass ====================
# Name: test_doctor_fix.py
# Description: Tests for doctor --fix remediation report (DPLAN-0177 Phase 2)
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""Tests for doctor_fix — remediation generation, text/JSON formatting, severity classification."""
import json
from pathlib import Path
from unittest.mock import patch
from aipass.aipass.apps.modules.doctor_fix import (
RemediationItem,
detect_project_name,
format_json_report,
format_text_report,
generate_remediation,
print_json_report,
print_remediation_report,
)
# =============================================================================
# Helpers
# =============================================================================
def _make_agent(tmp_path: Path, name: str, registry_id: str = "uuid-1", subdir: str = "") -> Path:
"""Create a minimal agent directory with passport."""
if subdir:
agent_dir = tmp_path / "src" / subdir / name
else:
agent_dir = tmp_path / "src" / name
trinity = agent_dir / ".trinity"
trinity.mkdir(parents=True, exist_ok=True)
passport = {
"branch_info": {"branch_name": name},
"citizenship": {"registry_id": registry_id},
}
(trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8")
return agent_dir
def _make_registry(tmp_path: Path, branches: list, prefix: str = "TEST") -> Path:
"""Create a registry file."""
reg = tmp_path / f"{prefix}_REGISTRY.json"
reg.write_text(json.dumps({"branches": branches}), encoding="utf-8")
return reg
# =============================================================================
# TestDetectProjectName
# =============================================================================
class TestDetectProjectName:
def test_from_registry(self, tmp_path: Path) -> None:
"""Derives name from COMPASS_REGISTRY.json → compass."""
_make_registry(tmp_path, [], prefix="COMPASS")
result = detect_project_name(tmp_path)
assert result == "compass"
def test_fallback_to_dirname(self, tmp_path: Path) -> None:
"""Falls back to directory name when no registry."""
result = detect_project_name(tmp_path)
assert result == tmp_path.name.lower()
def test_registry_name_lowered(self, tmp_path: Path) -> None:
"""Registry name is lowercased."""
_make_registry(tmp_path, [], prefix="AIPASS")
result = detect_project_name(tmp_path)
assert result == "aipass"
# =============================================================================
# TestGenerateRemediation
# =============================================================================
class TestGenerateRemediation:
def test_clean_project_empty(self, tmp_path: Path) -> None:
"""Clean project produces no remediation items."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
assert items == []
def test_pollution_is_critical(self, tmp_path: Path) -> None:
"""Duplicate registry_id produces critical severity."""
_make_agent(tmp_path, "orig", "uuid-dup")
_make_agent(tmp_path, "copy", "uuid-dup", subdir="pkg")
_make_registry(tmp_path, [])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
pollution = [i for i in items if i.category == "pollution"]
assert len(pollution) == 1
assert pollution[0].severity == "critical"
def test_pollution_command_uses_clean(self, tmp_path: Path) -> None:
"""Pollution fix command uses --clean-pollution."""
_make_agent(tmp_path, "orig", "uuid-dup")
_make_agent(tmp_path, "copy", "uuid-dup", subdir="pkg")
_make_registry(tmp_path, [], prefix="COMPASS")
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
pollution = [i for i in items if i.category == "pollution"]
assert "--clean-pollution" in pollution[0].fix_command
assert "@compass" in pollution[0].fix_command
def test_placement_is_warning(self, tmp_path: Path) -> None:
"""Misplaced agent produces warning severity."""
agent_dir = tmp_path / "stray" / "agent"
trinity = agent_dir / ".trinity"
trinity.mkdir(parents=True)
passport = {"branch_info": {"branch_name": "stray"}, "citizenship": {"registry_id": "uuid-s"}}
(trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8")
_make_registry(tmp_path, [])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
placement = [i for i in items if i.category == "placement"]
assert len(placement) >= 1
assert placement[0].severity == "warning"
def test_placement_command_uses_relocate(self, tmp_path: Path) -> None:
"""Placement fix command uses --relocate."""
agent_dir = tmp_path / "other" / "agent"
trinity = agent_dir / ".trinity"
trinity.mkdir(parents=True)
passport = {"branch_info": {"branch_name": "stray"}, "citizenship": {"registry_id": "uuid-s"}}
(trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8")
_make_registry(tmp_path, [], prefix="TEST")
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
placement = [i for i in items if i.category == "placement"]
assert any("--relocate" in i.fix_command for i in placement)
def test_registry_missing_path_is_warning(self, tmp_path: Path) -> None:
"""Missing registry path produces warning severity."""
_make_registry(tmp_path, [{"name": "ghost", "path": str(tmp_path / "src" / "ghost")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
registry = [i for i in items if i.category == "registry"]
assert len(registry) >= 1
assert registry[0].severity == "warning"
def test_registry_command_uses_dedup(self, tmp_path: Path) -> None:
"""Registry fix command uses --dedup-registry."""
_make_registry(tmp_path, [{"name": "ghost", "path": str(tmp_path / "src" / "ghost")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
items = generate_remediation(tmp_path)
registry = [i for i in items if i.category == "registry"]
assert "--dedup-registry" in registry[0].fix_command
def test_root_artifact_warn(self, tmp_path: Path) -> None:
"""Root artifact with warn severity produces warning remediation."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
(tmp_path / ".chroma").mkdir()
items = generate_remediation(tmp_path)
root_items = [i for i in items if i.category == "root_artifact"]
assert len(root_items) == 1
assert root_items[0].severity == "warning"
assert ".chroma" in root_items[0].description
def test_root_artifact_info(self, tmp_path: Path) -> None:
""".venv root artifact maps to info severity."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
(tmp_path / ".venv").mkdir()
items = generate_remediation(tmp_path)
root_items = [i for i in items if i.category == "root_artifact"]
assert len(root_items) == 1
assert root_items[0].severity == "info"
def test_root_artifact_command(self, tmp_path: Path) -> None:
"""Root artifact fix command uses --relocate-root."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}], prefix="TEST")
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
(tmp_path / "logs").mkdir()
items = generate_remediation(tmp_path)
root_items = [i for i in items if i.category == "root_artifact"]
assert "--relocate-root logs" in root_items[0].fix_command
assert "@test" in root_items[0].fix_command
def test_missing_pyproject_is_info(self, tmp_path: Path) -> None:
"""Missing pyproject.toml produces info severity."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
items = generate_remediation(tmp_path)
pyproject = [i for i in items if i.category == "pyproject"]
assert len(pyproject) == 1
assert pyproject[0].severity == "info"
def test_pyproject_command_uses_add(self, tmp_path: Path) -> None:
"""Pyproject fix command uses --add-pyproject."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
items = generate_remediation(tmp_path)
pyproject = [i for i in items if i.category == "pyproject"]
assert "--add-pyproject" in pyproject[0].fix_command
# =============================================================================
# TestFormatTextReport
# =============================================================================
class TestFormatTextReport:
def test_empty_items(self) -> None:
"""No items produces 'no issues' message."""
result = format_text_report([], "compass")
assert "No structure issues" in result
assert "@compass" in result
def test_header_shows_counts(self) -> None:
"""Header shows total and critical counts."""
items = [
RemediationItem("critical", "pollution", "dup", "fix1"),
RemediationItem("warning", "placement", "bad", "fix2"),
]
result = format_text_report(items, "test")
assert "2 found" in result
assert "1 critical" in result
def test_severity_tags_present(self) -> None:
"""Each item has [SEVERITY] tag."""
items = [
RemediationItem("critical", "pollution", "dup", "fix1"),
RemediationItem("warning", "placement", "bad", "fix2"),
RemediationItem("info", "pyproject", "missing", "fix3"),
]
result = format_text_report(items, "test")
assert "[CRITICAL]" in result
assert "[WARNING]" in result
assert "[INFO]" in result
def test_fix_commands_present(self) -> None:
"""Fix commands appear in output."""
items = [RemediationItem("warning", "placement", "bad", "drone @spawn repair @test --relocate a b")]
result = format_text_report(items, "test")
assert "drone @spawn repair @test --relocate a b" in result
def test_dry_run_hint(self) -> None:
"""Report ends with dry-run suggestion."""
items = [RemediationItem("info", "pyproject", "missing", "fix")]
result = format_text_report(items, "myproj")
assert "drone @spawn repair @myproj --dry-run" in result
def test_critical_sorted_first(self) -> None:
"""Critical items appear before warning and info."""
items = [
RemediationItem("info", "pyproject", "missing", "fix3"),
RemediationItem("critical", "pollution", "dup", "fix1"),
RemediationItem("warning", "placement", "bad", "fix2"),
]
result = format_text_report(items, "test")
crit_pos = result.index("[CRITICAL]")
warn_pos = result.index("[WARNING]")
info_pos = result.index("[INFO]")
assert crit_pos < warn_pos < info_pos
# =============================================================================
# TestFormatJsonReport
# =============================================================================
class TestFormatJsonReport:
def test_valid_json(self) -> None:
"""Output is valid parseable JSON."""
items = [RemediationItem("critical", "pollution", "dup", "fix1")]
result = format_json_report(items, "test")
parsed = json.loads(result)
assert isinstance(parsed, dict)
def test_json_structure(self) -> None:
"""JSON has expected top-level keys."""
items = [
RemediationItem("critical", "pollution", "dup", "fix1"),
RemediationItem("warning", "placement", "bad", "fix2"),
]
result = json.loads(format_json_report(items, "compass"))
assert result["project"] == "compass"
assert result["total_issues"] == 2
assert result["critical_count"] == 1
assert result["warning_count"] == 1
assert result["info_count"] == 0
def test_json_issues_array(self) -> None:
"""JSON issues array has correct fields per item."""
items = [RemediationItem("info", "pyproject", "missing toml", "drone @spawn repair @t --add-pyproject")]
result = json.loads(format_json_report(items, "t"))
issue = result["issues"][0]
assert issue["severity"] == "info"
assert issue["category"] == "pyproject"
assert issue["description"] == "missing toml"
assert issue["fix_command"] == "drone @spawn repair @t --add-pyproject"
def test_empty_report(self) -> None:
"""Empty items produces valid JSON with zero counts."""
result = json.loads(format_json_report([], "test"))
assert result["total_issues"] == 0
assert result["issues"] == []
def test_spawn_commands_in_json(self) -> None:
"""Fix commands in JSON match spawn's CLI interface."""
items = [
RemediationItem("critical", "pollution", "dup", "drone @spawn repair @p --clean-pollution"),
RemediationItem("warning", "registry", "miss", "drone @spawn repair @p --dedup-registry"),
RemediationItem("warning", "placement", "bad", "drone @spawn repair @p --relocate a b"),
]
result = json.loads(format_json_report(items, "p"))
commands = [i["fix_command"] for i in result["issues"]]
assert all(cmd.startswith("drone @spawn repair @p") for cmd in commands)
# =============================================================================
# TestPrintFunctions
# =============================================================================
class TestPrintFunctions:
def test_print_remediation_returns_count(self, tmp_path: Path) -> None:
"""print_remediation_report returns issue count."""
_make_agent(tmp_path, "orig", "uuid-dup")
_make_agent(tmp_path, "copy", "uuid-dup", subdir="pkg")
_make_registry(tmp_path, [])
count = print_remediation_report(tmp_path)
assert count >= 1
def test_print_remediation_clean_returns_zero(self, tmp_path: Path) -> None:
"""Clean project returns 0."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
count = print_remediation_report(tmp_path)
assert count == 0
def test_print_json_returns_count(self, tmp_path: Path) -> None:
"""print_json_report returns issue count."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
count = print_json_report(tmp_path)
assert count >= 1
# =============================================================================
# TestDoctorFixHandleCommand
# =============================================================================
class TestDoctorFixHandleCommand:
def test_wrong_command(self) -> None:
"""Non-doctor_fix commands are not handled."""
from aipass.aipass.apps.modules.doctor_fix import handle_command
assert handle_command("doctor", []) is False
assert handle_command("help", []) is False
def test_no_args_calls_introspection(self) -> None:
"""No args triggers print_introspection."""
from aipass.aipass.apps.modules.doctor_fix import handle_command
with patch("aipass.aipass.apps.modules.doctor_fix.print_introspection") as mock:
result = handle_command("doctor_fix", [])
assert result is True
mock.assert_called_once()
def test_info_flag(self) -> None:
"""--info triggers print_introspection."""
from aipass.aipass.apps.modules.doctor_fix import handle_command
with patch("aipass.aipass.apps.modules.doctor_fix.print_introspection") as mock:
result = handle_command("doctor_fix", ["--info"])
assert result is True
mock.assert_called_once()
@@ -13,9 +13,11 @@ from pathlib import Path
from unittest.mock import patch
from aipass.aipass.apps.handlers.structure_scan.structure_scanner import (
_detect_package_names,
check_placement,
check_pyproject,
check_registry_consistency,
check_root_artifacts,
detect_pollution,
find_project_root,
find_registry,
@@ -162,6 +164,113 @@ class TestCheckPlacement:
assert issues[0].agent_name == "stray"
# =============================================================================
# TestDetectPackageNames
# =============================================================================
class TestDetectPackageNames:
def test_no_pyproject(self, tmp_path: Path) -> None:
"""Returns empty set when no pyproject.toml."""
result = _detect_package_names(tmp_path)
assert result == set()
def test_hatch_packages(self, tmp_path: Path) -> None:
"""Detects package from hatch build config."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.hatch.build.targets.wheel]\npackages = ["src/aipl"]\n',
encoding="utf-8",
)
result = _detect_package_names(tmp_path)
assert "aipl" in result
def test_setuptools_packages(self, tmp_path: Path) -> None:
"""Detects package from setuptools config."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.setuptools]\npackages = ["mypackage"]\n',
encoding="utf-8",
)
result = _detect_package_names(tmp_path)
assert "mypackage" in result
def test_corrupt_pyproject(self, tmp_path: Path) -> None:
"""Returns empty set for corrupt TOML."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text("not valid toml {{{", encoding="utf-8")
result = _detect_package_names(tmp_path)
assert result == set()
def test_pyproject_without_packages(self, tmp_path: Path) -> None:
"""Returns empty set when pyproject has no package declarations."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text('[project]\nname = "test"\n', encoding="utf-8")
result = _detect_package_names(tmp_path)
assert result == set()
# =============================================================================
# TestPackageAwarePlacement
# =============================================================================
class TestPackageAwarePlacement:
def test_agent_inside_package_passes(self, tmp_path: Path) -> None:
"""Agent at src/<pkg>/<agent>/ passes when package is defined."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.hatch.build.targets.wheel]\npackages = ["src/aipl"]\n',
encoding="utf-8",
)
_make_agent(tmp_path, "polyglot", subdir="aipl")
agents = scan_agents(tmp_path)
issues = check_placement(agents, tmp_path)
assert issues == []
def test_agent_sibling_of_package_warned(self, tmp_path: Path) -> None:
"""Agent at src/<other>/ flagged when package exists at src/<pkg>/."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.hatch.build.targets.wheel]\npackages = ["src/aipl"]\n',
encoding="utf-8",
)
_make_agent(tmp_path, "polyglot")
agents = scan_agents(tmp_path)
issues = check_placement(agents, tmp_path)
assert len(issues) == 1
assert issues[0].agent_name == "polyglot"
assert "outside package framework" in issues[0].expected_pattern
def test_package_dir_itself_passes(self, tmp_path: Path) -> None:
"""Package dir at src/<pkg>/ with passport passes (it IS the package)."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
'[tool.hatch.build.targets.wheel]\npackages = ["src/aipl"]\n',
encoding="utf-8",
)
_make_agent(tmp_path, "aipl")
agents = scan_agents(tmp_path)
issues = check_placement(agents, tmp_path)
assert issues == []
def test_no_pyproject_unchanged(self, tmp_path: Path) -> None:
"""Without pyproject, src/<agent>/ still passes (original behavior)."""
_make_agent(tmp_path, "myagent")
agents = scan_agents(tmp_path)
issues = check_placement(agents, tmp_path)
assert issues == []
def test_pyproject_without_packages_unchanged(self, tmp_path: Path) -> None:
"""pyproject without package declarations doesn't flag anything."""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text('[project]\nname = "test"\n', encoding="utf-8")
_make_agent(tmp_path, "myagent")
agents = scan_agents(tmp_path)
issues = check_placement(agents, tmp_path)
assert issues == []
# =============================================================================
# TestDetectPollution
# =============================================================================
@@ -280,6 +389,49 @@ class TestCheckPyproject:
assert result["path"] == ""
# =============================================================================
# TestCheckRootArtifacts
# =============================================================================
class TestCheckRootArtifacts:
def test_no_artifacts(self, tmp_path: Path) -> None:
"""Clean root returns empty list."""
hits = check_root_artifacts(tmp_path)
assert hits == []
def test_chroma_detected(self, tmp_path: Path) -> None:
"""Detects .chroma/ at project root."""
(tmp_path / ".chroma").mkdir()
hits = check_root_artifacts(tmp_path)
assert len(hits) == 1
assert hits[0].name == ".chroma"
assert hits[0].artifact_type == "chroma"
assert hits[0].severity == "warn"
def test_venv_is_info(self, tmp_path: Path) -> None:
""".venv/ at root is info severity, not warn."""
(tmp_path / ".venv").mkdir()
hits = check_root_artifacts(tmp_path)
assert len(hits) == 1
assert hits[0].severity == "info"
def test_multiple_artifacts(self, tmp_path: Path) -> None:
"""Multiple misplaced items all detected."""
(tmp_path / ".chroma").mkdir()
(tmp_path / "logs").mkdir()
(tmp_path / ".ai_mail.local").mkdir()
hits = check_root_artifacts(tmp_path)
names = {h.name for h in hits}
assert names == {".chroma", "logs", ".ai_mail.local"}
def test_nonexistent_ignored(self, tmp_path: Path) -> None:
"""Only existing artifacts are reported."""
(tmp_path / ".chroma").mkdir()
hits = check_root_artifacts(tmp_path)
assert all(h.name != "logs" for h in hits)
# =============================================================================
# TestCheckStructureIntegration
# =============================================================================
@@ -309,6 +461,18 @@ class TestCheckStructureIntegration:
assert GLYPH_FAIL not in glyphs
assert GLYPH_WARN not in glyphs
def test_root_artifacts_reported(self, tmp_path: Path) -> None:
"""Root artifacts show up as WARN in structure check."""
_make_agent(tmp_path, "agent1", "uuid-1")
_make_registry(tmp_path, [{"name": "agent1", "path": str(tmp_path / "src" / "agent1")}])
(tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8")
(tmp_path / ".chroma").mkdir()
with patch("aipass.aipass.apps.modules.doctor.find_project_root", return_value=tmp_path):
results = _check_structure()
root_results = [r for r in results if "root:" in r.label]
assert len(root_results) >= 1
assert root_results[0].glyph == GLYPH_WARN
def test_pollution_reported(self, tmp_path: Path) -> None:
"""Duplicate registry_id shows up as FAIL in structure check."""
_make_agent(tmp_path, "orig", "uuid-dup")
@@ -17,7 +17,7 @@
"max_key_learnings": 25,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"note": "Auto-rollover to @memory when limits exceeded. Oldest entries trimmed first."
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries."
},
"status": {
"health": "healthy",
@@ -14,7 +14,7 @@
],
"limits": {
"max_lines": 600,
"note": "Auto-rollover when max_lines exceeded"
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
},
"status": {
"health": "healthy",
+36
View File
@@ -210,6 +210,42 @@
"standard": "windows_compat",
"lines": [249],
"reason": "fcntl import at L249 inside 'if lock_fd is not None:' — lock_fd is None on Windows (set at L206), so this line never executes on Windows."
},
{
"file": "apps/handlers/repair_ops.py",
"standard": "handlers",
"reason": "Imports registry functions — repair needs to load/save registry and update branch paths. Registry is a shared data handler."
},
{
"file": "apps/handlers/delete_ops.py",
"standard": "handlers",
"reason": "Imports ARCHIVE_EXCLUDE from repair_ops — shared constant for copytree ignore patterns. Avoids duplicating the exclude set across handlers."
},
{
"file": "tests/test_repair.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_repair.py",
"standard": "encapsulation",
"reason": "Test file — tests must import handlers directly to test them in isolation. Handler imports inside test methods are intentional."
},
{
"file": "tests/test_lifecycle.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_lifecycle.py",
"standard": "encapsulation",
"reason": "Test file — tests must import handlers directly to test them in isolation. Handler imports inside test methods are intentional."
},
{
"file": "apps/handlers/repair_ops.py",
"standard": "windows_compat",
"lines": [89],
"reason": "fcntl import guarded by sys.platform != 'win32' check — lock_fd is None on Windows, so this line never executes."
}
],
"notes": {
+10 -3
View File
@@ -78,6 +78,13 @@ drone @spawn sync-registry --fix # Rebuild .spawn/
drone @spawn sync-templates # Pull managed files from sources (partial — see Known Issues)
drone @spawn regenerate-registry # Regenerate builder template hashes
drone @spawn regenerate-registry --all # All template classes
# Repair
drone @spawn repair <project_path> # Scan project for structural issues
drone @spawn repair <project_path> --dry-run # Preview only
drone @spawn repair --relocate @branch src/pkg/branch # Move branch to new location
drone @spawn repair --relocate @branch path --relocate-artifacts # Move branch + .chroma/ into it
drone @spawn repair <project_path> --clean-pollution # Archive + remove duplicate dirs
```
### Introspection
@@ -140,7 +147,7 @@ spawn/
├── templates/
│ ├── builder/ # Full scaffold template (45 files, 24 dirs)
│ └── birthright/ # Minimal template
├── tests/ # 13 test files, 253 tests
├── tests/ # 14 test files, 316 tests
├── spawn_json/ # JSON tracking directory
├── tools/ # Branch verification utilities
├── docs/ # Documentation
@@ -191,7 +198,7 @@ spawn/
## Tests
**253 tests | 0 skipped | 0 failed** across 13 test files:
**316 tests | 0 skipped | 0 failed** across 14 test files:
| File | Focus |
|------|-------|
@@ -245,6 +252,6 @@ spawn/
---
*Last Updated: 2026-04-22*
*Last Updated: 2026-05-15*
[← Back to AIPass](../../../README.md)
+5 -4
View File
@@ -21,8 +21,9 @@ from aipass.spawn.apps.handlers.registry import (
find_registry,
load_registry,
save_registry,
_branches_as_list,
branches_as_list,
)
from aipass.spawn.apps.handlers.repair_ops import ARCHIVE_EXCLUDE
from aipass.spawn.apps.handlers.json import json_handler
# Branches that cannot be deleted (critical infrastructure)
@@ -41,7 +42,7 @@ def _resolve_branch_dir(branch_name, registry_path, registry):
Tuple of (branch_entry, branch_dir) or (None, None) if not found.
"""
project_root = registry_path.parent
for entry in _branches_as_list(registry.get("branches", [])):
for entry in branches_as_list(registry.get("branches", [])):
if entry.get("name", "").lower() == branch_name.lower():
rel_path = entry.get("path", "")
branch_dir = (project_root / rel_path).resolve() if rel_path else None
@@ -57,7 +58,7 @@ def _archive_branch(branch_dir, archive_dir):
"""
archive_dir.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.copytree(str(branch_dir), str(archive_dir))
shutil.copytree(str(branch_dir), str(archive_dir), ignore=shutil.ignore_patterns(*ARCHIVE_EXCLUDE))
logger.info(f"[delete] Archived to {archive_dir}")
return None
except Exception as exc:
@@ -91,7 +92,7 @@ def _remove_from_registry(registry, branch_name, registry_path):
registry["branches"] = branches
else:
registry["branches"] = [b for b in branches if b.get("name", "").lower() != branch_name.lower()]
registry["metadata"]["total_branches"] = len(_branches_as_list(registry["branches"]))
registry["metadata"]["total_branches"] = len(branches_as_list(registry["branches"]))
return save_registry(registry_path, registry)
+4 -4
View File
@@ -17,7 +17,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.spawn.apps.handlers.json import json_handler
def _branches_as_list(branches):
def branches_as_list(branches):
"""Normalize branches to a list regardless of storage format.
The registry may store branches as:
@@ -159,7 +159,7 @@ def get_next_citizen_number(registry_path):
"""
data = load_registry(registry_path)
branches = data.get("branches", [])
return len(_branches_as_list(branches)) + 1
return len(branches_as_list(branches)) + 1
def _validate_path_containment(branch_path, registry_path):
@@ -239,7 +239,7 @@ def add_to_registry(registry_path, branch_name, branch_path, profile, email, pur
else:
branches.append(entry)
registry["branches"] = branches
registry["metadata"]["total_branches"] = len(_branches_as_list(branches))
registry["metadata"]["total_branches"] = len(branches_as_list(branches))
json_handler.log_operation("registry_updated", data={"branch": branch_name})
@@ -310,7 +310,7 @@ def ensure_project_has_owner(registry_path):
"""If no agent in the project has owner:true, assign it to the earliest-created agent."""
registry_path = Path(registry_path)
reg_data = load_registry(registry_path)
branches = _branches_as_list(reg_data.get("branches", []))
branches = branches_as_list(reg_data.get("branches", []))
if not branches:
return False
@@ -0,0 +1,486 @@
# =================== AIPass ====================
# Name: repair_ops.py
# Description: Repair handler — move branches, update registry paths, clean pollution
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""Repair handler implementation for project structure fixes.
Contains core logic for moving branches to correct locations, updating
registry paths without re-creating entries, and detecting/cleaning init
pollution (duplicate nested directories).
"""
import shutil
import sys
from datetime import datetime
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
ARCHIVE_EXCLUDE = {".venv", ".git", "__pycache__", ".chroma", "node_modules", ".pytest_cache"}
from aipass.spawn.apps.handlers.registry import (
find_registry,
load_registry,
save_registry,
branches_as_list,
)
from aipass.spawn.apps.handlers.json import json_handler
# =============================================================================
# REGISTRY PATH UPDATE
# =============================================================================
def update_registry_path(registry_path, branch_name, new_path):
"""Update a branch's path in the registry without re-creating the entry.
Preserves creation date, citizen_number, status, and all other fields.
Uses file locking to prevent concurrent corruption.
Args:
registry_path: Path to *_REGISTRY.json
branch_name: Branch name (case-insensitive match)
new_path: New relative path for the branch
Returns:
True if updated, False if branch not found or error.
"""
registry_path = Path(registry_path)
lock_path = registry_path.parent / f".{registry_path.stem}.lock"
lock_fd = None
if sys.platform != "win32":
import fcntl # noqa: windows_compat — guarded by platform check
lock_fd = open(lock_path, "w", encoding="utf-8") # noqa: SIM115
fcntl.flock(lock_fd, fcntl.LOCK_EX)
try:
registry = load_registry(registry_path)
branches = registry.get("branches", [])
branch_list = branches_as_list(branches)
updated = False
for entry in branch_list:
if entry.get("name", "").lower() == branch_name.lower():
old_path = entry.get("path", "")
entry["path"] = Path(new_path).as_posix()
entry["last_active"] = datetime.now().strftime("%Y-%m-%d")
logger.info(
"[repair] Updated registry path for %s: %s → %s",
branch_name,
old_path,
new_path,
)
updated = True
break
if not updated:
logger.warning("[repair] Branch '%s' not found in registry", branch_name)
return False
registry["branches"] = branch_list
return save_registry(registry_path, registry)
finally:
if lock_fd is not None:
import fcntl
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
# =============================================================================
# PASSPORT PATH UPDATE
# =============================================================================
def _update_passport_paths(branch_dir, new_relative_path, project_name=None):
"""Update passport.json branch_info paths after a move.
Args:
branch_dir: New branch directory location
new_relative_path: New path relative to project root
project_name: Project package name (for module path)
Returns:
True if updated, False if passport missing or error.
"""
passport_path = branch_dir / ".trinity" / "passport.json"
if not passport_path.exists():
return False
try:
passport = json_handler.read_json(passport_path)
if passport is None:
return False
branch_info = passport.get("branch_info", {})
branch_info["path"] = new_relative_path
if project_name:
branch_name = branch_dir.name
branch_info["module"] = f"{project_name}.{branch_name}"
passport["branch_info"] = branch_info
return json_handler.write_json(passport_path, passport)
except Exception as e:
logger.error("[repair] Failed to update passport paths: %s", e)
return False
def _detect_project_package(branch_rel_path):
"""Detect the project package name from the branch's relative path.
E.g., "src/compass/navigator" → "compass"
"""
parts = Path(branch_rel_path).parts
if len(parts) >= 2 and parts[0] == "src":
return parts[1]
return None
# =============================================================================
# ARTIFACT RELOCATION
# =============================================================================
def _relocate_chroma(project_root, branch_dir, branches):
"""Move project-root .chroma/ into a branch when only one branch exists.
Only relocates when .chroma/ exists at project root AND the registry
has exactly one branch (the one being moved).
Args:
project_root: Project root directory
branch_dir: Target branch directory (already moved)
branches: List of branch entries from registry
Returns:
True if relocated, False otherwise.
"""
chroma_src = project_root / ".chroma"
if not chroma_src.is_dir():
return False
if len(branches) != 1:
logger.info("[repair] Skipping .chroma relocation — %d branches (need exactly 1)", len(branches))
return False
chroma_dest = branch_dir / ".chroma"
if chroma_dest.exists():
logger.warning("[repair] .chroma already exists in branch dir, skipping relocation")
return False
try:
shutil.move(str(chroma_src), str(chroma_dest))
logger.info("[repair] Relocated .chroma/ into %s", branch_dir.name)
return True
except Exception as e:
logger.error("[repair] Failed to relocate .chroma: %s", e)
return False
# =============================================================================
# BRANCH MOVE
# =============================================================================
def move_branch(branch_name, new_path, registry_path=None, dry_run=False, relocate_artifacts=False):
"""Move a branch to a new location, updating registry and passport.
Archive-first: creates backup before any destructive operation.
Args:
branch_name: Branch name (looked up in registry)
new_path: New absolute or relative-to-project-root path
registry_path: Path to registry (auto-discovered if None)
dry_run: If True, report what would happen without changes
relocate_artifacts: If True and project has only 1 branch, move .chroma/ into branch
Returns:
Dict with move results.
"""
if registry_path is None:
registry_path = find_registry()
registry_path = Path(registry_path)
project_root = registry_path.parent
registry = load_registry(registry_path)
branches = branches_as_list(registry.get("branches", []))
entry = None
for b in branches:
if b.get("name", "").lower() == branch_name.lower():
entry = b
break
if entry is None:
return {"success": False, "error": f"Branch '{branch_name}' not found in registry"}
old_rel_path = entry.get("path", "")
old_abs_path = (project_root / old_rel_path).resolve()
new_abs_path = Path(new_path)
if not new_abs_path.is_absolute():
new_abs_path = (project_root / new_path).resolve()
try:
new_rel_path = new_abs_path.relative_to(project_root).as_posix()
except ValueError:
logger.warning("[repair] Path %s is outside project root %s", new_abs_path, project_root)
return {
"success": False,
"error": f"New path {new_abs_path} is outside project root {project_root}",
}
if not old_abs_path.exists():
return {"success": False, "error": f"Source directory does not exist: {old_abs_path}"}
if new_abs_path.exists():
return {"success": False, "error": f"Target directory already exists: {new_abs_path}"}
if dry_run:
return {
"success": True,
"dry_run": True,
"branch": branch_name,
"old_path": old_rel_path,
"new_path": new_rel_path,
"actions": [
f"Archive {old_rel_path} to .archive/repair_moves/",
f"Move {old_rel_path} → {new_rel_path}",
"Update registry path",
"Update passport paths",
],
}
# Archive first
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_dir = project_root / ".archive" / "repair_moves" / f"{branch_name}_{timestamp}"
archive_dir.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.copytree(
str(old_abs_path),
str(archive_dir),
ignore=shutil.ignore_patterns(*ARCHIVE_EXCLUDE),
)
logger.info("[repair] Archived %s to %s", branch_name, archive_dir)
except Exception as e:
logger.error("[repair] Archive failed for %s: %s", branch_name, e)
return {"success": False, "error": f"Archive failed: {e}"}
# Move directory
new_abs_path.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.move(str(old_abs_path), str(new_abs_path))
logger.info("[repair] Moved %s → %s", old_abs_path, new_abs_path)
except Exception as e:
logger.error("[repair] Move failed for %s: %s", branch_name, e)
return {"success": False, "error": f"Move failed: {e}", "archive_path": str(archive_dir)}
# Update registry
reg_updated = update_registry_path(registry_path, branch_name, new_rel_path)
# Update passport
project_name = _detect_project_package(new_rel_path)
passport_updated = _update_passport_paths(new_abs_path, new_rel_path, project_name)
# Relocate .chroma into branch if requested and conditions met
chroma_relocated = False
if relocate_artifacts:
chroma_relocated = _relocate_chroma(project_root, new_abs_path, branches)
json_handler.log_operation(
"branch_moved",
data={"branch": branch_name, "old_path": old_rel_path, "new_path": new_rel_path},
)
return {
"success": True,
"branch": branch_name,
"old_path": old_rel_path,
"new_path": new_rel_path,
"archive_path": str(archive_dir),
"registry_updated": reg_updated,
"passport_updated": passport_updated,
"chroma_relocated": chroma_relocated,
}
# =============================================================================
# POLLUTION DETECTION & CLEANUP
# =============================================================================
def detect_pollution(project_root):
"""Detect init pollution — duplicate nested directories.
Init pollution: project_name/project_name/ exists (e.g., compass/compass/).
Args:
project_root: Path to the project root directory
Returns:
List of dicts describing pollution issues.
"""
project_root = Path(project_root)
issues = []
project_name = project_root.name
nested = project_root / project_name
if nested.is_dir():
issues.append(
{
"type": "duplicate_nested_dir",
"path": project_name,
"description": f"Duplicate nested directory: {project_name}/{project_name}/",
}
)
src_dir = project_root / "src"
if src_dir.is_dir():
for child in sorted(src_dir.iterdir()):
if child.is_dir() and not child.name.startswith(".") and not child.name.startswith("__"):
nested_dup = child / child.name
if nested_dup.is_dir():
rel = nested_dup.relative_to(project_root).as_posix()
issues.append(
{
"type": "duplicate_nested_dir",
"path": rel,
"description": f"Duplicate nested directory: src/{child.name}/{child.name}/",
}
)
return issues
def cleanup_pollution(project_root, dry_run=False):
"""Archive and remove detected pollution directories.
Args:
project_root: Path to the project root directory
dry_run: If True, report without changes
Returns:
Dict with cleanup results.
"""
project_root = Path(project_root)
issues = detect_pollution(project_root)
if not issues:
return {"success": True, "issues_found": 0, "cleaned": []}
if dry_run:
return {
"success": True,
"dry_run": True,
"issues_found": len(issues),
"issues": issues,
}
cleaned = []
errors = []
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
for issue in issues:
pollution_path = project_root / issue["path"]
if not pollution_path.exists():
continue
archive_dir = project_root / ".archive" / "pollution" / f"{pollution_path.name}_{timestamp}"
archive_dir.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.copytree(
str(pollution_path),
str(archive_dir),
ignore=shutil.ignore_patterns(*ARCHIVE_EXCLUDE),
)
shutil.rmtree(str(pollution_path))
logger.info("[repair] Cleaned pollution: %s (archived to %s)", issue["path"], archive_dir)
cleaned.append({"path": issue["path"], "archive": str(archive_dir)})
except Exception as e:
logger.error("[repair] Failed to clean pollution at %s: %s", issue["path"], e)
errors.append({"path": issue["path"], "error": str(e)})
json_handler.log_operation(
"pollution_cleaned",
data={"project": project_root.name, "cleaned": len(cleaned)},
)
return {
"success": len(errors) == 0,
"issues_found": len(issues),
"cleaned": cleaned,
"errors": errors,
}
# =============================================================================
# PROJECT SCAN
# =============================================================================
def repair_project(project_path, dry_run=False):
"""Scan a project for structural issues and report findings.
Checks:
1. Init pollution (duplicate nested dirs)
2. Registry path mismatches (registered path doesn't exist on disk)
Args:
project_path: Path to the project root
dry_run: If True, report only (same as default — scan is always read-only)
Returns:
Dict with scan results.
"""
project_path = Path(project_path).resolve()
if not project_path.is_dir():
return {"success": False, "error": f"Project path does not exist: {project_path}"}
registry_path = None
for f in sorted(project_path.glob("*_REGISTRY.json")):
registry_path = f
break
if registry_path is None:
return {"success": False, "error": f"No *_REGISTRY.json found in {project_path}"}
results = {
"project": project_path.name,
"registry": registry_path.name,
"dry_run": dry_run,
"pollution": detect_pollution(project_path),
"registry_mismatches": [],
"actions_taken": [],
}
registry = load_registry(registry_path)
branches = branches_as_list(registry.get("branches", []))
for entry in branches:
rel_path = entry.get("path", "")
abs_path = (project_path / rel_path).resolve()
if not abs_path.exists():
results["registry_mismatches"].append(
{
"branch": entry.get("name", "?"),
"registered_path": rel_path,
"issue": "Directory missing — registered path does not exist",
}
)
total_issues = len(results["pollution"]) + len(results["registry_mismatches"])
results["total_issues"] = total_issues
results["success"] = True
return results
@@ -26,7 +26,7 @@ from aipass.spawn.apps.handlers.registry import (
find_registry,
load_registry,
save_registry,
_branches_as_list,
branches_as_list,
fix_passport_registry_id,
)
from aipass.spawn.apps.handlers.meta_ops import (
@@ -109,7 +109,7 @@ def sync_registry(fix: bool = False) -> dict:
registry_path = find_registry()
project_root = registry_path.parent
registry = load_registry(registry_path)
branches = _branches_as_list(registry.get("branches", []))
branches = branches_as_list(registry.get("branches", []))
# Build lookup of registered branch names (lowercase) -> entry
registered: dict[str, dict] = {}
@@ -127,11 +127,19 @@ def sync_registry(fix: bool = False) -> dict:
healthy: list[str] = []
# Check registered branches against filesystem
resolved_root = project_root.resolve()
for name, entry in registered.items():
rel_path = entry.get("path", "")
branch_dir = (project_root / rel_path).resolve() if rel_path else None
if branch_dir and branch_dir.is_dir():
try:
branch_dir.relative_to(resolved_root)
except ValueError:
stale.append(name)
logger.info("[sync-registry] Entry '%s' path escapes project root: %s", name, rel_path)
continue
passport = branch_dir / ".trinity" / "passport.json"
if passport.exists():
healthy.append(name)
@@ -196,7 +204,7 @@ def sync_registry(fix: bool = False) -> dict:
logger.info(f"[sync-registry] Added unregistered branch: {name}")
# Update total and save
registry["metadata"]["total_branches"] = len(_branches_as_list(registry["branches"]))
registry["metadata"]["total_branches"] = len(branches_as_list(registry["branches"]))
save_result = save_registry(registry_path, registry)
fixed = save_result
+3 -3
View File
@@ -33,7 +33,7 @@ from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state
from aipass.spawn.apps.handlers.change_detection import detect_changes
from aipass.spawn.apps.handlers.json_ops import backup_json, deep_merge
from aipass.spawn.apps.handlers.placeholders import build_replacements_dict, replace_placeholders
from aipass.spawn.apps.handlers.registry import find_registry, load_registry, _branches_as_list
from aipass.spawn.apps.handlers.registry import find_registry, load_registry, branches_as_list
from aipass.spawn.apps.handlers.json import json_handler
@@ -269,7 +269,7 @@ def update_all(dry_run: bool = False, trace: bool = False, citizen_class: str |
"""
registry_path = find_registry()
registry = load_registry(registry_path)
branches = _branches_as_list(registry.get("branches", []))
branches = branches_as_list(registry.get("branches", []))
if not branches:
return []
@@ -361,7 +361,7 @@ def _resolve_branch_path(branch_name: str) -> Path | None:
project_root = registry_path.parent
registry = load_registry(registry_path)
for branch in _branches_as_list(registry.get("branches", [])):
for branch in branches_as_list(registry.get("branches", [])):
reg_name = branch.get("name", "")
if reg_name.lower() == branch_name.lower():
rel_path = branch.get("path", "")
+10
View File
@@ -180,6 +180,16 @@ def _spawn_agent(
else:
template = _get_template_dir(citizen_class)
# Guard: block creating agent inside another agent's directory
for parent in target.parents:
if (parent / ".trinity" / "passport.json").is_file():
return _error(
f"BLOCKED: Cannot create agent inside existing agent '{parent.name}' "
f"(found .trinity/passport.json at {parent})"
)
if parent == parent.parent:
break
# Validate
if target.exists():
# If target has a passport, adopt it (register without re-creating)
+274
View File
@@ -0,0 +1,274 @@
# =================== AIPass ====================
# Name: repair.py
# Description: Project structure repair — thin CLI layer for move, cleanup, scan
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""Repair orchestrator for project structure fixes.
Thin CLI module that parses arguments and delegates to the repair handler.
All implementation logic lives in apps/handlers/repair_ops.py.
"""
from pathlib import Path
from aipass.prax import logger
from aipass.cli.apps.modules import console, header, error, warning
from aipass.spawn.apps.handlers.repair_ops import (
move_branch,
cleanup_pollution,
repair_project,
)
from aipass.spawn.apps.handlers.json import json_handler
def print_introspection():
"""Display module introspection info."""
console.print()
console.print("repair Module")
console.print("Project structure repair — move branches, clean pollution, fix registries")
console.print()
console.print("Connected Handlers:")
console.print(" handlers/")
console.print(" - repair_ops.py (move_branch, cleanup_pollution, repair_project)")
console.print()
# =============================================================================
# DRONE ROUTING
# =============================================================================
def handle_command(command: str, args: list) -> bool:
"""Handle commands routed by the entry point."""
if command != "repair":
return False
if not args:
print_introspection()
return True
if "--help" in args:
_print_help()
return True
return handle_repair(args) == 0
# =============================================================================
# PUBLIC API
# =============================================================================
def handle_repair(args: list[str]) -> int:
"""Parse args and execute repair.
Args patterns:
["<project_path>"] -> scan and report
["<project_path>", "--dry-run"] -> scan only
["--relocate", "@branch", "new/path"] -> move branch
["<project_path>", "--clean-pollution"] -> archive duplicates
Returns exit code (0=success, 1=failure).
"""
if "--help" in args or "-h" in args:
_print_help()
return 0
if not args:
_print_help()
return 1
dry_run = "--dry-run" in args
if "--relocate" in args:
return _handle_relocate(args, dry_run)
if "--clean-pollution" in args:
return _handle_clean_pollution(args, dry_run)
return _handle_scan(args, dry_run)
# =============================================================================
# SUBCOMMAND HANDLERS
# =============================================================================
def _handle_relocate(args, dry_run):
"""Handle --relocate @branch new/path [--relocate-artifacts]."""
flags = {"--dry-run", "--relocate", "--relocate-artifacts"}
positional = [a for a in args if a not in flags]
relocate_artifacts = "--relocate-artifacts" in args
if len(positional) < 2:
error("--relocate requires @branch and new_path")
warning("Usage: drone @spawn repair --relocate @branch src/pkg/branch [--relocate-artifacts] [--dry-run]")
return 1
branch_name = positional[0].lstrip("@").lower()
new_path = positional[1]
registry_path = positional[2] if len(positional) > 2 else None
try:
result = move_branch(
branch_name,
new_path,
registry_path=registry_path,
dry_run=dry_run,
relocate_artifacts=relocate_artifacts,
)
except Exception as exc:
logger.error("[repair] Unexpected error relocating %s: %s", branch_name, exc)
error(f"Error relocating {branch_name}: {exc}")
return 1
_print_move_result(result, dry_run)
if result.get("success") and not dry_run:
json_handler.log_operation("repair_relocate", data={"branch": branch_name, "new_path": new_path})
return 0 if result.get("success") else 1
def _handle_clean_pollution(args, dry_run):
"""Handle --clean-pollution <project_path>."""
flags = {"--dry-run", "--clean-pollution"}
positional = [a for a in args if a not in flags]
if not positional:
error("project path required")
return 1
project_path = Path(positional[0]).resolve()
try:
result = cleanup_pollution(project_path, dry_run=dry_run)
except Exception as exc:
logger.error("[repair] Unexpected error cleaning pollution: %s", exc)
error(f"Error cleaning pollution: {exc}")
return 1
_print_pollution_result(result, dry_run)
return 0 if result.get("success") else 1
def _handle_scan(args, dry_run):
"""Handle default scan mode — report structural issues."""
flags = {"--dry-run"}
positional = [a for a in args if a not in flags]
if not positional:
error("project path required")
return 1
project_path = Path(positional[0]).resolve()
try:
result = repair_project(project_path, dry_run=dry_run)
except Exception as exc:
logger.error("[repair] Unexpected error scanning project: %s", exc)
error(f"Error scanning project: {exc}")
return 1
_print_scan_result(result)
return 0 if result.get("success") else 1
# =============================================================================
# OUTPUT HELPERS
# =============================================================================
def _print_help():
"""Display repair command help."""
warning("Usage: drone @spawn repair <project_path> [options]")
console.print()
console.print(" [green]<project_path>[/green] Path to project root")
console.print(" [green]--dry-run[/green] Preview changes without modifying")
console.print(" [green]--relocate[/green] @branch path Move a branch to a new location")
console.print(
" [green]--relocate-artifacts[/green] Move .chroma/ into branch (with --relocate, single-branch only)"
)
console.print(" [green]--clean-pollution[/green] Archive and remove duplicate dirs")
console.print()
def _print_move_result(result, dry_run):
"""Print relocate operation results."""
mode = "[dim](dry-run)[/dim] " if dry_run else ""
console.print()
if result.get("success"):
console.print(f"[green]Relocate {mode}{result.get('branch', '?')}[/green]")
console.print(f" From: {result.get('old_path', '?')}")
console.print(f" To: {result.get('new_path', '?')}")
if result.get("archive_path"):
console.print(f" Archive: {result['archive_path']}")
if dry_run and result.get("actions"):
console.print()
console.print(" [bold cyan]Would:[/bold cyan]")
for action in result["actions"]:
console.print(f" - {action}")
else:
error(f"Relocate FAILED: {result.get('error', 'unknown')}")
console.print()
def _print_pollution_result(result, dry_run):
"""Print pollution cleanup results."""
mode = "[dim](dry-run)[/dim] " if dry_run else ""
console.print()
issues_found = result.get("issues_found", 0)
if issues_found == 0:
console.print("[green]No pollution detected[/green]")
else:
warning(f"Pollution cleanup {mode}— {issues_found} issue(s)")
if dry_run:
for issue in result.get("issues", []):
console.print(f" - {issue['description']}")
console.print(f" Path: {issue['path']}")
else:
for item in result.get("cleaned", []):
console.print(f" Cleaned: {item['path']}")
for item in result.get("errors", []):
error(f"Failed: {item['path']} — {item['error']}")
console.print()
def _print_scan_result(result):
"""Print project scan results."""
console.print()
header(f"Structure Report — {result.get('project', '?')}")
console.print()
total = result.get("total_issues", 0)
if total == 0:
console.print("[green]No structural issues found[/green]")
else:
warning(f"{total} issue(s) found")
pollution = result.get("pollution", [])
if pollution:
console.print()
console.print(f"[bold]Pollution ({len(pollution)}):[/bold]")
for p in pollution:
console.print(f" - {p['description']}")
console.print(f" Fix: drone @spawn repair {result.get('project', '?')} --clean-pollution")
mismatches = result.get("registry_mismatches", [])
if mismatches:
console.print()
console.print(f"[bold]Registry mismatches ({len(mismatches)}):[/bold]")
for m in mismatches:
console.print(f" - {m['branch']}: {m['issue']}")
console.print(f" Registered: {m['registered_path']}")
console.print()
+7
View File
@@ -44,6 +44,7 @@ def print_help():
console.print(" [green]sync-registry[/green] Repair registry against filesystem")
console.print(" [green]sync-templates[/green] Pull managed files from source")
console.print(" [green]regenerate-registry[/green] Regenerate template registry hashes")
console.print(" [green]repair[/green] <project_path> Scan and fix project structure")
console.print()
console.print("[bold cyan]CITIZEN CLASSES:[/bold cyan]")
console.print()
@@ -197,6 +198,7 @@ def print_introspection():
console.print(" - sync_templates.py (handle_sync_templates — template synchronization)")
console.print(" - regenerate_registry.py (handle_regenerate_registry — regenerate template registry)")
console.print(" - passport.py (handle_passport — grant birthright citizenship)")
console.print(" - repair.py (handle_repair — project structure repair)")
console.print()
@@ -252,6 +254,11 @@ def main():
return handle_regenerate_registry(remaining)
if command == "repair":
from aipass.spawn.apps.modules.repair import handle_repair
return handle_repair(remaining)
error(f"Unknown command: {command}", suggestion="Run 'drone @spawn --help' for available commands")
return 1
@@ -17,7 +17,7 @@
"max_key_learnings": 25,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"note": "Auto-rollover to @memory when limits exceeded. Oldest entries trimmed first."
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries."
},
"status": {
"health": "healthy",
@@ -14,7 +14,7 @@
],
"limits": {
"max_lines": 600,
"note": "Auto-rollover when max_lines exceeded"
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
},
"status": {
"health": "healthy",
@@ -2,6 +2,7 @@ __pycache__/
*.pyc
*.pyo
.env
.venv/
*.egg-info/
.coverage
htmlcov/
@@ -1,7 +1,7 @@
{
"metadata": {
"version": "1.0.0",
"last_updated": "2026-05-12",
"last_updated": "2026-05-15",
"description": "Template file tracking registry for ID-based updates"
},
"files": {
@@ -50,7 +50,7 @@
"f008": {
"path": ".gitignore",
"name": ".gitignore",
"content_hash": "841dedb922da",
"content_hash": "2dd6758a96d5",
"has_branch_placeholder": false
},
"f009": {
@@ -86,13 +86,13 @@
"f012": {
"path": ".trinity/local.json",
"name": "local.json",
"content_hash": "f4e62d2d12ba",
"content_hash": "df2ba4a97f0a",
"has_branch_placeholder": false
},
"f013": {
"path": ".trinity/observations.json",
"name": "observations.json",
"content_hash": "25b957960a59",
"content_hash": "62160dfa243c",
"has_branch_placeholder": false
},
"f014": {
@@ -221,6 +221,12 @@
"content_hash": "7b39ba7bca40",
"has_branch_placeholder": false
},
"f045": {
"path": "requirements.project.txt",
"name": "requirements.project.txt",
"content_hash": "1facc521802b",
"has_branch_placeholder": false
},
"f036": {
"path": "templates/README.md",
"name": "README.md",
@@ -17,7 +17,7 @@
"max_key_learnings": 25,
"session_summary_max_chars": 150,
"learning_value_max_chars": 200,
"note": "Auto-rollover to @memory when limits exceeded. Oldest entries trimmed first."
"note": "DO NOT trim, prune, or delete entries. Rollover to @memory handles overflow automatically. Just add new entries."
},
"status": {
"health": "healthy",
@@ -14,7 +14,7 @@
],
"limits": {
"max_lines": 600,
"note": "Auto-rollover when max_lines exceeded"
"note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
},
"status": {
"health": "healthy",
@@ -0,0 +1,3 @@
# Project-specific Python packages only.
# These are installed into the AIPass venv: pip install -r requirements.project.txt
# Framework packages (drone, prax, chromadb, rich, etc.) are already available via AIPass.
+121
View File
@@ -437,6 +437,127 @@ class TestSyncRegistryCwdAware:
daemon_entry = next(b for b in reg["branches"] if b["name"] == "DAEMON")
assert daemon_entry["path"] == "src/daemon"
def test_escaped_paths_detected_as_stale(self, tmp_path):
"""Registry entries with ../paths that escape project root should be stale."""
from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry
# Simulate external project with stale cross-project entries
project = tmp_path / "myproject"
project.mkdir()
# Create a real agent that exists OUTSIDE this project (simulates AIPass branches)
external = tmp_path / "AIPass" / "src" / "aipass" / "ai_mail"
external.mkdir(parents=True)
(external / ".trinity").mkdir()
(external / ".trinity" / "passport.json").write_text('{"name": "AI_MAIL"}')
# Create a local agent that belongs to this project
local_agent = project / "src" / "polyglot"
local_agent.mkdir(parents=True)
(local_agent / ".trinity").mkdir()
(local_agent / ".trinity" / "passport.json").write_text(
json.dumps({"name": "POLYGLOT", "identity": {"citizen_class": "builder"}})
)
reg_path = project / "MYPROJECT_REGISTRY.json"
reg_path.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-05-15", "total_branches": 2},
"branches": [
{
"name": "AI_MAIL",
"path": "../AIPass/src/aipass/ai_mail",
"profile": "library",
"description": "Mail system",
"email": "@ai_mail",
"status": "active",
"created": "2026-05-01",
"last_active": "2026-05-01",
},
{
"name": "POLYGLOT",
"path": "src/polyglot",
"profile": "library",
"description": "Local agent",
"email": "@polyglot",
"status": "active",
"created": "2026-05-01",
"last_active": "2026-05-01",
},
],
}
)
)
with patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=reg_path):
result = sync_registry(fix=False)
assert "ai_mail" in result["stale"]
assert "polyglot" in result["healthy"]
def test_escaped_paths_pruned_on_fix(self, tmp_path):
"""sync_registry --fix should remove entries with ../paths escaping project root."""
from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry
project = tmp_path / "myproject"
project.mkdir()
# External directory exists with passport (would fool old code)
external = tmp_path / "AIPass" / "src" / "aipass" / "flow"
external.mkdir(parents=True)
(external / ".trinity").mkdir()
(external / ".trinity" / "passport.json").write_text('{"name": "FLOW"}')
# Local agent
local = project / "src" / "myagent"
local.mkdir(parents=True)
(local / ".trinity").mkdir()
(local / ".trinity" / "passport.json").write_text(
json.dumps({"name": "MYAGENT", "identity": {"citizen_class": "builder"}})
)
reg_path = project / "TEST_REGISTRY.json"
reg_path.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-05-15", "total_branches": 2},
"branches": [
{
"name": "FLOW",
"path": "../AIPass/src/aipass/flow",
"profile": "library",
"description": "Flow",
"email": "@flow",
"status": "active",
"created": "2026-05-01",
"last_active": "2026-05-01",
},
{
"name": "MYAGENT",
"path": "src/myagent",
"profile": "library",
"description": "Local",
"email": "@myagent",
"status": "active",
"created": "2026-05-01",
"last_active": "2026-05-01",
},
],
}
)
)
with patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=reg_path):
result = sync_registry(fix=True)
assert result["fixed"] is True
reg = json.loads(reg_path.read_text())
names = [b["name"] for b in reg["branches"]]
assert "FLOW" not in names
assert "MYAGENT" in names
assert reg["metadata"]["total_branches"] == 1
# ---------------------------------------------------------------------------
# ADOPT EXISTING Tests
+519
View File
@@ -0,0 +1,519 @@
# =================== AIPass ====================
# Name: test_repair.py
# Description: Tests for repair handler — move, registry path update, pollution cleanup
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""Tests for repair handler — move_branch, update_registry_path, pollution cleanup."""
import json
import shutil
from pathlib import Path
from unittest.mock import patch
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_project(tmp_path, project_name="testproj", branches=None):
"""Create a minimal project with registry and optional branches."""
project = tmp_path / project_name
project.mkdir()
branch_entries = []
for b in branches or []:
name = b["name"]
rel_path = b.get("path", name)
branch_dir = project / rel_path
branch_dir.mkdir(parents=True, exist_ok=True)
trinity = branch_dir / ".trinity"
trinity.mkdir()
passport = {
"branch_info": {
"branch_name": name.lower(),
"path": rel_path,
"module": f"{project_name}.{name.lower()}",
},
"identity": {"citizen_class": "builder"},
"citizenship": {"registered": True},
}
(trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8")
branch_entries.append(
{
"name": name,
"path": rel_path,
"profile": "test",
"description": b.get("purpose", "test branch"),
"email": f"@{name.lower()}",
"status": "active",
"created": "2026-01-01",
"last_active": "2026-01-01",
}
)
registry_path = project / f"{project_name.upper()}_REGISTRY.json"
registry_data = {
"metadata": {
"version": "1.0.0",
"last_updated": "2026-01-01",
"total_branches": len(branch_entries),
},
"branches": branch_entries,
}
registry_path.write_text(json.dumps(registry_data), encoding="utf-8")
return project, registry_path
# ---------------------------------------------------------------------------
# update_registry_path
# ---------------------------------------------------------------------------
class TestUpdateRegistryPath:
"""Tests for update_registry_path — path update without entry re-creation."""
def test_updates_path_preserves_fields(self, tmp_path):
"""Path updated, creation date and name preserved."""
from aipass.spawn.apps.handlers.repair_ops import update_registry_path
_project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
result = update_registry_path(reg, "NAV", "src/compass/navigator")
assert result is True
data = json.loads(reg.read_text())
entry = data["branches"][0]
assert entry["path"] == "src/compass/navigator"
assert entry["created"] == "2026-01-01"
assert entry["name"] == "NAV"
def test_not_found_returns_false(self, tmp_path):
"""Unknown branch returns False."""
from aipass.spawn.apps.handlers.repair_ops import update_registry_path
_project, reg = _make_project(tmp_path, branches=[])
result = update_registry_path(reg, "GHOST", "somewhere")
assert result is False
def test_case_insensitive_match(self, tmp_path):
"""Lowercase name matches uppercase registry entry."""
from aipass.spawn.apps.handlers.repair_ops import update_registry_path
_project, reg = _make_project(tmp_path, branches=[{"name": "POLY", "path": "polyglot"}])
result = update_registry_path(reg, "poly", "src/aipl/polyglot")
assert result is True
data = json.loads(reg.read_text())
assert data["branches"][0]["path"] == "src/aipl/polyglot"
# ---------------------------------------------------------------------------
# move_branch
# ---------------------------------------------------------------------------
class TestMoveBranch:
"""Tests for move_branch — relocate dir + registry + passport update."""
def test_moves_dir_updates_registry_and_passport(self, tmp_path):
"""Full move: directory relocated, registry path updated, passport paths updated."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
result = move_branch("NAV", "src/compass/navigator", registry_path=reg)
assert result["success"] is True
assert result["old_path"] == "navigator"
assert result["new_path"] == "src/compass/navigator"
assert not (project / "navigator").exists()
assert (project / "src" / "compass" / "navigator").is_dir()
data = json.loads(reg.read_text())
assert data["branches"][0]["path"] == "src/compass/navigator"
passport_path = project / "src" / "compass" / "navigator" / ".trinity" / "passport.json"
passport = json.loads(passport_path.read_text())
assert passport["branch_info"]["path"] == "src/compass/navigator"
def test_creates_archive(self, tmp_path):
"""Archive created before move contains original files."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
marker = project / "navigator" / "test_file.txt"
marker.write_text("hello")
result = move_branch("NAV", "src/compass/navigator", registry_path=reg)
assert result["success"] is True
archive_dir = Path(result["archive_path"])
assert archive_dir.is_dir()
assert (archive_dir / "test_file.txt").read_text() == "hello"
def test_dry_run_no_changes(self, tmp_path):
"""Dry run reports actions but makes no filesystem or registry changes."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
result = move_branch("NAV", "src/compass/navigator", registry_path=reg, dry_run=True)
assert result["success"] is True
assert result["dry_run"] is True
assert len(result["actions"]) > 0
assert (project / "navigator").is_dir()
data = json.loads(reg.read_text())
assert data["branches"][0]["path"] == "navigator"
def test_source_missing_fails(self, tmp_path):
"""Fails when source directory does not exist on disk."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
shutil.rmtree(project / "navigator")
result = move_branch("NAV", "src/nav", registry_path=reg)
assert result["success"] is False
assert "does not exist" in result["error"]
def test_target_exists_fails(self, tmp_path):
"""Fails when target directory already exists."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
(project / "src" / "compass" / "navigator").mkdir(parents=True)
result = move_branch("NAV", "src/compass/navigator", registry_path=reg)
assert result["success"] is False
assert "already exists" in result["error"]
def test_branch_not_in_registry(self, tmp_path):
"""Fails for branch name not found in registry."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
_project, reg = _make_project(tmp_path, branches=[])
result = move_branch("GHOST", "somewhere", registry_path=reg)
assert result["success"] is False
assert "not found" in result["error"]
def test_outside_project_root_fails(self, tmp_path):
"""Fails when target path escapes project root."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
_project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
result = move_branch("NAV", "/tmp/escape_attempt", registry_path=reg)
assert result["success"] is False
assert "outside project root" in result["error"]
# ---------------------------------------------------------------------------
# detect_pollution
# ---------------------------------------------------------------------------
class TestDetectPollution:
"""Tests for detect_pollution — duplicate nested directory detection."""
def test_finds_root_duplicate(self, tmp_path):
"""Detects project_name/project_name/ at root level."""
from aipass.spawn.apps.handlers.repair_ops import detect_pollution
project = tmp_path / "compass"
project.mkdir()
(project / "compass").mkdir()
issues = detect_pollution(project)
assert len(issues) == 1
assert issues[0]["type"] == "duplicate_nested_dir"
assert issues[0]["path"] == "compass"
def test_finds_src_duplicate(self, tmp_path):
"""Detects src/pkg/pkg/ duplication."""
from aipass.spawn.apps.handlers.repair_ops import detect_pollution
project = tmp_path / "myproj"
project.mkdir()
(project / "src" / "mypkg" / "mypkg").mkdir(parents=True)
issues = detect_pollution(project)
assert len(issues) == 1
assert "src/mypkg/mypkg" in issues[0]["path"]
def test_clean_project_no_issues(self, tmp_path):
"""Clean project returns empty issues list."""
from aipass.spawn.apps.handlers.repair_ops import detect_pollution
project = tmp_path / "clean"
project.mkdir()
(project / "src" / "pkg" / "agent").mkdir(parents=True)
issues = detect_pollution(project)
assert len(issues) == 0
# ---------------------------------------------------------------------------
# cleanup_pollution
# ---------------------------------------------------------------------------
class TestCleanupPollution:
"""Tests for cleanup_pollution — archive and remove duplicate dirs."""
def test_archives_and_removes(self, tmp_path):
"""Pollution dir archived then removed from filesystem."""
from aipass.spawn.apps.handlers.repair_ops import cleanup_pollution
project = tmp_path / "compass"
project.mkdir()
dup = project / "compass"
dup.mkdir()
(dup / "junk.txt").write_text("pollution")
result = cleanup_pollution(project)
assert result["success"] is True
assert result["issues_found"] == 1
assert len(result["cleaned"]) == 1
assert not dup.exists()
assert (project / ".archive" / "pollution").is_dir()
def test_dry_run_no_changes(self, tmp_path):
"""Dry run reports issues but leaves filesystem unchanged."""
from aipass.spawn.apps.handlers.repair_ops import cleanup_pollution
project = tmp_path / "compass"
project.mkdir()
dup = project / "compass"
dup.mkdir()
result = cleanup_pollution(project, dry_run=True)
assert result["success"] is True
assert result["dry_run"] is True
assert result["issues_found"] == 1
assert dup.exists()
def test_no_pollution_returns_empty(self, tmp_path):
"""Clean project returns zero issues."""
from aipass.spawn.apps.handlers.repair_ops import cleanup_pollution
project = tmp_path / "clean"
project.mkdir()
result = cleanup_pollution(project)
assert result["success"] is True
assert result["issues_found"] == 0
# ---------------------------------------------------------------------------
# repair_project
# ---------------------------------------------------------------------------
class TestRepairProject:
"""Tests for repair_project — scan and report structural issues."""
def test_detects_pollution_and_mismatches(self, tmp_path):
"""Finds both pollution and registry mismatches in one scan."""
from aipass.spawn.apps.handlers.repair_ops import repair_project
project, _reg = _make_project(
tmp_path,
project_name="compass",
branches=[{"name": "NAV", "path": "navigator"}],
)
(project / "compass").mkdir()
shutil.rmtree(project / "navigator")
result = repair_project(project)
assert result["success"] is True
assert result["total_issues"] == 2
assert len(result["pollution"]) == 1
assert len(result["registry_mismatches"]) == 1
def test_clean_project_no_issues(self, tmp_path):
"""Clean project reports zero issues."""
from aipass.spawn.apps.handlers.repair_ops import repair_project
project, _reg = _make_project(tmp_path, branches=[{"name": "AGENT", "path": "agent"}])
result = repair_project(project)
assert result["success"] is True
assert result["total_issues"] == 0
def test_no_registry_fails(self, tmp_path):
"""Fails when no *_REGISTRY.json found."""
from aipass.spawn.apps.handlers.repair_ops import repair_project
project = tmp_path / "empty"
project.mkdir()
result = repair_project(project)
assert result["success"] is False
assert "REGISTRY" in result["error"]
def test_nonexistent_path_fails(self, tmp_path):
"""Fails when project path does not exist."""
from aipass.spawn.apps.handlers.repair_ops import repair_project
result = repair_project(tmp_path / "does_not_exist")
assert result["success"] is False
assert "does not exist" in result["error"]
# ---------------------------------------------------------------------------
# CLI routing
# ---------------------------------------------------------------------------
class TestRepairCLI:
"""Tests for repair CLI integration — command routing and arg parsing."""
def test_repair_command_routes(self):
"""Verify spawn.py routes 'repair' to repair module."""
from aipass.spawn.apps.spawn import main
with patch("sys.argv", ["spawn", "repair", "--help"]):
result = main()
assert result == 0
def test_handle_repair_help(self):
"""--help returns exit code 0."""
from aipass.spawn.apps.modules.repair import handle_repair
result = handle_repair(["--help"])
assert result == 0
def test_handle_repair_no_args(self):
"""No args returns exit code 1."""
from aipass.spawn.apps.modules.repair import handle_repair
result = handle_repair([])
assert result == 1
# ---------------------------------------------------------------------------
# .chroma relocation
# ---------------------------------------------------------------------------
class TestChromaRelocation:
"""Tests for .chroma artifact relocation during branch moves."""
def test_relocates_chroma_single_branch(self, tmp_path):
"""Moves .chroma/ into branch dir when only 1 branch in registry."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
chroma = project / ".chroma"
chroma.mkdir()
(chroma / "data.bin").write_text("vectors")
result = move_branch("NAV", "src/compass/navigator", registry_path=reg, relocate_artifacts=True)
assert result["success"] is True
assert result["chroma_relocated"] is True
assert not chroma.exists()
assert (project / "src" / "compass" / "navigator" / ".chroma" / "data.bin").read_text() == "vectors"
def test_skips_chroma_multiple_branches(self, tmp_path):
"""Does not relocate .chroma/ when more than 1 branch exists."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(
tmp_path,
branches=[
{"name": "NAV", "path": "navigator"},
{"name": "LOG", "path": "logger"},
],
)
chroma = project / ".chroma"
chroma.mkdir()
result = move_branch("NAV", "src/compass/navigator", registry_path=reg, relocate_artifacts=True)
assert result["success"] is True
assert result["chroma_relocated"] is False
assert chroma.exists()
def test_skips_when_no_chroma(self, tmp_path):
"""Does not fail when .chroma/ does not exist at project root."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
_project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
result = move_branch("NAV", "src/compass/navigator", registry_path=reg, relocate_artifacts=True)
assert result["success"] is True
assert result["chroma_relocated"] is False
def test_skips_when_chroma_already_in_branch(self, tmp_path):
"""Does not overwrite existing .chroma/ inside the branch."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
(project / ".chroma").mkdir()
(project / "navigator" / ".chroma").mkdir()
result = move_branch("NAV", "src/compass/navigator", registry_path=reg, relocate_artifacts=True)
assert result["success"] is True
assert result["chroma_relocated"] is False
def test_no_relocation_without_flag(self, tmp_path):
"""Default relocate_artifacts=False leaves .chroma/ in place."""
from aipass.spawn.apps.handlers.repair_ops import move_branch
project, reg = _make_project(tmp_path, branches=[{"name": "NAV", "path": "navigator"}])
(project / ".chroma").mkdir()
result = move_branch("NAV", "src/compass/navigator", registry_path=reg)
assert result["success"] is True
assert result.get("chroma_relocated") is False
assert (project / ".chroma").exists()
# ---------------------------------------------------------------------------
# ARCHIVE_EXCLUDE shared constant
# ---------------------------------------------------------------------------
class TestArchiveExclude:
"""Tests for ARCHIVE_EXCLUDE constant shared between repair_ops and delete_ops."""
def test_archive_exclude_defined_in_repair_ops(self):
"""ARCHIVE_EXCLUDE is a set in repair_ops."""
from aipass.spawn.apps.handlers.repair_ops import ARCHIVE_EXCLUDE
assert isinstance(ARCHIVE_EXCLUDE, set)
assert ".venv" in ARCHIVE_EXCLUDE
assert ".git" in ARCHIVE_EXCLUDE
def test_delete_ops_imports_archive_exclude(self):
"""delete_ops imports ARCHIVE_EXCLUDE from repair_ops (same object)."""
from aipass.spawn.apps.handlers.delete_ops import ARCHIVE_EXCLUDE as del_exclude
from aipass.spawn.apps.handlers.repair_ops import ARCHIVE_EXCLUDE as rep_exclude
assert del_exclude is rep_exclude
# ---------------------------------------------------------------------------
# Template file checks
# ---------------------------------------------------------------------------
class TestTemplateFiles:
"""Tests for template file additions — .gitignore and requirements.project.txt."""
def test_builder_gitignore_has_venv(self):
"""Builder template .gitignore includes .venv/ entry."""
gitignore = Path(__file__).resolve().parent.parent / "templates" / "builder" / ".gitignore"
content = gitignore.read_text()
assert ".venv/" in content
def test_requirements_project_exists(self):
"""Builder template includes requirements.project.txt."""
req = Path(__file__).resolve().parent.parent / "templates" / "builder" / "requirements.project.txt"
assert req.exists()
content = req.read_text()
assert "Project-specific" in content