`ad2304e`

Add typed evidence provenance core

Authored by

espadonne 1 month ago

SHA: ad2304e9eac54db4448433967bf510cdca73066a
Parents: 2671488
Tree: 36ab49b

5 changed files

Status	File	+	-
M	`src/loader/runtime/completion_trace.py`	15	1
A	`src/loader/runtime/evidence_provenance.py`	93	0
M	`src/loader/runtime/policy_timeline.py`	3	0
M	`src/loader/runtime/workflow_policy.py`	19	1
A	`tests/test_evidence_provenance.py`	79	0

src/loader/runtime/completion_trace.pymodified

  from dataclasses import dataclass, field
  from typing import Any
 +from .evidence_provenance import (
 +    EvidenceProvenance,
 +    normalize_evidence_provenance,
 +    summarize_evidence_provenance,
 +)
  from .workflow_policy import WorkflowTimelineEntry
      decision_code: str
      decision_summary: str
      evidence_summary: list[str] = field(default_factory=list)
 +    evidence_provenance: list[EvidenceProvenance] = field(default_factory=list)
      def to_dict(self) -> dict[str, str]:
          """Serialize the entry into persisted session state."""
              "decision_code": self.decision_code,
              "decision_summary": self.decision_summary,
              "evidence_summary": list(self.evidence_summary),
 +            "evidence_provenance": [item.to_dict() for item in self.evidence_provenance],
+         }
      @classmethod
                  for item in data.get("evidence_summary", [])
                  if str(item).strip()
              ],
 +            evidence_provenance=normalize_evidence_provenance(
 +                data.get("evidence_provenance")
 +            ),
+         )
          outcome=entry.policy_outcome or _completion_outcome_from_kind(entry.kind),
          decision_code=entry.reason_code,
          decision_summary=summary,
 -        evidence_summary=list(entry.evidence_summary),
 +        evidence_summary=list(
 +            entry.evidence_summary
 +            or summarize_evidence_provenance(entry.evidence_provenance)
 +        ),
 +        evidence_provenance=list(entry.evidence_provenance),
+     )

src/loader/runtime/evidence_provenance.pyadded

 +"""Typed evidence provenance carried through runtime policy decisions."""
++
 +from __future__ import annotations
++
 +from dataclasses import dataclass
 +from enum import StrEnum
 +from typing import Any
++
++
 +class EvidenceProvenanceStatus(StrEnum):
 +    """How one evidence item relates to a runtime decision."""
++
 +    SUPPORTS = "supports"
 +    MISSING = "missing"
 +    CONTRADICTS = "contradicts"
 +    CONTEXT = "context"
++
++
 +@dataclass(slots=True)
 +class EvidenceProvenance:
 +    """One typed piece of evidence behind a completion or verification decision."""
++
 +    category: str
 +    source: str
 +    summary: str
 +    status: str = EvidenceProvenanceStatus.CONTEXT.value
 +    subject: str | None = None
 +    detail: str | None = None
++
 +    def to_dict(self) -> dict[str, Any]:
 +        """Serialize one provenance item for persisted runtime state."""
++
 +        return {
 +            "category": self.category,
 +            "source": self.source,
 +            "summary": self.summary,
 +            "status": self.status,
 +            "subject": self.subject,
 +            "detail": self.detail,
 +        }
++
 +    @classmethod
 +    def from_dict(cls, data: dict[str, Any]) -> EvidenceProvenance:
 +        """Load one persisted provenance item."""
++
 +        return cls(
 +            category=str(data.get("category", "")),
 +            source=str(data.get("source", "")),
 +            summary=str(data.get("summary", "")),
 +            status=str(data.get("status", EvidenceProvenanceStatus.CONTEXT.value)),
 +            subject=_optional_text(data.get("subject")),
 +            detail=_optional_text(data.get("detail")),
 +        )
++
 +    def render_summary(self) -> str:
 +        """Render one concise human-facing summary."""
++
 +        return self.summary
++
++
 +def normalize_evidence_provenance(value: Any) -> list[EvidenceProvenance]:
 +    """Coerce persisted provenance payloads into typed entries."""
