fix(drone): skip mismatched registries during walk-up instead of hard-failing

find_registry() now performs a lightweight credential check on each
candidate *_REGISTRY.json during the CWD walk-up. When a registry's
metadata.id conflicts with the nearest passport's registry_id, it is
skipped and the walk continues upward to find the correct registry.

Fixes the bug where an orphan DEVPULSE_REGISTRY.json inside a citizen's
tree caused RegistryMismatchError for all drone commands from devpulse CWD.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
AIPass
2026-05-03 18:28:08 -07:00
co-authored by Claude Opus 4.6
parent a2d84a78da
commit db55b6b63a
2 changed files with 109 additions and 5 deletions
@@ -77,27 +77,61 @@ def _first_registry_in(directory: Path) -> Optional[Path]:
return matches[0] if matches else None
def _registry_matches_credential(registry_path: Path) -> bool:
"""Check whether a candidate registry matches the nearest passport.
Returns True when the registry is acceptable (IDs match, or either
side is missing an ID). Returns False only when both IDs exist and
disagree — the caller should skip this registry and keep walking.
"""
try:
with open(registry_path, "r", encoding="utf-8") as f:
data = json.load(f)
registry_id = data.get("metadata", {}).get("id") if isinstance(data, dict) else None
if not registry_id:
return True
cwd = Path.cwd()
for parent in [cwd] + list(cwd.parents):
candidate = parent / ".trinity" / "passport.json"
if candidate.is_file():
with open(candidate, "r", encoding="utf-8") as f:
passport = json.load(f)
passport_id = passport.get("citizenship", {}).get("registry_id")
if not passport_id:
return True
return passport_id == registry_id
return True
except Exception as exc:
logger.warning("Credential pre-check failed for %s: %s", registry_path, exc)
return True
def find_registry() -> Path:
"""Find a *_REGISTRY.json by walking up from this file's location.
Search order:
1. Explicitly set path via set_registry_path()
2. AIPASS_REGISTRY environment variable
3. Walk up from cwd
3. Walk up from cwd (skipping registries that fail credential check)
4. AIPASS_HOME env var — for external projects where CWD walk finds nothing
5. Walk up from drone package location
6. Default: package-relative path
The first directory that contains any *_REGISTRY.json is treated
as the project boundary. If that directory holds more than one
match, the alphabetically-first file is returned.
When a candidate registry's metadata.id conflicts with the nearest
passport's registry_id, it is skipped and the walk continues upward.
"""
# Walk up from cwd FIRST — this is where the user is working
cwd = Path.cwd()
for parent in [cwd] + list(cwd.parents):
hit = _first_registry_in(parent)
if hit is not None:
return hit
if _registry_matches_credential(hit):
return hit
logger.warning(
"Skipping mismatched registry at %s — continuing walk-up",
hit,
)
# AIPASS_HOME fallback — for external projects where CWD walk finds nothing
aipass_home = os.environ.get("AIPASS_HOME")
@@ -19,6 +19,7 @@ import pytest
from aipass.drone.apps.handlers.registry_handler import (
_first_registry_in,
_registry_matches_credential,
_validate_branch_path,
_verify_registry_credential,
find_registry,
@@ -388,6 +389,75 @@ class TestFindRegistry:
assert isinstance(result, Path)
assert result.name == "AIPASS_REGISTRY.json"
def test_find_registry_skips_mismatched_continues_up(self, registry_dir: Path, monkeypatch):
"""find_registry() skips a nested registry with wrong credential and finds the correct one above."""
passport_id = "correct-registry-id"
wrong_id = "wrong-registry-id"
_write_registry(registry_dir, _minimal_registry(metadata_id=passport_id))
nested = registry_dir / "src" / "aipass" / "citizen"
nested.mkdir(parents=True)
_write_registry(nested, {"metadata": {"id": wrong_id}, "branches": []}, name="CITIZEN_REGISTRY.json")
_write_passport(nested, {"citizenship": {"registry_id": passport_id}})
monkeypatch.chdir(nested)
result = find_registry()
assert result.parent == registry_dir
assert result.name == "AIPASS_REGISTRY.json"
def test_find_registry_returns_matching_nested(self, registry_dir: Path, monkeypatch):
"""find_registry() returns a nested registry when credentials match."""
shared_id = "shared-id"
_write_registry(registry_dir, _minimal_registry(metadata_id="parent-id"))
nested = registry_dir / "sub"
nested.mkdir()
_write_registry(nested, _minimal_registry(metadata_id=shared_id), name="SUB_REGISTRY.json")
_write_passport(nested, {"citizenship": {"registry_id": shared_id}})
monkeypatch.chdir(nested)
result = find_registry()
assert result.parent == nested
assert result.name == "SUB_REGISTRY.json"
# ===================================================================
# 7b. _registry_matches_credential()
# ===================================================================
class TestRegistryMatchesCredential:
def test_returns_true_when_ids_match(self, registry_dir: Path, monkeypatch):
"""Returns True when passport and registry IDs agree."""
shared_id = "match-id"
path = _write_registry(registry_dir, _minimal_registry(metadata_id=shared_id))
_write_passport(registry_dir, {"citizenship": {"registry_id": shared_id}})
monkeypatch.chdir(registry_dir)
assert _registry_matches_credential(path) is True
def test_returns_false_on_mismatch(self, registry_dir: Path, monkeypatch):
"""Returns False when passport and registry IDs disagree."""
path = _write_registry(registry_dir, _minimal_registry(metadata_id="reg-A"))
_write_passport(registry_dir, {"citizenship": {"registry_id": "reg-B"}})
monkeypatch.chdir(registry_dir)
assert _registry_matches_credential(path) is False
def test_returns_true_when_registry_has_no_id(self, registry_dir: Path, monkeypatch):
"""Returns True when registry has no metadata.id."""
path = _write_registry(registry_dir, _minimal_registry())
_write_passport(registry_dir, {"citizenship": {"registry_id": "some-id"}})
monkeypatch.chdir(registry_dir)
assert _registry_matches_credential(path) is True
def test_returns_true_when_no_passport(self, registry_dir: Path, monkeypatch):
"""Returns True when no passport.json exists."""
path = _write_registry(registry_dir, _minimal_registry(metadata_id="some-id"))
monkeypatch.chdir(registry_dir)
assert _registry_matches_credential(path) is True
# ===================================================================
# 8, 9, 10. _verify_registry_credential()