feat(report): add LLM analysis provenance

src/skillspector/inference_usage.py

@@ -10,8 +10,11 @@
 
 from __future__ import annotations
 
+import base64
+import json
 import re
 import threading
+import weakref
 from collections.abc import Mapping, Sequence
 from typing import NotRequired, TypedDict
 
@@ -28,6 +31,80 @@
     "total_tokens",
 )
 _MAX_TOKEN_COUNT = (1 << 63) - 1
+_MIN_SAMPLING_SEED = -(1 << 63)
+_MAX_SAMPLING_SEED = (1 << 63) - 1
+_FORWARDED_CONTROL_NAMES = ("temperature", "seed", "reasoning_effort")
+# Public effort telemetry is an enum, even when a provider accepts arbitrary text.
+# Unrecognized provider-specific values must not become a channel for credentials.
+_REASONING_EFFORTS = frozenset({"none", "minimal", "low", "medium", "high", "xhigh", "max", "auto"})
+_CREDENTIAL_PREFIXES = (
+    "sk-",
+    "nvapi-",
+    "ghp_",
+    "gho_",
+    "ghu_",
+    "ghs_",
+    "ghr_",
+    "github_pat_",
+    "glpat-",
+    "bearer-",
+    "xoxb-",
+    "xoxp-",
+    "xoxa-",
+    "xoxr-",
+    "hf_",
+    "aiza",
+    "akia",
+    "asia",
+    "aws-secret-",
+)
+_UNPREFIXED_CREDENTIAL = re.compile(r"(?:[0-9a-fA-F]{32,64}|[A-Za-z0-9_+/=]{40,88})\Z")
+_AUTHORIZATION_CREDENTIAL = re.compile(r"(?:authorization\s*:\s*)?(?:bearer|basic)(?:\s|:)", re.I)
+_COMPACT_CREDENTIAL = re.compile(r"[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]*){2,4}\Z")
+
+_CHAT_MODEL_CONTROLS: dict[
+    int,
+    tuple[
+        weakref.ReferenceType[object],
+        dict[str, float | int | str | None],
+        dict[str, float | int | str | None],
+    ],
+] = {}
+_CHAT_MODEL_CONTROLS_LOCK = threading.Lock()
+
+
+def _is_compact_credential(value: str) -> bool:
+    """Recognize JSON JWT/JWE headers without rejecting dotted model versions."""
+    if not _COMPACT_CREDENTIAL.fullmatch(value):
+        return False
+    header = value.partition(".")[0]
+    try:
+        decoded = json.loads(base64.urlsafe_b64decode(header + "=" * (-len(header) % 4)))
+    except (ValueError, RecursionError):
+        return False
+    return isinstance(decoded, dict)
+
+
+def looks_like_credential(value: object) -> bool:
+    """Return whether a printable label resembles a common secret value."""
+    if not isinstance(value, str):
+        return False
+    candidate = value.strip()
+    lowered = candidate.lower()
+    return (
+        lowered.startswith(_CREDENTIAL_PREFIXES)
+        or bool(_AUTHORIZATION_CREDENTIAL.match(candidate))
+        or _is_compact_credential(candidate)
+        # A full hyphenated model name is not an opaque base64 credential.
+        # Still reject long opaque components embedded in a namespaced label.
+        or any(_UNPREFIXED_CREDENTIAL.fullmatch(part) for part in candidate.split("-"))
+    )
+
+
+def safe_reasoning_effort(value: object) -> str | None:
+    """Return a recognized effort value safe for public configuration telemetry."""
+    candidate = value.strip() if isinstance(value, str) else ""
+    return candidate if candidate in _REASONING_EFFORTS else None
 
 
 class InferenceUsageRecord(TypedDict):
@@ -45,6 +122,127 @@ class InferenceUsageRecord(TypedDict):
     cache_write_tokens: NotRequired[int]
     reasoning_tokens: NotRequired[int]
     total_tokens: NotRequired[int]
