From 5f8613e3023fc58f00d9ad79fa4a5e64dd8f3506 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 02:44:33 +0900 Subject: [PATCH 1/2] feat: consume TEPP terminal result contract --- .../2.20.0-tepp-terminal-result-consumer.md | 8 + backend/app/analysis_run_start.py | 57 +++++- docs/adr/0162-tepp-accepted-receipt.md | 11 +- .../adr/0178-tepp-terminal-result-consumer.md | 77 ++++++++ docs/product-technical-gap-baseline.md | 44 ++++- lineageweave/tepp_client.py | 141 ++++++++++++++ tests/test_tepp_accepted_receipt.py | 172 +++++++++++++++++- tests/test_tepp_client.py | 95 ++++++++++ 8 files changed, 585 insertions(+), 20 deletions(-) create mode 100644 CHANGELOG.d/2.20.0-tepp-terminal-result-consumer.md create mode 100644 docs/adr/0178-tepp-terminal-result-consumer.md diff --git a/CHANGELOG.d/2.20.0-tepp-terminal-result-consumer.md b/CHANGELOG.d/2.20.0-tepp-terminal-result-consumer.md new file mode 100644 index 000000000..1093a912f --- /dev/null +++ b/CHANGELOG.d/2.20.0-tepp-terminal-result-consumer.md @@ -0,0 +1,8 @@ +# 2.20.0 — TEPP terminal result consumer + +- Reads TEPP's versioned status/result contract from a stored accepted receipt + without resubmitting the measurement request. +- Revalidates every request binding and persists only digest-bound terminal + measurement evidence; accepted/running states remain measurement-free. +- Rejects changed result digests and maps a validated provider terminal failure + to a typed local failure without inventing a score or theta. diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index f60b6ce08..e3cc02e9d 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -36,7 +36,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" @@ -277,6 +282,32 @@ def classify_tepp_submission( return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") +def classify_tepp_status( + client: TeppClient, + request: AnalysisRunRequest, + remote_run_id: str, +) -> TeppSubmissionOutcome: + """Read one bounded, request-bound TEPP status without resubmitting work.""" + try: + response = client.read_analysis_run_status(remote_run_id, request) + except TeppNotAvailable: + return TeppSubmissionOutcome(_RUNNING, "", None, "") + except TeppInvalidResponse: + return TeppSubmissionOutcome(_FAILED, "tepp_result_not_persisted", None, "") + state = response["run_state"] + if state in {"accepted", "running"}: + return TeppSubmissionOutcome(_RUNNING, "", response, "") + terminal = response["terminal_result"] + if state == "failed": + return TeppSubmissionOutcome( + _FAILED, + str(terminal["failure_code"]), + terminal, + "", + ) + return TeppSubmissionOutcome(_SUCCEEDED, "", terminal, _PERSIST_RESULT) + + def _tepp_submission( client: TeppClient, request: AnalysisRunRequest, @@ -309,6 +340,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 @@ -973,10 +1017,13 @@ async def _deliver_tepp_measurement( knowledge_cutoff=locked["knowledge_cutoff"], corporate_entity_id=str(locked["corporate_entity_id"]), ) - 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: + receipt = await fetch_tepp_accepted_receipt(conn, analysis_run_id) + outcome = ( + classify_tepp_status(tepp_client, request, str(receipt["remote_run_id"])) + if receipt is not None + else classify_tepp_submission(tepp_client, request) + ) + if receipt is not None and outcome.status_code == _RUNNING: return False status_code = outcome.status_code failure_code = outcome.failure_code diff --git a/docs/adr/0162-tepp-accepted-receipt.md b/docs/adr/0162-tepp-accepted-receipt.md index 4b051ac75..0c798b7c4 100644 --- a/docs/adr/0162-tepp-accepted-receipt.md +++ b/docs/adr/0162-tepp-accepted-receipt.md @@ -5,8 +5,8 @@ **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) +**Refs:** Issue #277; terminal-result consumption continues in ADR 0178 after +[TEPP#156](https://github.com/ContextualWisdomLab/TEPP/issues/156) closed ## Context @@ -65,9 +65,10 @@ 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. +Completed-result request-binding revalidation is ADR 0178. Automatic polling +and backoff remain unavailable until TEPP publishes a provider-owned HTTP +status service and evidence-based retry policy. This ADR does not invent a +local psychometric substitute while waiting. ```mermaid sequenceDiagram diff --git a/docs/adr/0178-tepp-terminal-result-consumer.md b/docs/adr/0178-tepp-terminal-result-consumer.md new file mode 100644 index 000000000..7cca25790 --- /dev/null +++ b/docs/adr/0178-tepp-terminal-result-consumer.md @@ -0,0 +1,77 @@ +# ADR 0178 — Read TEPP terminal results through the provider contract + +**Decision status:** Accepted on this stacked PR; not protected-main truth until merge +**Date:** 2026-08-26 +**Depends on:** ADR 0022, ADR 0023, ADR 0162; TEPP PR #157 +**Refs:** LineageWeave issue #277; TEPP issues #156 and #249 + +## Context + +ADR 0162 correctly separates TEPP's accepted receipt from measurement, but it +predates TEPP's versioned `AnalysisRunStatus` and +`AnalysisRunTerminalResult` v1 contracts. TEPP PR #157 merged those Rust wire +contracts on 2026-08-25. It deliberately did not publish a production HTTP +status service, so LineageWeave must not guess a URL or poll interval. + +## Decision + +`TeppClient` exposes a pluggable status-read transport alongside its existing +submission transport. A stored accepted receipt causes the delivery retry to +read that remote run once instead of resubmitting the request. The read is +bounded by the supplied transport call and is crash-resumable because the +receipt and PostgreSQL outbox remain durable. + +The consumer requires the exact v1 status shape and revalidates remote run, +idempotency key, tenant/workspace, snapshot, cutoff, model contract, output +profile, terminal state, result schema, lowercase SHA-256 digest, bounded +identity-free summary, completion time, and failure code. Accepted/running +contains no terminal result and keeps the local run Running. A succeeded +terminal result is persisted before Succeeded. A terminal provider failure +appends its validated snake-case failure code without a result. Any mismatch +fails closed. Replaying the same remote run with a changed canonical result +digest is rejected. + +The configured HTTP client does not synthesize `GET /v1/analysis-runs/{id}` +while TEPP documents it only as a target endpoint. A production status +transport is enabled only when the owning TEPP service publishes that route. +No theta, score, estimator, poll cadence, or backoff coefficient is implemented +in LineageWeave. + +```mermaid +sequenceDiagram + participant Worker + participant Registry + participant TEPP + Worker->>Registry: read accepted receipt + immutable request + Worker->>TEPP: read status(remote run id) + alt accepted or running + Note over Registry: remain Running; outbox remains claimed + else succeeded and all bindings match + Worker->>Registry: persist terminal DTO + Succeeded atomically + else failed and all bindings match + Worker->>Registry: append typed Failed + else unavailable + Note over Registry: retain receipt; retry remains possible + else invalid or mismatched + Worker->>Registry: fail closed + end +``` + +## Consequences + +The contract consumer is testable now through an in-process or future HTTP +adapter without pretending TEPP has deployed a route. Automatic scheduled +polling remains unavailable until the provider owns an executable status +endpoint and an evidence-based retry policy. + +## 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/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 889ab85b5..ad932dccc 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,13 +1,38 @@ # Product & Technical Gap Baseline -> Audit scope: the current LineageWeave reader/source-context worktree and all -> 56 open PRs, compared with protected `main`, the UI/UX Standard Guide v3.0, +> Audit scope: the current LineageWeave TEPP terminal-result continuation and +> all 9 open PRs, compared with protected `main`, the UI/UX Standard Guide v3.0, > ADR 0118, the accepted TEPP contracts, and contextual-orchestrator. Real > source identifiers are deliberately replaced with case labels; they must not > enter repository artifacts. ## 1. Exact-head evidence +### 1.0 Current TEPP continuation + +Observed at `2026-08-26`: protected `main` is +`04e6b610655d0db91d5f7ba9486bdda1440e0b19`. Nine PRs target `main`: +#644 `c1018a0a`, #643 `041ec13b`, #640 `2fad1fe6`, #639 `aee02dca`, +#636 `f7b9a65f`, #632 `3e3f0ead`, #631 `c0022c97`, #629 `4b4d6707`, and +#579 `689a21b6`. All have auto-merge armed and zero unresolved review threads; +none has the independent exact-head approval required by protected rules, so +no protected merge is claimed. The older snapshots below remain historical. + +TEPP PR #157 merged as `7ce87c305981819f5333c7eb90ea0feafc0f7bf6` +and closed TEPP issue #156 by publishing `AnalysisRunStatus` and +`AnalysisRunTerminalResult` v1. The provider explicitly did not deploy a +production HTTP status service. This continuation therefore consumes the +strict contract through a pluggable status-read port, retains accepted/running +as transport evidence, persists only fully request-bound terminal results, +and rejects changed digests. It does not guess a provider URL, polling cadence, +backoff coefficient, theta, or score. Focused exact-source evidence is +`64 passed`; the full Python suite passed `1087` with `16` live-stack skips. +TEPP issue #249 now owns the executable HTTP status-service gap. The unrelated +Starlette `httpx2` migration and short synthetic JWT-test key warnings remain +pre-existing dependency/test-fixture gaps and are not suppressed in this +TEPP-scoped continuation. Protected checks and independent review remain +required. + ### 1.1 Current continuation head Observed at `2026-08-24T06:45:00+09:00`: protected `main` and `origin/main` @@ -551,7 +576,7 @@ adapter, fixture, or HTTP-shaped test double never upgrades a row to | Evidence-grounded chat and source navigation | `/chat`, `/ask`, citation/evidence UI | source + unit; synthetic orchestrator judge route verified, corpus chat/runtime evidence open | | OpenTelemetry across LineageWeave, contextual-orchestrator, Valkey, and GRC | LineageWeave PR #383 adds API/Valkey/session spans; contextual-orchestrator PR #818 carries session/provider telemetry; governance-risk-compliance PR #51 adds request telemetry, W3C trace context, OTLP export, and ADR 0009 | source + PR; protected merge and end-to-end collector evidence open | | PU/team/project weekly/monthly reports | report API/UI and grouping controls | source + unit; TEPP-backed live report open | -| TEPP calibrated measurement, dichotomous items, multilevel/MMM/time model | published import/REST boundary and TEPP ADR/PRD references | boundary-only; live-external open | +| TEPP calibrated measurement, dichotomous items, multilevel/MMM/time model | accepted receipt parent #496 plus TEPP terminal-result v1 consumer (ADR 0178); arithmetic remains TEPP-owned | strict contract consumer source + focused unit; provider HTTP status service and live-external evidence open | | contextual-orchestrator routing, VISION, embedding, schema repair | clients and provenance/session boundary; synthetic authenticated route returned a judge score of `0.98`, OCR succeeded, and region location returned five regions | source + local-integration partial; corpus backfill, capability/readiness evidence, and schema-repair workflow open | | HTML semantic units, tables, indentation, footnotes, formulas | parser modules and synthetic tests; adjacent open PR #367 at exact head `b628722cb000717b0198e4337d12306d4306922d` adds numbered-footnote, leading-empty-cell, and short-ID regressions; 11-case authenticated popup sweep had no popup errors and rendered the supplied footnote/table cases; bounded metric superscript/subscript normalization has backend/frontend focused coverage | source + unit + local-integration partial; PR #367 protected checks, arbitrary formula/semantic correctness, and corpus re-backfill remain open | | Base64/file image regions and multimodal evidence | image-region schema and VISION client boundary; live aggregate has 12,823 images, 25 described images, 421 failed images, 12,377 unavailable images, and 19 persisted regions; the bounded real-data queue run published three Valkey wake-ups and the worker claimed one | source + local-integration partial; supplied image-table case re-backfill and complete corpus coverage open | @@ -780,21 +805,22 @@ or an explicit unavailable result. digest mismatches before writes. The remaining acceptance work is to mount the authorized raw artifacts and run the real import/backfill; do not map an unrelated metadata column as body. -- **TEPP measurement — boundary accepted, runtime open:** LineageWeave must +- **TEPP measurement — terminal consumer implemented, runtime open:** LineageWeave must call TEPP through its published import/REST contract and must not implement a local theta, psychometric calibration, CAT, or judge score. TEPP owns the Rust numerical/psychometric layer and its multilevel/multiple-membership/time model. Live inspection on 2026-08-23 found that the upstream TEPP repository - currently publishes strict `AnalysisRunRequest` / `AnalysisRunAccepted` DTOs - and outbound HTTP exchange builders, but no executable HTTP server, completed - measurement response contract, snapshot-evidence ingest, or production + now publishes strict `AnalysisRunRequest`, `AnalysisRunAccepted`, + `AnalysisRunStatus`, and `AnalysisRunTerminalResult` v1 DTOs and outbound HTTP + exchange builders, but no production HTTP status server, snapshot-evidence + ingest, or production estimator entrypoint. The current request carries only a snapshot digest, so a service cannot calibrate the underlying observations without a new purpose-bound evidence artifact/API. `TEPP_TRANSPORT_URL` alone therefore cannot make measurement available. Close this in TEPP first with an ADR and PRD update covering authorized evidence transfer, Rust estimator authority, - durable lifecycle/idempotency, completed-result provenance, and CPU/GPU - parity; then pin that service in Compose and prove a persisted + durable lifecycle/idempotency, provider HTTP status route, and CPU/GPU parity; + then pin that service in Compose and prove a persisted `analysis_run_tepp_result`. An accepted-envelope shim is explicitly not an acceptable substitute. - **TEPP temporal context — source-connected, local runtime proven:** TEPP's diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index 5d7faf3a5..754927ea3 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -17,8 +17,12 @@ from __future__ import annotations +import json +import re +import unicodedata from collections.abc import Callable from dataclasses import dataclass +from datetime import datetime from typing import Any @@ -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. @@ -128,14 +141,25 @@ def __init__( transport: Callable[[dict[str, Any]], dict[str, Any]] = _no_transport, *, temporal_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._temporal_transport = temporal_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()) + 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 + def temporal_context(self, request: TemporalContextRequest) -> dict[str, Any]: """Return TEPP-owned ordering; callers must validate its claim boundary.""" response = self._temporal_transport(request.to_json()) @@ -144,6 +168,123 @@ def temporal_context(self, request: TemporalContextRequest) -> dict[str, Any]: return response +_SHA256 = re.compile(r"[0-9a-f]{64}") +_FAILURE_CODE = re.compile(r"[a-z][a-z0-9_]{0,63}") + + +def _nonempty(value: object) -> bool: + """Return whether a wire string contains non-whitespace text.""" + return ( + isinstance(value, str) + and bool(value.strip()) + and not any(unicodedata.category(char).startswith("C") for char in value) + ) + + +def _rfc3339(value: object) -> bool: + """Accept an RFC 3339 timestamp understood by the Python runtime.""" + if not isinstance(value, str): + 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() + if len(encoded) > 64 * 1024: + return False + except (TypeError, ValueError): + 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 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 + ) + + def _valid_temporal_response( request: TemporalContextRequest, response: object ) -> bool: diff --git a/tests/test_tepp_accepted_receipt.py b/tests/test_tepp_accepted_receipt.py index b1605f678..77f5baf16 100644 --- a/tests/test_tepp_accepted_receipt.py +++ b/tests/test_tepp_accepted_receipt.py @@ -27,9 +27,10 @@ async def __aexit__(self, exc_type, exc, traceback): class _ReceiptConnection: - def __init__(self, *, rows=None, row=None, error=None) -> None: + def __init__(self, *, rows=None, row=None, result_row=None, error=None) -> None: self.rows = [] if rows is None else rows self.row = row + self.result_row = result_row self.error = error self.transactions = 0 self.executions: list[tuple[object, ...]] = [] @@ -46,6 +47,8 @@ async def fetch(self, query, analysis_run_ids): async def fetchrow(self, query, analysis_run_id): if self.error is not None: raise self.error + if "from analysis_run_tepp_result" in query: + return self.result_row return self.row async def execute(self, *args): @@ -229,6 +232,153 @@ def test_completed_result_is_the_only_measurement_persist() -> None: assert outcome.failure_code == "" +def _terminal_status(state: str = "succeeded") -> dict: + request = _request() + failed = state == "failed" + terminal = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": state, + "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": None if failed else "artifact-1", + "result_sha256": None if failed else "ab" * 32, + "result_schema_version": None if failed else "tepp-result-v1", + "completed_at": "2026-01-12T13:00:00Z", + "summary": None + if failed + else { + "analysis_family": "temporal_event_measurement", + "evidence_count": 4, + "statistic_count": 2, + "validation_status": "validated", + }, + "failure_code": "estimation_failed" if failed else None, + } + return { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": state, + "idempotency_key": request.idempotency_key, + "terminal_result": terminal, + } + + +def test_status_read_promotes_only_validated_terminal_result() -> None: + status = _terminal_status() + client = TeppClient(status_transport=lambda _run_id: status) + + outcome = analysis_run_start_module.classify_tepp_status( + client, _request(), "remote-run-1" + ) + + assert outcome.status_code == "analysis_status_succeeded" + assert outcome.persist_kind == "result" + assert outcome.envelope == status["terminal_result"] + + +def test_status_read_maps_provider_terminal_failure_without_result() -> None: + status = _terminal_status("failed") + client = TeppClient(status_transport=lambda _run_id: status) + + outcome = analysis_run_start_module.classify_tepp_status( + client, _request(), "remote-run-1" + ) + + assert outcome.status_code == "analysis_status_failed" + assert outcome.failure_code == "estimation_failed" + assert outcome.persist_kind == "" + + +def test_status_read_keeps_provider_running_and_fails_closed_on_invalid() -> None: + request = _request() + running = { + "contract_version": 1, + "run_id": "remote-run-1", + "run_state": "running", + "idempotency_key": request.idempotency_key, + "terminal_result": None, + } + running_outcome = analysis_run_start_module.classify_tepp_status( + TeppClient(status_transport=lambda _run_id: running), + request, + "remote-run-1", + ) + invalid_outcome = analysis_run_start_module.classify_tepp_status( + TeppClient(status_transport=lambda _run_id: {}), + request, + "remote-run-1", + ) + + assert running_outcome.status_code == "analysis_status_running" + assert running_outcome.persist_kind == "" + assert invalid_outcome.status_code == "analysis_status_failed" + assert invalid_outcome.failure_code == "tepp_result_not_persisted" + + +def test_running_delivery_reads_and_persists_terminal_status(monkeypatch) -> None: + analysis_run_id = "11111111-1111-1111-1111-111111111111" + visible_calls = 0 + + async def fetch_visible(*_args, **_kwargs): + nonlocal visible_calls + visible_calls += 1 + return { + "analysis_run_id": analysis_run_id, + "status_code": ( + "analysis_status_running" + if visible_calls == 1 + else "analysis_status_succeeded" + ), + } + + 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, 12, 0, tzinfo=timezone.utc), + "idempotency_key": "buyer-tepp-2026-w07", + "analysis_source_snapshot_id": "snapshot-1", + "snapshot_sha256": "ab" * 32, + "corporate_entity_id": "11111111-1111-1111-1111-111111111111", + }, + 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), + } + ], + ) + status = _terminal_status() + client = TeppClient(status_transport=lambda _run_id: status) + + 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=client, + ) + ) + + assert result["status_code"] == "analysis_status_succeeded" + assert any( + "insert into analysis_run_tepp_result" in str(execution[0]) + for execution in connection.executions + ) + assert any( + "analysis_outbox_delivered" in execution for execution in connection.executions + ) + + def test_completed_remote_run_id_alias_persists_the_result() -> None: connection = _ReceiptConnection() persisted = asyncio.run( @@ -246,6 +396,26 @@ def test_completed_remote_run_id_alias_persists_the_result() -> None: assert connection.executions[0][2] == "remote-run" +def test_completed_result_replay_rejects_changed_digest() -> None: + connection = _ReceiptConnection( + result_row={"remote_run_id": "remote-run", "result_sha256": "0" * 64} + ) + persisted = asyncio.run( + analysis_run_start_module._persist_tepp_result( + connection, + analysis_run_id="local-run", + envelope={ + "contract_version": 1, + "run_id": "remote-run", + "run_state": "succeeded", + }, + ) + ) + + assert persisted is False + assert connection.executions == [] + + def test_receipt_insert_error_is_rolled_back_by_a_savepoint() -> None: connection = _ReceiptConnection(error=asyncpg.UniqueViolationError("duplicate")) persisted = asyncio.run( diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index 17f1717df..54ac95806 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -8,6 +8,7 @@ TemporalContextEvent, TemporalContextRequest, TeppClient, + TeppInvalidResponse, TeppNotAvailable, ) @@ -41,6 +42,8 @@ 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 test_custom_transport_receives_the_exact_wire_payload() -> None: @@ -58,6 +61,98 @@ def fake_transport(payload: dict) -> dict: assert received["snapshot_id"] == "demo-snapshot-1" +def _succeeded_status() -> dict: + """Return TEPP's identity-free terminal status v1 fixture.""" + 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_the_request_bound_terminal_contract() -> None: + status = _succeeded_status() + client = TeppClient(status_transport=lambda _run_id: status) + + assert client.read_analysis_run_status("tepp-run-1", _sample_request()) == status + + +@pytest.mark.parametrize( + "mutate", + [ + lambda status: status.clear(), + lambda status: status.update(run_id="other-run"), + lambda status: status["terminal_result"].update(snapshot_id="other-snapshot"), + lambda status: status["terminal_result"].update(result_sha256="not-a-digest"), + lambda status: status["terminal_result"].update(completed_at="2026-01-02"), + lambda status: status["terminal_result"].update(completed_at=None), + lambda status: status["terminal_result"].update(completed_at="not-a-time"), + lambda status: status["terminal_result"].update(summary={}), + lambda status: status["terminal_result"]["summary"].update( + analysis_family="family\u001fhidden" + ), + lambda status: status["terminal_result"].update( + result_artifact_id="x" * (64 * 1024) + ), + lambda status: status.update(extra=True), + lambda status: status.update(run_state="unknown", terminal_result=None), + lambda status: status.update(run_state="succeeded", terminal_result=[]), + ], +) +def test_status_reader_fails_closed_on_identity_shape_and_digest_mismatch(mutate) -> None: + status = _succeeded_status() + mutate(status) + 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("invalid", [None, {"unencodable": {1, 2}}]) +def test_status_reader_rejects_non_object_and_non_json_payloads(invalid) -> None: + client = TeppClient(status_transport=lambda _run_id: invalid) + + with pytest.raises(TeppInvalidResponse): + client.read_analysis_run_status("tepp-run-1", _sample_request()) + + +def test_status_reader_keeps_nonterminal_receipts_measurement_free() -> None: + status = { + "contract_version": 1, + "run_id": "tepp-run-1", + "run_state": "running", + "idempotency_key": _sample_request().idempotency_key, + "terminal_result": None, + } + client = TeppClient(status_transport=lambda _run_id: status) + + assert client.read_analysis_run_status("tepp-run-1", _sample_request()) == status + + def test_configured_transport_sends_optional_bearer_key(monkeypatch: pytest.MonkeyPatch) -> None: received = {} From 1142a76d212561c037ca993abb27499c6e3edf17 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 02:45:14 +0900 Subject: [PATCH 2/2] docs: record TEPP consumer pull request --- docs/product-technical-gap-baseline.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index ad932dccc..77fed159c 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -27,7 +27,8 @@ as transport evidence, persists only fully request-bound terminal results, and rejects changed digests. It does not guess a provider URL, polling cadence, backoff coefficient, theta, or score. Focused exact-source evidence is `64 passed`; the full Python suite passed `1087` with `16` live-stack skips. -TEPP issue #249 now owns the executable HTTP status-service gap. The unrelated +PR #656 carries this stacked consumer at exact head `5f8613e3`; TEPP issue #249 +now owns the executable HTTP status-service gap. The unrelated Starlette `httpx2` migration and short synthetic JWT-test key warnings remain pre-existing dependency/test-fixture gaps and are not suppressed in this TEPP-scoped continuation. Protected checks and independent review remain