Three autonomous agents. One negotiated decision.

# OPTIC — Document Intelligence Kernel
from typing import Mapping
from cortex.types import (
    SourceRef, DocumentEnvelope, ExtractedRecord,
    LayoutTree, Typed, ProvenanceTree,
)

def acquire(source: SourceRef) -> DocumentEnvelope:
    """Typed, source-agnostic acquisition boundary."""
    raw  = source.fetch()
    meta = extract_metadata(raw)
    return DocumentEnvelope(
        payload     = raw,
        mime        = meta.mime,
        provenance  = meta.provenance,
    )

# Composable, pure transformations
pipeline = compose(
    parse_pdf,
    segment_layout,        # → LayoutTree
    project_to_schema,     # → Mapping[str, Typed]
    calibrate_confidence,  # redundant-decoder agreement
)

record: ExtractedRecord = pipeline(envelope)
# record.fields      :: Mapping[str, Typed]
# record.confidence  :: Mapping[str, float ∈ [0,1]]
# record.layout      :: LayoutTree
# record.provenance  :: ProvenanceTree

# NEXUS — Cross-Source Validation Kernel
from nexus.types import (
    ExtractedRecord, Source, Verdict, Evidence,
)
from nexus.priors import RELIABILITY_PRIOR

THRESHOLD: float = 0.78  # empirically calibrated

def validate(
    record: ExtractedRecord,
    sources: list[Source],
) -> Verdict:
    """Agreement functional a(R, E) under heterogeneous reliability."""
    evidence: list[Evidence] = [
        s.lookup(record.key) for s in sources
    ]

    # Weighted agreement under reliability prior
    weights = [RELIABILITY_PRIOR[s.id] for s in sources]
    score   = weighted_agreement(record, evidence, weights)

    return Verdict(
        consistent  = score >= THRESHOLD,
        confidence  = score,
        evidence    = evidence,
        arbitration = arbitrate_disagreement(record, evidence),
        provenance  = build_dag(record, evidence, weights),
    )

verdict = validate(record, sources=[s1, s2, s3])
# verdict.consistent  :: bool
# verdict.confidence  :: float ∈ [0,1]
# verdict.provenance  :: DAG[Source, Transform]

# AEGIS — Symbolic Restriction Evaluator
from aegis.types import (
    ExtractedRecord, Verdict, Constraint,
    Decision, Justification, Status,
)
from aegis.catalogue import RESTRICTION_SET

def evaluate(
    record:   ExtractedRecord,
    verdict:  Verdict,
    rules:    frozenset[Constraint] = RESTRICTION_SET,
) -> Decision:
    """Monotone symbolic evaluation of C against (R, verdict)."""
    violations: set[Constraint] = {
        c for c in rules if not c.holds(record, verdict)
    }

    status: Status = (
        Status.APPROVE if not violations and verdict.consistent
        else Status.REVIEW if verdict.confidence >= 0.6
        else Status.REJECT
    )

    return Decision(
        status      = status,
        violations  = frozenset(violations),
        rationale   = Justification.from_violations(violations),
        provenance  = verdict.provenance.extend(rules),
    )

decision = evaluate(record, verdict, RESTRICTION_SET)
# decision.status     :: {approve, review, reject}
# decision.violations :: Set[Constraint]
# decision.rationale  :: Justification

# HELIOS — Process Observability Kernel
from helios.types import (
    Event, Trace, PhaseVisit, ProcessGraph, Metrics,
)

FAMILIES = ("PHASE_FIRST_ENTRY", "PHASE_REENTRY",
            "PHASE_EXIT", "FIELD_MUTATION", "ANNOTATION")

def canonicalise(events: list[Event]) -> Trace:
    """Normalise, order and repair a raw event stream."""
    ordered = sorted(events, key=lambda e: (e.case_id, e.t))
    return Trace(
        case_id = ordered[0].case_id,
        events  = [normalise_text(e) for e in ordered],
    )

def visits(trace: Trace) -> list[PhaseVisit]:
    """Duration attributed ONCE per visit, not per event row.

    A visit is keyed by (phase, visit_ordinal); the three event
    families emitted for the same visit carry an identical
    duration and must not be summed independently.
    """
    seen: dict[tuple[str, int], PhaseVisit] = {}
    for e in trace.events:
        if e.family not in ("PHASE_FIRST_ENTRY",
                            "PHASE_REENTRY", "PHASE_EXIT"):
            continue
        key = (e.phase, e.visit_ordinal)
        seen.setdefault(key, PhaseVisit(
            phase    = e.phase,
            ordinal  = e.visit_ordinal,
            duration = e.phase_duration_s,   # counted once
        ))
    return list(seen.values())

def measure(traces: list[Trace]) -> tuple[ProcessGraph, Metrics]:
    graph = directly_follows(traces)          # G_p = (Φ, ⟶)
    return graph, Metrics(
        cycle_time    = percentiles(cycle_times(traces)),
        residence     = residence_by_phase(traces),
        rework_ratio  = rework(traces),        # re-entries / visits
        queue_delay   = handoff_delay(graph),
        conformance   = conformance(traces, DECLARATIVE_MODEL),
        deviations    = as_discovery(non_conforming(traces)),
    )

# ORACLE — Residual Cycle-Time Estimator
import numpy as np
from oracle.types import CaseView, Prediction, WindowReport

FEATURES = (
    "phase_id", "phase_age_h", "reentries", "attachments",
    "annotations", "optic_mean_conf", "nexus_disagreement",
)

def featurise(case: CaseView) -> np.ndarray:
    """Strictly prediction-time observables — no future leakage."""
    return np.array([case.get(f) for f in FEATURES], dtype=float)

def evaluate_blocked(windows: list[list[CaseView]]) -> list[WindowReport]:
    """Train on window w, evaluate on w+1. Never a random split:
    random splitting leaks future process states and inflates R².
    """
    reports = []
    for w in range(len(windows) - 1):
        X_tr = np.stack([featurise(c) for c in windows[w]])
        y_tr = np.array([c.residual_h for c in windows[w]])
        model = ridge_fit(X_tr, y_tr, alpha=1.0)

        X_te = np.stack([featurise(c) for c in windows[w + 1]])
        y_te = np.array([c.residual_h for c in windows[w + 1]])
        y_hat = model.predict(X_te)

        reports.append(WindowReport(
            window       = w + 1,
            mae          = mae(y_te, y_hat),
            rmse         = rmse(y_te, y_hat),
            r2           = r2(y_te, y_hat),
            coefficients = dict(zip(FEATURES, model.coef_)),
        ))
    return reports

def predict(case: CaseView, model) -> Prediction:
    mu = float(model.predict(featurise(case)[None, :])[0])
    return Prediction(
        residual_hours = mu,
        interval       = residual_interval(model, mu),
        drivers        = top_coefficients(model, k=3),
        emitted_to     = "operator_console",  # never actuated alone
    )