+    # Internal-only construction evidence. ``sanitize_inference_usage`` never
+    # includes this field in the public token-usage projection; the provenance
+    # sanitizer consumes it separately after a provider response is observed.
+    requested_controls: NotRequired[dict[str, float | int | str | None]]
+    forwarded_controls: NotRequired[dict[str, float | int | str | None]]
+
+
+def _forwarded_controls(value: Mapping[str, object] | None) -> dict[str, float | int | str | None]:
+    """Return the fixed, non-secret sampling-control construction record."""
+    source = value or {}
+    controls: dict[str, float | int | str | None] = {}
+    for name in _FORWARDED_CONTROL_NAMES:
+        if name not in source:
+            continue
+        raw = source.get(name)
+        if raw is None:
+            controls[name] = None
+        elif name == "temperature":
+            if isinstance(raw, (int, float)) and not isinstance(raw, bool) and 0 <= float(raw) <= 1:
+                controls[name] = float(raw)
+        elif name == "seed":
+            if (
+                isinstance(raw, int)
+                and not isinstance(raw, bool)
+                and _MIN_SAMPLING_SEED <= raw <= _MAX_SAMPLING_SEED
+            ):
+                controls[name] = raw
+        elif (setting := safe_reasoning_effort(raw)) is not None:
+            controls[name] = setting
+    return controls
+
+
+def register_chat_model_controls(
+    chat_model: object,
+    forwarded_controls: Mapping[str, object],
+    *,
+    requested_controls: Mapping[str, object] | None = None,
+) -> None:
+    """Associate a model with requested and normalized request controls."""
+    model_id = id(chat_model)
+    sanitized_requested = _forwarded_controls(requested_controls)
+    sanitized_forwarded = _forwarded_controls(forwarded_controls)
+
+    def _discard(model_ref: weakref.ReferenceType[object]) -> None:
+        with _CHAT_MODEL_CONTROLS_LOCK:
+            current = _CHAT_MODEL_CONTROLS.get(model_id)
+            if current is not None and current[0] is model_ref:
+                _CHAT_MODEL_CONTROLS.pop(model_id, None)
+
+    try:
+        model_ref = weakref.ref(chat_model, _discard)
+    except TypeError:
+        return
+    with _CHAT_MODEL_CONTROLS_LOCK:
+        _CHAT_MODEL_CONTROLS[model_id] = (
+            model_ref,
+            sanitized_requested,
+            sanitized_forwarded,
+        )
+
+
+def chat_model_controls(chat_model: object | None) -> dict[str, float | int | str | None]:
+    """Return detached construction evidence for *chat_model*, when recorded."""
+    if chat_model is None:
+        return {}
+    with _CHAT_MODEL_CONTROLS_LOCK:
+        current = _CHAT_MODEL_CONTROLS.get(id(chat_model))
+        if current is None or current[0]() is not chat_model:
+            return {}
+        return current[2].copy()
+
+
+def chat_model_requested_controls(
+    chat_model: object | None,
+) -> dict[str, float | int | str | None]:
+    """Return the controls resolved when *chat_model* was constructed."""
+    if chat_model is None:
+        return {}
+    with _CHAT_MODEL_CONTROLS_LOCK:
+        current = _CHAT_MODEL_CONTROLS.get(id(chat_model))
+        if current is None or current[0]() is not chat_model:
+            return {}
+        return current[1].copy()
+
+
+def retained_chat_model_controls(
+    chat_model: object,
+    names: Sequence[str],
+) -> dict[str, float | int | str | None]:
+    """Return controls retained by the normalized provider request payload.
+
+    LangChain may accept a constructor option and then remove it for a
+    provider/model combination. Provenance must describe the request that the
+    adapter will send, not the raw constructor arguments supplied before that
+    normalization.
+    """
+    selected = [name for name in names if name in _FORWARDED_CONTROL_NAMES]
+    controls: dict[str, object] = dict.fromkeys(selected)
+    payload: Mapping[str, object] | None = None
+    payload_builder = getattr(chat_model, "_get_request_payload", None)
+    if callable(payload_builder):
+        try:
+            candidate = payload_builder("SkillSpector provenance probe")
+        except (TypeError, ValueError):
+            candidate = None
+        if isinstance(candidate, Mapping):
+            payload = candidate
+
+    if payload is not None:
+        output_config = payload.get("output_config")
+        output_config = output_config if isinstance(output_config, Mapping) else {}
+        for name in selected:
+            if name == "reasoning_effort":
+                controls[name] = payload.get(name, output_config.get("effort"))
+            else:
+                controls[name] = payload.get(name)
+    else:
+        for name in selected:
+            attribute = "effort" if name == "reasoning_effort" else name
+            controls[name] = getattr(chat_model, name, getattr(chat_model, attribute, None))
+    return _forwarded_controls(controls)
 
 
 def _mapping(value: object) -> Mapping[str, object]:
@@ -98,7 +296,12 @@ def _strict_label(value: object) -> str | None:
 def _strict_model_label(value: object) -> str | None:
     """Return a model label only when it cannot encode a URL or userinfo."""
     candidate = _strict_label(value)
-    if candidate is None or "://" in candidate or "@" in candidate:
+    if (
+        candidate is None
+        or "://" in candidate
+        or "@" in candidate
+        or looks_like_credential(candidate)
+    ):
         return None
     return candidate
 
@@ -110,14 +313,18 @@ def _model_label(value: object, fallback: str = "unknown") -> str:
 def provider_name(provider: object) -> str:
     """Return a stable provider label without endpoint or credential data."""
     names = {
+        "AntigravityCLIProvider": "antigravity_cli",
         "AnthropicProvider": "anthropic",
         "AnthropicProxyProvider": "anthropic_proxy",
+        "AzureOpenAIProvider": "azure_openai",
         "BedrockProvider": "bedrock",
         "ClaudeCLIProvider": "claude_cli",
         "CodexCLIProvider": "codex_cli",
         "GeminiCLIProvider": "gemini_cli",
         "NvBuildProvider": "nv_build",
         "NvInferenceProvider": "nv_inference",
+        "OllamaProvider": "ollama",
+        "OpenAICompatibleProvider": "openai_compatible",
         "OpenAIProvider": "openai",
         "OpencodeCLIProvider": "opencode_cli",
     }
