+ Measurement request accepted. Refresh this run to check whether results are ready. +
+ ); +} diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index 7dbfd886f..0dbc51c33 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -18,7 +18,11 @@ from __future__ import annotations +import json +import re +import unicodedata from dataclasses import dataclass +from datetime import datetime from typing import Any, Callable @@ -26,6 +30,10 @@ class TeppNotAvailable(RuntimeError): """Raised by the default transport: TEPP has no live REST API yet.""" +class TeppInvalidResponse(ValueError): + """Raised when TEPP returns a status payload outside its v1 contract.""" + + def _no_transport(request: dict[str, Any]) -> dict[str, Any]: """Implement the _no_transport operation for this channel.""" raise TeppNotAvailable( @@ -35,6 +43,11 @@ def _no_transport(request: dict[str, Any]) -> dict[str, Any]: ) +def _no_status_transport(remote_run_id: str) -> dict[str, Any]: + """Refuse status reads until a provider-owned read transport is supplied.""" + raise TeppNotAvailable(f"TEPP status transport unavailable for {remote_run_id!r}") + + @dataclass(frozen=True) class AnalysisRunRequest: """Mirrors TEPP's ``schemas/analysis_run_request_v1.json`` exactly. @@ -75,9 +88,177 @@ class exists so the rest of LineageWeave can be written against a touching any other module. """ - def __init__(self, transport: Callable[[dict[str, Any]], dict[str, Any]] = _no_transport) -> None: + def __init__( + self, + transport: Callable[[dict[str, Any]], dict[str, Any]] = _no_transport, + *, + status_transport: Callable[[str], dict[str, Any]] = _no_status_transport, + ) -> None: self._transport = transport + self._status_transport = status_transport def submit_analysis_run(self, request: AnalysisRunRequest) -> dict[str, Any]: """Submit a request; returns TEPP's ``AnalysisRunAccepted`` envelope.""" - return self._transport(request.to_json()) + response = self._transport(request.to_json()) + if ( + isinstance(response, dict) + and response.get("run_state") == "accepted" + and not _valid_analysis_run_accepted(request, response) + ): + raise TeppInvalidResponse("TEPP analysis-run accepted response was invalid") + return response + + def read_analysis_run_status( + self, remote_run_id: str, request: AnalysisRunRequest + ) -> dict[str, Any]: + """Read and validate TEPP's request-bound status/result v1 payload.""" + response = self._status_transport(remote_run_id) + if not _valid_analysis_run_status(remote_run_id, request, response): + raise TeppInvalidResponse("TEPP analysis-run status response was invalid") + return response + + +_SHA256 = re.compile(r"[0-9a-f]{64}") +_FAILURE_CODE = re.compile(r"[a-z][a-z0-9_]{0,63}") +_RFC3339 = re.compile( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})" +) + + +def _valid_analysis_run_accepted( + request: AnalysisRunRequest, response: object +) -> bool: + """Mirror TEPP's bounded, strict accepted-response v1 contract.""" + if not isinstance(response, dict) or set(response) != { + "contract_version", + "run_id", + "run_state", + "idempotency_key", + }: + return False + try: + encoded = json.dumps(response, separators=(",", ":"), ensure_ascii=False).encode() + except (TypeError, ValueError): + return False + return ( + len(encoded) <= 64 * 1024 + and response["contract_version"] == 1 + and response["run_state"] == "accepted" + and response["idempotency_key"] == request.idempotency_key + and _nonempty(response["run_id"]) + and _nonempty(response["idempotency_key"]) + ) + + +def _nonempty(value: object) -> bool: + """Return whether a wire string contains non-whitespace, non-control text.""" + return ( + isinstance(value, str) + and bool(value.strip()) + and not any(unicodedata.category(char) == "Cc" for char in value) + ) + + +def _rfc3339(value: object) -> bool: + """Accept a timezone-bearing RFC 3339 timestamp understood by Python.""" + if not isinstance(value, str) or _RFC3339.fullmatch(value) is None: + return False + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return False + return parsed.tzinfo is not None + + +def _valid_analysis_run_status( + remote_run_id: str, request: AnalysisRunRequest, response: object +) -> bool: + """Validate TEPP v1 status and every terminal request binding.""" + if not isinstance(response, dict): + return False + try: + encoded = json.dumps( + response, separators=(",", ":"), ensure_ascii=False + ).encode() + except (TypeError, ValueError): + return False + if len(encoded) > 64 * 1024: + return False + status_keys = { + "contract_version", + "run_id", + "run_state", + "idempotency_key", + "terminal_result", + } + if ( + set(response) != status_keys + or response["contract_version"] != 1 + or response["run_id"] != remote_run_id + or response["idempotency_key"] != request.idempotency_key + ): + return False + state = response["run_state"] + terminal = response["terminal_result"] + if state in {"accepted", "running"}: + return terminal is None + if state not in {"succeeded", "failed"} or not isinstance(terminal, dict): + return False + terminal_keys = { + "contract_version", + "run_id", + "run_state", + "idempotency_key", + "tenant_workspace_id", + "snapshot_id", + "knowledge_cutoff", + "model_contract_version", + "output_profile", + "result_artifact_id", + "result_sha256", + "result_schema_version", + "completed_at", + "summary", + "failure_code", + } + if ( + set(terminal) != terminal_keys + or terminal["contract_version"] != 1 + or terminal["run_id"] != remote_run_id + or terminal["run_state"] != state + or terminal["idempotency_key"] != request.idempotency_key + or terminal["tenant_workspace_id"] != request.tenant_workspace_id + or terminal["snapshot_id"] != request.snapshot_id + or terminal["knowledge_cutoff"] != request.knowledge_cutoff + or not _rfc3339(terminal["knowledge_cutoff"]) + or terminal["model_contract_version"] != request.model_contract_version + or terminal["output_profile"] != request.output_profile + or not _rfc3339(terminal["completed_at"]) + ): + return False + if state == "failed": + return ( + terminal["result_artifact_id"] is None + and terminal["result_sha256"] is None + and terminal["result_schema_version"] is None + and terminal["summary"] is None + and isinstance(terminal["failure_code"], str) + and _FAILURE_CODE.fullmatch(terminal["failure_code"]) is not None + ) + summary = terminal["summary"] + return ( + _nonempty(terminal["result_artifact_id"]) + and isinstance(terminal["result_sha256"], str) + and _SHA256.fullmatch(terminal["result_sha256"]) is not None + and _nonempty(terminal["result_schema_version"]) + and terminal["failure_code"] is None + and isinstance(summary, dict) + and set(summary) + == {"analysis_family", "evidence_count", "statistic_count", "validation_status"} + and _nonempty(summary["analysis_family"]) + and _nonempty(summary["validation_status"]) + and type(summary["evidence_count"]) is int + and 0 <= summary["evidence_count"] <= 1_000_000_000 + and type(summary["statistic_count"]) is int + and 0 <= summary["statistic_count"] <= 1_000_000_000 + ) diff --git a/migrations/0217_analysis_run_tepp_receipt.sql b/migrations/0217_analysis_run_tepp_receipt.sql new file mode 100644 index 000000000..1cbbfaa8f --- /dev/null +++ b/migrations/0217_analysis_run_tepp_receipt.sql @@ -0,0 +1,14 @@ +-- TEPP acceptance is durable transport evidence, never a measurement result. +create table if not exists analysis_run_tepp_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 = 'accepted'), + received_at timestamptz not null default clock_timestamp() +); + +create index if not exists analysis_run_tepp_receipt_received_idx + on analysis_run_tepp_receipt (received_at); diff --git a/migrations/rollback/0217_analysis_run_tepp_receipt.sql b/migrations/rollback/0217_analysis_run_tepp_receipt.sql new file mode 100644 index 000000000..143172d68 --- /dev/null +++ b/migrations/rollback/0217_analysis_run_tepp_receipt.sql @@ -0,0 +1 @@ +drop table if exists analysis_run_tepp_receipt; diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index 8f508e954..a3e317470 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -10,6 +10,8 @@ from backend.app.analysis_run_ingestion import reconstructed_edge_is_visible from backend.app.analysis_run_start import ( AnalysisRunStartError, + _persist_tepp_receipt, + _persist_tepp_terminal_result, _persist_tepp_result, configured_tepp_client, reconstruction_member_ids, @@ -216,6 +218,279 @@ def _tepp_request() -> AnalysisRunRequest: ) +def test_tepp_delivery_separates_acceptance_from_terminal_measurement() -> None: + locked = { + "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", + "remote_run_id": None, + } + plan = analysis_run_start._DeliveryPlan( + "analysis_run_tepp", datetime(2026, 1, 12, tzinfo=timezone.utc), locked + ) + accepted = analysis_run_start._execute_delivery_plan( + plan, + TeppClient( + transport=lambda _payload: { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "accepted", + "idempotency_key": "buyer-tepp-2026-w07", + } + ), + None, + ) + + assert accepted.status_code == "analysis_status_running" + assert accepted.persist_receipt + assert accepted.request == _tepp_request() + + +def test_tepp_delivery_reads_a_stored_remote_run_without_resubmitting() -> None: + request = _tepp_request() + status = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "running", + "idempotency_key": request.idempotency_key, + "terminal_result": None, + } + locked = { + "idempotency_key": request.idempotency_key, + "snapshot_sha256": request.snapshot_id, + "knowledge_cutoff": datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc), + "corporate_entity_id": request.tenant_workspace_id, + "remote_run_id": "remote-run-1", + } + outcome = analysis_run_start._execute_delivery_plan( + analysis_run_start._DeliveryPlan( + "analysis_run_tepp", datetime(2026, 1, 12, tzinfo=timezone.utc), locked + ), + TeppClient( + transport=lambda _payload: pytest.fail("accepted work was resubmitted"), + status_transport=lambda _run_id: status, + ), + None, + ) + + assert outcome.status_code == "analysis_status_running" + assert not outcome.persist_receipt + + +def test_tepp_delivery_keeps_the_full_terminal_status_for_persistence() -> None: + request = _tepp_request() + terminal = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "succeeded", + "idempotency_key": request.idempotency_key, + "tenant_workspace_id": request.tenant_workspace_id, + "snapshot_id": request.snapshot_id, + "knowledge_cutoff": request.knowledge_cutoff, + "model_contract_version": request.model_contract_version, + "output_profile": request.output_profile, + "result_artifact_id": "artifact-1", + "result_sha256": "ab" * 32, + "result_schema_version": "tepp-result-v1", + "completed_at": "2026-01-13T00:00:00Z", + "summary": { + "analysis_family": "temporal_topic_measurement", + "evidence_count": 1, + "statistic_count": 1, + "validation_status": "validated", + }, + "failure_code": None, + } + status = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "succeeded", + "idempotency_key": request.idempotency_key, + "terminal_result": terminal, + } + outcome = analysis_run_start._execute_delivery_plan( + analysis_run_start._DeliveryPlan( + "analysis_run_tepp", + datetime(2026, 1, 12, tzinfo=timezone.utc), + { + "idempotency_key": request.idempotency_key, + "snapshot_sha256": request.snapshot_id, + "knowledge_cutoff": datetime(2026, 1, 12, 12, tzinfo=timezone.utc), + "corporate_entity_id": request.tenant_workspace_id, + "remote_run_id": "remote-run-1", + }, + ), + TeppClient(status_transport=lambda _run_id: status), + None, + ) + + assert outcome.status_code == "analysis_status_succeeded" + assert outcome.persist_terminal_result + assert outcome.envelope == status + + +def test_tepp_terminal_result_replay_must_match() -> None: + class _Transaction: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + class _Connection: + def __init__(self) -> None: + self.existing = None + self.inserted: tuple[object, ...] | None = None + + def transaction(self): + return _Transaction() + + async def fetchrow(self, _query: str, *_args: object): + return self.existing + + async def execute(self, _query: str, *args: object): + self.inserted = args + + envelope = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "succeeded", + "terminal_result": {"result_artifact_id": "artifact-1"}, + } + conn = _Connection() + assert asyncio.run( + _persist_tepp_terminal_result( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + envelope=envelope, + ) + ) + assert conn.inserted is not None + conn.existing = { + "remote_run_id": conn.inserted[1], + "result_sha256": conn.inserted[3], + } + assert asyncio.run( + _persist_tepp_terminal_result( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + envelope=envelope, + ) + ) + conn.existing["result_sha256"] = "0" * 64 + assert not asyncio.run( + _persist_tepp_terminal_result( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + envelope=envelope, + ) + ) + + +def test_tepp_acceptance_receipt_replay_must_match() -> None: + """A provider replay cannot replace the remote identity or evidence digest.""" + + class _Transaction: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + class _Connection: + def __init__(self) -> None: + self.existing = None + self.inserted: tuple[object, ...] | None = None + + def transaction(self): + return _Transaction() + + async def fetchrow(self, _query: str, *_args: object): + return self.existing + + async def execute(self, _query: str, *args: object): + self.inserted = args + + request = _tepp_request() + receipt = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "accepted", + "idempotency_key": request.idempotency_key, + } + conn = _Connection() + assert asyncio.run( + _persist_tepp_receipt( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + request=request, + envelope=receipt, + ) + ) + assert conn.inserted is not None + conn.existing = { + "remote_run_id": conn.inserted[1], + "request_sha256": conn.inserted[2], + "receipt_sha256": conn.inserted[3], + } + assert asyncio.run( + _persist_tepp_receipt( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + request=request, + envelope=receipt, + ) + ) + conn.existing["remote_run_id"] = "changed-run" + assert not asyncio.run( + _persist_tepp_receipt( + conn, + analysis_run_id="11111111-1111-1111-1111-111111111111", + request=request, + envelope=receipt, + ) + ) + + +def test_tepp_persistence_conflict_rolls_back_savepoint() -> None: + """A provider identity conflict fails closed without aborting its caller.""" + + class _Transaction: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + class _Connection: + def transaction(self): + return _Transaction() + + async def fetchrow(self, _query: str, *_args: object): + return None + + async def execute(self, _query: str, *_args: object): + raise analysis_run_start.asyncpg.UniqueViolationError("duplicate remote run") + + request = _tepp_request() + assert not asyncio.run( + _persist_tepp_receipt( + _Connection(), + analysis_run_id="11111111-1111-1111-1111-111111111111", + request=request, + envelope={"run_id": "remote-run-1", "run_state": "accepted"}, + ) + ) + assert not asyncio.run( + _persist_tepp_terminal_result( + _Connection(), + analysis_run_id="11111111-1111-1111-1111-111111111111", + envelope={"run_id": "remote-run-1", "run_state": "succeeded"}, + ) + ) + + def test_tepp_run_request_is_the_published_wire_shape() -> None: """Start builds TEPP's seven-field request from the frozen run.""" request = _tepp_request() @@ -279,6 +554,9 @@ def transaction(self): async def execute(self, query: str, *args: object): self.queries.append((query, args)) + async def fetchrow(self, _query: str, *_args: object): + return None + cutoff = datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc) conn = _Connection() envelope = { @@ -393,6 +671,28 @@ def __init__(self) -> None: assert failure == "tepp_result_not_persisted" +def test_topic_lineage_submit_outcome_rejects_strict_async_acceptance() -> None: + """Topic lineage has no receipt/resume path, so acceptance fails closed.""" + + class _Accepting(TeppClient): + def __init__(self) -> None: + super().__init__( + transport=lambda payload: { + "contract_version": 1, + "run_id": "topic-run-1", + "run_state": "accepted", + "idempotency_key": payload["idempotency_key"], + } + ) + + status, failure, envelope = topic_lineage_submit_outcome( + _Accepting(), _topic_lineage_request() + ) + assert status == "analysis_status_failed" + assert failure == "tepp_result_not_persisted" + assert envelope is None + + def test_topic_lineage_submit_outcome_rejects_a_contentless_completed_envelope() -> None: """A 'completed' envelope missing the topic-identity/CHRONOS contract is Failed. diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 95c5c4fe4..af7f6d1fa 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -196,6 +196,36 @@ def test_topic_lineage_result_migration_is_idempotent_for_replay() -> None: assert "create index if not exists" in migration +def test_tepp_receipt_migration_is_replayable_and_digest_bound() -> None: + """Accepted transport evidence survives every-start migration replay.""" + migration_name = "0217_analysis_run_tepp_receipt.sql" + sql = ( + Path(__file__).resolve().parents[1] / "migrations" / migration_name + ).read_text(encoding="utf-8").casefold() + + assert re.fullmatch(r"[0-9]{4}_.+\.sql", migration_name) + assert "create table if not exists analysis_run_tepp_receipt" in sql + assert "remote_run_id text not null unique" in sql + assert "request_sha256 ~ '^[0-9a-f]{64}$'" in sql + assert "receipt_sha256 ~ '^[0-9a-f]{64}$'" in sql + assert "accepted_status_code = 'accepted'" in sql + assert "create index if not exists" in sql + + +def test_tepp_receipt_read_requires_the_replayed_schema() -> None: + """A missing required table must fail before it poisons a claim transaction.""" + source = ( + Path(__file__).resolve().parents[1] + / "backend" + / "app" + / "analysis_run_ingestion.py" + ).read_text(encoding="utf-8") + receipt_block = source.split('if row["run_kind_code"] == _TEPP_RUN_KIND:', 1)[1] + receipt_block = receipt_block.split("return detail", 1)[0] + + assert "from analysis_run_tepp_receipt" in receipt_block + assert "UndefinedTableError" not in receipt_block + def test_global_ask_job_migrations_are_idempotent_for_replay() -> None: """Existing volumes must replay the queue and authorization scope safely.""" migrations = Path(__file__).resolve().parents[1] / "migrations" diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index 01fc91a94..9824a684f 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -3,7 +3,12 @@ import pytest from backend.app.analysis_run_start import configured_tepp_client -from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable +from lineageweave.tepp_client import ( + AnalysisRunRequest, + TeppClient, + TeppInvalidResponse, + TeppNotAvailable, +) def _sample_request() -> AnalysisRunRequest: @@ -11,7 +16,7 @@ def _sample_request() -> AnalysisRunRequest: idempotency_key="demo-run-1", tenant_workspace_id="demo-workspace", snapshot_id="demo-snapshot-1", - knowledge_cutoff="2026-01-01", + knowledge_cutoff="2026-01-01T00:00:00Z", model_contract_version="v1", output_profile="graphml", ) @@ -25,7 +30,7 @@ def test_to_json_matches_tepp_published_schema_shape() -> None: "idempotency_key": "demo-run-1", "tenant_workspace_id": "demo-workspace", "snapshot_id": "demo-snapshot-1", - "knowledge_cutoff": "2026-01-01", + "knowledge_cutoff": "2026-01-01T00:00:00Z", "model_contract_version": "v1", "output_profile": "graphml", } @@ -35,6 +40,68 @@ def test_default_transport_fails_closed_until_tepp_ships_http() -> None: client = TeppClient() with pytest.raises(TeppNotAvailable): client.submit_analysis_run(_sample_request()) + with pytest.raises(TeppNotAvailable): + client.read_analysis_run_status("tepp-run-1", _sample_request()) + + +def _terminal_status() -> dict: + request = _sample_request() + return { + "contract_version": 1, + "run_id": "tepp-run-1", + "run_state": "succeeded", + "idempotency_key": request.idempotency_key, + "terminal_result": { + "contract_version": 1, + "run_id": "tepp-run-1", + "run_state": "succeeded", + "idempotency_key": request.idempotency_key, + "tenant_workspace_id": request.tenant_workspace_id, + "snapshot_id": request.snapshot_id, + "knowledge_cutoff": request.knowledge_cutoff, + "model_contract_version": request.model_contract_version, + "output_profile": request.output_profile, + "result_artifact_id": "artifact-1", + "result_sha256": "ab" * 32, + "result_schema_version": "tepp-result-v1", + "completed_at": "2026-01-02T03:04:05Z", + "summary": { + "analysis_family": "temporal_topic_measurement", + "evidence_count": 12, + "statistic_count": 4, + "validation_status": "validated", + }, + "failure_code": None, + }, + } + + +def test_status_reader_accepts_only_request_bound_terminal_results() -> None: + status = _terminal_status() + client = TeppClient(status_transport=lambda _run_id: status) + assert client.read_analysis_run_status("tepp-run-1", _sample_request()) == status + + status["terminal_result"]["snapshot_id"] = "other-snapshot" + with pytest.raises(TeppInvalidResponse): + client.read_analysis_run_status("tepp-run-1", _sample_request()) + + +def test_status_reader_requires_strict_rfc3339_terminal_time() -> None: + status = _terminal_status() + status["terminal_result"]["completed_at"] = "2026-01-02 03:04:05+00:00" + client = TeppClient(status_transport=lambda _run_id: status) + with pytest.raises(TeppInvalidResponse): + client.read_analysis_run_status("tepp-run-1", _sample_request()) + + +@pytest.mark.parametrize( + "status", + [None, {}, {"unencodable": {1}}, {"result_artifact_id": "x" * (64 * 1024)}], +) +def test_status_reader_rejects_invalid_or_oversized_payloads(status) -> None: + client = TeppClient(status_transport=lambda _run_id: status) + with pytest.raises(TeppInvalidResponse): + client.read_analysis_run_status("tepp-run-1", _sample_request()) def test_custom_transport_receives_the_exact_wire_payload() -> None: @@ -52,6 +119,20 @@ def fake_transport(payload: dict) -> dict: assert received["snapshot_id"] == "demo-snapshot-1" +def test_accepted_response_rejects_oversized_provider_identity() -> None: + client = TeppClient( + transport=lambda _payload: { + "contract_version": 1, + "run_id": "x" * (64 * 1024), + "run_state": "accepted", + "idempotency_key": _sample_request().idempotency_key, + } + ) + + with pytest.raises(TeppInvalidResponse): + client.submit_analysis_run(_sample_request()) + + def test_configured_transport_sends_tepp_consumer_contract_headers(monkeypatch: pytest.MonkeyPatch) -> None: received = {}