handle resume tokens gracefully (#1097)

* Fix telemetry deltas for resumed runs

* Fix resumed telemetry duration

strix/interface/utils.py

@@ -133,6 +133,27 @@ def format_vulnerability_report(report: dict[str, Any]) -> Text:  # noqa: PLR091
             text.append("CVSS Vector: ", style=field_style)
             text.append("/".join(cvss_parts), style="dim")
 
+    dependency_metadata = report.get("dependency_metadata") or {}
+    if dependency_metadata:
+        contextual_vector = dependency_metadata.get("contextual_cvss_vector")
+        if contextual_vector:
+            text.append("\n\n")
+            text.append("Contextual CVSS Vector: ", style=field_style)
+            text.append(contextual_vector, style="dim")
+
+        advisory_cvss = dependency_metadata.get("advisory_cvss")
+        if advisory_cvss is not None and advisory_cvss != report.get("cvss"):
+            text.append("\n\n")
+            text.append("Advisory CVSS: ", style=field_style)
+            text.append(f"{float(advisory_cvss):.1f}", style="dim")
+
+        contextual_reasoning = dependency_metadata.get("contextual_cvss_reasoning")
+        if contextual_reasoning:
+            text.append("\n\n")
+            text.append("Contextual CVSS Reasoning", style=field_style)
+            text.append("\n")
+            text.append(contextual_reasoning)
+
     description = report.get("description")
     if description:
         text.append("\n\n")

strix/report/pricing.py

@@ -0,0 +1,54 @@
+"""LiteLLM model-name resolution for local cost estimates."""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import Any, cast
+
+
+@lru_cache(maxsize=512)
+def resolve_litellm_model(model: str) -> str | None:
+    """Return a provider-qualified model name that LiteLLM can price."""
+    try:
+        import litellm
+
+        normalized = model.strip()
+        for prefix in ("litellm/", "any-llm/", "openai/"):
+            if normalized.startswith(prefix):
+                normalized = normalized.removeprefix(prefix)
+                break
+        if not normalized:
+            return None
+
+        model_cost = cast(
+            "dict[str, dict[str, Any]]",
+            getattr(litellm, "model_cost"),  # noqa: B009
+        )
+        bare_entry = model_cost.get(normalized)
+        if "/" not in normalized and isinstance(bare_entry, dict):
+            provider = bare_entry.get("litellm_provider")
+            if isinstance(provider, str) and provider:
+                return f"{provider}/{normalized}"
+        if "/" in normalized and isinstance(bare_entry, dict):
+            return normalized
+
+        names = [normalized]
+        if "/" in normalized:
+            names.append(normalized.rsplit("/", 1)[-1])
+        for name in names:
+            matches = sorted(key for key in model_cost if key.endswith(f"/{name}"))
+            if not matches:
+                continue
+            prices = {
+                (
+                    model_cost[key].get("input_cost_per_token"),
+                    model_cost[key].get("output_cost_per_token"),
+                )
+                for key in matches
+                if isinstance(model_cost.get(key), dict)
+            }
+            if len(matches) == 1 or len(prices) == 1:
+                return matches[0]
+        return None  # noqa: TRY300
+    except Exception:  # noqa: BLE001
+        return None

strix/report/state.py

@@ -14,6 +14,7 @@ from agents.usage import Usage
 from strix.config import codex
 from strix.config.loader import load_settings
 from strix.core.paths import run_dir_for
+from strix.report.pricing import resolve_litellm_model
 from strix.report.sarif import write_sarif
 from strix.report.usage import LLMUsageLedger
 from strix.report.writer import (
@@ -38,6 +39,13 @@ def _strix_version() -> str | None:
         return None
 
 
+def _number(value: Any) -> int | float:
+    try:
+        return float(value or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
 def _parse_repo_full_name(uri: str) -> str | None:
     """Extract ``owner/repo`` from a git URL or slug, else None."""
     text = uri.strip().removesuffix(".git")
@@ -114,6 +122,7 @@ class ReportState:
         self.run_name = run_name
         self.run_id = run_name or f"run-{uuid4().hex[:8]}"
         self.start_time = datetime.now(UTC).isoformat()
+        self.process_start_time = self.start_time
         self.end_time: str | None = None
 
         self.vulnerability_reports: list[dict[str, Any]] = []
@@ -122,6 +131,7 @@ class ReportState:
         self.scan_results: dict[str, Any] | None = None
         self.scan_config: dict[str, Any] | None = None
         self._llm_usage = LLMUsageLedger()
+        self._telemetry_llm_usage_baseline: dict[str, Any] = {}
         auth_mode = codex.auth_mode(load_settings().llm.model)
         self._llm_usage.zero_cost = auth_mode == "subscription"
         self.run_record: dict[str, Any] = {
@@ -187,6 +197,7 @@ class ReportState:
                 self.scan_results = scan_results
                 self.final_scan_result = self._format_final_scan_result(scan_results)
             self._hydrate_llm_usage(data.get("llm_usage"))
+            self._telemetry_llm_usage_baseline = self._build_llm_usage_record()
             logger.info("report state hydrated run.json from %s", run_dir)
 
         json_path = run_dir / "vulnerabilities.json"
@@ -330,6 +341,25 @@ class ReportState:
     def get_total_llm_usage(self) -> dict[str, Any]:
         return dict(self.run_record.get("llm_usage") or self._build_llm_usage_record())
 
+    def get_process_llm_usage(self) -> dict[str, int | float]:
+        """Return LLM usage accumulated since this process started."""
+        usage = self._llm_usage.to_record()
+        return {
+            key: max(
+                0, _number(usage.get(key)) - _number(self._telemetry_llm_usage_baseline.get(key))
+            )
+            for key in ("requests", "input_tokens", "output_tokens", "total_tokens", "cost")
+        }
+
+    def get_process_duration_seconds(self) -> float:
+        """Return this process's elapsed wall time for telemetry."""
+        try:
+            start = datetime.fromisoformat(self.process_start_time.replace("Z", "+00:00"))
+            duration = (datetime.now(start.tzinfo) - start).total_seconds()
+            return max(0.0, duration)
+        except (ValueError, TypeError, AttributeError):
+            return 0.0
+
     def get_total_llm_cost(self) -> float:
         """Live accumulated LLM cost, independent of the persisted run-record snapshot."""
         return self._llm_usage.total_cost
@@ -696,10 +726,13 @@ def _estimate_response_cost(kwargs: Any, completion_response: Any) -> float | No
         candidates.append(model.rsplit("/", 1)[-1])
 
     for candidate in candidates:
+        resolved = resolve_litellm_model(candidate)
+        if not resolved:
+            continue
         try:
             value = completion_cost(
-                completion_response={"model": candidate, "usage": usage_payload},
-                model=candidate,
+                completion_response={"model": resolved, "usage": usage_payload},
+                model=resolved,
             )
         except Exception:  # nosec B112  # noqa: BLE001, S112
             continue

strix/report/usage.py

@@ -7,6 +7,8 @@ from typing import Any
 
 from agents.usage import Usage, deserialize_usage, serialize_usage
 
+from strix.report.pricing import resolve_litellm_model
+
 
 logger = logging.getLogger(__name__)
 
@@ -18,7 +20,9 @@ class LLMUsageLedger:
         self._total_usage = Usage()
         self._agent_usage: dict[str, Usage] = {}
         self._agent_metadata: dict[str, dict[str, str]] = {}
-        self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._estimated_cost = 0.0
+        self._has_observed_cost = False
         # When True, tokens are still tracked but cost stays $0 — the run is on a
         # model subscription, so there is no metered per-token charge to report.
         self.zero_cost = False
@@ -44,10 +48,10 @@ class LLMUsageLedger:
         if model:
             metadata["model"] = model
 
-        if not self.zero_cost and not _is_litellm_routed(model):
+        if not self.zero_cost:
             estimated = _estimate_litellm_cost(usage, model)
             if estimated:
-                self._total_cost += estimated
+                self._estimated_cost += estimated
 
         return True
 
@@ -55,15 +59,18 @@ class LLMUsageLedger:
         if self.zero_cost:
             return
         if isinstance(cost, int | float) and cost > 0:
-            self._total_cost += float(cost)
+            self._observed_cost += float(cost)
+            self._has_observed_cost = True
 
     @property
     def total_cost(self) -> float:
-        return _round_cost(self._total_cost)
+        if self.zero_cost:
+            return 0.0
+        return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
 
     def to_record(self) -> dict[str, Any]:
         record = serialize_usage(self._total_usage)
-        record["cost"] = _round_cost(self._total_cost)
+        record["cost"] = self.total_cost
         record["agents"] = []
 
         agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
@@ -72,7 +79,7 @@ class LLMUsageLedger:
             usage = self._agent_usage[agent_id]
             metadata = self._agent_metadata.get(agent_id, {})
             agent_cost = (
-                self._total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
+                self.total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
             )
 
             agent_record = serialize_usage(usage)
@@ -92,7 +99,9 @@ class LLMUsageLedger:
         self._total_usage = Usage()
         self._agent_usage.clear()
         self._agent_metadata.clear()
-        self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._estimated_cost = 0.0
+        self._has_observed_cost = False
 
         if not isinstance(raw_usage, dict):
             return
@@ -103,7 +112,9 @@ class LLMUsageLedger:
             logger.exception("Failed to hydrate aggregate llm_usage from run.json")
             self._total_usage = Usage()
 
-        self._total_cost = _float_or_zero(raw_usage.get("cost"))
+        persisted_cost = _float_or_zero(raw_usage.get("cost"))
+        self._observed_cost = persisted_cost
+        self._estimated_cost = persisted_cost
 
         for raw_agent in raw_usage.get("agents") or []:
             if not isinstance(raw_agent, dict):
@@ -136,15 +147,6 @@ def _resolve_total_tokens(usage: Usage) -> int:
     return prompt + completion
 
 
-def _is_litellm_routed(model: str | None) -> bool:
-    if not model:
-        return False
-    name = model.strip().lower()
-    if "/" not in name:
-        return False
-    return not name.startswith("openai/")
-
-
 def _usage_has_activity(usage: Usage) -> bool:
     return bool(
         usage.requests
@@ -201,24 +203,23 @@ def _estimate_litellm_entry_cost(entry: Any, model: str) -> float | None:
 
     candidates = [model]
     if "/" in model:
-        candidates.append(model.split("/", 1)[-1])
+        candidates.append(model.rsplit("/", 1)[-1])
 
-    cost: Any = None
     for candidate in candidates:
+        resolved = resolve_litellm_model(candidate)
+        if not resolved:
+            continue
         try:
             cost = completion_cost(
-                completion_response={"model": candidate, "usage": usage_payload},
-                model=model,
+                completion_response={"model": resolved, "usage": usage_payload},
+                model=resolved,
             )
-            break
         except Exception:  # nosec B112  # noqa: BLE001, S112
             continue
-
-    if cost is None:
-        logger.debug("LiteLLM cost estimate unavailable for model %s", model)
-        return None
-
-    return cost if isinstance(cost, int | float) and cost >= 0 else None
+        if cost > 0:
+            return float(cost)
+    logger.debug("LiteLLM cost estimate unavailable for model %s", model)
+    return None
 
 
 def _litellm_model_name(model: str | None) -> str | None:

strix/report/writer.py

@@ -215,6 +215,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
     cvss = report.get("cvss")
     if cvss is not None:
         metadata.append(("CVSS", cvss))
+    advisory_cvss = dep_meta.get("advisory_cvss")
+    if advisory_cvss is not None and advisory_cvss != cvss:
+        metadata.append(("Advisory CVSS", advisory_cvss))
+    if dep_meta.get("contextual_cvss_vector"):
+        metadata.append(("Contextual CVSS Vector", dep_meta["contextual_cvss_vector"]))
     if report.get("fix_effort"):
         metadata.append(("Fix Effort", str(report["fix_effort"]).title()))
     for label, value in metadata:
@@ -241,6 +246,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         lines.append(str(report["technical_analysis"]))
         lines.append("")
 
+    if dep_meta.get("contextual_cvss_reasoning"):
+        lines.append("## Contextual CVSS\n")
+        lines.append(str(dep_meta["contextual_cvss_reasoning"]))
+        lines.append("")
+
     if report.get("poc_description") or report.get("poc_script_code"):
         lines.append("## Proof of Concept\n")
         if report.get("poc_description"):

strix/skills/custom/dependency_cve_scanning.md

@@ -161,7 +161,23 @@ fi
    verdict/evidence onto its siblings; run the symbol search against each
    CVE's own affected-symbol list. The import check (step 1) is the only
    part shared across a package's CVEs.
-3. If the analysis was not performed or is inconclusive (obfuscated code,
+3. **Source-to-sink trace — do this whenever step 2 found a symbol hit.** A
+   symbol hit alone says the code calls the vulnerable API; it does not say
+   who can reach it. Start at the sink (the exact line that calls the
+   vulnerable function) and walk backwards hop by hop to the source: the
+   entry point that carries untrusted input (HTTP route, CLI argument, queue
+   or webhook payload, uploaded file, config value). Read each intermediate
+   function; when a hop is a thin wrapper, go one step deeper — never stop at
+   the first caller. Record what each hop enforces: authentication, a role
+   check, validation, a feature flag, a size or type limit, a default that is
+   off in production.
+   Write the chain into `reachability_evidence` as
+   `entry point -> intermediate call -> package call` with a
+   repository-relative `file:line` for every hop, and say who controls the
+   input. If no source reaches the sink, say that too — the level stays
+   `vulnerable_symbol_used` (the call is real), and the trace is what tells
+   the reader it is only reachable from, say, an operator CLI.
+4. If the analysis was not performed or is inconclusive (obfuscated code,
    dynamic loading, unparsable sources) ⇒ `unknown` and say why in
    `assumptions`.
 
@@ -225,15 +241,83 @@ findings and rejects empty PoC fields):
   installed/affected version, fixed version, lockfile path, and the relevant
   trivy output excerpt.
 - **Always set `advisory_cvss` to the published advisory base score (0.0–10.0).**
-  Severity is derived *solely* from this number: read it off the advisory (`CVSS`
-  in trivy output, or the NVD/GHSA page) and pass the real value. The tool rejects
-  a call that omits it, because guessing a score both inflates low CVEs and
-  deflates critical ones.
+  It is the published reference, and it rates the finding whenever you give no
+  contextual breakdown: read it off the advisory (`CVSS` in trivy output, or the
+  NVD/GHSA page) and pass the real value. The tool rejects a call that omits it,
+  because guessing a score both inflates low CVEs and deflates critical ones.
 - Set `cwe` to the most specific `CWE-NNN` when the advisory names one.
 - Do NOT cap severity at LOW just because there is no dynamic reproduction — use
   the advisory score.
-- Set `reachability` + `reachability_evidence` from the usage analysis above;
+- Set `reachability` + `reachability_evidence` from the usage analysis above —
+  the tool rejects a report with no evidence, so for `unknown` write what you
+  searched and why the result is inconclusive;
   use `assumptions` for anything softer (confidence, caveats, analysis limits).
+- **Always set `contextual_cvss_breakdown` + `contextual_cvss_reasoning`.** Every
+  dependency finding carries a contextual rating of the CVE in this codebase
+  (see below). Start from the published metrics and change only what your
+  evidence proves.
+- Set every other field the report accepts when the information exists:
+  `package`, `ecosystem`, `installed_version`, `fixed_version`, `manifest_path`,
+  `introduced_by` for a transitive package, `dependency_path`, `cwe`,
+  `assumptions`, and the remediation instruction. A blank field costs the reader
+  a triage step.
+
+### Contextual CVSS
+
+The published score rates the CVE in the abstract. `contextual_cvss_breakdown`
+rates it **here**, in this codebase, and every dependency report must carry
+one. It is the same 8-metric CVSS v3.1 object as a
+normal finding's `cvss_breakdown` (`attack_vector`, `attack_complexity`,
+`privileges_required`, `user_interaction`, `scope`, `confidentiality`,
+`integrity`, `availability`). You never pass a score: the contextual score and
+vector are computed from the breakdown, and when you provide one it determines
+the finding's severity. `advisory_cvss` stays the published reference.
+
+Start from the advisory's own published metrics and change only what your
+evidence proves is different in this codebase:
+
+- `attack_vector` `N`/`A`/`L`/`P` — as deployed. A library reached only by a
+  local CLI is `L`, not `N`.
+- `attack_complexity` `L`/`H` — raise to `H` when the vulnerable path needs a
+  precondition the code enforces (input validation, a non-default flag, an
+  internal-only route).
+- `privileges_required` `N`/`L`/`H`, `user_interaction` `N`/`R` — what this
+  deployment requires before the path is reachable.
+- `scope` `U`/`C` — whether exploitation here escapes the component boundary.
+- `confidentiality`/`integrity`/`availability` `N`/`L`/`H` — the impact in this
+  codebase. `not_imported` code the build still ships is usually `N` across all
+  three.
+
+Ground every metric in the **source-to-sink trace** from the usage analysis
+(step 3 above), not in a general impression of the package. Derive the metrics
+from that chain: `attack_vector`, `privileges_required`, and `user_interaction`
+come from what the source requires; `attack_complexity` comes from the
+preconditions the hops enforce; `confidentiality`, `integrity`, and
+`availability` come from the data and privileges available at the sink.
+
+When you have no source-to-sink trace, still rate the finding: copy the
+published metrics, change only the metrics the usage level itself proves, and
+say so in the reasoning. For example, for a `not_imported` package that the
+build still ships, keep the published metrics and lower `confidentiality`,
+`integrity`, and `availability` to `N`, because no code path reaches the
+vulnerable symbol. Never invent a hop you did not read.
+
+`contextual_cvss_reasoning` is required with the breakdown. Write two to four
+sentences that another engineer can check without opening the repository. Name
+the chain hop by hop as `entry point -> intermediate call -> package call`, with
+a repository-relative `file:line` for each hop, say who controls the input, and
+say what the contextual rating changes. Example: lowering `attack_vector` to
+`L` and `confidentiality` to `L` with "The only caller of `yaml.load` is
+`parse_manifest` in `scripts/import.py:88`, which `cli/commands.py:212` invokes
+for an operator-supplied path behind the `--allow-unsafe-import` flag that
+`deploy/prod.yaml` never sets. No HTTP route reaches that function, so an
+attacker must already hold shell access on the job host, and the parsed data is
+build metadata rather than customer records."
+
+When the published rating already fits this codebase, repeat the published
+metrics in the breakdown and say in the reasoning that the deployment matches
+the advisory. A contextual rating is a claim you must be able to defend, and it
+never replaces `advisory_cvss` as the published reference.
 
 Verify the CVE with `web_search` when available before reporting. Never guess or
 hallucinate a CVE id.
@@ -244,10 +328,14 @@ hallucinate a CVE id.
   `create_dependency_report`.
 - Do not report a finding without a verified CVE id.
 - Do not batch multiple CVEs into one report.
-- Do not omit `advisory_cvss` — the tool rejects it, and it is the single input
-  that determines dependency severity.
+- Do not omit `advisory_cvss` — the tool rejects it, and it rates every finding
+  that carries no contextual breakdown.
 - Do not silently drop a known CVE because it lacks a dynamic PoC — that is the
   exact failure this skill prevents.
 - Do not downgrade advisory severity for lack of dynamic reproduction.
 - Do not claim a `reachability` level the evidence does not prove — `unknown`
   with a reason is always acceptable; an overclaimed level never is.
+- Do not send a report without `contextual_cvss_breakdown` and
+  `contextual_cvss_reasoning` — the reader rates and ranks the finding with them.
+- Do not use the contextual breakdown to quietly de-rate a CVE you could not
+  analyze. State the limit of the analysis in the reasoning instead.

strix/telemetry/posthog.py

@@ -1,5 +1,4 @@
 import logging
-from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
 import requests
@@ -105,17 +104,11 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
         if sev in vulnerabilities_counts:
             vulnerabilities_counts[sev] += 1
 
-    duration = 0.0
-    try:
-        start = datetime.fromisoformat(report_state.start_time.replace("Z", "+00:00"))
-        end_iso = report_state.end_time or datetime.now(start.tzinfo).isoformat()
-        duration = (datetime.fromisoformat(end_iso.replace("Z", "+00:00")) - start).total_seconds()
-    except (ValueError, TypeError, AttributeError):
-        pass
+    duration = report_state.get_process_duration_seconds()
 
     llm_props: dict[str, int | float] = {}
     try:
-        usage = report_state.get_total_llm_usage()
+        usage = report_state.get_process_llm_usage()
         if isinstance(usage, dict):
             llm_props = {
                 "llm_requests": int(usage.get("requests") or 0),

strix/telemetry/scarf.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import logging
 import urllib.parse
-from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
 import requests
@@ -114,19 +113,11 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
         if sev in vulnerabilities_counts:
             vulnerabilities_counts[sev] += 1
 
-    duration = 0.0
-    try:
-        scan_start = datetime.fromisoformat(report_state.start_time.replace("Z", "+00:00"))
-        end_iso = report_state.end_time or datetime.now(scan_start.tzinfo).isoformat()
-        duration = (
-            datetime.fromisoformat(end_iso.replace("Z", "+00:00")) - scan_start
-        ).total_seconds()
-    except (ValueError, TypeError, AttributeError):
-        pass
+    duration = report_state.get_process_duration_seconds()
 
     llm_props: dict[str, int | float] = {}
     try:
-        usage = report_state.get_total_llm_usage()
+        usage = report_state.get_process_llm_usage()
         if isinstance(usage, dict):
             llm_props = {
                 "llm_requests": int(usage.get("requests") or 0),

strix/tools/reporting/tool.py

@@ -749,6 +749,70 @@ def _validate_manifest_path(manifest_path: str | None) -> str | None:
     return None
 
 
+_MAX_CONTEXTUAL_REASONING_CHARS = 2000
+
+
+def _validate_contextual_cvss(
+    breakdown: dict[str, str] | None,
+    reasoning: str | None,
+) -> list[str]:
+    errors: list[str] = []
+    if not breakdown:
+        errors.append(
+            "contextual_cvss_breakdown is required: rate the CVE in this codebase with "
+            "all 8 CVSS v3.1 metrics (attack_vector, attack_complexity, "
+            "privileges_required, user_interaction, scope, confidentiality, integrity, "
+            "availability). When your trace does not change the published rating, repeat "
+            "the advisory's own metrics and adjust only what the usage level proves - a "
+            "package the code never imports is normally N on all three impact metrics."
+        )
+    else:
+        for name, valid in _CVSS_VALID.items():
+            value = breakdown.get(name)
+            if value not in valid:
+                errors.append(
+                    f"Invalid contextual_cvss_breakdown {name}: {value}. Must be one of: {valid}"
+                )
+    if not (reasoning or "").strip():
+        errors.append(
+            "contextual_cvss_reasoning is required: state what you observed in this "
+            "codebase that justifies the contextual rating. A contextual score with "
+            "no reasoning is not shown."
+        )
+    return errors
+
+
+def _validate_advisory_cvss(advisory_cvss: float | None) -> str | None:
+    if advisory_cvss is None:
+        return (
+            "advisory_cvss is required: read the published advisory base score "
+            "(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). It is the "
+            "published reference the finding is rated against — do not omit it "
+            "or the finding cannot be rated."
+        )
+    if not 0.0 <= advisory_cvss <= 10.0:
+        return f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}"
+    return None
+
+
+def _resolve_dependency_rating(
+    advisory_cvss: float | None,
+    contextual_cvss_breakdown: dict[str, str] | None,
+) -> tuple[float | None, str, float | None, str | None]:
+    """Rate the finding.
+
+    A contextual breakdown works exactly like a normal finding's
+    ``cvss_breakdown``: the agent supplies the 8 metrics as observed in this
+    codebase and the score/vector are computed from them. When provided it
+    rates the finding; the advisory score stays as the published reference.
+    """
+    if contextual_cvss_breakdown:
+        score, severity, vector = _calculate_cvss(contextual_cvss_breakdown)
+        return score, severity, score, vector
+    score, severity = _dependency_severity(advisory_cvss)
+    return score, severity, None, None
+
+
 def _build_dependency_metadata(
     *,
     package_name: str,
@@ -760,11 +824,18 @@ def _build_dependency_metadata(
     manifest_path: str | None = None,
     reachability: str | None = None,
     reachability_evidence: str | None = None,
-) -> dict[str, str]:
-    metadata = {
+    advisory_cvss: float | None = None,
+    contextual_cvss_breakdown: dict[str, str] | None = None,
+    contextual_cvss_score: float | None = None,
+    contextual_cvss_vector: str | None = None,
+    contextual_cvss_reasoning: str | None = None,
+) -> dict[str, Any]:
+    metadata: dict[str, Any] = {
         "package_name": package_name.strip(),
         "installed_version": installed_version.strip(),
     }
+    if advisory_cvss is not None:
+        metadata["advisory_cvss"] = advisory_cvss
     if package_ecosystem and package_ecosystem.strip():
         metadata["package_ecosystem"] = package_ecosystem.strip()
     if manifest_path and manifest_path.strip():
@@ -775,12 +846,24 @@ def _build_dependency_metadata(
         metadata["introduced_by"] = introduced_by.strip()
     if dependency_path and dependency_path.strip():
         metadata["dependency_path"] = dependency_path.strip()
-    # "unknown" is the absent case — omitting it keeps the jsonb contract clean,
-    # and evidence without a level would have nothing to qualify.
-    if reachability and reachability.strip() and reachability.strip() != "unknown":
+    if reachability and reachability.strip():
         metadata["reachability"] = reachability.strip()
         if reachability_evidence and reachability_evidence.strip():
             metadata["reachability_evidence"] = reachability_evidence.strip()
+    # Contextual CVSS is only meaningful as the full breakdown, its computed
+    # score/vector, and the reasoning a reader can check — an incomplete set
+    # is dropped.
+    reasoning = str(contextual_cvss_reasoning or "").strip()
+    if (
+        contextual_cvss_breakdown
+        and contextual_cvss_score is not None
+        and contextual_cvss_vector
+        and reasoning
+    ):
+        metadata["contextual_cvss_breakdown"] = contextual_cvss_breakdown
+        metadata["contextual_cvss_score"] = contextual_cvss_score
+        metadata["contextual_cvss_vector"] = contextual_cvss_vector
+        metadata["contextual_cvss_reasoning"] = reasoning[:_MAX_CONTEXTUAL_REASONING_CHARS]
     return metadata
 
 
@@ -852,6 +935,8 @@ async def _do_create_dependency(  # noqa: PLR0912
     manifest_path: str | None = None,
     reachability: str = "unknown",
     reachability_evidence: str | None = None,
+    contextual_cvss_breakdown: dict[str, str] | None = None,
+    contextual_cvss_reasoning: str | None = None,
     agent_id: str | None = None,
     agent_name: str | None = None,
 ) -> dict[str, Any]:
@@ -897,26 +982,29 @@ async def _do_create_dependency(  # noqa: PLR0912
         errors.append(
             f"Invalid reachability: {reachability!r}. Must be one of: {sorted(_VALID_REACHABILITY)}"
         )
-    elif reachability != "unknown" and not (reachability_evidence or "").strip():
+    elif not (reachability_evidence or "").strip():
         errors.append(
-            "reachability_evidence is required when reachability is not 'unknown': "
-            "cite the concrete proof (import file:line, matched symbol usage, or "
-            "govulncheck call path). Never claim a reachability level without evidence."
+            "reachability_evidence is required: cite the concrete proof (import "
+            "file:line, matched symbol usage, or govulncheck call path), or, for "
+            "'unknown', say what you searched and why the result is inconclusive. "
+            "Never claim a reachability level without evidence."
         )
 
-    if advisory_cvss is None:
-        errors.append(
-            "advisory_cvss is required: read the published advisory base score "
-            "(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). Severity is "
-            "derived solely from it — do not omit it or the finding cannot be rated."
-        )
-    elif not 0.0 <= advisory_cvss <= 10.0:
-        errors.append(f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}")
+    errors.extend(_validate_contextual_cvss(contextual_cvss_breakdown, contextual_cvss_reasoning))
+
+    advisory_err = _validate_advisory_cvss(advisory_cvss)
+    if advisory_err:
+        errors.append(advisory_err)
 
     if errors:
         return {"success": False, "error": "Validation failed", "errors": errors}
 
-    cvss_score, severity = _dependency_severity(advisory_cvss)
+    try:
+        cvss_score, severity, contextual_score, contextual_vector = _resolve_dependency_rating(
+            advisory_cvss, contextual_cvss_breakdown
+        )
+    except ValueError as exc:
+        return {"success": False, "error": "Validation failed", "errors": [str(exc)]}
     dependency_metadata = _build_dependency_metadata(
         package_name=package_name,
         installed_version=installed_version,
@@ -927,6 +1015,11 @@ async def _do_create_dependency(  # noqa: PLR0912
         manifest_path=manifest_path,
         reachability=reachability,
         reachability_evidence=reachability_evidence,
+        advisory_cvss=advisory_cvss,
+        contextual_cvss_breakdown=contextual_cvss_breakdown,
+        contextual_cvss_score=contextual_score,
+        contextual_cvss_vector=contextual_vector,
+        contextual_cvss_reasoning=contextual_cvss_reasoning,
     )
     evidence = _build_dependency_evidence(
         cve=parsed_cve,
@@ -1038,6 +1131,8 @@ async def create_dependency_report(
     dependency_path: str | None = None,
     reachability: str = "unknown",
     reachability_evidence: str | None = None,
+    contextual_cvss_breakdown: dict[str, str] | None = None,
+    contextual_cvss_reasoning: str | None = None,
 ) -> str:
     """File a known-CVE dependency (SCA) finding — one report per CVE x package.
 
@@ -1080,8 +1175,10 @@ async def create_dependency_report(
       proved a path from application code to the vulnerable function.
     - ``unknown`` — usage analysis was not performed or was inconclusive.
 
-    Severity is still derived solely from ``advisory_cvss`` — the
-    reachability level never changes the rating, only prioritization.
+    Severity comes from ``contextual_cvss_breakdown`` when you provide one
+    (computed exactly like a normal finding's ``cvss_breakdown``), otherwise
+    from ``advisory_cvss``. The reachability level alone never changes the
+    rating, only prioritization.
 
     **Formatting**: use markdown in text fields (``**bold**``, ``inline
     code`` for package/version identifiers, fenced code blocks for
@@ -1102,8 +1199,9 @@ async def create_dependency_report(
         cwe: ``CWE-NNN`` (most specific) if certain, else omit.
         advisory_cvss: **Required.** Published advisory base score
             (0.0-10.0) — read it off the advisory (trivy CVSS / NVD / GHSA).
-            Severity is derived solely from this score, so it must be the
-            real published value; do not guess or omit it.
+            It is the published reference the finding is rated against and
+            rates the finding whenever you give no contextual breakdown, so
+            it must be the real published value; do not guess or omit it.
         technical_analysis: Optional deeper mechanism/root-cause detail.
         fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
             (dependency upgrades are usually ``trivial``/``low``).
@@ -1127,10 +1225,58 @@ async def create_dependency_report(
             ``not_imported`` / ``imported`` / ``vulnerable_symbol_used`` /
             ``reachable_call_path`` / ``unknown``. Claim only what the
             evidence proves; when in doubt use ``unknown``.
-        reachability_evidence: The concrete proof for the claimed level
-            (required for any level other than ``unknown``): repo-relative
+        reachability_evidence: **Required.** The concrete proof for the
+            claimed level, or, for ``unknown``, what you searched and why
+            the result is inconclusive: repo-relative
             ``file:line`` of the import or symbol usage, the matched
             advisory symbols, or the govulncheck call-path excerpt.
+            Whenever you found the vulnerable symbol in use, also give the
+            **source-to-sink trace** here: start at the vulnerable package
+            call site and walk backwards hop by hop to the entry point
+            that carries untrusted input (HTTP route, CLI argument, queue
+            message, webhook, config file), going one step deeper whenever
+            a hop is a wrapper. Write it as ``entry point -> intermediate
+            call -> package call`` with a ``file:line`` per hop, name what
+            each hop enforces (auth, role check, validation, a flag that
+            is off in production), and say who controls the input. State
+            it plainly when no entry point reaches the sink — that is the
+            most useful result a reader can get.
+        contextual_cvss_breakdown: **Required.** Full CVSS v3.1 rating of this
+            CVE **in this codebase** — the same 8-metric object as
+            ``create_vulnerability_report``'s ``cvss_breakdown``:
+            ``attack_vector`` (N/A/L/P), ``attack_complexity`` (L/H),
+            ``privileges_required`` (N/L/H), ``user_interaction`` (N/R),
+            ``scope`` (U/C), ``confidentiality`` / ``integrity`` /
+            ``availability`` (N/L/H). All 8 metrics are required when the
+            field is set, and the contextual score/vector are computed
+            from them — you never supply a score. Start from the
+            advisory's published metrics and change only what the
+            **source-to-sink trace** you recorded in
+            ``reachability_evidence`` proves is different here: derive
+            ``attack_vector`` / ``privileges_required`` /
+            ``user_interaction`` from what the entry point actually
+            requires, ``attack_complexity`` from the preconditions the
+            hops enforce, and the impact metrics from the data and
+            privileges reachable at the sink. When provided, this rating
+            determines the finding's severity; ``advisory_cvss`` stays as
+            the published reference. Send it on every report: when the
+            trace does not change the published rating, or when you could
+            not complete the trace, repeat the advisory's own metrics and
+            adjust only what the usage level itself proves (a package the
+            code never imports is normally ``N`` on all three impact
+            metrics), then say so in the reasoning.
+        contextual_cvss_reasoning: **Required.** Two to four detailed
+            sentences that a reviewer can verify without opening the repo:
+            how the application uses the package, which call sites or
+            configuration you inspected (repo-relative ``file:line``),
+            which input reaches the vulnerable code and whether an
+            attacker controls it, and what the adjustment therefore
+            changes. State the source-to-sink chain explicitly, hop by
+            hop, as ``entry point -> intermediate call -> package call``
+            with a ``file:line`` for each hop. Cite concrete evidence,
+            never a generic statement such as "low risk". The user reads
+            this text next to the adjusted score, so an adjustment
+            without it is discarded.
     """
     agent_id, agent_name = _caller_identity(ctx)
 
@@ -1155,6 +1301,8 @@ async def create_dependency_report(
         manifest_path=manifest_path,
         reachability=reachability,
         reachability_evidence=reachability_evidence,
+        contextual_cvss_breakdown=contextual_cvss_breakdown,
+        contextual_cvss_reasoning=contextual_cvss_reasoning,
         agent_id=agent_id,
         agent_name=agent_name,
     )

tests/test_cost_tracking.py

@@ -143,7 +143,7 @@ def test_cost_callback_estimates_cost_with_bare_model_fallback() -> None:
     }
 
     def fake_completion_cost(**kwargs: object) -> float:
-        if kwargs["model"] == "gpt-4o-mini":
+        if kwargs["model"] == "openai/gpt-4o-mini":
             return 0.025
         raise ValueError(kwargs["model"])
 

tests/test_pricing.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import litellm
+from agents.usage import Usage
+
+from strix.report.pricing import resolve_litellm_model
+from strix.report.usage import LLMUsageLedger
+
+
+def test_resolves_common_bare_model_names() -> None:
+    resolve_litellm_model.cache_clear()
+    assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
+    assert resolve_litellm_model("openai/deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
+    assert resolve_litellm_model("grok-4.5") == "xai/grok-4.5"
+    assert resolve_litellm_model("MiniMax-M3") == "minimax/MiniMax-M3"
+
+
+def test_resolver_returns_none_for_unresolvable_model() -> None:
+    resolve_litellm_model.cache_clear()
+    assert resolve_litellm_model("provider/not-a-real-model") is None
+
+
+def test_ledger_uses_estimate_when_routed_provider_reports_no_cost() -> None:
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+
+    with patch("litellm.completion_cost", return_value=0.42):
+        ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
+
+    assert ledger.total_cost == 0.42
+
+
+def test_ledger_prefers_observed_cost_over_estimate() -> None:
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+
+    with patch("litellm.completion_cost", return_value=0.42):
+        ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
+    ledger.record_observed_cost(0.17)
+
+    assert ledger.total_cost == 0.17
+
+
+def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+    ledger.hydrate({"cost": 0.42})
+
+    with patch("litellm.completion_cost", return_value=0.17):
+        ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
+
+    assert ledger.total_cost == 0.59
+
+
+def test_zero_cost_disables_both_observed_and_estimated_costs() -> None:
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+    ledger.zero_cost = True
+
+    with patch("litellm.completion_cost", return_value=0.42) as estimate:
+        ledger.record(agent_id="a", usage=usage, model="deepseek-v4-flash")
+        ledger.record_observed_cost(1.0)
+
+    estimate.assert_not_called()
+    assert ledger.total_cost == 0.0
+
+
+def test_resolver_uses_provider_when_bare_entry_has_one() -> None:
+    original = litellm.model_cost
+    litellm.model_cost = {
+        "example": {
+            "litellm_provider": "example-provider",
+            "input_cost_per_token": 1.0,
+            "output_cost_per_token": 2.0,
+        }
+    }
+    try:
+        resolve_litellm_model.cache_clear()
+        assert resolve_litellm_model("example") == "example-provider/example"
+    finally:
+        litellm.model_cost = original
+        resolve_litellm_model.cache_clear()
+
+
+def test_resolver_does_not_guess_between_differently_priced_providers() -> None:
+    original = litellm.model_cost
+    litellm.model_cost = {
+        "provider-a/example": {
+            "input_cost_per_token": 1.0,
+            "output_cost_per_token": 2.0,
+        },
+        "provider-b/example": {
+            "input_cost_per_token": 3.0,
+            "output_cost_per_token": 4.0,
+        },
+    }
+    try:
+        resolve_litellm_model.cache_clear()
+        assert resolve_litellm_model("example") is None
+    finally:
+        litellm.model_cost = original
+        resolve_litellm_model.cache_clear()

tests/test_reporting_fields.py

@@ -37,6 +37,24 @@ _CVSS = {
 }
 
 
+_DEP_CONTEXT = {
+    "attack_vector": "N",
+    "attack_complexity": "L",
+    "privileges_required": "N",
+    "user_interaction": "N",
+    "scope": "U",
+    "confidentiality": "N",
+    "integrity": "N",
+    "availability": "H",
+}
+
+_DEP_CONTEXT_VECTOR = "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H"
+
+_DEP_EVIDENCE = "src/render.ts:14 imports the package."
+
+_DEP_REASONING = "Only scripts/import.py reaches the sink, so the impact is availability only."
+
+
 @pytest.fixture
 def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
     monkeypatch.chdir(tmp_path)
@@ -147,22 +165,33 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
         advisory_cvss=7.2,
         technical_analysis=None,
         fix_effort="trivial",
+        reachability="imported",
+        reachability_evidence=_DEP_EVIDENCE,
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
     assert result["success"] is True
     report = report_state.vulnerability_reports[0]
     assert report["finding_class"] == "dependency_cve"
     assert report["cve"] == "CVE-2021-23337"
     assert report["severity"] == "high"
-    assert report["evidence"] == (
+    assert report["evidence"].startswith(
         "**Advisory evidence:** `CVE-2021-23337` applies to `lodash` "
         "at installed version `4.17.20`. The advisory is fixed in `4.17.21`."
     )
     assert report["dependency_metadata"] == {
         "package_name": "lodash",
         "installed_version": "4.17.20",
+        "advisory_cvss": 7.2,
         "package_ecosystem": "npm",
         "manifest_path": "package-lock.json",
         "fixed_version": "4.17.21",
+        "reachability": "imported",
+        "reachability_evidence": _DEP_EVIDENCE,
+        "contextual_cvss_breakdown": _DEP_CONTEXT,
+        "contextual_cvss_score": pytest.approx(7.5, abs=0.05),
+        "contextual_cvss_vector": _DEP_CONTEXT_VECTOR,
+        "contextual_cvss_reasoning": _DEP_REASONING,
     }
 
 
@@ -186,6 +215,10 @@ async def test_dependency_report_records_transitive_chain(report_state: ReportSt
         fix_effort="trivial",
         introduced_by="express@4.18.1",
         dependency_path="express@4.18.1 > body-parser@1.20.0 > qs@6.10.2",
+        reachability="imported",
+        reachability_evidence=_DEP_EVIDENCE,
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
     assert result["success"] is True
     report = report_state.vulnerability_reports[0]
@@ -224,6 +257,10 @@ async def test_dependency_report_omits_blank_chain_fields(report_state: ReportSt
         fix_effort="trivial",
         introduced_by="  ",
         dependency_path=None,
+        reachability="imported",
+        reachability_evidence=_DEP_EVIDENCE,
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
     assert result["success"] is True
     report = report_state.vulnerability_reports[0]
@@ -231,7 +268,7 @@ async def test_dependency_report_omits_blank_chain_fields(report_state: ReportSt
     assert "dependency_path" not in report["dependency_metadata"]
 
 
-async def test_dependency_report_with_zero_cvss_remains_low_severity(
+async def test_dependency_report_with_no_contextual_impact_is_info(
     report_state: ReportState,
 ) -> None:
     result = await _do_create_dependency(
@@ -251,12 +288,16 @@ async def test_dependency_report_with_zero_cvss_remains_low_severity(
         advisory_cvss=0.0,
         technical_analysis=None,
         fix_effort="low",
+        reachability="not_imported",
+        reachability_evidence="No file imports the package.",
+        contextual_cvss_breakdown={**_DEP_CONTEXT, "availability": "N"},
+        contextual_cvss_reasoning="No application code imports the package.",
     )
 
     assert result["success"] is True
-    assert result["severity"] == "low"
+    assert result["severity"] == "info"
     report = report_state.vulnerability_reports[0]
-    assert report["severity"] == "low"
+    assert report["severity"] == "info"
     assert report["cvss"] == 0.0
 
 
@@ -280,6 +321,8 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
         fix_effort="low",
         reachability="vulnerable_symbol_used",
         reachability_evidence="src/render.ts:14 calls `_.template()`.",
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
 
     assert result["success"] is True
@@ -291,7 +334,8 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
     )
     assert "**Usage analysis:**" in report["evidence"]
     assert "not a proof of exploitability or of safety" in report["evidence"]
-    # The level must never influence the rating — that stays advisory_cvss only.
+    # The level must never influence the rating — that comes from the contextual
+    # breakdown, or from advisory_cvss when no breakdown applies.
     assert report["severity"] == "high"
 
 
@@ -352,7 +396,7 @@ async def test_dependency_report_rejects_unknown_reachability_level(
     assert not report_state.vulnerability_reports
 
 
-async def test_dependency_report_omits_unknown_reachability(report_state: ReportState) -> None:
+async def test_dependency_report_records_unknown_reachability(report_state: ReportState) -> None:
     result = await _do_create_dependency(
         title="CVE-2024-0001 in sample 1.0.0",
         description="Published advisory affects the pinned version.",
@@ -370,12 +414,15 @@ async def test_dependency_report_omits_unknown_reachability(report_state: Report
         advisory_cvss=5.0,
         technical_analysis=None,
         fix_effort="low",
+        reachability_evidence="Grep for the package found no import.",
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
 
-    assert result["success"] is True
+    assert result["success"] is True, result
     metadata = report_state.vulnerability_reports[0]["dependency_metadata"]
-    assert "reachability" not in metadata
-    assert "reachability_evidence" not in metadata
+    assert metadata["reachability"] == "unknown"
+    assert metadata["reachability_evidence"] == "Grep for the package found no import."
 
 
 async def test_dependency_report_requires_advisory_cvss(report_state: ReportState) -> None:
@@ -452,6 +499,10 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
         advisory_cvss=0.0,
         technical_analysis=None,
         fix_effort="low",
+        reachability="imported",
+        reachability_evidence=_DEP_EVIDENCE,
+        contextual_cvss_breakdown=_DEP_CONTEXT,
+        contextual_cvss_reasoning=_DEP_REASONING,
     )
 
     assert result["success"] is True
@@ -463,9 +514,16 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
         "dependency_metadata": {
             "package_name": "sample",
             "installed_version": "1.0.0",
+            "advisory_cvss": 0.0,
             "package_ecosystem": "npm",
             "manifest_path": "package-lock.json",
             "fixed_version": "1.0.1",
+            "reachability": "imported",
+            "reachability_evidence": _DEP_EVIDENCE,
+            "contextual_cvss_breakdown": _DEP_CONTEXT,
+            "contextual_cvss_score": pytest.approx(7.5, abs=0.05),
+            "contextual_cvss_vector": _DEP_CONTEXT_VECTOR,
+            "contextual_cvss_reasoning": _DEP_REASONING,
         },
         "technical_analysis": None,
     }
@@ -877,3 +935,155 @@ def test_vuln_tool_exposes_new_params() -> None:
     dep_required = create_dependency_report.params_json_schema["required"]
     assert "package_ecosystem" in dep_required
     assert "advisory_cvss" in dep_required
+
+
+def test_dep_tool_exposes_contextual_cvss_params() -> None:
+    dep_props = create_dependency_report.params_json_schema["properties"]
+    for field in (
+        "contextual_cvss_breakdown",
+        "contextual_cvss_reasoning",
+    ):
+        assert field in dep_props
+    assert "source-to-sink" in dep_props["contextual_cvss_breakdown"]["description"].lower()
+    assert "source-to-sink" in dep_props["reachability_evidence"]["description"].lower()
+    assert "file:line" in dep_props["contextual_cvss_reasoning"]["description"].lower()
+
+
+_CONTEXTUAL_BREAKDOWN = {
+    "attack_vector": "L",
+    "attack_complexity": "H",
+    "privileges_required": "H",
+    "user_interaction": "N",
+    "scope": "U",
+    "confidentiality": "L",
+    "integrity": "L",
+    "availability": "N",
+}
+
+
+@pytest.mark.asyncio
+async def test_dependency_report_computes_contextual_cvss(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2021-23337 in lodash 4.17.20",
+        description="Command injection via template.",
+        target="repo/package.json",
+        cve="CVE-2021-23337",
+        package_name="lodash",
+        installed_version="4.17.20",
+        impact="Arbitrary command execution.",
+        remediation_steps="Upgrade to 4.17.21.",
+        assumptions="Assumes the template sink is reachable.",
+        package_ecosystem="npm",
+        advisory_cvss=7.2,
+        technical_analysis=None,
+        fixed_version="4.17.21",
+        cwe="CWE-94",
+        fix_effort="trivial",
+        manifest_path="package-lock.json",
+        reachability="vulnerable_symbol_used",
+        reachability_evidence="scripts/import.py:88 calls `_.template()`.",
+        contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
+        contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
+    )
+    assert result["success"] is True, result
+    report = report_state.vulnerability_reports[0]
+    metadata = report["dependency_metadata"]
+    assert metadata["advisory_cvss"] == 7.2
+    assert metadata["contextual_cvss_breakdown"] == _CONTEXTUAL_BREAKDOWN
+    assert metadata["contextual_cvss_vector"] == ("CVSS:3.1/AV:L/AC:H/PR:H/UI:N/S:U/C:L/I:L/A:N")
+    assert metadata["contextual_cvss_score"] == pytest.approx(3.0, abs=0.05)
+    assert metadata["contextual_cvss_reasoning"] == "Only scripts/import.py reaches the sink."
+    # The contextual rating determines the finding's score/severity, exactly
+    # like a normal finding's cvss_breakdown.
+    assert report["cvss"] == metadata["contextual_cvss_score"]
+    assert report["severity"] == "low"
+
+
+@pytest.mark.asyncio
+async def test_dependency_report_requires_contextual_breakdown(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2021-23337 in lodash 4.17.20",
+        description="Command injection via template.",
+        target="repo/package.json",
+        cve="CVE-2021-23337",
+        package_name="lodash",
+        installed_version="4.17.20",
+        impact="Arbitrary command execution.",
+        remediation_steps="Upgrade to 4.17.21.",
+        assumptions="Assumes the template sink is reachable.",
+        package_ecosystem="npm",
+        advisory_cvss=7.2,
+        technical_analysis=None,
+        fixed_version="4.17.21",
+        cwe="CWE-94",
+        fix_effort="trivial",
+        manifest_path="package-lock.json",
+        reachability="imported",
+        reachability_evidence=_DEP_EVIDENCE,
+    )
+    assert result["success"] is False
+    assert any("contextual_cvss_breakdown is required" in error for error in result["errors"])
+    assert report_state.vulnerability_reports == []
+
+
+@pytest.mark.asyncio
+async def test_dependency_report_rejects_incomplete_contextual_breakdown(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2021-23337 in lodash 4.17.20",
+        description="Command injection via template.",
+        target="repo/package.json",
+        cve="CVE-2021-23337",
+        package_name="lodash",
+        installed_version="4.17.20",
+        impact="Arbitrary command execution.",
+        remediation_steps="Upgrade to 4.17.21.",
+        assumptions="Assumes the template sink is reachable.",
+        package_ecosystem="npm",
+        advisory_cvss=7.2,
+        technical_analysis=None,
+        fixed_version="4.17.21",
+        cwe="CWE-94",
+        fix_effort="trivial",
+        manifest_path="package-lock.json",
+        contextual_cvss_breakdown={"attack_vector": "L", "attack_complexity": "Z"},
+        contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
+    )
+    assert result["success"] is False
+    assert any("attack_complexity" in error for error in result["errors"])
+    assert any("privileges_required" in error for error in result["errors"])
+    assert report_state.vulnerability_reports == []
+
+
+@pytest.mark.asyncio
+async def test_dependency_report_rejects_contextual_breakdown_without_reasoning(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2021-23337 in lodash 4.17.20",
+        description="Command injection via template.",
+        target="repo/package.json",
+        cve="CVE-2021-23337",
+        package_name="lodash",
+        installed_version="4.17.20",
+        impact="Arbitrary command execution.",
+        remediation_steps="Upgrade to 4.17.21.",
+        assumptions="Assumes the template sink is reachable.",
+        package_ecosystem="npm",
+        advisory_cvss=7.2,
+        technical_analysis=None,
+        fixed_version="4.17.21",
+        cwe="CWE-94",
+        fix_effort="trivial",
+        manifest_path="package-lock.json",
+        contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
+        contextual_cvss_reasoning="   ",
+    )
+    assert result["success"] is False
+    assert any("contextual_cvss_reasoning is required" in error for error in result["errors"])
+    assert report_state.vulnerability_reports == []

tests/test_telemetry_resume.py

@@ -0,0 +1,89 @@
+"""Regression tests for telemetry emitted by resumed runs."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import pytest
+from agents.usage import Usage
+
+from strix.report.state import ReportState
+from strix.telemetry import posthog, scarf
+
+
+def _usage(requests: int, input_tokens: int, output_tokens: int, total_tokens: int) -> Usage:
+    return Usage(
+        requests=requests,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        total_tokens=total_tokens,
+    )
+
+
+def _capture(sent: list[dict[str, Any]], props: dict[str, Any]) -> bool:
+    sent.append(props)
+    return True
+
+
+@pytest.mark.parametrize("telemetry", [posthog, scarf])
+def test_scan_ended_reports_resumed_usage_delta(
+    telemetry: Any,
+    tmp_path: Any,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    initial = ReportState(run_name="resumed")
+    initial.record_sdk_usage(
+        agent_id="agent",
+        usage=_usage(10, 1000, 200, 1200),
+        model="unknown",
+    )
+    initial.record_observed_llm_cost(1.25)
+    initial.end_time = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+    initial.run_record["end_time"] = initial.end_time
+    initial.save_run_data()
+
+    resumed = ReportState(run_name="resumed")
+    resumed.hydrate_from_run_dir()
+    resumed.record_sdk_usage(
+        agent_id="agent",
+        usage=_usage(3, 300, 50, 350),
+        model="unknown",
+    )
+    resumed.record_observed_llm_cost(0.75)
+
+    sent: list[dict[str, Any]] = []
+    monkeypatch.setattr(telemetry, "_send", lambda _event, props: _capture(sent, props))
+    telemetry.end(resumed)
+
+    assert sent[0]["llm_requests"] == 3
+    assert sent[0]["llm_input_tokens"] == 300
+    assert sent[0]["llm_output_tokens"] == 50
+    assert sent[0]["llm_tokens"] == 350
+    assert sent[0]["llm_cost"] == pytest.approx(0.75)
+    assert 0 <= sent[0]["duration_seconds"] <= 2
+
+
+@pytest.mark.parametrize("telemetry", [posthog, scarf])
+def test_scan_ended_reports_all_fresh_run_usage(
+    telemetry: Any,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    state = ReportState()
+    state.record_sdk_usage(
+        agent_id="agent",
+        usage=_usage(3, 300, 50, 350),
+        model="unknown",
+    )
+    state.record_observed_llm_cost(0.75)
+
+    sent: list[dict[str, Any]] = []
+    monkeypatch.setattr(telemetry, "_send", lambda _event, props: _capture(sent, props))
+    telemetry.end(state)
+
+    assert sent[0]["llm_requests"] == 3
+    assert sent[0]["llm_input_tokens"] == 300
+    assert sent[0]["llm_output_tokens"] == 50
+    assert sent[0]["llm_tokens"] == 350
+    assert sent[0]["llm_cost"] == pytest.approx(0.75)