feat: persist TEPP asynchronous lifecycle evidence

CHANGELOG.d/2.20.0-tepp-terminal-lifecycle.md

@@ -0,0 +1,7 @@
+# 2.20.0 — TEPP terminal lifecycle
+
+- Persists strict TEPP accepted receipts as transport evidence while the local
+  run remains Running.
+- Reads and revalidates TEPP's terminal-result v1 contract without resubmitting
+  accepted work or implementing local psychometric arithmetic.
+- Rejects request-binding and replay-digest mismatches before local success.

backend/app/analysis_run_ingestion.py

@@ -422,6 +422,21 @@ async def fetch_visible_analysis_run(
                 json.loads(envelope) if isinstance(envelope, str) else envelope
             )
             detail["topic_lineage_result_sha256"] = topic_result["result_sha256"]
+    if row["run_kind_code"] == _TEPP_RUN_KIND:
+        receipt = await conn.fetchrow(
+            """
+            select remote_run_id, accepted_status_code, received_at
+            from analysis_run_tepp_receipt
+            where analysis_run_id = $1
+            """,
+            analysis_run_id,
+        )
+        if receipt is not None:
+            detail["tepp_accepted_receipt"] = {
+                "remote_run_id": str(receipt["remote_run_id"]),
+                "accepted_status_code": str(receipt["accepted_status_code"]),
+                "received_at": _iso(receipt["received_at"]),
+            }
     return detail
 
 

backend/app/analysis_run_start.py

@@ -38,7 +38,12 @@
 from lineageweave.http_client import HttpClientError, post_json
 from lineageweave.lineage_persistence import lineage_edge_specs
 from lineageweave.models import Edge
-from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable
+from lineageweave.tepp_client import (
+    AnalysisRunRequest,
+    TeppClient,
+    TeppInvalidResponse,
+    TeppNotAvailable,
+)
 
 _LINEAGE_KIND = "analysis_run_lineage"
 _TEPP_KIND = "analysis_run_tepp"
@@ -110,6 +115,9 @@ class _DeliveryOutcome:
     envelope: dict[str, Any] | None = None
     source_snapshot_sha256: str | None = None
     knowledge_cutoff: datetime | None = None
+    request: AnalysisRunRequest | None = None
+    persist_receipt: bool = False
+    persist_terminal_result: bool = False
 
 
 def reconstruction_result_digest(edges: list[Edge]) -> str:
