feat: recover TEPP validation on canonical Project history (v2.19.0)

CHANGELOG.md

@@ -4,6 +4,15 @@ All notable changes to this project are documented here. Format follows
 [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.19.0] - 2026-08-21
+
+### Added
+
+- Recovered the credential-free TEPP project-history validation boundary on top of
+  the canonical Buyer timeline. TEPP may return only cutoff-safe temporal
+  associations over the exact authorized events; the timeline remains readable
+  when TEPP is absent, and no result is labelled as a cause (ADR 0112).
+
 ## [2.18.0] - 2026-08-20
 
 ### Added

backend/app/analysis_run_start.py

@@ -11,7 +11,7 @@
 
 import hashlib
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 from uuid import UUID
 
@@ -21,13 +21,13 @@
     AnalysisRunCreateError,
     fetch_visible_analysis_run,
 )
-from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL
 from backend.app.analysis_run_outbox import (
     latest_outbox_delivery_is_claimed,
     latest_outbox_delivery_is_delivered,
     outbox_request_digest,
 )
 from backend.app.lineage_ingestion import records_from_source_posts
+from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL
 from lineageweave.adjudication_client import AdjudicationClient
 from lineageweave.http_client import HttpClientError, post_json
 from lineageweave.lineage_persistence import lineage_edge_specs
