Persist observed model provenance on learning lineage

reflexio/models/api_schema/domain/entities.py

@@ -507,6 +507,11 @@ class LineageEvent(BaseModel):
         request_id (str): Triggering request — part of the idempotency key.
         reason (str): Free-text rationale (no PII).
         created_at (int): Unix epoch seconds (0 = unset; storage stamps it).
+        from_status (str | None): Status before a transition.
+        to_status (str | None): Status after a transition.
+        status_namespace (str | None): Namespace for status values.
+        model_name (str | None): Observed model for a content-shaping operation.
+        provider (str | None): Observed provider for that operation.
     """
 
     event_id: int = 0
@@ -523,6 +528,8 @@ class LineageEvent(BaseModel):
     from_status: str | None = None
     to_status: str | None = None
     status_namespace: str | None = None
+    model_name: str | None = None
+    provider: str | None = None
 
 
 class LineageContext(BaseModel):
@@ -536,6 +543,8 @@ class LineageContext(BaseModel):
     source_ids: list[str] = []
     reason: str = ""
     request_id: str | None = None
+    model_name: str | None = None
+    provider: str | None = None
 
 
 class RecordRef(BaseModel):

reflexio/server/llm/_litellm_types.py

@@ -19,6 +19,22 @@
 from reflexio.models.config_schema import APIKeyConfig
 
 
+@dataclass(frozen=True)
+class ModelProvenance:
+    """Observed model and provider attribution for one completion."""
+
+    model_name: str | None = None
+    provider: str | None = None
+
+
+@dataclass(frozen=True)
+class CompletionResult[T]:
+    """Completion value paired with its non-serializing provenance."""
+
+    value: T
+    provenance: ModelProvenance
+
+
 @dataclass
 class LiteLLMConfig:
     """
@@ -104,17 +120,31 @@ class ToolCallingChatResponse:
 
 
 class LiteLLMClientError(Exception):
-    """Custom exception for LiteLLM client errors."""
+    """Custom exception for LiteLLM client errors.
+
+    ``first_parsed_provenance`` is populated when a later structured-output
+    repair transport failure leaves a parsed response available to a caller.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        first_parsed_provenance: ModelProvenance | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.first_parsed_provenance = first_parsed_provenance
 
 
 class StructuredOutputRepairError(LiteLLMClientError):
     """Raised when an opted-in structured-output repair ladder is exhausted.
 
     Field pairing caveat: ``raw_content``/``validation_errors`` describe the
     LAST attempt, while ``parsed_output`` falls back to the most recent attempt
-    that parsed at all — when the final attempt failed to parse, these fields
-    describe different attempts. Callers must not assume ``validation_errors``
-    were produced by validating ``parsed_output``.
+    that parsed at all. ``first_parsed_provenance`` is the first parse across the
+    whole multi-rung walk (not merely the final rung), so salvage callers can
+    pair it with the first accepted parsed content from a shared validator
+    closure.
     """
 
     def __init__(
@@ -126,8 +156,9 @@ def __init__(
         raw_content: str | None = None,
         parsed_output: BaseModel | None = None,
         validation_errors: tuple[str, ...] = (),
+        first_parsed_provenance: ModelProvenance | None = None,
     ) -> None:
-        super().__init__(message)
+        super().__init__(message, first_parsed_provenance=first_parsed_provenance)
         self.failure_kind = failure_kind
         self.model = model
         self.raw_content = raw_content
@@ -148,10 +179,12 @@ def __init__(
         *,
         raw_content: str | None = None,
         finish_reason: str | None = None,
+        provenance: ModelProvenance | None = None,
     ) -> None:
         super().__init__(message)
         self.raw_content = raw_content
         self.finish_reason = finish_reason
+        self.provenance = provenance
 
 
 class LLMHardTimeoutError(TimeoutError):

reflexio/server/llm/_provider_concurrency.py

@@ -16,12 +16,16 @@
 import threading
 from collections.abc import Iterator
 from contextlib import contextmanager
+from typing import TYPE_CHECKING
 
 import litellm
 
 from reflexio.server.env_utils import env_str
 from reflexio.server.llm.llm_utils import positive_int_env
 
+if TYPE_CHECKING:
+    from reflexio.server.llm._litellm_types import ModelProvenance
+
 logger = logging.getLogger(__name__)
 
 _DEFAULT_MAX_CONCURRENCY = 8
@@ -44,6 +48,15 @@ class ProviderCapSaturatedError(Exception):
     advance-worthy rung failure.
     """
 
