diff --git a/src/aipass/hooks/.seedgo/bypass.json b/src/aipass/hooks/.seedgo/bypass.json index 892f2145..56607918 100644 --- a/src/aipass/hooks/.seedgo/bypass.json +++ b/src/aipass/hooks/.seedgo/bypass.json @@ -869,6 +869,31 @@ "standard": "unused_function", "reason": "sandbox_launch(), build_policy(), build_srt_config(), and resolve_bwrap_command() are consumed CROSS-BRANCH by ai_mail's dispatch_monitor.py (the launch seam at Phase 4). seedgo's intra-branch static analysis cannot see these callers. Verified: dispatch_monitor imports sandbox module to wire build_policy + sandbox_launch at agent launch." }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "dead_code", + "reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.security.presence_gate.handle' — not statically imported by design. Wired in UserPromptSubmit.presence_gate + Stop.presence_release." + }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "unused_function", + "reason": "handle() and handle_stop() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Wired in UserPromptSubmit.presence_gate and Stop.presence_release." + }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "json_structure", + "reason": "Security gate uses stdlib json.dumps for hook protocol block responses — no JSON file ops needing json_handler." + }, + { + "file": "apps/modules/presence.py", + "standard": "json_structure", + "reason": "Presence service uses stdlib json for .ai_central/PRESENCE.central.json — shared runtime namespace, not branch json_handler storage." + }, + { + "file": "apps/modules/presence.py", + "standard": "unused_function", + "reason": "claim(), release(), refresh(), read_all() consumed by presence_gate.py handler (dynamically dispatched via engine). Static analysis cannot trace hooks.json → engine → handler → presence module." + }, { "file": "apps/handlers/notification/telegram_response.py", "standard": "unused_function", @@ -898,6 +923,81 @@ "file": "tests/test_telegram_response.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers." + }, + { + "file": "apps/modules/presence.py", + "standard": "json_structure", + "reason": "Uses stdlib json for .ai_central/PRESENCE.central.json shared runtime namespace — not branch data storage needing json_handler." + }, + { + "file": "apps/modules/presence.py", + "standard": "modules", + "reason": "Direct file ops on .ai_central/PRESENCE.central.json — a shared runtime namespace (gitignored, cross-branch). Not a handler-level concern; the module IS the presence service that owns this file." + }, + { + "file": "apps/modules/presence.py", + "standard": "dead_code", + "reason": "claim(), release(), refresh(), read_all() consumed by presence_gate.py handler (dynamically dispatched via engine). handle_command() and print_introspection() called by drone routing. Static analysis cannot trace hooks.json → engine → handler → presence module." + }, + { + "file": "apps/modules/presence.py", + "standard": "unused_function", + "reason": "claim(), release(), refresh(), read_all() consumed by presence_gate.py handler (dynamically dispatched via engine). handle_command() and print_introspection() called by drone routing." + }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "dead_code", + "reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.security.presence_gate.handle' — not statically imported by design. Wired in UserPromptSubmit.presence_gate + Stop.presence_release." + }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "unused_function", + "reason": "handle() and handle_stop() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Wired in UserPromptSubmit.presence_gate + Stop.presence_release." + }, + { + "file": "apps/handlers/security/presence_gate.py", + "standard": "json_structure", + "reason": "Security gate uses stdlib json.dumps for hook protocol block responses — no JSON file ops needing json_handler." + }, + { + "file": "tests/test_presence.py", + "standard": "architecture", + "reason": "Test files live in tests/, not in the 3-layer apps structure." + }, + { + "file": "tests/test_presence.py", + "standard": "documentation", + "reason": "Test methods use descriptive names as documentation per pytest convention." + }, + { + "file": "tests/test_presence.py", + "standard": "encapsulation", + "reason": "Tests import modules directly to test implementation details." + }, + { + "file": "tests/test_presence.py", + "standard": "meta", + "reason": "Test files do not need Version/Modified metadata headers." + }, + { + "file": "tests/test_presence_gate.py", + "standard": "architecture", + "reason": "Test files live in tests/, not in the 3-layer apps structure." + }, + { + "file": "tests/test_presence_gate.py", + "standard": "documentation", + "reason": "Test methods use descriptive names as documentation per pytest convention." + }, + { + "file": "tests/test_presence_gate.py", + "standard": "encapsulation", + "reason": "Tests import handlers directly to test implementation details." + }, + { + "file": "tests/test_presence_gate.py", + "standard": "meta", + "reason": "Test files do not need Version/Modified metadata headers." } ], "notes": { diff --git a/src/aipass/hooks/README.md b/src/aipass/hooks/README.md index d619a1a2..e46bda2c 100644 --- a/src/aipass/hooks/README.md +++ b/src/aipass/hooks/README.md @@ -52,6 +52,7 @@ src/aipass/hooks/ │ │ ├── engine.py # Core dispatch — routes events to handlers │ │ ├── hooksound.py # Sound control (drone @hooks hooksound on/off) │ │ ├── hookstatus.py # Config viewer (drone @hooks status) +│ │ ├── presence.py # Branch presence — claim/release/refresh for .ai_central/PRESENCE.central.json │ │ └── sandbox.py # Kernel sandbox — srt/bwrap wrapper + per-role policy generator │ ├── handlers/ │ │ ├── bridges/ # One per provider (thin normalization) @@ -64,6 +65,7 @@ src/aipass/hooks/ │ │ ├── security/ # Enforcement hooks │ │ │ ├── edit_gate.py # Blocks unsafe edits (cross-branch, inbox, diagnostics) │ │ │ ├── git_gate.py # Enforces git access tiers +│ │ │ ├── presence_gate.py # Single-session gate — blocks duplicate runtimes per branch │ │ │ ├── rm_gate.py # Guardrail — catches accidental rm -rf, teaches drone rm │ │ │ └── subagent_gate.py # Blocks sub-agent stop until clean │ │ ├── lifecycle/ # Session management hooks @@ -82,7 +84,7 @@ src/aipass/hooks/ │ └── diagnostics.py # JSONL logging for hook execution ├── logs/ │ └── engine.jsonl # JSONL diagnostics (every hook execution) -└── tests/ # 666 tests across 23 test files +└── tests/ # 705 tests across 25 test files ``` ## How It Works @@ -103,11 +105,11 @@ Handlers are called **dynamically at runtime** — the engine uses `importlib.im | Event | Hooks | Description | |---|---|---| -| UserPromptSubmit | identity, email, branch_loader, tier0_kernel, navmap | Prompt injection + inbox check | +| UserPromptSubmit | presence_gate, identity, email, branch_loader, tier0_kernel, navmap | Presence gate + prompt injection + inbox check | | PreToolUse | tool_sound, edit_gate, git_gate, rm_gate | Security gates + guardrails + sound | | PostToolUse | auto_fix, auto_watchdog | Diagnostics + watchdog | | SubagentStop | subagent_gate | Seedgo validation | -| Stop | stop_sound, telegram_response | Achievement bell + Telegram reply delivery | +| Stop | stop_sound, telegram_response, presence_release | Bell + Telegram delivery + presence release | | Notification | announce | Announcement tone | | PreCompact | compact, rollover | Memory archival + rollover | @@ -151,7 +153,7 @@ The @drone broker validates sandbox policy before agent launch. @ai_mail's dispa - All branches via hook dispatch — every Claude Code session routes through the engine - @ai_mail dispatch_monitor — sandbox_launch + build_policy for agent launch boundary -*Last Updated: 2026-06-10* +*Last Updated: 2026-06-29* --- diff --git a/src/aipass/hooks/apps/handlers/security/presence_gate.py b/src/aipass/hooks/apps/handlers/security/presence_gate.py new file mode 100644 index 00000000..9a4887bd --- /dev/null +++ b/src/aipass/hooks/apps/handlers/security/presence_gate.py @@ -0,0 +1,96 @@ +# =================== AIPass ==================== +# Name: presence_gate.py +# Version: 1.0.0 +# Description: Single-session gate — blocks duplicate Claude runtimes per branch +# Branch: hooks +# Layer: apps/handlers/security +# Created: 2026-06-29 +# Modified: 2026-06-29 +# ============================================= + +"""Single-session gate — blocks duplicate Claude runtimes per branch. + +Fires on UserPromptSubmit: calls presence.claim(). If OCCUPIED by a live PID, +blocks the prompt. If free or stale, acquires and proceeds. + +Fires on Stop: calls presence.release() to clean up. + +Skips sub-agents and dispatched/daemon session types. +""" + +import importlib +import json +import os +from pathlib import Path + +from aipass.prax.apps.modules.logger import system_logger as logger + +_ALLOW = {"exit_code": 0, "stdout": ""} +_NON_BLOCKING_SESSION_TYPES = frozenset({"dispatched", "daemon"}) + + +def handle(hook_data: dict) -> dict: + """UserPromptSubmit gate — enforce one live session per branch. + + Args: + hook_data: Parsed hook event dict from engine. + + Returns: + Result dict with stdout (block JSON or empty) and exit_code. + """ + try: + agent_type = hook_data.get("agent_type", "") + if agent_type and agent_type != "main": + return _ALLOW + + session_type = os.environ.get("AIPASS_SESSION_TYPE", "interactive") + if session_type in _NON_BLOCKING_SESSION_TYPES: + return _ALLOW + + branch = Path.cwd().name + session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "") + + presence = importlib.import_module("aipass.hooks.apps.modules.presence") + result = presence.claim( + branch=branch, + session_id=session_id, + session_type=session_type, + ) + + if result["status"] == "ACQUIRED": + return _ALLOW + + pid = result.get("pid", "?") + holder_type = result.get("session_type", "unknown") + reason = f"{branch} already live at PID {pid} (session_type: {holder_type}) — attach, do not spawn." + logger.warning("[presence_gate] BLOCKED: %s", reason) + return { + "exit_code": 2, + "stdout": json.dumps({"decision": "block", "reason": reason}), + "sound": "presence gate", + } + except Exception as exc: + logger.warning("[presence_gate] gate error (allowing): %s", exc) + return _ALLOW + + +def handle_stop(hook_data: dict) -> dict: + """Release presence on Stop event. + + Args: + hook_data: Parsed hook event dict from engine. + + Returns: + Result dict (always allows — Stop is informational). + """ + branch = Path.cwd().name + try: + presence = importlib.import_module("aipass.hooks.apps.modules.presence") + released = presence.release(branch) + if released: + logger.info("[presence_gate] released %s on Stop", branch) + else: + logger.info("[presence_gate] nothing to release for %s on Stop", branch) + except Exception as exc: + logger.warning("[presence_gate] release failed for %s: %s", branch, exc) + return _ALLOW diff --git a/src/aipass/hooks/apps/modules/presence.py b/src/aipass/hooks/apps/modules/presence.py new file mode 100644 index 00000000..215f2e2a --- /dev/null +++ b/src/aipass/hooks/apps/modules/presence.py @@ -0,0 +1,357 @@ +# =================== AIPass ==================== +# Name: presence.py +# Version: 1.0.0 +# Description: Branch presence service — claim/release/refresh for .ai_central/PRESENCE.central.json +# Branch: hooks +# Layer: apps/modules +# Created: 2026-06-29 +# Modified: 2026-06-29 +# ============================================= + +"""Branch presence service for concurrent session detection. + +Manages a shared PRESENCE.central.json in .ai_central/ at the AIPass project root. +Each branch can claim presence (one live session per branch), detect stale holders +via PID liveness + /proc/cwd verification, and release on exit. + +File locking uses flock (POSIX) / msvcrt (Windows) to prevent concurrent corruption. +""" + +import json +import os +import sys +from contextlib import contextmanager +from datetime import datetime +from pathlib import Path + +from aipass.cli.apps.modules import err_console +from aipass.prax.apps.modules.logger import system_logger as logger + +CONSOLE = err_console + +# Cross-platform flock +if sys.platform == "win32": + import msvcrt +else: + import fcntl + + +# --------------------------------------------------------------------------- +# Path resolution +# --------------------------------------------------------------------------- + + +def _find_ai_central() -> Path: + """Walk up from this file to find the directory containing .ai_central/.""" + current = Path(__file__).resolve().parent + for _ in range(20): # safety cap + candidate = current / ".ai_central" + if candidate.is_dir(): + return candidate + if current.parent == current: + break + current = current.parent + raise FileNotFoundError("Cannot locate .ai_central/ directory from presence module") + + +_PRESENCE_FILE_NAME = "PRESENCE.central.json" +_LOCK_FILE_NAME = ".presence.lock" + + +def _presence_path() -> Path: + """Return path to the PRESENCE.central.json file.""" + return _find_ai_central() / _PRESENCE_FILE_NAME + + +def _lock_path() -> Path: + """Return path to the .presence.lock file.""" + return _find_ai_central() / _LOCK_FILE_NAME + + +# --------------------------------------------------------------------------- +# File locking context manager +# --------------------------------------------------------------------------- + + +def _release_lock_fd(lock_fd) -> None: + """Release and close a lock file descriptor.""" + try: + if sys.platform == "win32": + msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN) + lock_fd.close() + except Exception as exc: + logger.warning("[PRESENCE] lock release failed: %s", exc) + try: + lock_fd.close() + except Exception as close_exc: + logger.warning("[PRESENCE] lock file close failed: %s", close_exc) + + +@contextmanager +def _presence_lock(): + """Acquire exclusive lock on the presence file for read-modify-write.""" + lock_file = _lock_path() + lock_fd = None + try: + lock_fd = open(lock_file, "w", encoding="utf-8") + if sys.platform == "win32": + msvcrt.locking(lock_fd.fileno(), msvcrt.LK_LOCK, 1) + else: + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX) + yield + finally: + if lock_fd is not None: + _release_lock_fd(lock_fd) + + +# --------------------------------------------------------------------------- +# JSON read/write helpers +# --------------------------------------------------------------------------- + + +def _read_presence() -> dict: + """Read the presence JSON. Returns empty dict if missing or corrupt.""" + fp = _presence_path() + if not fp.exists(): + return {} + try: + return json.loads(fp.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("[PRESENCE] Failed to read %s: %s", fp, exc) + return {} + + +def _write_presence(data: dict) -> None: + """Write the presence JSON.""" + fp = _presence_path() + fp.parent.mkdir(parents=True, exist_ok=True) + fp.write_text(json.dumps(data, indent=2), encoding="utf-8") + + +# --------------------------------------------------------------------------- +# Liveness detection +# --------------------------------------------------------------------------- + + +def _is_pid_alive(pid: int) -> bool: + """Check if a process with the given PID exists.""" + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + logger.info("[PRESENCE] PID %d not found (dead)", pid) + return False + except PermissionError: + # Process exists but we can't signal it — treat as alive + logger.info("[PRESENCE] PID %d exists but permission denied — treating as alive", pid) + return True + except OSError as exc: + logger.info("[PRESENCE] PID %d os.kill failed: %s — treating as dead", pid, exc) + return False + + +def _cwd_matches(pid: int, expected_dir: str) -> bool: + """Check if the process CWD matches the expected branch directory. + + Linux only — reads /proc//cwd. + On non-Linux platforms, returns True (skip the check, rely on os.kill alone). + """ + if sys.platform != "linux": + return True + try: + actual_cwd = os.readlink(f"/proc/{pid}/cwd") + return str(Path(actual_cwd).resolve()) == str(Path(expected_dir).resolve()) + except (OSError, PermissionError): + # Cannot read /proc — be conservative, treat as NOT matching (stale) + logger.info("[PRESENCE] Cannot read /proc/%d/cwd — treating as stale", pid) + return False + + +def _is_holder_alive(entry: dict) -> bool: + """Determine if the holder recorded in the presence entry is still alive. + + A holder is alive if: + 1. The PID exists (os.kill signal 0) + 2. AND the PID's CWD matches the branch work_dir (Linux /proc check) + + If either check fails, the holder is stale and can be reclaimed. + """ + pid = entry.get("pid") + if pid is None: + return False + if not _is_pid_alive(pid): + return False + work_dir = entry.get("work_dir", "") + if not work_dir: + return False + return _cwd_matches(pid, work_dir) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def claim( + branch: str, + session_id: str = "", + session_type: str = "interactive", + attach_handle: str = "", +) -> dict: + """Claim presence for a branch. + + Returns: + {"status": "ACQUIRED"} on success, or + {"status": "OCCUPIED", "pid": N, "session_id": "...", + "work_dir": "...", "session_type": "..."} + when a live session already owns the branch. + """ + my_pid = os.getpid() + now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") + cwd = os.getcwd() + + with _presence_lock(): + data = _read_presence() + existing = data.get(branch) + + if existing: + result = _handle_existing(data, existing, branch, my_pid, now, session_id) + if result is not None: + return result + + # No existing entry or stale holder — write the new entry + data[branch] = { + "pid": my_pid, + "session_id": session_id, + "work_dir": cwd, + "session_type": session_type, + "attach_handle": attach_handle, + "started": now, + "last_seen": now, + } + _write_presence(data) + logger.info("[PRESENCE] Acquired %s (PID %d)", branch, my_pid) + return {"status": "ACQUIRED"} + + +def _handle_existing( + data: dict, + existing: dict, + branch: str, + my_pid: int, + now: str, + session_id: str, +) -> dict | None: + """Handle an existing presence entry. Returns a result dict or None to proceed.""" + holder_pid = existing.get("pid") + + # Re-entry: same PID already holds it + if holder_pid == my_pid: + existing["last_seen"] = now + existing["session_id"] = session_id or existing.get("session_id", "") + _write_presence(data) + logger.info("[PRESENCE] Re-entry for %s (PID %d)", branch, my_pid) + return {"status": "ACQUIRED"} + + # Check if the holder is still alive + if _is_holder_alive(existing): + logger.info( + "[PRESENCE] %s occupied by PID %d (session_type=%s)", + branch, + holder_pid, + existing.get("session_type", "unknown"), + ) + return { + "status": "OCCUPIED", + "pid": holder_pid, + "session_id": existing.get("session_id", ""), + "work_dir": existing.get("work_dir", ""), + "session_type": existing.get("session_type", ""), + } + + # Stale holder — let caller reclaim + logger.info( + "[PRESENCE] Stale holder for %s (PID %d) — reclaiming", + branch, + holder_pid, + ) + return None + + +def release(branch: str) -> bool: + """Release presence for a branch. Only the holder PID can release its own entry.""" + my_pid = os.getpid() + with _presence_lock(): + data = _read_presence() + if branch not in data: + return False + if data[branch].get("pid") != my_pid: + logger.info( + "[PRESENCE] Release skipped for %s — not our PID (ours=%d, holder=%d)", + branch, + my_pid, + data[branch].get("pid", 0), + ) + return False + del data[branch] + _write_presence(data) + logger.info("[PRESENCE] Released %s (PID %d)", branch, my_pid) + return True + + +def refresh(branch: str) -> None: + """Update last_seen timestamp for the branch entry.""" + now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") + with _presence_lock(): + data = _read_presence() + if branch in data: + data[branch]["last_seen"] = now + _write_presence(data) + logger.info("[PRESENCE] Refreshed %s", branch) + + +def read_all() -> dict: + """Return the full presence JSON (all branches).""" + return _read_presence() + + +# ============================================================================= +# MODULE INTERFACE (drone @hooks routing) +# ============================================================================= + + +def print_introspection() -> None: + """Print presence state for drone routing.""" + CONSOLE.print("[bold cyan]presence[/bold cyan] Module") + try: + data = _read_presence() + if not data: + CONSOLE.print(" No branches currently present") + else: + for branch, entry in data.items(): + pid = entry.get("pid", "?") + stype = entry.get("session_type", "unknown") + last = entry.get("last_seen", "?") + alive = _is_holder_alive(entry) + status = "[green]live[/green]" if alive else "[dim]stale[/dim]" + CONSOLE.print(f" {branch}: PID {pid} type={stype} last_seen={last} {status}") + except FileNotFoundError as exc: + logger.info("[PRESENCE] introspection: %s", exc) + CONSOLE.print(" .ai_central/ not found") + + +def handle_command(command: str, args: list) -> bool: + """Route presence commands from drone @hooks.""" + if command in ("--help", "-h", "help"): + CONSOLE.print("[bold cyan]presence[/bold cyan] — Branch presence claim/release service") + CONSOLE.print() + CONSOLE.print(" drone @hooks presence Show current presence state for all branches") + return True + + if command == "presence": + if not args: + print_introspection() + return True + return False diff --git a/src/aipass/hooks/tests/test_presence.py b/src/aipass/hooks/tests/test_presence.py new file mode 100644 index 00000000..da627542 --- /dev/null +++ b/src/aipass/hooks/tests/test_presence.py @@ -0,0 +1,404 @@ +"""Tests for the presence service module.""" + +import json +from unittest.mock import patch + +import pytest + +from aipass.hooks.apps.modules import presence + + +@pytest.fixture +def presence_dir(tmp_path): + """Create a temporary .ai_central directory with PRESENCE.central.json.""" + ai_central = tmp_path / ".ai_central" + ai_central.mkdir() + return ai_central + + +@pytest.fixture +def presence_file(presence_dir): + """Return path to the presence file.""" + return presence_dir / "PRESENCE.central.json" + + +@pytest.fixture +def patch_paths(presence_dir): + """Patch _find_ai_central to use the temp directory.""" + with patch.object(presence, "_find_ai_central", return_value=presence_dir): + yield presence_dir + + +@pytest.fixture +def patch_flock(): + """Patch flock to be a no-op (avoid real file locking in tests).""" + with patch("aipass.hooks.apps.modules.presence.fcntl") as mock_fcntl: + mock_fcntl.LOCK_EX = 2 + mock_fcntl.LOCK_UN = 8 + yield mock_fcntl + + +# ── claim tests ────────────────────────────────────────────────────────── + + +class TestClaim: + def test_claim_empty_file(self, patch_paths, patch_flock, presence_file): + with patch("os.getpid", return_value=1000), patch("os.getcwd", return_value="/tmp/branch"): + result = presence.claim("devpulse", session_id="abc") + assert result["status"] == "ACQUIRED" + data = json.loads(presence_file.read_text()) + assert data["devpulse"]["pid"] == 1000 + assert data["devpulse"]["session_id"] == "abc" + assert data["devpulse"]["work_dir"] == "/tmp/branch" + + def test_claim_reentry_same_pid(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 1000, + "session_id": "old", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with patch("os.getpid", return_value=1000), patch("os.getcwd", return_value="/w"): + result = presence.claim("devpulse", session_id="new-id") + assert result["status"] == "ACQUIRED" + data = json.loads(presence_file.read_text()) + assert data["devpulse"]["session_id"] == "new-id" + + def test_claim_stale_dead_pid(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 9999, + "session_id": "old", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with ( + patch("os.getpid", return_value=2000), + patch("os.getcwd", return_value="/w2"), + patch.object(presence, "_is_pid_alive", return_value=False), + ): + result = presence.claim("devpulse", session_id="new") + assert result["status"] == "ACQUIRED" + data = json.loads(presence_file.read_text()) + assert data["devpulse"]["pid"] == 2000 + + def test_claim_occupied_live_pid(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 5000, + "session_id": "live", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with ( + patch("os.getpid", return_value=6000), + patch("os.getcwd", return_value="/w2"), + patch.object(presence, "_is_pid_alive", return_value=True), + patch.object(presence, "_cwd_matches", return_value=True), + ): + result = presence.claim("devpulse", session_id="new") + assert result["status"] == "OCCUPIED" + assert result["pid"] == 5000 + assert result["session_type"] == "interactive" + + def test_claim_stale_cwd_mismatch(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 5000, + "session_id": "old", + "work_dir": "/original", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with ( + patch("os.getpid", return_value=6000), + patch("os.getcwd", return_value="/new"), + patch.object(presence, "_is_pid_alive", return_value=True), + patch.object(presence, "_cwd_matches", return_value=False), + ): + result = presence.claim("devpulse", session_id="new") + assert result["status"] == "ACQUIRED" + + def test_claim_no_existing_file(self, patch_paths, patch_flock): + with patch("os.getpid", return_value=1000), patch("os.getcwd", return_value="/w"): + result = presence.claim("hooks") + assert result["status"] == "ACQUIRED" + + def test_claim_multiple_branches(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "api": { + "pid": 3000, + "session_id": "a", + "work_dir": "/api", + "session_type": "interactive-mirror", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with patch("os.getpid", return_value=4000), patch("os.getcwd", return_value="/hooks"): + result = presence.claim("hooks", session_id="h1") + assert result["status"] == "ACQUIRED" + data = json.loads(presence_file.read_text()) + assert "api" in data + assert "hooks" in data + + +# ── release tests ──────────────────────────────────────────────────────── + + +class TestRelease: + def test_release_existing(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 1000, + "session_id": "a", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with patch("os.getpid", return_value=1000): + result = presence.release("devpulse") + assert result is True + data = json.loads(presence_file.read_text()) + assert "devpulse" not in data + + def test_release_not_claimed(self, patch_paths, patch_flock, presence_file): + presence_file.write_text(json.dumps({})) + result = presence.release("devpulse") + assert result is False + + def test_release_wrong_pid_refused(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 1000, + "session_id": "a", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + with patch("os.getpid", return_value=9999): + result = presence.release("devpulse") + assert result is False + data = json.loads(presence_file.read_text()) + assert "devpulse" in data + assert data["devpulse"]["pid"] == 1000 + + def test_release_preserves_others(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "devpulse": { + "pid": 1000, + "session_id": "a", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + }, + "api": { + "pid": 2000, + "session_id": "b", + "work_dir": "/api", + "session_type": "interactive-mirror", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + }, + } + ) + ) + with patch("os.getpid", return_value=1000): + presence.release("devpulse") + data = json.loads(presence_file.read_text()) + assert "api" in data + assert "devpulse" not in data + + +# ── refresh tests ──────────────────────────────────────────────────────── + + +class TestRefresh: + def test_refresh_updates_last_seen(self, patch_paths, patch_flock, presence_file): + presence_file.write_text( + json.dumps( + { + "hooks": { + "pid": 1000, + "session_id": "a", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + ) + ) + presence.refresh("hooks") + data = json.loads(presence_file.read_text()) + assert data["hooks"]["last_seen"] != "2026-01-01T00:00:00" + + def test_refresh_nonexistent_noop(self, patch_paths, patch_flock, presence_file): + presence_file.write_text(json.dumps({})) + presence.refresh("hooks") + data = json.loads(presence_file.read_text()) + assert data == {} + + +# ── read_all tests ─────────────────────────────────────────────────────── + + +class TestReadAll: + def test_read_all_returns_data(self, patch_paths, patch_flock, presence_file): + expected = { + "hooks": { + "pid": 1000, + "session_id": "a", + "work_dir": "/w", + "session_type": "interactive", + "attach_handle": "", + "started": "2026-01-01T00:00:00", + "last_seen": "2026-01-01T00:00:00", + } + } + presence_file.write_text(json.dumps(expected)) + result = presence.read_all() + assert result == expected + + def test_read_all_empty(self, patch_paths, patch_flock): + result = presence.read_all() + assert result == {} + + +# ── liveness tests ─────────────────────────────────────────────────────── + + +class TestLiveness: + def test_is_pid_alive_true(self): + with patch("os.kill") as mock_kill: + assert presence._is_pid_alive(1234) is True + mock_kill.assert_called_once_with(1234, 0) + + def test_is_pid_alive_dead(self): + with patch("os.kill", side_effect=ProcessLookupError): + assert presence._is_pid_alive(1234) is False + + def test_is_pid_alive_permission_error(self): + with patch("os.kill", side_effect=PermissionError): + assert presence._is_pid_alive(1234) is True + + def test_cwd_matches_linux(self): + with ( + patch("sys.platform", "linux"), + patch("os.readlink", return_value="/tmp/project/src/aipass/hooks"), + ): + assert presence._cwd_matches(1234, "/tmp/project/src/aipass/hooks") is True + + def test_cwd_mismatch_linux(self): + with ( + patch("sys.platform", "linux"), + patch("os.readlink", return_value="/tmp/project/src/aipass/api"), + ): + assert presence._cwd_matches(1234, "/tmp/project/src/aipass/hooks") is False + + def test_cwd_matches_non_linux_skips(self): + with patch("sys.platform", "win32"): + assert presence._cwd_matches(1234, "/anything") is True + + def test_cwd_read_fails_treats_as_stale(self): + with ( + patch("sys.platform", "linux"), + patch("os.readlink", side_effect=OSError("no proc")), + ): + assert presence._cwd_matches(1234, "/w") is False + + def test_is_holder_alive_full_check(self): + entry = {"pid": 1234, "work_dir": "/w"} + with ( + patch.object(presence, "_is_pid_alive", return_value=True), + patch.object(presence, "_cwd_matches", return_value=True), + ): + assert presence._is_holder_alive(entry) is True + + def test_is_holder_alive_dead_pid(self): + entry = {"pid": 1234, "work_dir": "/w"} + with patch.object(presence, "_is_pid_alive", return_value=False): + assert presence._is_holder_alive(entry) is False + + def test_is_holder_alive_no_pid(self): + assert presence._is_holder_alive({}) is False + + def test_is_holder_alive_no_work_dir(self): + entry = {"pid": 1234, "work_dir": ""} + with patch.object(presence, "_is_pid_alive", return_value=True): + assert presence._is_holder_alive(entry) is False + + +# ── file locking tests ────────────────────────────────────────────────── + + +class TestFileLocking: + def test_presence_lock_acquires_flock(self, patch_paths): + with patch("aipass.hooks.apps.modules.presence.fcntl") as mock_fcntl: + mock_fcntl.LOCK_EX = 2 + mock_fcntl.LOCK_UN = 8 + with presence._presence_lock(): + pass + mock_fcntl.flock.assert_called() + + def test_corrupt_json_returns_empty(self, patch_paths, patch_flock, presence_file): + presence_file.write_text("not json {{{") + result = presence._read_presence() + assert result == {} diff --git a/src/aipass/hooks/tests/test_presence_gate.py b/src/aipass/hooks/tests/test_presence_gate.py new file mode 100644 index 00000000..64279b1d --- /dev/null +++ b/src/aipass/hooks/tests/test_presence_gate.py @@ -0,0 +1,104 @@ +"""Tests for the presence gate handler.""" + +import json +import os +from unittest.mock import MagicMock, patch + +from aipass.hooks.apps.handlers.security import presence_gate + + +def _make_presence_mock(claim_result, release_result=True): + """Build a mock presence module with given claim/release return values.""" + mock = MagicMock() + mock.claim.return_value = claim_result + mock.release.return_value = release_result + return mock + + +_ACQUIRED_MOCK = _make_presence_mock({"status": "ACQUIRED"}) +_OCCUPIED_MOCK = _make_presence_mock( + { + "status": "OCCUPIED", + "pid": 5000, + "session_id": "existing", + "work_dir": "/tmp/branch", + "session_type": "interactive", + } +) + + +class TestHandle: + def test_first_prompt_acquired(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True): + with patch("importlib.import_module", return_value=_ACQUIRED_MOCK): + result = presence_gate.handle({}) + assert result["exit_code"] == 0 + + def test_occupied_blocks(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True): + with patch("importlib.import_module", return_value=_OCCUPIED_MOCK): + result = presence_gate.handle({}) + assert result["exit_code"] == 2 + parsed = json.loads(result["stdout"]) + assert parsed["decision"] == "block" + assert "5000" in parsed["reason"] + + def test_subagent_skipped(self): + result = presence_gate.handle({"agent_type": "sub"}) + assert result["exit_code"] == 0 + assert result["stdout"] == "" + + def test_main_agent_not_skipped(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True): + with patch("importlib.import_module", return_value=_ACQUIRED_MOCK): + result = presence_gate.handle({"agent_type": "main"}) + assert result["exit_code"] == 0 + + def test_dispatched_session_skipped(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "dispatched"}, clear=True): + result = presence_gate.handle({}) + assert result["exit_code"] == 0 + assert result["stdout"] == "" + + def test_daemon_session_skipped(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "daemon"}, clear=True): + result = presence_gate.handle({}) + assert result["exit_code"] == 0 + assert result["stdout"] == "" + + def test_resume_dead_holder_acquires(self): + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True): + with patch("importlib.import_module", return_value=_ACQUIRED_MOCK): + result = presence_gate.handle({}) + assert result["exit_code"] == 0 + + def test_block_message_includes_branch(self): + mock_cwd = MagicMock() + mock_cwd.name = "devpulse" + with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True): + with patch("importlib.import_module", return_value=_OCCUPIED_MOCK): + with patch.object(presence_gate.Path, "cwd", return_value=mock_cwd): + result = presence_gate.handle({}) + parsed = json.loads(result["stdout"]) + assert "devpulse" in parsed["reason"] + assert "attach" in parsed["reason"].lower() + + +class TestHandleStop: + def test_stop_releases(self): + mock = _make_presence_mock({"status": "ACQUIRED"}) + with patch("importlib.import_module", return_value=mock): + result = presence_gate.handle_stop({}) + assert result["exit_code"] == 0 + mock.release.assert_called_once() + + def test_stop_nothing_to_release(self): + mock = _make_presence_mock({"status": "ACQUIRED"}, release_result=False) + with patch("importlib.import_module", return_value=mock): + result = presence_gate.handle_stop({}) + assert result["exit_code"] == 0 + + def test_stop_exception_handled(self): + with patch("importlib.import_module", side_effect=ImportError("no module")): + result = presence_gate.handle_stop({}) + assert result["exit_code"] == 0