From e4753425c152499b6e64f7dd734d087633dbf6a8 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 00:00:56 +0000 Subject: [PATCH] =?UTF-8?q?V2=20P0:=20Temporal=20Truth=20Engine=20?= =?UTF-8?q?=E2=80=94=20validity=20helpers=20and=20/memory/state=20(#70).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add temporal.validity ranking, TemporalStateService, and GET /memory/state for current vs historical claim queries. Co-authored-by: Abhinaysai Kamineni --- api/main.py | 2 + api/memory_state.py | 58 ++++++++++++++++++ temporal/__init__.py | 5 ++ temporal/state.py | 103 ++++++++++++++++++++++++++++++++ temporal/validity.py | 68 +++++++++++++++++++++ tests/temporal/test_validity.py | 44 ++++++++++++++ 6 files changed, 280 insertions(+) create mode 100644 api/memory_state.py create mode 100644 temporal/__init__.py create mode 100644 temporal/state.py create mode 100644 temporal/validity.py create mode 100644 tests/temporal/test_validity.py diff --git a/api/main.py b/api/main.py index dabdc8d..0a78d3f 100644 --- a/api/main.py +++ b/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") diff --git a/api/memory_state.py b/api/memory_state.py new file mode 100644 index 0000000..a1646aa --- /dev/null +++ b/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) diff --git a/temporal/__init__.py b/temporal/__init__.py new file mode 100644 index 0000000..4c85085 --- /dev/null +++ b/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"] diff --git a/temporal/state.py b/temporal/state.py new file mode 100644 index 0000000..31a05bb --- /dev/null +++ b/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() diff --git a/temporal/validity.py b/temporal/validity.py new file mode 100644 index 0000000..f0fa1a7 --- /dev/null +++ b/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 "")) diff --git a/tests/temporal/test_validity.py b/tests/temporal/test_validity.py new file mode 100644 index 0000000..86e70a1 --- /dev/null +++ b/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)