From db55b6b63aff38e23641df340bca79b5a7650ae1 Mon Sep 17 00:00:00 2001 From: AIPass Date: Sun, 3 May 2026 18:28:08 -0700 Subject: [PATCH] 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 --- .../drone/apps/handlers/registry_handler.py | 44 ++++++++++-- .../drone/tests/test_registry_handler.py | 70 +++++++++++++++++++ 2 files changed, 109 insertions(+), 5 deletions(-) diff --git a/src/aipass/drone/apps/handlers/registry_handler.py b/src/aipass/drone/apps/handlers/registry_handler.py index 2130ad9f..c1116cc3 100644 --- a/src/aipass/drone/apps/handlers/registry_handler.py +++ b/src/aipass/drone/apps/handlers/registry_handler.py @@ -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") diff --git a/src/aipass/drone/tests/test_registry_handler.py b/src/aipass/drone/tests/test_registry_handler.py index f883a954..940e9fb8 100644 --- a/src/aipass/drone/tests/test_registry_handler.py +++ b/src/aipass/drone/tests/test_registry_handler.py @@ -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()