fix: preserve locked uv package identities in dependency scopes

scripts/ci/record_dependency_review.py

@@ -9,6 +9,7 @@
 import subprocess
 import tomllib
 from pathlib import Path
+from urllib.parse import unquote
 from urllib.request import Request, urlopen
 
 from scripts.ci.submit_uv_lock_snapshot import uv_manifest
@@ -62,10 +63,17 @@ def _uv_lock_components(sha: str, manifest: str) -> set[tuple[str, str, str]]:
         return set()
     lock = tomllib.loads(result.stdout.decode())
     resolved = uv_manifest(lock, manifest)["resolved"]
-    return {
-        (key.rsplit("@", 1)[0], key.rsplit("@", 1)[1], package["scope"])
-        for key, package in resolved.items()
-    }
+    # Snapshot keys can distinguish sources; the comparison API reports only
+    # name/version. Project to that boundary after identity-specific traversal,
+    # using runtime precedence for a component present in both scopes (as npm).
+    components: dict[tuple[str, str], str] = {}
+    for package in resolved.values():
+        name, version = package["package_url"].removeprefix("pkg:pypi/").rsplit("@", 1)
+        identity = (unquote(name), unquote(version))
+        scope = package["scope"]
+        if identity not in components or scope == "runtime":
+            components[identity] = scope
+    return {(name, version, scope) for (name, version), scope in components.items()}
 
 
 def _validate_changes(

scripts/ci/submit_uv_lock_snapshot.py

@@ -19,24 +19,49 @@
 
 DEVELOPMENT_EXTRAS = frozenset({"dev", "docs"})
 CORRELATOR = "0-eventrelay-uv-lock-scoped"
+PackageId = tuple[str, str, str]
 
 
-def _reachable(roots: list[dict], packages_by_name: dict[str, list[dict]]) -> set[str]:
-    visited: set[tuple[str, tuple[str, ...]]] = set()
-    reachable: set[str] = set()
+def _package_id(package: dict) -> PackageId:
+    return (
+        package["name"], package["version"],
+        json.dumps(package["source"], sort_keys=True, separators=(",", ":")),
+    )
+
+
+def _resolve_dependency(dependency: dict, packages_by_name: dict[str, list[dict]]) -> dict:
+    candidates = [
+        package for package in packages_by_name.get(dependency["name"], [])
+        if all(field not in dependency or package[field] == dependency[field]
+               for field in ("version", "source"))
+    ]
+    if not candidates:
+        raise ValueError(f"Unresolved uv dependency: {dependency!r}")
+    if len(candidates) != 1:
+        raise ValueError(f"Ambiguous uv dependency: {dependency!r}")
+    return candidates[0]
+
+
+def _reachable(
+    roots: list[dict], packages_by_name: dict[str, list[dict]]
+) -> dict[PackageId, set[str]]:
+    # Extras are traversal state, not part of a locked package's identity.
+    # Keep each scope's requested extras separate even for a shared package.
+    visited: set[tuple[PackageId, tuple[str, ...]]] = set()
+    reachable: dict[PackageId, set[str]] = defaultdict(set)
     pending = deque(roots)
     while pending:
         dependency = pending.popleft()
-        name = dependency["name"]
-        extras = tuple(sorted(dependency.get("extra", [])))
-        if (name, extras) in visited:
+        package = _resolve_dependency(dependency, packages_by_name)
+        identity = _package_id(package)
+        extras = tuple(sorted(set(dependency.get("extra", []))))
+        if (identity, extras) in visited:
             continue
-        visited.add((name, extras))
-        reachable.add(name)
-        for package in packages_by_name.get(name, []):
-            pending.extend(package.get("dependencies", []))
-            for extra in extras:
-                pending.extend(package.get("optional-dependencies", {}).get(extra, []))
+        visited.add((identity, extras))
+        reachable[identity].update(extras)
+        pending.extend(package.get("dependencies", []))
+        for extra in extras:
+            pending.extend(package.get("optional-dependencies", {}).get(extra, []))
     return reachable
 
 
@@ -52,8 +77,25 @@ def uv_manifest(lock: dict, location: str = "uv.lock") -> dict:
         raise ValueError(f"Expected one uv project root; found {len(roots)}")
     root = roots[0]
     packages_by_name: dict[str, list[dict]] = defaultdict(list)
+    package_ids = set()
+    registry_ids: dict[tuple[str, str], list[PackageId]] = defaultdict(list)
     for package in packages:
+        identity = _package_id(package)
+        if identity in package_ids:
+            raise ValueError(f"Duplicate uv package: {identity!r}")
+        package_ids.add(identity)
         packages_by_name[package["name"]].append(package)
+        if "registry" in package["source"]:
+            registry_ids[identity[:2]].append(identity)
+    # Preserve existing keys for ordinary locks, but do not collapse distinct
+    # sources of the same name/version into one scope or adjacency list.
+    keys = {
+        identity: f"{name}@{version}" + (
+            f"?source={quote(identity[2], safe='')}" if len(identities) > 1 else ""
+        )
+        for (name, version), identities in registry_ids.items()
+        for identity in identities
+    }
 
     runtime_roots = list(root.get("dependencies", []))
     development_roots: list[dict] = []
@@ -64,33 +106,31 @@ def uv_manifest(lock: dict, location: str = "uv.lock") -> dict:
         target = development_roots if group in DEVELOPMENT_EXTRAS else runtime_roots
         target.extend(dependencies)
 
-    direct_runtime = {dependency["name"] for dependency in runtime_roots}
-    direct_development = {dependency["name"] for dependency in development_roots}
+    direct = {
+        _package_id(_resolve_dependency(dependency, packages_by_name))
+        for dependency in runtime_roots + development_roots
+    }
     runtime = _reachable(runtime_roots, packages_by_name)
     development = _reachable(development_roots, packages_by_name)
     resolved = {}
     for package in packages:
         if "registry" not in package.get("source", {}):
             continue
+        identity = _package_id(package)
         name, version = package["name"], package["version"]
-        key = f"{name}@{version}"
-        if key in resolved:
-            raise ValueError(f"Duplicate uv package: {key}")
+        key = keys[identity]
         dependency_keys = set()
-        for dependency in package.get("dependencies", []):
-            candidates = packages_by_name.get(dependency["name"], [])
-            if "version" in dependency:
-                candidates = [item for item in candidates if item["version"] == dependency["version"]]
-            if not candidates:
-                raise ValueError(f"Unresolved uv dependency: {name} -> {dependency['name']}")
-            dependency_keys.update(
-                f"{item['name']}@{item['version']}"
-                for item in candidates if "registry" in item.get("source", {})
-            )
+        dependencies = list(package.get("dependencies", []))
+        for extra in runtime.get(identity, set()) | development.get(identity, set()):
+            dependencies.extend(package.get("optional-dependencies", {}).get(extra, []))
+        for dependency in dependencies:
+            target = _package_id(_resolve_dependency(dependency, packages_by_name))
+            if target in keys:
+                dependency_keys.add(keys[target])
         entry = {
             "package_url": f"pkg:pypi/{quote(name, safe='')}@{quote(version, safe='')}",
-            "relationship": "direct" if name in direct_runtime | direct_development else "indirect",
-            "scope": "development" if name in development and name not in runtime else "runtime",
+            "relationship": "direct" if identity in direct else "indirect",
+            "scope": "development" if identity in development and identity not in runtime else "runtime",
             "dependencies": sorted(dependency_keys),
         }
         resolved[key] = entry

tests/unit/test_submit_uv_lock_snapshot.py

@@ -76,6 +76,143 @@ def test_requested_runtime_extra_overrides_dev_reachability() -> None:
     assert resolved["watchfiles@2.0"]["scope"] == "runtime"
 
 
+def _split_lock() -> dict:
+    registry = {"registry": "https://pypi.org/simple"}
+    return {
+        "version": 1,
+        "package": [
+            {
+                "name": "example", "version": "1.0", "source": {"editable": "."},
+                "dependencies": [{"name": "split", "version": "1.0", "source": registry}],
+                "optional-dependencies": {
+                    "dev": [{"name": "split", "version": "2.0", "source": registry}],
+                },
+            },
+            {"name": "split", "version": "1.0", "source": registry},
+            {
+                "name": "split", "version": "2.0", "source": registry,
+                "dependencies": [{"name": "dev-child"}],
+            },
+            {"name": "dev-child", "version": "1.0", "source": registry},
+        ],
+    }
+
+
+def test_split_versions_keep_development_subgraph_out_of_runtime() -> None:
+    resolved = MODULE["uv_manifest"](_split_lock())["resolved"]
+    assert {key: (row["scope"], row["relationship"]) for key, row in resolved.items()} == {
+        "split@1.0": ("runtime", "direct"),
+        "split@2.0": ("development", "direct"),
+        "dev-child@1.0": ("development", "indirect"),
+    }
+    assert resolved["split@1.0"]["dependencies"] == []
+    assert resolved["split@2.0"]["dependencies"] == ["dev-child@1.0"]
+
+
+def test_direct_relationship_is_specific_to_the_locked_version() -> None:
+    lock = _split_lock()
+    root, runtime = lock["package"][:2]
+    runtime_edge = root["dependencies"][0]
+    root["dependencies"] = [{"name": "parent"}]
+    lock["package"].append({
+        "name": "parent", "version": "1.0", "source": runtime["source"],
+        "dependencies": [runtime_edge],
+    })
+    resolved = MODULE["uv_manifest"](lock)["resolved"]
+    assert resolved["parent@1.0"]["dependencies"] == ["split@1.0"]
+    assert resolved["split@1.0"]["relationship"] == "indirect"
+    assert resolved["split@2.0"]["relationship"] == "direct"
+
+
+def test_same_version_sources_keep_scopes_extras_and_edges_separate() -> None:
+    lock = _split_lock()
+    root, runtime, development, child = lock["package"]
+    development["version"] = "1.0"
+    development["source"] = {"registry": "https://other.example/simple"}
+    development["optional-dependencies"] = {"feature": development.pop("dependencies")}
+    root["optional-dependencies"]["dev"] = [{
+        "name": "split", "version": "1.0", "source": development["source"],
+        "extra": ["feature"],
+    }]
+    # The same extra on the runtime source is NOT requested.
+    runtime["optional-dependencies"] = {"feature": [{"name": "unrequested"}]}
+    root["optional-dependencies"]["dev"].append({"name": "unrequested"})
+    lock["package"].append({**child, "name": "unrequested"})
+    # A transitive edge must point to the selected source's unique key too.
+    root["dependencies"].append({"name": "parent"})
+    lock["package"].append({
+        **child, "name": "parent", "dependencies": [root["dependencies"][0]],
+    })
+    resolved = MODULE["uv_manifest"](lock)["resolved"]
+    split = {key: row for key, row in resolved.items() if row["package_url"] == "pkg:pypi/split@1.0"}
+    assert len(split) == 2
+    assert {row["scope"] for row in split.values()} == {"runtime", "development"}
+    assert all(row["relationship"] == "direct" for row in split.values())
+    assert next(row for row in split.values() if row["scope"] == "runtime")["dependencies"] == []
+    assert next(row for row in split.values() if row["scope"] == "development")["dependencies"] == ["dev-child@1.0"]
+    assert resolved["dev-child@1.0"]["scope"] == "development"
+    assert resolved["unrequested@1.0"]["scope"] == "development"
+    runtime_key = next(key for key, row in split.items() if row["scope"] == "runtime")
+    assert resolved["parent@1.0"]["dependencies"] == [runtime_key]
+    lock["package"].reverse()
+    assert MODULE["uv_manifest"](lock)["resolved"] == resolved
+
+
+def test_extras_are_specific_to_version_and_scope() -> None:
+    lock = _split_lock()
+    root, runtime, development, child = lock["package"]
+    development["optional-dependencies"] = {"feature": development.pop("dependencies")}
+    runtime["optional-dependencies"] = {"feature": [{"name": "runtime-extra"}]}
+    root["optional-dependencies"]["dev"][0]["extra"] = ["feature"]
+    root["optional-dependencies"]["dev"].append({"name": "runtime-extra"})
+    lock["package"].append({**child, "name": "runtime-extra"})
+    resolved = MODULE["uv_manifest"](lock)["resolved"]
+    assert resolved["dev-child@1.0"]["scope"] == "development"
+    assert resolved["runtime-extra@1.0"]["scope"] == "development"
+    assert resolved["split@1.0"]["dependencies"] == []
+    assert resolved["split@2.0"]["dependencies"] == ["dev-child@1.0"]
+    # Sharing an identity across scopes must not promote its dev-only extra.
+    root["dependencies"] = [{"name": "split", "version": "2.0"}]
+    resolved = MODULE["uv_manifest"](lock)["resolved"]
+    assert resolved["split@2.0"]["scope"] == "runtime"
+    assert resolved["dev-child@1.0"]["scope"] == "development"
+
+
+def test_cycles_allow_later_requests_for_new_extras() -> None:
+    lock = _split_lock()
+    root, runtime, development, child = lock["package"]
+    runtime["dependencies"] = [{"name": "parent"}]
+    runtime["optional-dependencies"] = {"feature": [{"name": "dev-child"}]}
+    lock["package"].append({
+        **child, "name": "parent", "dependencies": [
+            {"name": "split", "version": "1.0", "extra": ["feature"]},
+        ],
+    })
+    resolved = MODULE["uv_manifest"](lock)["resolved"]
+    assert resolved["dev-child@1.0"]["scope"] == "runtime"
+    assert resolved["split@2.0"]["scope"] == "development"
+    assert resolved["split@1.0"]["dependencies"] == ["dev-child@1.0", "parent@1.0"]
+
+
+@pytest.mark.parametrize("edge, error", [
+    ({"name": "split"}, "Ambiguous uv dependency"),
+    ({"name": "split", "version": "3.0"}, "Unresolved uv dependency"),
+    ({"name": "split", "version": "1.0", "source": {"registry": "https://missing.example"}}, "Unresolved uv dependency"),
+])
+def test_uv_manifest_rejects_unresolved_or_ambiguous_identity(edge: dict, error: str) -> None:
+    lock = _split_lock()
+    lock["package"][0]["dependencies"] = [edge]
+    with pytest.raises(ValueError, match=error):
+        MODULE["uv_manifest"](lock)
+
+
+def test_uv_manifest_rejects_duplicate_locked_identity() -> None:
+    lock = _split_lock()
+    lock["package"].append(dict(lock["package"][1]))
+    with pytest.raises(ValueError, match="Duplicate uv package"):
+        MODULE["uv_manifest"](lock)
+
+
 def test_current_uv_lock_keeps_dev_only_changes_out_of_runtime() -> None:
     with (ROOT / "uv.lock").open("rb") as stream:
         resolved = MODULE["uv_manifest"](tomllib.load(stream))["resolved"]