@@ -297,11 +504,15 @@ def __init__(
         request_kind: str,
         provider: str,
         requested_model: str,
+        requested_controls: Mapping[str, object] | None = None,
+        forwarded_controls: Mapping[str, object] | None = None,
     ) -> None:
         self._node = node
         self._request_kind = request_kind
         self._provider = provider
         self._requested_model = requested_model
+        self._requested_controls = _forwarded_controls(requested_controls)
+        self._forwarded_controls = _forwarded_controls(forwarded_controls)
         self._records: list[InferenceUsageRecord] = []
         self._response_received = False
         self._lock = threading.Lock()
@@ -328,12 +539,32 @@ def on_llm_end(self, response: LLMResult, **kwargs: object) -> None:
         with self._lock:
             self._response_received = True
             if record is not None:
+                if self._requested_controls:
+                    record["requested_controls"] = self._requested_controls.copy()
+                if self._forwarded_controls:
+                    record["forwarded_controls"] = self._forwarded_controls.copy()
                 self._records.append(record)
+            else:
+                self._records.append(self._response_observation())
+
+    def _response_observation(self) -> InferenceUsageRecord:
+        """Build counter-less, internal-only evidence of a completed response."""
+        return {
+            "node": _label(self._node),
+            "request_kind": _label(self._request_kind),
+            "provider": _label(self._provider),
+            "model": _model_label(self._requested_model),
+            "model_source": "requested_model",
+            "usage_source": "provider_response",
+            "requested_controls": self._requested_controls.copy(),
+            "forwarded_controls": self._forwarded_controls.copy(),
+        }
 
     def mark_response_received(self) -> None:
         """Record a completed response from a non-LangChain transport."""
         with self._lock:
             self._response_received = True
+            self._records.append(self._response_observation())
 
     def set_provider(self, provider: str) -> None:
         """Set the effective provider before the first response is observed."""
@@ -343,16 +574,36 @@ def set_provider(self, provider: str) -> None:
                 raise RuntimeError("cannot change inference provider after a response")
             self._provider = label
 
+    def set_controls(
+        self,
+        requested: Mapping[str, object] | None,
+        forwarded: Mapping[str, object] | None,
+    ) -> None:
+        """Update constructor/request evidence before the next response."""
+        with self._lock:
+            self._requested_controls = _forwarded_controls(requested)
+            self._forwarded_controls = _forwarded_controls(forwarded)
+
     @property
     def response_received(self) -> bool:
         """Whether the provider returned, even when it reported no token usage."""
         with self._lock:
             return self._response_received
 
     def snapshot(self) -> list[InferenceUsageRecord]:
-        """Return detached copies safe for graph-state serialization."""
+        """Return usage plus counter-less response evidence for provenance."""
         with self._lock:
-            return [record.copy() for record in self._records]
+            snapshot: list[InferenceUsageRecord] = []
+            for record in self._records:
+                detached = record.copy()
+                controls = record.get("forwarded_controls")
+                if isinstance(controls, dict):
+                    detached["forwarded_controls"] = controls.copy()
+                requested = record.get("requested_controls")
+                if isinstance(requested, dict):
+                    detached["requested_controls"] = requested.copy()
+                snapshot.append(detached)
+            return snapshot
 
 
 def sanitize_inference_usage(

src/skillspector/llm_analyzer_base.py

@@ -41,7 +41,11 @@
 from langchain_openai import ChatOpenAI
 from pydantic import BaseModel, Field, ValidationError, field_validator
 
-from skillspector.inference_usage import InferenceUsageRecord
+from skillspector.inference_usage import (
+    InferenceUsageRecord,
+    chat_model_controls,
+    chat_model_requested_controls,
+)
 from skillspector.inspection_ledger import (
     AnalyzerStatusEvent,
     InspectionLedgerEvent,
@@ -57,6 +61,7 @@
     _AgentCLIMessage,
     _ainvoke_with_usage,
     _invoke_with_usage,
+    chat_model_provider_name,
     get_chat_model,
     new_inference_usage_collector,
 )
@@ -724,6 +729,13 @@ def _model_for_call(self) -> tuple[object, object | None]:
             return self._llm, self._structured_llm
         llm = get_chat_model(model=self.model, timeout=remaining)
         _uses_native_connection_retries(llm, max_retries=0)
+        effective_provider = chat_model_provider_name(llm)
+        if effective_provider is not None:
+            self._usage_collector.set_provider(effective_provider)
+        self._usage_collector.set_controls(
+            chat_model_requested_controls(llm),
+            chat_model_controls(llm),
+        )
         structured = (
             llm.with_structured_output(self.response_schema) if self.response_schema else None
         )