feat(aipass): add doctor structure scanner (DPLAN-0177 Phase 1)
This commit is contained in:
AIOSAI
@@ -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",
|
||||
|
||||
@@ -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/<package>/<agent>/ or src/<agent>/.
|
||||
|
||||
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/<package>/<agent>/ or src/<agent>/",
|
||||
severity="warn",
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
parts = rel.parts
|
||||
if len(parts) == 1:
|
||||
# src/<agent>/ — valid single-agent layout
|
||||
continue
|
||||
elif len(parts) == 2:
|
||||
# src/<package>/<agent>/ — valid multi-agent package layout
|
||||
continue
|
||||
else:
|
||||
issues.append(
|
||||
PlacementIssue(
|
||||
agent_name=agent.name,
|
||||
actual_path=str(agent.path),
|
||||
expected_pattern="src/<package>/<agent>/ (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 "",
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
# =============================================================================
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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/<agent>/ 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/<package>/<agent>/ 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/<a>/<b>/<c>/ 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
|
||||
Reference in New Issue
Block a user