triage: deterministic dependency-manifest resolver (dep_manifest + Lane A)

backend/cliff/agents/triage_deep/integration.py

@@ -75,7 +75,7 @@ async def maybe_deep_dive(
 
     knowledge = {
         name: mgr.read_artifact(repo.id, name)
-        for name in ("profile", "code_map", "threat")
+        for name in ("profile", "code_map", "threat", "dep_manifest")
     }
     active_runner = runner or DeepDiveRunner(
         build_tier_models(ai_env, model_full_id),

backend/cliff/agents/triage_deep/runner.py

@@ -272,6 +272,7 @@ async def _run(
             "profile": repo_knowledge.get("profile"),
             "code_map": repo_knowledge.get("code_map"),
             "threat": repo_knowledge.get("threat"),
+            "dep_manifest": repo_knowledge.get("dep_manifest"),
         }
 
         def prov(exit_stage: str) -> TriageProvenance:

backend/cliff/agents/triage_dep_manifest.py

@@ -0,0 +1,139 @@
+"""Deterministic dependency-manifest resolver (SP1 / ADR-0053 dep_manifest).
+
+Clears a **dependency** finding (``location`` shape ``pkg@version``) as
+``false_positive`` BEFORE the LLM Deep dive when the package provably does not
+ship: every root reaching its ``(name, version)`` node is dev/test/docs/build
+(no prod, no optional) AND it is not imported by any first-party shipping source.
+
+Pure — no LLM, no network, no filesystem — keyless and CI-testable. This is a
+*structural* clear like ``code_map`` (facts from the lockfile graph, not a model
+hunch), so it is tier-independent under ADR-0054.
+
+Safety (never clear a shipping dependency): the clear requires TWO independent
+gates to both hold — non-prod scope AND zero shipping import sites — plus a
+resolved import name. Transitive reachability of a *vulnerable API* is NOT
+decided here (that is Lane B / the Deep dive); this resolver only clears the
+"doesn't ship at all" case.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from cliff.agents.schemas import TriageCheck, TriageOutput, TriageProvenance
+
+_CONF_DEP_CLEAR = 0.9
+
+#: The ONLY scopes that are safe to clear (allowlist, not denylist). A node may be
+#: cleared only if *every* scope reaching it is one of these — any other value
+#: ("prod", "optional", or an unrecognized/future scope) blocks the clear, so a new
+#: shipping scope can never silently pass the gate.
+_CLEARABLE_SCOPES = frozenset({"dev", "test", "docs", "build"})
+
+
+def _parse_location(loc: str) -> tuple[str, str] | None:
+    """``"pkg@version"`` -> ``(name, version)``; scoped npm ``@scope/name@ver`` handled.
+
+    Splits on the LAST ``@`` so the leading ``@`` of a scoped name is preserved.
+    """
+    s = loc.strip()
+    if "@" not in s:
+        return None
+    name, _, version = s.rpartition("@")
+    if not name or not version:
+        return None
+    return name, version
+
+
+def _pep503(name: str) -> str:
+    """PEP 503 canonical form (lowercase; runs of -_. → single -)."""
+    return re.sub(r"[-_.]+", "-", name).lower()
+
+
+def _name_matches(node: dict[str, Any], raw_name: str) -> bool:
+    """The finding's scanner-reported name vs a node's stored name.
+
+    pypi nodes are stored PEP-503-normalized (``pyyaml``) but scanners report raw
+    names (``PyYAML``, ``ruamel.yaml``) — normalize before comparing. npm names
+    are compared verbatim (they are not PEP-503-normalized; ``lodash.merge`` ≠
+    ``lodash-merge``).
+    """
+    node_name = node.get("name")
+    if node.get("ecosystem") == "pypi":
+        return isinstance(node_name, str) and _pep503(raw_name) == node_name
+    return raw_name == node_name
+
+
+def _clear(*, detail: str) -> TriageOutput:
+    return TriageOutput(
+        verdict="false_positive",
+        confidence=_CONF_DEP_CLEAR,
+        checks=[
+            TriageCheck(
+                eyebrow="Dependency does not ship",
+                result="dev/test-only dependency",
+                kind="pass",
+                detail=detail,
+            )
+        ],
+        provenance=TriageProvenance(
+            steps_run=["dep_manifest_resolver"],
+            exit_stage="dep_manifest_resolver",
+            escalated=False,
+        ),
+    )
+
+
+def resolve_by_dep_manifest(
+    finding: dict[str, Any], dep_manifest: dict[str, Any] | None
+) -> TriageOutput | None:
+    """Clear *finding* if its ``pkg@version`` is a dev/test-only, unimported
+    dependency; else ``None`` (fall through to Lane B / the Deep dive)."""
+    if not dep_manifest:
+        return None
+    loc = finding.get("location")
+    if not isinstance(loc, str):
+        return None
+    parsed = _parse_location(loc)
+    if parsed is None:
+        return None
+    name, version = parsed
+
+    nodes = dep_manifest.get("nodes")
+    if not isinstance(nodes, list):
+        return None
+    matches = [
+        n
+        for n in nodes
+        if isinstance(n, dict) and _name_matches(n, name) and n.get("version") == version
+    ]
+    if len(matches) != 1:
+        # not found, or a cross-ecosystem (name, version) collision → conservative
+        return None
+    node = matches[0]
+
+    scopes = node.get("scopes")
+    if not isinstance(scopes, list) or not scopes:
+        return None  # unknown reachability → cannot prove non-prod → don't clear
+    scope_set = {s for s in scopes if isinstance(s, str)}
+    if not scope_set <= _CLEARABLE_SCOPES:
+        return None  # a prod/optional/unknown scope reaches it → don't clear (allowlist)
+
+    # Import-site gate — fail CLOSED: only a present, empty list proves "not imported".
+    # A missing / non-list / non-empty import_sites blocks the clear (pitfall 1).
+    import_sites = node.get("import_sites")
+    if not isinstance(import_sites, list) or import_sites:
+        return None  # imported in first-party shipping code, or untrustworthy → don't clear
+
+    if not node.get("import_name"):
+        # import name unresolved → couldn't run the import gate → don't clear (pitfall 3)
+        return None
+
+    return _clear(
+        detail=(
+            f"{name}@{version} ({node.get('ecosystem', '?')}) reaches only "
+            f"{sorted(scope_set)} roots and is imported by no shipping first-party "
+            f"source (declared_in={node.get('declared_in') or 'transitive'})"
+        )
+    )

backend/cliff/agents/triage_runner.py

@@ -22,6 +22,7 @@
 from cliff.agents.sidebar_mapper import map_and_upsert
 from cliff.agents.triage_codemap import resolve_by_code_map
 from cliff.agents.triage_deep.integration import maybe_deep_dive
+from cliff.agents.triage_dep_manifest import resolve_by_dep_manifest
 from cliff.db.repo_agent_run import (
     create_agent_run,
     list_latest_runs_by_workspace_ids,
@@ -107,29 +108,37 @@ def _structured_output(runs: dict[str, Any], agent_type: str) -> dict[str, Any]
     return run.structured_output if run is not None else None
 
 
-async def _load_code_map(db: aiosqlite.Connection, repo_url: str | None) -> dict[str, Any] | None:
-    """The repo's cached code_map, or None when there's no ready profile — the
-    same resolution the Deep dive uses (cliff/agents/triage_deep/integration.py).
+async def _load_artifact(
+    db: aiosqlite.Connection, repo_url: str | None, name: str
+) -> dict[str, Any] | None:
+    """The repo's cached profile artifact *name*, or None when there's no ready
+    profile — the same resolution the Deep dive uses
+    (cliff/agents/triage_deep/integration.py).
 
-    Returns None (falls through to Deep dive) on any read/parse failure, or
-    when the artifact is not a dict — triage must never crash on a corrupt
-    code_map.
+    Returns None (falls through to Deep dive) on any read/parse failure, or when
+    the artifact is not a dict — triage must never crash on a corrupt artifact.
     """
     if not repo_url:
         return None
     repo = await get_repo_by_url(db, repo_url)
     if repo is None or repo.profile_status != "ready":
         return None
     try:
-        result = default_repo_dir_manager().read_artifact(repo.id, "code_map")
+        result = default_repo_dir_manager().read_artifact(repo.id, name)
     except Exception:
-        logger.debug(
-            "code_map unreadable for repo %s — falling through to Deep dive", repo_url
-        )
-        return None
-    if not isinstance(result, dict):
+        logger.debug("%s unreadable for repo %s — falling through to Deep dive", name, repo_url)
         return None
-    return result
+    return result if isinstance(result, dict) else None
+
+
+async def _load_code_map(db: aiosqlite.Connection, repo_url: str | None) -> dict[str, Any] | None:
+    return await _load_artifact(db, repo_url, "code_map")
+
+
+async def _load_dep_manifest(
+    db: aiosqlite.Connection, repo_url: str | None
+) -> dict[str, Any] | None:
+    return await _load_artifact(db, repo_url, "dep_manifest")
 
 
 async def run_triage(
@@ -262,6 +271,22 @@ async def _run_scanner_triage(
         await _persist_synthesis(db, workspace.id, cleared)
         return cleared
 
+    # Deterministic dep_manifest gate (SP1): clear a dependency finding whose
+    # package provably does not ship (dev/test-only, unimported). The dep finding
+    # carries pkg@version in raw_payload; code_map's file-path gate never matches
+    # it, so this runs only for type=="dependency" and is a pure structural clear.
+    if finding.type == "dependency":
+        rp = finding.raw_payload or {}
+        pkg, ver = rp.get("package"), rp.get("version")
+        if pkg and ver:
+            dep_manifest = await _load_dep_manifest(db, workspace.repo_url)
+            dep_cleared = resolve_by_dep_manifest(
+                {**finding_ctx, "location": f"{pkg}@{ver}"}, dep_manifest
+            )
+            if dep_cleared is not None:
+                await _persist_synthesis(db, workspace.id, dep_cleared)
+                return dep_cleared
+
     # Escalate to the agentic Deep dive when warranted (ADR-0052). Best-effort:
     # any failure keeps the cheap Quick-read verdict — triage never breaks.
     final = quick

backend/cliff/repos/deps/__init__.py

@@ -0,0 +1,16 @@
+"""Deterministic dependency-manifest builder (ADR-0053 §3, dep_manifest artifact).
+
+Parses lockfiles/manifests across all workspaces into a graph keyed by
+``(name, version)``, propagates scopes (prod dominates), finds first-party import
+sites, and emits a ``DepManifest``. Pure — no LLM, no network.
+
+The public entry point ``build_dep_manifest`` is assembled in ``build.py`` and
+re-exported here (SP1 Task 7).
+"""
+
+from __future__ import annotations
+
+from .build import build_dep_manifest
+from .graph import DepGraph, Root
+
+__all__ = ["DepGraph", "Root", "build_dep_manifest"]

backend/cliff/repos/deps/build.py

@@ -0,0 +1,104 @@
+"""Assemble the ``dep_manifest`` artifact (SP1 Task 7).
+
+Discover manifests → parse npm + pypi graphs (each ecosystem independently) →
+propagate scopes → resolve import names → attach first-party import sites (one
+walk per ecosystem) → emit a ``DepManifest`` dict.
+
+Robustness: this builder runs in-process at scan time and parses *untrusted*
+lockfiles. It MUST NOT raise — an uncaught parser error would fail the whole
+profile build (``profile_status='error'``), which silently disables the code_map
+gate, the dep_manifest gate, AND the Deep dive for that repo. Any per-ecosystem
+failure is caught and recorded in ``unresolved``; the artifact is still emitted.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from ..schemas import DepManifest, DepNode
+from .imports import collect_import_sites
+from .manifests import discover_manifests
+from .npm import parse_npm
+from .pypi import dist_to_import, parse_pypi
+
+logger = logging.getLogger(__name__)
+
+
+def _import_name(name: str, ecosystem: str) -> str | None:
+    if ecosystem == "npm":
+        return name  # npm import name == package name (scoped kept)
+    if ecosystem == "pypi":
+        return dist_to_import(name)
+    return None
+
+
+def _lookup_key(iname: str, ecosystem: str) -> str:
+    # pypi import sites are bucketed by TOP-LEVEL module; a dotted import name
+    # (``google.protobuf``) must be looked up by its first component (``google``).
+    return iname.split(".")[0] if ecosystem == "pypi" else iname
+
+
+def _ecosystem_nodes(root: Path, ms: list, parser, eco: str) -> tuple[list[dict], list[str]]:
+    """Parse one ecosystem into DepNode dicts + its unresolved manifests. Never raises."""
+    try:
+        graph, unresolved = parser(root, ms)
+        node_map = graph.nodes
+        if not node_map:
+            return [], list(unresolved)
+        scopes = graph.propagate_scopes()
+        direct = graph.direct_nodes()
+        declared = graph.declared_in_map()
+        imports = collect_import_sites(root, eco)  # one walk per ecosystem
+        nodes: list[dict] = []
+        for (name, version), node_eco in node_map.items():
+            iname = _import_name(name, node_eco)
+            sites = imports.get(_lookup_key(iname, node_eco), []) if iname else []
+            nodes.append(
+                DepNode(
+                    name=name,
+                    version=version,
+                    ecosystem=node_eco,
+                    scopes=sorted(scopes.get((name, version), set())),
+                    direct=(name, version) in direct,
+                    declared_in=declared.get((name, version), []),
+                    import_name=iname,
+                    import_sites=sites,
+                ).model_dump()
+            )
+        return nodes, list(unresolved)
+    except Exception:  # noqa: BLE001 - a bad lockfile must not fail the whole profile
+        logger.warning(
+            "dep_manifest: %s parse failed — recording as unresolved", eco, exc_info=True
+        )
+        return [], [f"<{eco}-parse-error>"]
+
+
+def build_dep_manifest(root) -> dict:  # noqa: ANN001 - Path | str
+    root = Path(root)
+    try:
+        manifests = discover_manifests(root)
+    except Exception:  # noqa: BLE001 - never let discovery crash the profile build
+        logger.warning("dep_manifest: manifest discovery failed", exc_info=True)
+        return DepManifest(unresolved=["<discovery-error>"]).model_dump()
+
+    npm_ms = [m for m in manifests if m.ecosystem == "npm"]
+    py_ms = [m for m in manifests if m.ecosystem == "pypi"]
+    nodes: list[dict] = []
+    unresolved: list[str] = []
+    ecosystems: list[str] = []
+    for ms, parser, eco in ((npm_ms, parse_npm, "npm"), (py_ms, parse_pypi, "pypi")):
+        if not ms:
+            continue
+        eco_nodes, eco_unresolved = _ecosystem_nodes(root, ms, parser, eco)
+        unresolved.extend(eco_unresolved)
+        if eco_nodes:
+            ecosystems.append(eco)
+            nodes.extend(eco_nodes)
+
+    return DepManifest(
+        ecosystems=ecosystems,
+        workspaces=[m.path for m in manifests],
+        nodes=nodes,
+        unresolved=unresolved,
+    ).model_dump()