fidelity_framework_v1.py

"""
Judgment Fidelity Framework v1.0

Principle:
- Runtime emits facts only (ObservedEvent).
- Offline computes findings (Assessor).
- Obligations are the only durable outputs (ObligationLedger).
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple, Protocol
from contextvars import ContextVar
from uuid import uuid4
import json
import re
import warnings


SCHEMA_VERSION = "1.0"

# -----------------------------
# Extension Interfaces (No-Op)
# -----------------------------

class ActorResolver(Protocol):
    """STUB interface: map actor_claim to verified identity in forks."""
    def resolve(self, actor_claim: str) -> Optional[str]: ...


class ProvenanceProvider(Protocol):
    """STUB interface: compute callsite fingerprint in forks."""
    def callsite_fingerprint(self) -> Optional[str]: ...


class SinkPolicy(Protocol):
    """STUB interface: swap regex sink scan for AST or wrapper enforcement."""
    def scan(self, code_path: str) -> List[Tuple[int, str, str]]: ...


class GatePolicy(Protocol):
    """STUB interface: optional gating logic in forks."""
    def should_block(self, obligations: "ObligationLedger") -> bool: ...


class LedgerBackend(Protocol):
    """STUB interface: persist obligations in forks (ticketing/workflow)."""
    def persist(self, ledger: "ObligationLedger") -> None: ...


@dataclass
class NoOpActorResolver:
    """STUB: visibility-only default. Returns no verification."""
    def resolve(self, actor_claim: str) -> Optional[str]:
        warnings.warn("NoOpActorResolver is a stub; no verification is performed.")
        return None


@dataclass
class NoOpProvenanceProvider:
    """STUB: visibility-only default. No provenance fingerprints."""
    def callsite_fingerprint(self) -> Optional[str]:
        warnings.warn("NoOpProvenanceProvider is a stub; no provenance is captured.")
        return None


@dataclass
class NoOpGatePolicy:
    """STUB: visibility-only default. Never blocks."""
    def should_block(self, obligations: "ObligationLedger") -> bool:
        warnings.warn("NoOpGatePolicy is a stub; no enforcement is applied.")
        return False


@dataclass
class NoOpLedgerBackend:
    """STUB: visibility-only default. No persistence."""
    def persist(self, ledger: "ObligationLedger") -> None:
        warnings.warn("NoOpLedgerBackend is a stub; obligations are not persisted.")
        return None


@dataclass
class RingProfile:
    name: str
    ring_level: int  # 0 = heuristic visibility only
    sink_policy: SinkPolicy
    actor_resolver: ActorResolver
    provenance_provider: ProvenanceProvider
    gate_policy: GatePolicy
    ledger_backend: LedgerBackend

# -----------------------------
# Layer 1: Observation Runtime
# -----------------------------

@dataclass
class ObservedEvent:
    event_id: str
    timestamp: datetime
    correlation_id: Optional[str]
    anchor_id: str
    location_id: str
    event_type: str
    decision: str  # allow | deny | unknown
    actor_claim: str  # who/what claims authority
    parent_event_id: Optional[str] = None
    step_id: Optional[str] = None
    data: Dict[str, Any] = field(default_factory=dict)
    schema_version: str = SCHEMA_VERSION


_correlation_id_ctx: ContextVar[Optional[str]] = ContextVar("correlation_id", default=None)
_trace_log: ContextVar[List[ObservedEvent]] = ContextVar("observed_events", default=[])
_parent_event_ctx: ContextVar[Optional[str]] = ContextVar("parent_event_id", default=None)


def set_correlation_id(correlation_id: str) -> None:
    _correlation_id_ctx.set(correlation_id)


def get_correlation_id() -> Optional[str]:
    return _correlation_id_ctx.get()


def clear_correlation_id() -> None:
    _correlation_id_ctx.set(None)


def clear_events() -> None:
    _trace_log.set([])
    _parent_event_ctx.set(None)


def get_events() -> List[ObservedEvent]:
    return list(_trace_log.get())


def emit_observed_event(
    *,
    anchor_id: str,
    location_id: str,
    event_type: str,
    decision: str,
    actor_claim: str,
    step_id: Optional[str] = None,
    **data: Any,
) -> ObservedEvent:
    """
    Emit an ObservedEvent. No plan access, no assessment logic.

    Missing correlation_id is recorded as a fact via TRACE_CONTEXT_MISSING event.
    """
    correlation_id = get_correlation_id()
    if correlation_id is None:
        missing_event = ObservedEvent(
            event_id=str(uuid4()),
            timestamp=datetime.now(timezone.utc),
            correlation_id=None,
            anchor_id="trace",
            location_id="trace_context",
            event_type="TRACE_CONTEXT_MISSING",
            decision="unknown",
            actor_claim=actor_claim,
            parent_event_id=_parent_event_ctx.get(),
            step_id=step_id,
            data={
                "missing": "correlation_id",
                "attempted_anchor_id": anchor_id,
                "attempted_location_id": location_id,
                "attempted_event_type": event_type,
            },
            schema_version=SCHEMA_VERSION,
        )
        _trace_log.get().append(missing_event)
        _parent_event_ctx.set(missing_event.event_id)
        correlation_id = None

    event = ObservedEvent(
        event_id=str(uuid4()),
        timestamp=datetime.now(timezone.utc),
        correlation_id=correlation_id,
        anchor_id=anchor_id,
        location_id=location_id,
        event_type=event_type,
        decision=decision,
        actor_claim=actor_claim,
        parent_event_id=_parent_event_ctx.get(),
        step_id=step_id,
        data=data,
        schema_version=SCHEMA_VERSION,
    )
    _trace_log.get().append(event)
    _parent_event_ctx.set(event.event_id)
    return event


# -----------------------------
# Plan Schema (Typed IDs)
# -----------------------------

@dataclass
class AnchorSpec:
    anchor_id: str
    location_id: str
    event_types: List[str]
    event_order: List[str]
    allowed_decisions: List[str]
    allowed_actor_claims: List[str]
    finality: str


@dataclass
class DecisionAuthority:
    decision_point: str
    authority_holder: str
    escalation_path: str
    conditions: str


@dataclass
class Plan:
    version: str
    name: str
    owner: str
    signature: str
    anchor_specs: List[AnchorSpec]
    decision_authorities: List[DecisionAuthority]
    schema_versions: List[str]
    stakeholders: List[str]

    def validate(self) -> Tuple[bool, str]:
        if not self.owner:
            return False, "Plan missing owner"
        if not self.signature:
            return False, "Plan missing signature"
        if not self.anchor_specs:
            return False, "Plan missing anchor specs"
        return True, "ok"


# -----------------------------
# Layer 2: Assessor (Findings)
# -----------------------------

@dataclass
class AssessmentFinding:
    finding_id: str
    finding_type: str  # coverage_missing | schema_mismatch | order_violation | unassessable
    severity: str  # info | review | critical
    message: str
    evidence_event_ids: List[str]
    location_id: Optional[str] = None
    kind: str = "other"  # coverage | schema | ordering | integrity


@dataclass
class AssessmentResult:
    plan_version: str
    code_reference: str
    findings: List[AssessmentFinding]
    unassessable: bool


def assess_plan(plan: Plan, events: Iterable[ObservedEvent], code_reference: str) -> AssessmentResult:
    valid, msg = plan.validate()
    if not valid:
        return AssessmentResult(
            plan_version=plan.version,
            code_reference=code_reference,
            findings=[AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="plan_invalid",
                severity="critical",
                message=msg,
                evidence_event_ids=[],
                kind="integrity",
            )],
            unassessable=True,
        )

    events_list = list(events)
    findings: List[AssessmentFinding] = []
    unassessable = False

    # Schema version compatibility
    for event in events_list:
        if event.schema_version not in plan.schema_versions:
            findings.append(AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="schema_mismatch",
                severity="review",
                message=f"Unexpected schema_version {event.schema_version}",
                evidence_event_ids=[event.event_id],
                location_id=event.location_id,
                kind="schema",
            ))
            unassessable = True

    # Correlation context missing
    for event in events_list:
        if event.event_type == "TRACE_CONTEXT_MISSING":
            findings.append(AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="unassessable",
                severity="critical",
                message="Missing correlation context; trace integrity compromised.",
                evidence_event_ids=[event.event_id],
                kind="integrity",
            ))
            unassessable = True
        elif event.correlation_id is None:
            findings.append(AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="unassessable",
                severity="critical",
                message="Event missing correlation_id; trace integrity compromised.",
                evidence_event_ids=[event.event_id],
                location_id=event.location_id,
                kind="integrity",
            ))
            unassessable = True

    # Coverage and schema checks per anchor
    for spec in plan.anchor_specs:
        relevant = [
            e for e in events_list
            if e.anchor_id == spec.anchor_id and e.location_id == spec.location_id
        ]
        if not relevant:
            findings.append(AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="coverage_missing",
                severity="review",
                message=f"No observed events for anchor {spec.anchor_id}:{spec.location_id}",
                evidence_event_ids=[],
                location_id=spec.location_id,
                kind="coverage",
            ))
            continue

        # Event type validation
        if spec.event_types:
            for event in relevant:
                if event.event_type not in spec.event_types:
                    findings.append(AssessmentFinding(
                        finding_id=str(uuid4()),
                        finding_type="schema_mismatch",
                        severity="review",
                        message=f"Event type '{event.event_type}' not allowed for {spec.location_id}",
                        evidence_event_ids=[event.event_id],
                        location_id=spec.location_id,
                        kind="schema",
                    ))

        # Decision and actor claim validation
        for event in relevant:
            if event.decision not in spec.allowed_decisions:
                findings.append(AssessmentFinding(
                    finding_id=str(uuid4()),
                    finding_type="schema_mismatch",
                    severity="review",
                    message=f"Decision '{event.decision}' not allowed for {spec.location_id}",
                    evidence_event_ids=[event.event_id],
                    location_id=spec.location_id,
                    kind="schema",
                ))
            if spec.allowed_actor_claims and event.actor_claim not in spec.allowed_actor_claims:
                findings.append(AssessmentFinding(
                    finding_id=str(uuid4()),
                    finding_type="schema_mismatch",
                    severity="review",
                    message=f"Actor claim '{event.actor_claim}' not allowed for {spec.location_id}",
                    evidence_event_ids=[event.event_id],
                    location_id=spec.location_id,
                    kind="schema",
                ))

        # Order check (use observed order, not timestamps)
        observed_order = [e.event_type for e in relevant]
        if spec.event_order and observed_order != spec.event_order:
            findings.append(AssessmentFinding(
                finding_id=str(uuid4()),
                finding_type="order_violation",
                severity="review",
                message=f"Expected order {spec.event_order}, got {observed_order}",
                evidence_event_ids=[e.event_id for e in relevant],
                location_id=spec.location_id,
                kind="ordering",
            ))

    return AssessmentResult(
        plan_version=plan.version,
        code_reference=code_reference,
        findings=findings,
        unassessable=unassessable,
    )


# -----------------------------
# Layer 3: Obligation Ledger
# -----------------------------

@dataclass
class Obligation:
    obligation_id: str
    obligation_type: str
    owner: str
    required_action: str
    evidence_finding_ids: List[str]
    due_by: Optional[str] = None
    status: str = "open"  # open | accepted | resolved | superseded


@dataclass
class ObligationLedger:
    obligations: List[Obligation] = field(default_factory=list)

    def add(self, obligation: Obligation) -> None:
        self.obligations.append(obligation)

    def open_obligations(self) -> List[Obligation]:
        return [o for o in self.obligations if o.status == "open"]


def findings_to_obligations(result: AssessmentResult, owner: str) -> ObligationLedger:
    ledger = ObligationLedger()
    for finding in result.findings:
        if finding.finding_type == "unassessable":
            action = "Restore trace integrity; rerun assessment"
            obligation_type = "trace_integrity"
        elif finding.finding_type == "coverage_missing":
            action = "Add instrumentation or justify missing anchor coverage"
            obligation_type = "coverage"
        elif finding.finding_type == "order_violation":
            action = "Review event ordering; fix or update plan"
            obligation_type = "ordering"
        elif finding.finding_type == "schema_mismatch":
            action = "Resolve schema mismatch or update plan"
            obligation_type = "schema"
        else:
            action = "Review finding and decide next steps"
            obligation_type = "review"

        ledger.add(Obligation(
            obligation_id=str(uuid4()),
            obligation_type=obligation_type,
            owner=owner,
            required_action=action,
            evidence_finding_ids=[finding.finding_id],
        ))
    return ledger


def render_report(result: AssessmentResult, ledger: ObligationLedger) -> str:
    """
    Render a human-readable report with no approval semantics.
    Statuses are: UNASSESSABLE | OBLIGATIONS_PRESENT | NO_NEW_OBLIGATIONS_CREATED.
    """
    obligations = ledger.open_obligations()
    if result.unassessable:
        status = "UNASSESSABLE"
    elif obligations:
        status = "OBLIGATIONS_PRESENT"
    else:
        status = "NO_NEW_OBLIGATIONS_CREATED"

    coverage_missing = sum(1 for f in result.findings if f.finding_type == "coverage_missing")
    lines = []
    lines.append("=" * 80)
    lines.append("JUDGMENT STRUCTURE REPORT (VISIBILITY ONLY)")
    lines.append("=" * 80)
    lines.append("DISCLAIMER:")
    lines.append("- This report does NOT verify authority, intent, or completeness.")
    lines.append("- It records observed facts and derives obligations for review.")
    lines.append("")
    lines.append(f"STATUS: {status}")
    lines.append(f"Plan version: {result.plan_version}")
    lines.append(f"Code reference: {result.code_reference}")
    lines.append("")
    lines.append("FINDINGS SUMMARY")
    lines.append(f"- Total findings: {len(result.findings)}")
    lines.append(f"- Coverage missing: {coverage_missing}")
    lines.append(f"- Obligations created: {len(obligations)}")
    lines.append("")
    if obligations:
        lines.append("OBLIGATIONS")
        lines.append("-" * 80)
        for obligation in obligations:
            lines.append(f"* {obligation.obligation_type} ({obligation.obligation_id})")
            lines.append(f"  Owner: {obligation.owner}")
            lines.append(f"  Action: {obligation.required_action}")
            lines.append(f"  Evidence: {', '.join(obligation.evidence_finding_ids)}")
            lines.append("")
    return "\n".join(lines)


# -----------------------------
# Transcripts (Continuity Tests)
# -----------------------------

@dataclass
class TranscriptEventMatcher:
    anchor_id: str
    event_type: str
    required_fields: Dict[str, Any] = field(default_factory=dict)


@dataclass
class GoldenTranscript:
    name: str
    expected_subsequence: List[TranscriptEventMatcher]
    allow_extra_events: bool = True


def assert_anchor_order(
    events: Iterable[ObservedEvent],
    anchor_id: str,
    location_id: str,
    expected_order: List[str],
) -> None:
    """Raise if observed events for an anchor are not in expected order."""
    relevant = [
        e for e in events
        if e.anchor_id == anchor_id and e.location_id == location_id
    ]
    observed = [e.event_type for e in relevant]
    if expected_order and observed != expected_order:
        raise ValueError(f"Expected order {expected_order}, got {observed}")


def assert_all_anchors_covered(plan: Plan, events: Iterable[ObservedEvent]) -> None:
    """Raise if any AnchorSpec has zero observed events."""
    events_list = list(events)
    missing = []
    for spec in plan.anchor_specs:
        relevant = [
            e for e in events_list
            if e.anchor_id == spec.anchor_id and e.location_id == spec.location_id
        ]
        if not relevant:
            missing.append(f"{spec.anchor_id}:{spec.location_id}")
    if missing:
        raise ValueError(f"Missing coverage for anchors: {missing}")


def _canonicalize_value(value: Any) -> Any:
    if isinstance(value, str):
        return value.replace("\\", "/").strip()
    if isinstance(value, list):
        return sorted(_canonicalize_value(v) for v in value)
    if isinstance(value, dict):
        return {k: _canonicalize_value(v) for k, v in sorted(value.items())}
    return value


def _event_matches(event: ObservedEvent, matcher: TranscriptEventMatcher) -> bool:
    if event.anchor_id != matcher.anchor_id or event.event_type != matcher.event_type:
        return False
    for key, expected in matcher.required_fields.items():
        if key in event.data:
            actual = event.data.get(key)
        elif hasattr(event, key):
            actual = getattr(event, key)
        else:
            return False
        if expected is None:
            continue
        if _canonicalize_value(actual) != _canonicalize_value(expected):
            return False
    return True


def match_transcript(events: Iterable[ObservedEvent], transcript: GoldenTranscript) -> Tuple[bool, str]:
    events_list = list(events)
    if not transcript.allow_extra_events:
        if len(events_list) != len(transcript.expected_subsequence):
            return False, "Transcript requires exact match (no extra events)."
        for event, matcher in zip(events_list, transcript.expected_subsequence):
            if not _event_matches(event, matcher):
                return False, f"Missing required event: {matcher.anchor_id}.{matcher.event_type}"
        return True, "Transcript matched (exact)."

    cursor = 0
    for matcher in transcript.expected_subsequence:
        matched = False
        while cursor < len(events_list):
            if _event_matches(events_list[cursor], matcher):
                matched = True
                cursor += 1
                break
            cursor += 1
        if not matched:
            return False, f"Missing required event: {matcher.anchor_id}.{matcher.event_type}"
    return True, "Transcript matched (subsequence)."


# -----------------------------
# Forbidden Sink Scanning (Tiered)
# -----------------------------

DEFAULT_FORBIDDEN_SINKS = [
    r"\bopen\(.*?,\s*['\"]([wax].*)['\"]",
    r"\.write_text\(",
    r"\.write_bytes\(",
    r"\bos\.replace\(",
    r"\bos\.rename\(",
    r"\bshutil\.copy",
    r"\bshutil\.copyfile",
    r"\bshutil\.move",
    r"\.to_csv\(",
    r"\.to_json\(",
]

DEFAULT_ALLOWED_WRAPPERS = [
    r"write_output_atomic",
    r"anchor\.commit",
    r"_write_atomic",
    r"_write_csv",
]

@dataclass
class RegexSinkPolicy:
    forbidden_patterns: Optional[List[str]] = None
    allowed_wrappers: Optional[List[str]] = None

    def scan(self, code_path: str) -> List[Tuple[int, str, str]]:
        return scan_forbidden_sinks(code_path, self.forbidden_patterns, self.allowed_wrappers)


def scan_forbidden_sinks(
    code_path: str,
    forbidden_patterns: Optional[List[str]] = None,
    allowed_wrappers: Optional[List[str]] = None,
) -> List[Tuple[int, str, str]]:
    forbidden_patterns = forbidden_patterns or DEFAULT_FORBIDDEN_SINKS
    allowed_wrappers = allowed_wrappers or DEFAULT_ALLOWED_WRAPPERS
    findings: List[Tuple[int, str, str]] = []
    with open(code_path, "r", encoding="utf-8") as handle:
        for idx, line in enumerate(handle, 1):
            if any(re.search(pat, line) for pat in allowed_wrappers):
                continue
            for pattern in forbidden_patterns:
                if re.search(pattern, line):
                    findings.append((idx, line.rstrip(), pattern))
                    break
    return findings


def assert_no_forbidden_sinks(
    code_path: str,
    forbidden_patterns: Optional[List[str]] = None,
    allowed_wrappers: Optional[List[str]] = None,
) -> None:
    findings = scan_forbidden_sinks(code_path, forbidden_patterns, allowed_wrappers)
    if findings:
        summary = "; ".join(f"line {line_no}: {pattern}" for line_no, _, pattern in findings)
        raise ValueError(f"Forbidden sink usage detected: {summary}")


# -----------------------------
# Serialization Helpers
# -----------------------------


def serialize_event(event: ObservedEvent) -> Dict[str, Any]:
    return {
        "event_id": event.event_id,
        "timestamp": event.timestamp.isoformat(),
        "correlation_id": event.correlation_id,
        "anchor_id": event.anchor_id,
        "location_id": event.location_id,
        "event_type": event.event_type,
        "decision": event.decision,
        "actor_claim": event.actor_claim,
        "parent_event_id": event.parent_event_id,
        "step_id": event.step_id,
        "data": event.data,
        "schema_version": event.schema_version,
    }


def serialize_events(events: Iterable[ObservedEvent]) -> str:
    payload = [serialize_event(event) for event in events]
    return json.dumps(payload, indent=2, sort_keys=True)


def default_profile() -> RingProfile:
    """Ring-0 visibility profile with no enforcement."""
    return RingProfile(
        name="ring-0-visibility",
        ring_level=0,
        sink_policy=RegexSinkPolicy(),
        actor_resolver=NoOpActorResolver(),
        provenance_provider=NoOpProvenanceProvider(),
        gate_policy=NoOpGatePolicy(),
        ledger_backend=NoOpLedgerBackend(),
    )