"""Explanation records: the auditable answer to "why do you believe this?". Every asserted claim gets exactly one :class:`Explanation`, written to the operation log (``explanation.add``) *before* the ``claim.assert`` that references it. An explanation is self-contained: it embeds the deriver identity, the human-readable reasoning steps the deriver produced, a detail line for every input (with an excerpt of the underlying evidence), and the confidence arithmetic -- so a delegated node that receives only the claim + explanation can answer "why" without access to the raw evidence payloads. """ from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any, List, Mapping, Optional from mnema.derive.derivers.base import CandidateClaim, DeriverInfo from mnema.derive.graph import DerivationGraph, UnknownNodeError from mnema.derive.model import content_id @dataclass class Explanation: explanation_id: str claim_id: str deriver: str deriver_version: str summary: str reasoning: List[str] inputs: List[dict] # [{id, type, kind|predicate, excerpt}] confidence: float confidence_account: dict = field(default_factory=dict) generated_at: str = "" def to_dict(self) -> dict: return { "explanation_id": self.explanation_id, "claim_id": self.claim_id, "deriver": self.deriver, "deriver_version": self.deriver_version, "summary": self.summary, "reasoning": list(self.reasoning), "inputs": list(self.inputs), "confidence": self.confidence, "confidence_account": self.confidence_account, "generated_at": self.generated_at, } @classmethod def from_dict(cls, d: Mapping[str, Any]) -> "Explanation": return cls( explanation_id=d["explanation_id"], claim_id=d["claim_id"], deriver=d.get("deriver", "unknown"), deriver_version=d.get("deriver_version", "0"), summary=d.get("summary", ""), reasoning=list(d.get("reasoning", [])), inputs=list(d.get("inputs", [])), confidence=float(d.get("confidence", 0.0)), confidence_account=dict(d.get("confidence_account", {})), generated_at=d.get("generated_at", ""), ) def _truncate(s: str, limit: int = 100) -> str: s = " ".join(str(s).split()) return s if len(s) <= limit else s[: limit - 1] + "\u2026" def _evidence_excerpt(kind: str, payload: dict, observed_at: str) -> str: if kind == "calendar.event": title = payload.get("title") or payload.get("summary") or "(untitled event)" start = payload.get("start") or payload.get("start_time") or observed_at loc = payload.get("location") return _truncate(f"{title} @ {start}" + (f" ({loc})" if loc else "")) if kind == "note": text = payload.get("text") or payload.get("body") or payload.get("content") or "" title = payload.get("title") head = f"{title}: " if title else "" return _truncate(f"note {head}{text}") if kind == "photo.meta": when = payload.get("taken_at") or observed_at place = payload.get("place_name") or payload.get("place") if not place and payload.get("lat") is not None: place = f"{payload.get('lat')},{payload.get('lon') or payload.get('lng')}" return _truncate(f"photo taken {when}" + (f" at {place}" if place else "")) return _truncate(json.dumps(payload, sort_keys=True, ensure_ascii=False)) def input_detail(graph: DerivationGraph, node_id: str) -> dict: """Describe one provenance input for embedding in an explanation.""" try: ev = graph.get_evidence(node_id) return { "id": node_id, "type": "evidence", "kind": ev.kind, "source": ev.source, "excerpt": _evidence_excerpt(ev.kind, ev.payload, ev.observed_at), } except UnknownNodeError: pass try: claim = graph.get_claim(node_id) return { "id": node_id, "type": "claim", "predicate": claim.predicate, "deriver": claim.deriver, "confidence": round(claim.confidence, 6), "excerpt": _truncate( json.dumps(claim.identity, sort_keys=True, ensure_ascii=False) ), } except UnknownNodeError: return {"id": node_id, "type": "unknown", "excerpt": "(not in local graph)"} def build_explanation( claim_id: str, candidate: CandidateClaim, info: DeriverInfo, graph: DerivationGraph, generated_at: str, ) -> Explanation: return Explanation( explanation_id=content_id("exp", {"claim": claim_id, "at": generated_at}), claim_id=claim_id, deriver=info.deriver_id, deriver_version=info.version, summary=candidate.summary, reasoning=list(candidate.reasoning), inputs=[input_detail(graph, nid) for nid in candidate.inputs], confidence=candidate.confidence, confidence_account=dict(candidate.confidence_account), generated_at=generated_at, ) def render_explanation(exp: Explanation) -> str: """Human-readable multi-line rendering for the CLI.""" lines = [ f"WHY: {exp.summary}", f" deriver: {exp.deriver} v{exp.deriver_version}", f" confidence: {exp.confidence:.4f}", ] if exp.reasoning: lines.append(" reasoning:") for i, step in enumerate(exp.reasoning, 1): lines.append(f" {i}. {step}") if exp.inputs: lines.append(f" derived from {len(exp.inputs)} input(s):") for detail in exp.inputs: kind = detail.get("kind") or detail.get("predicate") or detail.get("type") lines.append(f" - [{kind}] {detail['id']}: {detail.get('excerpt', '')}") if exp.confidence_account: method = exp.confidence_account.get("method", "?") terms = { k: v for k, v in exp.confidence_account.items() if k not in ("method",) and v is not None } lines.append(f" confidence math: {method}") for k, v in terms.items(): lines.append(f" {k} = {v}") return "\n".join(lines)