portfolio-optimiser/src/portfolio_optimiser/provenance.py

"""First-class Pydantic provenance stamp.

Provenance is authoritative framework data — **independent** of MAF's ``Annotation`` type,
which silently drops on the Python streaming path (#4316, research 02 Dim 2). A
``ProvenanceStamp`` is a Pydantic model that must carry at least one ``Citation`` (a
``min_length=1`` constraint), the model + role that produced the proposal, the validator's
decision, and the token usage. ``to_annotations()`` maps to MAF ``Annotation`` dicts for
DISPLAY only — never the source of truth.

The locator type (``TextSpan``) is owned by ``retrieval.py`` (Step 5) and imported here, so
a citation's span is the same exact object the retriever produced.
"""

from __future__ import annotations

from typing import Literal

from agent_framework import Annotation, TextSpanRegion
from pydantic import BaseModel, Field

from portfolio_optimiser.okf import ResolvedBundleId
from portfolio_optimiser.retrieval import TextSpan


class Citation(BaseModel):
    """One exact-span citation into a source document (locator owned by retrieval.py)."""

    file: str
    locator: TextSpan
    snippet: str


class ExternalCall(BaseModel):
    """One external service call a run actually made (Trekk B4).

    **What this is evidence of, and what it is not.** It records that ``tool`` was invoked and which
    configured ``server`` it belongs to. It is NOT evidence that the service's answer reached the
    proposal, and it is not a verified rendering of what the service returned — the framework hands
    the answer to the agent, and what the agent does with it is the agent's. Reading this as "the
    figure came from the price register" would claim more than the record supports.

    ``server`` is ``""`` when the tool name cannot be attributed to exactly one configured server.
    MEASURED against a real MCP stdio subprocess: MAF passes the BARE tool name to function
    middleware, with no server prefix, so two servers exposing one tool name are indistinguishable
    at this seam. Unattributed is the honest answer there; naming the first match would put a
    service in the record that may never have been contacted.
    """

    server: str
    tool: str


class ProvenanceStamp(BaseModel):
    """Authoritative provenance for one proposal — at least one citation is mandatory."""

    citations: list[Citation] = Field(min_length=1)
    model: str
    role: str
    validator_decision: Literal["validated", "rejected"]
    token_usage: int
    #: Was the deterministic gate ANCHORED to the project's own cost lines? ``True`` means
    #: ``validate_proposal`` ran its stage-0 reconciliation (every ``affected_item`` checked against
    #: a real ``CostBaseline`` line, within tolerance, BEFORE the solver); ``False`` means the
    #: bundle shipped no ``cost-baseline.json``, so that stage was SKIPPED and the gate reasoned
    #: only about numbers the proposal itself supplied. Anchoring stays OPTIONAL (a pre-amendment
    #: bundle is legitimately un-anchored) — this field does not gate anything, it makes the skip
    #: legible. A STRUCTURED field rather than prose, for the reason ``BudgetExceeded`` carries
    #: ``kind``/``limit``/``observed`` as fields (kø-(y)): "was the falsifier anchored" is an
    #: operative question that must be readable by machine.
    #:
    #: REQUIRED, with no default, because both defaults lie: ``True`` would let a constructor that
    #: forgot claim an anchoring that never happened, and ``False`` would under-claim a real one.
    #: A binary fact about a falsifier has no honest default.
    cost_baseline_anchored: bool
    #: WHICH knowledge base this proposal was reasoned from, and how its identity was established —
    #: ``okf.ResolvedBundleId`` (declared id / origin / the mount it was read from), or ``None`` on
    #: the road path, where no knowledge base was involved at all.
    #:
    #: REQUIRED WITHOUT DEFAULT, for ``cost_baseline_anchored``'s reason: ``None`` is a VALUE here
    #: (a run with no base), so letting an omitted field mean it would make "no base" and "nobody
    #: filled this in" the same answer. A stamp that cannot say which corpus it judged cannot be
    #: joined back to one, and the artefact leaves the process.
    #:
    #: Not on ``RunResult`` beside ``skipped_links``, and the distinction is deliberate:
    #: ``skipped_links`` is diagnostic about the NAVIGATION, while this is the identity of the
    #: artefact that was judged — the same class as ``cost_baseline_anchored``, which also resolves
    #: once per run and is stamped per proposal.
    bundle_id_source: ResolvedBundleId | None
    #: P19/B2 — each ``affected_item`` code as ``"identifier"`` or ``"prose"``, by whether it is
    #: SHAPED like an identifier of the delivered corpora at all (``validator.classify_codes``).
    #:
    #: A REPORT, never the gate. The gate is P19/B3 inside ``validate_proposal``, and it fires only
    #: where the input demonstrably offers identifier forms; this says what the run classified
    #: whatever the gate then did, which is what makes an un-anchored or form-less run readable
    #: rather than silent.
    #:
    #: REQUIRED WITHOUT DEFAULT, for ``cost_baseline_anchored``'s reason, sharpened by the shape:
    #: an empty map would read as "this proposal has no codes", and ``affected_items`` cannot be
    #: empty (``min_length=1``), so the empty default could never be true. The only honest
    #: alternative to a value here is a constructor that says what it saw.
    code_forms: dict[str, str]
    #: External service calls the run made (B4). EMPTY is a positive statement — "nothing outside
    #: this process was contacted" — not an absent field, which is why it is always serialized.
    external_calls: list[ExternalCall] = Field(default_factory=list)

    def to_annotations(self) -> list[Annotation]:
        """Map to MAF ``Annotation`` dicts for display only (NOT the source of truth)."""
        return [
            Annotation(
                type="citation",
                snippet=c.snippet,
                annotated_regions=[
                    TextSpanRegion(
                        type="text_span",
                        start_index=c.locator.start_index,
                        end_index=c.locator.end_index,
                    )
                ],
                additional_properties={"file": c.file},
            )
            for c in self.citations
        ]