feat(hooks): presence service + single-session gate, dormant (FPLAN-0289 P1)

One live Claude runtime per branch. presence.py manages .ai_central/PRESENCE.central.json
(claim/release/refresh, PID + /proc/cwd liveness, stale-reclaim, PID-guarded release so a
non-holder can never release the holder). presence_gate.py: UserPromptSubmit blocks a
duplicate (exit 2 + decision:block), Stop releases; skips sub-agents + dispatched/daemon.

SessionStart can't block in Claude Code (inject-only) → gate is UserPromptSubmit, like the
edit/git gates. NOT wired into hooks.json yet — dormant, zero behavior change until enabled.

705 tests pass, seedgo 100%. Live cross-process block + PID-guard verified (devpulse).
Design: DPLAN-0225. Next: P2 telegram relay follows the pointer (@skills).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GFihce1oLtp6UDAPGryYSv
This commit is contained in:
AIOSAI
2026-06-29 13:51:48 -07:00
co-authored by Claude Opus 4.8
parent 2217b96054
commit f460cd577e
6 changed files with 1067 additions and 4 deletions
+100
View File
@@ -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": {
+6 -4
View File
@@ -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*
---
@@ -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
+357
View File
@@ -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/<pid>/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
+404
View File
@@ -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 == {}
@@ -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