setup.sh + aipass doctor: recurrence-prevention for silent hook-wiring break — setup.sh merge now drops orphaned empty hook events (the exact bug: an event left as [] fires nothing, written silently) and announces the drop; aipass doctor surfaces a wire_verify check under Services + re-verifies after --fix. Proven: orphan dropped / valid kept / user hooks preserved; doctor shows green. 629 aipass green

This commit is contained in:
AIOSAI
2026-07-09 16:25:35 -07:00
parent 5fea5bbf44
commit cdbd1dc821
4 changed files with 120 additions and 2 deletions
+9 -1
View File
@@ -712,7 +712,15 @@ for event in set(existing_hooks) | set(aipass_hooks):
entry for entry in existing_hooks.get(event, [])
if "bridges/claude.py" not in json.dumps(entry)
]
merged_hooks[event] = aipass_hooks.get(event, []) + user_entries
merged = aipass_hooks.get(event, []) + user_entries
# Never emit an empty hook event. If this event had only stale AIPass bridge
# entries (no current aipass_hooks definition AND no user-wired hooks), the
# filter above orphans it to [] — a half-wired event that fires nothing,
# written silently. Drop the key instead and say so, so the state stays honest.
if not merged:
print(f" ! dropped orphaned hook event (no live entries): {event}")
continue
merged_hooks[event] = merged
settings["hooks"] = merged_hooks
# Inject AIPASS_HOME into env block so dispatched agents find AIPass
+8
View File
@@ -60,6 +60,7 @@ from aipass.aipass.apps.modules.doctor_fix import (
)
from aipass.aipass.apps.modules.doctor_wire import (
_auto_wire_provider,
check_wire_verify,
prompt_auto_wire,
reconcile_stale_deny,
)
@@ -468,6 +469,9 @@ def _check_services(verbose: bool = False) -> List[CheckResult]:
manifest_checks = _check_provider_manifest()
results.extend(manifest_checks)
# wire_verify guard — catch empty/orphaned/duplicate provider hook entries
results.extend(CheckResult(*r) for r in check_wire_verify())
# stale rm deny rules — detect only (fix runs in run_doctor when --fix)
for tup in reconcile_stale_deny(fix=False):
results.append(CheckResult(*tup))
@@ -901,6 +905,10 @@ def run_doctor(verbose: bool = False, interactive: bool = False, fix: bool = Fal
services = groups.get("Services", [])
groups["Services"] = [r for r in services if r.label != "rm deny migration"] + stale_results
wire_recheck = [CheckResult(*r) for r in check_wire_verify()]
services = groups.get("Services", [])
groups["Services"] = [r for r in services if r.label != "wire verify"] + wire_recheck
pass_count = 0
warn_count = 0
error_count = 0
+50 -1
View File
@@ -20,9 +20,10 @@ from __future__ import annotations
import json
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List
from typing import Dict, List, NamedTuple
from aipass.cli.apps.modules import console
from aipass.prax import logger
@@ -304,3 +305,51 @@ def handle_command(command: str, args: list[str]) -> bool:
json_handler.log_operation("doctor_wire_noop", {"command": command})
return False
# =============================================================================
# WIRE VERIFY GUARD (doctor check row)
# =============================================================================
class WireCheckResult(NamedTuple):
"""Single doctor check result (mirrors doctor.CheckResult without importing it)."""
label: str
glyph: str
detail: str
remediation: str
_GLYPH_PASS = "[green]✓[/green]"
_GLYPH_FAIL = "[red]✗[/red]"
_GLYPH_WARN = "[yellow]![/yellow]"
def check_wire_verify() -> list[WireCheckResult]:
"""Run the hooks wire_verify guard — catch empty/orphaned/duplicate provider entries."""
try:
proc = subprocess.run(
["drone", "@hooks", "verify"],
capture_output=True,
text=True,
timeout=10,
)
if proc.returncode == 0:
return [WireCheckResult("wire verify", _GLYPH_PASS, "provider hooks wired correctly", "")]
lines = [ln.strip() for ln in proc.stdout.splitlines() if ln.strip()]
detail = lines[-1] if lines else "errors detected"
return [
WireCheckResult(
"wire verify",
_GLYPH_FAIL,
detail,
"Run 'aipass doctor --fix' to re-wire, then re-run doctor to confirm",
)
]
except FileNotFoundError as exc:
logger.warning("[doctor] drone not found for wire_verify: %s", exc)
return [WireCheckResult("wire verify", _GLYPH_WARN, "drone not found", "")]
except subprocess.TimeoutExpired as exc:
logger.warning("[doctor] wire_verify timed out: %s", exc)
return [WireCheckResult("wire verify", _GLYPH_WARN, "timed out", "")]
+53
View File
@@ -774,3 +774,56 @@ class TestReconcileStaleDeny:
results = reconcile_stale_deny(fix=False)
assert len(results) == 1
assert results[0][1] == GLYPH_PASS
class TestCheckWireVerify:
"""Tests for check_wire_verify() — hooks wire_verify guard."""
def test_pass_on_zero_exit(self) -> None:
"""Exit 0 from drone @hooks verify produces a PASS row."""
from aipass.aipass.apps.modules.doctor_wire import check_wire_verify
fake = MagicMock(returncode=0, stdout="✓ Wire check passed\n\n0 errors, 0 warnings\n")
with patch("aipass.aipass.apps.modules.doctor_wire.subprocess.run", return_value=fake):
results = check_wire_verify()
assert len(results) == 1
assert results[0].label == "wire verify"
assert results[0].glyph == "[green]✓[/green]"
def test_fail_on_nonzero_exit(self) -> None:
"""Non-zero exit from drone @hooks verify produces a FAIL row."""
from aipass.aipass.apps.modules.doctor_wire import check_wire_verify
fake = MagicMock(returncode=1, stdout="ERROR empty array\n2 errors, 0 warnings\n")
with patch("aipass.aipass.apps.modules.doctor_wire.subprocess.run", return_value=fake):
results = check_wire_verify()
assert len(results) == 1
assert results[0].glyph == "[red]✗[/red]"
assert "errors" in results[0].detail
def test_warn_on_drone_not_found(self) -> None:
"""FileNotFoundError (drone missing) produces a WARN row."""
from aipass.aipass.apps.modules.doctor_wire import check_wire_verify
with patch(
"aipass.aipass.apps.modules.doctor_wire.subprocess.run",
side_effect=FileNotFoundError("drone"),
):
results = check_wire_verify()
assert len(results) == 1
assert results[0].glyph == "[yellow]![/yellow]"
def test_warn_on_timeout(self) -> None:
"""TimeoutExpired produces a WARN row."""
import subprocess as sp
from aipass.aipass.apps.modules.doctor_wire import check_wire_verify
with patch(
"aipass.aipass.apps.modules.doctor_wire.subprocess.run",
side_effect=sp.TimeoutExpired(cmd="drone", timeout=10),
):
results = check_wire_verify()
assert len(results) == 1
assert results[0].glyph == "[yellow]![/yellow]"
assert "timed out" in results[0].detail