feat(lineage): require TEPP criterion anchor for channel weights

backend/app/analysis_run_start.py

@@ -48,8 +48,9 @@
 _RUNNING = "analysis_status_running"
 _SUCCEEDED = "analysis_status_succeeded"
 _FAILED = "analysis_status_failed"
-_TEPP_MODEL_CONTRACT = "tepp-analysis-run-v1"
-_TEPP_OUTPUT_PROFILE = "calibrated_event_measurement"
+_TEPP_MODEL_CONTRACT = "tepp-lineage-criterion-v1"
+_TEPP_OUTPUT_PROFILE = "lineage_pair_criterion_anchor"
+_TEPP_LINEAGE_ANCHOR_SCHEMA = "tepp.lineage_criterion_anchor.v1"
 _TOPIC_LINEAGE_MODEL_CONTRACT = "tepp-topic-lineage-v1"
 _TOPIC_LINEAGE_OUTPUT_PROFILE = "topic_identity_lineage"
 
@@ -107,6 +108,8 @@ class _DeliveryOutcome:
     status_code: str = _SUCCEEDED
     failure_code: str = ""
     envelope: dict[str, Any] | None = None
+    source_snapshot_sha256: str | None = None
+    knowledge_cutoff: datetime | None = None
 
 
 def reconstruction_result_digest(edges: list[Edge]) -> str:
@@ -199,7 +202,7 @@ def tepp_run_request(
         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(timezone.utc).isoformat().replace("+00:00", "Z"),
         model_contract_version=_TEPP_MODEL_CONTRACT,
         output_profile=_TEPP_OUTPUT_PROFILE,
     )
@@ -227,7 +230,7 @@ def topic_lineage_run_request(
         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(timezone.utc).isoformat().replace("+00:00", "Z"),
         model_contract_version=_TOPIC_LINEAGE_MODEL_CONTRACT,
         output_profile=_TOPIC_LINEAGE_OUTPUT_PROFILE,
     )
@@ -318,8 +321,10 @@ async def _persist_tepp_result(
     *,
     analysis_run_id: str,
     envelope: dict[str, Any],
+    expected_snapshot_sha256: str,
+    expected_knowledge_cutoff: datetime,
 ) -> bool:
-    """Persist only a validated, remote-completed TEPP envelope."""
+    """Persist a completed TEPP envelope and any exact lineage anchor projection."""
     remote_run_id = envelope.get("analysis_run_id") or envelope.get("run_id")
     if not isinstance(remote_run_id, str) or not remote_run_id.strip():
         return False
@@ -339,6 +344,53 @@ async def _persist_tepp_result(
                 result_json,
                 result_sha256,
             )
+            anchor = envelope.get("result")
+            if (
+                envelope.get("result_schema_version") == _TEPP_LINEAGE_ANCHOR_SCHEMA
+                and isinstance(anchor, dict)
+            ):
+                try:
+                    raw_estimation_run_id = str(anchor["estimation_run_id"])
+                    estimation_run_id = str(UUID(raw_estimation_run_id))
+                    anchor_cutoff = datetime.fromisoformat(
+                        str(anchor["knowledge_cutoff"]).replace("Z", "+00:00")
+                    )
+                except (KeyError, TypeError, ValueError):
+                    anchor = None
+                expected_cutoff = expected_knowledge_cutoff
+                if expected_cutoff.tzinfo is None:
+                    expected_cutoff = expected_cutoff.replace(tzinfo=timezone.utc)
+                if anchor is not None and (
+                    anchor.get("anchor_kind_code") != "lineage_pair_criterion"
+                    or anchor.get("contract_version") != 1
+                    or raw_estimation_run_id != estimation_run_id
+                    or anchor.get("source_snapshot_sha256") != expected_snapshot_sha256
+                    or anchor_cutoff != expected_cutoff
+                    or anchor.get("criterion_validity_status") != "accepted"
+                    or type(anchor.get("validated_pair_count")) is not int
+                    or anchor["validated_pair_count"] <= 0
+                ):
+                    anchor = None
+                if anchor is not None:
+                    await conn.execute(
+                        """
+                        insert into lineage_weight_tepp_anchor
+                            (estimation_run_id, tepp_analysis_run_id,
+                             anchor_kind_code, anchor_contract_version,
+                             source_snapshot_sha256, knowledge_cutoff,
+                             criterion_validity_status_code, validated_pair_count)
+                        values ($1, $2, $3, $4, $5, $6, $7, $8)
+                        on conflict (estimation_run_id) do nothing
+                        """,
+                        estimation_run_id,
+                        analysis_run_id,
+                        anchor["anchor_kind_code"],
+                        anchor["contract_version"],
+                        anchor["source_snapshot_sha256"],
+                        anchor_cutoff,
+                        anchor["criterion_validity_status"],
+                        anchor["validated_pair_count"],
+                    )
     except (asyncpg.PostgresError, TypeError, ValueError):
         return False
     return True