++
 +    if not isinstance(value, list):
 +        return []
 +    entries: list[EvidenceProvenance] = []
 +    for item in value:
 +        if isinstance(item, dict):
 +            entries.append(EvidenceProvenance.from_dict(item))
 +    return entries
++
++
 +def summarize_evidence_provenance(
 +    entries: list[EvidenceProvenance],
 +    *,
 +    max_items: int | None = None,
 +) -> list[str]:
 +    """Project typed provenance into concise evidence-summary strings."""
++
 +    summaries: list[str] = []
 +    limit = len(entries) if max_items is None else max_items
 +    for entry in entries[:limit]:
 +        summary = entry.render_summary().strip()
 +        if summary and summary not in summaries:
 +            summaries.append(summary)
 +    return summaries
++
++
 +def _optional_text(value: Any) -> str | None:
 +    if value is None:
 +        return None
 +    text = str(value).strip()
 +    return text or None

src/loader/runtime/policy_timeline.pymodified

  from .context import RuntimeContext
  from .events import TurnSummary
 +from .evidence_provenance import EvidenceProvenance
  from .workflow_policy import (
      WorkflowDecisionKind,
      WorkflowTimelineEntry,
      policy_outcome: str | None = None,
      decision_kind: WorkflowDecisionKind | str | None = WorkflowDecisionKind.FORCED,
      evidence_summary: list[str] | None = None,
 +    evidence_provenance: list[EvidenceProvenance] | None = None,
  ) -> WorkflowTimelineEntry:
      """Append one typed completion/repair accountability event."""
          prompt_format=context.prompt_format,
          prompt_sections=context.prompt_sections,
          evidence_summary=evidence_summary,
 +        evidence_provenance=evidence_provenance,
+     )
      context.session.append_workflow_timeline_entry(entry)
      summary.workflow_timeline = list(context.session.workflow_timeline)

src/loader/runtime/workflow_policy.pymodified

      describe_clarify_pressure_kind,
      describe_clarify_slot,
