From 09cd76fd4a0b068ae890ebdac3298145bf95ddeb Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Fri, 15 May 2026 11:14:40 -0700 Subject: [PATCH] feat(aipass): add doctor structure scanner (DPLAN-0177 Phase 1) --- src/aipass/aipass/.seedgo/bypass.json | 10 + .../apps/handlers/structure_scan/__init__.py | 0 .../structure_scan/structure_scanner.py | 264 ++++++++++++++ src/aipass/aipass/apps/modules/doctor.py | 210 ++++++------ src/aipass/aipass/apps/modules/doctor_wire.py | 76 +++++ src/aipass/aipass/tests/test_doctor.py | 39 ++- .../aipass/tests/test_structure_scan.py | 322 ++++++++++++++++++ 7 files changed, 792 insertions(+), 129 deletions(-) create mode 100644 src/aipass/aipass/apps/handlers/structure_scan/__init__.py create mode 100644 src/aipass/aipass/apps/handlers/structure_scan/structure_scanner.py create mode 100644 src/aipass/aipass/tests/test_structure_scan.py diff --git a/src/aipass/aipass/.seedgo/bypass.json b/src/aipass/aipass/.seedgo/bypass.json index 36300167..55b1ab9c 100644 --- a/src/aipass/aipass/.seedgo/bypass.json +++ b/src/aipass/aipass/.seedgo/bypass.json @@ -216,6 +216,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_structure_scan.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_structure_scan.py", + "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": "apps/handlers/init/bootstrap.py", "standard": "debug_print", diff --git a/src/aipass/aipass/apps/handlers/structure_scan/__init__.py b/src/aipass/aipass/apps/handlers/structure_scan/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/aipass/aipass/apps/handlers/structure_scan/structure_scanner.py b/src/aipass/aipass/apps/handlers/structure_scan/structure_scanner.py new file mode 100644 index 00000000..b2e5f68c --- /dev/null +++ b/src/aipass/aipass/apps/handlers/structure_scan/structure_scanner.py @@ -0,0 +1,264 @@ +# =================== AIPass ==================== +# Name: structure_scanner.py +# Description: Project structure validation for aipass doctor +# Version: 1.0.0 +# Created: 2026-05-14 +# Modified: 2026-05-14 +# ============================================= + +""" +Structure Scanner — detect misplaced agents, init pollution, registry mismatches. + +Pure scanning logic. Returns plain dicts/lists — no Rich markup. +Display concerns belong to the doctor module. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, NamedTuple, Optional + +from aipass.aipass.apps.handlers.json import json_handler +from aipass.prax import logger + + +class AgentInfo(NamedTuple): + """An agent discovered on disk via .trinity/passport.json.""" + + name: str + path: Path + registry_id: str + passport_data: Dict[str, Any] + + +class PlacementIssue(NamedTuple): + """Agent placement problem.""" + + agent_name: str + actual_path: str + expected_pattern: str + severity: str # "warn" or "fail" + + +class PollutionHit(NamedTuple): + """Same registry_id found at multiple locations.""" + + registry_id: str + agent_name: str + locations: List[str] + + +class RegistryIssue(NamedTuple): + """Registry path mismatch.""" + + branch_name: str + registered_path: str + problem: str # "missing" or "mismatch" + + +# ============================================================================= +# PROJECT ROOT DETECTION +# ============================================================================= + + +def find_project_root(start: Path) -> Optional[Path]: + """Walk up from *start* looking for a registry file or pyproject.toml with src/. + + Returns: + Project root Path, or None if not found. + """ + p = start.resolve() + for parent in (p, *p.parents): + if list(parent.glob("*_REGISTRY.json")): + return parent + if (parent / "pyproject.toml").exists() and (parent / "src").is_dir(): + return parent + if parent == parent.parent: + break + return None + + +# ============================================================================= +# AGENT DETECTION +# ============================================================================= + + +def scan_agents(project_root: Path) -> List[AgentInfo]: + """Find all agents by scanning for .trinity/passport.json under project_root. + + Returns: + List of AgentInfo for each valid passport found. + """ + agents: List[AgentInfo] = [] + for passport_path in sorted(project_root.rglob(".trinity/passport.json")): + try: + data = json.loads(passport_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("[structure_scan] unreadable passport: %s — %s", passport_path, exc) + continue + + branch_info = data.get("branch_info", {}) + citizenship = data.get("citizenship", {}) + name = branch_info.get("branch_name", passport_path.parent.parent.name) + registry_id = citizenship.get("registry_id", "") + agent_dir = passport_path.parent.parent + + agents.append( + AgentInfo( + name=name, + path=agent_dir, + registry_id=registry_id, + passport_data=data, + ) + ) + + logger.info("[structure_scan] found %d agents under %s", len(agents), project_root) + json_handler.log_operation("scan_agents", {"count": len(agents), "root": str(project_root)}) + return agents + + +# ============================================================================= +# PLACEMENT VALIDATION +# ============================================================================= + + +def check_placement(agents: List[AgentInfo], project_root: Path) -> List[PlacementIssue]: + """Check whether each agent is in src/// or src//. + + Returns: + List of PlacementIssue for agents in unexpected locations. + """ + src_dir = project_root / "src" + issues: List[PlacementIssue] = [] + + for agent in agents: + rel = None + try: + rel = agent.path.relative_to(src_dir) + except ValueError: + logger.warning("[structure_scan] agent %s outside src/: %s", agent.name, agent.path) + issues.append( + PlacementIssue( + agent_name=agent.name, + actual_path=str(agent.path), + expected_pattern="src/// or src//", + severity="warn", + ) + ) + continue + + parts = rel.parts + if len(parts) == 1: + # src// — valid single-agent layout + continue + elif len(parts) == 2: + # src/// — valid multi-agent package layout + continue + else: + issues.append( + PlacementIssue( + agent_name=agent.name, + actual_path=str(agent.path), + expected_pattern="src/// (too deeply nested)", + severity="warn", + ) + ) + + return issues + + +# ============================================================================= +# POLLUTION DETECTION +# ============================================================================= + + +def detect_pollution(agents: List[AgentInfo]) -> List[PollutionHit]: + """Find duplicate registry_ids — same agent found at multiple locations. + + Returns: + List of PollutionHit for each duplicated registry_id. + """ + id_to_locations: Dict[str, List[AgentInfo]] = {} + for agent in agents: + if not agent.registry_id: + continue + id_to_locations.setdefault(agent.registry_id, []).append(agent) + + hits: List[PollutionHit] = [] + for rid, group in id_to_locations.items(): + if len(group) > 1: + hits.append( + PollutionHit( + registry_id=rid, + agent_name=group[0].name, + locations=[str(a.path) for a in group], + ) + ) + return hits + + +# ============================================================================= +# REGISTRY CONSISTENCY +# ============================================================================= + + +def find_registry(project_root: Path) -> Optional[Path]: + """Find *_REGISTRY.json under project_root.""" + candidates = list(project_root.glob("*_REGISTRY.json")) + return candidates[0] if candidates else None + + +def check_registry_consistency( + registry_path: Path, + agents: List[AgentInfo], +) -> List[RegistryIssue]: + """Validate that registry branches[].path entries match actual filesystem. + + Returns: + List of RegistryIssue for each problem found. + """ + try: + data = json.loads(registry_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("[structure_scan] registry unreadable: %s", exc) + return [RegistryIssue("(registry)", str(registry_path), "unreadable")] + + branches = data.get("branches", []) + agent_paths = {str(a.path.resolve()) for a in agents} + issues: List[RegistryIssue] = [] + + for branch in branches: + name = branch.get("name", "unknown") + path_str = branch.get("path", "") + if not path_str: + issues.append(RegistryIssue(name, "", "missing")) + continue + + reg_path = Path(path_str).resolve() + if not reg_path.exists(): + issues.append(RegistryIssue(name, path_str, "missing")) + elif str(reg_path) not in agent_paths: + trinity = reg_path / ".trinity" / "passport.json" + if not trinity.exists(): + issues.append(RegistryIssue(name, path_str, "no_passport")) + + return issues + + +# ============================================================================= +# PYPROJECT CHECK +# ============================================================================= + + +def check_pyproject(project_root: Path) -> Dict[str, Any]: + """Check for pyproject.toml presence at project root. + + Returns: + Dict with 'found' bool and optional 'path' string. + """ + pyproject = project_root / "pyproject.toml" + return { + "found": pyproject.exists(), + "path": str(pyproject) if pyproject.exists() else "", + } diff --git a/src/aipass/aipass/apps/modules/doctor.py b/src/aipass/aipass/apps/modules/doctor.py index 484f2002..9aefcb7a 100644 --- a/src/aipass/aipass/apps/modules/doctor.py +++ b/src/aipass/aipass/apps/modules/doctor.py @@ -36,10 +36,18 @@ 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, + detect_pollution, + find_project_root, + find_registry, + scan_agents, +) from aipass.aipass.apps.modules.doctor_wire import ( - ENV_DESCRIPTIONS, - HOOK_DESCRIPTIONS, _auto_wire_provider, + prompt_auto_wire, ) from aipass.aipass.apps.handlers.system_detect.system_detector import ( detect_cpu, @@ -362,7 +370,7 @@ def _check_provider_manifest(interactive: bool = False, fix: bool = False) -> Li console.print(f"[green]✓[/green] {action}") wired = bool(actions) else: - wired = _prompt_auto_wire(manifest_path, results, missing_hooks, missing_env, missing_deny, missing_ask) + wired = prompt_auto_wire(manifest_path, missing_hooks, missing_env, missing_deny, missing_ask) if wired: return _check_provider_manifest(interactive=False, fix=False) @@ -370,78 +378,6 @@ def _check_provider_manifest(interactive: bool = False, fix: bool = False) -> Li return results -def _prompt_auto_wire( - manifest_path: Path, - results: List[CheckResult], - missing_hooks: List[str], - missing_env: List[str], - missing_deny: List[str], - missing_ask: List[str], -) -> bool: - """Prompt user to auto-wire provider settings, or print manual warning. - - Returns True if wiring was performed. - """ - hook_count = len(missing_hooks) - env_count = len(missing_env) - perm_count = len(missing_deny) + len(missing_ask) - logger.warning("[doctor] %d hooks, %d env vars, %d permissions missing", hook_count, env_count, perm_count) - parts = [] - if hook_count: - parts.append(f"{hook_count} hooks") - if env_count: - parts.append(f"{env_count} env vars") - if perm_count: - parts.append(f"{perm_count} permissions") - console.print(f"\n[bold]{', '.join(parts)} missing[/bold]") - console.print("[dim]Review details: .claude/hooks/README.md[/dim]") - - try: - answer = input("Auto-wire provider settings? [y/N]: ").strip().lower() - except (EOFError, KeyboardInterrupt) as exc: - logger.info("[doctor] auto-wire prompt interrupted: %s", type(exc).__name__) - answer = "n" - - if answer in ("y", "yes"): - actions = _auto_wire_provider(manifest_path, interactive=True) - for action in actions: - console.print(f"[green]✓[/green] {action}") - return bool(actions) - - _print_manual_wire_warning(missing_hooks, missing_env, missing_deny, missing_ask) - return False - - -def _print_manual_wire_warning( - missing_hooks: List[str], - missing_env: List[str], - missing_deny: List[str], - missing_ask: List[str], -) -> None: - """Print detailed warning when user declines auto-wire.""" - logger.warning("[doctor] provider settings not wired — user declined auto-wire") - console.print("\n[bold]Provider settings not wired. Required for full AIPass functionality:[/bold]\n") - if missing_hooks: - console.print("[bold]Hooks (code quality enforcement):[/bold]") - for hook in missing_hooks: - desc = HOOK_DESCRIPTIONS.get(hook, hook) - console.print(f" [dim]•[/dim] {hook} — {desc}") - console.print() - if missing_env: - console.print("[bold]Env vars:[/bold]") - for var in missing_env: - desc = ENV_DESCRIPTIONS.get(var, var) - console.print(f" [dim]•[/dim] {var} — {desc}") - console.print() - if missing_deny or missing_ask: - console.print( - f"{len(missing_deny)} deny rules + {len(missing_ask)} ask rules" - " (protect ~/.secrets/, block destructive git)" - ) - console.print() - console.print("[dim]Wire manually when ready — see .claude/hooks/README.md[/dim]") - - def _check_services(verbose: bool = False) -> List[CheckResult]: """Run Services group checks.""" results: List[CheckResult] = [] @@ -535,24 +471,94 @@ def _check_community() -> List[CheckResult]: return results +# ============================================================================= +# STRUCTURE CHECK GROUP +# ============================================================================= + + +def _check_structure() -> List[CheckResult]: + """Run Structure group checks — agent placement, pollution, registry consistency.""" + results: List[CheckResult] = [] + + project_root = find_project_root(Path.cwd()) + if project_root is None: + results.append( + CheckResult("project root", GLYPH_WARN, "not detected", "Run from inside an AIPass project directory") + ) + return results + + agents = scan_agents(project_root) + results.append(CheckResult("agents found", GLYPH_PASS, f"{len(agents)} agents", "")) + + # Placement + placement_issues = check_placement(agents, project_root) + if placement_issues: + for issue in placement_issues: + glyph = GLYPH_WARN if issue.severity == "warn" else GLYPH_FAIL + results.append( + CheckResult(f"placement: {issue.agent_name}", glyph, issue.actual_path, issue.expected_pattern) + ) + else: + results.append(CheckResult("placement", GLYPH_PASS, "all agents correctly placed", "")) + + # Pollution + pollution = detect_pollution(agents) + if pollution: + for hit in pollution: + locs = ", ".join(hit.locations) + results.append( + CheckResult( + f"pollution: {hit.agent_name}", + GLYPH_FAIL, + f"{len(hit.locations)} copies", + f"Duplicate registry_id at: {locs}", + ) + ) + else: + results.append(CheckResult("pollution", GLYPH_PASS, "no duplicates", "")) + + # Registry consistency + reg_path = find_registry(project_root) + if reg_path: + reg_issues = check_registry_consistency(reg_path, agents) + if reg_issues: + for issue in reg_issues: + glyph = GLYPH_FAIL if issue.problem == "missing" else GLYPH_WARN + results.append( + CheckResult(f"registry: {issue.branch_name}", glyph, issue.problem, issue.registered_path) + ) + else: + results.append(CheckResult("registry paths", GLYPH_PASS, "all paths valid", "")) + else: + results.append(CheckResult("registry", GLYPH_WARN, "not found", "Expected *_REGISTRY.json in project root")) + + # Pyproject + pyproject = check_pyproject(project_root) + if pyproject["found"]: + results.append(CheckResult("pyproject.toml", GLYPH_PASS, "present", "")) + else: + results.append(CheckResult("pyproject.toml", GLYPH_WARN, "missing", "Create pyproject.toml for pip packaging")) + + return results + + # ============================================================================= # MAIN DOCTOR RUN # ============================================================================= def run_doctor(verbose: bool = False, interactive: bool = False, fix: bool = False) -> int: - """Run all four groups and print results. Returns error count.""" + """Run all five groups and print results. Returns error count.""" console.print() console.print("[bold cyan]aipass doctor[/bold cyan]") console.print() - # Run each check group inside a transient progress spinner so the user - # sees what is happening during slow checks (e.g. pytest --collect-only). group_specs = [ ("System", _check_system), ("Identity", _check_identity), ("Services", lambda: _check_services(verbose=verbose)), ("Community", _check_community), + ("Structure", _check_structure), ] groups: Dict[str, List[CheckResult]] = {} with make_doctor_progress() as progress: @@ -609,29 +615,12 @@ 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]Connected Handlers:[/yellow]") - console.print(" [cyan]handlers/system_detect/[/cyan]") - console.print(" [dim]- system_detector.py (python, git, shell, OS, RAM, CPU, install)[/dim]") + 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(" [cyan]handlers/ui/[/cyan]") - console.print(" [dim]- progress.py (GLYPH_PASS/WARN/FAIL, format_check, make_doctor_progress)[/dim]") - console.print() - console.print(" [cyan]handlers/json/[/cyan]") - console.print(" [dim]- json_handler.py (operation logging)[/dim]") - console.print() - - console.print("[yellow]Check Groups:[/yellow]") - console.print(" [dim]System — Python, git, shell, OS, RAM, CPU, install method[/dim]") - console.print(" [dim]Identity — AIPASS_HOME, registry, passport[/dim]") - console.print(" [dim]Services — drone routing, pytest collect, hooks[/dim]") - console.print(" [dim]Community — ai_mail, dropbox[/dim]") - 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(" [green]aipass doctor --help[/green] [dim]# Full usage[/dim]") + 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() @@ -639,24 +628,17 @@ 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") + 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 --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[/green] [dim]# Run all checks (interactive)[/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() - - console.print("[yellow]OUTPUT:[/yellow]") - console.print(" [green]✓[/green] green — check passed") - console.print(" [yellow]![/yellow] yellow — warning (non-blocking)") - console.print(" [red]✗[/red] red — error (remediation shown below)") - console.print() - - console.print("[yellow]EXIT CODES:[/yellow]") - console.print(" 0 — all checks pass or warn only") - console.print(" 1 — one or more errors found") + 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() diff --git a/src/aipass/aipass/apps/modules/doctor_wire.py b/src/aipass/aipass/apps/modules/doctor_wire.py index 367cbab8..943dd404 100644 --- a/src/aipass/aipass/apps/modules/doctor_wire.py +++ b/src/aipass/aipass/apps/modules/doctor_wire.py @@ -223,6 +223,82 @@ def _auto_wire_provider(manifest_path: Path, interactive: bool = True) -> List[s return actions +# ============================================================================= +# INTERACTIVE WIRE PROMPTS +# ============================================================================= + + +def prompt_auto_wire( + manifest_path: Path, + missing_hooks: List[str], + missing_env: List[str], + missing_deny: List[str], + missing_ask: List[str], +) -> bool: + """Prompt user to auto-wire provider settings, or print manual warning. + + Returns True if wiring was performed. + """ + hook_count = len(missing_hooks) + env_count = len(missing_env) + perm_count = len(missing_deny) + len(missing_ask) + logger.warning("[doctor] %d hooks, %d env vars, %d permissions missing", hook_count, env_count, perm_count) + parts = [] + if hook_count: + parts.append(f"{hook_count} hooks") + if env_count: + parts.append(f"{env_count} env vars") + if perm_count: + parts.append(f"{perm_count} permissions") + console.print(f"\n[bold]{', '.join(parts)} missing[/bold]") + console.print("[dim]Review details: .claude/hooks/README.md[/dim]") + + try: + answer = input("Auto-wire provider settings? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt) as exc: + logger.info("[doctor] auto-wire prompt interrupted: %s", type(exc).__name__) + answer = "n" + + if answer in ("y", "yes"): + actions = _auto_wire_provider(manifest_path, interactive=True) + for action in actions: + console.print(f"[green]✓[/green] {action}") + return bool(actions) + + _print_manual_wire_warning(missing_hooks, missing_env, missing_deny, missing_ask) + return False + + +def _print_manual_wire_warning( + missing_hooks: List[str], + missing_env: List[str], + missing_deny: List[str], + missing_ask: List[str], +) -> None: + """Print detailed warning when user declines auto-wire.""" + logger.warning("[doctor] provider settings not wired — user declined auto-wire") + console.print("\n[bold]Provider settings not wired. Required for full AIPass functionality:[/bold]\n") + if missing_hooks: + console.print("[bold]Hooks (code quality enforcement):[/bold]") + for hook in missing_hooks: + desc = HOOK_DESCRIPTIONS.get(hook, hook) + console.print(f" [dim]•[/dim] {hook} — {desc}") + console.print() + if missing_env: + console.print("[bold]Env vars:[/bold]") + for var in missing_env: + desc = ENV_DESCRIPTIONS.get(var, var) + console.print(f" [dim]•[/dim] {var} — {desc}") + console.print() + if missing_deny or missing_ask: + console.print( + f"{len(missing_deny)} deny rules + {len(missing_ask)} ask rules" + " (protect ~/.secrets/, block destructive git)" + ) + console.print() + console.print("[dim]Wire manually when ready — see .claude/hooks/README.md[/dim]") + + # ============================================================================= # OUTPUT FORMATTING # ============================================================================= diff --git a/src/aipass/aipass/tests/test_doctor.py b/src/aipass/aipass/tests/test_doctor.py index f546cc0a..798e2a32 100644 --- a/src/aipass/aipass/tests/test_doctor.py +++ b/src/aipass/aipass/tests/test_doctor.py @@ -350,11 +350,14 @@ class TestRunDoctor: def test_run_doctor_returns_int(self) -> None: """run_doctor returns an integer error count.""" - with patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]): - result = run_doctor() + with ( + patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_structure", return_value=[]), + ): + result = run_doctor() assert isinstance(result, int) assert result == 0 @@ -363,11 +366,14 @@ class TestRunDoctor: from aipass.aipass.apps.modules.doctor import CheckResult fail_check = CheckResult("test", GLYPH_FAIL, "bad", "fix it") - with patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[fail_check]): - with patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]): - result = run_doctor() + with ( + patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[fail_check]), + patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_structure", return_value=[]), + ): + result = run_doctor() assert result == 1 def test_run_doctor_counts_only_errors_not_warnings(self) -> None: @@ -375,11 +381,14 @@ class TestRunDoctor: from aipass.aipass.apps.modules.doctor import CheckResult warn_check = CheckResult("test", GLYPH_WARN, "minor", "") - with patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[warn_check]): - with patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]): - with patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]): - result = run_doctor() + with ( + patch("aipass.aipass.apps.modules.doctor._check_system", return_value=[warn_check]), + patch("aipass.aipass.apps.modules.doctor._check_identity", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_services", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_community", return_value=[]), + patch("aipass.aipass.apps.modules.doctor._check_structure", return_value=[]), + ): + result = run_doctor() assert result == 0 diff --git a/src/aipass/aipass/tests/test_structure_scan.py b/src/aipass/aipass/tests/test_structure_scan.py new file mode 100644 index 00000000..37d3c44a --- /dev/null +++ b/src/aipass/aipass/tests/test_structure_scan.py @@ -0,0 +1,322 @@ +# =================== AIPass ==================== +# Name: test_structure_scan.py +# Description: Tests for doctor structure scanner (DPLAN-0177) +# Version: 1.0.0 +# Created: 2026-05-14 +# Modified: 2026-05-14 +# ============================================= + +"""Tests for structure scanner handler — agent detection, placement, pollution, registry.""" + +import json +from pathlib import Path +from unittest.mock import patch + +from aipass.aipass.apps.handlers.structure_scan.structure_scanner import ( + check_placement, + check_pyproject, + check_registry_consistency, + detect_pollution, + find_project_root, + find_registry, + scan_agents, +) +from aipass.aipass.apps.handlers.ui.progress import GLYPH_FAIL, GLYPH_WARN +from aipass.aipass.apps.modules.doctor import _check_structure + + +# ============================================================================= +# 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) -> Path: + """Create a registry file.""" + reg = tmp_path / "TEST_REGISTRY.json" + reg.write_text(json.dumps({"branches": branches}), encoding="utf-8") + return reg + + +# ============================================================================= +# TestFindProjectRoot +# ============================================================================= + + +class TestFindProjectRoot: + def test_finds_by_registry(self, tmp_path: Path) -> None: + """Finds project root via *_REGISTRY.json.""" + (tmp_path / "AIPASS_REGISTRY.json").write_text("{}", encoding="utf-8") + deep = tmp_path / "src" / "pkg" / "agent" + deep.mkdir(parents=True) + result = find_project_root(deep) + assert result == tmp_path + + def test_finds_by_pyproject_with_src(self, tmp_path: Path) -> None: + """Finds project root via pyproject.toml + src/ dir.""" + (tmp_path / "pyproject.toml").write_text("", encoding="utf-8") + (tmp_path / "src").mkdir() + result = find_project_root(tmp_path / "src") + assert result == tmp_path + + def test_returns_none_when_nothing(self, tmp_path: Path) -> None: + """Returns None when no markers found.""" + result = find_project_root(tmp_path) + assert result is None + + +# ============================================================================= +# TestScanAgents +# ============================================================================= + + +class TestScanAgents: + def test_finds_agents(self, tmp_path: Path) -> None: + """Discovers agents by scanning for .trinity/passport.json.""" + _make_agent(tmp_path, "alpha", "uuid-a") + _make_agent(tmp_path, "beta", "uuid-b") + agents = scan_agents(tmp_path) + assert len(agents) == 2 + names = {a.name for a in agents} + assert names == {"alpha", "beta"} + + def test_skips_corrupt_passport(self, tmp_path: Path) -> None: + """Skips passports that aren't valid JSON.""" + _make_agent(tmp_path, "good", "uuid-g") + bad_dir = tmp_path / "src" / "bad" / ".trinity" + bad_dir.mkdir(parents=True) + (bad_dir / "passport.json").write_text("not json{{{", encoding="utf-8") + agents = scan_agents(tmp_path) + assert len(agents) == 1 + assert agents[0].name == "good" + + def test_empty_project(self, tmp_path: Path) -> None: + """Returns empty list when no passports found.""" + agents = scan_agents(tmp_path) + assert agents == [] + + def test_extracts_registry_id(self, tmp_path: Path) -> None: + """Extracts registry_id from passport citizenship section.""" + _make_agent(tmp_path, "agent1", "my-uuid-123") + agents = scan_agents(tmp_path) + assert agents[0].registry_id == "my-uuid-123" + + +# ============================================================================= +# TestCheckPlacement +# ============================================================================= + + +class TestCheckPlacement: + def test_valid_single_agent(self, tmp_path: Path) -> None: + """src// is valid placement.""" + _make_agent(tmp_path, "myagent") + agents = scan_agents(tmp_path) + issues = check_placement(agents, tmp_path) + assert issues == [] + + def test_valid_package_agent(self, tmp_path: Path) -> None: + """src/// is valid placement.""" + _make_agent(tmp_path, "myagent", subdir="mypkg") + agents = scan_agents(tmp_path) + issues = check_placement(agents, tmp_path) + assert issues == [] + + def test_deeply_nested_flagged(self, tmp_path: Path) -> None: + """src//// is too deeply nested.""" + agent_dir = tmp_path / "src" / "a" / "b" / "c" + trinity = agent_dir / ".trinity" + trinity.mkdir(parents=True) + passport = {"branch_info": {"branch_name": "deep"}, "citizenship": {"registry_id": "uuid-d"}} + (trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8") + agents = scan_agents(tmp_path) + issues = check_placement(agents, tmp_path) + assert len(issues) == 1 + assert "too deeply nested" in issues[0].expected_pattern + + def test_outside_src_flagged(self, tmp_path: Path) -> None: + """Agent outside src/ directory is flagged.""" + 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") + agents = scan_agents(tmp_path) + issues = check_placement(agents, tmp_path) + assert len(issues) == 1 + assert issues[0].agent_name == "stray" + + +# ============================================================================= +# TestDetectPollution +# ============================================================================= + + +class TestDetectPollution: + def test_no_duplicates(self, tmp_path: Path) -> None: + """Unique registry_ids produce no pollution hits.""" + _make_agent(tmp_path, "a", "uuid-1") + _make_agent(tmp_path, "b", "uuid-2", subdir="pkg") + agents = scan_agents(tmp_path) + hits = detect_pollution(agents) + assert hits == [] + + def test_duplicate_detected(self, tmp_path: Path) -> None: + """Same registry_id at two locations is flagged.""" + _make_agent(tmp_path, "original", "uuid-dup") + _make_agent(tmp_path, "copy", "uuid-dup", subdir="pkg") + agents = scan_agents(tmp_path) + hits = detect_pollution(agents) + assert len(hits) == 1 + assert len(hits[0].locations) == 2 + assert hits[0].registry_id == "uuid-dup" + + def test_empty_registry_id_ignored(self, tmp_path: Path) -> None: + """Agents without registry_id are skipped.""" + _make_agent(tmp_path, "noid", "") + agents = scan_agents(tmp_path) + hits = detect_pollution(agents) + assert hits == [] + + +# ============================================================================= +# TestRegistryConsistency +# ============================================================================= + + +class TestRegistryConsistency: + def test_all_paths_valid(self, tmp_path: Path) -> None: + """All registry paths exist and have passports.""" + agent_dir = _make_agent(tmp_path, "alpha", "uuid-a") + reg = _make_registry(tmp_path, [{"name": "alpha", "path": str(agent_dir)}]) + agents = scan_agents(tmp_path) + issues = check_registry_consistency(reg, agents) + assert issues == [] + + def test_missing_path_detected(self, tmp_path: Path) -> None: + """Registry path that doesn't exist is flagged.""" + reg = _make_registry(tmp_path, [{"name": "ghost", "path": str(tmp_path / "src" / "ghost")}]) + issues = check_registry_consistency(reg, []) + assert len(issues) == 1 + assert issues[0].branch_name == "ghost" + assert issues[0].problem == "missing" + + def test_no_passport_detected(self, tmp_path: Path) -> None: + """Registry path exists but has no passport.""" + no_passport_dir = tmp_path / "src" / "bare" + no_passport_dir.mkdir(parents=True) + reg = _make_registry(tmp_path, [{"name": "bare", "path": str(no_passport_dir)}]) + issues = check_registry_consistency(reg, []) + assert len(issues) == 1 + assert issues[0].problem == "no_passport" + + def test_corrupt_registry(self, tmp_path: Path) -> None: + """Corrupt registry JSON returns unreadable issue.""" + reg = tmp_path / "BAD_REGISTRY.json" + reg.write_text("not json", encoding="utf-8") + issues = check_registry_consistency(reg, []) + assert len(issues) == 1 + assert issues[0].problem == "unreadable" + + def test_empty_path_flagged(self, tmp_path: Path) -> None: + """Empty path string in registry is flagged as missing.""" + reg = _make_registry(tmp_path, [{"name": "empty", "path": ""}]) + issues = check_registry_consistency(reg, []) + assert len(issues) == 1 + assert issues[0].problem == "missing" + + +# ============================================================================= +# TestFindRegistry +# ============================================================================= + + +class TestFindRegistry: + def test_finds_registry(self, tmp_path: Path) -> None: + """Finds *_REGISTRY.json in project root.""" + (tmp_path / "AIPASS_REGISTRY.json").write_text("{}", encoding="utf-8") + result = find_registry(tmp_path) + assert result is not None + assert result.name == "AIPASS_REGISTRY.json" + + def test_returns_none(self, tmp_path: Path) -> None: + """Returns None when no registry file.""" + result = find_registry(tmp_path) + assert result is None + + +# ============================================================================= +# TestCheckPyproject +# ============================================================================= + + +class TestCheckPyproject: + def test_pyproject_present(self, tmp_path: Path) -> None: + """Found when pyproject.toml exists.""" + (tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8") + result = check_pyproject(tmp_path) + assert result["found"] is True + assert result["path"] != "" + + def test_pyproject_missing(self, tmp_path: Path) -> None: + """Not found when pyproject.toml absent.""" + result = check_pyproject(tmp_path) + assert result["found"] is False + assert result["path"] == "" + + +# ============================================================================= +# TestCheckStructureIntegration +# ============================================================================= + + +class TestCheckStructureIntegration: + def test_no_project_root(self, tmp_path: Path, monkeypatch) -> None: + """Returns warning when no project root detected.""" + monkeypatch.chdir(tmp_path) + with patch( + "aipass.aipass.apps.modules.doctor.find_project_root", + return_value=None, + ): + results = _check_structure() + assert len(results) == 1 + assert results[0].glyph == GLYPH_WARN + assert "project root" in results[0].label + + def test_clean_project(self, tmp_path: Path) -> None: + """Clean project with agents, registry, pyproject gets all PASS.""" + _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") + with patch("aipass.aipass.apps.modules.doctor.find_project_root", return_value=tmp_path): + results = _check_structure() + glyphs = {r.glyph for r in results} + assert GLYPH_FAIL not in glyphs + assert GLYPH_WARN not in glyphs + + 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") + _make_agent(tmp_path, "copy", "uuid-dup", subdir="pkg") + _make_registry(tmp_path, []) + (tmp_path / "pyproject.toml").write_text("[project]", encoding="utf-8") + with patch("aipass.aipass.apps.modules.doctor.find_project_root", return_value=tmp_path): + results = _check_structure() + pollution_results = [r for r in results if "pollution" in r.label] + assert len(pollution_results) == 1 + assert pollution_results[0].glyph == GLYPH_FAIL