@@ -948,6 +1000,8 @@ def _execute_delivery_plan(
         status_code=status_code,
         failure_code=failure_code,
         envelope=envelope,
+        source_snapshot_sha256=str(plan.locked["snapshot_sha256"]),
+        knowledge_cutoff=plan.locked["knowledge_cutoff"],
     )
 
 
@@ -980,14 +1034,24 @@ async def _persist_delivery_outcome(
             conn, analysis_run_id=analysis_run_id, edges=outcome.edges, finished=finished
         )
     elif outcome.status_code == _SUCCEEDED and outcome.envelope is not None:
-        persist = (
-            _persist_topic_lineage_result
-            if outcome.work_kind_code == _TOPIC_LINEAGE_KIND
-            else _persist_tepp_result
-        )
-        if not await persist(
-            conn, analysis_run_id=analysis_run_id, envelope=outcome.envelope
+        if outcome.work_kind_code == _TOPIC_LINEAGE_KIND:
+            persisted = await _persist_topic_lineage_result(
+                conn, analysis_run_id=analysis_run_id, envelope=outcome.envelope
+            )
+        elif (
+            outcome.source_snapshot_sha256 is not None
+            and outcome.knowledge_cutoff is not None
         ):
+            persisted = await _persist_tepp_result(
+                conn,
+                analysis_run_id=analysis_run_id,
+                envelope=outcome.envelope,
+                expected_snapshot_sha256=outcome.source_snapshot_sha256,
+                expected_knowledge_cutoff=outcome.knowledge_cutoff,
+            )
+        else:
+            persisted = False
+        if not persisted:
             status_code = _FAILED
             failure_code = "tepp_result_not_persisted"
     if outcome.work_kind_code != _LINEAGE_KIND:

backend/app/global_ask_queue.py

@@ -28,6 +28,7 @@
 import redis.asyncio as redis
 from fastapi import HTTPException, status
 
+from lineageweave.ask_delivery import build_ask_delivery
 from lineageweave.http_client import HttpClientError
 from lineageweave.observability import record_server_failure
 from lineageweave.post_chat import (
@@ -251,6 +252,7 @@ def can_see(row: asyncpg.Record) -> bool:
             "Ask Agent is unavailable: authorized evidence could not be assembled",
         ) from exc
     if not sources:
+        delivery = build_ask_delivery("", (), ())
         return {
             "answer_text": "",
             "cited_post_ids": [],
@@ -260,6 +262,7 @@ def can_see(row: asyncpg.Record) -> bool:
             "lineage_graph": {"nodes": [], "edges": [], "truncated": False},
             "cited_post_images": [],
             "next_action": "No authorized source posts are available for this question.",
+            "delivery": delivery,
         }
     try:
         answer = await asyncio.to_thread(
@@ -297,14 +300,17 @@ def can_see(row: asyncpg.Record) -> bool:
     async with pool.acquire() as conn:
         lineage_graph = await lineage_graphs_for_posts(conn, can_see, cited_ids)
         images = await cited_post_images(conn, cited_ids)
+    cited_posts = cited_post_summaries(sources, cited_ids)
+    cited_evidence = cited_post_evidence(sources, cited_ids)
     return {
         "answer_text": answer.answer_text,
         "cited_post_ids": cited_ids,
-        "cited_posts": cited_post_summaries(sources, cited_ids),
-        "cited_post_evidence": cited_post_evidence(sources, cited_ids),
+        "cited_posts": cited_posts,
+        "cited_post_evidence": cited_evidence,
         "cited_post_images": images,
         "source_post_ids": [source.post_id for source in sources],
         "lineage_graph": lineage_graph,
+        "delivery": build_ask_delivery(answer.answer_text, cited_posts, cited_evidence),
     }
 
 

backend/app/lineage_ingestion.py

@@ -31,14 +31,9 @@
 ISOLATION_NO_COMPARISON_GROUP = "no_comparison_group"
 ISOLATION_COMPARISON_CANDIDATES_AVAILABLE = "comparison_candidates_available"
 
-# Accepted ADR 0200 (points 2-3) authorizes exactly one anchor method:
-# expected-information estimates honestly labeled as validated by the
-# channels' internal response structure only, pending the TEPP
-# criterion-validity gate. When that gate exists, a set that fails it is
-# retired and this stays the only place an anchor method is ever added
-# -- ADR-first, per ADR 0145's original condition.
+# ADR 0205 authorizes only a completed, persisted TEPP criterion anchor.
 _SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset(
-    {"unanchored_internal_structure"}
+    {"tepp_lineage_criterion_v1"}
 )
 
 
@@ -165,12 +160,11 @@ async def load_estimated_channel_weights(
 ) -> dict[str, float] | None:
     """Load only a complete vector from an independently anchored method.
 
-    No anchor method is currently authorized (ADR 0200 point 3 names the
-    conditions under which one becomes authorized). A partial or invalid
-    vector returns ``None`` rather than being repaired. A database that has
-    not applied migration 0135 is likewise an unavailable state, detected
-    without issuing a statement that would abort the caller's outer
-    PostgreSQL transaction.
+    ADR 0205 authorizes only the exact persisted TEPP lineage-criterion
+    contract. A partial, internally anchored, or identity-mismatched vector
+    returns ``None`` rather than being repaired. A database missing either
+    persistence table is likewise an unavailable state, detected without an
+    aborting query inside the caller's transaction.
 
     Since migration 0200 one weight set is persisted per active-channel
     combination (``channel_set_code``): the corpus-wide rebuild's three
@@ -184,6 +178,11 @@ async def load_estimated_channel_weights(
     )
     if not table_exists:
         return None
+    anchor_table_exists = await conn.fetchval(
+        "select to_regclass('public.lineage_weight_tepp_anchor') is not null"
+    )
+    if not anchor_table_exists:
+        return None
     # Pre-0200 schemas lack channel_set_code; probe via the catalog (never
     # a failing statement, which would abort the caller's transaction).
     # Pre-0200 rows form one implicit deterministic set.
@@ -194,14 +193,30 @@ async def load_estimated_channel_weights(
         "  and column_name = 'channel_set_code')"
     )
     set_column_sql = (
-        "channel_set_code" if set_column_exists else "'channel_set_deterministic'"
+        "weight.channel_set_code" if set_column_exists else "'channel_set_deterministic'"
     )
     all_rows = await conn.fetch(
         f"select {set_column_sql} as channel_set_code, "
-        "channel_code, weight_value, estimation_run_id, "
-        "estimation_method_code, estimator_version, anchor_method_code, "
-        "source_snapshot_sha256, sample_pair_count, knowledge_cutoff "
-        "from lineage_channel_weight"
+        "weight.channel_code, weight.weight_value, weight.estimation_run_id, "
+        "weight.estimation_method_code, weight.estimator_version, weight.anchor_method_code, "
+        "weight.source_snapshot_sha256, weight.sample_pair_count, weight.knowledge_cutoff, "
+        "anchor.anchor_kind_code, anchor.anchor_contract_version, "
+        "anchor.source_snapshot_sha256 as anchor_snapshot_sha256, "
+        "anchor.knowledge_cutoff as anchor_knowledge_cutoff, "
+        "anchor.criterion_validity_status_code, anchor.validated_pair_count, "
+        "tepp_result.result_sha256 as tepp_result_sha256, "
+        "tepp_run.run_kind_code as tepp_run_kind_code, "
+        "tepp_snapshot.snapshot_sha256 as tepp_snapshot_sha256, "
+        "tepp_run.knowledge_cutoff as tepp_knowledge_cutoff "
+        "from lineage_channel_weight weight "
+        "left join lineage_weight_tepp_anchor anchor "
+        "on anchor.estimation_run_id = weight.estimation_run_id "
+        "left join analysis_run_tepp_result tepp_result "
+        "on tepp_result.analysis_run_id = anchor.tepp_analysis_run_id "
+        "left join analysis_run tepp_run "
+        "on tepp_run.analysis_run_id = tepp_result.analysis_run_id "
+        "left join analysis_source_snapshot tepp_snapshot "
+        "on tepp_snapshot.analysis_source_snapshot_id = tepp_run.analysis_source_snapshot_id"
     )
     sets: dict[str, list] = {}
     for row in all_rows:
@@ -264,6 +279,51 @@ async def load_estimated_channel_weights(
         or not isinstance(knowledge_cutoff, datetime)
     ):
         return None
+    if anchor_method == "tepp_lineage_criterion_v1":
+        anchor_values = {
+            (
+                row.get("anchor_kind_code"),
+                row.get("anchor_contract_version"),
+                row.get("anchor_snapshot_sha256"),
+                row.get("anchor_knowledge_cutoff"),
+                row.get("criterion_validity_status_code"),
+                row.get("validated_pair_count"),
+                row.get("tepp_result_sha256"),
+                row.get("tepp_run_kind_code"),
+                row.get("tepp_snapshot_sha256"),
+                row.get("tepp_knowledge_cutoff"),
+            )
+            for row in rows
+        }
+        if len(anchor_values) != 1:
+            return None
+        (
+            anchor_kind,
+            anchor_version,
+            anchor_snapshot,
+            anchor_cutoff,
+            validity_status,
+            validated_pairs,
+            tepp_digest,
+            tepp_run_kind,
+            tepp_snapshot,
+            tepp_cutoff,
+        ) = next(iter(anchor_values))
+        if (
+            estimation_method != "mls2plm_expected_information"
+            or anchor_kind != "lineage_pair_criterion"
+            or anchor_version != 1
+            or validity_status != "accepted"
+            or validated_pairs != sample_pair_count
+            or anchor_snapshot != snapshot_digest
+            or tepp_snapshot != snapshot_digest
+            or anchor_cutoff != knowledge_cutoff
+            or tepp_cutoff != knowledge_cutoff
+            or tepp_run_kind != "analysis_run_tepp"
+            or not isinstance(tepp_digest, str)
+            or re.fullmatch(r"[0-9a-f]{64}", tepp_digest) is None
+        ):
+            return None
     return persisted
 
 

backend/app/main.py

@@ -24,7 +24,7 @@
 import logging
 from contextlib import asynccontextmanager
 from dataclasses import asdict
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
 from typing import Any, Literal
 from uuid import UUID
 
@@ -160,6 +160,7 @@
     update_ticket,
     upsert_commitment_ticket,
 )
+from backend.app.operations_dashboard import fetch_operations_dashboard
 from backend.app.keyman_ingestion import ingest_post_keymen
 from backend.app.knowledge_graph import (
     corporate_entity_exists,
@@ -743,6 +744,24 @@ async def read_me(
     }
 
 
+@app.get("/api/dashboard")
+async def operations_dashboard(
+    period_start: date | None = Query(None),
+    period_end: date | None = Query(None),
+    account: CurrentAccount = Depends(get_current_account),
+    pool: asyncpg.Pool = Depends(get_pool),
+) -> dict[str, Any]:
+    """Show quantified operational cases backed by visible source evidence."""
+    _require_post_read(account)
+    async with pool.acquire() as conn:
+        try:
+            return await fetch_operations_dashboard(
+                conn, account.corporate_entity_ids, period_start, period_end
+            )
+        except ValueError as exc:
+            raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc)) from exc
+
+
 class LocalePreferenceRequest(BaseModel):
     """Body of a PATCH /api/me/preferences request."""