+ )
 +from .evidence_provenance import (
 +    EvidenceProvenance,
 +    normalize_evidence_provenance,
 +    summarize_evidence_provenance,
 +)
  from .workflow_signals import WorkflowSignalExtractor, WorkflowSignalPacket
      unresolved_questions: list[str] = field(default_factory=list)
      signal_summary: list[str] = field(default_factory=list)
      evidence_summary: list[str] = field(default_factory=list)
 +    evidence_provenance: list[EvidenceProvenance] = field(default_factory=list)
      clarify_stage: str | None = None
      clarify_pressure_kind: str | None = None
      pressure_pass_complete: bool = False
              "unresolved_questions": list(self.unresolved_questions),
              "signal_summary": list(self.signal_summary),
              "evidence_summary": list(self.evidence_summary),
 +            "evidence_provenance": [
 +                item.to_dict() for item in self.evidence_provenance
 +            ],
              "clarify_stage": self.clarify_stage,
              "clarify_pressure_kind": self.clarify_pressure_kind,
              "pressure_pass_complete": self.pressure_pass_complete,
              unresolved_questions=_string_list(data.get("unresolved_questions")),
              signal_summary=_string_list(data.get("signal_summary")),
              evidence_summary=_string_list(data.get("evidence_summary")),
 +            evidence_provenance=normalize_evidence_provenance(
 +                data.get("evidence_provenance")
 +            ),
              clarify_stage=_optional_text(data.get("clarify_stage")),
              clarify_pressure_kind=_optional_text(data.get("clarify_pressure_kind")),
              pressure_pass_complete=bool(data.get("pressure_pass_complete", False)),
              unresolved_questions=list(decision.unresolved_questions),
              signal_summary=list(decision.signal_summary),
              evidence_summary=list(decision.evidence_summary),
 +            evidence_provenance=[],
              clarify_stage=decision.clarify_stage,
              clarify_pressure_kind=decision.clarify_pressure_kind,
              pressure_pass_complete=decision.pressure_pass_complete,
          prompt_sections: list[str] | None = None,
          signal_summary: list[str] | None = None,
          evidence_summary: list[str] | None = None,
 +        evidence_provenance: list[EvidenceProvenance] | None = None,
          artifact_paths: list[str] | None = None,
      ) -> WorkflowTimelineEntry:
          """Build one typed non-routing accountability entry."""
              resolved_decision_kind = None
          else:
              resolved_decision_kind = str(decision_kind)
 +        resolved_provenance = list(evidence_provenance or [])
          return cls(
              timestamp=_utc_now(),
              kind=kind.value,
              summary=summary,
              decision_kind=resolved_decision_kind,
              signal_summary=list(signal_summary or []),
 -            evidence_summary=list(evidence_summary or []),
 +            evidence_summary=list(
 +                evidence_summary or summarize_evidence_provenance(resolved_provenance)
 +            ),
 +            evidence_provenance=resolved_provenance,
              policy_stage=policy_stage,
              policy_outcome=policy_outcome,
              prompt_format=prompt_format,

tests/test_evidence_provenance.pyadded

 +"""Tests for typed evidence provenance on policy timelines and traces."""
++
 +from __future__ import annotations
++
 +from loader.runtime.completion_trace import completion_trace_from_workflow_timeline
 +from loader.runtime.evidence_provenance import (
 +    EvidenceProvenance,
 +    EvidenceProvenanceStatus,
 +)
 +from loader.runtime.workflow_policy import WorkflowTimelineEntry, WorkflowTimelineEntryKind
++
++
 +def test_workflow_timeline_entry_derives_evidence_summary_from_provenance() -> None:
 +    entry = WorkflowTimelineEntry.accountability(
 +        kind=WorkflowTimelineEntryKind.COMPLETION_FINALIZE,
 +        mode="execute",
 +        reason_code="continuation_budget_exhausted",
 +        summary="completion: stopped because follow-through evidence was still missing",
 +        policy_stage="continuation_check",
 +        policy_outcome="finalize",
 +        evidence_provenance=[
 +            EvidenceProvenance(
 +                category="verification",
 +                source="dod.evidence",
 +                summary="verification evidence was still missing for `pytest -q`",
 +                status=EvidenceProvenanceStatus.MISSING.value,
 +                subject="pytest -q",
 +            )
 +        ],
 +    )
++
 +    assert entry.evidence_summary == [
 +        "verification evidence was still missing for `pytest -q`"
 +    ]
 +    assert entry.evidence_provenance[0].status == EvidenceProvenanceStatus.MISSING.value
++
++
 +def test_completion_trace_projection_preserves_evidence_provenance() -> None:
 +    timeline = [
 +        WorkflowTimelineEntry.accountability(
 +            kind=WorkflowTimelineEntryKind.COMPLETION_FINALIZE,
 +            mode="execute",
 +            reason_code="continuation_budget_exhausted",
 +            summary="completion: stopped because follow-through evidence was still missing",
 +            policy_stage="continuation_check",
 +            policy_outcome="finalize",
 +            evidence_provenance=[
 +                EvidenceProvenance(
 +                    category="verification",
 +                    source="dod.evidence",
 +                    summary="verification evidence was still missing for `pytest -q`",
 +                    status=EvidenceProvenanceStatus.MISSING.value,
 +                    subject="pytest -q",
 +                ),
 +                EvidenceProvenance(
 +                    category="action",
 +                    source="actions_taken",
 +                    summary="recorded work already showed the requested edit happened",
 +                    status=EvidenceProvenanceStatus.SUPPORTS.value,
 +                ),
 +            ],
 +        )
 +    ]
++
 +    trace = completion_trace_from_workflow_timeline(
 +        timeline,
 +        last_decision_code="continuation_budget_exhausted",
 +    )
++
 +    assert len(trace) == 1
 +    assert trace[0].decision_code == "continuation_budget_exhausted"
 +    assert trace[0].evidence_summary == [
 +        "verification evidence was still missing for `pytest -q`",
 +        "recorded work already showed the requested edit happened",
 +    ]
 +    assert [item.status for item in trace[0].evidence_provenance] == [
 +        EvidenceProvenanceStatus.MISSING.value,
 +        EvidenceProvenanceStatus.SUPPORTS.value,
 +    ]