+    def __init__(
+        self,
+        message: str,
+        *,
+        first_parsed_provenance: "ModelProvenance | None" = None,
+    ) -> None:
+        super().__init__(message)
+        self.first_parsed_provenance = first_parsed_provenance
+
 
 def _parse_fail_closed() -> frozenset[str]:
     raw = env_str("REFLEXIO_LLM_FAIL_CLOSED_PROVIDERS", "")

reflexio/server/llm/providers/claude_code_provider.py

@@ -28,6 +28,7 @@
 import tempfile
 import time
 from contextlib import suppress
+from dataclasses import replace
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -530,8 +531,11 @@ def _run_claude_stream(
     except FileNotFoundError as exc:
         raise ClaudeCodeCLIError(f"claude CLI not found at {cli_path}") from exc
 
-    return parse_stream_json(
-        proc.stdout, exit_code=proc.returncode, stderr_text=proc.stderr
+    return replace(
+        parse_stream_json(
+            proc.stdout, exit_code=proc.returncode, stderr_text=proc.stderr
+        ),
+        cli_binary=_cli_name(),
     )
 
 
@@ -592,6 +596,7 @@ def _run_codex_stream(
         terminal_text=terminal_text,
         stderr_text=proc.stderr,
         raw_lines_parsed=1 if terminal_text else 0,
+        cli_binary="codex",
     )
 
 
@@ -612,6 +617,10 @@ def _build_model_response(
     model: str,
     terminal_text: str,
     elapsed_seconds: float,
+    *,
+    served_model: str | None = None,
+    served_provider: str | None = None,
+    cli_binary: str | None = None,
 ) -> ModelResponse:
     """Wrap the CLI's terminal text in a LiteLLM ``ModelResponse``.
 
@@ -621,9 +630,12 @@ def _build_model_response(
 
     Args:
         model (str): The model string originally requested
-            (e.g. ``claude-code/default``).
+            (e.g. ``claude-code/default``). Populates the public LiteLLM
+            ``ModelResponse.model`` field the same way other providers do.
         terminal_text (str): The terminal ``result`` text from the CLI.
         elapsed_seconds (float): Wall time the subprocess took — for logging only.
+        served_model: Observed served model from stream-json, if any. Stamped on
+            hidden metadata for provenance; not used to overwrite ``model``.
 
     Returns:
         ModelResponse: Shaped to match what callers of ``litellm.completion`` expect.
@@ -639,6 +651,12 @@ def _build_model_response(
         object="chat.completion",
         usage=usage,
     )
+    _set_cli_response_metadata(
+        response,
+        served_model=served_model,
+        served_provider=served_provider,
+        cli_binary=cli_binary,
+    )
     _LOGGER.debug(
         "claude-code provider: model=%s elapsed=%.2fs",
         model,
@@ -647,6 +665,25 @@ def _build_model_response(
     return response
 
 
+def _set_cli_response_metadata(
+    response: ModelResponse,
+    *,
+    served_model: str | None,
+    served_provider: str | None,
+    cli_binary: str | None,
+) -> None:
+    """Stamp truthful route metadata on a CLI-backed completion response."""
+    hidden = dict(getattr(response, "_hidden_params", {}) or {})
+    hidden["reflexio_provider"] = PROVIDER_KEY
+    if cli_binary:
+        hidden["reflexio_cli_binary"] = cli_binary
+    if served_model:
+        hidden["reflexio_served_model"] = served_model
+    if served_provider:
+        hidden["reflexio_served_provider"] = served_provider
+    response._hidden_params = hidden
+
+
 _TOOL_USE_INSTRUCTION_TEMPLATE = (
     "## EXTERNAL TOOL-CALLING MODE\n"
     "\n"
@@ -785,6 +822,9 @@ def _build_model_response_with_tool_call(
     terminal_text: str,
     elapsed_seconds: float,
     tool_use: dict[str, Any],
+    served_model: str | None = None,
+    served_provider: str | None = None,
+    cli_binary: str | None = None,
 ) -> ModelResponse:
     """Wrap the CLI terminal text as a ``ModelResponse`` carrying one ``tool_calls`` entry.
 
@@ -794,7 +834,6 @@ def _build_model_response_with_tool_call(
     this — usage is informational, not load-bearing.
 
     Args:
-        model: Model string passed in by LiteLLM.
         terminal_text: The terminal ``result`` text from the CLI (retained
             for signature parity with the plain-text branch; surfaced via
             logging only).
@@ -826,6 +865,12 @@ def _build_model_response_with_tool_call(
         object="chat.completion",
         usage=usage,
     )
+    _set_cli_response_metadata(
+        response,
+        served_model=served_model,
+        served_provider=served_provider,
+        cli_binary=cli_binary,
+    )
     _LOGGER.debug(
         "claude-code provider: tool_call name=%s elapsed=%.2fs",
         tool_use["name"],
@@ -1018,6 +1063,9 @@ def completion(  # type: ignore[override]
                         terminal_text=result.terminal_text,
                         elapsed_seconds=elapsed,
                         tool_use=tool_use,
+                        served_model=result.served_model,
+                        served_provider=result.served_provider,
+                        cli_binary=result.cli_binary,
                     )
                 # Log a metadata-only warning (no raw payload) — the model
                 # output can carry user content / source code; deferring the
@@ -1038,6 +1086,9 @@ def completion(  # type: ignore[override]
                 model=model,
                 terminal_text=result.terminal_text,
                 elapsed_seconds=elapsed,
+                served_model=result.served_model,
+                served_provider=result.served_provider,
+                cli_binary=result.cli_binary,
             )
 
         self._record_stall_safely(result)

reflexio/server/llm/providers/claude_code_stream_parser.py

@@ -63,6 +63,9 @@ class ParseResult:
     stderr_text: str = ""
     raw_lines_parsed: int = 0
     raw_lines_failed: int = 0
+    served_model: str | None = None
+    served_provider: str | None = None
+    cli_binary: str | None = None
 
     @property
     def stall_candidate(self) -> str | None:
@@ -95,6 +98,12 @@ def parse_stream_json(
     parsed = 0
     failed = 0
     saw_terminal = False
+    init_model: str | None = None
+    assistant_model: str | None = None
+    assistant_provider: str | None = None
+    usage_model: str | None = None
+    result_model: str | None = None
+    result_provider: str | None = None
     for line in stdout.splitlines():
         if not line.strip():
             continue
@@ -107,6 +116,10 @@ def parse_stream_json(
         if not isinstance(event, dict):
             continue
         match event.get("type"), event.get("subtype"):
+            case ("system", "init"):
+                model = event.get("model")
+                if isinstance(model, str) and model.strip():
+                    init_model = model
             case ("system", "api_retry"):
                 err = event.get("error")
                 if isinstance(err, str):
@@ -116,13 +129,36 @@ def parse_stream_json(
                 if isinstance(text, str):
                     terminal_text = text
                     saw_terminal = True
+                model_usage = event.get("modelUsage")
+                if isinstance(model_usage, dict) and len(model_usage) == 1:
+                    model = next(iter(model_usage))
+                    if isinstance(model, str) and model.strip():
+                        usage_model = model
+                model = event.get("model")
+                if isinstance(model, str) and model.strip():
+                    result_model = model
+                provider = event.get("provider")
+                if isinstance(provider, str) and provider.strip():
+                    result_provider = provider
+            case ("assistant", _):
+                message = event.get("message")
+                model = message.get("model") if isinstance(message, dict) else None
+                if isinstance(model, str) and model.strip():
+                    assistant_model = model
+                provider = (
+                    message.get("provider") if isinstance(message, dict) else None
+                )
+                if isinstance(provider, str) and provider.strip():
+                    assistant_provider = provider
     return ParseResult(
         success=(exit_code == 0 and saw_terminal and bool(terminal_text)),
         terminal_text=terminal_text,
         retry_errors=retry_errors,
         stderr_text=stderr_text,
         raw_lines_parsed=parsed,
         raw_lines_failed=failed,
+        served_model=result_model or assistant_model or init_model or usage_model,
+        served_provider=result_provider or assistant_provider,
     )