Merge pull request #543 from AIOSAI/work/system-dplan-0168-phase-5-doctor-auto-wire-re-check-after

feat(system): DPLAN-0168 Phase 5: doctor auto-wire + re-check after wiring + sub-agent PR rule
This commit is contained in:
AIPass
2026-05-08 23:22:44 -07:00
committed by GitHub
4 changed files with 374 additions and 12 deletions
+2
View File
@@ -90,6 +90,8 @@ Every branch is an expert in its domain. When you don't know something, ask the
Branches operate semi-autonomously. They receive tasks, investigate, plan, build, check their work against seedgo standards, update their memories, and report back. The system teaches itself through this cycle.
**Sub-agents build, managers PR.** Sub-agents never create PRs — they build the code and run tests. The manager (devpulse or the dispatching branch) reviews the work, then creates the PR. The review gate is the manager's responsibility.
---
## Message from the AIPass Developer
+97 -11
View File
@@ -18,9 +18,9 @@ Flutter-doctor-style health check across four groups:
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.
Pure reads — never mutates (unless --fix or interactive auto-wire accepted).
Run: aipass doctor [--verbose]
Run: aipass doctor [--verbose] [--fix]
"""
from __future__ import annotations
@@ -36,6 +36,11 @@ 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.modules.doctor_wire import (
ENV_DESCRIPTIONS,
HOOK_DESCRIPTIONS,
_auto_wire_provider,
)
from aipass.aipass.apps.handlers.system_detect.system_detector import (
detect_cpu,
detect_git,
@@ -59,7 +64,6 @@ from aipass.aipass.apps.handlers.ui.progress import (
_BRANCH_ROOT = Path(__file__).resolve().parents[2]
class CheckResult(NamedTuple):
"""Single doctor check result."""
@@ -224,7 +228,7 @@ def _find_manifest() -> Path | None:
return None
def _check_provider_manifest() -> List[CheckResult]:
def _check_provider_manifest(interactive: bool = False, fix: bool = False) -> List[CheckResult]:
"""Check provider settings against manifest. Returns hook/env/permission results."""
results: List[CheckResult] = []
@@ -275,6 +279,7 @@ def _check_provider_manifest() -> List[CheckResult]:
)
# --- Env vars in provider settings ---
missing_env: List[str] = []
manifest_env = claude_section.get("env", {})
if manifest_env:
provider_settings_path = Path.home() / ".claude" / "settings.json"
@@ -299,6 +304,8 @@ def _check_provider_manifest() -> List[CheckResult]:
)
# --- Permissions ---
missing_deny: List[str] = []
missing_ask: List[str] = []
manifest_perms = claude_section.get("permissions", {})
manifest_deny = manifest_perms.get("deny", [])
manifest_ask = manifest_perms.get("ask", [])
@@ -333,10 +340,87 @@ def _check_provider_manifest() -> List[CheckResult]:
)
)
# --- Interactive auto-wire prompt / --fix auto-accept ---
if (interactive or fix) and any(r.glyph != GLYPH_PASS for r in results):
wired = False
if fix:
actions = _auto_wire_provider(manifest_path, interactive=False)
for action in actions:
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)
if wired:
return _check_provider_manifest(interactive=False, fix=False)
return results
def _check_services(verbose: bool = False) -> List[CheckResult]:
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)
console.print(f"\n[bold]{hook_count} hooks, {env_count} env vars, {perm_count} permissions 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, interactive: bool = False, fix: bool = False) -> List[CheckResult]:
"""Run Services group checks."""
results: List[CheckResult] = []
@@ -397,7 +481,7 @@ def _check_services(verbose: bool = False) -> List[CheckResult]:
results.append(CheckResult("pytest collect", GLYPH_WARN, "timed out", ""))
# hooks + env + permissions — manifest-driven provider check
manifest_checks = _check_provider_manifest()
manifest_checks = _check_provider_manifest(interactive=interactive, fix=fix)
results.extend(manifest_checks)
return results
@@ -434,7 +518,7 @@ def _check_community() -> List[CheckResult]:
# =============================================================================
def run_doctor(verbose: bool = False) -> int:
def run_doctor(verbose: bool = False, interactive: bool = False, fix: bool = False) -> int:
"""Run all four groups and print results. Returns error count."""
console.print()
console.print("[bold cyan]aipass doctor[/bold cyan]")
@@ -445,7 +529,7 @@ def run_doctor(verbose: bool = False) -> int:
group_specs = [
("System", _check_system),
("Identity", _check_identity),
("Services", lambda: _check_services(verbose=verbose)),
("Services", lambda: _check_services(verbose=verbose, interactive=interactive, fix=fix)),
("Community", _check_community),
]
groups: Dict[str, List[CheckResult]] = {}
@@ -530,8 +614,9 @@ def print_help() -> None:
console.print()
console.print("[yellow]USAGE:[/yellow]")
console.print(" [green]aipass doctor[/green] [dim]# Run all checks[/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]")
@@ -573,8 +658,9 @@ def handle_command(command: str, args: list[str]) -> bool:
return True
verbose = "--verbose" in args or "-v" in args
error_count = run_doctor(verbose=verbose)
json_handler.log_operation("doctor_run", {"error_count": error_count})
fix_mode = "--fix" in args
error_count = run_doctor(verbose=verbose, interactive=True, fix=fix_mode)
json_handler.log_operation("doctor_run", {"error_count": error_count, "fix": fix_mode})
if error_count > 0:
raise SystemExit(1)
return True
@@ -0,0 +1,274 @@
# =================== AIPass ====================
# Name: doctor_wire.py
# Description: Auto-wire provider settings from manifest into user config
# Version: 1.0.0
# Created: 2026-05-08
# Modified: 2026-05-08
# =============================================
"""
doctor_wire — auto-wire provider settings
Extracted from doctor.py to keep module sizes manageable.
Provides:
- HOOK_DESCRIPTIONS / ENV_DESCRIPTIONS — human-readable hook/env purpose
- _resolve_hook_source() — locate hook scripts (repo or pip package)
- _auto_wire_provider() — additive merge of manifest into ~/.claude/settings.json
"""
from __future__ import annotations
import json
import os
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List
from aipass.cli.apps.modules import console
from aipass.prax import logger
from aipass.aipass.apps.handlers.json import json_handler
# =============================================================================
# HOOK & ENV DESCRIPTIONS (for interactive "no" warning)
# =============================================================================
HOOK_DESCRIPTIONS: Dict[str, str] = {
"pre_edit_gate.py": "blocks edits outside agent's branch",
"subagent_stop_gate.py": "validates agent output on exit",
"auto_fix_diagnostics.py": "auto-fixes lint issues after edits",
"global_prompt_loader.py": "injects branch context on each turn",
"identity_injector.py": "injects agent identity on each turn",
"email_notification.py": "notifies on incoming agent mail",
"branch_prompt_loader.py": "loads branch-specific prompts",
"pre_compact.py": "saves state before context compaction",
}
ENV_DESCRIPTIONS: Dict[str, str] = {
"AIPASS_HOME": "tells agents where AIPass lives",
"CLAUDE_CODE_DISABLE_AUTO_MEMORY": "prevents conflict with .trinity/ memory system",
}
# =============================================================================
# HOOK SOURCE RESOLUTION
# =============================================================================
def _resolve_hook_source(manifest_path: Path) -> Path | None:
"""Find where hook scripts live.
First: look for ``.claude/hooks/`` relative to the manifest path (clone/fork users).
Fallback: find the pip-installed package location via ``aipass/_hooks/``.
Returns the directory Path, or None if neither found.
"""
# Repo-local hooks (manifest lives in .claude/, hooks are in .claude/hooks/)
repo_hooks = manifest_path.parent / "hooks"
if repo_hooks.is_dir():
return repo_hooks
# Pip-installed package — _hooks directory next to top-level aipass __init__.py
try:
import importlib.resources as _resources
ref = _resources.files("aipass").joinpath("_hooks")
pkg_hooks = Path(str(ref))
if pkg_hooks.is_dir():
return pkg_hooks
except Exception as exc:
logger.info("[doctor] importlib.resources lookup failed: %s", exc)
# Manual fallback: walk up from this file to find the aipass package root
pkg_root = Path(__file__).resolve()
for _ in range(10):
pkg_root = pkg_root.parent
candidate = pkg_root / "_hooks"
if candidate.is_dir():
return candidate
if pkg_root == pkg_root.parent:
break
return None
# =============================================================================
# AUTO-WIRE
# =============================================================================
def _auto_wire_provider(manifest_path: Path, interactive: bool = True) -> List[str]:
"""Auto-wire provider settings from manifest into ~/.claude/settings.json.
Additive merge only — never removes or overwrites existing keys/values.
Returns list of action descriptions (for logging/display).
"""
actions: List[str] = []
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
claude_section = manifest.get("cli", {}).get("claude", {})
if not claude_section:
return actions
# Read existing settings
settings_path = Path.home() / ".claude" / "settings.json"
if settings_path.exists():
settings = json.loads(settings_path.read_text(encoding="utf-8"))
else:
settings = {}
# Backup
if settings_path.exists():
date_stamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")
backup_path = settings_path.with_suffix(f".json.bak.{date_stamp}")
shutil.copy2(settings_path, backup_path)
actions.append(f"Backed up settings to {backup_path.name}")
# Hooks — copy user-source scripts and add settings entries
manifest_hooks = claude_section.get("hooks", [])
hook_source_dir = _resolve_hook_source(manifest_path)
user_hooks_dir = Path.home() / ".claude" / "hooks"
for hook in manifest_hooks:
script = hook.get("script", "")
if not script:
continue
source_type = hook.get("source", "repo")
# Only user-source hooks get copied to ~/.claude/hooks/
if source_type == "user":
target = user_hooks_dir / script
if not target.exists() and hook_source_dir:
src_file = hook_source_dir / script
if src_file.exists():
os.makedirs(user_hooks_dir, exist_ok=True)
shutil.copy2(src_file, target)
actions.append(f"Copied hook {script} to ~/.claude/hooks/")
# Add hook entry to settings.json if not already present
if "hooks" not in settings:
settings["hooks"] = {}
event = hook.get("event", "")
if event not in settings["hooks"]:
settings["hooks"][event] = []
event_hooks = settings["hooks"][event]
if not isinstance(event_hooks, list):
event_hooks = [event_hooks]
settings["hooks"][event] = event_hooks
already_wired = any(
isinstance(h, dict) and script in h.get("command", "")
for h in event_hooks
)
if not already_wired:
if source_type == "user":
hook_path = f"~/.claude/hooks/{script}"
else:
hook_path = f".claude/hooks/{script}"
entry: Dict[str, object] = {
"type": "command",
"command": f"{sys.executable} {hook_path}",
}
if hook.get("matcher"):
entry["matcher"] = hook["matcher"]
if hook.get("timeout"):
entry["timeout"] = hook["timeout"]
event_hooks.append(entry)
actions.append(f"Wired hook {script} -> {event}")
# Env vars
manifest_env = claude_section.get("env", {})
if manifest_env:
if "env" not in settings:
settings["env"] = {}
repo_root = str(manifest_path.parent.parent)
project_root = str(Path.cwd())
for key, value in manifest_env.items():
if key not in settings["env"]:
resolved = value.replace("{{REPO_ROOT}}", repo_root)
resolved = resolved.replace("{{PROJECT_ROOT}}", project_root)
settings["env"][key] = resolved
actions.append(f"Set env {key}={resolved}")
# Permissions
manifest_perms = claude_section.get("permissions", {})
manifest_deny = manifest_perms.get("deny", [])
manifest_ask = manifest_perms.get("ask", [])
if manifest_deny or manifest_ask:
if "permissions" not in settings:
settings["permissions"] = {}
if "deny" not in settings["permissions"]:
settings["permissions"]["deny"] = []
if "ask" not in settings["permissions"]:
settings["permissions"]["ask"] = []
existing_deny = set(settings["permissions"]["deny"])
for rule in manifest_deny:
if rule not in existing_deny:
settings["permissions"]["deny"].append(rule)
actions.append(f"Added deny rule: {rule}")
existing_ask = set(settings["permissions"]["ask"])
for rule in manifest_ask:
if rule not in existing_ask:
settings["permissions"]["ask"].append(rule)
actions.append(f"Added ask rule: {rule}")
# Write settings back
os.makedirs(settings_path.parent, exist_ok=True)
settings_path.write_text(json.dumps(settings, indent=2) + "\n", encoding="utf-8")
actions.append("Updated ~/.claude/settings.json")
return actions
# =============================================================================
# OUTPUT FORMATTING
# =============================================================================
def print_introspection() -> None:
"""Display module info for doctor_wire."""
console.print()
console.print("[bold cyan]doctor_wire Module[/bold cyan]")
console.print("Auto-wire provider settings from manifest into user config")
console.print()
console.print("[yellow]Provides:[/yellow]")
console.print(" [dim]- HOOK_DESCRIPTIONS / ENV_DESCRIPTIONS[/dim]")
console.print(" [dim]- _resolve_hook_source() — locate hook scripts[/dim]")
console.print(" [dim]- _auto_wire_provider() — additive merge into settings[/dim]")
console.print()
# =============================================================================
# COMMAND HANDLER
# =============================================================================
def handle_command(command: str, args: list[str]) -> bool:
"""Handle command routing. This is a helper module — no standalone commands.
Args:
command: Command name.
args: Additional arguments.
Returns:
True if handled, False otherwise.
"""
if command != "doctor_wire":
return False
if not args:
print_introspection()
json_handler.log_operation("doctor_wire_info", {"command": command})
return True
if args[0] in ("--info", "info"):
print_introspection()
json_handler.log_operation("doctor_wire_info", {"command": command})
return True
json_handler.log_operation("doctor_wire_noop", {"command": command})
return False
+1 -1
View File
@@ -332,7 +332,7 @@ class TestDoctorHandleCommand:
with patch("aipass.aipass.apps.modules.doctor.run_doctor", return_value=0) as mock_run:
with patch("aipass.aipass.apps.modules.doctor.json_handler"):
handle_command("doctor", ["--verbose"])
mock_run.assert_called_once_with(verbose=True)
mock_run.assert_called_once_with(verbose=True, interactive=True, fix=False)
# =============================================================================