@@ -250,18 +258,145 @@ def _tepp_submission(
         response = client.submit_analysis_run(request)
     except TeppNotAvailable:
         return _FAILED, "tepp_not_available", None
+    except TeppInvalidResponse:
+        return _FAILED, "tepp_result_not_persisted", None
     if not isinstance(response, dict):
         return _FAILED, "tepp_result_not_persisted", None
-    if response.get("status") not in {"completed", "succeeded"}:
+    state = response.get("status") or response.get("run_state")
+    remote_run_id = response.get("analysis_run_id") or response.get("run_id")
+    if state == "accepted":
+        if (
+            set(response)
+            == {"contract_version", "run_id", "run_state", "idempotency_key"}
+            and response["contract_version"] == 1
+            and response["idempotency_key"] == request.idempotency_key
+            and isinstance(remote_run_id, str)
+            and remote_run_id.strip()
+        ):
+            return _RUNNING, "", response
+        return _FAILED, "tepp_result_not_persisted", None
+    if state not in {"completed", "succeeded"}:
         return _FAILED, "tepp_result_not_persisted", None
     if not isinstance(response.get("result"), dict):
         return _FAILED, "tepp_result_not_persisted", None
-    remote_run_id = response.get("analysis_run_id") or response.get("run_id")
     if not isinstance(remote_run_id, str) or not remote_run_id.strip():
         return _FAILED, "tepp_result_not_persisted", None
     return _SUCCEEDED, "", response
 
 
+def _tepp_status(
+    client: TeppClient,
+    request: AnalysisRunRequest,
+    remote_run_id: str,
+) -> tuple[str, str, dict[str, Any] | None]:
+    """Read one strict provider status; unavailable reads remain retryable."""
+    try:
+        response = client.read_analysis_run_status(remote_run_id, request)
+    except TeppNotAvailable:
+        return _RUNNING, "", None
+    except TeppInvalidResponse:
+        return _FAILED, "tepp_result_not_persisted", None
+    if response["run_state"] in {"accepted", "running"}:
+        return _RUNNING, "", response
+    terminal = response["terminal_result"]
+    if response["run_state"] == "failed":
+        return _FAILED, str(terminal["failure_code"]), response
+    return _SUCCEEDED, "", response
+
+
+async def _persist_tepp_terminal_result(
+    conn: asyncpg.Connection,
+    *,
+    analysis_run_id: str,
+    envelope: dict[str, Any],
+) -> bool:
+    """Persist a validated TEPP status envelope without reshaping its evidence."""
+    remote_run_id = envelope.get("run_id")
+    if envelope.get("run_state") != "succeeded" or not isinstance(remote_run_id, str):
+        return False
+    result_json = json.dumps(envelope, separators=(",", ":"), sort_keys=True)
+    result_sha256 = hashlib.sha256(result_json.encode("utf-8")).hexdigest()
+    try:
+        async with conn.transaction():
+            existing = await conn.fetchrow(
+                """
+                select remote_run_id, result_sha256
+                from analysis_run_tepp_result
+                where analysis_run_id = $1
+                """,
+                analysis_run_id,
+            )
+            if existing is not None:
+                return (
+                    str(existing["remote_run_id"]) == remote_run_id
+                    and str(existing["result_sha256"]) == result_sha256
+                )
+            await conn.execute(
+                """
+                insert into analysis_run_tepp_result
+                    (analysis_run_id, remote_run_id, result_json, result_sha256)
+                values ($1, $2, $3::jsonb, $4)
+                """,
+                analysis_run_id,
+                remote_run_id,
+                result_json,
+                result_sha256,
+            )
+    except (asyncpg.PostgresError, TypeError, ValueError):
+        return False
+    return True
+
+
+async def _persist_tepp_receipt(
+    conn: asyncpg.Connection,
+    *,
+    analysis_run_id: str,
+    request: AnalysisRunRequest,
+    envelope: dict[str, Any],
+) -> bool:
+    """Persist TEPP acceptance as transport evidence, never measurement."""
+    remote_run_id = envelope.get("analysis_run_id") or envelope.get("run_id")
+    state = envelope.get("status") or envelope.get("run_state")
+    if not isinstance(remote_run_id, str) or state != "accepted":
+        return False
+    request_json = json.dumps(request.to_json(), separators=(",", ":"), sort_keys=True)
+    receipt_json = json.dumps(envelope, separators=(",", ":"), sort_keys=True)
+    request_sha256 = hashlib.sha256(request_json.encode()).hexdigest()
+    receipt_sha256 = hashlib.sha256(receipt_json.encode()).hexdigest()
+    try:
+        async with conn.transaction():
+            existing = await conn.fetchrow(
+                """
+                select remote_run_id, request_sha256, receipt_sha256
+                from analysis_run_tepp_receipt
+                where analysis_run_id = $1
+                """,
+                analysis_run_id,
+            )
+            if existing is not None:
+                return (
+                    str(existing["remote_run_id"]) == remote_run_id
+                    and str(existing["request_sha256"]) == request_sha256
+                    and str(existing["receipt_sha256"]) == receipt_sha256
+                )
+            await conn.execute(
+                """
+                insert into analysis_run_tepp_receipt
+                    (analysis_run_id, remote_run_id, request_sha256, receipt_sha256,
+                     accepted_status_code)
+                values ($1, $2, $3, $4, $5)
+                """,
+                analysis_run_id,
+                remote_run_id,
+                request_sha256,
+                receipt_sha256,
+                state,
+            )
+    except (asyncpg.PostgresError, TypeError, ValueError):
+        return False
+    return True
+
+
 def tepp_submit_outcome(
     client: TeppClient,
     request: AnalysisRunRequest,
@@ -309,6 +444,8 @@ def topic_lineage_submit_outcome(
     item 3), not silently persisted as a topic-lineage result.
     """
     status_code, failure_code, envelope = _tepp_submission(client, request)
+    if status_code == _RUNNING:
+        return _FAILED, "tepp_result_not_persisted", None
     if status_code == _SUCCEEDED and not (
         envelope is not None and _topic_lineage_envelope_is_valid(envelope)
     ):
@@ -332,6 +469,19 @@ async def _persist_tepp_result(
     result_sha256 = hashlib.sha256(result_json.encode("utf-8")).hexdigest()
     try:
         async with conn.transaction():
+            existing = await conn.fetchrow(
+                """
+                select remote_run_id, result_sha256
+                from analysis_run_tepp_result
+                where analysis_run_id = $1
+                """,
+                analysis_run_id,
+            )
+            if existing is not None:
+                return (
+                    str(existing["remote_run_id"]) == remote_run_id
+                    and str(existing["result_sha256"]) == result_sha256
+                )
             await conn.execute(
                 """
                 insert into analysis_run_tepp_result
@@ -899,12 +1049,14 @@ async def _claim_delivery_plan(
         select outbox.analysis_run_id, outbox.work_kind_code,
                run.knowledge_cutoff, run.idempotency_key,
                run.analysis_source_snapshot_id, snapshot.snapshot_sha256,
-               scope.corporate_entity_id
+               scope.corporate_entity_id, receipt.remote_run_id
         from analysis_run_outbox outbox
         join analysis_run run on run.analysis_run_id = outbox.analysis_run_id
         join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id
         join analysis_source_snapshot snapshot
           on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id
+        left join analysis_run_tepp_receipt receipt
+          on receipt.analysis_run_id = run.analysis_run_id
         where outbox.analysis_run_id = $1
         for update of outbox
         """,
@@ -1003,12 +1155,20 @@ def _execute_delivery_plan(
         knowledge_cutoff=plan.locked["knowledge_cutoff"],
         corporate_entity_id=str(plan.locked["corporate_entity_id"]),
     )
+    persist_receipt = False
+    persist_terminal_result = False
     if plan.work_kind_code == _TOPIC_LINEAGE_KIND:
         status_code, failure_code, envelope = topic_lineage_submit_outcome(
             tepp_client, request
         )
+    elif plan.locked.get("remote_run_id"):
+        status_code, failure_code, envelope = _tepp_status(
+            tepp_client, request, str(plan.locked["remote_run_id"])
+        )
+        persist_terminal_result = status_code == _SUCCEEDED and envelope is not None
     else:
         status_code, failure_code, envelope = _tepp_submission(tepp_client, request)
+        persist_receipt = status_code == _RUNNING and envelope is not None
     return _DeliveryOutcome(
         plan.work_kind_code,
         plan.started_at,
@@ -1017,6 +1177,9 @@ def _execute_delivery_plan(
         envelope=envelope,
         source_snapshot_sha256=str(plan.locked["snapshot_sha256"]),
         knowledge_cutoff=plan.locked["knowledge_cutoff"],
+        request=request,
+        persist_receipt=persist_receipt,
+        persist_terminal_result=persist_terminal_result,
     )
 
 
@@ -1044,12 +1207,41 @@ async def _persist_delivery_outcome(
     finished = max(datetime.now(timezone.utc), outcome.started_at)
     status_code = outcome.status_code
     failure_code = outcome.failure_code
+    if outcome.work_kind_code == _TEPP_KIND and outcome.status_code == _RUNNING:
+        if outcome.persist_receipt:
+            persisted_receipt = (
+                outcome.envelope is not None
+                and outcome.request is not None
+                and await _persist_tepp_receipt(
+                    conn,
+                    analysis_run_id=analysis_run_id,
+                    request=outcome.request,
+                    envelope=outcome.envelope,
+                )
+            )
+            if not persisted_receipt:
+                status_code = _FAILED
+                failure_code = "tepp_receipt_not_persisted"
+            else:
+                return await _visible_or_404(
+                    conn, analysis_run_id, account_id, affiliated_entity_ids
+                )
+        else:
+            return await _visible_or_404(
+                conn, analysis_run_id, account_id, affiliated_entity_ids
+            )
     if outcome.work_kind_code == _LINEAGE_KIND:
         await _persist_lineage_reconstruction(
             conn, analysis_run_id=analysis_run_id, edges=outcome.edges, finished=finished
         )
     elif outcome.status_code == _SUCCEEDED and outcome.envelope is not None:
-        if outcome.work_kind_code == _TOPIC_LINEAGE_KIND:
+        if outcome.persist_terminal_result:
+            persisted = await _persist_tepp_terminal_result(
+                conn,
+                analysis_run_id=analysis_run_id,
+                envelope=outcome.envelope,
+            )
+        elif outcome.work_kind_code == _TOPIC_LINEAGE_KIND:
             persisted = await _persist_topic_lineage_result(
                 conn, analysis_run_id=analysis_run_id, envelope=outcome.envelope
             )