V2 P0: Evidence Graph + explain API (#69)

.env.example

@@ -125,6 +125,8 @@ CORTEX_CONTRADICTION_KAFKA=true
 CORTEX_SEMANTIC_ENABLED=false
 # V2 Evidence Graph writers (Claim/Evidence). Schema V010 applies regardless; writes gated here.
 CORTEX_EVIDENCE_GRAPH=false
+# Optional JSON overrides for authority class scores (merged onto defaults).
+# CORTEX_AUTHORITY_SCORES_JSON={"casual_chat":0.4,"merged_PR":0.85}
 # Decay worker interval when Compose profile `api` is up (seconds). Default 24h.
 CORTEX_DECAY_INTERVAL_SECONDS=86400
 

api/evidence.py

@@ -0,0 +1,48 @@
+"""Evidence Graph HTTP routes — explain Claim confidence."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, status
+from pydantic import BaseModel, Field
+
+from api.deps import RolesDep
+from evidence.explain import EvidenceExplainer
+
+router = APIRouter(prefix="/evidence", tags=["evidence"])
+
+
+class ExplainResponse(BaseModel):
+    claim: str
+    claim_id: str
+    confidence: float | None = None
+    status: str | None = None
+    assertion_type: str | None = None
+    temporal: dict[str, Any] = Field(default_factory=dict)
+    supporting_evidence: list[dict[str, Any]] = Field(default_factory=list)
+    conflicting_evidence: list[dict[str, Any]] = Field(default_factory=list)
+
+
+@router.get(
+    "/explain/{claim_id}",
+    response_model=ExplainResponse,
+    summary="Explain a claim via supporting and conflicting evidence",
+)
+def explain_claim(
+    claim_id: str,
+    workspace_id: str,
+    _roles: RolesDep,
+) -> ExplainResponse:
+    """Return evidence-backed explanation for a Claim node."""
+    explainer = EvidenceExplainer()
+    try:
+        payload = explainer.explain(claim_id=claim_id, workspace_id=workspace_id)
+    finally:
+        explainer.close()
+    if payload is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Claim {claim_id} not found in workspace {workspace_id}",
+        )
+    return ExplainResponse(**payload)

api/main.py

@@ -26,6 +26,7 @@
 from slowapi.errors import RateLimitExceeded
 
 from api.contradictions import router as contradictions_router
+from api.evidence import router as evidence_router
 from api.gdpr import router as gdpr_router
 from api.metrics import record_http_request, record_query, render_metrics
 from api.decisions import router as decisions_router
@@ -105,6 +106,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 app.include_router(gdpr_router)
 app.include_router(decisions_router)
 app.include_router(remember_router)
+app.include_router(evidence_router)
 
 
 @app.middleware("http")

evidence/__init__.py

@@ -1,15 +1,18 @@
-"""Evidence graph package — Claim/Evidence adapters and (later) explain APIs."""
+"""Evidence graph package — adapters, authority, writer, explain."""
 
 from evidence.adapters import (
     claim_evidence_bundle,
     decision_to_claim,
     decision_to_evidence,
     evidence_graph_enabled,
 )
+from evidence.authority import compare_authority, score_for_class
 
 __all__ = [
     "claim_evidence_bundle",
+    "compare_authority",
     "decision_to_claim",
     "decision_to_evidence",
     "evidence_graph_enabled",
+    "score_for_class",
 ]

evidence/authority.py

@@ -0,0 +1,54 @@
+"""Configurable source-authority policy for Evidence Graph."""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from shared.v2_models import AuthorityClass
+
+# Default ordering: higher = more authoritative (do not hardcode without tests).
+DEFAULT_AUTHORITY_SCORES: dict[str, float] = {
+    "formal_policy": 0.98,
+    "approved_ADR": 0.95,
+    "production_state": 0.90,
+    "merged_PR": 0.82,
+    "approved_ticket": 0.75,
+    "incident_record": 0.78,
+    "verified_human_statement": 0.65,
+    "casual_chat": 0.45,
+    "agent_inference": 0.35,
+    "external_unverified": 0.25,
+}
+
+
+def load_authority_scores() -> dict[str, float]:
+    """Load authority scores from CORTEX_AUTHORITY_SCORES_JSON or defaults."""
+    raw = os.environ.get("CORTEX_AUTHORITY_SCORES_JSON", "").strip()
+    if not raw:
+        return dict(DEFAULT_AUTHORITY_SCORES)
+    try:
+        parsed: dict[str, Any] = json.loads(raw)
+        out = dict(DEFAULT_AUTHORITY_SCORES)
+        for key, value in parsed.items():
+            out[str(key)] = float(value)
+        return out
+    except (json.JSONDecodeError, TypeError, ValueError):
+        return dict(DEFAULT_AUTHORITY_SCORES)
+
+
+def score_for_class(authority_class: AuthorityClass | str) -> float:
+    """Return configured authority score for a class."""
+    scores = load_authority_scores()
+    return float(scores.get(str(authority_class), 0.30))
+
+
+def compare_authority(a: str, b: str) -> int:
+    """Return 1 if a > b, -1 if a < b, 0 if equal."""
+    sa, sb = score_for_class(a), score_for_class(b)
+    if sa > sb:
+        return 1
+    if sa < sb:
+        return -1
+    return 0

evidence/explain.py

@@ -0,0 +1,81 @@
+"""Explain Claim confidence via supporting / conflicting evidence."""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import structlog
+from neo4j import Driver, GraphDatabase
+
+log = structlog.get_logger(__name__)
+
+_EXPLAIN_CYPHER = """
+MATCH (c:Claim {id: $claim_id})
+WHERE c.workspace_id = $workspace_id
+OPTIONAL MATCH (s:Evidence)-[:SUPPORTS]->(c)
+OPTIONAL MATCH (x:Evidence)-[:CONTRADICTS]->(c)
+RETURN c {
+  .id, .workspace_id, .subject, .predicate, .object, .content,
+  .confidence, .status, .assertion_type, .decision_id,
+  .valid_from, .valid_to, .observed_at
+} AS claim,
+collect(DISTINCT s {
+  .id, .source_type, .source_id, .authority_class, .authority_score,
+  .snippet, .integrity_state
+}) AS supporting,
+collect(DISTINCT x {
+  .id, .source_type, .source_id, .authority_class, .authority_score,
+  .snippet, .integrity_state
+}) AS conflicting
+"""
+
+
+class EvidenceExplainer:
+    """Assemble explainability payload for a Claim."""
+
+    def __init__(self, driver: Driver | None = None) -> None:
+        self._owns_driver = driver is None
+        if driver is None:
+            uri = os.environ.get("NEO4J_URI", "bolt://localhost:7687")
+            user = os.environ.get("NEO4J_USER", "neo4j")
+            password = os.environ.get("NEO4J_PASSWORD", "cortex_local")
+            self._driver = GraphDatabase.driver(uri, auth=(user, password))
+        else:
+            self._driver = driver
+
+    def explain(self, *, claim_id: str, workspace_id: str) -> dict[str, Any] | None:
+        """Return claim + evidence breakdown or None if missing."""
+        with self._driver.session() as session:
+            record = session.run(
+                _EXPLAIN_CYPHER,
+                claim_id=claim_id,
+                workspace_id=workspace_id,
+            ).single()
+        if record is None or record["claim"] is None:
+            return None
+        supporting = [e for e in (record["supporting"] or []) if e and e.get("id")]
+        conflicting = [e for e in (record["conflicting"] or []) if e and e.get("id")]
+        claim = dict(record["claim"])
+        return {
+            "claim": claim.get("content") or self._claim_text(claim),
+            "claim_id": claim.get("id"),
+            "confidence": claim.get("confidence"),
+            "status": claim.get("status"),
+            "assertion_type": claim.get("assertion_type"),
+            "temporal": {
+                "valid_from": claim.get("valid_from"),
+                "valid_to": claim.get("valid_to"),
+                "observed_at": claim.get("observed_at"),
+            },
+            "supporting_evidence": supporting,
+            "conflicting_evidence": conflicting,
+        }
+
+    @staticmethod
+    def _claim_text(claim: dict[str, Any]) -> str:
+        return f"{claim.get('subject')} {claim.get('predicate')} {claim.get('object')}"
+
+    def close(self) -> None:
+        if self._owns_driver:
+            self._driver.close()