diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d59bb7d59..bf0fcc169 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -489,11 +489,13 @@ more than one), then open the Pending row to confirm the cutoff corpus. `POST /api/analysis-runs/{id}/start` then commits Running plus a durable outbox row, wakes Valkey, and delivers ThreadWeave on that frozen bag (ADR 0021 / ADR 0023) or submits TEPP through -`tepp_client` (ADR 0022). It does not invent a TEPP score. +`tepp_client` (ADR 0022 / ADR 0162). It does not invent a TEPP score. Request a lineage reconstruction from the home list, open the Pending row, then start reconstruction. A Pending TEPP row starts a measurement; a missing transport stays Failed / -`tepp_not_available`. Hover the Result digest +`tepp_not_available`. A live `accepted` envelope with a remote run id +persists as transport evidence and keeps the run Running — that +receipt is not a calibrated result. Hover the Result digest prefix, then confirm the designed A-100 fork before treating the live Event Lineage panel as that run's tree. `make seed` also records a TEPP measurement run through diff --git a/CHANGELOG.d/2.12.11-tepp-accepted-receipt.md b/CHANGELOG.d/2.12.11-tepp-accepted-receipt.md new file mode 100644 index 000000000..f722398ae --- /dev/null +++ b/CHANGELOG.d/2.12.11-tepp-accepted-receipt.md @@ -0,0 +1,8 @@ +# 2.12.11 TEPP accepted receipt + +A live TEPP `accepted` / `queued` / `running` envelope with a remote +run id persists as transport evidence and leaves the local analysis +run Running. The receipt is not a measurement and does not invent a +theta. Empty accepted envelopes and missing transport stay Failed. +Completed envelopes still persist `analysis_run_tepp_result`. Login +no longer mounts Admin settings with an undefined token. diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dbd797ef..04ee6ef67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -235,6 +235,24 @@ All notable changes to this project are documented here. Format follows lineage run and Start still recovers the designed A-100 fork. Never invent a theta. +## [2.12.11] - 2026-08-23 + +### Added + +- A live TEPP `accepted` / `queued` / `running` envelope that carries a + remote run id is stored as transport evidence + (`analysis_run_tepp_accepted_receipt`). The local analysis run stays + Running. The receipt is not a calibrated measurement and does not + invent a theta. Empty accepted envelopes and a missing transport stay + Failed. Completed envelopes still persist `analysis_run_tepp_result`. + Completed-result polling remains blocked on TEPP#156. + +### Fixed + +- Unauthenticated login no longer mounts Admin settings with an + undefined access token, and login persists a validated OIDC return + URL before the redirect. + ## [2.12.6] - 2026-08-20 ### Added diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index 1817b0b16..587ac9f9f 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -252,15 +252,16 @@ async def fetch_outbox_deliveries( entry ids stay off the payload -- they are not reader-facing evidence. """ try: - rows = await conn.fetch( - """ - select delivery_ordinal, delivery_status_code, occurred_at - from analysis_run_outbox_delivery - where analysis_run_id = $1::uuid - order by delivery_ordinal - """, - analysis_run_id, - ) + async with conn.transaction(): + rows = await conn.fetch( + """ + select delivery_ordinal, delivery_status_code, occurred_at + from analysis_run_outbox_delivery + where analysis_run_id = $1::uuid + order by delivery_ordinal + """, + analysis_run_id, + ) except asyncpg.UndefinedTableError: return [] labels = await labels_for_codes( @@ -289,6 +290,9 @@ async def _serialize_runs( if not rows: return [] count_rows = await _counts_by_run(conn, [str(row["analysis_run_id"]) for row in rows]) + receipts = await fetch_tepp_accepted_receipts( + conn, [str(row["analysis_run_id"]) for row in rows] + ) labels = await labels_for_codes( conn, [row["run_kind_code"] for row in rows] @@ -334,6 +338,8 @@ async def _serialize_runs( grouping_key = scope_grouping_key(row) if grouping_key: item["scope_grouping_key"] = grouping_key + if run_id in receipts: + item["tepp_accepted_receipt"] = receipts[run_id] payload.append(item) return payload @@ -427,6 +433,49 @@ def reconstructed_edge_is_visible( return parent_visible and child_visible +async def fetch_tepp_accepted_receipt( + conn: asyncpg.Connection, + analysis_run_id: str, +) -> dict[str, Any] | None: + """Transport receipt for one already-visible run, or None. + + Missing table (migration 0171 not applied) is not a 500. The + receipt is not a measurement and never includes a theta. + """ + return (await fetch_tepp_accepted_receipts(conn, [analysis_run_id])).get( + analysis_run_id + ) + + +async def fetch_tepp_accepted_receipts( + conn: asyncpg.Connection, + analysis_run_ids: list[str], +) -> dict[str, dict[str, Any]]: + """Transport receipts for visible runs, isolated from optional-schema errors.""" + if not analysis_run_ids: + return {} + try: + async with conn.transaction(): + rows = await conn.fetch( + """ + select analysis_run_id, remote_run_id, accepted_status_code, received_at + from analysis_run_tepp_accepted_receipt + where analysis_run_id = any($1::uuid[]) + """, + analysis_run_ids, + ) + except asyncpg.UndefinedTableError: + return {} + return { + str(row["analysis_run_id"]): { + "remote_run_id": str(row["remote_run_id"]), + "accepted_status_code": str(row["accepted_status_code"]), + "received_at": _iso(row["received_at"]), + } + for row in rows + } + + async def fetch_reconstructed_edges( conn: asyncpg.Connection, analysis_run_id: str, @@ -439,14 +488,15 @@ async def fetch_reconstructed_edges( Titles follow the same public-or-affiliated rule as ``visible_posts``. """ try: - header = await conn.fetchrow( - """ - select result_sha256 - from analysis_run_reconstruction - where analysis_run_id = $1 - """, - analysis_run_id, - ) + async with conn.transaction(): + header = await conn.fetchrow( + """ + select result_sha256 + from analysis_run_reconstruction + where analysis_run_id = $1 + """, + analysis_run_id, + ) except asyncpg.UndefinedTableError: return None, [] if header is None: diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 5deb8347c..f60b6ce08 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -2,15 +2,17 @@ ADR 0021 reconstructs lineage. ADR 0022 starts TEPP through ``tepp_client`` only. ADR 0023 enqueues that work on a durable outbox -so a crash after Running does not lose the item. Period-report stays -another path. Neither start invents a theta or a calibrated report -score. +so a crash after Running does not lose the item. ADR 0162 persists a +TEPP accepted envelope as transport evidence and keeps the local run +Running. Period-report stays another path. Neither start invents a +theta or a calibrated report score. """ from __future__ import annotations import hashlib import json +from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from urllib.parse import urlparse @@ -20,6 +22,7 @@ from backend.app.analysis_run_ingestion import ( AnalysisRunCreateError, + fetch_tepp_accepted_receipt, fetch_visible_analysis_run, ) from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL @@ -44,12 +47,31 @@ _FAILED = "analysis_status_failed" _TEPP_MODEL_CONTRACT = "tepp-analysis-run-v1" _TEPP_OUTPUT_PROFILE = "calibrated_event_measurement" +_TEPP_TRANSPORT_STATES = frozenset({"accepted", "queued", "running"}) +_TEPP_COMPLETED_STATES = frozenset({"completed", "succeeded"}) +_PERSIST_RESULT = "result" +_PERSIST_RECEIPT = "receipt" class AnalysisRunStartError(AnalysisRunCreateError): """Fail-closed start: HTTP status plus a next-action detail string.""" +@dataclass(frozen=True) +class TeppSubmissionOutcome: + """Classified TEPP envelope: local status, optional persist kind. + + ``persist_kind`` is ``result`` (completed measurement), ``receipt`` + (accepted transport evidence), or empty when nothing may be stored. + A receipt is never a measurement and never invents a theta. + """ + + status_code: str + failure_code: str + envelope: dict[str, Any] | None + persist_kind: str + + def reconstruction_result_digest(edges: list[Edge]) -> str: """SHA-256 of the ordered parent choices. Never hashes a post body.""" material = json.dumps( @@ -172,30 +194,96 @@ def tepp_run_request( ) -def _tepp_submission( +def tepp_envelope_status(envelope: dict[str, Any]) -> str: + """Normalize TEPP ``status`` or ``run_state``. Empty when neither is a string.""" + raw = envelope.get("status") + if not isinstance(raw, str) or not raw.strip(): + raw = envelope.get("run_state") + if isinstance(raw, str): + return raw.strip().casefold() + return "" + + +def tepp_remote_run_id(envelope: dict[str, Any]) -> str: + """Remote run identity from TEPP's published envelope keys.""" + for key in ("analysis_run_id", "run_id", "remote_run_id"): + value = envelope.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def tepp_accepted_status_code(envelope: dict[str, Any]) -> str: + """Transport state when the envelope is accepted/queued/running.""" + status = tepp_envelope_status(envelope) + return status if status in _TEPP_TRANSPORT_STATES else "" + + +def tepp_request_digest(request: AnalysisRunRequest) -> str: + """SHA-256 of the published seven-field request. Never hashes a theta.""" + material = json.dumps(request.to_json(), separators=(",", ":"), sort_keys=True) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def tepp_receipt_digest( + *, + remote_run_id: str, + accepted_status_code: str, + model_contract_version: str, + snapshot_id: str, + knowledge_cutoff: str, +) -> str: + """SHA-256 of persisted receipt columns. Never hashes a result body.""" + material = json.dumps( + { + "accepted_status_code": accepted_status_code, + "knowledge_cutoff": knowledge_cutoff, + "model_contract_version": model_contract_version, + "remote_run_id": remote_run_id, + "snapshot_id": snapshot_id, + }, + separators=(",", ":"), + sort_keys=True, + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def classify_tepp_submission( client: TeppClient, request: AnalysisRunRequest, -) -> tuple[str, str, dict[str, Any] | None]: - """Submit through ``tepp_client`` and require a completed result envelope. +) -> TeppSubmissionOutcome: + """Classify a TEPP envelope without inventing a measurement. - TEPP's target HTTP contract is asynchronous. An ``accepted`` response is - therefore not a measurement and remains ``tepp_result_not_persisted``. - Only a provider-authoritative completed envelope can enter the database. + Accepted/queued/running with a remote run id is transport evidence. + Only a completed envelope with a result dict may persist a + measurement. An empty accepted envelope stays unpersistable. """ try: response = client.submit_analysis_run(request) except TeppNotAvailable: - return _FAILED, "tepp_not_available", None + return TeppSubmissionOutcome(_FAILED, "tepp_not_available", None, "") if not isinstance(response, dict): - return _FAILED, "tepp_result_not_persisted", None - if response.get("status") 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 + return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") + status = tepp_envelope_status(response) + remote_run_id = tepp_remote_run_id(response) + if status in _TEPP_COMPLETED_STATES: + if isinstance(response.get("result"), dict) and remote_run_id: + return TeppSubmissionOutcome(_SUCCEEDED, "", response, _PERSIST_RESULT) + return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") + if status in _TEPP_TRANSPORT_STATES: + if remote_run_id: + return TeppSubmissionOutcome(_RUNNING, "", response, _PERSIST_RECEIPT) + return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") + return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") + + +def _tepp_submission( + client: TeppClient, + request: AnalysisRunRequest, +) -> tuple[str, str, dict[str, Any] | None]: + """Compatibility projection used by older start tests.""" + outcome = classify_tepp_submission(client, request) + return outcome.status_code, outcome.failure_code, outcome.envelope def tepp_submit_outcome( @@ -203,8 +291,8 @@ def tepp_submit_outcome( request: AnalysisRunRequest, ) -> tuple[str, str]: """Compatibility projection of the TEPP submission outcome.""" - status_code, failure_code, _ = _tepp_submission(client, request) - return status_code, failure_code + outcome = classify_tepp_submission(client, request) + return outcome.status_code, outcome.failure_code async def _persist_tepp_result( @@ -214,8 +302,8 @@ async def _persist_tepp_result( envelope: dict[str, Any], ) -> bool: """Persist only a validated, remote-completed TEPP envelope.""" - 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(): + remote_run_id = tepp_remote_run_id(envelope) + if not remote_run_id: return False result_json = json.dumps(envelope, separators=(",", ":"), sort_keys=True) result_sha256 = hashlib.sha256(result_json.encode("utf-8")).hexdigest() @@ -238,6 +326,71 @@ async def _persist_tepp_result( return True +async def _persist_tepp_accepted_receipt( + conn: asyncpg.Connection, + *, + analysis_run_id: str, + envelope: dict[str, Any], + request: AnalysisRunRequest, + knowledge_cutoff: datetime, +) -> bool: + """Persist transport evidence. Never writes a result or a theta.""" + remote_run_id = tepp_remote_run_id(envelope) + accepted_status_code = tepp_accepted_status_code(envelope) + if not remote_run_id or not accepted_status_code: + return False + cutoff = knowledge_cutoff + if cutoff.tzinfo is None: + cutoff = cutoff.replace(tzinfo=timezone.utc) + cutoff_iso = cutoff.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + request_sha256 = tepp_request_digest(request) + receipt_sha256 = tepp_receipt_digest( + remote_run_id=remote_run_id, + accepted_status_code=accepted_status_code, + model_contract_version=request.model_contract_version, + snapshot_id=request.snapshot_id, + knowledge_cutoff=cutoff_iso, + ) + try: + async with conn.transaction(): + existing = await conn.fetchrow( + """ + select remote_run_id, request_sha256 + from analysis_run_tepp_accepted_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 + ) + # Safe SQL: fixed insert text; every external value is bound as $1 through $8. + await conn.execute( # nosemgrep: python.django.security.injection.sql.sql-injection-using-db-cursor-execute.sql-injection-db-cursor-execute, python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli -- code-scanning alert 226 + """ + insert into analysis_run_tepp_accepted_receipt + (analysis_run_id, remote_run_id, request_sha256, receipt_sha256, + accepted_status_code, model_contract_version, snapshot_id, + knowledge_cutoff) + values ($1, $2, $3, $4, $5, $6, $7, $8) + """, + analysis_run_id, + remote_run_id, + request_sha256, + receipt_sha256, + accepted_status_code, + request.model_contract_version, + request.snapshot_id, + cutoff, + ) + except asyncpg.UndefinedTableError: + return False + except (asyncpg.PostgresError, TypeError, ValueError): + return False + return True + + def start_write_conflict_error() -> AnalysisRunStartError: """Next action when a concurrent start already wrote this run.""" return AnalysisRunStartError( @@ -612,7 +765,8 @@ async def deliver_queued_analysis_run( A delivered row replays the stored result. Missing work is 409. TEPP stays Failed when the transport is missing or the envelope is - not persistable. No theta is invented. + not persistable. An accepted receipt keeps the run Running and + leaves the outbox claimed. No theta is invented. """ try: UUID(analysis_run_id) @@ -669,7 +823,7 @@ async def deliver_queued_analysis_run( valkey_stream_entry_id, ) if outbox["work_kind_code"] == _TEPP_KIND: - await _deliver_tepp_measurement( + terminal = await _deliver_tepp_measurement( conn, analysis_run_id=analysis_run_id, locked=outbox, @@ -683,17 +837,19 @@ 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 - await _append_outbox_delivery( - conn, - analysis_run_id, - await _next_outbox_delivery_ordinal(conn, analysis_run_id), - "analysis_outbox_delivered", - finished, - valkey_stream_entry_id, - ) + terminal = True + if terminal: + finished = datetime.now(timezone.utc) + if finished < now: + finished = now + await _append_outbox_delivery( + conn, + analysis_run_id, + await _next_outbox_delivery_ordinal(conn, analysis_run_id), + "analysis_outbox_delivered", + finished, + valkey_stream_entry_id, + ) except asyncpg.UniqueViolationError as exc: raise start_write_conflict_error() from exc return await _visible_or_404( @@ -803,27 +959,51 @@ async def _deliver_tepp_measurement( analysis_run_id: str, locked: asyncpg.Record, tepp_client: TeppClient, -) -> None: - """Submit the frozen snapshot through ``tepp_client``. Never persist a theta.""" +) -> bool: + """Submit the frozen snapshot through ``tepp_client``. Never persist a theta. + + Returns True when a terminal status was appended. An accepted + receipt leaves the run Running and returns False so the outbox + stays claimed until a completed result (TEPP#156) or a typed + failure arrives. + """ 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: + outcome = classify_tepp_submission(tepp_client, request) + if not outcome.persist_kind and await fetch_tepp_accepted_receipt( + conn, analysis_run_id + ) is not None: + return False + status_code = outcome.status_code + failure_code = outcome.failure_code + if outcome.persist_kind == _PERSIST_RESULT and outcome.envelope is not None: if not await _persist_tepp_result( conn, analysis_run_id=analysis_run_id, - envelope=envelope, + envelope=outcome.envelope, ): status_code = _FAILED failure_code = "tepp_result_not_persisted" + elif outcome.persist_kind == _PERSIST_RECEIPT and outcome.envelope is not None: + if await _persist_tepp_accepted_receipt( + conn, + analysis_run_id=analysis_run_id, + envelope=outcome.envelope, + request=request, + knowledge_cutoff=locked["knowledge_cutoff"], + ): + return False + status_code = _FAILED + failure_code = "tepp_receipt_not_persisted" await _append_status( conn, analysis_run_id, await _next_status_ordinal(conn, analysis_run_id), status_code, - failure_code, + failure_code or None, ) + return True diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index ac490b993..35cde9bec 100644 --- a/docker/postgres-init/migrate.sh +++ b/docker/postgres-init/migrate.sh @@ -18,7 +18,7 @@ for migration in /opt/lineageweave/migrations/*.sql; do migration_name=${migration##*/} case "$migration_name" in 0012_*|0013_*|0014_*|0015_*|0016_*|0017_*|0018_*|0019_*|0020_*|0021_*|0022_*|0023_*|0024_*|0025_*|0026_*|0027_*|0028_*|0029_*|0030_*|0031_*|0032_*|0033_*|0034_*|0035_*|0036_*|0037_*|0038_*|0039_*|0040_*|0041_*|0042_*|0043_*|0044_*|0045_*|0046_*|0047_*|0048_*|0049_*|0050_*) ;; - 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*|0130_*|0133_*|0134_*|0136_*|0137_*|0138_*|0139_*|0172_*|0173_*|0176_*|0177_*) ;; + 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*|0130_*|0133_*|0134_*|0136_*|0137_*|0138_*|0139_*|0171_*|0172_*|0173_*|0176_*|0177_*) ;; *) continue ;; esac printf 'Applying %s\n' "$migration_name" diff --git a/docs/adr/0022-authorized-tepp-start.md b/docs/adr/0022-authorized-tepp-start.md index 6e02f54cf..79e89f22e 100644 --- a/docs/adr/0022-authorized-tepp-start.md +++ b/docs/adr/0022-authorized-tepp-start.md @@ -1,6 +1,8 @@ # ADR 0022 — Operators start a pending TEPP measurement through tepp_client **Decision status:** Accepted on this active PR; not protected-main truth until merge +**Amended by:** ADR 0162 (accepted receipts with a remote run id stay +Running as transport evidence; empty accepted envelopes remain Failed) **Date:** 2026-08-17 **Depends on:** ADR 0013 registry; ADR 0017 authorized create; ADR 0021 authorized lineage start diff --git a/docs/adr/0162-tepp-accepted-receipt.md b/docs/adr/0162-tepp-accepted-receipt.md new file mode 100644 index 000000000..4b051ac75 --- /dev/null +++ b/docs/adr/0162-tepp-accepted-receipt.md @@ -0,0 +1,119 @@ +# ADR 0162 — Persist TEPP AnalysisRunAccepted as transport evidence + +**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Date:** 2026-08-23 +**Depends on:** ADR 0022 authorized TEPP start; ADR 0023 analysis-run outbox +**Amends:** ADR 0022 (accepted envelopes are no longer Failed / +`tepp_result_not_persisted` when they carry a remote run id) +**Refs:** Issue #277; blocked completed-result poll remains +[TEPP#156](https://github.com/ContextualWisdomLab/TEPP/issues/156) + +## Context + +TEPP's published start response is `AnalysisRunAccepted`: a durable +submission receipt with a remote `run_id`. It is not a temporal +measurement. ADR 0022 treated any non-completed envelope as Failed / +`tepp_result_not_persisted` so LineageWeave would not stamp Succeeded +from transport evidence. That is honest about measurement authority +and dishonest about operator progress: a normal live TEPP accept +looks like a product failure. + +Issue #277 requires a split lifecycle. This slice covers the +unblocked half: persist the accepted receipt, keep the local run +`Running`, and never invent a theta. Polling TEPP's versioned +completed-result contract stays blocked on TEPP#156. + +## Decision + +Classify a `TeppClient.submit_analysis_run` envelope as follows. On a re-check +with a stored receipt, an outcome carrying neither another persistable receipt +nor a completed result leaves the run Running; only durable evidence may +advance or revoke durable acceptance. A new receipt still must pass the +immutable remote-run and request-digest checks below. + +1. `TeppNotAvailable` with no stored receipt → Failed / + `tepp_not_available`. Seed keeps this default. Do not change seed to + Running. +2. `status` or `run_state` in `{completed, succeeded}` plus a result + object plus a remote run id → Succeeded, persist + `analysis_run_tepp_result` (migration 0027). +3. `status` or `run_state` in `{accepted, queued, running}` plus a + remote run id (`analysis_run_id`, `run_id`, or `remote_run_id`) → + persist `analysis_run_tepp_accepted_receipt`, leave the local run + Running, do not append a terminal status, and do not mark the + outbox delivered. +4. `accepted` / `queued` / `running` without a remote run id → Failed / + `tepp_result_not_persisted` when no receipt was previously stored. Empty + re-check envelopes cannot revoke an existing receipt. +5. Anything else → Failed / `tepp_result_not_persisted` when no receipt was + previously stored. + +The receipt table stores transport evidence only: remote run id, +request digest, receipt digest, accepted status code, model contract +version, snapshot id, knowledge cutoff, and received time. It does +not store a result JSON, a theta, or a calibrated score. A changed +remote run id or request digest for the same local run fails closed. +Accepted, queued, and running transport-state progression for the +same remote run and request remains idempotent; the first receipt stays +the durable acceptance evidence. + +`GET /api/analysis-runs` and `GET /api/analysis-runs/{id}` may attach +`tepp_accepted_receipt` +`{remote_run_id, accepted_status_code, received_at}` so the operator +can see that TEPP accepted the work. Missing migration 0171 is an +empty attachment, not a 500; list rows load receipts in one bounded +query rather than one query per run. Global Ask must not promote this +receipt into an answer claim. + +Completed-result polling, bounded backoff, and request-binding +revalidation remain TEPP#156. This ADR does not invent a local +psychometric substitute while waiting. + +```mermaid +sequenceDiagram + participant Operator + participant API + participant TeppClient + participant Registry + Operator->>API: POST /api/analysis-runs/{id}/start + API->>Registry: Running + outbox + API->>TeppClient: AnalysisRunRequest v1 + alt TeppNotAvailable without a stored receipt + Registry->>Registry: Failed tepp_not_available + Registry->>Registry: outbox delivered + else unavailable or unpersistable re-check after a stored receipt + Note over Registry: stay Running, outbox stays claimed + else accepted with remote run id + Registry->>Registry: persist accepted receipt + Note over Registry: stay Running, outbox stays claimed + API-->>Operator: 200 Running + receipt + else completed with result + Registry->>Registry: persist tepp result + Succeeded + Registry->>Registry: outbox delivered + else accepted without remote run id + Registry->>Registry: Failed tepp_result_not_persisted + Registry->>Registry: outbox delivered + end +``` + +## Consequences + +A connected TEPP transport that returns `accepted` plus a remote run +id no longer looks like a product failure. The operator refreshes a +Running run. Refresh may replay the same idempotent submit; an unavailable or +unpersistable response after that durable receipt keeps the run Running and its +outbox claimed, and it still must not stamp Succeeded from the receipt. Seed +and an initial empty `accepted` envelope stay Failed. +Do not invent a theta. + +## References — APA 7th + +Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns: +Designing, building, and deploying messaging solutions*. Addison-Wesley. + +International Organization for Standardization. (2019). *ISO 8601-1:2019: +Date and time—Representations for information interchange—Part 1: Basic +rules* (confirmed 2024; Amendment 1:2022). + +Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*. +World Wide Web Consortium. https://www.w3.org/TR/prov-dm/ diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index a28848838..f532edd79 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -3,6 +3,7 @@ import userEvent from "@testing-library/user-event"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import App from "./App"; import { setLocale } from "./i18n"; +import { OIDC_RETURN_URL_STORAGE_KEY } from "./oidcReturnUrl"; const signinRedirect = vi.fn(); const signoutRedirect = vi.fn(); @@ -48,6 +49,7 @@ describe("App, unauthenticated", () => { it("shows a login button that starts the real OIDC redirect", async () => { window.sessionStorage.clear(); render(); + expect(screen.queryByRole("heading", { name: /admin settings/i })).toBeNull(); const button = screen.getByRole("button", { name: /log in/i }); await userEvent.click(button); expect(signinRedirect).toHaveBeenCalledTimes(1); @@ -56,7 +58,8 @@ describe("App, unauthenticated", () => { state: expect.objectContaining({ returnUrl: expect.stringMatching(/^\//) }), }), ); - expect(window.sessionStorage.getItem("lineageweave.oidc.returnUrl")).toBe("/"); + expect(window.sessionStorage.getItem(OIDC_RETURN_URL_STORAGE_KEY)).toMatch(/^\//); + expect(window.localStorage.getItem(OIDC_RETURN_URL_STORAGE_KEY)).toMatch(/^\//); }); it("remembers a same-origin post deep link before the OIDC redirect", async () => { @@ -174,6 +177,7 @@ describe("App, authenticated", () => { succeededReportRun?: boolean; succeededTeppRun?: boolean; pendingTeppRun?: boolean; + acceptedTeppRun?: boolean; pluralAffiliations?: boolean; manyAffiliations?: boolean; noAffiliations?: boolean; @@ -301,6 +305,28 @@ describe("App, authenticated", () => { }, ]; + const teppStatus = options?.acceptedTeppRun + ? "analysis_status_running" + : options?.succeededTeppRun + ? "analysis_status_succeeded" + : options?.pendingTeppRun + ? "analysis_status_pending" + : "analysis_status_failed"; + const teppLabel = options?.acceptedTeppRun + ? "Running" + : options?.succeededTeppRun + ? "Succeeded" + : options?.pendingTeppRun + ? "Pending" + : "Failed"; + const teppAcceptedReceipt = options?.acceptedTeppRun + ? { + remote_run_id: "tepp-run-accepted-1", + accepted_status_code: "accepted" as const, + received_at: "2026-01-12T12:36:30Z", + } + : undefined; + let releaseMe = () => {}; let releasePosts = () => {}; const meReady = options?.deferMe @@ -552,16 +578,6 @@ describe("App, authenticated", () => { ); } if (url.endsWith("/api/analysis-runs/run-demo-tepp")) { - const teppStatus = options?.succeededTeppRun - ? "analysis_status_succeeded" - : options?.pendingTeppRun - ? "analysis_status_pending" - : "analysis_status_failed"; - const teppLabel = options?.succeededTeppRun - ? "Succeeded" - : options?.pendingTeppRun - ? "Pending" - : "Failed"; return Promise.resolve( jsonResponse({ analysis_run_id: "run-demo-tepp", @@ -582,6 +598,7 @@ describe("App, authenticated", () => { }, ], visible_posts: [{ post_id: "post-1", post_title: "Public post" }], + ...(teppAcceptedReceipt ? { tepp_accepted_receipt: teppAcceptedReceipt } : {}), status_history: options?.pendingTeppRun ? [ { @@ -591,31 +608,46 @@ describe("App, authenticated", () => { occurred_at: "2026-01-12T12:35:00Z", }, ] - : [ - { - status_ordinal: 1, - status_code: "analysis_status_pending", - status_label: "Pending", - occurred_at: "2026-01-12T12:35:00Z", - }, - { - status_ordinal: 2, - status_code: "analysis_status_running", - status_label: "Running", - occurred_at: "2026-01-12T12:36:00Z", - }, - { - status_ordinal: 3, - status_code: options?.succeededTeppRun - ? "analysis_status_succeeded" - : "analysis_status_failed", - status_label: options?.succeededTeppRun ? "Succeeded" : "Failed", - occurred_at: "2026-01-12T12:37:00Z", - ...(options?.succeededTeppRun - ? {} - : { failure_code: "tepp_not_available" }), - }, - ], + : options?.acceptedTeppRun + ? [ + { + status_ordinal: 1, + status_code: "analysis_status_pending", + status_label: "Pending", + occurred_at: "2026-01-12T12:35:00Z", + }, + { + status_ordinal: 2, + status_code: "analysis_status_running", + status_label: "Running", + occurred_at: "2026-01-12T12:36:00Z", + }, + ] + : [ + { + status_ordinal: 1, + status_code: "analysis_status_pending", + status_label: "Pending", + occurred_at: "2026-01-12T12:35:00Z", + }, + { + status_ordinal: 2, + status_code: "analysis_status_running", + status_label: "Running", + occurred_at: "2026-01-12T12:36:00Z", + }, + { + status_ordinal: 3, + status_code: options?.succeededTeppRun + ? "analysis_status_succeeded" + : "analysis_status_failed", + status_label: options?.succeededTeppRun ? "Succeeded" : "Failed", + occurred_at: "2026-01-12T12:37:00Z", + ...(options?.succeededTeppRun + ? {} + : { failure_code: "tepp_not_available" }), + }, + ], }), ); } @@ -926,16 +958,8 @@ describe("App, authenticated", () => { scope_kind_code: "analysis_scope_corporate_entity", scope_kind_label: "Corporate entity", scope_entity_name: "Demo Corp", - status_code: options?.succeededTeppRun - ? "analysis_status_succeeded" - : options?.pendingTeppRun - ? "analysis_status_pending" - : "analysis_status_failed", - status_label: options?.succeededTeppRun - ? "Succeeded" - : options?.pendingTeppRun - ? "Pending" - : "Failed", + status_code: teppStatus, + status_label: teppLabel, knowledge_cutoff: "2026-01-12T12:00:00Z", requested_at: "2026-01-12T12:34:00Z", source_counts: [ @@ -945,6 +969,7 @@ describe("App, authenticated", () => { count_value: 3, }, ], + ...(teppAcceptedReceipt ? { tepp_accepted_receipt: teppAcceptedReceipt } : {}), }, { analysis_run_id: "run-demo-report", @@ -5097,6 +5122,40 @@ describe("App, authenticated", () => { expect(screen.queryByText(/replace Failed/i)).not.toBeInTheDocument(); }); + it("keeps an accepted TEPP receipt Running and does not invent a theta", async () => { + stubBackend({ acceptedTeppRun: true }); + render(); + + const listButton = await screen.findByRole("button", { + name: "Open analysis run: TEPP measurement · Running · Demo Corp", + }); + expect(listButton).toHaveTextContent( + "TEPP accepted this measurement. Check its status to retrieve the completed result. This receipt is not a calibrated score.", + ); + expect(listButton).not.toHaveTextContent("theta"); + expect(listButton).not.toHaveTextContent("calibrated result"); + + await userEvent.click(listButton); + expect( + await screen.findByRole("heading", { name: "TEPP measurement · Running · Demo Corp" }), + ).toBeInTheDocument(); + expect( + screen.getAllByText( + "TEPP accepted this measurement. Check its status to retrieve the completed result. This receipt is not a calibrated score.", + ).length, + ).toBeGreaterThan(0); + expect( + screen.getByRole("button", { name: "Check TEPP measurement status" }), + ).toBeInTheDocument(); + const history = screen.getByRole("list", { name: "Analysis run status history" }); + expect(history).toHaveTextContent("Running 2026-01-12 12:36"); + expect(history).not.toHaveTextContent("Succeeded"); + expect(history).not.toHaveTextContent("Failed"); + expect(history).not.toHaveTextContent("tepp_not_available"); + expect(screen.queryByText(/theta/i)).not.toBeInTheDocument(); + expect(screen.queryByText(/tepp-run-accepted-1/)).not.toBeInTheDocument(); + }); + it("records a pending lineage run and opens the authorized detail", async () => { const fetchMock = stubBackend(); render(); diff --git a/frontend/src/analysisRunGuidance.test.ts b/frontend/src/analysisRunGuidance.test.ts index bd405c9d0..8ebca6656 100644 --- a/frontend/src/analysisRunGuidance.test.ts +++ b/frontend/src/analysisRunGuidance.test.ts @@ -177,6 +177,31 @@ describe("analysisRunGuidance", () => { expect(analysisRunCanRefresh(runningTepp)).toBe(true); }); + it("names an accepted TEPP receipt as transport evidence, not a calibrated score", () => { + const runningTepp = run({ + run_kind_code: "analysis_run_tepp", + status_code: "analysis_status_running", + }); + const withReceipt = { + ...runningTepp, + tepp_accepted_receipt: { + remote_run_id: "tepp-run-accepted-1", + accepted_status_code: "accepted" as const, + received_at: "2026-01-12T12:36:00Z", + }, + }; + const copy = analysisRunNextAction(withReceipt) ?? ""; + expect(copy).toBe( + "TEPP accepted this measurement. Check its status to retrieve the completed result. This receipt is not a calibrated score.", + ); + expect(copy.toLowerCase()).not.toMatch(/theta/); + expect(analysisRunStartLabel(withReceipt)).toBe("Check TEPP measurement status"); + expect(analysisRunNextAction(runningTepp)).toMatch(/already queued/i); + expect(analysisRunCanStart(withReceipt)).toBe(true); + expect(analysisRunCanRefresh(withReceipt)).toBe(false); + expect(analysisRunCanStart(runningTepp)).toBe(false); + }); + it("keeps succeeded report copy free of unbuilt/rebuild/reconstruct/measure language", () => { const succeeded = run({ run_kind_code: "analysis_run_report", diff --git a/frontend/src/analysisRunGuidance.ts b/frontend/src/analysisRunGuidance.ts index 6ce21434e..89697c64f 100644 --- a/frontend/src/analysisRunGuidance.ts +++ b/frontend/src/analysisRunGuidance.ts @@ -48,6 +48,9 @@ export function analysisRunNextAction(run: AnalysisRun): string | null { case "analysis_run_lineage": return "Refresh this run. Reconstruction is already queued on the durable outbox."; case "analysis_run_tepp": + if (run.tepp_accepted_receipt) { + return "TEPP accepted this measurement. Check its status to retrieve the completed result. This receipt is not a calibrated score."; + } return "Refresh this run. Measurement is already queued on the durable outbox."; case "analysis_run_report": return "Refresh this run. The period report is already queued on the durable outbox."; @@ -114,8 +117,15 @@ export function analysisRunCorpusHint(run: AnalysisRun): string | null { } } -/** Start is only for a Pending lineage or TEPP row. Running work is already queued. */ +/** Start is for Pending lineage/TEPP, or a Running TEPP receipt re-check. */ export function analysisRunCanStart(run: AnalysisRun): boolean { + if ( + run.run_kind_code === "analysis_run_tepp" && + run.status_code === "analysis_status_running" && + run.tepp_accepted_receipt + ) { + return true; + } return ( (run.run_kind_code === "analysis_run_lineage" || run.run_kind_code === "analysis_run_tepp") && run.status_code === "analysis_status_pending" @@ -123,13 +133,17 @@ export function analysisRunCanStart(run: AnalysisRun): boolean { } export function analysisRunCanRefresh(run: AnalysisRun): boolean { + if (run.run_kind_code === "analysis_run_tepp" && run.tepp_accepted_receipt) { + return false; + } return run.status_code === "analysis_status_running"; } export function analysisRunStartLabel(run: AnalysisRun): string { - return run.run_kind_code === "analysis_run_tepp" - ? "Start TEPP measurement" - : "Start reconstruction"; + if (run.run_kind_code !== "analysis_run_tepp") return "Start reconstruction"; + return run.status_code === "analysis_status_running" + ? "Check TEPP measurement status" + : "Start TEPP measurement"; } export function analysisRunRefreshLabel(): string { diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 53e840bc9..b33b3c8bd 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1275,6 +1275,13 @@ export interface AnalysisRunReconstructedEdge { fused_score: number; } +/** Transport receipt from TEPP AnalysisRunAccepted. Not a measurement. */ +export interface AnalysisRunTeppAcceptedReceipt { + remote_run_id: string; + accepted_status_code: "accepted" | "queued" | "running"; + received_at: string; +} + export interface AnalysisRunVisiblePost { post_id: string; post_title: string; @@ -1302,6 +1309,7 @@ export interface AnalysisRun { visible_posts?: AnalysisRunVisiblePost[]; reconstructed_edges?: AnalysisRunReconstructedEdge[]; reconstruction_result_sha256?: string; + tepp_accepted_receipt?: AnalysisRunTeppAcceptedReceipt; code_revision_sha?: string; configuration_sha256?: string; } diff --git a/frontend/src/components/AnalysisRunNextAction.tsx b/frontend/src/components/AnalysisRunNextAction.tsx index e21f8c05d..d5775ded8 100644 --- a/frontend/src/components/AnalysisRunNextAction.tsx +++ b/frontend/src/components/AnalysisRunNextAction.tsx @@ -43,7 +43,9 @@ export function AnalysisRunNextAction({ > {starting ? run.run_kind_code === "analysis_run_tepp" - ? "Submitting the TEPP request..." + ? run.status_code === "analysis_status_running" + ? "Checking TEPP measurement status..." + : "Submitting the TEPP request..." : "Reconstructing the cutoff bag..." : startLabel} diff --git a/migrations/0171_analysis_run_tepp_accepted_receipt.sql b/migrations/0171_analysis_run_tepp_accepted_receipt.sql new file mode 100644 index 000000000..1d5fe35ff --- /dev/null +++ b/migrations/0171_analysis_run_tepp_accepted_receipt.sql @@ -0,0 +1,19 @@ +-- Persist TEPP AnalysisRunAccepted as transport evidence (ADR 0162). +-- This is not a measurement: no result JSON and no Succeeded stamp. +-- Completed envelopes stay on analysis_run_tepp_result (migration 0027). +create table if not exists analysis_run_tepp_accepted_receipt ( + analysis_run_id uuid primary key + references analysis_run(analysis_run_id) on delete cascade, + remote_run_id text not null unique check (btrim(remote_run_id) <> ''), + request_sha256 text not null check (request_sha256 ~ '^[0-9a-f]{64}$'), + receipt_sha256 text not null check (receipt_sha256 ~ '^[0-9a-f]{64}$'), + accepted_status_code text not null + check (accepted_status_code in ('accepted', 'queued', 'running')), + model_contract_version text not null check (btrim(model_contract_version) <> ''), + snapshot_id text not null check (btrim(snapshot_id) <> ''), + knowledge_cutoff timestamptz not null, + received_at timestamptz not null default clock_timestamp() +); + +create index if not exists analysis_run_tepp_accepted_receipt_received_idx + on analysis_run_tepp_accepted_receipt (received_at); diff --git a/migrations/rollback/0171_analysis_run_tepp_accepted_receipt.sql b/migrations/rollback/0171_analysis_run_tepp_accepted_receipt.sql new file mode 100644 index 000000000..a774e2c18 --- /dev/null +++ b/migrations/rollback/0171_analysis_run_tepp_accepted_receipt.sql @@ -0,0 +1,25 @@ +-- Fail-closed rollback for migration 0171. +-- +-- Accepted TEPP receipts are transport evidence. Export or explicitly +-- delete them under an approved retention procedure before dropping. + +begin; + +do $$ +declare + relation_has_rows boolean; +begin + if to_regclass('public.analysis_run_tepp_accepted_receipt') is not null then + execute 'select exists (select 1 from analysis_run_tepp_accepted_receipt)' + into relation_has_rows; + if relation_has_rows then + raise exception 'analysis_run_tepp_accepted_receipt_not_empty'; + end if; + end if; +end +$$; + +drop index if exists analysis_run_tepp_accepted_receipt_received_idx; +drop table if exists analysis_run_tepp_accepted_receipt; + +commit; diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index dc0274dcd..ecd0b107b 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -12,6 +12,7 @@ from backend.app.analysis_run_start import ( AnalysisRunStartError, _append_status, + classify_tepp_submission, configured_tepp_client, reconstruction_member_ids, reconstruction_result_digest, @@ -139,7 +140,7 @@ def test_tepp_submit_outcome_drops_a_missing_transport() -> None: def test_tepp_submit_outcome_does_not_persist_an_empty_envelope() -> None: - """An accepted envelope is not a persistable measurement.""" + """An accepted envelope without a remote run id is not persistable.""" class _Accepting(TeppClient): def __init__(self) -> None: @@ -150,6 +151,26 @@ def __init__(self) -> None: assert failure == "tepp_result_not_persisted" +def test_tepp_submit_outcome_keeps_accepted_receipt_running() -> None: + """A valid accepted envelope is transport evidence, not a measurement.""" + + class _Accepting(TeppClient): + def __init__(self) -> None: + super().__init__( + transport=lambda _payload: { + "status": "accepted", + "run_id": "tepp-run-accepted-1", + } + ) + + status, failure = tepp_submit_outcome(_Accepting(), _tepp_request()) + assert status == "analysis_status_running" + assert failure == "" + outcome = classify_tepp_submission(_Accepting(), _tepp_request()) + assert outcome.persist_kind == "receipt" + assert outcome.persist_kind != "result" + + def test_configured_tepp_client_stays_unavailable_without_http() -> None: """Empty or non-http URLs keep the default dropped channel.""" assert isinstance(configured_tepp_client(""), TeppClient) diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 83dbc04c8..8c96ef044 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -121,6 +121,18 @@ def test_migrate_sh_replays_tenant_identity_metadata_migration_on_existing_volum assert "tenant_settings_copyright_year_range_check" in migration +def test_migrate_sh_replays_tepp_accepted_receipt_on_existing_volumes() -> None: + """Existing volumes must receive the TEPP accepted-receipt table.""" + root = Path(__file__).resolve().parents[1] + script = (root / "docker" / "postgres-init" / "migrate.sh").read_text(encoding="utf-8") + migration = (root / "migrations" / "0171_analysis_run_tepp_accepted_receipt.sql").read_text( + encoding="utf-8" + ) + + assert "0171_*" in script + assert "create table if not exists analysis_run_tepp_accepted_receipt" in migration + + def test_migrate_sh_replays_analysis_run_status_same_clock_on_existing_volumes() -> None: """Existing volumes must replace the analysis-run status trigger from one clock.""" root = Path(__file__).resolve().parents[1] diff --git a/tests/test_static_sql_review_contracts.py b/tests/test_static_sql_review_contracts.py index e9d16db42..6e4c2ccd0 100644 --- a/tests/test_static_sql_review_contracts.py +++ b/tests/test_static_sql_review_contracts.py @@ -27,7 +27,7 @@ ) ASYNC_STATEMENT_METHODS = {"execute", "fetch", "fetchrow", "fetchval"} SQL_REVIEW_RULE = "python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli" -EXPECTED_SQL_SUPPRESSION_COUNT = 38 +EXPECTED_SQL_SUPPRESSION_COUNT = 39 @pytest.mark.parametrize("relative_path", SQL_REVIEW_PATHS) diff --git a/tests/test_tepp_accepted_receipt.py b/tests/test_tepp_accepted_receipt.py new file mode 100644 index 000000000..b1605f678 --- /dev/null +++ b/tests/test_tepp_accepted_receipt.py @@ -0,0 +1,427 @@ +"""TEPP accepted receipts are transport evidence, never a measurement.""" + +import asyncio +from datetime import datetime, timezone +from pathlib import Path + +import asyncpg +import pytest + +import backend.app.analysis_run_start as analysis_run_start_module +from backend.app.analysis_run_ingestion import ( + fetch_outbox_deliveries, + fetch_reconstructed_edges, + fetch_tepp_accepted_receipts, +) +from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable + +_ROOT = Path(__file__).resolve().parents[1] + + +class _Transaction: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback): + return False + + +class _ReceiptConnection: + def __init__(self, *, rows=None, row=None, error=None) -> None: + self.rows = [] if rows is None else rows + self.row = row + self.error = error + self.transactions = 0 + self.executions: list[tuple[object, ...]] = [] + + def transaction(self): + self.transactions += 1 + return _Transaction() + + async def fetch(self, query, analysis_run_ids): + if self.error is not None: + raise self.error + return self.rows + + async def fetchrow(self, query, analysis_run_id): + if self.error is not None: + raise self.error + return self.row + + async def execute(self, *args): + self.executions.append(args) + if self.error is not None: + raise self.error + + async def fetchval(self, query, *args): + if "select delivery_status_code" in query: + return "analysis_outbox_claimed" + if "max(status_ordinal)" in query or "max(delivery_ordinal)" in query: + return 1 + raise AssertionError(f"Unexpected fetchval query: {query}") + + +def _request() -> AnalysisRunRequest: + return analysis_run_start_module.tepp_run_request( + idempotency_key="buyer-tepp-2026-w07", + snapshot_sha256="ab" * 32, + knowledge_cutoff=datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc), + corporate_entity_id="11111111-1111-1111-1111-111111111111", + ) + + +class _EnvelopeClient(TeppClient): + def __init__(self, envelope: object) -> None: + super().__init__(transport=lambda _payload: envelope) # type: ignore[arg-type] + + +def test_missing_transport_stays_failed_not_available() -> None: + outcome = analysis_run_start_module.classify_tepp_submission(TeppClient(), _request()) + assert outcome.status_code == "analysis_status_failed" + assert outcome.failure_code == "tepp_not_available" + assert outcome.persist_kind == "" + assert outcome.envelope is None + + +@pytest.mark.parametrize( + "tepp_client", + [TeppClient(), _EnvelopeClient({"status": "accepted"})], + ids=["transport-unavailable", "empty-accepted-envelope"], +) +def test_unpersistable_recheck_keeps_an_already_accepted_run_claimed( + monkeypatch, tepp_client, +) -> None: + analysis_run_id = "11111111-1111-1111-1111-111111111111" + visible = { + "analysis_run_id": analysis_run_id, + "status_code": "analysis_status_running", + } + + async def fetch_visible(*_args, **_kwargs): + return visible + + monkeypatch.setattr( + analysis_run_start_module, + "fetch_visible_analysis_run", + fetch_visible, + ) + connection = _ReceiptConnection( + row={ + "analysis_run_id": analysis_run_id, + "work_kind_code": "analysis_run_tepp", + "knowledge_cutoff": datetime(2026, 1, 12, tzinfo=timezone.utc), + "idempotency_key": "buyer-tepp-2026-w07", + "analysis_source_snapshot_id": "snapshot-1", + "snapshot_sha256": "ab" * 32, + "corporate_entity_id": "22222222-2222-2222-2222-222222222222", + }, + rows=[ + { + "analysis_run_id": analysis_run_id, + "remote_run_id": "remote-run-1", + "accepted_status_code": "accepted", + "received_at": datetime(2026, 1, 12, tzinfo=timezone.utc), + } + ], + ) + + result = asyncio.run( + analysis_run_start_module.deliver_queued_analysis_run( + connection, + analysis_run_id=analysis_run_id, + account_id="account-1", + affiliated_entity_ids=[], + tepp_client=tepp_client, + ) + ) + + assert result["status_code"] == "analysis_status_running" + assert not any( + "insert into analysis_run_status_event" in str(execution[0]) + for execution in connection.executions + ) + assert not any( + "analysis_outbox_delivered" in execution + for execution in connection.executions + ) + + +def test_initial_unavailability_without_a_receipt_remains_terminal() -> None: + connection = _ReceiptConnection(rows=[]) + + terminal = asyncio.run( + analysis_run_start_module._deliver_tepp_measurement( + connection, + analysis_run_id="11111111-1111-1111-1111-111111111111", + locked={ + "knowledge_cutoff": datetime(2026, 1, 12, tzinfo=timezone.utc), + "idempotency_key": "buyer-tepp-2026-w07", + "snapshot_sha256": "ab" * 32, + "corporate_entity_id": "22222222-2222-2222-2222-222222222222", + }, + tepp_client=TeppClient(), + ) + ) + + assert terminal is True + assert any( + execution[3:] == ("analysis_status_failed", "tepp_not_available") + for execution in connection.executions + ) + + +def test_empty_accepted_envelope_is_not_a_receipt() -> None: + outcome = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient({"status": "accepted"}), _request() + ) + assert outcome.status_code == "analysis_status_failed" + assert outcome.failure_code == "tepp_result_not_persisted" + assert outcome.persist_kind == "" + status, failure = analysis_run_start_module.tepp_submit_outcome( + _EnvelopeClient({"status": "accepted"}), _request() + ) + assert status == "analysis_status_failed" + assert failure == "tepp_result_not_persisted" + + +def test_accepted_with_remote_run_id_is_running_receipt() -> None: + envelope = {"status": "accepted", "run_id": "tepp-run-accepted-1"} + outcome = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient(envelope), _request() + ) + assert outcome.status_code == "analysis_status_running" + assert outcome.failure_code == "" + assert outcome.persist_kind == "receipt" + assert outcome.envelope == envelope + assert "theta" not in str(outcome.envelope).casefold() + + +def test_queued_and_running_envelopes_are_receipts_not_results() -> None: + queued = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient({"run_state": "queued", "analysis_run_id": "tepp-queued-1"}), + _request(), + ) + running = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient({"status": "running", "remote_run_id": "tepp-running-1"}), + _request(), + ) + assert queued.persist_kind == "receipt" + assert running.persist_kind == "receipt" + assert queued.status_code == "analysis_status_running" + assert running.status_code == "analysis_status_running" + assert queued.persist_kind != "result" + assert running.persist_kind != "result" + + +def test_completed_result_is_the_only_measurement_persist() -> None: + outcome = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient( + { + "status": "completed", + "analysis_run_id": "tepp-completed-1", + "result": {"schema_version": "tepp-result-v1", "event_count": 3}, + } + ), + _request(), + ) + assert outcome.status_code == "analysis_status_succeeded" + assert outcome.persist_kind == "result" + assert outcome.failure_code == "" + + +def test_completed_remote_run_id_alias_persists_the_result() -> None: + connection = _ReceiptConnection() + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_result( + connection, + analysis_run_id="local-run", + envelope={ + "status": "completed", + "remote_run_id": "remote-run", + "result": {"schema_version": "tepp-result-v1"}, + }, + ) + ) + assert persisted is True + assert connection.executions[0][2] == "remote-run" + + +def test_receipt_insert_error_is_rolled_back_by_a_savepoint() -> None: + connection = _ReceiptConnection(error=asyncpg.UniqueViolationError("duplicate")) + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_accepted_receipt( + connection, + analysis_run_id="local-run", + envelope={"status": "accepted", "run_id": "remote-run"}, + request=_request(), + knowledge_cutoff=datetime(2026, 1, 12, tzinfo=timezone.utc), + ) + ) + assert persisted is False + assert connection.transactions == 1 + + +def test_receipt_transport_progression_keeps_the_same_run_running() -> None: + request = _request() + connection = _ReceiptConnection( + row={ + "remote_run_id": "remote-run", + "request_sha256": analysis_run_start_module.tepp_request_digest(request), + } + ) + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_accepted_receipt( + connection, + analysis_run_id="local-run", + envelope={"status": "running", "run_id": "remote-run"}, + request=request, + knowledge_cutoff=datetime(2026, 1, 12, tzinfo=timezone.utc), + ) + ) + assert persisted is True + assert connection.executions == [] + + +def test_receipt_replay_fails_closed_when_remote_run_id_changes() -> None: + request = _request() + connection = _ReceiptConnection( + row={ + "remote_run_id": "different-remote-run", + "request_sha256": analysis_run_start_module.tepp_request_digest(request), + } + ) + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_accepted_receipt( + connection, + analysis_run_id="local-run", + envelope={"status": "running", "run_id": "remote-run"}, + request=request, + knowledge_cutoff=datetime.fromisoformat(request.knowledge_cutoff), + ) + ) + assert persisted is False + assert connection.executions == [] + + +def test_receipt_replay_fails_closed_when_request_digest_changes() -> None: + request = _request() + connection = _ReceiptConnection( + row={ + "remote_run_id": "remote-run", + "request_sha256": "cd" * 32, + } + ) + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_accepted_receipt( + connection, + analysis_run_id="local-run", + envelope={"status": "running", "run_id": "remote-run"}, + request=request, + knowledge_cutoff=datetime.fromisoformat(request.knowledge_cutoff), + ) + ) + assert persisted is False + assert connection.executions == [] + + +def test_missing_receipt_table_isolated_from_the_callers_transaction() -> None: + connection = _ReceiptConnection(error=asyncpg.UndefinedTableError("missing")) + receipts = asyncio.run(fetch_tepp_accepted_receipts(connection, ["local-run"])) + assert receipts == {} + assert connection.transactions == 1 + + +def test_legacy_optional_reads_isolate_missing_tables() -> None: + outbox = _ReceiptConnection(error=asyncpg.UndefinedTableError("missing")) + reconstruction = _ReceiptConnection(error=asyncpg.UndefinedTableError("missing")) + + assert asyncio.run(fetch_outbox_deliveries(outbox, "local-run")) == [] + assert asyncio.run(fetch_reconstructed_edges(reconstruction, "local-run", [])) == ( + None, + [], + ) + assert outbox.transactions == reconstruction.transactions == 1 + + +def test_completed_without_result_is_not_a_measurement() -> None: + outcome = analysis_run_start_module.classify_tepp_submission( + _EnvelopeClient({"status": "succeeded", "run_id": "tepp-empty-1"}), + _request(), + ) + assert outcome.status_code == "analysis_status_failed" + assert outcome.failure_code == "tepp_result_not_persisted" + assert outcome.persist_kind == "" + + +def test_receipt_digest_is_stable_and_omits_result_bodies() -> None: + request = _request() + first = analysis_run_start_module.tepp_receipt_digest( + remote_run_id="tepp-run-accepted-1", + accepted_status_code="accepted", + model_contract_version=request.model_contract_version, + snapshot_id=request.snapshot_id, + knowledge_cutoff=request.knowledge_cutoff, + ) + second = analysis_run_start_module.tepp_receipt_digest( + remote_run_id="tepp-run-accepted-1", + accepted_status_code="accepted", + model_contract_version=request.model_contract_version, + snapshot_id=request.snapshot_id, + knowledge_cutoff=request.knowledge_cutoff, + ) + assert first == second + assert len(first) == 64 + assert analysis_run_start_module.tepp_request_digest(request) != first + assert "theta" not in analysis_run_start_module.tepp_request_digest(request) + + +def test_unavailable_transport_still_raises_on_direct_submit() -> None: + with pytest.raises(TeppNotAvailable): + TeppClient().submit_analysis_run(_request()) + + +def test_migration_0171_is_transport_evidence_not_a_result_table() -> None: + sql = (_ROOT / "migrations" / "0171_analysis_run_tepp_accepted_receipt.sql").read_text( + encoding="utf-8" + ) + rollback = ( + _ROOT / "migrations" / "rollback" / "0171_analysis_run_tepp_accepted_receipt.sql" + ).read_text(encoding="utf-8") + assert "analysis_run_tepp_accepted_receipt" in sql + assert "create table if not exists analysis_run_tepp_accepted_receipt" in sql + assert "accepted_status_code in ('accepted', 'queued', 'running')" in sql + assert "remote_run_id text not null unique" in sql + assert "request_sha256" in sql + assert "receipt_sha256" in sql + create_body = sql.split("create table", 1)[1] + assert "jsonb" not in create_body.casefold() + assert "theta" not in create_body.casefold() + assert "result_json" not in create_body + assert "drop table if exists analysis_run_tepp_accepted_receipt" in rollback + table_names = [ + name + for name in ("analysis_run_tepp_accepted_receipt",) + if name in sql + ] + assert all(len(name.split("_")) >= 2 for name in table_names) + + +def test_migrate_sh_replays_accepted_receipt_migration() -> None: + script = (_ROOT / "docker" / "postgres-init" / "migrate.sh").read_text(encoding="utf-8") + tenant_settings = (_ROOT / "migrations" / "0103_tenant_settings.sql").read_text( + encoding="utf-8" + ) + assert "0103_*" in script + assert "0171_*" in script + assert "CREATE TABLE IF NOT EXISTS tenant_settings" in tenant_settings + assert "ON CONFLICT (id) DO NOTHING" in tenant_settings + + +def test_adr_0162_keeps_measurement_authority_with_tepp() -> None: + adr = (_ROOT / "docs" / "adr" / "0162-tepp-accepted-receipt.md").read_text(encoding="utf-8") + assert "transport evidence" in adr.casefold() + assert "do not invent a theta" in adr.casefold() + assert "TEPP#156" in adr + assert "analysis_run_tepp_accepted_receipt" in adr + assert "stay Running" in adr or "leave the local run" in adr + assert "GET /api/analysis-runs` and `GET /api/analysis-runs/{id}`" in adr