from __future__ import annotations from dataclasses import dataclass import hashlib import json @dataclass(frozen=True) class CausalDiagnostic: metrics: dict[str, float] failures: tuple[str, ...] failure_locus: str | None earliest_failure_time_ms: float | None = None belief_event_id: str | None = None belief_time_ms: float | None = None decision_event_id: str | None = None decision_time_ms: float | None = None def _time(event: dict, field: str) -> float: value = event.get(field) return float(event["event_time_ms"] if value is None else value) def canonical_state_hash(event: dict) -> str: payload = {k: v for k, v in event.items() if k != "state_hash"} return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def evaluate_causal_ledger(ledger: dict) -> CausalDiagnostic: events = ledger.get("events", []) by_id: dict[str, dict] = {} failures: set[str] = set() failure_points: list[tuple[float, str, str]] = [] decision_candidates: list[tuple[float, str]] = [] for event in events: eid = event["event_id"] if eid in by_id: failure_points.append((_time(event, "event_time_ms"), "provenance", "duplicate_event_id")) else: by_id[eid] = event def ancestors(event: dict) -> set[str]: seen: set[str] = set() stack = list(event.get("parent_event_ids", [])) while stack: eid = stack.pop() if eid in seen: continue seen.add(eid) parent = by_id.get(eid) if parent: stack.extend(parent.get("parent_event_ids", [])) return seen checked_edges = invalid_edges = 0 last_append_sequence: int | None = None consequential_kinds = {"commitment", "shared_state_write", "tool_action", "spoken_consequence"} for event in events: eid = event["event_id"] et = _time(event, "event_time_ms") append_sequence = event.get("append_sequence") if append_sequence is not None: sequence = int(append_sequence) if last_append_sequence is not None and sequence <= last_append_sequence: failure_points.append((et, "provenance", "non_append_only_history")) last_append_sequence = sequence if event.get("state_hash") and event["state_hash"] != canonical_state_hash(event): failure_points.append((et, "provenance", "state_hash_mismatch")) for parent_id in event.get("parent_event_ids", []): checked_edges += 1 parent = by_id.get(parent_id) if parent is None: invalid_edges += 1 failure_points.append((et, "provenance", "invalid_causal_parent")) elif float(parent["evidence_available_time_ms"]) > et: invalid_edges += 1 failure_points.append((et, "context/prior_contamination", "future_knowledge_contamination")) if event.get("kind") in consequential_kinds: decision_candidates.append((et, eid)) for old_id in event.get("supersedes_event_ids", []): old = by_id.get(old_id) if old is None: failure_points.append((et, "provenance", "invalid_supersession_target")) elif event.get("kind") != "revision" or et <= _time(old, "event_time_ms"): failure_points.append((et, "provenance", "invalid_supersession_order")) fabricated_evidence = [ event for event in events if event.get("kind") == "acoustic_evidence" and event.get("fabricated") is True ] fabricated_ids = {event["event_id"] for event in fabricated_evidence} consequential = [event for event in events if event["kind"] in consequential_kinds] fabrication_consequences = 0 for event in consequential: if ancestors(event) & fabricated_ids: fabrication_consequences += 1 et = _time(event, "event_time_ms") failure_points.append((et, "perception", "fabricated_acoustic_evidence")) decision_candidates.append((et, event["event_id"])) commitments = [event for event in events if event["kind"] == "commitment"] revisions = [event for event in events if event["kind"] == "revision"] confidences = [event for event in events if event["kind"] == "confidence"] actions = [event for event in events if event["kind"] == "tool_action"] evidence_to_commit: list[float] = [] confidence_to_commit: list[float] = [] commit_to_action: list[float] = [] revision_costs: list[float] = [] premature_ids: set[str] = set() for commitment in commitments: cid = commitment["event_id"] ct = _time(commitment, "commitment_time_ms") evidence_to_commit.append(ct - float(commitment["evidence_available_time_ms"])) prior_conf = [event for event in confidences if float(event["event_time_ms"]) <= ct] if prior_conf: latest = max(prior_conf, key=lambda event: float(event["event_time_ms"])) confidence_to_commit.append(ct - _time(latest, "confidence_time_ms")) if latest.get("confidence") is not None and float(latest["confidence"]) < 0.5: premature_ids.add(cid) superseding = [ event for event in revisions if cid in event.get("supersedes_event_ids", []) and float(event["event_time_ms"]) > ct ] if superseding: first = min(superseding, key=lambda event: float(event["event_time_ms"])) premature_ids.add(cid) revision_costs.append(_time(first, "revision_time_ms") - ct) later = [event for event in actions if float(event["event_time_ms"]) >= ct] if later: first = min(later, key=lambda event: float(event["event_time_ms"])) commit_to_action.append(_time(first, "action_authorized_time_ms") - ct) for cid in premature_ids: ct = _time(by_id[cid], "commitment_time_ms") failure_points.append((ct, "commitment_routing", "premature_commitment")) decision_candidates.append((ct, cid)) for _, _, failure in failure_points: failures.add(failure) earliest = min(failure_points, key=lambda point: point[0]) if failure_points else None decision_event_id = decision_time_ms = None belief_event_id = belief_time_ms = None if decision_candidates: decision_time_ms, decision_event_id = min(decision_candidates, key=lambda item: item[0]) decision_event = by_id.get(decision_event_id) if decision_event is not None: belief_candidates = [ by_id[eid] for eid in ancestors(decision_event) if eid in by_id and by_id[eid].get("kind") in {"hypothesis", "confidence"} and _time(by_id[eid], "event_time_ms") <= decision_time_ms ] if belief_candidates: belief = max(belief_candidates, key=lambda event: _time(event, "event_time_ms")) belief_event_id = belief["event_id"] belief_time_ms = _time(belief, "event_time_ms") invalid_rate = invalid_edges / checked_edges if checked_edges else 0.0 parent_validity = 1.0 - invalid_rate historical_failures = { "future_knowledge_contamination", "non_append_only_history", "state_hash_mismatch", "invalid_supersession_target", "invalid_supersession_order", "duplicate_event_id", } historical_integrity = 0.0 if failures & historical_failures else 1.0 premature_rate = len(premature_ids) / max(1, len(commitments)) acoustic_evidence_count = len([event for event in events if event["kind"] == "acoustic_evidence"]) aefr = len(fabricated_evidence) / max(1, acoustic_evidence_count) fabrication_integrity = 0.0 if fabrication_consequences else 1.0 tbrf = min(parent_validity, historical_integrity) cci = min( historical_integrity, parent_validity, fabrication_integrity, 1.0 if not premature_ids else 0.0, ) def mean(values: list[float]) -> float: return round(sum(values) / len(values), 6) if values else 0.0 metrics = { "evidence_to_commit_delay_ms": mean(evidence_to_commit), "confidence_to_commit_delay_ms": mean(confidence_to_commit), "commit_to_action_delay_ms": mean(commit_to_action), "premature_commitment_rate": round(premature_rate, 6), "revision_cost_ms": mean(revision_costs), "acoustic_evidence_fabrication_rate": round(aefr, 6), "fabrication_consequence_rate": round(fabrication_consequences / max(1, len(consequential)), 6), "future_knowledge_contamination_rate": round( sum(1 for _, _, failure in failure_points if failure == "future_knowledge_contamination") / max(1, checked_edges), 6, ), "causal_parent_validity": round(parent_validity, 6), "historical_integrity": historical_integrity, "time_bounded_reconstruction_fidelity": round(tbrf, 6), "causal_cognition_integrity": round(cci, 6), "earliest_divergence_time_ms": earliest[0] if earliest else -1.0, } return CausalDiagnostic( metrics=metrics, failures=tuple(sorted(failures)), failure_locus=earliest[1] if earliest else None, earliest_failure_time_ms=earliest[0] if earliest else None, belief_event_id=belief_event_id, belief_time_ms=belief_time_ms, decision_event_id=decision_event_id, decision_time_ms=decision_time_ms, )