diff --git a/src/aipass/spawn/apps/handlers/delete_ops.py b/src/aipass/spawn/apps/handlers/delete_ops.py index 3c5b7787..e9c36901 100644 --- a/src/aipass/spawn/apps/handlers/delete_ops.py +++ b/src/aipass/spawn/apps/handlers/delete_ops.py @@ -19,6 +19,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger from aipass.spawn.apps.handlers.registry import ( find_registry, + is_protected, load_registry, save_registry, branches_as_list, @@ -26,9 +27,6 @@ from aipass.spawn.apps.handlers.registry import ( from aipass.spawn.apps.handlers.repair_ops import ARCHIVE_EXCLUDE from aipass.spawn.apps.handlers.json import json_handler -# Branches that cannot be deleted (critical infrastructure) -_PROTECTED_BRANCHES = {"spawn", "devpulse", "drone"} - # ============================================================================= # PUBLIC API @@ -120,14 +118,15 @@ def delete_branch( Returns: Dict with deletion results. """ - # Safety: check protected branches - if branch_name.lower() in _PROTECTED_BRANCHES: - msg = f"Cannot delete '{branch_name}' — protected branch ({', '.join(sorted(_PROTECTED_BRANCHES))})" + # 1. Resolve registry (needed for protection check and branch resolution) + registry_path = find_registry() + + # Safety: check protected branches (hardcoded floor + registry owner + active passport) + protected, reason = is_protected(branch_name, registry_path=registry_path) + if protected: + msg = f"Cannot delete '{branch_name}' — protected ({reason})" logger.warning(f"[delete] {msg}") return _error_result(branch_name, msg) - - # 1. Resolve branch path from registry - registry_path = find_registry() project_root = registry_path.parent registry = load_registry(registry_path) branch_entry, branch_dir = _resolve_branch_dir(branch_name, registry_path, registry) diff --git a/src/aipass/spawn/apps/handlers/registry.py b/src/aipass/spawn/apps/handlers/registry.py index 59875b24..9ed39920 100644 --- a/src/aipass/spawn/apps/handlers/registry.py +++ b/src/aipass/spawn/apps/handlers/registry.py @@ -36,6 +36,59 @@ from aipass.prax.apps.modules.logger import system_logger as logger from aipass.spawn.apps.handlers.json import json_handler +_PROTECTED_FLOOR = frozenset({"spawn", "devpulse", "drone"}) + + +def is_protected(branch_name, branch_dir=None, registry_path=None): + """Check if a branch is protected from deletion and pollution cleanup. + + Protection layers (any one is sufficient): + 1. Hardcoded floor — spawn, devpulse, drone. + 2. Registry owner flag — entry has ``owner: true``. + 3. Active passport — ``.trinity/passport.json`` with + ``citizenship.registered == True``. + + Args: + branch_name: Branch name to check (case-insensitive). + branch_dir: Path to the branch directory (for passport check). + Auto-resolved from registry when omitted. + registry_path: Path to ``*_REGISTRY.json``. + Auto-discovered via ``find_registry()`` when omitted. + + Returns: + Tuple of (protected: bool, reason: str). + """ + name_lower = branch_name.lower() + + if name_lower in _PROTECTED_FLOOR: + return True, f"infrastructure ({', '.join(sorted(_PROTECTED_FLOOR))})" + + try: + rp = Path(registry_path) if registry_path else find_registry() + reg_data = load_registry(rp) + for entry in branches_as_list(reg_data.get("branches", [])): + if entry.get("name", "").lower() == name_lower: + if entry.get("owner") is True: + return True, "registry owner" + if branch_dir is None: + branch_dir = (rp.parent / entry.get("path", "")).resolve() + break + except Exception: + pass + + if branch_dir is not None: + passport_path = Path(branch_dir) / ".trinity" / "passport.json" + if passport_path.is_file(): + try: + passport = json_handler.read_json(passport_path) + if passport and passport.get("citizenship", {}).get("registered") is True: + return True, "active citizen" + except Exception: + pass + + return False, "" + + def branches_as_list(branches): """Normalize branches to a list regardless of storage format. diff --git a/src/aipass/spawn/apps/handlers/repair_ops.py b/src/aipass/spawn/apps/handlers/repair_ops.py index 2ee756f3..dac06411 100644 --- a/src/aipass/spawn/apps/handlers/repair_ops.py +++ b/src/aipass/spawn/apps/handlers/repair_ops.py @@ -24,6 +24,7 @@ ARCHIVE_EXCLUDE = {".venv", ".git", "__pycache__", ".chroma", "node_modules", ". from aipass.spawn.apps.handlers.registry import ( find_registry, + is_protected, load_registry, save_registry, branches_as_list, @@ -322,6 +323,8 @@ def detect_pollution(project_root): """Detect init pollution — duplicate nested directories. Init pollution: project_name/project_name/ exists (e.g., compass/compass/). + Skips directories that are protected branches (active passport, registry + owner, or infrastructure floor). Args: project_root: Path to the project root directory @@ -333,15 +336,26 @@ def detect_pollution(project_root): issues = [] project_name = project_root.name + registry_path = None + for f in sorted(project_root.glob("*_REGISTRY.json")): + registry_path = f + break + nested = project_root / project_name if nested.is_dir(): - issues.append( - { - "type": "duplicate_nested_dir", - "path": project_name, - "description": f"Duplicate nested directory: {project_name}/{project_name}/", - } + protected, _reason = is_protected( + project_name, + branch_dir=nested, + registry_path=registry_path, ) + if not protected: + issues.append( + { + "type": "duplicate_nested_dir", + "path": project_name, + "description": f"Duplicate nested directory: {project_name}/{project_name}/", + } + ) src_dir = project_root / "src" if src_dir.is_dir(): @@ -349,14 +363,20 @@ def detect_pollution(project_root): if child.is_dir() and not child.name.startswith(".") and not child.name.startswith("__"): nested_dup = child / child.name if nested_dup.is_dir(): - rel = nested_dup.relative_to(project_root).as_posix() - issues.append( - { - "type": "duplicate_nested_dir", - "path": rel, - "description": f"Duplicate nested directory: src/{child.name}/{child.name}/", - } + protected, _reason = is_protected( + child.name, + branch_dir=nested_dup, + registry_path=registry_path, ) + if not protected: + rel = nested_dup.relative_to(project_root).as_posix() + issues.append( + { + "type": "duplicate_nested_dir", + "path": rel, + "description": f"Duplicate nested directory: src/{child.name}/{child.name}/", + } + ) return issues 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 d9227c3b..7dc05fe8 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-17", + "last_updated": "2026-07-19", "version": "1.0.0" } } diff --git a/src/aipass/spawn/tests/test_repair.py b/src/aipass/spawn/tests/test_repair.py index 84e3c08d..630365c1 100644 --- a/src/aipass/spawn/tests/test_repair.py +++ b/src/aipass/spawn/tests/test_repair.py @@ -473,6 +473,223 @@ class TestChromaRelocation: assert (project / ".chroma").exists() +# --------------------------------------------------------------------------- +# is_protected shared helper +# --------------------------------------------------------------------------- + + +class TestIsProtected: + """Tests for is_protected() — shared protection helper across repair and delete.""" + + def test_hardcoded_floor_spawn(self, tmp_path): + """spawn is protected by the hardcoded floor.""" + from aipass.spawn.apps.handlers.registry import is_protected + + protected, reason = is_protected("spawn") + assert protected is True + assert "infrastructure" in reason + + def test_hardcoded_floor_case_insensitive(self, tmp_path): + """Floor check is case-insensitive.""" + from aipass.spawn.apps.handlers.registry import is_protected + + protected, _reason = is_protected("DEVPULSE") + assert protected is True + + def test_registry_owner_protected(self, tmp_path): + """Branch with owner:true in registry is protected.""" + from aipass.spawn.apps.handlers.registry import is_protected + + project, reg = _make_project(tmp_path, branches=[{"name": "MYOWNER", "path": "myowner"}]) + reg_data = json.loads(reg.read_text()) + reg_data["branches"][0]["owner"] = True + reg.write_text(json.dumps(reg_data)) + + protected, reason = is_protected("myowner", registry_path=reg) + assert protected is True + assert "owner" in reason + + def test_active_passport_protected(self, tmp_path): + """Branch with citizenship.registered=True passport is protected.""" + from aipass.spawn.apps.handlers.registry import is_protected + + project, reg = _make_project(tmp_path, branches=[{"name": "CITIZEN", "path": "citizen"}]) + protected, reason = is_protected("citizen", registry_path=reg) + assert protected is True + assert "citizen" in reason + + def test_no_passport_not_protected(self, tmp_path): + """Branch without passport (no citizenship.registered) is not protected.""" + from aipass.spawn.apps.handlers.registry import is_protected + + project = tmp_path / "proj" + project.mkdir() + branch = project / "ephemeral" + branch.mkdir() + + reg = project / "TEST_REGISTRY.json" + reg.write_text( + json.dumps( + { + "metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "total_branches": 1}, + "branches": [{"name": "EPHEMERAL", "path": "ephemeral", "status": "active"}], + } + ) + ) + + protected, _reason = is_protected("ephemeral", registry_path=reg) + assert protected is False + + def test_minimal_passport_not_protected(self, tmp_path): + """Passport without citizenship.registered is not protected.""" + from aipass.spawn.apps.handlers.registry import is_protected + + branch = tmp_path / "minimal" + branch.mkdir() + (branch / ".trinity").mkdir() + (branch / ".trinity" / "passport.json").write_text(json.dumps({"name": "MINIMAL", "role": "test"})) + + protected, _reason = is_protected("minimal", branch_dir=branch) + assert protected is False + + def test_unknown_branch_not_protected(self): + """Completely unknown branch is not protected.""" + from aipass.spawn.apps.handlers.registry import is_protected + + protected, _reason = is_protected("nonexistent", branch_dir=None, registry_path=None) + assert protected is False + + +# --------------------------------------------------------------------------- +# detect_pollution skips protected branches +# --------------------------------------------------------------------------- + + +class TestDetectPollutionProtection: + """Tests for detect_pollution skipping protected branches.""" + + def test_skips_branch_with_active_passport(self, tmp_path): + """src/pkg/pkg/ with active passport is NOT flagged as pollution.""" + from aipass.spawn.apps.handlers.repair_ops import detect_pollution + + project = tmp_path / "myproj" + project.mkdir() + src_pkg = project / "src" / "mypkg" / "mypkg" + src_pkg.mkdir(parents=True) + + trinity = src_pkg / ".trinity" + trinity.mkdir() + (trinity / "passport.json").write_text( + json.dumps( + { + "branch_info": {"branch_name": "mypkg"}, + "identity": {"citizen_class": "aipass_framework"}, + "citizenship": {"registered": True}, + } + ) + ) + + reg = project / "MYPROJ_REGISTRY.json" + reg.write_text( + json.dumps( + { + "metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "total_branches": 1}, + "branches": [{"name": "MYPKG", "path": "src/mypkg/mypkg", "status": "active"}], + } + ) + ) + + issues = detect_pollution(project) + assert len(issues) == 0 + + def test_still_flags_real_pollution(self, tmp_path): + """src/pkg/pkg/ without passport IS flagged as pollution.""" + from aipass.spawn.apps.handlers.repair_ops import detect_pollution + + project = tmp_path / "myproj" + project.mkdir() + (project / "src" / "mypkg" / "mypkg").mkdir(parents=True) + + issues = detect_pollution(project) + assert len(issues) == 1 + assert issues[0]["type"] == "duplicate_nested_dir" + + def test_skips_owner_branch_at_root(self, tmp_path): + """project/project/ with owner flag is NOT flagged as pollution.""" + from aipass.spawn.apps.handlers.repair_ops import detect_pollution + + project = tmp_path / "compass" + project.mkdir() + nested = project / "compass" + nested.mkdir() + + reg = project / "COMPASS_REGISTRY.json" + reg.write_text( + json.dumps( + { + "metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "total_branches": 1}, + "branches": [{"name": "COMPASS", "path": "compass", "status": "active", "owner": True}], + } + ) + ) + + issues = detect_pollution(project) + assert len(issues) == 0 + + +# --------------------------------------------------------------------------- +# delete_branch refuses owner branches +# --------------------------------------------------------------------------- + + +class TestDeleteOwnerProtection: + """Tests for delete_branch refusing registry-owner branches.""" + + def test_delete_owner_refused(self, tmp_path): + """Cannot delete a branch with owner:true in registry.""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + project = tmp_path / "repo" + project.mkdir() + branch = project / "src" / "aipass" / "aipass_branch" + branch.mkdir(parents=True) + (branch / ".trinity").mkdir() + (branch / ".trinity" / "passport.json").write_text( + json.dumps( + { + "identity": {"citizen_class": "manager"}, + "citizenship": {"registered": True}, + } + ) + ) + + reg = project / "AIPASS_REGISTRY.json" + reg.write_text( + json.dumps( + { + "metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "total_branches": 1}, + "branches": [ + { + "name": "AIPASS_BRANCH", + "path": "src/aipass/aipass_branch", + "status": "active", + "owner": True, + "email": "@aipass_branch", + } + ], + } + ) + ) + + with patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=reg): + result = delete_branch("aipass_branch", confirm=False) + + assert result["success"] is False + assert "protected" in result.get("error", "").lower() + assert "owner" in result.get("error", "").lower() + assert branch.exists() + + # --------------------------------------------------------------------------- # ARCHIVE_EXCLUDE shared constant # ---------------------------------------------------------------------------