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
|
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:
|
def find_registry() -> Path:
|
||||||
"""Find a *_REGISTRY.json by walking up from this file's location.
|
"""Find a *_REGISTRY.json by walking up from this file's location.
|
||||||
|
|
||||||
Search order:
|
Search order:
|
||||||
1. Explicitly set path via set_registry_path()
|
1. Explicitly set path via set_registry_path()
|
||||||
2. AIPASS_REGISTRY environment variable
|
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
|
4. AIPASS_HOME env var — for external projects where CWD walk finds nothing
|
||||||
5. Walk up from drone package location
|
5. Walk up from drone package location
|
||||||
6. Default: package-relative path
|
6. Default: package-relative path
|
||||||
|
|
||||||
The first directory that contains any *_REGISTRY.json is treated
|
When a candidate registry's metadata.id conflicts with the nearest
|
||||||
as the project boundary. If that directory holds more than one
|
passport's registry_id, it is skipped and the walk continues upward.
|
||||||
match, the alphabetically-first file is returned.
|
|
||||||
"""
|
"""
|
||||||
# Walk up from cwd FIRST — this is where the user is working
|
# Walk up from cwd FIRST — this is where the user is working
|
||||||
cwd = Path.cwd()
|
cwd = Path.cwd()
|
||||||
for parent in [cwd] + list(cwd.parents):
|
for parent in [cwd] + list(cwd.parents):
|
||||||
hit = _first_registry_in(parent)
|
hit = _first_registry_in(parent)
|
||||||
if hit is not None:
|
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 fallback — for external projects where CWD walk finds nothing
|
||||||
aipass_home = os.environ.get("AIPASS_HOME")
|
aipass_home = os.environ.get("AIPASS_HOME")
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ import pytest
|
|||||||
|
|
||||||
from aipass.drone.apps.handlers.registry_handler import (
|
from aipass.drone.apps.handlers.registry_handler import (
|
||||||
_first_registry_in,
|
_first_registry_in,
|
||||||
|
_registry_matches_credential,
|
||||||
_validate_branch_path,
|
_validate_branch_path,
|
||||||
_verify_registry_credential,
|
_verify_registry_credential,
|
||||||
find_registry,
|
find_registry,
|
||||||
@@ -388,6 +389,75 @@ class TestFindRegistry:
|
|||||||
assert isinstance(result, Path)
|
assert isinstance(result, Path)
|
||||||
assert result.name == "AIPASS_REGISTRY.json"
|
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()
|
# 8, 9, 10. _verify_registry_credential()
|
||||||
|
|||||||
Reference in New Issue
Block a user