merge main into #370 — take Track E v1.3.0 (compatible with #370's auto_fix lint additions)

This commit is contained in:
AIOSAI
2026-04-21 08:26:59 -07:00
24 changed files with 3024 additions and 63 deletions
+58 -36
View File
@@ -1,34 +1,34 @@
#!/usr/bin/env python3
"""
PreToolUse Gate — Blocks edits when unresolved type errors exist.
PreToolUse Gate — Blocks unsafe edits at the hook layer.
Two-hook system:
PostToolUse (auto_fix_diagnostics.py) → detects errors, saves to state file
PreToolUse (this file) → reads state file, blocks edits to OTHER files
Rules (checked in order):
1. Inbox lock — any write targeting *.ai_mail.local/inbox.json is BLOCKED.
Use `drone @ai_mail email` instead.
2. Cross-branch — writes to src/aipass/X/** from a CWD inside src/aipass/Y/**
are BLOCKED unless the calling branch is in TRUSTED_CROSS_WRITERS.
3. State-file — edits to OTHER .py files while the current branch has unresolved
type errors are BLOCKED. (original v1.2.0 logic)
Logic:
- No state file or empty → ALLOW
- Editing the SAME file that has errors → ALLOW (they're fixing it)
- Errored file in a DIFFERENT branch → ALLOW (not your problem)
- Editing a DIFFERENT file in SAME branch → BLOCK (fix errors first)
Version: 1.2.0
Track E additions: rules 1 + 2 (DPLAN-0139).
Version: 1.3.0
"""
import json
import os
import sys
from pathlib import Path
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
# Single source of truth lives in permissions.py — inline here as fallback
# so the hook works even when aipass package is not on sys.path.
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
def _get_branch(file_path: str) -> str:
"""Extract AIPass branch name from file path.
Looks for src/aipass/{branch}/ pattern. Returns branch name
or empty string if not in a branch.
"""
"""Extract AIPass branch name from a file path (src/aipass/{branch}/ pattern)."""
parts = Path(file_path).parts
for i, part in enumerate(parts):
if part == "aipass" and i > 0 and parts[i - 1] == "src" and i + 1 < len(parts):
@@ -36,6 +36,11 @@ def _get_branch(file_path: str) -> str:
return ""
def _block(reason: str) -> None:
print(json.dumps({"decision": "block", "reason": reason}))
sys.exit(2)
def main():
try:
input_data = json.load(sys.stdin)
@@ -43,60 +48,77 @@ def main():
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
# Only gate edit tools
if tool_name not in EDIT_TOOLS:
return
# Only gate Python files
if not file_path:
return
# ------------------------------------------------------------------
# Rule 1: Inbox lock — block all writes to *.ai_mail.local/inbox.json
# ------------------------------------------------------------------
fp = Path(file_path)
if fp.name == "inbox.json" and ".ai_mail.local" in fp.parts:
_block(
"Direct writes to inbox.json are blocked.\n"
"Use: drone @ai_mail email @<branch> \"Subject\" \"Body\""
)
# ------------------------------------------------------------------
# Rule 2: Cross-branch write enforcement
# ------------------------------------------------------------------
cwd = input_data.get("cwd", "") or os.getcwd()
cwd_branch = _get_branch(cwd)
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if cwd_branch and target_branch and cwd_branch != target_branch:
if cwd_branch not in TRUSTED_CROSS_WRITERS:
_block(
f"Cross-branch write blocked: '{cwd_branch}' cannot write to '{target_branch}'.\n"
f"Trusted cross-writers: {', '.join(TRUSTED_CROSS_WRITERS)}"
)
# ------------------------------------------------------------------
# Rule 3: State-file (original v1.2.0) — .py files only
# ------------------------------------------------------------------
if not file_path.endswith(".py"):
return
# No state file → no pending errors → allow
if not STATE_FILE.exists():
return
try:
state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, IOError):
return # Corrupted state → allow
return
errored_file = state.get("file", "")
errors = state.get("errors", [])
# No errors in state → allow
if not errors:
return
# Resolve both paths for comparison
try:
current = str(Path(file_path).resolve())
errored = str(Path(errored_file).resolve())
except (OSError, ValueError):
return # Path resolution failed → allow
return
# Editing the file WITH errors → allow (they're fixing it)
if current == errored:
return
# Different branch → allow (cross-branch errors aren't your problem)
# If errored file is outside AIPass entirely → allow (external projects)
current_branch = _get_branch(current)
errored_branch = _get_branch(errored)
if not errored_branch:
return # Errored file is outside src/aipass/ — don't gate
return
if current_branch and errored_branch and current_branch != errored_branch:
return
# Editing a DIFFERENT file in SAME branch while errors exist → BLOCK
error_summary = "\n".join(f" L{e['line']}: {e['message']}" for e in errors[:5])
reason = f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n{error_summary}"
output = {
"decision": "block",
"reason": reason
}
print(json.dumps(output))
sys.exit(2)
_block(
f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n"
f"{error_summary}"
)
except Exception:
pass # Silent fail → allow
@@ -0,0 +1,21 @@
# Hook Probe Matrix — Q12 Findings 2026-04-20
Generated by `drone @seedgo hooks probe --matrix`.
Total entries: 1
## Event Matrix
| Event | Count | ProjectDir T/F | AIPassHome T/F | Unique Agents |
|-------|-------|----------------|----------------|---------------|
| PostToolUse | 1 | 0/1 | 1/0 | 1 |
## Notes
- `ProjectDir T/F`: entries where `CLAUDE_PROJECT_DIR` env var was set (T) vs unset (F)
- `AIPassHome T/F`: entries where `AIPASS_HOME` env var was set (T) vs unset (F)
- Unique Agents: distinct `CLAUDE_CODE_SESSION_ID` values seen for this event
## Raw entry count by event
- PostToolUse: 1
+105
View File
@@ -0,0 +1,105 @@
# Hook Probe Suite
This directory contains ping-response probe scripts for each Claude Code hook event type.
Probes are **opt-in** — they are never auto-wired. See below for how to enable them.
---
## What this directory is
Each `probe_*.py` script in this directory is a passive observer for one Claude Code hook event.
When enabled in `settings.json`, a probe fires on its event, records a structured entry to
`last_ping.jsonl`, and exits 0 immediately — it never blocks execution.
The log is used by `drone @seedgo hooks probe` to display event tables and generate reports.
---
## Probe scripts
| Script | Hook event |
|--------------------------------|-------------------|
| `probe_pre_tool_use.py` | PreToolUse |
| `probe_post_tool_use.py` | PostToolUse |
| `probe_user_prompt_submit.py` | UserPromptSubmit |
| `probe_subagent_stop.py` | SubagentStop |
| `probe_pre_compact.py` | PreCompact |
| `probe_stop.py` | Stop |
| `probe_notification.py` | Notification |
---
## How to enable probes (settings.json snippets)
Add any subset of the following to your `.claude/settings.json` `hooks` object.
**Replace `/path/to/AIPass` with your actual repo root.**
```json
{
"hooks": {
"PreToolUse": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_pre_tool_use.py"}]}
],
"PostToolUse": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_post_tool_use.py"}]}
],
"UserPromptSubmit": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_user_prompt_submit.py"}]}
],
"SubagentStop": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_subagent_stop.py"}]}
],
"PreCompact": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_pre_compact.py"}]}
],
"Stop": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_stop.py"}]}
],
"Notification": [
{"hooks": [{"type": "command", "command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_notification.py"}]}
]
}
}
```
---
## Example `last_ping.jsonl` output
```jsonl
{"event": "PreToolUse", "tool": "Bash", "cwd": "/home/user/Projects/AIPass", "agent_id": "sess-abc123", "timestamp": "2026-04-20T12:00:00.123456Z", "script_elapsed_ms": 2.1, "cli_version": "1.0.0", "env_has_claude_project_dir": true, "env_has_aipass_home": false}
{"event": "PostToolUse", "tool": "Read", "cwd": "/home/user/Projects/AIPass", "agent_id": "sess-abc123", "timestamp": "2026-04-20T12:00:01.456789Z", "script_elapsed_ms": 1.8, "cli_version": "1.0.0", "env_has_claude_project_dir": true, "env_has_aipass_home": false}
{"event": "Stop", "tool": "", "cwd": "/home/user/Projects/AIPass", "agent_id": "sess-abc123", "timestamp": "2026-04-20T12:05:00.000000Z", "script_elapsed_ms": 1.5, "cli_version": "1.0.0", "env_has_claude_project_dir": true, "env_has_aipass_home": false}
```
---
## How to run `drone @seedgo hooks probe`
```bash
# Display a table of recent probe entries
drone @seedgo hooks probe
# Test whether PostToolUse and SubagentStop fire in headless mode
drone @seedgo hooks probe --subagent
# Generate a full matrix report grouped by event type
drone @seedgo hooks probe --matrix
```
### Flag reference
| Flag | What it does |
|---------------|--------------|
| *(no flag)* | Read `last_ping.jsonl`, display Rich table of recent entries |
| `--subagent` | Spawn a headless Claude Code process, then check if PostToolUse / SubagentStop fired |
| `--matrix` | Group all entries by event, show counts and env-var truth table, write markdown report |
---
## Notes
- `last_ping.jsonl` is gitignored — it is a live log file, not source.
- Probes are opt-in. The AIPass repo does **not** auto-wire them into `settings.json`.
- Each probe script contains its own `settings.json` snippet in its module docstring.
- Probes are pure stdlib Python — no aipass imports, no third-party packages.
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: Notification
Fires on the Claude Code [Notification] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"Notification": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_notification.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "Notification"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: PostToolUse
Fires on the Claude Code [PostToolUse] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"PostToolUse": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_post_tool_use.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "PostToolUse"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
+94
View File
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: PreCompact
Fires on the Claude Code [PreCompact] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"PreCompact": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_pre_compact.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "PreCompact"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: PreToolUse
Fires on the Claude Code [PreToolUse] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"PreToolUse": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_pre_tool_use.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "PreToolUse"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
+94
View File
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: Stop
Fires on the Claude Code [Stop] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"Stop": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_stop.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "Stop"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: SubagentStop
Fires on the Claude Code [SubagentStop] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"SubagentStop": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_subagent_stop.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "SubagentStop"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""
Hook probe: UserPromptSubmit
Fires on the Claude Code [UserPromptSubmit] hook event.
Records a structured entry to last_ping.jsonl. Never blocks. Silent-fail on any exception.
To enable — add this snippet to AIPass/.claude/settings.json (inside "hooks"):
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "python3 /path/to/AIPass/.claude/hooks/probes/probe_user_prompt_submit.py"
}
]
}
]
Replace /path/to/AIPass with the actual AIPass repo root path.
"""
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Log file lives next to this script
_LOG_FILE = Path(__file__).parent / "last_ping.jsonl"
_EVENT = "UserPromptSubmit"
def main() -> None:
start = time.monotonic()
# --- Read stdin (tolerant of parse failures) ---
payload: dict = {}
try:
raw = sys.stdin.read()
if raw.strip():
payload = json.loads(raw)
except Exception:
pass
# --- Extract fields ---
tool = (
payload.get("tool_name")
or payload.get("hook_event_name")
or ""
)
cwd = payload.get("cwd") or os.getcwd()
agent_id = (
os.environ.get("CLAUDE_CODE_SESSION_ID")
or os.environ.get("CLAUDE_SESSION_ID")
or "unknown"
)
cli_version = os.environ.get("CLAUDE_CODE_VERSION", "unknown")
env_has_claude_project_dir = bool(os.environ.get("CLAUDE_PROJECT_DIR"))
env_has_aipass_home = bool(os.environ.get("AIPASS_HOME"))
elapsed_ms = (time.monotonic() - start) * 1000.0
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
entry = {
"event": _EVENT,
"tool": tool,
"cwd": cwd,
"agent_id": agent_id,
"timestamp": timestamp,
"script_elapsed_ms": round(elapsed_ms, 3),
"cli_version": cli_version,
"env_has_claude_project_dir": env_has_claude_project_dir,
"env_has_aipass_home": env_has_aipass_home,
}
# --- Append to log (never block) ---
try:
_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(_LOG_FILE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
except Exception:
pass
if __name__ == "__main__":
try:
main()
except Exception:
pass
sys.exit(0)
+5 -1
View File
@@ -3,7 +3,11 @@
"CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS": "1"
},
"permissions": {
"allow": [],
"allow": [
"Write(.claude/hooks/probes/**)",
"Edit(.claude/hooks/probes/**)",
"Bash(mkdir*)"
],
"deny": [
"EnterPlanMode",
"Bash(git add -f*)",
+2
View File
@@ -136,3 +136,5 @@ STATUS.local.md
src/aipass/*/apps/integrations/**
!src/aipass/*/apps/integrations/README.md
.coverage
claude_4_7_transition_notes.md
.claude/hooks/probes/last_ping.jsonl
@@ -378,6 +378,69 @@ def deliver_email_to_branch(
return True, ""
def deliver_to_inbox_file(inbox_file: Path, email_data: Dict) -> Tuple[bool, str, str]:
"""Write *email_data* to an inbox.json file and fire a desktop notification.
Single canonical path for direct-path delivery (used by cross-project
reply.py to replace the raw-write backdoor). Always fires notify-send.
Args:
inbox_file: Absolute path to the target inbox.json.
email_data: Dict with at minimum ``from``, ``to``, ``subject``,
``message``, ``timestamp``. An ``id`` key is assigned
internally if absent.
Returns:
``(success, error_msg, reply_id)`` — ``reply_id`` is the 8-char hex
string assigned to the message (empty string on failure).
"""
if not inbox_file.exists():
return False, f"inbox not found: {inbox_file}", ""
try:
with _get_inbox_lock()(inbox_file):
try:
with open(inbox_file, "r", encoding="utf-8") as fh:
inbox_data = json.load(fh)
except Exception as exc:
logger.warning("[delivery] deliver_to_inbox_file read failed %s: %s", inbox_file, exc)
return False, f"Failed to read inbox: {exc}", ""
inbox_data = _migrate_inbox_format(inbox_data, inbox_file)
reply_id = str(uuid.uuid4())[:8]
email_data = dict(email_data)
email_data.setdefault("id", reply_id)
reply_id = email_data["id"]
inbox_data.setdefault("messages", []).insert(0, email_data)
inbox_data["total_messages"] = len(inbox_data["messages"])
inbox_data["unread_count"] = sum(
1
for m in inbox_data["messages"]
if m.get("status") == "new" or (m.get("status") is None and not m.get("read", False))
)
try:
with open(inbox_file, "w", encoding="utf-8") as fh:
json.dump(inbox_data, fh, indent=2, ensure_ascii=False)
except Exception as exc:
logger.warning("[delivery] deliver_to_inbox_file write failed %s: %s", inbox_file, exc)
return False, f"Failed to write inbox: {exc}", ""
except OSError as exc:
logger.warning("[delivery] deliver_to_inbox_file lock failed %s: %s", inbox_file, exc)
return False, f"Failed to acquire inbox lock: {exc}", ""
_send_desktop_notification(
email_data.get("from", "@unknown"),
email_data.get("to", str(inbox_file)),
email_data.get("subject", ""),
email_data.get("message", ""),
)
return True, "", reply_id
_NOTIFICATION_TIMESTAMPS: Dict[str, List[float]] = {}
# Rate limit: max notifications per recipient within time window
@@ -20,6 +20,7 @@ from datetime import datetime
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.ai_mail.apps.handlers.json import json_handler
from aipass.ai_mail.apps.handlers.email.delivery import deliver_to_inbox_file
# Services imported in __main__ only (handlers should not display)
@@ -185,31 +186,12 @@ def _deliver_via_reply_path(
Tuple of (success, message, reply_id or None)
"""
inbox_file = Path(reply_path)
if not inbox_file.exists():
return False, f"reply_path inbox not found: {reply_path}", None
success, error_msg, reply_id = deliver_to_inbox_file(inbox_file, reply_email_data)
if not success:
logger.warning("[reply] _deliver_via_reply_path failed for %s: %s", reply_path, error_msg)
return False, f"Failed to deliver to reply_path: {error_msg}", None
try:
with open(inbox_file, "r", encoding="utf-8") as f:
inbox_data = json.load(f)
except Exception as e:
logger.warning("[reply] _deliver_via_reply_path read failed %s: %s", reply_path, e)
return False, f"Failed to read target inbox: {e}", None
reply_id = str(uuid.uuid4())[:8]
reply_email_data["id"] = reply_id
inbox_data.setdefault("messages", []).insert(0, reply_email_data)
inbox_data["total_messages"] = len(inbox_data["messages"])
new_count = sum(1 for m in inbox_data["messages"] if m.get("status") == "new" or not m.get("read", False))
inbox_data["unread_count"] = new_count
try:
with open(inbox_file, "w", encoding="utf-8") as f:
json.dump(inbox_data, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.warning("[reply] _deliver_via_reply_path write failed %s: %s", reply_path, e)
return False, f"Failed to write to target inbox: {e}", None
logger.info("[reply] Cross-project reply delivered to %s", reply_path)
# Save to sender's sent folder
@@ -0,0 +1,179 @@
# =================== AIPass ====================
# Name: status_handler_gitpython.py
# Description: GitPython prototype for scoped git status (DPLAN-0140 Phase 1)
# Version: 0.1.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""
GitPython prototype for scoped git status -- DPLAN-0140 Phase 1.
Drop-in replacement for status_handler.py that uses GitPython's ``Repo``
object instead of ``subprocess.run(["git", "status", "--porcelain"])``.
The return dict format is identical to the subprocess version::
{
"files": [{"status": str, "path": str}, ...],
"total": int,
"message": str,
}
Status codes mapped from GitPython change_type:
M modified (staged or unstaged)
A added / new in index
D deleted
R renamed
? untracked (working-tree new, not staged)
Design note (two-library split):
GitHub CLI interactions (gh pr create, gh pr list, gh pr merge) are kept
as subprocess calls because they require the gh binary's authentication
context and REST logic. GitPython covers all *local* git operations.
This split is intentional and documented in the Phase 1 investigation
report at docs.local/gitpython_investigation_2026-04-20.md.
"""
from __future__ import annotations
from pathlib import Path
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
try:
import git as _git_module
_GITPYTHON_AVAILABLE = True
except ImportError:
_GITPYTHON_AVAILABLE = False
# Map GitPython diff change_type codes to porcelain-compatible single letters.
_STAGED_STATUS_MAP: dict[str, str] = {
"A": "A",
"D": "D",
"M": "M",
"R": "R",
"C": "C",
"T": "T",
"U": "U",
}
_UNSTAGED_STATUS_MAP: dict[str, str] = {
"D": "D",
"M": "M",
"R": "R",
"A": "A",
}
def _collect_staged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
"""Return staged changes that fall under the branch directory."""
files: list[dict] = []
try:
staged_diffs = repo.head.commit.diff()
except Exception as exc: # empty repo or detached HEAD
logger.debug("status_handler_gitpython: could not get staged diffs: %s", exc)
return files
for diff in staged_diffs:
path = diff.b_path or diff.a_path
if not path:
continue
if not (path.startswith(rel_prefix) or path == rel_dir):
continue
code = _STAGED_STATUS_MAP.get(diff.change_type, diff.change_type)
files.append({"status": code, "path": path})
return files
def _collect_unstaged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
"""Return unstaged working-tree changes that fall under the branch directory."""
files: list[dict] = []
for diff in repo.index.diff(None):
path = diff.b_path or diff.a_path
if not path:
continue
if not (path.startswith(rel_prefix) or path == rel_dir):
continue
code = _UNSTAGED_STATUS_MAP.get(diff.change_type, diff.change_type)
files.append({"status": code, "path": path})
return files
def _collect_untracked(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
"""Return untracked files that fall under the branch directory."""
files: list[dict] = []
for upath in repo.untracked_files:
if upath.startswith(rel_prefix) or upath == rel_dir:
files.append({"status": "?", "path": upath})
return files
def get_branch_status(branch_dir: Path) -> dict:
"""Get git status filtered to files under branch_dir using GitPython.
This is a drop-in replacement for status_handler.get_branch_status().
The return format is identical; callers do not need to change.
Args:
branch_dir: Absolute path to the branch directory to scope output to.
Returns:
Dict with:
files -- list of {"status": str, "path": str} dicts
total -- int count of changed files
message -- human-readable summary string
"""
if not _GITPYTHON_AVAILABLE:
logger.error(
"status_handler_gitpython: GitPython is not installed. "
"Run: pip install gitpython"
)
return {
"files": [],
"total": 0,
"message": "GitPython not available -- install with: pip install gitpython",
}
repo_root = find_repo_root()
try:
repo = _git_module.Repo(str(repo_root))
except _git_module.InvalidGitRepositoryError as exc:
logger.error("status_handler_gitpython: not a git repository at %s: %s", repo_root, exc)
return {"files": [], "total": 0, "message": f"Not a git repository: {exc}"}
except _git_module.GitCommandNotFound as exc:
logger.error("status_handler_gitpython: git not found: %s", exc)
return {"files": [], "total": 0, "message": f"git not found: {exc}"}
# Compute relative scope for filtering -- identical logic to subprocess version.
try:
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
except ValueError:
logger.warning(
"get_branch_status: branch_dir %s not relative to repo root %s, using absolute",
branch_dir,
repo_root,
)
rel_dir = branch_dir
rel_prefix = str(rel_dir) + "/"
rel_dir_str = str(rel_dir)
files: list[dict] = []
files.extend(_collect_staged(repo, rel_prefix, rel_dir_str))
files.extend(_collect_unstaged(repo, rel_prefix, rel_dir_str))
files.extend(_collect_untracked(repo, rel_prefix, rel_dir_str))
total = len(files)
message = f"{total} file(s) changed under {rel_dir}"
json_handler.log_operation(
"get_branch_status_gitpython",
{"branch_dir": str(branch_dir), "total": total},
)
logger.info(message)
return {"files": files, "total": total, "message": message}
@@ -20,8 +20,9 @@ from pathlib import Path
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
ALLOWED_CALLERS: list[str] = ["devpulse"]
ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
def verify_caller() -> str:
@@ -52,7 +53,7 @@ def verify_caller() -> str:
logger.error(msg)
raise PermissionError(msg)
if name not in ALLOWED_CALLERS:
msg = f"Branch '{name}' is not authorized for system-pr. Allowed callers: {ALLOWED_CALLERS}"
msg = f"Branch '{name}' is not authorized for system-pr. Trusted cross-writers: {ALLOWED_CALLERS}"
logger.error(msg)
raise PermissionError(msg)
json_handler.log_operation(
@@ -0,0 +1,231 @@
# DPLAN-0140 Phase 1 — GitPython Investigation Report
**Date:** 2026-04-21
**Author:** @drone (builder agent)
**Branch:** proto/drone-dplan-0140-phase1
**Scope:** Phase 1 only — investigation, prototype, benchmarks. Phase 2/3 not included.
---
## 1. Current Subprocess Inventory
~40 subprocess calls across 8 files. Organized by file:
### `lock_handler.py` (1 call)
| Command | Purpose |
|---------|---------|
| `git rev-parse --show-toplevel` | find_repo_root() fallback when AIPASS_REGISTRY.json walk fails |
### `status_handler.py` (1 call)
| Command | Purpose |
|---------|---------|
| `git status --porcelain` | Full working-tree status, string-parsed line-by-line |
### `sync_handler.py` (5 calls)
| Command | Purpose |
|---------|---------|
| `git checkout main` | Switch to main branch |
| `git fetch origin` | Fetch remote refs |
| `git rev-list --left-right --count main...origin/main` | Ahead/behind count, string split + int() |
| `git merge origin/main --no-edit` | Fast-forward merge |
| `git pull --rebase` | Rebase pull |
| `git stash` / `git stash pop` | Autostash before/after sync |
### `pr_handler.py` (8 calls — mixed git + gh)
| Command | Purpose |
|---------|---------|
| `git rev-parse --abbrev-ref HEAD` | Get current branch name |
| `git add <path>/` | Stage branch directory |
| `git diff --cached --quiet` | Check if anything staged |
| `git commit -m <msg> -- <path>/` | Commit staged changes |
| `git branch -f <feature>` | Force-move feature branch pointer |
| `git push --force-with-lease` | Push feature branch |
| `git branch -D <feature>` | Delete local feature branch |
| `gh pr create`, `gh pr list` | GitHub API (stays subprocess — see Section 5) |
### `merge_plugin.py` (6 calls — mixed git + gh)
| Command | Purpose |
|---------|---------|
| `gh pr merge` | Merge PR via GitHub API |
| `git stash` / `git stash pop` | State preservation |
| `git pull --rebase` | Sync after merge |
| `git rev-parse HEAD` | Get current commit SHA |
| `gh pr view` | Read PR metadata (GitHub API) |
### `pr_plugin.py` / system-pr (8 calls — mixed)
| Command | Purpose |
|---------|---------|
| `git rev-parse --abbrev-ref HEAD` | Branch name |
| `git add -A` | Stage everything |
| `git reset HEAD .git_pr.lock` | Unstage lock file |
| `git diff --cached --quiet` | Check staged state |
| `git commit -m <msg>` | Commit |
| `git fetch origin main` | Fetch main |
| `git rev-list --count origin/main..HEAD` | Commit count ahead |
| `git branch -f`, `git push --force-with-lease`, `git branch -D` | Branch management |
| `gh pr create` | GitHub API |
### `sync_plugin.py` (smart-sync, 6 calls)
| Command | Purpose |
|---------|---------|
| `git fetch origin` | Fetch remote |
| `git rev-list --left-right --count main...origin/main` | Ahead/behind, string-parsed |
| `git merge origin/main --no-edit` | Merge |
| `git diff --name-only --diff-filter=U` | List conflict files, string-parsed |
| `git merge --abort` | Abort failed merge |
| `git rebase origin/main` / `git rebase --abort` | Rebase path |
### `fix_plugin.py` (9 calls)
| Command | Purpose |
|---------|---------|
| `git rebase --abort` | Abort rebase |
| `git symbolic-ref -q HEAD` | Detect detached HEAD state |
| `git checkout main` | Switch to main |
| `git fetch origin` | Fetch remote |
| `git rev-list --left-right --count main...origin/main` | Ahead/behind |
| `git merge origin/main --no-edit` | Merge |
| `git diff --name-only --diff-filter=U` | Conflict file list |
| `git merge --abort` | Abort merge |
| `git diff --cached --name-only` | Staged file list |
| `git reset HEAD` | Unstage all |
**Total: ~44 subprocess calls, 8 files.** GitHub CLI calls (gh) account for ~8 of these and must remain as subprocess regardless of library choice.
---
## 2. Library Comparison Matrix
| Criterion | GitPython 3.1.46 | pygit2 1.19.2 | dulwich 1.1.0 |
|-----------|-----------------|---------------|----------------|
| **Latest release** | 3.1.46 (2025) | 1.19.2 (2025) | 1.1.0 (2025) |
| **PyPI release count** | 99 releases | Active | Active |
| **Maintenance health** | Active, well-maintained | Active | Active |
| **API style** | Pythonic, high-level | C-extension wrapping libgit2, lower-level | Pure Python, porcelain-style |
| **Native deps** | None (pure Python: gitdb + smmap) | libgit2 shared library required | None (pure Python) |
| **Windows support** | Excellent — no native deps, pip install works everywhere | Problematic — libgit2 must be available, wheel availability varies | Good — pure Python |
| **API coverage** | High-level for common ops; shell fallback for exotic commands | Full libgit2 surface, lower-level | Limited high-level API |
| **Error handling** | GitCommandError with stdout/stderr captured | GitError (C-level), less descriptive | Exceptions from pure Python |
| **Avg invocation time** | 27.9ms (fresh Repo()) / 26.9ms (cached) | 30.2ms | 585.3ms |
| **Min invocation time** | 21.4ms | 28.2ms | 564.1ms |
| **Subprocess overhead** | ~14ms baseline (current) | ~14ms baseline | ~14ms baseline |
| **Learning curve** | Low — familiar Python object model | Medium — libgit2 concepts leak through | Low — porcelain API simple but limited |
| **Documentation** | Good, stable | Good, thorough | Adequate |
### Notes on benchmark conditions
- All measurements: 20 iterations, Python 3.12, Linux 6.17, AIPass repo (clean working tree except one untracked file).
- Subprocess baseline (current `status_handler.py`): avg 13.9ms, min 11.7ms.
- GitPython is ~2x slower than subprocess on a clean repo. The delta collapses for dirty repos where parsing overhead matters.
- dulwich (585ms avg) is disqualifying for interactive use — internal reimplementation of pack/object reads in Python accounts for the slowdown.
- pygit2 (30.2ms) is fast but requires libgit2 native library — this is a hard blocker for Windows compatibility.
---
## 3. Recommendation
**Use GitPython.**
Rationale: GitPython is pure Python (no native deps), works identically on Windows and Linux, has the most Pythonic API of the three candidates, and covers all ~36 local git operations in the audit with first-class support. The 2x overhead vs subprocess (28ms vs 14ms) is acceptable given that drone's git operations are not hot paths — they run at PR/sync cadence, not in tight loops.
pygit2 would be faster but libgit2 dependency breaks Windows support, which is a stated requirement for @cli. dulwich is disqualified on performance alone (585ms vs 14ms).
---
## 4. Prototype Benchmarks
Benchmark environment: Python 3.12.x, Linux 6.17, AIPass repo, 20 iterations each, clean working tree with 1 untracked file.
| Implementation | Avg | Min | Max |
|----------------|-----|-----|-----|
| subprocess (current) | 13.9ms | 11.7ms | 26.1ms |
| GitPython (fresh Repo() per call) | 27.9ms | 21.4ms | 49.2ms |
| GitPython (cached Repo object) | 26.9ms | 19.7ms | n/a |
| pygit2 (fresh Repository() per call) | 30.2ms | 28.2ms | n/a |
| dulwich | 585.3ms | 564.1ms | n/a |
**Verdict:** GitPython adds ~14ms overhead per call. At drone's usage cadence this is imperceptible. The overhead buys: no process fork, structured error objects, and type-safe diff iteration.
---
## 5. @git pr Trade-offs: Two-Library Split
**Question:** Can we use GitPython for local git work while keeping `gh` subprocess for GitHub API calls?
**Answer: Yes. The split is correct and clean.**
Reasoning:
1. `gh` is an OAuth-authenticated CLI that manages GitHub REST API state (PR creation, merge, review status, checks). GitPython has no equivalent — it only knows the local `.git` directory.
2. The two surfaces don't overlap. Local commits, branches, diffs, staging, stash = GitPython. GitHub PR lifecycle = gh subprocess.
3. This pattern is standard in Git tooling (e.g. hub, lab, glab all work this way).
4. Error handling stays clean: GitPython raises `git.GitCommandError`; gh failures surface through returncode + stderr as before.
Concrete split for drone's files:
| File | GitPython replaces | gh stays subprocess |
|------|--------------------|---------------------|
| status_handler.py | `git status --porcelain` | — |
| lock_handler.py | `git rev-parse --show-toplevel` | — |
| sync_handler.py | fetch, merge, rebase, stash, rev-list | — |
| pr_handler.py | add, diff, commit, branch, push | `gh pr create`, `gh pr list` |
| merge_plugin.py | stash, pull, rev-parse | `gh pr merge`, `gh pr view` |
| pr_plugin.py | add, reset, diff, commit, fetch, rev-list, branch, push | `gh pr create` |
| sync_plugin.py | fetch, merge, rebase, diff | — |
| fix_plugin.py | rebase, symbolic-ref, checkout, fetch, merge, diff, reset | — |
---
## 6. Known Pain Points
### Pathspec Scope Limitation
**Problem:** `drone @git pr` stages only the caller's branch directory via `git add <path>/`. This path-scoped add cannot reach cross-directory paths such as repo-root `.claude/hooks/` or `.aipass/registry.json`.
**Impact:** @seedgo hit this limitation 3x during hook consolidation work (PRs #371, #372, #373) — hook files at `.claude/hooks/` were not staged because they live outside the branch directory prefix.
**Current subprocess behavior:** `git add <branch_dir>/` — silently ignores everything outside that prefix.
**GitPython fix available:**
```python
# Current (subprocess):
subprocess.run(["git", "add", str(branch_dir) + "/"], ...)
# GitPython replacement:
repo.index.add(["src/aipass/seedgo/", ".claude/hooks/post_tool_use.py"])
```
`repo.index.add()` accepts an explicit path list, enabling multi-directory staging without accidentally bundling unrelated files. This is the recommended fix for Phase 2 — the caller explicitly opts in to each path, eliminating silent-omission bugs.
**Workaround until Phase 2:** Callers that need cross-directory staging must issue a separate `drone @git pr` invocation from the repo root, or use the system-pr plugin (which uses `git add -A` + `git reset` to exclude lock files).
---
## 7. Proposed Phase 2 Surface Expansion Priorities
From DPLAN-0140 planning notes:
**Tier 1 — Replace first (high value, low risk):**
- `git stash` / `git stash pop` — GitPython: `repo.git.stash()` / `repo.git.stash("pop")`
- `git fetch origin` — GitPython: `repo.remote("origin").fetch()`
- `git rev-parse --abbrev-ref HEAD` — GitPython: `repo.active_branch.name`
- `git rev-parse HEAD` — GitPython: `repo.head.commit.hexsha`
- `git diff --cached --quiet` — GitPython: `bool(repo.index.diff("HEAD"))`
- `git add <path>` — GitPython: `repo.index.add([path])` (fixes pathspec bug above)
- `git commit -m <msg>` — GitPython: `repo.index.commit(msg)`
- `git status --porcelain` — DONE (this prototype)
- `git rev-parse --show-toplevel` — GitPython: `Repo.working_tree_dir`
**Tier 2 — Replace second (more complex, higher value):**
- `git reset HEAD` — GitPython: `repo.index.reset()`
- `git revert` — GitPython: `repo.git.revert()`
- `git cherry-pick` — GitPython: `repo.git.cherry_pick(sha)`
- `git rev-list --count` / `--left-right` — GitPython: `repo.iter_commits()` + `repo.merge_base()`
- `git branch -f`, `git branch -D` — GitPython: `repo.create_head()`, `repo.delete_head()`
**Tier 3 — Later (rarely used, lower ROI for Phase 2):**
- `git tag`, `git bisect`, `git blame`, `git reflog`
**Stays subprocess forever:**
- All `gh` commands (GitHub API, no GitPython equivalent)
- `git symbolic-ref -q HEAD` (GitPython equivalent is `repo.head.is_detached`)
@@ -43,7 +43,7 @@ from aipass.drone.apps.handlers.exceptions import (
@pytest.fixture
def registry_dir() -> Generator[Path, None, None]:
"""Isolated temp directory for registry tests; cleaned up after."""
d = Path(tempfile.mkdtemp(prefix="reg_test_"))
d = Path(tempfile.mkdtemp(prefix="reg_test_")).resolve()
yield d
shutil.rmtree(d, ignore_errors=True)
+20
View File
@@ -10,6 +10,11 @@
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_hooks_probe.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_checkers_batch5.py",
"standard": "encapsulation",
@@ -219,6 +224,21 @@
"file": "templates/",
"standard": "unused_function",
"reason": "Template files are reference implementations for other branches to copy. They contain function definitions that are not called within seedgo itself."
},
{
"file": "tests/test_hooks_track_e.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_hooks_track_e.py",
"standard": "encapsulation",
"reason": "Unit tests must import handlers directly to test them in isolation. Same pattern as test_checkers_batch5.py."
},
{
"file": "apps/modules/permissions.py",
"standard": "unused_function",
"reason": "is_trusted_caller() and identify_caller() are public API consumed by pre_edit_gate.py (hook layer) and drone auth.py. Checker cannot trace cross-file dynamic dispatch to the hook scripts."
}
],
"notes": {
+559
View File
@@ -0,0 +1,559 @@
# =================== AIPass ====================
# Name: hooks.py
# Description: Hook Probe Display Module
# Version: 1.0.0
# Created: 2026-04-20
# Modified: 2026-04-20
# =============================================
"""
Hook Probe Display Module
Reads last_ping.jsonl produced by .claude/hooks/probes/ scripts and
surfaces probe data via Rich tables and reports.
Run: drone @seedgo hooks probe [--subagent|--matrix]
Subcommands:
hooks probe Show recent probe entries as a Rich table
hooks probe --subagent Spawn headless Claude, check if PostToolUse/SubagentStop fired
hooks probe --matrix Full event matrix + markdown report
"""
import json
import subprocess
import sys
import time
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import List
# =============================================================================
# INFRASTRUCTURE SETUP
# =============================================================================
# IMPORTS
# =============================================================================
# Prax logger (system-wide, always first)
from aipass.prax import logger
# CLI services (display/output formatting)
from aipass.cli import console
from aipass.cli.apps.modules import warning
# JSON handler for tracking
from aipass.seedgo.apps.handlers.json import json_handler
# Rich output
from rich.panel import Panel
from rich.table import Table
# =============================================================================
# PATHS
# =============================================================================
_REPO_ROOT: Path | None = None
# Enable snippet shown when no probe data exists — uses a placeholder path
# so the help_text checker does not flag a literal drone command path.
_PROBE_ENABLE_NOTE = (
"Add probe hook commands to .claude/settings.json (inside the hooks object).\n"
"See .claude/hooks/probes/README.md for per-event snippets."
)
def _get_repo_root() -> Path:
"""Return the AIPass repo root derived from this file's location."""
global _REPO_ROOT
if _REPO_ROOT is not None:
return _REPO_ROOT
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
_REPO_ROOT = parent
return parent
# Fallback: assume 5 levels up from this file
_REPO_ROOT = Path(__file__).resolve().parents[5]
return _REPO_ROOT
def _log_file() -> Path:
"""Return path to last_ping.jsonl."""
return _get_repo_root() / ".claude" / "hooks" / "probes" / "last_ping.jsonl"
def _probes_dir() -> Path:
"""Return path to probes directory."""
return _get_repo_root() / ".claude" / "hooks" / "probes"
# =============================================================================
# JSONL READING
# =============================================================================
def _parse_jsonl_line(line: str) -> dict | None:
"""Parse a single JSONL line. Returns None on parse failure."""
stripped = line.strip()
if not stripped:
return None
try:
return json.loads(stripped)
except json.JSONDecodeError as exc:
logger.info("hooks.py: skipping malformed JSONL line: %s", exc)
return None
def _read_entries(log_path: Path | None = None) -> List[dict]:
"""Read all entries from last_ping.jsonl. Returns empty list if missing or unreadable."""
path = log_path or _log_file()
if not path.exists():
return []
try:
with open(path, "r", encoding="utf-8") as fh:
raw_lines = fh.readlines()
except OSError as exc:
logger.info("hooks.py: could not read %s: %s", path, exc)
return []
entries = []
for line in raw_lines:
entry = _parse_jsonl_line(line)
if entry is not None:
entries.append(entry)
return entries
def _truncate(s: str, n: int) -> str:
"""Truncate string to n chars with ellipsis."""
if len(s) <= n:
return s
return s[: n - 3] + "..."
# =============================================================================
# SUBCOMMAND: hooks probe (no flags)
# =============================================================================
def _cmd_probe_display(log_path: Path | None = None) -> None:
"""Display recent probe entries as a Rich table."""
entries = _read_entries(log_path)
if not entries:
console.print(
Panel(
"[yellow]No probe data yet.[/yellow]\n\n"
"Enable probes by adding hook commands to [cyan].claude/settings.json[/cyan].\n"
f"{_PROBE_ENABLE_NOTE}",
title="[bold cyan][PROBE][/bold cyan]",
border_style="cyan",
)
)
return
# Show last 50 entries
recent = entries[-50:]
table = Table(
title="[bold cyan][PROBE][/bold cyan] Recent Hook Pings",
show_header=True,
header_style="bold cyan",
border_style="dim",
expand=False,
)
table.add_column("Event", style="cyan", no_wrap=True)
table.add_column("Tool", style="green")
table.add_column("Timestamp", style="dim")
table.add_column("CWD", style="dim")
table.add_column("Agent ID", style="dim")
table.add_column("Elapsed ms", justify="right", style="yellow")
table.add_column("CLI Ver", style="dim")
table.add_column("ProjectDir", justify="center")
table.add_column("AIPassHome", justify="center")
for e in recent:
table.add_row(
e.get("event", "?"),
_truncate(e.get("tool", ""), 20),
_truncate(e.get("timestamp", ""), 24),
_truncate(e.get("cwd", ""), 30),
_truncate(e.get("agent_id", ""), 14),
str(round(e.get("script_elapsed_ms", 0), 1)),
_truncate(e.get("cli_version", "?"), 10),
"[green]Y[/green]" if e.get("env_has_claude_project_dir") else "[red]N[/red]",
"[green]Y[/green]" if e.get("env_has_aipass_home") else "[red]N[/red]",
)
console.print()
console.print(table)
console.print(f"\n[dim]Showing {len(recent)} of {len(entries)} total entries from {_log_file()}[/dim]\n")
json_handler.log_operation("hooks_probe_display", {"entries_shown": len(recent), "total": len(entries)})
# =============================================================================
# SUBCOMMAND: hooks probe --subagent
# =============================================================================
def _run_headless_claude() -> int:
"""Spawn headless claude with a Read tool call. Returns exit code."""
canary = Path("/tmp/probe_canary.txt")
try:
canary.write_text("probe canary 2026-04-20\n", encoding="utf-8")
except OSError as exc:
logger.info("hooks.py: could not write canary: %s", exc)
console.print("[dim]Spawning headless claude...[/dim]")
# --permission-mode bypassPermissions is the AIPass-approved bypass flag
try:
result = subprocess.run(
[
"claude",
"-p",
"--permission-mode",
"bypassPermissions",
"read the file /tmp/probe_canary.txt",
"--allowedTools",
"Read",
],
capture_output=True,
text=True,
timeout=30,
)
console.print(f"[dim]claude exited: {result.returncode}[/dim]")
return result.returncode
except FileNotFoundError as exc:
logger.info("hooks.py: claude binary not found: %s", exc)
warning("claude binary not found — skipping spawn test")
return -1
except subprocess.TimeoutExpired as exc:
logger.info("hooks.py: claude timed out: %s", exc)
warning("claude timed out after 30s")
return -2
except OSError as exc:
logger.info("hooks.py: spawn error: %s", exc)
warning(f"spawn error: {exc}")
return -3
def _cmd_probe_subagent() -> None:
"""Spawn a headless Claude Code process and check if PostToolUse/SubagentStop fired."""
console.print()
console.print(
Panel(
"[bold cyan]Headless Probe — PostToolUse + SubagentStop[/bold cyan]\n\n"
"This test spawns a headless claude process with a Read tool call, "
"then checks last_ping.jsonl for "
"[yellow]PostToolUse[/yellow] and [yellow]SubagentStop[/yellow] entries "
"created within the last 10 seconds.\n\n"
"[dim]Requires PostToolUse and SubagentStop probes to be enabled in settings.json.[/dim]",
title="[bold cyan][PROBE][/bold cyan]",
border_style="cyan",
)
)
start_ts = time.time()
_run_headless_claude()
# Wait a beat for hooks to flush
time.sleep(0.5)
# Check last_ping.jsonl for recent entries
cutoff = start_ts - 1.0 # 1s before spawn
entries = _read_entries()
recent_events = {e.get("event") for e in entries if _entry_ts(e) >= cutoff}
post_tool_use_fired = "PostToolUse" in recent_events
subagent_stop_fired = "SubagentStop" in recent_events
result_table = Table(show_header=True, header_style="bold cyan", border_style="dim")
result_table.add_column("Check", style="cyan")
result_table.add_column("Result", justify="center")
result_table.add_row(
"PostToolUse fired in headless",
"[green]YES[/green]" if post_tool_use_fired else "[red]NO[/red]",
)
result_table.add_row(
"SubagentStop fired in headless",
"[green]YES[/green]" if subagent_stop_fired else "[red]NO[/red]",
)
console.print(result_table)
console.print()
# Manual test section
console.print(
Panel(
"[bold yellow]Manual Interactive Agent-Tool Test[/bold yellow]\n\n"
"To test hooks fired by the interactive Agent tool:\n\n"
"1. Enable PostToolUse and SubagentStop probes in .claude/settings.json\n"
"2. In an interactive Claude Code session, invoke an Agent tool call\n"
"3. After the agent completes, run:\n"
" [cyan]drone @seedgo hooks probe[/cyan]\n"
"4. Check for SubagentStop and PostToolUse entries near the agent's timestamp\n\n"
"[dim]Headless (-p) and interactive Agent tool have different session contexts.[/dim]\n"
"[dim]SubagentStop fires when an agent tool invocation completes.[/dim]",
title="[bold cyan][PROBE][/bold cyan] Manual Test",
border_style="dim",
)
)
json_handler.log_operation(
"hooks_probe_subagent",
{"post_tool_use_fired": post_tool_use_fired, "subagent_stop_fired": subagent_stop_fired},
)
def _entry_ts(entry: dict) -> float:
"""Parse entry timestamp to unix float. Returns 0 on failure."""
ts_str = entry.get("timestamp", "")
if not ts_str:
return 0.0
try:
normalized = ts_str[:-1] + "+00:00" if ts_str.endswith("Z") else ts_str
dt = datetime.fromisoformat(normalized)
return dt.timestamp()
except ValueError as exc:
logger.info("hooks.py: could not parse timestamp %r: %s", ts_str, exc)
return 0.0
# =============================================================================
# SUBCOMMAND: hooks probe --matrix
# =============================================================================
def _build_matrix_rows(entries: list) -> tuple[list, dict]:
"""Group entries by event and build matrix row dicts. Returns (rows, groups)."""
groups: dict = defaultdict(list)
for e in entries:
groups[e.get("event", "unknown")].append(e)
matrix_rows = []
for event_name in sorted(groups.keys()):
evs = groups[event_name]
count = len(evs)
pd_true = sum(1 for e in evs if e.get("env_has_claude_project_dir"))
pd_false = count - pd_true
ah_true = sum(1 for e in evs if e.get("env_has_aipass_home"))
ah_false = count - ah_true
agents = len({e.get("agent_id", "unknown") for e in evs})
matrix_rows.append(
{
"event": event_name,
"count": count,
"project_dir_true": pd_true,
"project_dir_false": pd_false,
"aipass_home_true": ah_true,
"aipass_home_false": ah_false,
"unique_agents": agents,
}
)
return matrix_rows, groups
def _cmd_probe_matrix() -> None:
"""Group entries by event, show matrix, write markdown report."""
entries = _read_entries()
if not entries:
console.print(
Panel(
f"[yellow]No probe data yet — enable probes first.[/yellow]\n\n{_PROBE_ENABLE_NOTE}",
title="[bold cyan][PROBE] Matrix[/bold cyan]",
border_style="cyan",
)
)
return
matrix_rows, groups = _build_matrix_rows(entries)
# Build matrix table
table = Table(
title="[bold cyan][PROBE][/bold cyan] Event Matrix",
show_header=True,
header_style="bold cyan",
border_style="dim",
)
table.add_column("Event", style="cyan", no_wrap=True)
table.add_column("Count", justify="right", style="yellow")
table.add_column("ProjectDir T/F", justify="center")
table.add_column("AIPassHome T/F", justify="center")
table.add_column("Unique Agents", justify="right")
for row in matrix_rows:
table.add_row(
row["event"],
str(row["count"]),
f"[green]{row['project_dir_true']}[/green]/[red]{row['project_dir_false']}[/red]",
f"[green]{row['aipass_home_true']}[/green]/[red]{row['aipass_home_false']}[/red]",
str(row["unique_agents"]),
)
console.print()
console.print(table)
console.print()
# Write markdown report
report_path = _probes_dir() / "Q12_findings_2026-04-20.md"
_write_matrix_report(report_path, matrix_rows, entries)
console.print(f"[green]Report written:[/green] {report_path}\n")
json_handler.log_operation("hooks_probe_matrix", {"events": len(groups), "total_entries": len(entries)})
def _write_matrix_report(report_path: Path, matrix_rows: list, entries: list) -> None:
"""Write markdown matrix report to disk."""
lines = [
"# Hook Probe Matrix — Q12 Findings 2026-04-20",
"",
"Generated by `drone @seedgo hooks probe --matrix`.",
"",
f"Total entries: {len(entries)}",
"",
"## Event Matrix",
"",
"| Event | Count | ProjectDir T/F | AIPassHome T/F | Unique Agents |",
"|-------|-------|----------------|----------------|---------------|",
]
for row in matrix_rows:
lines.append(
f"| {row['event']} | {row['count']} "
f"| {row['project_dir_true']}/{row['project_dir_false']} "
f"| {row['aipass_home_true']}/{row['aipass_home_false']} "
f"| {row['unique_agents']} |"
)
lines += [
"",
"## Notes",
"",
"- `ProjectDir T/F`: entries where `CLAUDE_PROJECT_DIR` env var was set (T) vs unset (F)",
"- `AIPassHome T/F`: entries where `AIPASS_HOME` env var was set (T) vs unset (F)",
"- Unique Agents: distinct `CLAUDE_CODE_SESSION_ID` values seen for this event",
"",
"## Raw entry count by event",
"",
]
event_counts = Counter(e.get("event", "unknown") for e in entries)
for ev, cnt in sorted(event_counts.items()):
lines.append(f"- {ev}: {cnt}")
lines.append("")
try:
report_path.parent.mkdir(parents=True, exist_ok=True)
with open(report_path, "w", encoding="utf-8") as fh:
fh.write("\n".join(lines))
except OSError as exc:
logger.info("hooks.py: failed to write report to %s: %s", report_path, exc)
# =============================================================================
# INTROSPECTION
# =============================================================================
def print_introspection() -> None:
"""Display module info and usage."""
console.print()
console.print("[bold cyan]hooks Module[/bold cyan]")
console.print("Hook probe display — reads last_ping.jsonl from .claude/hooks/probes/")
console.print()
log = _log_file()
if log.exists():
entries = _read_entries()
console.print(f"[yellow]Probe log:[/yellow] {log} ([green]{len(entries)} entries[/green])")
else:
console.print(f"[yellow]Probe log:[/yellow] {log} [dim](not yet created — enable probes first)[/dim]")
console.print()
console.print("[yellow]Subcommands:[/yellow]")
console.print(" [green]drone @seedgo hooks probe[/green] [dim]# Display recent entries table[/dim]")
console.print(
" [green]drone @seedgo hooks probe --subagent[/green] [dim]# Test headless PostToolUse/SubagentStop[/dim]"
)
console.print(
" [green]drone @seedgo hooks probe --matrix[/green] [dim]# Full event matrix + markdown report[/dim]"
)
console.print()
console.print("[yellow]Probe scripts:[/yellow]")
probes_dir = _probes_dir()
for script in sorted(probes_dir.glob("probe_*.py")):
console.print(f" [dim]{script.name}[/dim]")
console.print()
console.print("[yellow]Connected Handlers:[/yellow]")
console.print(" [cyan]handlers/json/[/cyan]")
console.print(" [dim]- json_handler.py (log_operation — operation tracking)[/dim]")
console.print()
console.print("[yellow]External Dependencies:[/yellow]")
console.print(" [dim]- aipass.prax (logger)[/dim]")
console.print(" [dim]- aipass.cli (console)[/dim]")
console.print()
# =============================================================================
# COMMAND HANDLER
# =============================================================================
def handle_command(command: str, args: List[str]) -> bool:
"""
Handle 'hooks' command — hook probe display and testing.
Args:
command: Command name
args: Additional arguments
[] -> print_introspection()
["probe"] -> display last_ping.jsonl table
["probe", "--subagent"] -> headless probe test
["probe", "--matrix"] -> full event matrix + report
["--help"] | ["help"] -> print_introspection()
Returns:
True if handled, False if not this module's command
"""
if command != "hooks":
return False
# No args or help -> introspection
if not args or args[0] in ("--help", "-h", "help"):
print_introspection()
return True
subcommand = args[0]
if subcommand == "probe":
sub_args = args[1:]
if "--subagent" in sub_args:
_cmd_probe_subagent()
elif "--matrix" in sub_args:
_cmd_probe_matrix()
else:
_cmd_probe_display()
return True
# Unknown subcommand — show introspection
console.print(f"[dim]Unknown subcommand: {subcommand!r} — showing help[/dim]")
print_introspection()
return True
# =============================================================================
# STANDALONE EXECUTION
# =============================================================================
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] in ("--help", "-h", "help"):
print_introspection()
sys.exit(0)
logger.info("Prax logger connected to hooks")
handle_command("hooks", sys.argv[1:])
@@ -0,0 +1,120 @@
# =================== AIPass ====================
# Name: inbox_audit.py
# Description: Inbox ID validator — scans all inbox.json files for non-8-hex ids
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""Inbox ID validator for the drone @seedgo audit inbox-ids command.
Walks all .ai_mail.local/inbox.json files in the AIPass repo and flags any
message ids that are not 8-character lowercase hex strings. Alerts devpulse
when violations are found.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import List
from aipass.prax import logger
from aipass.cli import console, header
from aipass.seedgo.apps.handlers.json import json_handler
_HEX8_RE = re.compile(r"^[0-9a-f]{8}$")
def _find_repo_root() -> Path:
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return current
def _scan_inbox(inbox_path: Path) -> List[dict]:
"""Return a list of violation dicts for messages with bad ids in *inbox_path*."""
violations: List[dict] = []
try:
data = json.loads(inbox_path.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("[inbox_audit] could not read %s: %s", inbox_path, exc)
return violations
for msg in data.get("messages", []):
msg_id = msg.get("id", "")
if not _HEX8_RE.match(str(msg_id)):
violations.append(
{
"inbox": str(inbox_path),
"id": msg_id,
"subject": msg.get("subject", ""),
"from": msg.get("from", ""),
}
)
return violations
def _run_inbox_id_scan() -> int:
"""Scan all inbox.json files; return number of violations found."""
json_handler.log_operation("inbox_audit_scan", {})
repo_root = _find_repo_root()
inbox_files = list(repo_root.rglob(".ai_mail.local/inbox.json"))
console.print()
header("SEEDGO — Inbox ID Validator")
console.print(f"[dim]Scanning {len(inbox_files)} inbox file(s) for non-8-hex message ids...[/dim]")
console.print()
all_violations: List[dict] = []
for inbox_path in sorted(inbox_files):
violations = _scan_inbox(inbox_path)
all_violations.extend(violations)
if not all_violations:
console.print("[green]✓[/green] All message ids are valid 8-char hex strings.")
console.print()
return 0
console.print(f"[red]✗[/red] Found [bold]{len(all_violations)}[/bold] id violation(s):\n")
for v in all_violations:
rel = Path(v["inbox"]).relative_to(repo_root) if Path(v["inbox"]).is_absolute() else v["inbox"]
console.print(
f" [red]•[/red] [bold]{rel}[/bold] id=[yellow]{v['id']!r}[/yellow] from={v['from']} subject={v['subject']!r}"
)
console.print()
console.print("[yellow]Action:[/yellow] Alert devpulse — run:")
console.print(
f' [green]drone @ai_mail email @devpulse "inbox-id violations" '
f'"Found {len(all_violations)} bad message id(s) — run drone @seedgo audit inbox-ids for details"[/green]'
)
console.print()
return len(all_violations)
def print_introspection() -> None:
"""Show inbox_audit module structure."""
console.print("[bold cyan]inbox_audit[/bold cyan] — Inbox ID validator")
console.print(" Connected Handlers: none (uses stdlib + pathlib only)")
console.print(" Command: drone @seedgo audit inbox-ids")
def handle_command(command: str, args: List[str]) -> bool:
"""Handle `audit inbox-ids` — return True only for that exact subcommand."""
if command not in ("audit", "standards_audit"):
return False
if not args:
print_introspection()
return True
if args[0] in ("--help", "-h", "help"):
console.print("Usage: drone @seedgo audit inbox-ids")
console.print(" Scans all .ai_mail.local/inbox.json files for non-8-hex message ids.")
return True
if args[0] != "inbox-ids":
return False
_run_inbox_id_scan()
return True
@@ -0,0 +1,77 @@
# =================== AIPass ====================
# Name: permissions.py
# Description: Shared trust list for hook layer and drone authorization
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""Shared permission definitions for cross-branch write authorization.
Single source of truth for which branches may write outside their own
directory. Consumed by pre_edit_gate.py (hook layer) and
drone/apps/plugins/devpulse_ops/auth.py (drone layer).
"""
from __future__ import annotations
import json
from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
def is_trusted_caller(name: str) -> bool:
"""Return True if *name* is in TRUSTED_CROSS_WRITERS."""
json_handler.log_operation("is_trusted_caller", {"name": name})
return name.lower() in TRUSTED_CROSS_WRITERS
def identify_caller(cwd: str | None = None) -> str:
"""Walk up from *cwd* (default: CWD) to find passport.json, return branch_name.
Returns the branch name string, or empty string if no passport is found
or the file cannot be parsed.
"""
start = Path(cwd).resolve() if cwd else Path.cwd().resolve()
current = start
for _ in range(10):
passport = current / ".trinity" / "passport.json"
if passport.exists():
try:
data = json.loads(passport.read_text(encoding="utf-8"))
name = data.get("branch_info", {}).get("branch_name")
if not name:
name = data.get("identity", {}).get("name")
return name or ""
except Exception as exc:
logger.warning("[permissions] identify_caller: failed to parse passport at %s: %s", passport, exc)
return ""
parent = current.parent
if parent == current:
break
current = parent
return ""
def print_introspection() -> None:
"""Display permissions module info."""
from aipass.cli import console
console.print("[bold cyan]permissions[/bold cyan] — shared trust list for hook + drone layers")
console.print(f" TRUSTED_CROSS_WRITERS: {TRUSTED_CROSS_WRITERS}")
console.print(" Functions: is_trusted_caller(name), identify_caller(cwd)")
def handle_command(command: str, args: list) -> bool:
"""Library module — not a command handler. Returns False for all commands."""
if not args:
print_introspection()
return False
if args[0] in ("--help", "-h", "help"):
print_introspection()
return False
return False
+515
View File
@@ -0,0 +1,515 @@
"""Tests for the hook probe scripts and hooks module.
# =================== META ====================
# Name: test_hooks_probe.py
# Description: Tests for hook probe scripts and hooks seedgo module
# Version: 1.0.0
# Created: 2026-04-20
# Modified: 2026-04-20
# =============================================
"""
import importlib.util
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# Helpers — probe script loader
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
"""Walk up from this file to find the git repo root."""
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4] # fallback
_PROBES_DIR = _find_repo_root() / ".claude" / "hooks" / "probes"
_PROBE_SCRIPTS = [
"probe_pre_tool_use.py",
"probe_post_tool_use.py",
"probe_user_prompt_submit.py",
"probe_subagent_stop.py",
"probe_pre_compact.py",
"probe_stop.py",
"probe_notification.py",
]
_EVENT_NAMES = [
"PreToolUse",
"PostToolUse",
"UserPromptSubmit",
"SubagentStop",
"PreCompact",
"Stop",
"Notification",
]
def _load_probe(script_name: str):
"""Import a probe script by filename via importlib (outside package)."""
path = _PROBES_DIR / script_name
if not path.exists():
pytest.skip(f"Probe script not found: {path}")
spec = importlib.util.spec_from_file_location(path.stem, path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore[union-attr]
return module
# ---------------------------------------------------------------------------
# Fixtures — infrastructure mocks for hooks module
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_hooks_infrastructure(monkeypatch):
"""Mock aipass infrastructure imports for hooks module tests."""
mock_logger = MagicMock()
mock_console = MagicMock()
mock_warning = MagicMock()
mock_json_handler = MagicMock()
# -- prax ---------------------------------------------------------------
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
# -- cli ----------------------------------------------------------------
cli_mod = MagicMock()
cli_mod.console = mock_console
monkeypatch.setitem(sys.modules, "aipass.cli", cli_mod)
cli_apps = MagicMock()
monkeypatch.setitem(sys.modules, "aipass.cli.apps", cli_apps)
cli_modules = MagicMock()
cli_modules.warning = mock_warning
monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules)
# -- seedgo json handler ------------------------------------------------
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
# Force re-import of hooks module
monkeypatch.delitem(sys.modules, "aipass.seedgo.apps.modules.hooks", raising=False)
# ---------------------------------------------------------------------------
# Tests — probe scripts: happy-path stdin
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_happy_path(script_name, event_name, tmp_path, monkeypatch):
"""Each probe writes a correctly-shaped entry to last_ping.jsonl."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
payload = json.dumps(
{
"tool_name": "Read",
"cwd": "/tmp/test_cwd",
"session_id": "test-session-123",
}
)
monkeypatch.setattr("sys.stdin", io.StringIO(payload))
probe.main()
assert log_file.exists(), f"{script_name} did not create log file"
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
assert len(lines) == 1, f"Expected 1 entry, got {len(lines)}"
entry = json.loads(lines[0])
assert entry["event"] == event_name
assert entry["tool"] == "Read"
assert entry["cwd"] == "/tmp/test_cwd"
assert "timestamp" in entry
assert "script_elapsed_ms" in entry
assert isinstance(entry["script_elapsed_ms"], float)
assert "agent_id" in entry
assert "cli_version" in entry
assert "env_has_claude_project_dir" in entry
assert "env_has_aipass_home" in entry
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_appends_not_overwrites(script_name, event_name, tmp_path, monkeypatch):
"""Each probe appends to an existing log file instead of overwriting."""
import io
log_file = tmp_path / "last_ping.jsonl"
existing = json.dumps(
{
"event": "existing",
"tool": "",
"cwd": "/",
"agent_id": "x",
"timestamp": "2026-01-01T00:00:00Z",
"script_elapsed_ms": 0.1,
"cli_version": "0",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
}
)
log_file.write_text(existing + "\n", encoding="utf-8")
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
assert len(lines) == 2, "Probe should append, not overwrite"
assert json.loads(lines[0])["event"] == "existing"
assert json.loads(lines[1])["event"] == event_name
# ---------------------------------------------------------------------------
# Tests — probe scripts: malformed stdin
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_malformed_stdin_exits_cleanly(script_name, event_name, tmp_path, monkeypatch):
"""Each probe handles malformed stdin without raising an exception."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO("not json at all !!!"))
# main() returns normally on parse failure — no exception should propagate.
# sys.exit(0) is only called from the __main__ block, not from main() itself.
probe.main() # must not raise
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_empty_stdin_exits_cleanly(script_name, event_name, tmp_path, monkeypatch):
"""Each probe handles empty stdin without raising an exception."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO(""))
probe.main() # must not raise
# ---------------------------------------------------------------------------
# Tests — probe env var extraction
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_reads_session_id_from_env(script_name, event_name, tmp_path, monkeypatch):
"""Probe picks up CLAUDE_CODE_SESSION_ID from environment."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setenv("CLAUDE_CODE_SESSION_ID", "sess-env-test-999")
monkeypatch.delenv("CLAUDE_SESSION_ID", raising=False)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "sess-env-test-999"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_falls_back_to_claude_session_id(script_name, event_name, tmp_path, monkeypatch):
"""Probe falls back to CLAUDE_SESSION_ID when CLAUDE_CODE_SESSION_ID is absent."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
monkeypatch.setenv("CLAUDE_SESSION_ID", "fallback-session-42")
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "fallback-session-42"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_unknown_agent_id_when_no_env(script_name, event_name, tmp_path, monkeypatch):
"""Probe uses 'unknown' when no session env vars are set."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
monkeypatch.delenv("CLAUDE_SESSION_ID", raising=False)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "unknown"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_env_has_aipass_home(script_name, event_name, tmp_path, monkeypatch):
"""Probe records env_has_aipass_home correctly."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setenv("AIPASS_HOME", "/home/user/Projects/AIPass")
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["env_has_aipass_home"] is True
# ---------------------------------------------------------------------------
# Tests — hooks module: handle_command routing
# ---------------------------------------------------------------------------
def test_handle_command_wrong_command_returns_false():
"""handle_command returns False for unrecognised commands."""
from aipass.seedgo.apps.modules.hooks import handle_command
assert handle_command("wrong_command", []) is False
def test_handle_command_no_args_calls_introspection():
"""No args triggers introspection (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", [])
assert result is True
def test_handle_command_help_flag():
"""--help flag is handled without error."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["--help"])
assert result is True
def test_handle_command_probe_no_flags():
"""hooks probe with no extra flags calls display (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["probe"])
assert result is True
def test_handle_command_unknown_subcommand():
"""Unknown subcommand falls back to introspection (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["nonexistent_subcommand"])
assert result is True
# ---------------------------------------------------------------------------
# Tests — hooks module: probe display with mock data
# ---------------------------------------------------------------------------
def test_probe_display_renders_table(tmp_path):
"""_cmd_probe_display renders a Rich table without error given sample data."""
entries = [
{
"event": "PreToolUse",
"tool": "Bash",
"cwd": "/tmp/test",
"agent_id": "sess-abc",
"timestamp": "2026-04-20T12:00:00.000Z",
"script_elapsed_ms": 2.1,
"cli_version": "1.0.0",
"env_has_claude_project_dir": True,
"env_has_aipass_home": False,
},
{
"event": "PostToolUse",
"tool": "Read",
"cwd": "/tmp/test",
"agent_id": "sess-abc",
"timestamp": "2026-04-20T12:00:01.000Z",
"script_elapsed_ms": 1.5,
"cli_version": "1.0.0",
"env_has_claude_project_dir": True,
"env_has_aipass_home": True,
},
]
log_file = tmp_path / "last_ping.jsonl"
with open(log_file, "w", encoding="utf-8") as fh:
for e in entries:
fh.write(json.dumps(e) + "\n")
from aipass.seedgo.apps.modules.hooks import _cmd_probe_display
_cmd_probe_display(log_path=log_file)
def test_probe_display_empty_log(tmp_path):
"""_cmd_probe_display handles missing log file gracefully."""
from aipass.seedgo.apps.modules.hooks import _cmd_probe_display
missing = tmp_path / "no_such_file.jsonl"
_cmd_probe_display(log_path=missing)
def test_read_entries_skips_bad_lines(tmp_path):
"""_read_entries skips malformed lines and returns valid ones."""
log_file = tmp_path / "last_ping.jsonl"
good = json.dumps(
{
"event": "Stop",
"tool": "",
"cwd": "/",
"agent_id": "x",
"timestamp": "2026-04-20T00:00:00Z",
"script_elapsed_ms": 1.0,
"cli_version": "1",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
}
)
log_file.write_text(f"not json\n{good}\nalso not json\n", encoding="utf-8")
from aipass.seedgo.apps.modules.hooks import _read_entries
entries = _read_entries(log_path=log_file)
assert len(entries) == 1
assert entries[0]["event"] == "Stop"
def test_truncate_helper():
"""_truncate shortens long strings correctly."""
from aipass.seedgo.apps.modules.hooks import _truncate
assert _truncate("short", 20) == "short"
assert _truncate("a" * 30, 10) == "a" * 7 + "..."
assert len(_truncate("x" * 50, 15)) == 15
def test_entry_ts_parses_z_suffix():
"""_entry_ts handles ISO 8601 Z-suffix timestamps."""
from aipass.seedgo.apps.modules.hooks import _entry_ts
ts = _entry_ts({"timestamp": "2026-04-20T12:00:00.000000Z"})
assert ts > 0
def test_entry_ts_returns_zero_on_bad_input():
"""_entry_ts returns 0.0 for unparseable timestamps."""
from aipass.seedgo.apps.modules.hooks import _entry_ts
assert _entry_ts({"timestamp": "not-a-date"}) == 0.0
assert _entry_ts({}) == 0.0
# ---------------------------------------------------------------------------
# Tests — matrix builder
# ---------------------------------------------------------------------------
def test_build_matrix_rows_groups_correctly():
"""_build_matrix_rows groups entries by event and counts correctly."""
entries = [
{
"event": "Stop",
"env_has_claude_project_dir": True,
"env_has_aipass_home": False,
"agent_id": "a",
},
{
"event": "Stop",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
"agent_id": "b",
},
{
"event": "PreToolUse",
"env_has_claude_project_dir": True,
"env_has_aipass_home": True,
"agent_id": "a",
},
]
from aipass.seedgo.apps.modules.hooks import _build_matrix_rows
rows, groups = _build_matrix_rows(entries)
assert len(rows) == 2
stop_row = next(r for r in rows if r["event"] == "Stop")
assert stop_row["count"] == 2
assert stop_row["project_dir_true"] == 1
assert stop_row["project_dir_false"] == 1
assert stop_row["unique_agents"] == 2
def test_probe_matrix_writes_report(tmp_path):
"""_cmd_probe_matrix creates the markdown report given sample data."""
from aipass.seedgo.apps.modules.hooks import _write_matrix_report
entries = [
{
"event": "Stop",
"tool": "",
"cwd": "/tmp",
"agent_id": "sess-1",
"timestamp": "2026-04-20T10:00:00Z",
"script_elapsed_ms": 1.2,
"cli_version": "1.0",
"env_has_claude_project_dir": False,
"env_has_aipass_home": True,
}
]
matrix_rows = [
{
"event": "Stop",
"count": 1,
"project_dir_true": 0,
"project_dir_false": 1,
"aipass_home_true": 1,
"aipass_home_false": 0,
"unique_agents": 1,
}
]
report_path = tmp_path / "Q12_findings_2026-04-20.md"
_write_matrix_report(report_path, matrix_rows, entries)
assert report_path.exists()
content = report_path.read_text(encoding="utf-8")
assert "Stop" in content
assert "Q12 Findings" in content
@@ -0,0 +1,402 @@
# =================== AIPass ====================
# Name: test_hooks_track_e.py
# Description: DPLAN-0139 Track E — single-path enforcement tests
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""Tests for DPLAN-0139 Track E — single-path enforcement.
Covers:
- permissions.py: TRUSTED_CROSS_WRITERS, is_trusted_caller(), identify_caller()
- pre_edit_gate.py v1.3.0: inbox lock rule + cross-branch write rule
- drone auth.py: ALLOWED_CALLERS derived from TRUSTED_CROSS_WRITERS
- inbox_audit.py: handle_command routing + _scan_inbox validation
- delivery.py: deliver_to_inbox_file single-path helper
"""
import importlib.util
import io
import json
from pathlib import Path
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
"""Walk up from this file to find the git repo root."""
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4]
REPO_ROOT = _find_repo_root()
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
def _load_hook(name: str):
"""Import a hook script by filename via importlib (outside package)."""
path = HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
# ---------------------------------------------------------------------------
# permissions.py
# ---------------------------------------------------------------------------
def test_trusted_cross_writers_contains_expected_members():
"""TRUSTED_CROSS_WRITERS must include devpulse, seedgo, and spawn."""
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
assert "devpulse" in TRUSTED_CROSS_WRITERS
assert "seedgo" in TRUSTED_CROSS_WRITERS
assert "spawn" in TRUSTED_CROSS_WRITERS
def test_is_trusted_caller_returns_true_for_devpulse():
"""devpulse is a trusted cross-writer."""
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
assert is_trusted_caller("devpulse") is True
def test_is_trusted_caller_returns_true_for_seedgo():
"""seedgo is a trusted cross-writer."""
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
assert is_trusted_caller("seedgo") is True
def test_is_trusted_caller_returns_true_for_spawn():
"""spawn is a trusted cross-writer."""
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
assert is_trusted_caller("spawn") is True
def test_is_trusted_caller_returns_false_for_unknown():
"""Regular branches are not trusted cross-writers."""
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
assert is_trusted_caller("flow") is False
assert is_trusted_caller("memory") is False
assert is_trusted_caller("random_branch") is False
def test_identify_caller_returns_empty_when_no_passport(tmp_path):
"""identify_caller returns empty string when no passport.json is found."""
from aipass.seedgo.apps.modules.permissions import identify_caller
result = identify_caller(str(tmp_path))
assert result == ""
def test_identify_caller_reads_branch_name_from_passport(tmp_path):
"""identify_caller reads branch_name from branch_info section."""
from aipass.seedgo.apps.modules.permissions import identify_caller
trinity = tmp_path / ".trinity"
trinity.mkdir()
passport = trinity / "passport.json"
passport.write_text(json.dumps({"branch_info": {"branch_name": "testbranch"}}), encoding="utf-8")
result = identify_caller(str(tmp_path))
assert result == "testbranch"
def test_identify_caller_falls_back_to_identity_name(tmp_path):
"""identify_caller falls back to identity.name when branch_info absent."""
from aipass.seedgo.apps.modules.permissions import identify_caller
trinity = tmp_path / ".trinity"
trinity.mkdir()
passport = trinity / "passport.json"
passport.write_text(json.dumps({"identity": {"name": "fallback_branch"}}), encoding="utf-8")
result = identify_caller(str(tmp_path))
assert result == "fallback_branch"
# ---------------------------------------------------------------------------
# pre_edit_gate.py v1.3.0 — Track E rules
# ---------------------------------------------------------------------------
def test_gate_allows_non_edit_tool():
"""Non-edit tools (Read) must pass through without blocking."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": "/tmp/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
def test_gate_blocks_inbox_json_write(capsys):
"""Any write targeting .ai_mail.local/inbox.json must be blocked."""
mod = _load_hook("pre_edit_gate.py")
inbox_path = "/home/user/Projects/AIPass/src/aipass/flow/.ai_mail.local/inbox.json"
payload = json.dumps({"tool_name": "Write", "tool_input": {"file_path": inbox_path}})
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
result = json.loads(captured.out)
assert result["decision"] == "block"
assert "inbox.json" in result["reason"].lower() or "drone" in result["reason"].lower()
def test_gate_blocks_cross_branch_write_from_untrusted(capsys):
"""Untrusted branch writing to a different branch must be blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/memory",
}
)
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
result = json.loads(captured.out)
assert result["decision"] == "block"
def test_gate_allows_cross_branch_write_from_devpulse(capsys):
"""devpulse may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/devpulse",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_cross_branch_write_from_seedgo(capsys):
"""seedgo may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/seedgo",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_cross_branch_write_from_spawn(capsys):
"""spawn may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Write",
"tool_input": {"file_path": "/repo/src/aipass/prax/apps/modules/bar.py"},
"cwd": "/repo/src/aipass/spawn",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_same_branch_write(capsys):
"""Writes within the same branch must not be blocked by the cross-branch rule."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/flow",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
# ---------------------------------------------------------------------------
# drone auth.py — ALLOWED_CALLERS derived from permissions
# ---------------------------------------------------------------------------
def test_drone_auth_allowed_callers_matches_permissions():
"""Hook and drone must reach the same decision for the same caller."""
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
for branch in TRUSTED_CROSS_WRITERS:
assert branch in ALLOWED_CALLERS, f"'{branch}' in TRUSTED_CROSS_WRITERS but missing from drone ALLOWED_CALLERS"
def test_drone_auth_allowed_callers_includes_devpulse():
"""devpulse must remain in drone ALLOWED_CALLERS."""
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
assert "devpulse" in ALLOWED_CALLERS
def test_drone_auth_allowed_callers_includes_seedgo():
"""seedgo must be in drone ALLOWED_CALLERS."""
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
assert "seedgo" in ALLOWED_CALLERS
def test_drone_auth_allowed_callers_includes_spawn():
"""spawn must be in drone ALLOWED_CALLERS."""
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
assert "spawn" in ALLOWED_CALLERS
# ---------------------------------------------------------------------------
# inbox_audit.py — handle_command routing + _scan_inbox
# ---------------------------------------------------------------------------
def test_inbox_audit_ignores_non_audit_command():
"""handle_command returns False for non-audit command names."""
from aipass.seedgo.apps.modules.inbox_audit import handle_command
assert handle_command("standards_query", ["inbox-ids"]) is False
assert handle_command("checklist", ["inbox-ids"]) is False
def test_inbox_audit_handles_inbox_ids_subcommand():
"""handle_command returns True and runs scan for `audit inbox-ids`."""
from aipass.seedgo.apps.modules.inbox_audit import handle_command
with patch("aipass.seedgo.apps.modules.inbox_audit._run_inbox_id_scan", return_value=0):
result = handle_command("audit", ["inbox-ids"])
assert result is True
def test_inbox_audit_ignores_other_audit_subcommands():
"""handle_command returns False for audit subcommands other than inbox-ids."""
from aipass.seedgo.apps.modules.inbox_audit import handle_command
assert handle_command("audit", ["aipass"]) is False
assert handle_command("audit", ["flow"]) is False
def test_inbox_audit_scan_detects_bad_id(tmp_path):
"""_scan_inbox flags message ids that are not 8-char lowercase hex."""
from aipass.seedgo.apps.modules.inbox_audit import _scan_inbox
inbox = tmp_path / "inbox.json"
inbox.write_text(
json.dumps(
{
"messages": [
{"id": "not-hex!", "subject": "bad", "from": "@test", "status": "new"},
{"id": "a1b2c3d4", "subject": "ok", "from": "@test", "status": "new"},
]
}
),
encoding="utf-8",
)
violations = _scan_inbox(inbox)
assert len(violations) == 1
assert violations[0]["id"] == "not-hex!"
def test_inbox_audit_scan_passes_valid_ids(tmp_path):
"""_scan_inbox returns empty list when all message ids are valid 8-hex."""
from aipass.seedgo.apps.modules.inbox_audit import _scan_inbox
inbox = tmp_path / "inbox.json"
inbox.write_text(
json.dumps(
{
"messages": [
{"id": "a1b2c3d4", "subject": "ok1", "from": "@x", "status": "new"},
{"id": "deadbeef", "subject": "ok2", "from": "@y", "status": "new"},
]
}
),
encoding="utf-8",
)
violations = _scan_inbox(inbox)
assert violations == []
# ---------------------------------------------------------------------------
# delivery.py — deliver_to_inbox_file single-path helper
# Load by file path to avoid cross-branch package import restriction.
# ---------------------------------------------------------------------------
def _load_delivery():
"""Load ai_mail delivery.py by file path (bypasses cross-branch import check)."""
delivery_path = REPO_ROOT / "src" / "aipass" / "ai_mail" / "apps" / "handlers" / "email" / "delivery.py"
if not delivery_path.exists():
pytest.skip(f"delivery.py not found: {delivery_path}")
spec = importlib.util.spec_from_file_location("delivery", delivery_path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
def test_deliver_to_inbox_file_returns_false_for_missing_inbox(tmp_path):
"""deliver_to_inbox_file returns (False, error, '') when inbox does not exist."""
delivery = _load_delivery()
missing = tmp_path / "inbox.json"
success, error_msg, reply_id = delivery.deliver_to_inbox_file(
missing,
{"from": "@x", "to": "@y", "subject": "s", "message": "m", "timestamp": "t"},
)
assert success is False
assert reply_id == ""
def test_deliver_to_inbox_file_writes_message_and_returns_id(tmp_path):
"""deliver_to_inbox_file writes to inbox and returns the assigned 8-char id."""
delivery = _load_delivery()
inbox = tmp_path / "inbox.json"
inbox.write_text(
json.dumps({"mailbox": "inbox", "total_messages": 0, "unread_count": 0, "messages": []}),
encoding="utf-8",
)
email_data = {
"from": "@sender",
"to": "@recv",
"subject": "Hello",
"message": "body",
"timestamp": "2026-04-21 00:00:00",
}
with patch.object(delivery, "_send_desktop_notification"):
success, error_msg, reply_id = delivery.deliver_to_inbox_file(inbox, email_data)
assert success is True
assert len(reply_id) == 8
data = json.loads(inbox.read_text())
assert len(data["messages"]) == 1
assert data["messages"][0]["id"] == reply_id