seedgo+hooks: json_handler empty-guard (#667) + fix silent hook-wiring break — new wire_verify guard fails loud on empty/orphaned/dup provider hook events (the real bug: half-wired hooks written silently); presence_gate marked provider_wired:false (dormant by design, not a break); snapshot fixture corrected (drop presence_gate, add SessionStart:cadence_reset); load_json empty-guard. Live SessionStart orphan re-wired separately. 1138 seedgo + 831 hooks green

This commit is contained in:
AIOSAI
2026-07-09 16:12:44 -07:00
parent d0a31fd862
commit 5fea5bbf44
9 changed files with 717 additions and 13 deletions
+2 -1
View File
@@ -6,7 +6,8 @@
"presence_gate": {
"enabled": true,
"handler": "aipass.hooks.apps.handlers.security.presence_gate.handle",
"matcher": ""
"matcher": "",
"provider_wired": false
},
"identity_injector": {
"enabled": true,
+30
View File
@@ -366,6 +366,11 @@
"standard": "json_structure",
"reason": "Read-only config viewer \u2014 delegates JSON loading to config/loader.py, no direct JSON file ops."
},
{
"file": "apps/modules/wire_verify.py",
"standard": "json_structure",
"reason": "Reads ~/.claude/settings.json (external provider settings) with stdlib json \u2014 not branch data storage needing json_handler."
},
{
"file": "apps/modules/cadence.py",
"standard": "modules",
@@ -1103,6 +1108,31 @@
"file": "tests/test_session_start.py",
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_wire_verify.py",
"standard": "architecture",
"reason": "Test files live in tests/, not in the 3-layer apps structure."
},
{
"file": "tests/test_wire_verify.py",
"standard": "documentation",
"reason": "Test methods use descriptive names as documentation per pytest convention."
},
{
"file": "tests/test_wire_verify.py",
"standard": "encapsulation",
"reason": "Tests import modules directly to test implementation details."
},
{
"file": "tests/test_wire_verify.py",
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_wire_verify.py",
"standard": "help_text",
"reason": "Test fixture _BRIDGE_CMD contains 'python3' as part of a mock provider command string — not user-facing help text."
}
],
"notes": {
+4 -2
View File
@@ -26,6 +26,7 @@ Every hook event flows through one engine. Platform bridges normalize the event
| `drone @hooks hooksound off` | Mute all hook sounds |
| `drone @hooks hooksound on` | Unmute all hook sounds |
| `drone @hooks cadence` | Show prompt injection cadence config and state |
| `drone @hooks verify` | Cross-check provider settings vs project hook config |
| `drone @hooks --help` | Full help reference |
| `drone @hooks --version` | Version info |
@@ -54,7 +55,8 @@ src/aipass/hooks/
│ │ ├── hooksound.py # Sound control (drone @hooks hooksound on/off)
│ │ ├── hookstatus.py # Config viewer (drone @hooks status)
│ │ ├── presence.py # Branch presence — claim/release/refresh for .ai_central/PRESENCE.central.json
│ │ └── sandbox.py # Kernel sandbox — srt/bwrap wrapper + per-role policy generator
│ │ ├── sandbox.py # Kernel sandbox — srt/bwrap wrapper + per-role policy generator
│ │ └── wire_verify.py # Wire verification — provider ↔ project hook wiring checker
│ ├── handlers/
│ │ ├── bridges/ # One per provider (thin normalization)
│ │ │ └── claude.py # Claude Code bridge
@@ -86,7 +88,7 @@ src/aipass/hooks/
│ └── diagnostics.py # JSONL logging for hook execution
├── logs/
│ └── engine.jsonl # JSONL diagnostics (every hook execution)
└── tests/ # 705 tests across 25 test files
└── tests/ # 825 tests across 27 test files
```
## How It Works
@@ -27,6 +27,7 @@ EVENT_TYPES = [
"SubagentStop",
"Stop",
"Notification",
"SessionStart",
"PreCompact",
]
@@ -0,0 +1,233 @@
# =================== AIPass ====================
# Name: wire_verify.py
# Version: 1.0.0
# Description: Wire verification — cross-checks provider settings vs project hook config
# Branch: hooks
# Layer: apps/modules
# Created: 2026-07-09
# Modified: 2026-07-09
# =============================================
"""Wire verification — catches silent hook-wiring breaks.
Cross-checks ~/.claude/settings.json (provider hooks) against
.aipass/hooks.json (project hook config). Detects:
- Empty provider hook arrays (event key exists but nothing fires)
- Enabled handlers with no provider bridge entry (handler never dispatched)
- Duplicate provider entries (handler fires multiple times)
- Orphaned provider entries (bridge entry with no project config handler)
Invoked via: drone @hooks verify
"""
import json
from pathlib import Path
from aipass.cli.apps.modules import err_console
from aipass.hooks.apps.handlers.config.loader import find_project_config
from aipass.prax.apps.modules.logger import system_logger as logger
CONSOLE = err_console
HELP_COMMANDS = [
("verify", "Cross-check provider settings vs project hook config"),
]
_BRIDGE_MARKER = "bridges/claude.py"
_META_KEYS = frozenset({"_comment", "hooks_enabled"})
def _read_provider_hooks(path=None):
"""Read hook events from provider settings. Returns {event: [entries]}."""
settings_path = Path(path) if path else Path.home() / ".claude" / "settings.json"
try:
raw = settings_path.read_text(encoding="utf-8")
data = json.loads(raw)
return data.get("hooks", {})
except (OSError, json.JSONDecodeError) as exc:
logger.info("[WIRE_VERIFY] cannot read provider settings: %s", exc)
return {}
def _extract_bridge_arg(entry):
"""Extract the bridge event arg from a provider entry's command string.
Returns e.g. 'UserPromptSubmit:tier0_kernel' or 'Stop', or None if not a bridge entry.
"""
for hook in entry.get("hooks", []):
cmd = hook.get("command", "")
if _BRIDGE_MARKER not in cmd:
continue
parts = cmd.split()
for i, part in enumerate(parts):
if part.endswith("claude.py") or _BRIDGE_MARKER in part:
if i + 1 < len(parts):
return parts[i + 1]
return None
def _build_provider_index(provider_hooks, errors):
"""Parse provider hook entries into a lookup index.
Returns {event: {"filtered": {hook_name: {matcher: count}}, "unfiltered": int, "empty": bool}}.
Appends to *errors* for empty arrays.
"""
index = {}
for event, entries in provider_hooks.items():
idx = {"filtered": {}, "unfiltered": 0, "empty": False}
if not entries:
errors.append(f"{event}: provider entry exists but hooks array is EMPTY — nothing fires")
idx["empty"] = True
for entry in entries:
arg = _extract_bridge_arg(entry)
if arg is None:
continue
if ":" in arg:
hook_name = arg.split(":", 1)[1]
matcher = entry.get("matcher", "")
if hook_name not in idx["filtered"]:
idx["filtered"][hook_name] = {}
idx["filtered"][hook_name][matcher] = idx["filtered"][hook_name].get(matcher, 0) + 1
else:
idx["unfiltered"] += 1
index[event] = idx
return index
def _check_event_wiring(event_type, hooks_group, pidx, errors, warnings, info):
"""Check one project config event against its provider index entry."""
enabled_hooks = {
name: defn for name, defn in hooks_group.items() if isinstance(defn, dict) and defn.get("enabled", False)
}
if not enabled_hooks:
return
provider_wired_hooks = {
name: defn for name, defn in enabled_hooks.items() if defn.get("provider_wired", True) is not False
}
if pidx is None:
if provider_wired_hooks:
errors.append(
f"{event_type}: {len(provider_wired_hooks)} enabled handler(s) in project config"
f" but NO provider event entry — handlers never fire"
)
return
if pidx["empty"]:
return
filtered = pidx["filtered"]
if pidx["unfiltered"] > 0:
if pidx["unfiltered"] > 1:
warnings.append(f"{event_type}: {pidx['unfiltered']} duplicate unfiltered provider entries")
info.append(f"{event_type}: unfiltered bridge, {len(enabled_hooks)} enabled hooks OK")
else:
for hook_name, hook_defn in enabled_hooks.items():
if hook_defn.get("provider_wired", True) is False:
continue
if hook_name not in filtered:
errors.append(
f"{event_type}:{hook_name}: enabled in project config"
" but no provider bridge entry — handler never fires"
)
continue
for matcher, count in filtered[hook_name].items():
if count > 1:
warnings.append(
f"{event_type}:{hook_name}: {count} duplicate provider entries (matcher={matcher or 'none'})"
)
for hook_name in filtered:
if hook_name not in hooks_group:
warnings.append(
f"{event_type}:{hook_name}: provider entry exists but no handler in project config (orphaned)"
)
def verify_wiring(provider_path=None, project_config=None):
"""Cross-check provider settings against project hook config.
Returns dict with keys: errors (list), warnings (list), info (list), ok (bool).
"""
errors = []
warnings = []
info = []
provider_hooks = _read_provider_hooks(provider_path)
if not provider_hooks:
errors.append("No provider hooks found in ~/.claude/settings.json")
return {"errors": errors, "warnings": warnings, "info": info, "ok": False}
config = project_config if project_config is not None else find_project_config()
if config is None:
errors.append("No .aipass/hooks.json found in directory tree")
return {"errors": errors, "warnings": warnings, "info": info, "ok": False}
provider_index = _build_provider_index(provider_hooks, errors)
for event_type, hooks_group in config.items():
if event_type in _META_KEYS or not isinstance(hooks_group, dict):
continue
pidx = provider_index.get(event_type)
_check_event_wiring(event_type, hooks_group, pidx, errors, warnings, info)
for event in provider_hooks:
if event not in config and not provider_index.get(event, {}).get("empty"):
info.append(f"{event}: provider-only event (no project config section)")
return {
"errors": errors,
"warnings": warnings,
"info": info,
"ok": len(errors) == 0,
}
def _render_results(results):
"""Render verification results to console."""
CONSOLE.print()
if results["ok"]:
CONSOLE.print("[bold green]✓ Wire check passed[/bold green]")
else:
CONSOLE.print("[bold red]✗ Wire check FAILED[/bold red]")
CONSOLE.print()
for error in results["errors"]:
CONSOLE.print(f" [red]ERROR[/red] {error}")
for warning in results["warnings"]:
CONSOLE.print(f" [yellow]WARN[/yellow] {warning}")
for item in results["info"]:
CONSOLE.print(f" [dim]OK[/dim] {item}")
CONSOLE.print()
CONSOLE.print(f"[bold]{len(results['errors'])} errors, {len(results['warnings'])} warnings[/bold]")
def print_introspection():
"""Print module structure for drone routing."""
CONSOLE.print("[bold cyan]wire_verify[/bold cyan] — Provider ↔ project hook wiring checker")
def handle_command(command, args) -> bool:
"""Route verify commands from drone @hooks."""
if command != "verify":
return False
if args and args[0] in ("--help", "-h", "help"):
CONSOLE.print("[bold cyan]wire_verify[/bold cyan] — Provider ↔ project hook wiring checker")
CONSOLE.print()
CONSOLE.print(" drone @hooks verify Cross-check provider settings vs project config")
CONSOLE.print()
CONSOLE.print("Reads ~/.claude/settings.json and .aipass/hooks.json,")
CONSOLE.print("verifies every enabled handler has a working provider bridge entry.")
CONSOLE.print("Exits non-zero on any ERROR finding.")
return True
results = verify_wiring()
_render_results(results)
return True
+380
View File
@@ -0,0 +1,380 @@
"""Tests for wire_verify module — provider ↔ project hook wiring checker."""
import json
from unittest.mock import patch
from aipass.hooks.apps.modules import wire_verify
_BRIDGE_CMD = "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py"
def _provider_entry(event_arg, timeout=None, matcher=None):
hook = {"type": "command", "command": f"{_BRIDGE_CMD} {event_arg}"}
if timeout:
hook["timeout"] = timeout
entry: dict = {"hooks": [hook]}
if matcher is not None:
entry["matcher"] = matcher
return entry
GOOD_PROVIDER = {
"UserPromptSubmit": [
_provider_entry("UserPromptSubmit:identity_injector"),
_provider_entry("UserPromptSubmit:branch_prompt"),
],
"Stop": [_provider_entry("Stop")],
"PreToolUse": [_provider_entry("PreToolUse")],
}
GOOD_PROJECT = {
"hooks_enabled": True,
"UserPromptSubmit": {
"identity_injector": {"enabled": True, "handler": "x.handle", "matcher": ""},
"branch_prompt": {"enabled": True, "handler": "y.handle", "matcher": ""},
},
"Stop": {
"stop_sound": {"enabled": True, "handler": "s.handle", "matcher": ""},
},
"PreToolUse": {
"tool_sound": {"enabled": True, "handler": "t.handle", "matcher": "Bash|Edit"},
},
}
class TestExtractBridgeArg:
def test_filtered_arg(self):
entry = _provider_entry("UserPromptSubmit:tier0_kernel")
assert wire_verify._extract_bridge_arg(entry) == "UserPromptSubmit:tier0_kernel"
def test_unfiltered_arg(self):
entry = _provider_entry("Stop")
assert wire_verify._extract_bridge_arg(entry) == "Stop"
def test_no_bridge_marker(self):
entry = {"hooks": [{"command": "echo hello"}]}
assert wire_verify._extract_bridge_arg(entry) is None
def test_empty_hooks(self):
assert wire_verify._extract_bridge_arg({"hooks": []}) is None
def test_no_hooks_key(self):
assert wire_verify._extract_bridge_arg({}) is None
class TestBuildProviderIndex:
def test_builds_filtered_index(self):
provider = {
"UserPromptSubmit": [
_provider_entry("UserPromptSubmit:identity_injector"),
_provider_entry("UserPromptSubmit:branch_prompt"),
],
}
errors = []
idx = wire_verify._build_provider_index(provider, errors)
assert errors == []
assert idx["UserPromptSubmit"]["filtered"]["identity_injector"] == {"": 1}
assert idx["UserPromptSubmit"]["filtered"]["branch_prompt"] == {"": 1}
assert idx["UserPromptSubmit"]["unfiltered"] == 0
def test_builds_unfiltered_index(self):
provider = {"Stop": [_provider_entry("Stop")]}
errors = []
idx = wire_verify._build_provider_index(provider, errors)
assert idx["Stop"]["unfiltered"] == 1
assert idx["Stop"]["filtered"] == {}
def test_empty_array_errors(self):
provider = {"SessionStart": []}
errors = []
idx = wire_verify._build_provider_index(provider, errors)
assert len(errors) == 1
assert "EMPTY" in errors[0]
assert idx["SessionStart"]["empty"] is True
def test_duplicate_filtered_counted(self):
provider = {
"PreCompact": [
_provider_entry("PreCompact:pre_compact"),
_provider_entry("PreCompact:pre_compact"),
],
}
errors = []
idx = wire_verify._build_provider_index(provider, errors)
assert idx["PreCompact"]["filtered"]["pre_compact"] == {"": 2}
def test_distinct_matchers_not_duplicate(self):
provider = {
"PreCompact": [
_provider_entry("PreCompact:pre_compact", matcher="manual"),
_provider_entry("PreCompact:pre_compact", matcher="auto"),
],
}
errors = []
idx = wire_verify._build_provider_index(provider, errors)
assert idx["PreCompact"]["filtered"]["pre_compact"] == {"manual": 1, "auto": 1}
class TestCheckEventWiring:
def test_unfiltered_ok(self):
pidx = {"filtered": {}, "unfiltered": 1, "empty": False}
hooks_group = {"stop_sound": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("Stop", hooks_group, pidx, errors, warnings, info)
assert errors == []
assert any("unfiltered" in i for i in info)
def test_missing_provider_event(self):
hooks_group = {"cadence_reset": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("SessionStart", hooks_group, None, errors, warnings, info)
assert len(errors) == 1
assert "NO provider event entry" in errors[0]
def test_missing_per_hook_entry(self):
pidx = {"filtered": {"identity_injector": {"": 1}}, "unfiltered": 0, "empty": False}
hooks_group = {
"identity_injector": {"enabled": True, "handler": "x"},
"presence_gate": {"enabled": True, "handler": "y"},
}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("UserPromptSubmit", hooks_group, pidx, errors, warnings, info)
assert len(errors) == 1
assert "presence_gate" in errors[0]
assert "never fires" in errors[0]
def test_duplicate_per_hook_warns(self):
pidx = {"filtered": {"pre_compact": {"": 2}}, "unfiltered": 0, "empty": False}
hooks_group = {"pre_compact": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("PreCompact", hooks_group, pidx, errors, warnings, info)
assert errors == []
assert len(warnings) == 1
assert "duplicate" in warnings[0]
def test_distinct_matchers_no_warning(self):
pidx = {"filtered": {"pre_compact": {"manual": 1, "auto": 1}}, "unfiltered": 0, "empty": False}
hooks_group = {"pre_compact": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("PreCompact", hooks_group, pidx, errors, warnings, info)
assert errors == []
assert warnings == []
def test_orphaned_provider_entry(self):
pidx = {"filtered": {"ghost_hook": {"": 1}}, "unfiltered": 0, "empty": False}
hooks_group = {"real_hook": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("UserPromptSubmit", hooks_group, pidx, errors, warnings, info)
assert any("orphaned" in w for w in warnings)
def test_disabled_hooks_skipped(self):
pidx = {"filtered": {}, "unfiltered": 0, "empty": False}
hooks_group = {"disabled_hook": {"enabled": False, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("UserPromptSubmit", hooks_group, pidx, errors, warnings, info)
assert errors == []
def test_empty_provider_skipped(self):
pidx = {"filtered": {}, "unfiltered": 0, "empty": True}
hooks_group = {"cadence_reset": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("SessionStart", hooks_group, pidx, errors, warnings, info)
assert errors == []
def test_duplicate_unfiltered_warns(self):
pidx = {"filtered": {}, "unfiltered": 3, "empty": False}
hooks_group = {"stop_sound": {"enabled": True, "handler": "x"}}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("Stop", hooks_group, pidx, errors, warnings, info)
assert len(warnings) == 1
assert "duplicate unfiltered" in warnings[0]
def test_provider_wired_false_skips_error(self):
pidx = {"filtered": {"identity_injector": {"": 1}}, "unfiltered": 0, "empty": False}
hooks_group = {
"identity_injector": {"enabled": True, "handler": "x"},
"presence_gate": {"enabled": True, "handler": "y", "provider_wired": False},
}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("UserPromptSubmit", hooks_group, pidx, errors, warnings, info)
assert errors == []
def test_provider_wired_false_no_event_no_error(self):
hooks_group = {
"presence_gate": {"enabled": True, "handler": "y", "provider_wired": False},
}
errors, warnings, info = [], [], []
wire_verify._check_event_wiring("UserPromptSubmit", hooks_group, None, errors, warnings, info)
assert errors == []
class TestVerifyWiring:
def test_all_good(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": GOOD_PROVIDER}))
result = wire_verify.verify_wiring(provider_path=settings, project_config=GOOD_PROJECT)
assert result["ok"] is True
assert result["errors"] == []
def test_empty_provider_array(self, tmp_path):
provider = {**GOOD_PROVIDER, "SessionStart": []}
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": provider}))
project = {
**GOOD_PROJECT,
"SessionStart": {
"cadence_reset": {"enabled": True, "handler": "x"},
},
}
result = wire_verify.verify_wiring(provider_path=settings, project_config=project)
assert result["ok"] is False
assert any("EMPTY" in e for e in result["errors"])
def test_missing_provider_file(self, tmp_path):
result = wire_verify.verify_wiring(
provider_path=tmp_path / "nonexistent.json",
project_config=GOOD_PROJECT,
)
assert result["ok"] is False
assert any("No provider" in e for e in result["errors"])
def test_no_project_config(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": GOOD_PROVIDER}))
with patch.object(wire_verify, "find_project_config", return_value=None):
result = wire_verify.verify_wiring(provider_path=settings)
assert result["ok"] is False
assert any("hooks.json" in e for e in result["errors"])
def test_missing_per_hook_entry_is_error(self, tmp_path):
provider = {
"UserPromptSubmit": [
_provider_entry("UserPromptSubmit:identity_injector"),
],
}
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": provider}))
project = {
"hooks_enabled": True,
"UserPromptSubmit": {
"identity_injector": {"enabled": True, "handler": "x"},
"presence_gate": {"enabled": True, "handler": "y"},
},
}
result = wire_verify.verify_wiring(provider_path=settings, project_config=project)
assert result["ok"] is False
assert any("presence_gate" in e for e in result["errors"])
def test_provider_wired_false_passes(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": GOOD_PROVIDER}))
project = {
**GOOD_PROJECT,
"UserPromptSubmit": {
**GOOD_PROJECT["UserPromptSubmit"],
"presence_gate": {"enabled": True, "handler": "z", "provider_wired": False},
},
}
result = wire_verify.verify_wiring(provider_path=settings, project_config=project)
assert result["ok"] is True
assert not any("presence_gate" in e for e in result["errors"])
def test_distinct_matchers_no_dupe_warning(self, tmp_path):
provider = {
**GOOD_PROVIDER,
"PreCompact": [
_provider_entry("PreCompact:pre_compact", matcher="manual"),
_provider_entry("PreCompact:pre_compact", matcher="auto"),
],
}
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": provider}))
project = {
**GOOD_PROJECT,
"PreCompact": {
"pre_compact": {"enabled": True, "handler": "x"},
},
}
result = wire_verify.verify_wiring(provider_path=settings, project_config=project)
assert result["ok"] is True
assert not any("duplicate" in w for w in result["warnings"])
def test_provider_only_event_info(self, tmp_path):
provider = {**GOOD_PROVIDER, "CustomEvent": [_provider_entry("CustomEvent")]}
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": provider}))
result = wire_verify.verify_wiring(provider_path=settings, project_config=GOOD_PROJECT)
assert any("provider-only" in i for i in result["info"])
def test_meta_keys_ignored(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": GOOD_PROVIDER}))
project = {**GOOD_PROJECT, "_comment": "test", "hooks_enabled": True}
result = wire_verify.verify_wiring(provider_path=settings, project_config=project)
assert result["ok"] is True
class TestReadProviderHooks:
def test_reads_file(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": {"Stop": []}}))
result = wire_verify._read_provider_hooks(settings)
assert "Stop" in result
def test_missing_file_returns_empty(self, tmp_path):
result = wire_verify._read_provider_hooks(tmp_path / "missing.json")
assert result == {}
def test_malformed_json_returns_empty(self, tmp_path):
settings = tmp_path / "settings.json"
settings.write_text("not json{{{")
result = wire_verify._read_provider_hooks(settings)
assert result == {}
class TestHandleCommand:
def test_returns_false_for_non_verify(self):
assert wire_verify.handle_command("status", []) is False
def test_routes_verify(self):
mock_result = {"ok": True, "errors": [], "warnings": [], "info": []}
with patch.object(wire_verify, "verify_wiring", return_value=mock_result):
assert wire_verify.handle_command("verify", []) is True
def test_help_flag(self):
assert wire_verify.handle_command("verify", ["--help"]) is True
def test_help_word(self):
assert wire_verify.handle_command("verify", ["help"]) is True
class TestRenderResults:
def test_renders_pass(self):
from io import StringIO
from rich.console import Console
buf = StringIO()
test_console = Console(file=buf, force_terminal=False)
with patch.object(wire_verify, "CONSOLE", test_console):
wire_verify._render_results({"ok": True, "errors": [], "warnings": [], "info": ["x"]})
output = buf.getvalue()
assert "passed" in output
def test_renders_fail(self):
from io import StringIO
from rich.console import Console
buf = StringIO()
test_console = Console(file=buf, force_terminal=False)
with patch.object(wire_verify, "CONSOLE", test_console):
wire_verify._render_results({"ok": False, "errors": ["bad"], "warnings": [], "info": []})
output = buf.getvalue()
assert "FAILED" in output
assert "bad" in output
class TestPrintIntrospection:
def test_runs_without_error(self):
wire_verify.print_introspection()
@@ -125,14 +125,27 @@ def ensure_json_exists(module_name: str, json_type: str) -> bool:
def load_json(module_name: str, json_type: str) -> Optional[Any]:
"""Load JSON file, auto-create if missing"""
"""Load JSON file, auto-create if missing.
Guards against an empty/whitespace file — e.g. a concurrent writer caught
mid-truncate in the TOCTOU window between ensure_json_exists() and this
read. Rather than raising JSONDecodeError, fall back to the type's default
template so callers always get a valid structure. A non-empty but malformed
file still raises (fail honestly — that is real corruption, not a race).
"""
if not ensure_json_exists(module_name, json_type):
return None
json_path = get_json_path(module_name, json_type)
with open(json_path, "r", encoding="utf-8") as f:
return json.load(f)
content = f.read()
if not content.strip():
logger.warning("JSON file empty, using default template: %s", json_path)
return _create_default(json_type, module_name)
return json.loads(content)
def save_json(module_name: str, json_type: str, data: Any) -> bool:
@@ -1,13 +1,5 @@
{
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py UserPromptSubmit:presence_gate"
}
]
},
{
"hooks": [
{
@@ -171,5 +163,16 @@
}
]
}
],
"SessionStart": [
{
"hooks": [
{
"type": "command",
"command": "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py SessionStart:cadence_reset",
"timeout": 30
}
]
}
]
}
@@ -620,6 +620,47 @@ def test_load_json_empty_file(tmp_path: Path) -> None:
assert isinstance(result, dict), "load_json must return dict even for empty file"
def test_load_json_empty_at_read_survives_race(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""#667: empty file at load_json's OWN read.
The single-threaded case above passes because ensure_json_exists repairs the
empty file first. The real bug is a TOCTOU race: ensure_json_exists reports
OK, then a concurrent writer truncates the file before load_json re-reads it.
Simulate by stubbing ensure_json_exists to pass without repairing.
"""
json_dir = _json_dir_as_path(tmp_path)
json_dir.mkdir(parents=True, exist_ok=True)
# whitespace-only — what a writer caught mid-truncate can leave behind
(json_dir / "raced_config.json").write_text(" \n", encoding="utf-8")
monkeypatch.setattr(json_handler, "ensure_json_exists", lambda *a, **k: True)
result = json_handler.load_json("raced", "config")
assert isinstance(result, dict), "empty-at-read must fall back to default, not crash"
def test_load_json_empty_at_read_log_returns_list(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""#667: empty-at-read for a log falls back to the [] default, not a crash."""
json_dir = _json_dir_as_path(tmp_path)
json_dir.mkdir(parents=True, exist_ok=True)
(json_dir / "raced_log.json").write_text("", encoding="utf-8")
monkeypatch.setattr(json_handler, "ensure_json_exists", lambda *a, **k: True)
result = json_handler.load_json("raced", "log")
assert result == [], "empty-at-read log must fall back to the [] default"
def test_load_json_malformed_nonempty_still_raises(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""#667: a non-empty but malformed file still raises (fail honestly).
The guard only swallows empty/whitespace (a race artifact). Real corruption
must surface, not be masked by a silent default.
"""
json_dir = _json_dir_as_path(tmp_path)
json_dir.mkdir(parents=True, exist_ok=True)
(json_dir / "corrupt_config.json").write_text("{bad json", encoding="utf-8")
monkeypatch.setattr(json_handler, "ensure_json_exists", lambda *a, **k: True)
with pytest.raises(json.JSONDecodeError):
json_handler.load_json("corrupt", "config")
def test_get_json_path_returns_pathlib_path(tmp_path: Path) -> None:
"""paths_return_path: get_json_path returns a pathlib.Path instance."""
result = json_handler.get_json_path("pathmod", "config")