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:
co-authored by
Claude Opus 4.6
parent
a2d84a78da
commit
db55b6b63a
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user