V2 P0: Temporal Truth Engine + /memory/state (#70)

api/main.py

@@ -28,6 +28,7 @@
 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.memory_state import router as memory_state_router
 from api.metrics import record_http_request, record_query, render_metrics
 from api.decisions import router as decisions_router
 from api.deps import RolesDep, memory, set_memory_service
@@ -107,6 +108,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 app.include_router(decisions_router)
 app.include_router(remember_router)
 app.include_router(evidence_router)
+app.include_router(memory_state_router)
 
 
 @app.middleware("http")

api/memory_state.py

@@ -0,0 +1,58 @@
+"""GET /memory/state — current or historical claim state for an entity."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from fastapi import APIRouter, Query
+from pydantic import BaseModel, Field
+
+from api.deps import RolesDep
+from temporal.state import TemporalStateService
+
+router = APIRouter(prefix="/memory", tags=["memory"])
+
+
+class MemoryStateResponse(BaseModel):
+    entity: str
+    workspace_id: str
+    at: str
+    mode: str
+    claims: list[dict[str, Any]] = Field(default_factory=list)
+    total: int
+
+
+@router.get(
+    "/state",
+    response_model=MemoryStateResponse,
+    summary="Current or historical memory state for an entity",
+)
+def memory_state(
+    _roles: RolesDep,
+    entity: str = Query(..., min_length=1),
+    workspace_id: str = Query(...),
+    at: datetime | None = Query(
+        default=None,
+        description="UTC timestamp for historical state; default now",
+    ),
+    current_only: bool = Query(
+        default=True,
+        description="If true, prefer active valid claims only",
+    ),
+) -> MemoryStateResponse:
+    """Return temporally filtered claims about an entity."""
+    when = at or datetime.now(UTC)
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=UTC)
+    svc = TemporalStateService()
+    try:
+        payload = svc.state_at(
+            entity=entity,
+            workspace_id=workspace_id,
+            at=when,
+            current_only=current_only,
+        )
+    finally:
+        svc.close()
+    return MemoryStateResponse(**payload)

temporal/__init__.py

@@ -0,0 +1,5 @@
+"""Temporal Truth package."""
+
+from temporal.validity import is_valid_at, parse_utc, rank_key
+
+__all__ = ["is_valid_at", "parse_utc", "rank_key"]

temporal/state.py

@@ -0,0 +1,103 @@
+"""Temporal state queries over Claim nodes."""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from typing import Any
+
+from neo4j import Driver, GraphDatabase
+
+from temporal.validity import is_valid_at, parse_utc, rank_key
+
+_CLAIMS_FOR_ENTITY = """
+MATCH (c:Claim)
+WHERE c.workspace_id = $workspace_id
+  AND (c.subject = $entity OR c.object = $entity OR c.decision_id = $entity)
+RETURN c {
+  .id, .subject, .predicate, .object, .content, .confidence, .status,
+  .valid_from, .valid_to, .observed_at, .invalidated_at, .superseded_by,
+  .assertion_type, .decision_id
+} AS claim
+"""
+
+
+class TemporalStateService:
+    """Query current or historical claim state for an entity."""
+
+    def __init__(self, driver: Driver | None = None) -> None:
+        self._owns = 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 state_at(
+        self,
+        *,
+        entity: str,
+        workspace_id: str,
+        at: datetime | None = None,
+        current_only: bool = False,
+    ) -> dict[str, Any]:
+        """Return claims for entity filtered/ranked for timestamp `at`."""
+        at = at or datetime.now(UTC)
+        with self._driver.session() as session:
+            rows = list(
+                session.run(
+                    _CLAIMS_FOR_ENTITY,
+                    workspace_id=workspace_id,
+                    entity=entity,
+                )
+            )
+        claims = [dict(r["claim"]) for r in rows if r.get("claim")]
+        if current_only:
+            claims = [
+                c
+                for c in claims
+                if is_valid_at(
+                    status=str(c.get("status") or ""),
+                    valid_from=c.get("valid_from"),
+                    valid_to=c.get("valid_to"),
+                    at=at,
+                )
+                and str(c.get("status")) not in {"SUPERSEDED", "ARCHIVED"}
+            ]
+        else:
+            # Historical: include claims whose validity window covers `at`
+            claims = [
+                c
+                for c in claims
+                if is_valid_at(
+                    status=str(c.get("status") or ""),
+                    valid_from=c.get("valid_from"),
+                    valid_to=c.get("valid_to"),
+                    at=at,
+                )
+                or (
+                    # include superseded that were still open at `at` via valid_from
+                    parse_utc(c.get("valid_from")) is not None
+                    and parse_utc(c.get("valid_from")) <= at  # type: ignore[operator]
+                    and (
+                        parse_utc(c.get("valid_to")) is None
+                        or parse_utc(c.get("valid_to")) > at  # type: ignore[operator]
+                    )
+                    and str(c.get("status")) == "SUPERSEDED"
+                )
+            ]
+        claims.sort(key=lambda c: rank_key(c, at=at), reverse=True)
+        return {
+            "entity": entity,
+            "workspace_id": workspace_id,
+            "at": at.isoformat(),
+            "mode": "current" if current_only else "historical",
+            "claims": claims,
+            "total": len(claims),
+        }
+
+    def close(self) -> None:
+        if self._owns:
+            self._driver.close()

temporal/validity.py

@@ -0,0 +1,68 @@
+"""Temporal validity helpers — current vs historical claim state."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+
+def parse_utc(value: Any) -> datetime | None:
+    """Parse ISO / Neo4j temporal to timezone-aware UTC."""
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value.astimezone(UTC)
+    if hasattr(value, "to_native"):
+        native = value.to_native()
+        if isinstance(native, datetime):
+            return parse_utc(native)
+    if isinstance(value, str):
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+    return None
+
+
+def is_valid_at(
+    *,
+    status: str,
+    valid_from: Any,
+    valid_to: Any,
+    at: datetime,
+) -> bool:
+    """Return True if claim is temporally valid at timestamp `at`."""
+    if status in {"QUARANTINED", "ERASED"}:
+        return False
+    start = parse_utc(valid_from)
+    end = parse_utc(valid_to)
+    if start and at < start:
+        return False
+    if end and at >= end:
+        return False
+    return True
+
+
+def rank_key(claim: dict[str, Any], *, at: datetime) -> tuple[int, float, str]:
+    """Sort key for current-state ranking (higher better via reverse sort)."""
+    status = str(claim.get("status") or "")
+    valid = is_valid_at(
+        status=status,
+        valid_from=claim.get("valid_from"),
+        valid_to=claim.get("valid_to"),
+        at=at,
+    )
+    status_rank = {
+        "VERIFIED": 5,
+        "ACTIVE": 4,
+        "CHALLENGED": 3,
+        "CANDIDATE": 2,
+        "SUPERSEDED": 1,
+        "ARCHIVED": 0,
+    }.get(status, 0)
+    if not valid:
+        status_rank = -1
+    conf = float(claim.get("confidence") or 0.0)
+    return (status_rank, conf, str(claim.get("id") or ""))

tests/temporal/test_validity.py

@@ -0,0 +1,44 @@
+"""Tests for temporal validity and ranking."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from temporal.validity import is_valid_at, rank_key
+
+
+def test_active_claim_valid_now() -> None:
+    now = datetime.now(UTC)
+    assert is_valid_at(
+        status="ACTIVE",
+        valid_from=now - timedelta(days=10),
+        valid_to=None,
+        at=now,
+    )
+
+
+def test_ended_claim_not_valid() -> None:
+    now = datetime.now(UTC)
+    assert not is_valid_at(
+        status="ACTIVE",
+        valid_from=now - timedelta(days=30),
+        valid_to=now - timedelta(days=1),
+        at=now,
+    )
+
+
+def test_quarantined_never_valid() -> None:
+    now = datetime.now(UTC)
+    assert not is_valid_at(
+        status="QUARANTINED",
+        valid_from=None,
+        valid_to=None,
+        at=now,
+    )
+
+
+def test_ranking_prefers_active_over_superseded() -> None:
+    now = datetime.now(UTC)
+    active = {"id": "a", "status": "ACTIVE", "confidence": 0.5}
+    superseded = {"id": "s", "status": "SUPERSEDED", "confidence": 0.99}
+    assert rank_key(active, at=now) > rank_key(superseded, at=now)