Reuse path resolution within each registry inspection

loopx/registry.py

@@ -389,6 +389,14 @@ def inspect_registry(path: Path) -> dict[str, Any]:
     seen_ids: set[str] = set()
     repo_goal_ids: dict[str, list[str]] = {}
     state_goal_ids: dict[str, list[str]] = {}
+    path_keys: dict[str, str] = {}
+
+    def inspection_path_key(path: Path) -> str:
+        # Group and report from the same resolution within this registry read.
+        spelling = os.fspath(path)
+        if spelling not in path_keys:
+            path_keys[spelling] = stable_path_key(path)
+        return path_keys[spelling]
 
     def add_problem(
         code: str,
@@ -416,10 +424,10 @@ def add_problem(
         repo_text = str(raw_goal.get("repo") or "")
         repo = Path(repo_text).expanduser() if repo_text else None
         if repo:
-            repo_goal_ids.setdefault(stable_path_key(repo), []).append(goal_id)
+            repo_goal_ids.setdefault(inspection_path_key(repo), []).append(goal_id)
             state_file = resolve_state_file(repo, raw_goal.get("state_file"))
             if state_file:
-                state_goal_ids.setdefault(stable_path_key(state_file), []).append(goal_id)
+                state_goal_ids.setdefault(inspection_path_key(state_file), []).append(goal_id)
 
     for raw_goal in goals:
         if not isinstance(raw_goal, dict):
@@ -431,7 +439,7 @@ def add_problem(
         repo_text = str(raw_goal.get("repo") or "")
         repo = Path(repo_text).expanduser() if repo_text else None
         state_file = resolve_state_file(repo, raw_goal.get("state_file")) if repo else None
-        repo_ids = repo_goal_ids.get(stable_path_key(repo), []) if repo else []
+        repo_ids = repo_goal_ids.get(inspection_path_key(repo), []) if repo else []
         adapter = cast(dict[str, Any], raw_goal.get("adapter")) if isinstance(raw_goal.get("adapter"), dict) else {}
         spawn_policy = cast(dict[str, Any], raw_goal.get("spawn_policy")) if isinstance(raw_goal.get("spawn_policy"), dict) else {}
         orchestration = compact_orchestration_policy(spawn_policy)
@@ -490,7 +498,7 @@ def add_problem(
                 "repo_goal_count": len(repo_ids),
                 "repo_goal_ids": repo_ids,
                 "state_file": raw_goal.get("state_file"),
-                "state_file_abs": stable_path_key(state_file) if state_file else None,
+                "state_file_abs": inspection_path_key(state_file) if state_file else None,
                 "state_file_exists": bool(state_file and state_file.exists()),
                 "adapter_kind": adapter.get("kind"),
                 "adapter_status": adapter.get("status"),

tests/presentation/test_goal_source.py

@@ -12,6 +12,7 @@
 
 from loopx.capabilities.explore.result_log import build_explore_node_event, explore_history_rows
 from loopx.control_plane.status.goal_source import GoalSource, GoalSourceReader, SourceScopeError
+from loopx.registry import inspect_registry
 
 
 def manifest(*, connection=None, environment=None, scope=None):
@@ -132,3 +133,35 @@ def test_real_native_host_lists_goals_and_reads_canonical_history(tmp_path):
     with pytest.raises((SourceScopeError, subprocess.SubprocessError)):
         reader.status("unregistered")
     assert reader.status("first")["run_history"]["goals"][0]["id"] == "first"
+
+
+def test_registry_inspection_rechecks_retargeted_source_paths(tmp_path):
+    first, second = tmp_path / "first", tmp_path / "second"
+    for project in (first, second):
+        project.mkdir()
+        (project / "active.md").write_text("# Research\n")
+    alias = tmp_path / "original"
+    try:
+        alias.symlink_to(first, target_is_directory=True)
+    except (OSError, NotImplementedError):
+        pytest.skip("Directory symlinks are not available")
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"schema_version": 1, "goals": [
+        {"id": goal_id, "domain": "source-read", "status": "active",
+         "repo": str(repo), "state_file": "active.md", "adapter": {"kind": "test_v0"}}
+        for goal_id, repo in (("first", first), ("second", alias))
+    ]}))
+    shared = inspect_registry(registry)
+    assert shared["ok"] is False
+    assert any(row["code"] == "registry_state_file_shared" for row in shared["problem_diagnostics"])
+    assert [row["repo_goal_ids"] for row in shared["goals"]] == [["first", "second"]] * 2
+    assert [row["state_file_abs"] for row in shared["goals"]] == [str(first.resolve() / "active.md")] * 2
+
+    alias.unlink()
+    alias.symlink_to(second, target_is_directory=True)
+    separated = inspect_registry(registry)
+    assert separated["ok"] is True
+    assert [row["repo_goal_ids"] for row in separated["goals"]] == [["first"], ["second"]]
+    assert [row["state_file_abs"] for row in separated["goals"]] == [
+        str(first.resolve() / "active.md"), str(second.resolve() / "active.md"),
+    ]