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

fix(spawn): shared is_protected() guards repair + delete — S304 F30/F84. 3-layer check (infra floor, registry owner, active passport) replaces name-match pollution detection and 3-name delete guard. Live-verified: detect_pollution on AIPass root = 0 issues, delete aipass/devpulse both refused. 357 tests (11 new), seedgo 31/31. DPLAN-0250 Track A, dispatched to @spawn

Browse Source
This commit is contained in:
AIOSAI
2026-07-19 00:49:46 -07:00
parent 7811381d54
commit defee1f513
5 changed files with 312 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/spawn/apps/handlers/delete_ops.py
+8 -9
View File
@@ -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)
src/aipass/spawn/apps/handlers/registry.py
+53
View File
@@ -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.
src/aipass/spawn/apps/handlers/repair_ops.py
+33 -13
View File
@@ -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
src/aipass/spawn/templates/aipass_framework/.spawn/.template_registry.json
+1 -1
View File
@@ -429,7 +429,7 @@
},
"metadata": {
"description": "Template file tracking registry for ID-based updates",
"last_updated": "2026-07-17",
"last_updated": "2026-07-19",
"version": "1.0.0"
}
}
src/aipass/spawn/tests/test_repair.py
+217
View File
@@ -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
# ---------------------------------------------------------------------------