Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.d/2.20.0-tepp-status-read-client.md
Original file line number Diff line number Diff line change
@@ -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.
23 changes: 21 additions & 2 deletions backend/app/analysis_run_start.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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}"
Comment thread
seonghobae marked this conversation as resolved.
Resolved
return get_json(
f"{url.rstrip('/')}/{quote(run_id, safe='')}",
Comment thread
seonghobae marked this conversation as resolved.
Resolved
headers=headers,
timeout=30.0,
service_peer_name="tepp",
)
Comment thread
seonghobae marked this conversation as resolved.
Resolved
except (HttpClientError, OSError, ValueError, TypeError) as exc:
raise TeppNotAvailable("TEPP status transport unavailable") from exc
Comment thread
seonghobae marked this conversation as resolved.
Resolved

return TeppClient(transport=transport, status_transport=status_transport)


def tepp_run_request(
Expand Down
33 changes: 33 additions & 0 deletions docs/adr/0217-tepp-terminal-status-read-boundary.md
Original file line number Diff line number Diff line change
@@ -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
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
**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.

2 changes: 1 addition & 1 deletion docs/product-technical-gap-baseline.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
37 changes: 21 additions & 16 deletions lineageweave/tepp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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())
Comment thread
seonghobae marked this conversation as resolved.
Resolved
Comment thread
seonghobae marked this conversation as resolved.
Resolved
38 changes: 38 additions & 0 deletions tests/test_tepp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
Loading