merge main into #373 — take Track B version of hooks.py (superset of Track A)
This commit is contained in:
AIOSAI
@@ -12,10 +12,9 @@ Key behaviors:
|
||||
- Saves ruff lint AND pyright errors to state file for PreToolUse gate (hard block)
|
||||
- Surfaces ALL errors in additionalContext so Claude sees them
|
||||
|
||||
Version: 5.3.0
|
||||
Version: 5.2.0
|
||||
|
||||
CHANGELOG:
|
||||
- v5.3.0 (2026-04-20): [SILENT-FIX] label + IDE fallback with loud announce.
|
||||
- v5.2.0 (2026-04-20): Save ruff lint errors to state file for hard-block enforcement.
|
||||
Pre-edit gate now blocks on F401/lint just like type errors.
|
||||
- v5.1.0 (2026-04-19): Added ruff format --check to surface format drift.
|
||||
@@ -177,15 +176,11 @@ def run_ruff_lint_structured(file_path: str) -> list[dict]:
|
||||
return []
|
||||
|
||||
|
||||
def run_pyright_check(file_path: str) -> tuple[list[dict], bool]:
|
||||
"""Run pyright on a single file. Returns (errors, fallback_needed).
|
||||
|
||||
fallback_needed is True when pyright is unavailable (FileNotFoundError).
|
||||
Timeout and generic exceptions return ([], False) — silent.
|
||||
"""
|
||||
def run_pyright_check(file_path: str) -> list[dict]:
|
||||
"""Run pyright on a single file. Returns list of error dicts."""
|
||||
# Skip hook files - they don't follow project standards
|
||||
if '/.claude/hooks/' in file_path:
|
||||
return [], False
|
||||
return []
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
@@ -198,7 +193,7 @@ def run_pyright_check(file_path: str) -> tuple[list[dict], bool]:
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return [], False
|
||||
return []
|
||||
|
||||
errors = []
|
||||
for diag in data.get("generalDiagnostics", []):
|
||||
@@ -211,14 +206,14 @@ def run_pyright_check(file_path: str) -> tuple[list[dict], bool]:
|
||||
"message": message[:100]
|
||||
})
|
||||
|
||||
return errors[:10], False # Max 10 errors
|
||||
return errors[:10] # Max 10 errors
|
||||
|
||||
except FileNotFoundError:
|
||||
return [], True # pyright not installed — signal fallback needed
|
||||
return [] # pyright not installed
|
||||
except subprocess.TimeoutExpired:
|
||||
return [], False # Timeout — don't block
|
||||
return [] # Timeout — don't block
|
||||
except Exception:
|
||||
return [], False
|
||||
return []
|
||||
|
||||
|
||||
def save_diagnostics_state(file_path: str, errors: list[dict]):
|
||||
@@ -343,7 +338,6 @@ def main():
|
||||
# Collect all errors
|
||||
errors = []
|
||||
file_type = ""
|
||||
fallback_needed = False
|
||||
|
||||
if file_path.endswith(".py"):
|
||||
file_type = "Python"
|
||||
@@ -355,9 +349,7 @@ def main():
|
||||
errors.append(f"SEEDGO: {v}")
|
||||
|
||||
# Pyright type errors (single file)
|
||||
type_errors, fallback_needed = run_pyright_check(file_path)
|
||||
if fallback_needed:
|
||||
print("[FALLBACK ENGAGED: IDE diagnostics — VS Code only]", file=sys.stderr, flush=True)
|
||||
type_errors = run_pyright_check(file_path)
|
||||
for te in type_errors:
|
||||
errors.append(f"TYPE: L{te['line']}: {te['message']}")
|
||||
|
||||
@@ -374,16 +366,17 @@ def main():
|
||||
# Build output
|
||||
if errors:
|
||||
error_text = "\n".join(f" - {e}" for e in errors)
|
||||
context = f"[SILENT-FIX] {len(errors)} error(s) in {Path(file_path).name}:\n{error_text}\n\nFix these silently — do not announce or explain, just fix and continue."
|
||||
if fallback_needed:
|
||||
context += "\n\n[FALLBACK] pyright unavailable — run mcp__ide__getDiagnostics to check type errors. VS Code only. Non-VS-Code users: this check is a no-op."
|
||||
context = f"""[AUTO-FIX] {len(errors)} error(s) in {Path(file_path).name}:
|
||||
{error_text}
|
||||
|
||||
Fix these errors in {Path(file_path).name} now. Do not skip or defer."""
|
||||
|
||||
output = {
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": "PostToolUse",
|
||||
"additionalContext": context
|
||||
},
|
||||
"systemMessage": f"[SILENT-FIX] {len(errors)} error(s) — fix before continuing"
|
||||
"systemMessage": f"[AUTO-FIX] {len(errors)} error(s) — fix before continuing"
|
||||
}
|
||||
print(json.dumps(output))
|
||||
else:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -224,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": {
|
||||
|
||||
@@ -1,20 +1,24 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: ruff_check.py
|
||||
# Description: Ruff Linter Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Version: 1.1.0
|
||||
# Created: 2026-04-16
|
||||
# Modified: 2026-04-16
|
||||
# Modified: 2026-04-20
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Ruff Linter Standards Checker Handler
|
||||
|
||||
Runs ruff against a branch's apps/ directory and scores based on violation
|
||||
count. Prevents ruff debt from silently re-accumulating after a cleanup.
|
||||
Two modes:
|
||||
- check_branch(): runs ruff across entire apps/ tree (used by audit pipeline,
|
||||
AUDIT_SCOPE = branch_level, ADVISORY = always-passes)
|
||||
- check_module(): runs ruff on a single file (used by checklist/per-file hooks,
|
||||
returns passed=False on violations so subagent_stop_gate can block)
|
||||
|
||||
AUDIT_SCOPE: branch_level — runs once per branch, ruff walks the tree.
|
||||
ADVISORY: surfaces violations and score but always passes overall.
|
||||
Promote to required once all branches are clean.
|
||||
AUDIT_SCOPE: branch_level — audit pipeline uses check_branch() once per branch.
|
||||
Checkers that also implement check_module() are eligible for per-file checklist runs.
|
||||
ADVISORY: check_branch() surfaces violations but always passes (advisory score).
|
||||
check_module() returns passed=False so checklist/hooks can block.
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -95,6 +99,123 @@ def _score_from_count(count: int) -> int:
|
||||
return 25
|
||||
|
||||
|
||||
def _find_ruff_bypass_from_file(file_path: str) -> list:
|
||||
"""Walk up from file_path to find .seedgo/ruff_bypass.json at the branch root."""
|
||||
fp = Path(file_path).resolve()
|
||||
for parent in list(fp.parents):
|
||||
candidate = parent / ".seedgo" / "ruff_bypass.json"
|
||||
if candidate.exists():
|
||||
try:
|
||||
data = json.loads(candidate.read_text(encoding="utf-8"))
|
||||
return data if isinstance(data, list) else []
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to load ruff_bypass.json at %s: %s", candidate, exc)
|
||||
return []
|
||||
if (parent / ".git").exists():
|
||||
break
|
||||
return []
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""Run ruff check on a single file.
|
||||
|
||||
Used by checklist mode and subagent_stop_gate for per-file enforcement.
|
||||
Returns passed=False when violations exist so hooks can block.
|
||||
|
||||
Args:
|
||||
module_path: Absolute path to the Python file to check.
|
||||
bypass_rules: Standard bypass rules from .seedgo/bypass.json
|
||||
|
||||
Returns:
|
||||
dict with passed, checks, score, standard keys.
|
||||
"""
|
||||
fp = Path(module_path)
|
||||
|
||||
if is_bypassed(module_path, "ruff_check", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [{"name": "Ruff check", "passed": True, "message": "Standard bypassed via .seedgo/bypass.json"}],
|
||||
"score": 100,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
|
||||
if shutil.which("ruff") is None:
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [{"name": "Ruff check", "passed": True, "message": "ruff not installed — check skipped"}],
|
||||
"score": 100,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
|
||||
ruff_bypass = _find_ruff_bypass_from_file(module_path)
|
||||
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
["ruff", "check", str(fp), "--output-format=json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("ruff check_module timed out on %s", module_path)
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [{"name": "Ruff check", "passed": True, "message": "ruff check timed out — skipped"}],
|
||||
"score": 100,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.warning("ruff check_module failed on %s: %s", module_path, exc)
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [{"name": "Ruff check", "passed": True, "message": f"ruff error — skipped: {exc}"}],
|
||||
"score": 100,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
|
||||
violations: list = []
|
||||
if proc.stdout.strip():
|
||||
try:
|
||||
violations = json.loads(proc.stdout)
|
||||
if not isinstance(violations, list):
|
||||
violations = []
|
||||
except (json.JSONDecodeError, ValueError) as exc:
|
||||
logger.warning("ruff JSON parse failed for %s: %s", module_path, exc)
|
||||
violations = []
|
||||
|
||||
active = [v for v in violations if not _is_ruff_bypassed(v, ruff_bypass)]
|
||||
count = len(active)
|
||||
|
||||
if count == 0:
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": module_path, "score": 100, "standard": "ruff_check"},
|
||||
)
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [{"name": "Ruff check", "passed": True, "message": "No ruff violations found"}],
|
||||
"score": 100,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
|
||||
top = active[:5]
|
||||
msgs = [f"{v.get('code', '?')} L{v.get('location', {}).get('row', '?')}: {v.get('message', '?')[:80]}" for v in top]
|
||||
suffix = f" (and {count - 5} more)" if count > 5 else ""
|
||||
detail = f"{count} violation(s) — " + "; ".join(msgs) + suffix
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": module_path, "score": 0, "standard": "ruff_check", "violations": count},
|
||||
)
|
||||
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [{"name": "Ruff check", "passed": False, "message": detail}],
|
||||
"score": 0,
|
||||
"standard": "RUFF_CHECK",
|
||||
}
|
||||
|
||||
|
||||
def check_branch(branch_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""Run ruff against the branch and score based on violation count.
|
||||
|
||||
|
||||
@@ -86,13 +86,15 @@ def _is_applicable(checker, file_path: str) -> bool:
|
||||
Rules based on AUDIT_SCOPE:
|
||||
- "entry_point" (default) -> only apps/{name}.py files
|
||||
- "all_files" -> any .py file
|
||||
- "branch_level" -> not applicable to single-file checks
|
||||
- "branch_level" -> normally skipped, but eligible if checker
|
||||
also implements check_module() for per-file use
|
||||
"""
|
||||
scope = getattr(checker, "AUDIT_SCOPE", "entry_point")
|
||||
|
||||
# Branch-level checkers need a branch path, not a single file
|
||||
# Branch-level checkers skip per-file runs UNLESS they also implement
|
||||
# check_module() for targeted single-file validation (e.g., ruff_check)
|
||||
if scope == "branch_level":
|
||||
return False
|
||||
return hasattr(checker, "check_module") and file_path.endswith(".py")
|
||||
|
||||
# Only check_module() capable checkers
|
||||
if not hasattr(checker, "check_module"):
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user