feat(system): feat(system): fix(trigger): dead events + zombie handler + stale registry + per-branch disable + email check (DPLAN-0159)

Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-28 17:46:06 -07:00
co-authored by @devpulse
parent 477e3a58ef
commit 78b7fb6ac7
39 changed files with 354 additions and 618 deletions
@@ -201,28 +201,45 @@ def is_kill_switch_active(config: Dict[str, Any]) -> bool:
def _write_pid_file() -> bool:
"""Write current PID to daemon.pid. Returns False if another daemon is running."""
if DAEMON_PID_FILE.exists():
try:
old_pid = int(DAEMON_PID_FILE.read_text().strip())
try:
os.kill(old_pid, 0)
# Process exists — another daemon is running
logger.info(f"Another daemon already running (PID {old_pid}). Exiting.")
return False
except ProcessLookupError:
# Stale PID file — process is dead, we can take over
logger.info(f"Removing stale PID file (PID {old_pid} is dead)")
except PermissionError:
# Process exists but we can't signal it
logger.info(f"Another daemon already running (PID {old_pid}, permission denied). Exiting.")
return False
except (ValueError, OSError):
logger.info("Corrupt PID file — removing")
"""Write current PID to daemon.pid atomically. Returns False if another daemon is running."""
DAEMON_PID_FILE.parent.mkdir(parents=True, exist_ok=True)
DAEMON_PID_FILE.write_text(str(os.getpid()))
return True
try:
fd = os.open(str(DAEMON_PID_FILE), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
try:
os.write(fd, str(os.getpid()).encode("utf-8"))
finally:
os.close(fd)
return True
except FileExistsError:
logger.info("[daemon] PID file already exists, checking owner")
# PID file exists — check if the owning process is alive
try:
old_pid = int(DAEMON_PID_FILE.read_text().strip())
try:
os.kill(old_pid, 0)
logger.info(f"Another daemon already running (PID {old_pid}). Exiting.")
return False
except ProcessLookupError:
logger.info(f"Removing stale PID file (PID {old_pid} is dead)")
except PermissionError:
logger.info(f"Another daemon already running (PID {old_pid}, permission denied). Exiting.")
return False
except (ValueError, OSError):
logger.info("Corrupt PID file — removing")
# Stale or corrupt — remove and retry atomically
DAEMON_PID_FILE.unlink(missing_ok=True)
try:
fd = os.open(str(DAEMON_PID_FILE), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
try:
os.write(fd, str(os.getpid()).encode("utf-8"))
finally:
os.close(fd)
return True
except FileExistsError:
logger.info("Another daemon raced us for the PID file. Exiting.")
return False
def _remove_pid_file() -> None:
@@ -245,6 +262,17 @@ def get_registered_branches() -> list:
return data.get("branches", [])
def _is_registered_sender(sender: str) -> bool:
"""Check if sender email exists in the branch registry (DPLAN-0159 S2)."""
registry = _read_json(BRANCH_REGISTRY)
if registry is None:
return True # fail open if registry unreadable
for branch in registry.get("branches", []):
if branch.get("email") == sender:
return True
return False
def check_inbox_for_dispatch(branch_path: Path) -> Optional[Dict[str, Any]]:
"""
Check a branch's inbox for unprocessed --dispatch emails.
@@ -311,6 +339,10 @@ def spawn_agent(
subject = message.get("subject", "")
max_turns = config.get("max_turns_per_wake", 100)
if message.get("auto_execute") and not _is_registered_sender(sender):
logger.warning("[daemon] Dispatch from unregistered sender %s — rejecting", sender)
return False
lock_file_path = str(branch_path / ".ai_mail.local" / ".dispatch.lock")
# Prompt — only interpolate system-generated metadata (id, sender email).
@@ -164,6 +164,30 @@ def send_reply(from_branch_path: Path, original_email: Dict, reply_message: str)
return True, f"Reply sent to {reply_destination}, original closed", reply_id
def _validate_reply_path(reply_path: str) -> Tuple[bool, str]:
"""Validate that reply_path points to a legitimate inbox.json.
Checks: (a) path resolves, (b) ends with .ai_mail.local/inbox.json,
(c) an AIPASS_REGISTRY.json exists in an ancestor directory.
"""
try:
path = Path(reply_path).resolve()
except (OSError, ValueError) as e:
logger.warning("[reply] _validate_reply_path resolution failed: %s", e)
return False, f"Path resolution failed: {e}"
if path.name != "inbox.json" or path.parent.name != ".ai_mail.local":
return False, f"Path does not end with .ai_mail.local/inbox.json: {path}"
for parent in path.parents:
if (parent / "AIPASS_REGISTRY.json").exists():
return True, ""
if parent == parent.parent:
break
return False, f"No AIPASS_REGISTRY.json found in ancestors of {path}"
def _deliver_via_reply_path(
reply_path: str,
reply_email_data: Dict,
@@ -185,7 +209,12 @@ def _deliver_via_reply_path(
Returns:
Tuple of (success, message, reply_id or None)
"""
inbox_file = Path(reply_path)
valid, reason = _validate_reply_path(reply_path)
if not valid:
logger.warning("[reply] reply_path rejected: %s", reason)
return False, f"Invalid reply_path: {reason}", None
inbox_file = Path(reply_path).resolve()
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)
+168
View File
@@ -773,6 +773,7 @@ from aipass.ai_mail.apps.handlers.dispatch.daemon import (
_handle_signal,
_check_lock,
_acquire_lock,
_is_registered_sender,
poll_cycle,
_write_pid_file,
_remove_pid_file,
@@ -1773,3 +1774,170 @@ def test_poll_cycle_spawn_failure_not_counted(tmp_path, monkeypatch):
result = poll_cycle(config, state)
assert result == 0
# ---- _is_registered_sender tests (DPLAN-0159 S2) ----------------
def test_is_registered_sender_found(tmp_path, monkeypatch):
"""Registered sender returns True."""
registry = {"branches": [{"email": "@flow"}, {"email": "@backup"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
assert _is_registered_sender("@flow") is True
def test_is_registered_sender_not_found(tmp_path, monkeypatch):
"""Unregistered sender returns False."""
registry = {"branches": [{"email": "@flow"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
assert _is_registered_sender("@evil") is False
def test_is_registered_sender_missing_registry(tmp_path, monkeypatch):
"""Missing registry fails open (returns True)."""
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", tmp_path / "missing.json")
assert _is_registered_sender("@anyone") is True
def test_is_registered_sender_empty_branches(tmp_path, monkeypatch):
"""Empty branches list returns False for any sender."""
registry = {"branches": []}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
assert _is_registered_sender("@flow") is False
# ---- spawn_agent sender auth tests (DPLAN-0159 S2) ----------------
def test_spawn_agent_rejects_unregistered_auto_execute(tmp_path, monkeypatch):
"""Auto-execute dispatch from unregistered sender is rejected."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
registry = {"branches": [{"email": "@flow"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
message = {"from": "@forged", "id": "msg1", "subject": "Fake", "auto_execute": True}
config = {"max_turns_per_wake": 50}
state = {"daily_counts": {}, "session_cycles": {}}
result = spawn_agent(branch_path, "@testbranch", message, config, state)
assert result is False
def test_spawn_agent_allows_registered_auto_execute(tmp_path, monkeypatch):
"""Auto-execute dispatch from registered sender proceeds to spawn."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
(branch_path / "logs").mkdir()
registry = {"branches": [{"email": "@devpulse"}, {"email": "@flow"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
message = {"from": "@devpulse", "id": "msg1", "subject": "Task", "auto_execute": True}
config = {"max_turns_per_wake": 50}
state = {"daily_counts": {}, "session_cycles": {}}
mock_process = MagicMock()
mock_process.pid = 54321
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen",
return_value=mock_process,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock",
return_value=(True, "Lock acquired"),
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification",
create=True,
),
):
result = spawn_agent(branch_path, "@testbranch", message, config, state)
assert result is True
def test_spawn_agent_no_auth_check_without_auto_execute(tmp_path, monkeypatch):
"""Non-auto_execute email skips sender auth check."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
(branch_path / "logs").mkdir()
registry = {"branches": []}
reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json.dumps(registry), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "BRANCH_REGISTRY", reg_file)
message = {"from": "@unknown", "id": "msg1", "subject": "Manual"}
config = {"max_turns_per_wake": 50}
state = {"daily_counts": {}, "session_cycles": {}}
mock_process = MagicMock()
mock_process.pid = 54321
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen",
return_value=mock_process,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock",
return_value=(True, "Lock acquired"),
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification",
create=True,
),
):
result = spawn_agent(branch_path, "@testbranch", message, config, state)
assert result is True
# ---- _write_pid_file atomic tests (DPLAN-0159 S5) ----------------
def test_write_pid_file_atomic_no_existing(tmp_path, monkeypatch):
"""Atomic creation succeeds when no PID file exists."""
pid_file = tmp_path / "daemon.pid"
monkeypatch.setattr(daemon_mod, "DAEMON_PID_FILE", pid_file)
result = _write_pid_file()
assert result is True
assert pid_file.exists()
assert int(pid_file.read_text().strip()) == os.getpid()
def test_write_pid_file_atomic_race_second_loses(tmp_path, monkeypatch):
"""Second daemon loses the race when both try O_CREAT|O_EXCL."""
pid_file = tmp_path / "daemon.pid"
monkeypatch.setattr(daemon_mod, "DAEMON_PID_FILE", pid_file)
# First daemon wins
pid_file.write_text(str(os.getpid()), encoding="utf-8")
# Second daemon: file exists, owner is alive → returns False
result = _write_pid_file()
assert result is False
@@ -130,8 +130,20 @@ class MonitoringFileHandler(FileSystemEventHandler):
return f"⚡ Bash: {desc[:120]}"
if tool_name in ("Grep", "Glob"):
return f"🔍 {tool_name}: {inp.get('pattern', '')[:80]}"
if tool_name == "Task":
if tool_name in ("Task", "Agent"):
return f"🚀 Agent: {inp.get('description', '')[:80]}"
if tool_name == "WebFetch":
return f"🌐 WebFetch: {inp.get('url', '')[:80]}"
if tool_name == "WebSearch":
return f"🔍 WebSearch: {inp.get('query', '')[:80]}"
if tool_name == "Monitor":
return f"⚡ Monitor: {inp.get('command', '')[:120]}"
if tool_name == "Skill":
return f"🎯 Skill: {inp.get('skill', '')[:80]}"
if tool_name == "NotebookEdit":
fp = inp.get("notebook_path", "")
short = fp.split("/")[-1] if "/" in fp else fp
return f"🔧 NotebookEdit: {short}"
return f"🔧 {tool_name}"
@staticmethod
+5
View File
@@ -81,6 +81,11 @@
],
"reason": "Same-branch cross-handler import (allowed per architecture standard). Diagnostics handler imports bypass/ignore_handler for audit ignore patterns."
},
{
"file": "apps/handlers/aipass_standards/",
"standard": "handlers",
"reason": "All 33 checkers import is_bypassed from bypass/utils.py — shared utility extracted from 33 duplicate copies (DPLAN-0159 A2). Same-branch cross-handler import, intentional."
},
{
"file": "apps/handlers/aipass_standards/cli_check.py",
"standard": "debug_print",
@@ -21,6 +21,7 @@ from typing import Dict, List, Optional
from aipass.seedgo.apps.handlers.bypass.ignore_handler import get_template_ignore_patterns
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
@@ -33,26 +34,6 @@ _SRC_PKG_ROOT = PACK_ROOT.parent.parent # apps/ -> seedgo/ -> src/aipass/
SPAWN_TEMPLATES_DIR = _SRC_PKG_ROOT / "spawn" / "templates"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows architecture standards
@@ -19,31 +19,12 @@ from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows CLI standards
@@ -27,28 +27,12 @@ from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: entry points only (apps/{name}.py)
AUDIT_SCOPE = "entry_point"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows CLI flags standards
@@ -28,6 +28,7 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: scan every .py file, not just entry point
AUDIT_SCOPE = "all_files"
@@ -36,23 +37,6 @@ AUDIT_SCOPE = "all_files"
_COMMENTED_LOGGER_RE = re.compile(r"#\s*logger\.(error|warning|warn|info|exception|critical|debug)\s*\(")
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check a Python file for commented-out logger calls.
@@ -23,6 +23,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "branch_level"
@@ -58,23 +59,6 @@ _SKIP_DIRS = {
# =============================================
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# =============================================
# FILE COLLECTION
# =============================================
@@ -20,6 +20,7 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -34,26 +35,6 @@ _DOCTEST_RE = re.compile(r"^\s*(\.\.\.|>>>)\s")
_TEST_FILE_RE = re.compile(r"^(test_.+|.+_test|conftest)\.py$")
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def _is_in_main_block(lines: list[str], lineno: int) -> bool:
"""
Return True if the line at *lineno* (1-based) is inside an
@@ -22,6 +22,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -35,23 +36,6 @@ DEPTH_LIMIT = 4
# -- Bypass helper -----------------------------------------------------------
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# -- AST depth analysis ------------------------------------------------------
@@ -19,28 +19,12 @@ from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows documentation standards.
@@ -23,6 +23,7 @@ from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
@@ -38,26 +39,6 @@ def _find_registry() -> Path:
return Path.cwd() / "AIPASS_REGISTRY.json"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def get_branch_from_path(file_path: str) -> Optional[Dict]:
"""Detect which branch a file belongs to using AIPASS_REGISTRY.json."""
try:
@@ -17,28 +17,12 @@ from pathlib import Path
from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""Check if module follows error handling standards"""
checks = []
@@ -20,6 +20,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "branch_level"
@@ -29,28 +30,6 @@ AUDIT_SCOPE = "branch_level"
# =============================================
def is_bypassed(
file_path: str,
standard: str,
line: int | None = None,
bypass_rules: list | None = None,
) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# =============================================
# BRANCH-LEVEL CHECK (audit pipeline entry)
# =============================================
@@ -19,31 +19,12 @@ from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if handler follows handler standards
@@ -20,6 +20,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -98,23 +99,6 @@ _PAT_REGEX_CONTEXT = re.compile(r"""re\.compile|r["']|\\[dws\^]""")
# -- Helpers ----------------------------------------------------------------
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def _is_placeholder(key_value: str) -> bool:
"""Return True if the captured key value looks like a placeholder."""
if _PLACEHOLDER_VALUE_RE.search(key_value):
@@ -26,6 +26,7 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -68,23 +69,6 @@ def _line_has_python_instruction(line: str) -> bool:
# ── Bypass helper ───────────────────────────────────────────────────────
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# ── Main checker entry point ────────────────────────────────────────────
@@ -20,28 +20,12 @@ from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows import standards for pip packages.
@@ -23,31 +23,12 @@ from pathlib import Path
from typing import Dict, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Run on ALL .py files so modules (apps/modules/*.py) are checked, not just entry points
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows introspection standards
@@ -26,28 +26,12 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: scan every .py file, not just entry point
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows JSON structure standards.
@@ -25,28 +25,12 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows log handler standards
@@ -24,29 +24,13 @@ import re
from pathlib import Path
from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
from aipass.seedgo.apps.handlers.json import json_handler
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows log level hygiene standards
@@ -21,6 +21,7 @@ from pathlib import Path
from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
@@ -50,23 +51,6 @@ def _find_branch_root(file_path: Path) -> Path:
return file_path.parent
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check module logging structure against the two-tier model.
@@ -25,6 +25,7 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
@@ -34,23 +35,6 @@ _GETLOGGER_PAT = r"logging" + r"\.getLogger\s*\("
_FILEHANDLER_PAT = r"logging" + r"\.FileHandler\s*\("
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module's logging is visible to Prax monitor (system_logs/).
@@ -31,6 +31,7 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -50,23 +51,6 @@ REQUIRED_FIELDS = {
}
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module has a valid library-profile META block.
@@ -19,31 +19,12 @@ from pathlib import Path
from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows module standards
@@ -18,31 +18,12 @@ from pathlib import Path
from typing import Dict, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module follows naming standards
@@ -26,28 +26,12 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: all Python files
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def _get_non_code_lines(lines: List[str]) -> set:
"""
Build a set of line numbers that are inside docstrings or comments.
@@ -27,28 +27,12 @@ from pathlib import Path
from typing import Dict, List, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: entry points only (apps/{name}.py)
AUDIT_SCOPE = "entry_point"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if branch README follows standards
@@ -31,25 +31,12 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "branch_level"
ADVISORY = True
def is_bypassed(file_path: str, standard: str, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed via standard bypass.json rules."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
return True
return False
def _load_ruff_bypass(branch_path: Path) -> list:
"""Load .seedgo/ruff_bypass.json for ruff-specific bypass rules."""
bypass_file = branch_path / ".seedgo" / "ruff_bypass.json"
@@ -19,28 +19,12 @@ from pathlib import Path
from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: scan every .py file, not just entry point
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if a Python file contains a shebang line.
@@ -26,6 +26,7 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
# Audit scope: scan every .py file, not just entry point
AUDIT_SCOPE = "all_files"
@@ -34,23 +35,6 @@ AUDIT_SCOPE = "all_files"
_LOGGING_ATTRS = frozenset({"error", "warning", "warn", "info", "debug", "exception", "critical"})
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# -- AST helpers (extracted from devpulse silent_catch_scanner_v2) ---------
@@ -27,29 +27,13 @@ import re
from pathlib import Path
from typing import Dict, List
from aipass.prax import logger
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
from aipass.seedgo.apps.handlers.json import json_handler
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module routes error/warning output to stderr via CLI display functions.
@@ -28,6 +28,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "branch_level"
@@ -175,28 +176,6 @@ TOTAL_ITEMS = _PATTERN_ITEMS + _MODULE_COVERAGE_ITEMS
# =============================================
def is_bypassed(
file_path: str,
standard: str,
line: int | None = None,
bypass_rules: list | None = None,
) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# =============================================
# FILE HELPERS
# =============================================
@@ -21,6 +21,7 @@ from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "all_files"
@@ -35,26 +36,6 @@ _TAG_RE = re.compile(
)
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
# Must match standard
if rule.get("standard") and rule.get("standard") != standard:
continue
# Must match file (check if rule file path is in the full path)
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
# Check line-specific bypass
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check a Python file for TODO/FIXME/HACK/XXX comments.
@@ -31,6 +31,7 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
AUDIT_SCOPE = "branch_level"
@@ -78,23 +79,6 @@ _MAIN_BLOCK_RE = re.compile(
# -- Bypass helper ------------------------------------------------------------
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
return True
return False
# -- File collection ----------------------------------------------------------
@@ -0,0 +1,52 @@
# =================== AIPass ====================
# Name: utils.py
# Description: Shared bypass checking utility for standards checkers
# Version: 1.0.0
# Created: 2026-04-27
# Modified: 2026-04-27
# =============================================
"""Shared bypass checking utility for standards checkers."""
from aipass.seedgo.apps.handlers.json import json_handler
def is_bypassed(
file_path: str,
standard: str,
line: int | None = None,
bypass_rules: list | None = None,
) -> bool:
"""Check if a violation should be bypassed.
Args:
file_path: Path to the file being checked
standard: Standard name (e.g., 'cli', 'imports')
line: Optional specific line number of the violation
bypass_rules: List of bypass rules from .seedgo/bypass.json
Returns:
True if this violation should be bypassed
"""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get("standard") and rule.get("standard") != standard:
continue
rule_file = rule.get("file", "")
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get("lines", [])
if rule_lines and line is not None and line not in rule_lines:
continue
json_handler.log_operation(
"bypass_matched",
{
"file": file_path,
"standard": standard,
"line": line,
"rule_file": rule_file,
},
)
return True
return False