From 874c7fed2ea2fed802de4d7a10eb75379278a230 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Fri, 10 Jul 2026 00:46:17 -0700 Subject: [PATCH] =?UTF-8?q?#678=20owner-capability:=20seal=20project=20own?= =?UTF-8?q?ership=20in=20the=20registry=20+=20wake=20the=20OWNER=20back=20?= =?UTF-8?q?on=20dispatch=20completion.=20TDPLAN-0012=20=E2=80=94=203=20par?= =?UTF-8?q?ts=20built=20in=20parallel=20against=20a=20frozen=20is=5Fowner?= =?UTF-8?q?=20contract,=20verified=20end-to-end=20by=20devpulse.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit @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. --- .aipass/hooks.json | 5 + .../handlers/dispatch/dispatch_monitor.py | 84 ++++ src/aipass/ai_mail/apps/modules/dispatch.py | 58 +-- .../ai_mail/tests/test_dispatch_module.py | 207 +------- .../ai_mail/tests/test_dispatch_monitor.py | 424 ++++++++++++++++ src/aipass/hooks/.seedgo/bypass.json | 45 ++ src/aipass/hooks/README.md | 5 +- .../apps/handlers/security/registry_gate.py | 166 ++++++ src/aipass/hooks/tests/test_registry_gate.py | 337 +++++++++++++ src/aipass/spawn/.seedgo/bypass.json | 20 + src/aipass/spawn/apps/handlers/registry.py | 128 ++++- src/aipass/spawn/apps/modules/core.py | 3 + .../.spawn/.template_registry.json | 2 +- .../spawn/tests/test_citizen_classes.py | 50 +- src/aipass/spawn/tests/test_owner_resolver.py | 472 ++++++++++++++++++ 15 files changed, 1719 insertions(+), 287 deletions(-) create mode 100644 src/aipass/hooks/apps/handlers/security/registry_gate.py create mode 100644 src/aipass/hooks/tests/test_registry_gate.py create mode 100644 src/aipass/spawn/tests/test_owner_resolver.py diff --git a/.aipass/hooks.json b/.aipass/hooks.json index a35c4000..d50add22 100644 --- a/.aipass/hooks.json +++ b/.aipass/hooks.json @@ -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", diff --git a/src/aipass/ai_mail/apps/handlers/dispatch/dispatch_monitor.py b/src/aipass/ai_mail/apps/handlers/dispatch/dispatch_monitor.py index 6a85b9f0..406536b1 100644 --- a/src/aipass/ai_mail/apps/handlers/dispatch/dispatch_monitor.py +++ b/src/aipass/ai_mail/apps/handlers/dispatch/dispatch_monitor.py @@ -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) diff --git a/src/aipass/ai_mail/apps/modules/dispatch.py b/src/aipass/ai_mail/apps/modules/dispatch.py index ee2fafb8..296e81d3 100644 --- a/src/aipass/ai_mail/apps/modules/dispatch.py +++ b/src/aipass/ai_mail/apps/modules/dispatch.py @@ -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") diff --git a/src/aipass/ai_mail/tests/test_dispatch_module.py b/src/aipass/ai_mail/tests/test_dispatch_module.py index fdce3792..426804c2 100644 --- a/src/aipass/ai_mail/tests/test_dispatch_module.py +++ b/src/aipass/ai_mail/tests/test_dispatch_module.py @@ -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 diff --git a/src/aipass/ai_mail/tests/test_dispatch_monitor.py b/src/aipass/ai_mail/tests/test_dispatch_monitor.py index ff5f2f24..23aa3608 100644 --- a/src/aipass/ai_mail/tests/test_dispatch_monitor.py +++ b/src/aipass/ai_mail/tests/test_dispatch_monitor.py @@ -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" diff --git a/src/aipass/hooks/.seedgo/bypass.json b/src/aipass/hooks/.seedgo/bypass.json index e246f446..0cd1aa09 100644 --- a/src/aipass/hooks/.seedgo/bypass.json +++ b/src/aipass/hooks/.seedgo/bypass.json @@ -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", diff --git a/src/aipass/hooks/README.md b/src/aipass/hooks/README.md index 6a69124a..4b295cd0 100644 --- a/src/aipass/hooks/README.md +++ b/src/aipass/hooks/README.md @@ -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 | diff --git a/src/aipass/hooks/apps/handlers/security/registry_gate.py b/src/aipass/hooks/apps/handlers/security/registry_gate.py new file mode 100644 index 00000000..29f9880b --- /dev/null +++ b/src/aipass/hooks/apps/handlers/security/registry_gate.py @@ -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 diff --git a/src/aipass/hooks/tests/test_registry_gate.py b/src/aipass/hooks/tests/test_registry_gate.py new file mode 100644 index 00000000..15ebb9a2 --- /dev/null +++ b/src/aipass/hooks/tests/test_registry_gate.py @@ -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() diff --git a/src/aipass/spawn/.seedgo/bypass.json b/src/aipass/spawn/.seedgo/bypass.json index 13915e60..d6e509d8 100644 --- a/src/aipass/spawn/.seedgo/bypass.json +++ b/src/aipass/spawn/.seedgo/bypass.json @@ -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": { diff --git a/src/aipass/spawn/apps/handlers/registry.py b/src/aipass/spawn/apps/handlers/registry.py index a79581d4..43409ae4 100644 --- a/src/aipass/spawn/apps/handlers/registry.py +++ b/src/aipass/spawn/apps/handlers/registry.py @@ -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 diff --git a/src/aipass/spawn/apps/modules/core.py b/src/aipass/spawn/apps/modules/core.py index 66782407..5464461e 100644 --- a/src/aipass/spawn/apps/modules/core.py +++ b/src/aipass/spawn/apps/modules/core.py @@ -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 diff --git a/src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json b/src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json index 69a37988..71b35635 100644 --- a/src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json +++ b/src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json @@ -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" } } diff --git a/src/aipass/spawn/tests/test_citizen_classes.py b/src/aipass/spawn/tests/test_citizen_classes.py index 553b8b9e..fd93fc79 100644 --- a/src/aipass/spawn/tests/test_citizen_classes.py +++ b/src/aipass/spawn/tests/test_citizen_classes.py @@ -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 diff --git a/src/aipass/spawn/tests/test_owner_resolver.py b/src/aipass/spawn/tests/test_owner_resolver.py new file mode 100644 index 00000000..7fdc219a --- /dev/null +++ b/src/aipass/spawn/tests/test_owner_resolver.py @@ -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