From 7388e183c1e6ff18c579a215a61cb0d78e5faf9e Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 04:18:34 +0900 Subject: [PATCH 1/5] feat(tepp): add terminal status read boundary --- CHANGELOG.d/2.20.0-tepp-status-read-client.md | 5 +++ backend/app/analysis_run_start.py | 23 ++++++++++- ...0217-tepp-terminal-status-read-boundary.md | 33 ++++++++++++++++ docs/product-technical-gap-baseline.md | 2 +- lineageweave/tepp_client.py | 37 ++++++++++-------- tests/test_tepp_client.py | 38 +++++++++++++++++++ 6 files changed, 119 insertions(+), 19 deletions(-) create mode 100644 CHANGELOG.d/2.20.0-tepp-status-read-client.md create mode 100644 docs/adr/0217-tepp-terminal-status-read-boundary.md diff --git a/CHANGELOG.d/2.20.0-tepp-status-read-client.md b/CHANGELOG.d/2.20.0-tepp-status-read-client.md new file mode 100644 index 000000000..77714bc94 --- /dev/null +++ b/CHANGELOG.d/2.20.0-tepp-status-read-client.md @@ -0,0 +1,5 @@ +# Added + +- Add a fail-closed TEPP analysis-run status/read client boundary so durable + lifecycle work can poll the upstream Rust-owned contract without treating an + accepted receipt as measurement. diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index c08810078..82cd1652b 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -16,6 +16,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import Any +from urllib.parse import quote from uuid import UUID import asyncpg @@ -35,7 +36,7 @@ records_from_source_posts, ) from lineageweave.adjudication_client import AdjudicationClient -from lineageweave.http_client import HttpClientError, post_json +from lineageweave.http_client import HttpClientError, get_json, post_json from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable @@ -184,7 +185,25 @@ def transport(payload: dict[str, Any]) -> dict[str, Any]: # message stays generic, never the raw provider exception text. raise TeppNotAvailable("TEPP transport unavailable") from exc - return TeppClient(transport=transport) + def status_transport(run_id: str) -> dict[str, Any]: + """GET TEPP's request-bound status envelope without interpreting it.""" + try: + headers = { + "tepp-consumer": "lineageweave", + "tepp-contract-version": "1", + } + if api_key.strip(): + headers["authorization"] = f"Bearer {api_key}" + return get_json( + f"{url.rstrip('/')}/{quote(run_id, safe='')}", + headers=headers, + timeout=30.0, + service_peer_name="tepp", + ) + except (HttpClientError, OSError, ValueError, TypeError) as exc: + raise TeppNotAvailable("TEPP status transport unavailable") from exc + + return TeppClient(transport=transport, status_transport=status_transport) def tepp_run_request( diff --git a/docs/adr/0217-tepp-terminal-status-read-boundary.md b/docs/adr/0217-tepp-terminal-status-read-boundary.md new file mode 100644 index 000000000..4d8dd0371 --- /dev/null +++ b/docs/adr/0217-tepp-terminal-status-read-boundary.md @@ -0,0 +1,33 @@ +# ADR 0217 — Read TEPP status without treating transport as measurement + +**Decision status:** Accepted on this branch; not protected-main truth until merge +**Date:** 2026-08-26 +**Depends on:** ADR 0022; issue #277; ContextualWisdomLab/TEPP PR #157 + +## Context + +TEPP now publishes a versioned status/read contract for accepted, running, +succeeded, and failed analysis runs. LineageWeave can submit a request but its +client has no read operation, so issue #277 cannot poll a remote run without +bypassing the existing provider boundary. + +## Decision + +TeppClient.get_analysis_run_status reads one opaque remote run identity +through a separately injected status transport. The configured HTTP client +derives the item URL from TEPP_TRANSPORT_URL, percent-encodes the opaque +identity, reuses the TEPP consumer/version headers, and fails closed behind +TeppNotAvailable. + +The method returns the unmodified status envelope. This slice does not poll, +persist, validate a terminal digest, append Succeeded, or interpret any +accepted/running envelope as measurement. Those lifecycle operations remain +issue #277 work and must bind the terminal result to the persisted request and +accepted receipt in one transaction. + +## Consequences + +The later durable worker can reuse the same client instead of introducing a +second HTTP path. Missing transport, malformed identity, and provider errors +cannot manufacture a result. + diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 9a65c2eb6..e555bec40 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -346,7 +346,7 @@ this file per §3.5 of the prior snapshot). | #271 | Evidence-honest knowledge-cutoff scope on Global Ask | Ask stack | | #272 | Verify Global Ask KG/ontology/semantic claims with public SearXNG evidence | Ask stack | | #274 | Persist and explain Event Lineage channel evidence | #387 | -| #277 | TEPP: persist accepted receipts, poll completed results, keep measurement authority distinct | #468, #417 | +| #277 | TEPP: persist accepted receipts, poll completed results, keep measurement authority distinct; TEPP PR #157 is merged and this exact head adds the fail-closed status/read client boundary, while durable polling and terminal persistence remain open | #468, #417, ADR 0217 | | #280 | Full project-lifecycle history and handover intervals | Tracked with issue #284; no active delivery PR confirmed | | #284 | Authoritative lifecycle ingestion and idempotent reconciliation | No active delivery PR confirmed | | #289 | Activate the optional lineage LLM channel through a bounded asynchronous rebuild | #434 | diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index 7dbfd886f..d76ca1061 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -8,12 +8,9 @@ lineage scores as TEPP's calibrated psychometric measurement (they answer different questions -- see docs/lineage-bi-research-notes.md). -TEPP does not expose a live HTTP endpoint yet (as of this writing it is -Rust-crate-only; see ``docs/API_CONTRACT.md`` in that repo). This client -builds and validates the exact wire shape TEPP has published -(``schemas/analysis_run_request_v1.json``) so wiring in a real transport is -a one-line change (:meth:`TeppClient.__init__`'s ``transport`` argument) once -that endpoint exists, instead of a redesign. +TEPP publishes versioned submit and status/read contracts. This client keeps +those transports separate so an accepted receipt cannot be mistaken for a +terminal measurement result. """ from __future__ import annotations @@ -35,6 +32,11 @@ def _no_transport(request: dict[str, Any]) -> dict[str, Any]: ) +def _no_status_transport(run_id: str) -> dict[str, Any]: + """Fail closed when no TEPP status/read transport is configured.""" + raise TeppNotAvailable("TEPP status transport unavailable") + + @dataclass(frozen=True) class AnalysisRunRequest: """Mirrors TEPP's ``schemas/analysis_run_request_v1.json`` exactly. @@ -65,19 +67,22 @@ def to_json(self) -> dict[str, Any]: class TeppClient: - """Submits :class:`AnalysisRunRequest` through a pluggable transport. - - The default transport always raises :class:`TeppNotAvailable` -- this - class exists so the rest of LineageWeave can be written against a - stable interface today, and gains a real TEPP integration by supplying - a ``transport`` (an HTTP POST to TEPP's future ``/v1/analysis-runs``, or - an in-process call into the ``tepp_api`` Rust crate via FFI) without - touching any other module. - """ + """Submit and read TEPP analysis runs through separate transports.""" - 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()) + + def get_analysis_run_status(self, run_id: str) -> dict[str, Any]: + """Read TEPP's status envelope for one opaque remote run id.""" + if not isinstance(run_id, str) or not run_id.strip(): + raise ValueError("run_id must be a non-empty string") + return self._status_transport(run_id.strip()) diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index 01fc91a94..97c4ee9d6 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -35,6 +35,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, match="status transport unavailable"): + client.get_analysis_run_status("remote-run-1") def test_custom_transport_receives_the_exact_wire_payload() -> None: @@ -86,6 +88,42 @@ def fake_post_json( assert received["service_peer_name"] == "tepp" +def test_configured_transport_reads_opaque_remote_run_status( + monkeypatch: pytest.MonkeyPatch, +) -> None: + received = {} + + def fake_get_json(url: str, **kwargs) -> dict: + received.update(url=url, **kwargs) + return {"contract_version": 1, "run_state": "running"} + + monkeypatch.setattr("backend.app.analysis_run_start.get_json", fake_get_json) + client = configured_tepp_client( + "https://tepp.example/v1/analysis-runs", + api_key="runtime-only", + ) + + status = client.get_analysis_run_status("remote/run 1") + + assert status["run_state"] == "running" + assert received == { + "url": "https://tepp.example/v1/analysis-runs/remote%2Frun%201", + "headers": { + "tepp-consumer": "lineageweave", + "tepp-contract-version": "1", + "authorization": "Bearer runtime-only", + }, + "timeout": 30.0, + "service_peer_name": "tepp", + } + + +@pytest.mark.parametrize("run_id", ["", " ", None]) +def test_status_read_rejects_missing_remote_run_identity(run_id) -> None: + with pytest.raises(ValueError, match="run_id must be a non-empty string"): + TeppClient().get_analysis_run_status(run_id) + + def test_configured_transport_hides_raw_provider_exception_chain( monkeypatch: pytest.MonkeyPatch, ) -> None: From 10bee7d0cfe10149a01d6bfeab8d42cf711c2579 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 04:21:09 +0900 Subject: [PATCH 2/5] fix(tepp): share the published contract version --- backend/app/analysis_run_start.py | 9 +++++++-- lineageweave/tepp_client.py | 4 +++- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 82cd1652b..49c63e511 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -39,7 +39,12 @@ from lineageweave.http_client import HttpClientError, get_json, 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 ( + ANALYSIS_RUN_CONTRACT_VERSION, + AnalysisRunRequest, + TeppClient, + TeppNotAvailable, +) _LINEAGE_KIND = "analysis_run_lineage" _TEPP_KIND = "analysis_run_tepp" @@ -190,7 +195,7 @@ def status_transport(run_id: str) -> dict[str, Any]: try: headers = { "tepp-consumer": "lineageweave", - "tepp-contract-version": "1", + "tepp-contract-version": str(ANALYSIS_RUN_CONTRACT_VERSION), } if api_key.strip(): headers["authorization"] = f"Bearer {api_key}" diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index d76ca1061..8f4c6c0d7 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -18,6 +18,8 @@ from dataclasses import dataclass from typing import Any, Callable +ANALYSIS_RUN_CONTRACT_VERSION = 1 + class TeppNotAvailable(RuntimeError): """Raised by the default transport: TEPP has no live REST API yet.""" @@ -51,7 +53,7 @@ class AnalysisRunRequest: knowledge_cutoff: str model_contract_version: str output_profile: str - contract_version: int = 1 + contract_version: int = ANALYSIS_RUN_CONTRACT_VERSION def to_json(self) -> dict[str, Any]: """Serialize the accepted TEPP result into its wire representation.""" From 0afbdd5ba7ed15fc93b1fe77e806bcc820bf25f6 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 04:24:21 +0900 Subject: [PATCH 3/5] fix(tepp): preserve opaque status run identities --- lineageweave/tepp_client.py | 2 +- tests/test_tepp_client.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index 8f4c6c0d7..7f18ff09a 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -87,4 +87,4 @@ def get_analysis_run_status(self, run_id: str) -> dict[str, Any]: """Read TEPP's status envelope for one opaque remote run id.""" if not isinstance(run_id, str) or not run_id.strip(): raise ValueError("run_id must be a non-empty string") - return self._status_transport(run_id.strip()) + return self._status_transport(run_id) diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index 97c4ee9d6..89182fbda 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -103,11 +103,11 @@ def fake_get_json(url: str, **kwargs) -> dict: api_key="runtime-only", ) - status = client.get_analysis_run_status("remote/run 1") + status = client.get_analysis_run_status(" remote/run 1 ") assert status["run_state"] == "running" assert received == { - "url": "https://tepp.example/v1/analysis-runs/remote%2Frun%201", + "url": "https://tepp.example/v1/analysis-runs/%20remote%2Frun%201%20", "headers": { "tepp-consumer": "lineageweave", "tepp-contract-version": "1", From 4158da86f4b8658c1a019a84ebb7b2086635e083 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 04:25:57 +0900 Subject: [PATCH 4/5] fix(tepp): leave unpublished status route unavailable --- backend/app/analysis_run_start.py | 23 ++-------------- ...0217-tepp-terminal-status-read-boundary.md | 15 ++++++----- tests/test_tepp_client.py | 27 +++++-------------- 3 files changed, 16 insertions(+), 49 deletions(-) diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 49c63e511..2597d73c2 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -16,7 +16,6 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import Any -from urllib.parse import quote from uuid import UUID import asyncpg @@ -36,7 +35,7 @@ records_from_source_posts, ) from lineageweave.adjudication_client import AdjudicationClient -from lineageweave.http_client import HttpClientError, get_json, post_json +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 ( @@ -190,25 +189,7 @@ def transport(payload: dict[str, Any]) -> dict[str, Any]: # message stays generic, never the raw provider exception text. raise TeppNotAvailable("TEPP transport unavailable") from exc - def status_transport(run_id: str) -> dict[str, Any]: - """GET TEPP's request-bound status envelope without interpreting it.""" - try: - headers = { - "tepp-consumer": "lineageweave", - "tepp-contract-version": str(ANALYSIS_RUN_CONTRACT_VERSION), - } - if api_key.strip(): - headers["authorization"] = f"Bearer {api_key}" - return get_json( - f"{url.rstrip('/')}/{quote(run_id, safe='')}", - headers=headers, - timeout=30.0, - service_peer_name="tepp", - ) - except (HttpClientError, OSError, ValueError, TypeError) as exc: - raise TeppNotAvailable("TEPP status transport unavailable") from exc - - return TeppClient(transport=transport, status_transport=status_transport) + return TeppClient(transport=transport) def tepp_run_request( diff --git a/docs/adr/0217-tepp-terminal-status-read-boundary.md b/docs/adr/0217-tepp-terminal-status-read-boundary.md index 4d8dd0371..69dd70af2 100644 --- a/docs/adr/0217-tepp-terminal-status-read-boundary.md +++ b/docs/adr/0217-tepp-terminal-status-read-boundary.md @@ -14,10 +14,11 @@ bypassing the existing provider boundary. ## Decision TeppClient.get_analysis_run_status reads one opaque remote run identity -through a separately injected status transport. The configured HTTP client -derives the item URL from TEPP_TRANSPORT_URL, percent-encodes the opaque -identity, reuses the TEPP consumer/version headers, and fails closed behind -TeppNotAvailable. +through a separately injected status transport. TEPP PR #157 publishes wire +types but no executable HTTP status route. The configured HTTP client therefore +keeps status reads unavailable instead of deriving an item URL from the submit +collection URL. A later owning-repository route contract may inject a transport +without changing or locally interpreting the opaque identity. The method returns the unmodified status envelope. This slice does not poll, persist, validate a terminal digest, append Succeeded, or interpret any @@ -28,6 +29,6 @@ accepted receipt in one transaction. ## Consequences The later durable worker can reuse the same client instead of introducing a -second HTTP path. Missing transport, malformed identity, and provider errors -cannot manufacture a result. - +second client abstraction. Missing transport and malformed identity cannot +manufacture a result; route construction remains unavailable until TEPP owns +and publishes it. diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index 89182fbda..57c611302 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -88,34 +88,19 @@ def fake_post_json( assert received["service_peer_name"] == "tepp" -def test_configured_transport_reads_opaque_remote_run_status( - monkeypatch: pytest.MonkeyPatch, -) -> None: - received = {} +def test_injected_status_transport_receives_opaque_remote_run_id_unchanged() -> None: + received: list[str] = [] - def fake_get_json(url: str, **kwargs) -> dict: - received.update(url=url, **kwargs) + def fake_status_transport(run_id: str) -> dict: + received.append(run_id) return {"contract_version": 1, "run_state": "running"} - monkeypatch.setattr("backend.app.analysis_run_start.get_json", fake_get_json) - client = configured_tepp_client( - "https://tepp.example/v1/analysis-runs", - api_key="runtime-only", - ) + client = TeppClient(status_transport=fake_status_transport) status = client.get_analysis_run_status(" remote/run 1 ") assert status["run_state"] == "running" - assert received == { - "url": "https://tepp.example/v1/analysis-runs/%20remote%2Frun%201%20", - "headers": { - "tepp-consumer": "lineageweave", - "tepp-contract-version": "1", - "authorization": "Bearer runtime-only", - }, - "timeout": 30.0, - "service_peer_name": "tepp", - } + assert received == [" remote/run 1 "] @pytest.mark.parametrize("run_id", ["", " ", None]) From 92534118d2f202c59518092664a872260d5d8d8a Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 04:32:09 +0900 Subject: [PATCH 5/5] refactor(tepp): drop unused contract import --- backend/app/analysis_run_start.py | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 2597d73c2..4b6476081 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -39,7 +39,6 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge from lineageweave.tepp_client import ( - ANALYSIS_RUN_CONTRACT_VERSION, AnalysisRunRequest, TeppClient, TeppNotAvailable,