@@ -119,12 +119,12 @@ def tepp_run_request(
     """Build TEPP's published request from the frozen run, never a theta."""
     cutoff = knowledge_cutoff
     if cutoff.tzinfo is None:
-        cutoff = cutoff.replace(tzinfo=timezone.utc)
+        cutoff = cutoff.replace(tzinfo=UTC)
     return AnalysisRunRequest(
         idempotency_key=idempotency_key,
         tenant_workspace_id=str(corporate_entity_id),
         snapshot_id=snapshot_sha256,
-        knowledge_cutoff=cutoff.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        knowledge_cutoff=cutoff.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
         model_contract_version=_TEPP_MODEL_CONTRACT,
         output_profile=_TEPP_OUTPUT_PROFILE,
     )
@@ -610,7 +610,7 @@ async def deliver_queued_analysis_run(
         return await _visible_or_404(
             conn, analysis_run_id, account_id, affiliated_entity_ids
         )
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     try:
         if not latest_outbox_delivery_is_claimed(latest):
             await _append_outbox_delivery(
@@ -636,9 +636,8 @@ async def deliver_queued_analysis_run(
                 affiliated_entity_ids=affiliated_entity_ids,
                 adjudication_client=adjudication_client,
             )
-        finished = datetime.now(timezone.utc)
-        if finished < now:
-            finished = now
+        finished = datetime.now(UTC)
+        finished = max(finished, now)
         await _append_outbox_delivery(
             conn,
             analysis_run_id,
@@ -699,7 +698,7 @@ async def _deliver_lineage_reconstruction(
     adjudication_client: AdjudicationClient | None = None,
 ) -> None:
     """Persist ThreadWeave parent choices for the frozen bag."""
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     member_rows = await _snapshot_member_posts(
         conn,
         locked["analysis_source_snapshot_id"],
@@ -715,9 +714,8 @@ async def _deliver_lineage_reconstruction(
         )
     edges = lineage_edge_specs(records_from_source_posts(rows), llm=adjudication_client)
     digest = reconstruction_result_digest(edges)
-    finished = datetime.now(timezone.utc)
-    if finished < now:
-        finished = now
+    finished = datetime.now(UTC)
+    finished = max(finished, now)
     await conn.execute(
         """
         insert into analysis_run_reconstruction
@@ -759,25 +757,23 @@ async def _deliver_tepp_measurement(
     tepp_client: TeppClient,
 ) -> None:
     """Submit the frozen snapshot through ``tepp_client``. Never persist a theta."""
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     request = tepp_run_request(
         idempotency_key=str(locked["idempotency_key"]),
         snapshot_sha256=str(locked["snapshot_sha256"]),
         knowledge_cutoff=locked["knowledge_cutoff"],
         corporate_entity_id=str(locked["corporate_entity_id"]),
     )
     status_code, failure_code, envelope = _tepp_submission(tepp_client, request)
-    if status_code == _SUCCEEDED and envelope is not None:
-        if not await _persist_tepp_result(
-            conn,
-            analysis_run_id=analysis_run_id,
-            envelope=envelope,
-        ):
-            status_code = _FAILED
-            failure_code = "tepp_result_not_persisted"
-    finished = datetime.now(timezone.utc)
-    if finished < now:
-        finished = now
+    if status_code == _SUCCEEDED and envelope is not None and not await _persist_tepp_result(
+        conn,
+        analysis_run_id=analysis_run_id,
+        envelope=envelope,
+    ):
+        status_code = _FAILED
+        failure_code = "tepp_result_not_persisted"
+    finished = datetime.now(UTC)
+    finished = max(finished, now)
     await _append_status(
         conn,
         analysis_run_id,

backend/app/main.py

@@ -196,6 +196,10 @@
     fetch_project_history_index,
     fetch_project_history_projection,
 )
+from backend.app.tepp_project_history import (
+    tenant_workspace_reference,
+    validate_project_history_with_tepp,
+)
 from lineageweave.project_history import normalize_project_key
 from backend.app.demo_scope import (
     fetch_demo_corporate_entity_ids,
@@ -3439,7 +3443,7 @@ async def read_project_history(
             ) from exc
     async with pool.acquire() as conn:
         try:
-            return await fetch_project_history_projection(
+            projection = await fetch_project_history_projection(
                 conn,
                 project_key=project_key,
                 focus_post_id=focus_post_id,
@@ -3449,6 +3453,13 @@ async def read_project_history(
             )
         except ProjectHistoryNotFound as exc:
             raise HTTPException(status.HTTP_404_NOT_FOUND, "project history not found") from exc
+    projection["tepp_validation"] = await asyncio.to_thread(
+        validate_project_history_with_tepp,
+        projection=projection,
+        tenant_workspace_id=tenant_workspace_reference(account.corporate_entity_ids),
+        transport_url=load_settings().tepp_transport_url,
+    )
+    return projection
 
 
 @app.get("/api/rankings")

backend/app/tepp_project_history.py

@@ -0,0 +1,208 @@
+"""Map the canonical Buyer project history into TEPP's strict wire contract."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Iterable, Mapping, Sequence
+from typing import Any
+
+from lineageweave.tepp_project_history import (
+    PROJECT_HISTORY_CONTRACT_VERSION,
+    TeppProjectHistoryClient,
+    TeppProjectHistoryInvalidResponse,
+    TeppProjectHistoryUnavailable,
+    parse_rfc3339_utc,
+    project_history_event_sort_key,
+    validate_tepp_project_history_request,
+)
+
+
+def tenant_workspace_reference(corporate_entity_ids: Iterable[str]) -> str:
+    """Return a deterministic opaque workspace reference for the ABAC scope."""
+
+    normalized = sorted({str(value).strip() for value in corporate_entity_ids if str(value).strip()})
+    material = "\u001f".join(normalized) if normalized else "public-only"
+    digest = hashlib.sha256(material.encode("utf-8")).hexdigest()
+    return f"lw-workspace-{digest}"
+
+
+def _utc_text(value: object, field_name: str) -> str:
+    """Return canonical UTC text from one offset-aware source timestamp."""
+
+    return parse_rfc3339_utc(value, field_name)[1]
+
+
+def _opaque_actor_ids(
+    event: Mapping[str, Any],
+    *,
+    tenant_workspace_id: str,
+) -> list[str]:
+    """Hash canonical actor keys so names and local identifiers do not cross."""
+
+    raw_roles = event.get("responsibility_evidence")
+    if raw_roles is None:
+        raw_roles = event.get("observed_responsibilities")
+    if not isinstance(raw_roles, Sequence) or isinstance(raw_roles, (str, bytes)):
+        raw_roles = ()
+    actor_ids: set[str] = set()
+    for role in raw_roles:
+        if not isinstance(role, Mapping):
+            continue
+        actor_key = str(role.get("actor_key") or "").strip()
+        if not actor_key:
+            continue
+        material = f"{tenant_workspace_id}\u0000{actor_key}".encode("utf-8")
+        actor_ids.add(f"lw-actor-{hashlib.sha256(material).hexdigest()}")
+    return sorted(actor_ids)
+
+
+def _evidence_text(event: Mapping[str, Any]) -> str:
+    """Build bounded source-field evidence without sending a post body."""
+
+    title = str(event.get("event_title") or "").strip()
+    event_type = str(event.get("event_type_code") or "").strip()
+    if not title or not event_type:
+        raise TeppProjectHistoryUnavailable("canonical event title and type are required")
+    parts = [title, f"event_type={event_type}"]
+    for key in ("source_stage_code", "source_detail_state_code", "voc_type_code"):
+        value = str(event.get(key) or "").strip()
+        if value:
+            parts.append(f"{key}={value}")
+    rendered = " | ".join(parts)
+    encoded = rendered.encode("utf-8")
+    if len(encoded) <= 4096:
+        return rendered
+    return encoded[:4096].decode("utf-8", errors="ignore").rstrip()
+
+
+def _idempotency_key(request_without_key: Mapping[str, Any]) -> str:
+    """Hash the exact authorized evidence bundle into a stable request key."""
+
+    material = json.dumps(
+        request_without_key,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    digest = hashlib.sha256(material.encode("utf-8")).hexdigest()
+    return f"lineageweave-project-history-{digest}"
+
+
+def build_tepp_project_history_request(
+    *,
+    projection: Mapping[str, Any],
+    tenant_workspace_id: str,
+) -> dict[str, Any]:
+    """Build TEPP #159 input from the already-authorized canonical timeline."""
+
+    if projection.get("contract_version") != 1:
+        raise TeppProjectHistoryUnavailable("unsupported canonical project-history version")
+    events_value = projection.get("events")
+    if not isinstance(events_value, Sequence) or isinstance(events_value, (str, bytes)):
+        raise TeppProjectHistoryUnavailable("canonical project history has no event list")
+    cutoff = _utc_text(projection.get("knowledge_cutoff"), "knowledge_cutoff")
+    events: list[dict[str, Any]] = []
+    for value in events_value:
+        if not isinstance(value, Mapping):
+            raise TeppProjectHistoryUnavailable("canonical project event must be an object")
+        occurred_at = _utc_text(value.get("occurred_at"), "occurred_at")
+        event_id = str(value.get("event_id") or "").strip()
+        source_post_id = str(value.get("source_post_id") or "").strip()
+        if not event_id or not source_post_id:
+            raise TeppProjectHistoryUnavailable("canonical project event identity is missing")
+        events.append(
+            {
+                "event_id": event_id,
+                "event_type_code": str(value.get("event_type_code") or "").strip(),
+                "event_title": str(value.get("event_title") or "").strip(),
+                "occurred_at": occurred_at,
+                # The canonical timeline explicitly declares source-post creation
+                # time as its fallback clock. It is therefore also the earliest
+                # evidence-availability instant LineageWeave can substantiate.
+                "available_at": occurred_at,
+                "source_post_id": source_post_id,
+                "evidence_text": _evidence_text(value),
+                "actor_ids": _opaque_actor_ids(
+                    value,
+                    tenant_workspace_id=tenant_workspace_id,
+                ),
+            }
+        )
+    events.sort(key=project_history_event_sort_key)
+    request: dict[str, Any] = {
+        "contract_version": PROJECT_HISTORY_CONTRACT_VERSION,
+        "tenant_workspace_id": tenant_workspace_id,
+        "project_key": str(projection.get("project_key") or "").strip(),
+        "project_name": str(projection.get("project_name") or "").strip(),
+        "knowledge_cutoff": cutoff,
+        "focus_event_id": str(projection.get("focus_event_id") or "").strip(),
+        "events": events,
+    }
+    request["idempotency_key"] = _idempotency_key(request)
+    return validate_tepp_project_history_request(request)
+
+
+def _buyer_metadata(projection: Mapping[str, Any]) -> dict[str, Any]:
+    """Strip duplicate event rows while preserving TEPP findings and evidence IDs."""
+
+    events = projection["events"]
+    return {
+        "contract_version": projection["contract_version"],
+        "project_key": projection["project_key"],
+        "project_name": projection["project_name"],
+        "focus_event_id": projection["focus_event_id"],
+        "knowledge_cutoff": projection["knowledge_cutoff"],
+        "history_span_start": projection["history_span_start"],
+        "history_span_end": projection["history_span_end"],
+        "participant_count": projection["participant_count"],
+        "inference_status": projection["inference_status"],
+        "event_count": len(events),
+        "findings": projection["findings"],
+    }
+
+
+def validate_project_history_with_tepp(
+    *,
+    projection: Mapping[str, Any],
+    tenant_workspace_id: str,
+    transport_url: str,
+) -> dict[str, Any]:
+    """Return optional TEPP metadata without hiding the canonical timeline."""
+
+    if not transport_url.strip():
+        return {
+            "status": "not_configured",
+            "project_history": None,
+            "next_action_code": "configure_tepp_project_history",
+        }
+    try:
+        request = build_tepp_project_history_request(
+            projection=projection,
+            tenant_workspace_id=tenant_workspace_id,
+        )
+    except TeppProjectHistoryUnavailable:
+        return {
+            "status": "invalid_evidence",
+            "project_history": None,
+            "next_action_code": "open_source_evidence",
+        }
+    try:
+        validated = TeppProjectHistoryClient(transport_url).project(request)
+    except TeppProjectHistoryInvalidResponse:
+        return {
+            "status": "invalid_evidence",
+            "project_history": None,
+            "next_action_code": "open_source_evidence",
+        }
+    except TeppProjectHistoryUnavailable:
+        return {
+            "status": "unavailable",
+            "project_history": None,
+            "next_action_code": "retry_tepp_project_history",
+        }
+    return {
+        "status": "validated",
+        "project_history": _buyer_metadata(validated),
+        "next_action_code": "open_source_evidence",
+    }