slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#678 owner-capability: seal project ownership in the registry + wake the OWNER back on dispatch completion. TDPLAN-0012 — 3 parts built in parallel against a frozen is_owner contract, verified end-to-end by devpulse.

Browse Source 
@spawn: owner + registry_id written into the SEALED registry entries (authority lives in registry, not the self-editable passport). ensure_project_has_owner() now keys off citizen_class=manager (was earliest-created, which mislabeled @aipass) and writes the registry entry. get_owner()/is_owner() resolvers added. 315 tests, seedgo 100%.
@hooks: new registry_gate PreToolUse handler seals *_REGISTRY.json — blocks raw writes/tee/sed/rm + Edit/Write/MultiEdit, redirects to drone @spawn; per-clause bypass defeats compound-command smuggling; reads allowed. 82 tests, seedgo 100%.
@ai_mail: wake-back reslope — SKIP_SENDERS blocklist replaced by an is_owner allowlist. Only the project owner is woken when their dispatched agent completes; all other guards intact (depth cap, lock, occupancy, honest messaging, dispatch_wake.log). seedgo 100% on the changed file.

devpulse cross-part verify (REAL unmocked resolver): is_owner resolves devpulse-only; gate 13/13 incl compound-smuggle blocked + reads/drone-@spawn allowed; wake-back wakes owner / skips non-owner / respects depth-cap; 195 new-suite tests green together. Note: AIPASS_REGISTRY.json is gitignored — this ships the CODE; owner data regenerates per-install via ensure_project_has_owner. Still open (PART 4): gate watchdog+feedback on is_owner; portability of owner-only privileges across projects.
This commit is contained in:
AIOSAI
2026-07-10 00:46:17 -07:00
parent ae8f4a843a
commit 874c7fed2e
15 changed files with 1719 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.aipass/hooks.json
+5
View File
@@ -63,6 +63,11 @@
"handler": "aipass.hooks.apps.handlers.security.rm_gate.handle",
"matcher": "Bash"
},
"registry_gate": {
"enabled": true,
"handler": "aipass.hooks.apps.handlers.security.registry_gate.handle",
"matcher": "Bash|Edit|MultiEdit|Write|NotebookEdit"
},
"engine_test_sound": {
"enabled": false,
"command": "python3 $AIPASS_HOME/.claude/hooks/engine_test_sound.py",
src/aipass/ai_mail/apps/handlers/dispatch/dispatch_monitor.py
+84
View File
@@ -80,6 +80,86 @@ def _connect_broker(repo_root: Path, branch_name: str) -> socket.socket:
return create_identified_connection(socket_path, secret_path, branch_name)
MAX_WAKE_DEPTH = 3
def _wake_sender(sender: str, branch_email: str, exit_code: int, lock_file: str) -> str:
"""Wake the dispatcher back after target completion.
Wake-back is owner-only: only the project owner (sealed registry)
gets woken. Non-owners silently skipped.
Returns a result tag for the dispatch_wake.log:
success, blocked_occupied, blocked_locked, blocked_depth,
skipped_sender, skipped_not_owner, failed
"""
if not sender or not sender.strip():
logger.info("[monitor] Wake-back skipped — no sender")
return "skipped_sender"
normalized = f"@{sender.lstrip('@').lower()}"
try:
from aipass.spawn.apps.handlers.registry import is_owner
except ImportError:
logger.warning("[monitor] Wake-back skipped — is_owner import failed")
return "failed"
if not is_owner(normalized):
logger.info("[monitor] Wake-back skipped — sender %s is not project owner", sender)
return "skipped_not_owner"
depth = int(os.environ.get("AIPASS_WAKE_DEPTH", "0"))
if depth >= MAX_WAKE_DEPTH:
logger.warning("[monitor] Wake-back skipped — depth %d >= max %d", depth, MAX_WAKE_DEPTH)
return "blocked_depth"
try:
from aipass.ai_mail.apps.handlers.dispatch.wake import wake_branch
os.environ["AIPASS_WAKE_DEPTH"] = str(depth + 1)
wake_status, success = wake_branch(sender, auto=True, sender="@ai_mail")
if success:
logger.info("[monitor] Wake-back: %s woken after %s completed (exit %d)", sender, branch_email, exit_code)
return "success"
summary = wake_status.summary
lower = summary.lower()
if "interactive" in lower or "occupancy" in lower or "occupied" in lower:
logger.info("[monitor] Wake-back blocked — sender %s has interactive session: %s", sender, summary)
return "blocked_occupied"
if "active agent" in lower or "lock" in lower:
logger.info("[monitor] Wake-back blocked — sender %s has active lock: %s", sender, summary)
return "blocked_locked"
logger.info("[monitor] Wake-back: %s not woken — %s", sender, summary)
return "failed"
except Exception as e:
logger.warning("[monitor] Wake-back failed for %s: %s", sender, e)
return "failed"
def _log_wake_result(branch_email: str, sender: str, exit_code: int, result: str, lock_file: str):
"""Append a wake-back result line to dispatch_wake.log under target's logs/."""
lock_path = Path(lock_file).resolve()
logs_dir = lock_path.parent.parent / "logs"
log_file = logs_dir / "dispatch_wake.log"
try:
logs_dir.mkdir(parents=True, exist_ok=True)
line = (
f"{time.strftime('%Y-%m-%dT%H:%M:%S')}"
f" target={branch_email}"
f" sender={sender}"
f" exit_code={exit_code}"
f" wake_result={result}\n"
)
with open(log_file, "a", encoding="utf-8") as f:
f.write(line)
except OSError as e:
logger.info("[monitor] Failed to write dispatch_wake.log: %s", e)
def _send_bounce(branch_email: str, reason: str, sender: str, lock_file: str, stderr_log: str) -> bool:
"""Send return-to-sender bounce email via drone."""
subject = f"BOUNCE: Dispatch to {branch_email} failed"
@@ -585,6 +665,10 @@ def main():
except Exception:
logger.info("[monitor] Desktop notification unavailable")
# ─── Wake-back: wake the dispatcher ────────────────────
wake_result = _wake_sender(sender, branch_email, exit_code, lock_file)
_log_wake_result(branch_email, sender, exit_code, wake_result, lock_file)
sys.exit(0 if exit_code == 0 else 1)
src/aipass/ai_mail/apps/modules/dispatch.py
+2 -56
View File
@@ -14,7 +14,6 @@ Delegates all business logic to handlers.
"""
import os
import subprocess
import sys
from pathlib import Path
from typing import List
@@ -48,7 +47,6 @@ DISPATCH (send + wake):
drone @ai_mail dispatch @branch "Subject" "Body" --fresh # Send + fresh wake
drone @ai_mail dispatch @branch "Subject" "Body" --model opus # Send + wake with Opus
drone @ai_mail dispatch @branch "Subject" "Body" --no-memory-save
drone @ai_mail dispatch @branch "Subject" "Body" --no-watchdog # Skip auto-watchdog
WAKE ONLY:
drone @ai_mail dispatch wake @branch # Wake with default inbox check
@@ -226,7 +224,6 @@ def _orchestrate_dispatch_send(args: List[str]) -> bool:
# Parse flags
use_fresh = False
no_memory_save = False
no_watchdog = False
from_branch = None
use_model = None
filtered = []
@@ -241,7 +238,6 @@ def _orchestrate_dispatch_send(args: List[str]) -> bool:
i += 1
continue
if args[i] == "--no-watchdog":
no_watchdog = True
i += 1
continue
if args[i] == "--from" and i + 1 < len(args):
@@ -349,62 +345,12 @@ def _orchestrate_dispatch_send(args: List[str]) -> bool:
if not wake_ok:
logger.warning("[dispatch] Wake failed for %s — email was sent", target)
error(f"Email sent but wake failed — retry: drone @ai_mail dispatch wake {target}")
elif not no_watchdog:
_spawn_watchdog(target)
else:
console.print(f"[dim]Wake-back enabled — sender will be woken when {target} completes (if available)[/dim]")
return True
def _spawn_watchdog(target: str) -> None:
"""Auto-spawn devpulse watchdog as a detached background process."""
from aipass.ai_mail.apps.handlers.registry.read import get_branch_by_email
from aipass.ai_mail.apps.handlers.paths import find_repo_root
devpulse_info = get_branch_by_email("@devpulse")
if not devpulse_info:
logger.warning("[dispatch] Cannot spawn watchdog — @devpulse not in registry")
return
_repo_root = find_repo_root()
devpulse_path = devpulse_info.get("path", "")
if not devpulse_path:
logger.warning("[dispatch] Cannot spawn watchdog — @devpulse has no path")
return
devpulse_dir = Path(devpulse_path)
if not devpulse_dir.is_absolute():
devpulse_dir = _repo_root / devpulse_dir
if not devpulse_dir.is_dir():
logger.warning("[dispatch] Cannot spawn watchdog — devpulse dir not found: %s", devpulse_dir)
return
cmd = ["drone", "@devpulse", "watchdog", "agent", target]
spawn_env = os.environ.copy()
local_bin = str(Path.home() / ".local" / "bin")
if local_bin not in spawn_env.get("PATH", ""):
spawn_env["PATH"] = local_bin + ":" + spawn_env.get("PATH", "")
_detach_kwargs: dict = {}
if sys.platform == "win32":
_detach_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
else:
_detach_kwargs["start_new_session"] = True
try:
subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
cwd=str(devpulse_dir),
env=spawn_env,
**_detach_kwargs,
)
console.print(f"[green]Watchdog armed for {target}[/green]")
except Exception as e:
logger.warning("[dispatch] Watchdog spawn failed for %s: %s", target, e)
def _orchestrate_daemon() -> bool:
"""Orchestrate daemon startup."""
logger.info("[dispatch] Starting dispatch daemon")
src/aipass/ai_mail/tests/test_dispatch_module.py
+18 -189
View File
@@ -16,8 +16,6 @@ All handler dependencies are mocked -- these tests verify orchestration
logic, not business logic.
"""
import subprocess
import sys
from contextlib import ExitStack
import pytest
@@ -971,172 +969,18 @@ class TestPrintIntrospection:
# ===========================================================================
# _spawn_watchdog
# Wake-back messaging (TDPLAN-0012 — retired _spawn_watchdog)
# ===========================================================================
class TestSpawnWatchdog:
"""Tests for _spawn_watchdog."""
def test_spawns_detached_subprocess(self, monkeypatch, tmp_path):
"""Successful watchdog spawn calls Popen with correct args."""
devpulse_dir = tmp_path / "src" / "aipass" / "devpulse"
devpulse_dir.mkdir(parents=True)
class TestWakeBackMessaging:
"""Tests for honest wake-back messaging after watchdog retirement."""
def test_wake_back_message_on_successful_wake(self, monkeypatch):
"""Successful send + wake prints wake-back enabled message."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
popen_calls: list[dict] = []
mock_popen = MagicMock()
def tracking_popen(cmd, **kwargs):
"""Capture Popen arguments."""
popen_calls.append({"cmd": cmd, **kwargs})
return mock_popen
with (
patch(
f"{_H_REG}.get_branch_by_email",
return_value={"email": "@devpulse", "path": str(devpulse_dir)},
),
patch(f"{MOD}.subprocess.Popen", side_effect=tracking_popen),
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
assert len(popen_calls) == 1
assert popen_calls[0]["cmd"] == ["drone", "@devpulse", "watchdog", "agent", "@flow"]
if sys.platform == "win32":
assert popen_calls[0].get("creationflags") == subprocess.CREATE_NEW_PROCESS_GROUP
assert "start_new_session" not in popen_calls[0]
else:
assert popen_calls[0]["start_new_session"] is True
assert popen_calls[0]["cwd"] == str(devpulse_dir)
combined = " ".join(printed)
assert "Watchdog armed for @flow" in combined
def test_devpulse_not_in_registry(self, monkeypatch):
"""No spawn when @devpulse not found in registry."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(f"{_H_REG}.get_branch_by_email", return_value=None),
patch(f"{MOD}.subprocess.Popen") as mock_popen,
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
mock_popen.assert_not_called()
def test_devpulse_no_path(self, monkeypatch):
"""No spawn when @devpulse has empty path."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
f"{_H_REG}.get_branch_by_email",
return_value={"email": "@devpulse", "path": ""},
),
patch(f"{MOD}.subprocess.Popen") as mock_popen,
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
mock_popen.assert_not_called()
def test_devpulse_dir_missing(self, monkeypatch, tmp_path):
"""No spawn when devpulse directory doesn't exist."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
f"{_H_REG}.get_branch_by_email",
return_value={"email": "@devpulse", "path": str(tmp_path / "nonexistent")},
),
patch(f"{MOD}.subprocess.Popen") as mock_popen,
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
mock_popen.assert_not_called()
def test_popen_failure_warns_but_does_not_raise(self, monkeypatch, tmp_path):
"""Popen failure logs warning but doesn't propagate."""
devpulse_dir = tmp_path / "src" / "aipass" / "devpulse"
devpulse_dir.mkdir(parents=True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
f"{_H_REG}.get_branch_by_email",
return_value={"email": "@devpulse", "path": str(devpulse_dir)},
),
patch(f"{MOD}.subprocess.Popen", side_effect=FileNotFoundError("drone not found")),
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
# Should not raise — watchdog is optional
def test_relative_devpulse_path_resolved(self, monkeypatch, tmp_path):
"""Relative path from registry is resolved against repo root."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
popen_calls: list[dict] = []
def tracking_popen(cmd, **kwargs):
"""Capture Popen arguments."""
popen_calls.append({"cmd": cmd, **kwargs})
return MagicMock()
from aipass.ai_mail.apps.modules import dispatch as dispatch_mod
real_repo_root = dispatch_mod.Path(__file__).resolve().parents[4]
devpulse_dir = real_repo_root / "src" / "aipass" / "devpulse"
with (
patch(
f"{_H_REG}.get_branch_by_email",
return_value={"email": "@devpulse", "path": "src/aipass/devpulse"},
),
patch(f"{MOD}.subprocess.Popen", side_effect=tracking_popen),
):
from aipass.ai_mail.apps.modules.dispatch import _spawn_watchdog
_spawn_watchdog("@flow")
if devpulse_dir.is_dir():
assert len(popen_calls) == 1
assert "devpulse" in popen_calls[0]["cwd"]
else:
assert len(popen_calls) == 0
class TestDispatchSendWatchdogIntegration:
"""Tests for watchdog integration in _orchestrate_dispatch_send."""
def test_watchdog_spawned_after_successful_wake(self, monkeypatch):
"""Watchdog is spawned after successful send + wake."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
watchdog_calls: list[str] = []
monkeypatch.setattr(
f"{MOD}._spawn_watchdog",
lambda target: watchdog_calls.append(target),
)
patches = _send_patches()
with patches:
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_dispatch_send
@@ -1144,21 +988,17 @@ class TestDispatchSendWatchdogIntegration:
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is True
assert watchdog_calls == ["@target"]
combined = " ".join(printed)
assert "Wake-back enabled" in combined
assert "Watchdog armed" not in combined
def test_watchdog_not_spawned_on_wake_failure(self, monkeypatch):
"""Watchdog is NOT spawned when wake fails."""
def test_no_wake_back_message_on_wake_failure(self, monkeypatch):
"""No wake-back message when wake fails."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
watchdog_calls: list[str] = []
monkeypatch.setattr(
f"{MOD}._spawn_watchdog",
lambda target: watchdog_calls.append(target),
)
mock_status = MagicMock()
mock_status.format.return_value = "WAKE FAILED"
patches = _send_patches(
@@ -1171,19 +1011,14 @@ class TestDispatchSendWatchdogIntegration:
_orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert watchdog_calls == []
combined = " ".join(printed)
assert "Wake-back enabled" not in combined
def test_no_watchdog_flag_skips_spawn(self, monkeypatch):
"""--no-watchdog flag prevents watchdog spawn."""
def test_no_watchdog_flag_still_accepted(self, monkeypatch):
"""--no-watchdog flag is consumed without error (backward compat)."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
watchdog_calls: list[str] = []
monkeypatch.setattr(
f"{MOD}._spawn_watchdog",
lambda target: watchdog_calls.append(target),
)
patches = _send_patches()
with patches:
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_dispatch_send
@@ -1191,21 +1026,14 @@ class TestDispatchSendWatchdogIntegration:
result = _orchestrate_dispatch_send(["@target", "Subject", "Body", "--no-watchdog"])
assert result is True
assert watchdog_calls == []
def test_watchdog_not_spawned_on_send_failure(self, monkeypatch):
"""Watchdog is NOT spawned when send fails."""
def test_no_watchdog_message_on_send_failure(self, monkeypatch):
"""No wake-back message when send fails."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
watchdog_calls: list[str] = []
monkeypatch.setattr(
f"{MOD}._spawn_watchdog",
lambda target: watchdog_calls.append(target),
)
patches = _send_patches(
{
f"{_H_SEND}.send_to_single": MagicMock(return_value=(False, "error")),
@@ -1216,4 +1044,5 @@ class TestDispatchSendWatchdogIntegration:
_orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert watchdog_calls == []
combined = " ".join(printed)
assert "Wake-back enabled" not in combined
src/aipass/ai_mail/tests/test_dispatch_monitor.py
+424
View File
@@ -19,15 +19,18 @@ from unittest.mock import MagicMock
import aipass.ai_mail.apps.handlers.dispatch.dispatch_monitor as mod
from aipass.ai_mail.apps.handlers.dispatch.dispatch_monitor import (
MAX_WAKE_DEPTH,
_check_jsonl_activity,
_check_rate_limited,
_get_jsonl_projects_dir,
_is_sandbox_enabled,
_kill_process,
_log_wake_result,
_make_fresh_cmd,
_run_with_startup_check,
_send_bounce,
_snapshot_jsonl_sizes,
_wake_sender,
_wrap_for_sandbox,
main,
)
@@ -52,6 +55,13 @@ def _suppress_logger(monkeypatch):
monkeypatch.setattr(mod, "logger", MagicMock())
@pytest.fixture(autouse=True)
def _suppress_wake(monkeypatch):
"""Prevent _wake_sender from importing/calling real wake_branch in unrelated tests."""
monkeypatch.setattr(mod, "_wake_sender", MagicMock(return_value="skipped_sender"))
monkeypatch.setattr(mod, "_log_wake_result", MagicMock())
@pytest.fixture
def stderr_log(tmp_path):
"""Create a stderr log file and return its path string."""
@@ -1824,3 +1834,417 @@ finally:
finally:
broker.stop()
t.join(timeout=3)
# === Wake-back tests (TDPLAN-0012) ==========================================
class TestWakeSender:
"""_wake_sender guards and owner-allowlist dispatch."""
@pytest.fixture(autouse=True)
def _mock_is_owner(self, monkeypatch):
"""Default: is_owner returns False (non-owner). Tests override as needed."""
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=False),
)
def test_skips_empty_sender(self, monkeypatch):
"""Empty sender returns skipped_sender."""
monkeypatch.setattr(mod, "logger", MagicMock())
result = _wake_sender("", "@target", 0, "/fake/lock")
assert result == "skipped_sender"
def test_skips_whitespace_sender(self, monkeypatch):
"""Whitespace-only sender returns skipped_sender."""
monkeypatch.setattr(mod, "logger", MagicMock())
result = _wake_sender(" ", "@target", 0, "/fake/lock")
assert result == "skipped_sender"
def test_skips_non_owner_sender(self, monkeypatch):
"""Non-owner sender returns skipped_not_owner."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=False),
)
result = _wake_sender("@someagent", "@target", 0, "/fake/lock")
assert result == "skipped_not_owner"
def test_skips_ai_mail_when_not_owner(self, monkeypatch):
"""@ai_mail is not owner — skipped."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=False),
)
result = _wake_sender("@ai_mail", "@target", 0, "/fake/lock")
assert result == "skipped_not_owner"
def test_skips_human_when_not_owner(self, monkeypatch):
"""@human is not owner — skipped."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=False),
)
result = _wake_sender("@human", "@target", 0, "/fake/lock")
assert result == "skipped_not_owner"
def test_owner_passes_guard(self, monkeypatch):
"""Owner sender passes the is_owner guard and reaches wake_branch."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
mock_status = MagicMock()
mock_status.summary = "ok"
mock_wake = MagicMock(return_value=(mock_status, True))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
mock_wake,
)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "success"
mock_wake.assert_called_once()
def test_is_owner_called_with_normalized_sender(self, monkeypatch):
"""is_owner receives normalized @-prefixed lowercase sender."""
monkeypatch.setattr(mod, "logger", MagicMock())
mock_is_owner = MagicMock(return_value=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
mock_is_owner,
)
_wake_sender("DevPulse", "@target", 0, "/fake/lock")
mock_is_owner.assert_called_once_with("@devpulse")
def test_is_owner_import_failure(self, monkeypatch):
"""ImportError from is_owner returns failed."""
monkeypatch.setattr(mod, "logger", MagicMock())
import builtins
real_import = builtins.__import__
def fail_import(name, *args, **kwargs):
if name == "aipass.spawn.apps.handlers.registry":
raise ImportError("no spawn")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", fail_import)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "failed"
def test_depth_cap_blocks(self, monkeypatch):
"""AIPASS_WAKE_DEPTH >= MAX_WAKE_DEPTH returns blocked_depth."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
monkeypatch.setenv("AIPASS_WAKE_DEPTH", str(MAX_WAKE_DEPTH))
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "blocked_depth"
def test_depth_cap_over_max_blocks(self, monkeypatch):
"""Depth above max also blocks."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
monkeypatch.setenv("AIPASS_WAKE_DEPTH", str(MAX_WAKE_DEPTH + 5))
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "blocked_depth"
def test_success_on_wake(self, monkeypatch):
"""Successful wake_branch call returns success."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
mock_status = MagicMock()
mock_status.summary = "ok"
mock_wake = MagicMock(return_value=(mock_status, True))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
mock_wake,
)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "success"
mock_wake.assert_called_once_with("@devpulse", auto=True, sender="@ai_mail")
def test_blocked_locked_on_lock_failure(self, monkeypatch):
"""wake_branch failing with lock-related message returns blocked_locked."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
mock_status = MagicMock()
mock_status.summary = "lock: Active agent (PID 1234)"
mock_wake = MagicMock(return_value=(mock_status, False))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
mock_wake,
)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "blocked_locked"
def test_blocked_occupied_on_interactive(self, monkeypatch):
"""wake_branch failing with occupancy message returns blocked_occupied."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
mock_status = MagicMock()
mock_status.summary = "blocked: Cannot spawn — interactive session running"
mock_wake = MagicMock(return_value=(mock_status, False))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
mock_wake,
)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "blocked_occupied"
def test_failed_on_exception(self, monkeypatch):
"""Exception during wake returns failed."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
MagicMock(side_effect=RuntimeError("broken")),
)
result = _wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert result == "failed"
def test_depth_incremented_before_wake(self, monkeypatch):
"""AIPASS_WAKE_DEPTH is incremented before calling wake_branch."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.setenv("AIPASS_WAKE_DEPTH", "1")
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
captured_depth = []
def capture_wake(*args, **kwargs):
captured_depth.append(os.environ.get("AIPASS_WAKE_DEPTH"))
mock_status = MagicMock()
mock_status.summary = "ok"
return mock_status, True
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
capture_wake,
)
_wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert captured_depth == ["2"]
def test_wake_called_on_failure_exit(self, monkeypatch):
"""Wake fires on non-zero exit code too."""
monkeypatch.setattr(mod, "logger", MagicMock())
monkeypatch.delenv("AIPASS_WAKE_DEPTH", raising=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
MagicMock(return_value=True),
)
mock_status = MagicMock()
mock_status.summary = "ok"
mock_wake = MagicMock(return_value=(mock_status, True))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
mock_wake,
)
result = _wake_sender("@devpulse", "@target", 1, "/fake/lock")
assert result == "success"
mock_wake.assert_called_once()
def test_sender_normalization_for_is_owner(self, monkeypatch):
"""Sender with or without @ prefix is normalized before is_owner call."""
monkeypatch.setattr(mod, "logger", MagicMock())
mock_is_owner = MagicMock(return_value=False)
monkeypatch.setattr(
"aipass.spawn.apps.handlers.registry.is_owner",
mock_is_owner,
)
_wake_sender("devpulse", "@target", 0, "/fake/lock")
_wake_sender("@devpulse", "@target", 0, "/fake/lock")
assert mock_is_owner.call_count == 2
for call in mock_is_owner.call_args_list:
assert call[0][0] == "@devpulse"
class TestLogWakeResult:
"""_log_wake_result writes to dispatch_wake.log."""
def test_creates_log_file(self, tmp_path):
"""Log file created under target's logs/ directory."""
lock = tmp_path / "branch" / ".ai_mail.local" / ".dispatch.lock"
lock.parent.mkdir(parents=True)
lock.write_text("{}", encoding="utf-8")
logs_dir = tmp_path / "branch" / "logs"
_log_wake_result("@target", "@sender", 0, "success", str(lock))
log_file = logs_dir / "dispatch_wake.log"
assert log_file.exists()
content = log_file.read_text(encoding="utf-8")
assert "target=@target" in content
assert "sender=@sender" in content
assert "exit_code=0" in content
assert "wake_result=success" in content
def test_appends_to_existing(self, tmp_path):
"""Subsequent calls append, not overwrite."""
lock = tmp_path / "branch" / ".ai_mail.local" / ".dispatch.lock"
lock.parent.mkdir(parents=True)
lock.write_text("{}", encoding="utf-8")
logs_dir = tmp_path / "branch" / "logs"
logs_dir.mkdir(parents=True)
log_file = logs_dir / "dispatch_wake.log"
log_file.write_text("existing line\n", encoding="utf-8")
_log_wake_result("@target", "@sender", 0, "success", str(lock))
lines = log_file.read_text(encoding="utf-8").strip().split("\n")
assert len(lines) == 2
assert lines[0] == "existing line"
assert "wake_result=success" in lines[1]
def test_all_result_values(self, tmp_path):
"""All result enum values are logged correctly."""
lock = tmp_path / "branch" / ".ai_mail.local" / ".dispatch.lock"
lock.parent.mkdir(parents=True)
lock.write_text("{}", encoding="utf-8")
for result_tag in (
"success",
"blocked_occupied",
"blocked_locked",
"blocked_depth",
"skipped_sender",
"failed",
):
_log_wake_result("@t", "@s", 0, result_tag, str(lock))
log_file = tmp_path / "branch" / "logs" / "dispatch_wake.log"
lines = log_file.read_text(encoding="utf-8").strip().split("\n")
assert len(lines) == 6
class TestWakeBackIntegration:
"""Wake-back wired into main() — fires after lock cleanup on both paths."""
def test_wake_called_on_success(self, monkeypatch, main_argv):
"""_wake_sender called with correct args after successful agent run."""
argv, lock_file, stderr_log = main_argv
wake_calls = []
def track_wake(sender, branch_email, exit_code, lf):
wake_calls.append((sender, branch_email, exit_code))
return "success"
monkeypatch.setattr("sys.argv", argv)
monkeypatch.setattr(mod, "_run_with_startup_check", MagicMock(return_value=(0, False)))
monkeypatch.setattr(mod, "_send_bounce", MagicMock())
monkeypatch.setattr(mod, "_check_rate_limited", MagicMock(return_value=False))
monkeypatch.setattr(mod, "_wake_sender", track_wake)
monkeypatch.setattr(mod, "_log_wake_result", MagicMock())
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.paths.find_repo_root",
MagicMock(return_value=Path("/fake/repo")),
)
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 0
assert len(wake_calls) == 1
assert wake_calls[0] == ("@sender", "@test_branch", 0)
def test_wake_called_on_failure(self, monkeypatch, main_argv):
"""_wake_sender called after all attempts fail (in addition to bounce)."""
argv, lock_file, stderr_log = main_argv
wake_calls = []
mock_bounce = MagicMock()
def track_wake(sender, branch_email, exit_code, lf):
wake_calls.append((sender, branch_email, exit_code))
return "success"
monkeypatch.setattr("sys.argv", argv)
monkeypatch.setattr(mod, "_run_with_startup_check", MagicMock(side_effect=[(1, False), (1, False), (1, False)]))
monkeypatch.setattr(mod, "_send_bounce", mock_bounce)
monkeypatch.setattr(mod, "_check_rate_limited", MagicMock(return_value=False))
monkeypatch.setattr(mod, "_wake_sender", track_wake)
monkeypatch.setattr(mod, "_log_wake_result", MagicMock())
monkeypatch.setattr(
mod,
"time",
MagicMock(time=time.time, strftime=time.strftime, sleep=MagicMock()),
)
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.paths.find_repo_root",
MagicMock(return_value=Path("/fake/repo")),
)
with pytest.raises(SystemExit):
main()
mock_bounce.assert_called_once()
assert len(wake_calls) == 1
assert wake_calls[0][2] != 0
def test_log_wake_result_called(self, monkeypatch, main_argv):
"""_log_wake_result called with wake result after main completes."""
argv, lock_file, stderr_log = main_argv
log_calls = []
monkeypatch.setattr("sys.argv", argv)
monkeypatch.setattr(mod, "_run_with_startup_check", MagicMock(return_value=(0, False)))
monkeypatch.setattr(mod, "_send_bounce", MagicMock())
monkeypatch.setattr(mod, "_check_rate_limited", MagicMock(return_value=False))
monkeypatch.setattr(mod, "_wake_sender", MagicMock(return_value="success"))
monkeypatch.setattr(mod, "_log_wake_result", lambda *a: log_calls.append(a))
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.paths.find_repo_root",
MagicMock(return_value=Path("/fake/repo")),
)
with pytest.raises(SystemExit):
main()
assert len(log_calls) == 1
branch_email, sender, exit_code, result, lf = log_calls[0]
assert branch_email == "@test_branch"
assert sender == "@sender"
assert exit_code == 0
assert result == "success"
src/aipass/hooks/.seedgo/bypass.json
+45
View File
@@ -166,6 +166,51 @@
"standard": "open_encoding",
"reason": "NamedTemporaryFile creates binary wav for Piper TTS \u2014 encoding not applicable to binary audio."
},
{
"file": "apps/handlers/security/registry_gate.py",
"standard": "dead_code",
"reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.security.registry_gate.handle' — not statically imported by design. Wired in PreToolUse.registry_gate."
},
{
"file": "apps/handlers/security/registry_gate.py",
"standard": "unused_function",
"reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Wired in PreToolUse.registry_gate."
},
{
"file": "apps/handlers/security/registry_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_registry_gate.py",
"standard": "architecture",
"reason": "Test files live in tests/, not in the 3-layer apps structure."
},
{
"file": "tests/test_registry_gate.py",
"standard": "documentation",
"reason": "Test methods use descriptive names as documentation per pytest convention."
},
{
"file": "tests/test_registry_gate.py",
"standard": "encapsulation",
"reason": "Tests import handlers directly to test implementation details."
},
{
"file": "tests/test_registry_gate.py",
"standard": "meta",
"reason": "Test files do not need Version/Modified metadata headers."
},
{
"file": "tests/test_registry_gate.py",
"standard": "hardcoded_path",
"reason": "Test fixture CWD string and bash command strings contain paths as test input data — not real filesystem operations."
},
{
"file": "tests/test_registry_gate.py",
"standard": "windows_compat",
"reason": "Test fixture bash command strings contain /tmp paths as scanner input — not real filesystem operations. The handler itself is platform-agnostic (string scanning only)."
},
{
"file": "apps/handlers/lifecycle/auto_fix.py",
"standard": "dead_code",
src/aipass/hooks/README.md
+3 -2
View File
@@ -69,6 +69,7 @@ src/aipass/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
│ │ │ ├── registry_gate.py # Seals *_REGISTRY.json — blocks raw writes/edits/deletes, redirects to drone @spawn
│ │ │ ├── rm_gate.py # Guardrail — catches accidental rm -rf, teaches drone rm
│ │ │ └── subagent_gate.py # Blocks sub-agent stop until clean
│ │ ├── lifecycle/ # Session management hooks
@@ -88,7 +89,7 @@ src/aipass/hooks/
│ └── diagnostics.py # JSONL logging for hook execution
├── logs/
│ └── engine.jsonl # JSONL diagnostics (every hook execution)
└── tests/ # 825 tests across 27 test files
└── tests/ # 913 tests across 28 test files
```
## How It Works
@@ -110,7 +111,7 @@ Handlers are called **dynamically at runtime** — the engine uses `importlib.im
| Event | Hooks | Description |
|---|---|---|
| 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 |
| PreToolUse | tool_sound, edit_gate, git_gate, rm_gate, registry_gate | Security gates + guardrails + sound |
| PostToolUse | auto_fix, auto_watchdog | Diagnostics + watchdog |
| SubagentStop | subagent_gate | Seedgo validation |
| Stop | stop_sound, telegram_response, presence_release | Bell + Telegram delivery + presence release |
src/aipass/hooks/apps/handlers/security/registry_gate.py
+166
View File
@@ -0,0 +1,166 @@
# =================== AIPass ====================
# Name: registry_gate.py
# Version: 1.0.0
# Description: Blocks raw writes, edits, and deletions of *_REGISTRY.json (PreToolUse)
# Branch: hooks
# Layer: apps/handlers/security
# Created: 2026-07-10
# Modified: 2026-07-10
# =============================================
"""Blocks raw writes, edits, and deletions of *_REGISTRY.json files.
Sealed-authority enforcement: the registry is the single source of truth
for project ownership. Only drone @spawn (tier-gated) may write it.
"""
import json
import re
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
REGISTRY_RE = re.compile(r"\w+_REGISTRY\.json$")
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
REGISTRY_REDIRECT = (
"{file} is a sealed registry — direct writes are blocked.\nUse drone @spawn to manage registry entries."
)
_BLOCK_ALLOW = {"stdout": "", "exit_code": 0}
_REDIRECT_RE = re.compile(r">{1,2}\s*\S*_REGISTRY\.json\b")
_TEE_RE = re.compile(r"\btee\b[^&;|]*\S*_REGISTRY\.json\b")
_SED_I_RE = re.compile(r"\bsed\b\s[^&;|]*-i[^&;|]*\S*_REGISTRY\.json\b")
def _block(reason: str) -> dict:
return {"stdout": json.dumps({"decision": "block", "reason": reason}), "exit_code": 2, "sound": "registry gate"}
def _is_registry_file(name: str) -> bool:
return bool(REGISTRY_RE.search(Path(name).name))
def _strip_quotes(cmd: str) -> str:
cmd = re.sub(r'"(?:[^"\\]|\\.)*"', '""', cmd)
cmd = re.sub(r"'(?:[^'\\]|\\.)*'", "''", cmd)
return cmd
def _split_clauses(cmd: str) -> list[str]:
parts = re.split(r"&&|\|\||[;|]", cmd)
clauses: list[str] = []
for part in parts:
clauses.extend(re.split(r"[$()`]", part))
return clauses
def _is_drone_spawn(clause: str) -> bool:
stripped = clause.strip()
return stripped.startswith("drone @spawn") or stripped.startswith("drone spawn")
def _clause_targets_registry(clause: str) -> bool:
if _is_drone_spawn(clause):
return False
if _REDIRECT_RE.search(clause):
return True
if _TEE_RE.search(clause):
return True
if _SED_I_RE.search(clause):
return True
tokens = clause.split()
if not tokens:
return False
for i, tok in enumerate(tokens):
if tok in ("mv",) or tok.endswith("/mv"):
if i > 0 and tokens[i - 1] == "drone":
continue
remaining = tokens[i + 1 :]
args = [t for t in remaining if not t.startswith("-")]
for arg in args:
if _is_registry_file(arg):
return True
if tok in ("cp",) or tok.endswith("/cp"):
if i > 0 and tokens[i - 1] == "drone":
continue
remaining = tokens[i + 1 :]
args = [t for t in remaining if not t.startswith("-")]
if len(args) >= 2 and _is_registry_file(args[-1]):
return True
if tok in ("rm", "unlink") or tok.endswith("/rm"):
if i > 0 and tokens[i - 1] == "drone":
continue
remaining = tokens[i + 1 :]
for arg in remaining:
if arg.startswith("-"):
continue
if _is_registry_file(arg):
return True
return False
def _find_registry_name(text: str) -> str:
match = REGISTRY_RE.search(text)
return match.group(0) if match else "*_REGISTRY.json"
def _check_bash(tool_input: dict) -> dict:
cmd = tool_input.get("command", "")
if not cmd:
return _BLOCK_ALLOW
if "_REGISTRY.json" not in cmd:
return _BLOCK_ALLOW
scan = _strip_quotes(cmd)
if "_REGISTRY.json" not in scan:
return _BLOCK_ALLOW
for clause in _split_clauses(scan):
if _clause_targets_registry(clause):
return _block(REGISTRY_REDIRECT.format(file=_find_registry_name(clause)))
return _BLOCK_ALLOW
def _check_edit(tool_input: dict) -> dict:
file_path = tool_input.get("file_path") or tool_input.get("notebook_path") or ""
if not file_path:
return _BLOCK_ALLOW
if _is_registry_file(file_path):
return _block(REGISTRY_REDIRECT.format(file=Path(file_path).name))
return _BLOCK_ALLOW
def handle(hook_data: dict) -> dict:
"""Block raw writes, edits, and deletions of *_REGISTRY.json files.
Args:
hook_data: Parsed hook event dict from engine.
Returns:
Result dict with stdout (block JSON or empty) and exit_code.
"""
try:
tool_name = hook_data.get("tool_name", "")
tool_input = hook_data.get("tool_input", {})
if tool_name == "Bash":
return _check_bash(tool_input)
if tool_name in EDIT_TOOLS:
return _check_edit(tool_input)
return _BLOCK_ALLOW
except Exception as exc:
logger.info("[HOOKS] registry_gate: unexpected error (allowing): %s", exc)
return _BLOCK_ALLOW
src/aipass/hooks/tests/test_registry_gate.py
+337
View File
@@ -0,0 +1,337 @@
# =================== AIPass ====================
# Name: test_registry_gate.py
# Version: 1.0.0
# Description: Tests for registry_gate security handler
# Branch: hooks
# Created: 2026-07-10
# Modified: 2026-07-10
# =============================================
"""Tests for handlers/security/registry_gate.py."""
import json
from unittest.mock import patch
from aipass.hooks.apps.handlers.security.registry_gate import (
_clause_targets_registry,
_find_registry_name,
_is_drone_spawn,
_is_registry_file,
_split_clauses,
_strip_quotes,
handle,
)
class TestIsRegistryFile:
def test_aipass_registry(self):
assert _is_registry_file("AIPASS_REGISTRY.json") is True
def test_vera_registry(self):
assert _is_registry_file("VERA_REGISTRY.json") is True
def test_full_path(self):
assert _is_registry_file("/tmp/projects/AIPass/AIPASS_REGISTRY.json") is True
def test_not_registry(self):
assert _is_registry_file("config.json") is False
def test_partial_match(self):
assert _is_registry_file("REGISTRY.json") is False
def test_wrong_extension(self):
assert _is_registry_file("AIPASS_REGISTRY.yaml") is False
def test_registry_in_path(self):
assert _is_registry_file("/path/to/FOO_REGISTRY.json") is True
class TestIsDroneSpawn:
def test_drone_at_spawn(self):
assert _is_drone_spawn("drone @spawn register") is True
def test_drone_spawn(self):
assert _is_drone_spawn("drone spawn register") is True
def test_leading_space(self):
assert _is_drone_spawn(" drone @spawn list") is True
def test_not_drone(self):
assert _is_drone_spawn("echo drone @spawn") is False
def test_empty(self):
assert _is_drone_spawn("") is False
class TestStripQuotes:
def test_double_quotes(self):
assert _strip_quotes('echo "AIPASS_REGISTRY.json"') == 'echo ""'
def test_single_quotes(self):
assert _strip_quotes("echo 'AIPASS_REGISTRY.json'") == "echo ''"
def test_no_quotes(self):
assert _strip_quotes("rm AIPASS_REGISTRY.json") == "rm AIPASS_REGISTRY.json"
class TestSplitClauses:
def test_and_operator(self):
clauses = _split_clauses("echo hi && rm AIPASS_REGISTRY.json")
assert any("AIPASS_REGISTRY" in c for c in clauses)
def test_semicolon(self):
clauses = _split_clauses("echo hi; rm AIPASS_REGISTRY.json")
assert any("AIPASS_REGISTRY" in c for c in clauses)
def test_pipe(self):
clauses = _split_clauses("cat foo | tee AIPASS_REGISTRY.json")
assert any("tee" in c for c in clauses)
def test_subshell(self):
clauses = _split_clauses("echo $(cat AIPASS_REGISTRY.json)")
assert any("AIPASS_REGISTRY" in c for c in clauses)
class TestClauseTargetsRegistry:
def test_redirect_overwrite(self):
assert _clause_targets_registry("echo data > AIPASS_REGISTRY.json") is True
def test_redirect_append(self):
assert _clause_targets_registry("echo data >> AIPASS_REGISTRY.json") is True
def test_redirect_with_path(self):
assert _clause_targets_registry("echo data > /path/to/AIPASS_REGISTRY.json") is True
def test_tee(self):
assert _clause_targets_registry(" tee AIPASS_REGISTRY.json") is True
def test_tee_append(self):
assert _clause_targets_registry(" tee -a AIPASS_REGISTRY.json") is True
def test_sed_inplace(self):
assert _clause_targets_registry("sed -i 's/foo/bar/' AIPASS_REGISTRY.json") is True
def test_sed_inplace_backup(self):
assert _clause_targets_registry("sed -i.bak 's/foo/bar/' AIPASS_REGISTRY.json") is True
def test_mv_onto_registry(self):
assert _clause_targets_registry("mv temp.json AIPASS_REGISTRY.json") is True
def test_mv_registry_away(self):
assert _clause_targets_registry("mv AIPASS_REGISTRY.json backup.json") is True
def test_mv_with_flag(self):
assert _clause_targets_registry("mv -f temp.json AIPASS_REGISTRY.json") is True
def test_cp_onto_registry(self):
assert _clause_targets_registry("cp temp.json AIPASS_REGISTRY.json") is True
def test_cp_from_registry_allowed(self):
assert _clause_targets_registry("cp AIPASS_REGISTRY.json backup.json") is False
def test_rm_registry(self):
assert _clause_targets_registry("rm AIPASS_REGISTRY.json") is True
def test_rm_with_flag(self):
assert _clause_targets_registry("rm -f AIPASS_REGISTRY.json") is True
def test_unlink_registry(self):
assert _clause_targets_registry("unlink AIPASS_REGISTRY.json") is True
def test_absolute_path_rm(self):
assert _clause_targets_registry("/usr/bin/rm AIPASS_REGISTRY.json") is True
def test_drone_spawn_allowed(self):
assert _clause_targets_registry("drone @spawn register AIPASS_REGISTRY.json") is False
def test_drone_spawn_no_at_allowed(self):
assert _clause_targets_registry("drone spawn update AIPASS_REGISTRY.json") is False
def test_cat_allowed(self):
assert _clause_targets_registry("cat AIPASS_REGISTRY.json") is False
def test_jq_read_allowed(self):
assert _clause_targets_registry("jq '.branches' AIPASS_REGISTRY.json") is False
def test_head_allowed(self):
assert _clause_targets_registry("head -5 AIPASS_REGISTRY.json") is False
def test_empty_clause(self):
assert _clause_targets_registry("") is False
def test_no_registry_mention(self):
assert _clause_targets_registry("rm some_file.txt") is False
def test_vera_registry(self):
assert _clause_targets_registry("rm VERA_REGISTRY.json") is True
class TestFindRegistryName:
def test_finds_aipass(self):
assert _find_registry_name("rm AIPASS_REGISTRY.json") == "AIPASS_REGISTRY.json"
def test_finds_vera(self):
assert _find_registry_name("rm VERA_REGISTRY.json") == "VERA_REGISTRY.json"
def test_fallback(self):
assert _find_registry_name("no match here") == "*_REGISTRY.json"
class TestHandleBash:
CWD = "/home/patrick/Projects/AIPass/src/aipass/hooks"
def _bash(self, command: str) -> dict:
return handle({"tool_name": "Bash", "tool_input": {"command": command}, "cwd": self.CWD})
def _assert_blocked(self, result: dict):
assert result["exit_code"] == 2
parsed = json.loads(result["stdout"])
assert parsed["decision"] == "block"
assert "drone @spawn" in parsed["reason"]
def _assert_allowed(self, result: dict):
assert result["exit_code"] == 0
assert result["stdout"] == ""
def test_block_redirect_overwrite(self):
self._assert_blocked(self._bash("echo '{}' > AIPASS_REGISTRY.json"))
def test_block_redirect_append(self):
self._assert_blocked(self._bash("echo data >> AIPASS_REGISTRY.json"))
def test_block_tee(self):
self._assert_blocked(self._bash("echo data | tee AIPASS_REGISTRY.json"))
def test_block_sed_inplace(self):
self._assert_blocked(self._bash("sed -i 's/old/new/' AIPASS_REGISTRY.json"))
def test_block_mv_onto(self):
self._assert_blocked(self._bash("mv temp.json AIPASS_REGISTRY.json"))
def test_block_mv_away(self):
self._assert_blocked(self._bash("mv AIPASS_REGISTRY.json /tmp/backup.json"))
def test_block_cp_onto(self):
self._assert_blocked(self._bash("cp temp.json AIPASS_REGISTRY.json"))
def test_block_rm(self):
self._assert_blocked(self._bash("rm AIPASS_REGISTRY.json"))
def test_block_rm_force(self):
self._assert_blocked(self._bash("rm -f AIPASS_REGISTRY.json"))
def test_block_unlink(self):
self._assert_blocked(self._bash("unlink AIPASS_REGISTRY.json"))
def test_block_compound_rm(self):
self._assert_blocked(self._bash("echo done && rm AIPASS_REGISTRY.json"))
def test_block_subshell_rm(self):
self._assert_blocked(self._bash("echo $(rm AIPASS_REGISTRY.json)"))
def test_block_vera_registry(self):
self._assert_blocked(self._bash("rm VERA_REGISTRY.json"))
def test_allow_drone_spawn(self):
self._assert_allowed(self._bash("drone @spawn register --project ."))
def test_allow_drone_spawn_with_registry(self):
self._assert_allowed(self._bash("drone @spawn update AIPASS_REGISTRY.json"))
def test_allow_cat(self):
self._assert_allowed(self._bash("cat AIPASS_REGISTRY.json"))
def test_allow_jq_read(self):
self._assert_allowed(self._bash("jq '.branches' AIPASS_REGISTRY.json"))
def test_allow_grep(self):
self._assert_allowed(self._bash("grep owner AIPASS_REGISTRY.json"))
def test_allow_cp_from_registry(self):
self._assert_allowed(self._bash("cp AIPASS_REGISTRY.json /tmp/backup.json"))
def test_allow_head(self):
self._assert_allowed(self._bash("head -5 AIPASS_REGISTRY.json"))
def test_allow_no_registry(self):
self._assert_allowed(self._bash("echo hello"))
def test_allow_registry_in_quotes(self):
self._assert_allowed(self._bash('echo "modifying AIPASS_REGISTRY.json"'))
def test_empty_command(self):
self._assert_allowed(self._bash(""))
def test_no_tool_input(self):
result = handle({"tool_name": "Bash"})
assert result["exit_code"] == 0
def test_empty_hook_data(self):
result = handle({})
assert result["exit_code"] == 0
def test_sound_key_on_block(self):
result = self._bash("rm AIPASS_REGISTRY.json")
assert result.get("sound") == "registry gate"
class TestHandleEditTools:
CWD = "/home/patrick/Projects/AIPass/src/aipass/hooks"
def _edit(self, tool_name: str, file_path: str) -> dict:
return handle({"tool_name": tool_name, "tool_input": {"file_path": file_path}, "cwd": self.CWD})
def _assert_blocked(self, result: dict):
assert result["exit_code"] == 2
parsed = json.loads(result["stdout"])
assert parsed["decision"] == "block"
assert "drone @spawn" in parsed["reason"]
def _assert_allowed(self, result: dict):
assert result["exit_code"] == 0
assert result["stdout"] == ""
def test_block_edit(self):
self._assert_blocked(self._edit("Edit", "/path/AIPASS_REGISTRY.json"))
def test_block_write(self):
self._assert_blocked(self._edit("Write", "/path/AIPASS_REGISTRY.json"))
def test_block_multi_edit(self):
self._assert_blocked(self._edit("MultiEdit", "/path/AIPASS_REGISTRY.json"))
def test_block_notebook_edit(self):
self._assert_blocked(
handle(
{
"tool_name": "NotebookEdit",
"tool_input": {"notebook_path": "/path/AIPASS_REGISTRY.json"},
"cwd": self.CWD,
}
)
)
def test_block_vera_registry(self):
self._assert_blocked(self._edit("Edit", "/path/VERA_REGISTRY.json"))
def test_allow_normal_file(self):
self._assert_allowed(self._edit("Edit", "/path/config.json"))
def test_allow_empty_path(self):
self._assert_allowed(self._edit("Edit", ""))
def test_allow_read_tool(self):
result = handle({"tool_name": "Read", "tool_input": {"file_path": "/path/AIPASS_REGISTRY.json"}})
self._assert_allowed(result)
def test_sound_key_on_block(self):
result = self._edit("Edit", "/path/AIPASS_REGISTRY.json")
assert result.get("sound") == "registry gate"
class TestHandleExceptionSafety:
@patch("aipass.hooks.apps.handlers.security.registry_gate.logger")
def test_exception_allows(self, mock_logger):
result = handle({"tool_name": "Bash", "tool_input": None, "cwd": "/tmp"})
assert result["exit_code"] == 0
mock_logger.info.assert_called()
src/aipass/spawn/.seedgo/bypass.json
+20
View File
@@ -307,6 +307,26 @@
"file": "apps/modules/core.py",
"standard": "encapsulation",
"reason": "Local import of render_all_meta_tabs from memory.apps.handlers.tracking.tab_renderer — @memory has no module entry point for this API. Cross-branch contract per FPLAN-0286."
},
{
"file": "apps/handlers/registry.py",
"standard": "unused_function",
"reason": "backfill_owner_and_registry_id() is a migration function for TDPLAN-0012 — called manually to backfill existing registries. is_owner() is part of the frozen interface contract (TDPLAN-0012) — consumed by @ai_mail and @devpulse via direct import."
},
{
"file": "tests/test_owner_resolver.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_owner_resolver.py",
"standard": "encapsulation",
"reason": "Test file — tests must import handlers directly to test them in isolation. Handler imports inside test methods are intentional."
},
{
"file": "tests/test_owner_resolver.py",
"standard": "documentation",
"reason": "Test file — pytest test_* functions are self-documenting via descriptive names; docstrings not required on individual test cases."
}
],
"notes": {
src/aipass/spawn/apps/handlers/registry.py
+109 -19
View File
@@ -136,7 +136,7 @@ def _validate_path_containment(branch_path, registry_path):
return False
def add_to_registry(registry_path, branch_name, branch_path, profile, email, purpose=""):
def add_to_registry(registry_path, branch_name, branch_path, profile, email, purpose="", registry_id=""):
"""
Add a new branch entry to the registry.
@@ -194,6 +194,8 @@ def add_to_registry(registry_path, branch_name, branch_path, profile, email, pur
"created": today,
"last_active": today,
}
if registry_id:
entry["registry_id"] = registry_id
if isinstance(branches, dict):
branches[branch_name] = entry
@@ -206,7 +208,7 @@ def add_to_registry(registry_path, branch_name, branch_path, profile, email, pur
return save_registry(registry_path, registry)
finally:
if lock_fd is not None:
if lock_fd is not None and sys.platform != "win32":
import fcntl
fcntl.flock(lock_fd, fcntl.LOCK_UN)
@@ -268,7 +270,59 @@ def fix_passport_registry_id(branch_dir: Path, registry_path: Path) -> bool:
def ensure_project_has_owner(registry_path):
"""If no agent in the project has owner:true, assign it to the earliest-created agent."""
"""Ensure exactly one branch entry in the registry has owner:true.
Owner is determined by citizen_class=manager (read from passport).
Falls back to citizen_number==1 if no manager found.
Writes to the REGISTRY ENTRY (sealed authority), not the passport.
"""
registry_path = Path(registry_path)
reg_data = load_registry(registry_path)
branches = branches_as_list(reg_data.get("branches", []))
if not branches:
return False
for branch in branches:
if branch.get("owner") is True:
return False
registry_root = registry_path.parent
owner_branch = None
for branch in branches:
branch_path = registry_root / branch.get("path", "")
passport_path = branch_path / ".trinity" / "passport.json"
if passport_path.exists():
passport = json_handler.read_json(passport_path)
if passport and passport.get("identity", {}).get("citizen_class") == "manager":
owner_branch = branch
break
if owner_branch is None:
for branch in branches:
branch_path = registry_root / branch.get("path", "")
passport_path = branch_path / ".trinity" / "passport.json"
if passport_path.exists():
passport = json_handler.read_json(passport_path)
if passport and passport.get("citizenship", {}).get("owner") is True:
owner_branch = branch
break
if owner_branch is None:
return False
owner_branch["owner"] = True
save_registry(registry_path, reg_data)
logger.info("[registry] Set owner=true on %s (registry entry)", owner_branch.get("name", "?"))
return True
def backfill_owner_and_registry_id(registry_path):
"""Backfill owner and registry_id fields into all registry branch entries.
- Sets registry_id from each branch's passport citizenship.registry_id
- Sets owner:true on the manager branch (devpulse in AIPass)
"""
registry_path = Path(registry_path)
reg_data = load_registry(registry_path)
branches = branches_as_list(reg_data.get("branches", []))
@@ -276,23 +330,59 @@ def ensure_project_has_owner(registry_path):
return False
registry_root = registry_path.parent
changed = False
for branch in branches:
branch_path = registry_root / branch.get("path", "")
passport_path = branch_path / ".trinity" / "passport.json"
if passport_path.exists():
passport = json_handler.read_json(passport_path)
if passport and passport.get("citizenship", {}).get("owner") is True:
return False
if not passport_path.exists():
continue
passport = json_handler.read_json(passport_path)
if not passport:
continue
by_created = sorted(branches, key=lambda b: b.get("created", "9999-99-99"))
for branch in by_created:
branch_path = registry_root / branch.get("path", "")
passport_path = branch_path / ".trinity" / "passport.json"
if passport_path.exists():
passport = json_handler.read_json(passport_path)
if passport:
passport.setdefault("citizenship", {})["owner"] = True
json_handler.write_json(passport_path, passport)
logger.info("[registry] Retroactively set owner=true on %s", branch.get("name", "?"))
return True
return False
rid = passport.get("citizenship", {}).get("registry_id", "")
if rid and "registry_id" not in branch:
branch["registry_id"] = rid
changed = True
citizen_class = passport.get("identity", {}).get("citizen_class", "")
if citizen_class == "manager" and not branch.get("owner"):
branch["owner"] = True
changed = True
if changed:
save_registry(registry_path, reg_data)
logger.info("[registry] Backfilled owner + registry_id into registry entries")
return changed
def get_owner(start_path=None):
"""Return the branch entry dict whose owner==true, or None.
Walks up from start_path (default CWD) to find *_REGISTRY.json.
"""
registry_path = find_registry(start_path=start_path)
if not registry_path.exists():
return None
reg_data = load_registry(registry_path)
for branch in branches_as_list(reg_data.get("branches", [])):
if branch.get("owner") is True:
return branch
return None
def is_owner(email, start_path=None):
"""True iff email matches the owner entry's email.
Normalizes email — tolerates with/without leading '@'.
"""
if not email:
return False
normalized = email if email.startswith("@") else f"@{email}"
owner = get_owner(start_path=start_path)
if owner is None:
return False
owner_email = owner.get("email", "")
owner_normalized = owner_email if owner_email.startswith("@") else f"@{owner_email}"
return normalized == owner_normalized
src/aipass/spawn/apps/modules/core.py
+3
View File
@@ -283,11 +283,13 @@ def _spawn_agent(
# Step 2b: Set owner field — first agent in the project is the owner
passport_path = target / ".trinity" / "passport.json"
passport_registry_id = ""
if passport_path.exists():
passport_data = json_handler.read_json(passport_path)
if passport_data:
passport_data.setdefault("citizenship", {})["owner"] = citizen_number == 1
json_handler.write_json(passport_path, passport_data)
passport_registry_id = passport_data.get("citizenship", {}).get("registry_id", "")
# Step 3: Regenerate .template_registry.json with fresh hashes
regenerate_template_registry(target)
@@ -312,6 +314,7 @@ def _spawn_agent(
detected_profile,
f"@{branch_lower}",
purpose or "New agent - purpose TBD",
registry_id=passport_registry_id,
)
# Step 5: Ensure at least one agent in the project is the owner
src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json
+1 -1
View File
@@ -429,7 +429,7 @@
},
"metadata": {
"description": "Template file tracking registry for ID-based updates",
"last_updated": "2026-07-04",
"last_updated": "2026-07-10",
"version": "1.0.0"
}
}
src/aipass/spawn/tests/test_citizen_classes.py
+30 -20
View File
@@ -383,31 +383,30 @@ class TestRetroactiveOwner:
"""Tests for retroactive owner assignment on legacy projects."""
def test_retroactive_owner_on_legacy_agents(self, tmp_path):
"""Creating a new agent in a project where no agent has owner sets the alphabetically first."""
"""ensure_project_has_owner picks manager passport, falls back to passport owner flag."""
from aipass.spawn.apps.modules.core import _spawn_agent
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
# Names chosen so legacy agents sort first alphabetically (alpha < beta < zeta)
_spawn_agent(str(tmp_path / "alpha"), registry_path=str(reg))
_spawn_agent(str(tmp_path / "beta"), registry_path=str(reg))
for name in ["alpha", "beta"]:
pp = tmp_path / name / ".trinity" / "passport.json"
data = json.loads(pp.read_text())
del data["citizenship"]["owner"]
pp.write_text(json.dumps(data, indent=2))
# Strip owner from registry entries to simulate legacy state (no sealed owner)
# Keep alpha's passport owner=True as the fallback signal
reg_data = json.loads(reg.read_text())
for b in reg_data["branches"]:
b.pop("owner", None)
reg.write_text(json.dumps(reg_data, indent=2))
# Create a third agent — triggers retroactive fix on alphabetically first
_spawn_agent(str(tmp_path / "zeta"), registry_path=str(reg))
pa = json.loads((tmp_path / "alpha" / ".trinity" / "passport.json").read_text())
pb = json.loads((tmp_path / "beta" / ".trinity" / "passport.json").read_text())
pz = json.loads((tmp_path / "zeta" / ".trinity" / "passport.json").read_text())
assert pa["citizenship"]["owner"] is True
assert pb["citizenship"].get("owner") is not True
assert pz["citizenship"]["owner"] is False
# Owner now lives in the registry entry — alpha picked via passport fallback
reg_data = json.loads(reg.read_text())
entries = {b["name"]: b for b in reg_data["branches"]}
assert entries["ALPHA"].get("owner") is True
assert entries["BETA"].get("owner") is not True
assert entries["ZETA"].get("owner") is not True
def test_no_retroactive_if_owner_exists(self, tmp_path):
"""If an existing agent already has owner:true, no retroactive change."""
@@ -428,7 +427,7 @@ class TestRetroactiveOwner:
assert p3["citizenship"]["owner"] is False
def test_ensure_project_has_owner_direct(self, tmp_path):
"""Direct call to ensure_project_has_owner fixes a legacy project."""
"""Direct call to ensure_project_has_owner sets owner in registry entry."""
from aipass.spawn.apps.handlers.registry import ensure_project_has_owner
from aipass.spawn.apps.modules.core import _spawn_agent
@@ -438,17 +437,28 @@ class TestRetroactiveOwner:
_spawn_agent(str(tmp_path / "agent_x"), registry_path=str(reg))
_spawn_agent(str(tmp_path / "agent_y"), registry_path=str(reg))
# Strip owner from both
# Strip owner from passports AND registry entries
for name in ["agent_x", "agent_y"]:
pp = tmp_path / name / ".trinity" / "passport.json"
data = json.loads(pp.read_text())
data["citizenship"].pop("owner", None)
pp.write_text(json.dumps(data, indent=2))
reg_data = json.loads(reg.read_text())
for b in reg_data["branches"]:
b.pop("owner", None)
reg.write_text(json.dumps(reg_data, indent=2))
# Re-add passport owner on agent_x to simulate fallback signal
px_pp = tmp_path / "agent_x" / ".trinity" / "passport.json"
px_data = json.loads(px_pp.read_text())
px_data["citizenship"]["owner"] = True
px_pp.write_text(json.dumps(px_data, indent=2))
result = ensure_project_has_owner(reg)
assert result is True
px = json.loads((tmp_path / "agent_x" / ".trinity" / "passport.json").read_text())
py = json.loads((tmp_path / "agent_y" / ".trinity" / "passport.json").read_text())
assert px["citizenship"]["owner"] is True
assert py["citizenship"].get("owner") is not True
# Owner is now in the registry entry
reg_data = json.loads(reg.read_text())
entries = {b["name"]: b for b in reg_data["branches"]}
assert entries["AGENT_X"].get("owner") is True
assert entries["AGENT_Y"].get("owner") is not True
src/aipass/spawn/tests/test_owner_resolver.py
+472
View File
@@ -0,0 +1,472 @@
# =================== META ====================
# Name: test_owner_resolver.py
# Description: Tests for owner resolver and registry authority
# Version: 1.0.0
# Created: 2026-07-10
# Modified: 2026-07-10
# =============================================
"""Tests for owner resolver: get_owner, is_owner, ensure_project_has_owner, backfill."""
import json
import pytest
from unittest.mock import patch
@pytest.fixture
def registry_with_owner(tmp_path):
"""Create a registry file with one owner branch."""
reg = tmp_path / "AIPASS_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 3},
"branches": [
{
"name": "alpha",
"path": "src/alpha",
"email": "@alpha",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-01",
"last_active": "2026-01-01",
},
{
"name": "devpulse",
"path": "src/devpulse",
"email": "@devpulse",
"status": "active",
"profile": "library",
"description": "orchestrator",
"created": "2026-01-02",
"last_active": "2026-01-02",
"owner": True,
"registry_id": "abc-123",
},
{
"name": "gamma",
"path": "src/gamma",
"email": "@gamma",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-03",
"last_active": "2026-01-03",
},
],
}
),
encoding="utf-8",
)
return reg
@pytest.fixture
def registry_no_owner(tmp_path):
"""Create a registry file with no owner set."""
reg = tmp_path / "AIPASS_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 2},
"branches": [
{
"name": "alpha",
"path": "src/alpha",
"email": "@alpha",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-04-16",
"last_active": "2026-04-16",
},
{
"name": "devpulse",
"path": "src/devpulse",
"email": "@devpulse",
"status": "active",
"profile": "library",
"description": "orchestrator",
"created": "2026-04-28",
"last_active": "2026-04-28",
},
],
}
),
encoding="utf-8",
)
return reg
class TestGetOwner:
"""Tests for get_owner()."""
def test_returns_owner_entry(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import get_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
result = get_owner(start_path=tmp_path)
assert result is not None
assert result["name"] == "devpulse"
assert result["owner"] is True
def test_returns_none_when_no_owner(self, registry_no_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import get_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_no_owner):
result = get_owner(start_path=tmp_path)
assert result is None
def test_returns_none_when_registry_missing(self, tmp_path):
from aipass.spawn.apps.handlers.registry import get_owner
missing = tmp_path / "MISSING_REGISTRY.json"
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=missing):
result = get_owner(start_path=tmp_path)
assert result is None
def test_default_start_path_uses_cwd(self, registry_with_owner):
from aipass.spawn.apps.handlers.registry import get_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
result = get_owner()
assert result is not None
assert result["name"] == "devpulse"
class TestIsOwner:
"""Tests for is_owner()."""
def test_true_for_owner_email_with_at(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
assert is_owner("@devpulse", start_path=tmp_path) is True
def test_true_for_owner_email_without_at(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
assert is_owner("devpulse", start_path=tmp_path) is True
def test_false_for_non_owner(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
assert is_owner("@alpha", start_path=tmp_path) is False
def test_false_for_empty_email(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
assert is_owner("", start_path=tmp_path) is False
def test_false_for_none_email(self, registry_with_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_with_owner):
assert is_owner(None, start_path=tmp_path) is False
def test_false_when_no_owner_in_registry(self, registry_no_owner, tmp_path):
from aipass.spawn.apps.handlers.registry import is_owner
with patch("aipass.spawn.apps.handlers.registry.find_registry", return_value=registry_no_owner):
assert is_owner("@devpulse", start_path=tmp_path) is False
class TestEnsureProjectHasOwner:
"""Tests for ensure_project_has_owner() — registry-entry based."""
def test_sets_owner_on_manager_branch(self, tmp_path):
from aipass.spawn.apps.handlers.registry import ensure_project_has_owner
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 2},
"branches": [
{
"name": "alpha",
"path": "src/alpha",
"email": "@alpha",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-01",
"last_active": "2026-01-01",
},
{
"name": "devpulse",
"path": "src/devpulse",
"email": "@devpulse",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-02",
"last_active": "2026-01-02",
},
],
}
),
encoding="utf-8",
)
alpha_dir = tmp_path / "src" / "alpha" / ".trinity"
alpha_dir.mkdir(parents=True)
(alpha_dir / "passport.json").write_text(
json.dumps(
{
"identity": {"citizen_class": "aipass_framework"},
"citizenship": {"registry_id": "abc"},
}
),
encoding="utf-8",
)
dp_dir = tmp_path / "src" / "devpulse" / ".trinity"
dp_dir.mkdir(parents=True)
(dp_dir / "passport.json").write_text(
json.dumps(
{
"identity": {"citizen_class": "manager"},
"citizenship": {"registry_id": "abc"},
}
),
encoding="utf-8",
)
result = ensure_project_has_owner(reg)
assert result is True
data = json.loads(reg.read_text(encoding="utf-8"))
devpulse_entry = next(b for b in data["branches"] if b["name"] == "devpulse")
alpha_entry = next(b for b in data["branches"] if b["name"] == "alpha")
assert devpulse_entry.get("owner") is True
assert alpha_entry.get("owner") is None or alpha_entry.get("owner") is not True
def test_noop_when_owner_already_set(self, registry_with_owner):
from aipass.spawn.apps.handlers.registry import ensure_project_has_owner
result = ensure_project_has_owner(registry_with_owner)
assert result is False
def test_returns_false_for_empty_registry(self, tmp_path):
from aipass.spawn.apps.handlers.registry import ensure_project_has_owner
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 0},
"branches": [],
}
),
encoding="utf-8",
)
result = ensure_project_has_owner(reg)
assert result is False
class TestBackfillOwnerAndRegistryId:
"""Tests for backfill_owner_and_registry_id()."""
def test_backfills_registry_id_and_owner(self, tmp_path):
from aipass.spawn.apps.handlers.registry import backfill_owner_and_registry_id
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 2},
"branches": [
{
"name": "alpha",
"path": "src/alpha",
"email": "@alpha",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-01",
"last_active": "2026-01-01",
},
{
"name": "devpulse",
"path": "src/devpulse",
"email": "@devpulse",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-02",
"last_active": "2026-01-02",
},
],
}
),
encoding="utf-8",
)
alpha_dir = tmp_path / "src" / "alpha" / ".trinity"
alpha_dir.mkdir(parents=True)
(alpha_dir / "passport.json").write_text(
json.dumps(
{
"identity": {"citizen_class": "aipass_framework"},
"citizenship": {"registry_id": "uuid-alpha"},
}
),
encoding="utf-8",
)
dp_dir = tmp_path / "src" / "devpulse" / ".trinity"
dp_dir.mkdir(parents=True)
(dp_dir / "passport.json").write_text(
json.dumps(
{
"identity": {"citizen_class": "manager"},
"citizenship": {"registry_id": "uuid-dp"},
}
),
encoding="utf-8",
)
result = backfill_owner_and_registry_id(reg)
assert result is True
data = json.loads(reg.read_text(encoding="utf-8"))
alpha_entry = next(b for b in data["branches"] if b["name"] == "alpha")
dp_entry = next(b for b in data["branches"] if b["name"] == "devpulse")
assert alpha_entry["registry_id"] == "uuid-alpha"
assert dp_entry["registry_id"] == "uuid-dp"
assert dp_entry["owner"] is True
assert alpha_entry.get("owner") is None or alpha_entry.get("owner") is not True
def test_noop_when_already_backfilled(self, tmp_path):
from aipass.spawn.apps.handlers.registry import backfill_owner_and_registry_id
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 1},
"branches": [
{
"name": "devpulse",
"path": "src/devpulse",
"email": "@devpulse",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-01",
"last_active": "2026-01-01",
"owner": True,
"registry_id": "uuid-dp",
},
],
}
),
encoding="utf-8",
)
dp_dir = tmp_path / "src" / "devpulse" / ".trinity"
dp_dir.mkdir(parents=True)
(dp_dir / "passport.json").write_text(
json.dumps(
{
"identity": {"citizen_class": "manager"},
"citizenship": {"registry_id": "uuid-dp"},
}
),
encoding="utf-8",
)
result = backfill_owner_and_registry_id(reg)
assert result is False
def test_skips_branches_without_passport(self, tmp_path):
from aipass.spawn.apps.handlers.registry import backfill_owner_and_registry_id
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 1},
"branches": [
{
"name": "ghost",
"path": "src/ghost",
"email": "@ghost",
"status": "active",
"profile": "library",
"description": "test",
"created": "2026-01-01",
"last_active": "2026-01-01",
},
],
}
),
encoding="utf-8",
)
result = backfill_owner_and_registry_id(reg)
assert result is False
class TestAddToRegistryWithRegistryId:
"""Tests for add_to_registry with registry_id parameter."""
def test_includes_registry_id_when_provided(self, tmp_path):
from aipass.spawn.apps.handlers.registry import add_to_registry
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 0},
"branches": [],
}
),
encoding="utf-8",
)
result = add_to_registry(
reg,
"NEW_BRANCH",
"src/new_branch",
"library",
"@new_branch",
purpose="test branch",
registry_id="uuid-new",
)
assert result is True
data = json.loads(reg.read_text(encoding="utf-8"))
entry = data["branches"][0]
assert entry["registry_id"] == "uuid-new"
def test_omits_registry_id_when_empty(self, tmp_path):
from aipass.spawn.apps.handlers.registry import add_to_registry
reg = tmp_path / "TEST_REGISTRY.json"
reg.write_text(
json.dumps(
{
"metadata": {"version": "1.0.0", "last_updated": "2026-07-10", "total_branches": 0},
"branches": [],
}
),
encoding="utf-8",
)
add_to_registry(reg, "NEW_BRANCH", "src/new_branch", "library", "@new_branch")
data = json.loads(reg.read_text(encoding="utf-8"))
entry = data["branches"][0]
assert "registry_id" not in entry