diff --git a/.env.example b/.env.example index 03b0fb45e..0f89fc44a 100644 --- a/.env.example +++ b/.env.example @@ -28,3 +28,7 @@ BACKEND_PORT=18420 ORCHESTRATOR_BASE_URL= ORCHESTRATOR_API_KEY= VISION_MODEL= + +# Optional. Empty = TEPP submit stays fail-closed (no invented theta). +# Point at a live TEPP HTTP root to POST /v1/analysis-runs. +TEPP_BASE_URL= diff --git a/AGENTS.md b/AGENTS.md index c790995c1..9ef184d7d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -97,3 +97,10 @@ pnpm run lint && pnpm run test && pnpm run build Do not weaken, skip, or `continue-on-error` a failing check -- fix the underlying cause or, for a genuine false positive in a third-party scanner, add a narrow, documented suppression referencing the specific finding. + +## TEPP measurement + +TEPP thetas come only from TEPP. `TeppClient.submit_fail_closed` and +`outbox:tepp` never store a score. Empty `TEPP_BASE_URL` is +`tepp_not_available`, not a guessed number. Do not copy IRT period-report +θ into a TEPP envelope. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f8a83ceb1..c91ecd7fa 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -65,7 +65,7 @@ flowchart LR | `embedding_client.py` | Pluggable text-embedding channel (`Null` default, `OpenAiCompatible` real impl) + `chunked_max_similarity` | | `adjudication_client.py` | Pluggable LLM-judgment channel (`Null` default, `ContextualOrchestrator` real impl) | | `image_content.py` | Pluggable vision channel: OCR + object recognition/tagging for embedded images (`Null` default, `OpenAiCompatibleVisionClient` real impl) | -| `tepp_client.py` | TEPP's published `AnalysisRunRequest` wire contract, pluggable transport | +| `tepp_client.py` | TEPP's published `AnalysisRunRequest` plus HTTP `/v1/analysis-runs` and a fail-closed envelope (ADR 0022). Never invents a theta. | | `rankweave_client.py` | Fail-closed RankWeave ranking port (`weighted_reciprocal_rank_fuse` in-process; never invent a fused score or a theta) | | `reconstruct.py` | The pipeline: group → candidate window → score → fuse → thread | | `lineage_persistence.py` | Flattens reconstruct trees into `post_lineage_edge` row specs (parent, child, fused_score) | @@ -112,12 +112,12 @@ flowchart LR `ponytail`-tagged in `reconstruct.py`: keeps per-group cost `O(n * window)` instead of `O(n^2)` for large groups; raise it if recall against a labeled set ever shows true parents falling outside the window. -- **TEPP is a wire contract, not an import.** `tepp_client.py`'s default - transport raises `TeppNotAvailable` rather than silently no-op'ing, - because TEPP has no live HTTP endpoint yet; the shape is validated - (`AnalysisRunRequest.to_json()` mirrors TEPP's published JSON Schema - exactly, `additionalProperties: false` and all) so wiring in a real - transport is additive, not a rewrite. +- **TEPP is a wire contract, not an invented score.** `tepp_client.py` + POSTs the published `AnalysisRunRequest` to + `{TEPP_BASE_URL}/v1/analysis-runs` when that URL is set. Empty or + failing TEPP returns a fail-closed envelope (`tepp_not_available`) + and an `outbox:tepp` Valkey row (ADR 0022 / 0023). The envelope has + no theta. - **RankWeave is an in-process library, not an HTTP host.** `rankweave_client.py`'s default transport raises `RankWeaveNotAvailable`. `GET /api/rankings` then returns diff --git a/CHANGELOG.d/0.79.0-tepp-fail-closed.md b/CHANGELOG.d/0.79.0-tepp-fail-closed.md new file mode 100644 index 000000000..a0852b049 --- /dev/null +++ b/CHANGELOG.d/0.79.0-tepp-fail-closed.md @@ -0,0 +1,6 @@ +# 0.79.0 — TEPP HTTP + fail-closed outbox + +## Added + +- Request TEPP measurement from Period reports. Missing TEPP stays + fail-closed on the Valkey outbox. No invented theta. diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bfcaa28f..1c8c3768e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,19 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.79.0] - 2026-08-17 + +### Added + +- TEPP HTTP port and a fail-closed submit envelope (ADR 0022 / 0023). + After `make seed`, Period reports shows the last Valkey outbox + row: TEPP is not available and no score was invented. post_admin + can Request TEPP measurement; LineageWeave POSTs the published + `AnalysisRunRequest` to `TEPP_BASE_URL/v1/analysis-runs` when that + URL is set, otherwise the same fail-closed envelope. Thetas still + come only from TEPP or from `calibrate_period_report` on the IRT + panel -- never copied into the TEPP envelope. + ## [0.75.0] - 2026-08-17 ### Added diff --git a/backend/app/config.py b/backend/app/config.py index 68cad3439..8b1a17d43 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -52,6 +52,9 @@ class Settings: # RankWeaveNotAvailable -- never invent a fused score. Default false # uses the in-process library already required by reconstruct.py. rankweave_disabled: bool + # Target TEPP HTTP root (ADR 0022). Empty = crate-only default + # transport; submit stays fail-closed and never invents a theta. + tepp_base_url: str @property def keycloak_jwks_uri(self) -> str: @@ -88,4 +91,5 @@ def load_settings() -> Settings: .strip() .lower() in {"1", "true", "yes", "on"}, + tepp_base_url=os.environ.get("TEPP_BASE_URL", ""), ) diff --git a/backend/app/main.py b/backend/app/main.py index 27f679119..089258b07 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -21,6 +21,7 @@ from contextlib import asynccontextmanager from typing import Any +from uuid import uuid4 import asyncpg import redis.asyncio as redis @@ -55,6 +56,7 @@ from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient from lineageweave.rankweave_client import build_rankweave_client +from lineageweave.tepp_client import AnalysisRunRequest, client_from_base_url, submit_fail_closed from backend.app.activity_stream import ( create_valkey_client, @@ -83,6 +85,7 @@ parse_period_code, rebuild_period_reports, ) +from backend.app.tepp_outbox import list_tepp_outbox, publish_tepp_outbox from backend.app.relation_verification_ingestion import verify_post_relations from backend.app.issue_ticket_ingestion import ( create_ticket, @@ -1150,3 +1153,57 @@ async def read_rankings( return _rankweave_client().as_api_payload( posts, can_see_post=lambda _row: True ) +class TeppAnalysisRunRequest(BaseModel): + snapshot_id: str + knowledge_cutoff: str + output_profile: str = "graphml" + idempotency_key: str | None = None + model_contract_version: str = "v1" + + +def _tenant_workspace_id(account: CurrentAccount) -> str: + """Stable tenant identity from an affiliated corp, never a guessed score.""" + if account.corporate_entity_ids: + return sorted(account.corporate_entity_ids)[0] + return account.user_account_id + + +@app.get("/api/tepp/outbox") +async def read_tepp_outbox( + account: CurrentAccount = Depends(get_current_account), + valkey: redis.Redis = Depends(get_valkey), +) -> dict[str, Any]: + """Recent TEPP submit envelopes. post_read. No theta is stored.""" + _require_post_read(account) + events = await list_tepp_outbox(valkey) + return {"events": events} + + +@app.post("/api/tepp/analysis-runs") +async def submit_tepp_analysis_run( + body: TeppAnalysisRunRequest, + account: CurrentAccount = Depends(get_current_account), + valkey: redis.Redis = Depends(get_valkey), +) -> dict[str, Any]: + """Submit TEPP's published request. post_admin. Fail-closed, no invented theta.""" + _require_post_admin(account) + snapshot_id = body.snapshot_id.strip() + knowledge_cutoff = body.knowledge_cutoff.strip() + output_profile = body.output_profile.strip() + if not snapshot_id or not knowledge_cutoff or not output_profile: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "snapshot_id, knowledge_cutoff, and output_profile are required") + idempotency_key = (body.idempotency_key or "").strip() or str(uuid4()) + request = AnalysisRunRequest( + idempotency_key=idempotency_key, + tenant_workspace_id=_tenant_workspace_id(account), + snapshot_id=snapshot_id, + knowledge_cutoff=knowledge_cutoff, + model_contract_version=(body.model_contract_version or "v1").strip() or "v1", + output_profile=output_profile, + ) + envelope = submit_fail_closed(client_from_base_url(load_settings().tepp_base_url), request) + event_id = await publish_tepp_outbox(valkey, envelope, account.user_account_id) + payload = envelope.to_json() + payload["event_id"] = event_id + return payload + diff --git a/backend/app/tepp_outbox.py b/backend/app/tepp_outbox.py new file mode 100644 index 000000000..c623dab0c --- /dev/null +++ b/backend/app/tepp_outbox.py @@ -0,0 +1,65 @@ +"""Valkey outbox for TEPP submit envelopes (ADR 0023). + +The stream is the durable attempt log -- not a second score table. +Fields are outcome + next action + the published request identity. +No theta is written. +""" + +from __future__ import annotations + +from typing import Any + +from lineageweave.fail_closed import FailClosedEnvelope + +TEPP_OUTBOX_STREAM = "outbox:tepp" + + +def outbox_fields(envelope: FailClosedEnvelope, actor_account_id: str) -> dict[str, str]: + """String fields Valkey can XADD. Measurement keys are never included.""" + request = envelope.request or {} + return { + "channel_code": envelope.channel_code, + "outcome_code": envelope.outcome_code, + "next_action": envelope.next_action, + "actor_account_id": str(actor_account_id), + "idempotency_key": str(request.get("idempotency_key", "")), + "snapshot_id": str(request.get("snapshot_id", "")), + "knowledge_cutoff": str(request.get("knowledge_cutoff", "")), + "tenant_workspace_id": str(request.get("tenant_workspace_id", "")), + } + + +def publish_tepp_outbox_sync(client: Any, envelope: FailClosedEnvelope, actor_account_id: str) -> str | None: + """Sync ``XADD`` for ``make seed``. Skips a matching idempotency key.""" + fields = outbox_fields(envelope, actor_account_id) + existing = client.xrevrange(TEPP_OUTBOX_STREAM, count=50) + if any(row.get("idempotency_key") == fields["idempotency_key"] for _entry_id, row in existing): + return None + return client.xadd(TEPP_OUTBOX_STREAM, fields, maxlen=1000, approximate=True) + + +async def publish_tepp_outbox(client: Any, envelope: FailClosedEnvelope, actor_account_id: str) -> str: + """``XADD`` one fail-closed (or accepted) TEPP attempt.""" + return await client.xadd( + TEPP_OUTBOX_STREAM, + outbox_fields(envelope, actor_account_id), + maxlen=1000, + approximate=True, + ) + + +async def list_tepp_outbox(client: Any, count: int = 20) -> list[dict[str, Any]]: + """Newest first. Payloads are labels and request identity only.""" + entries = await client.xrevrange(TEPP_OUTBOX_STREAM, count=count) + return [ + { + "event_id": entry_id, + "channel_code": fields.get("channel_code", ""), + "outcome_code": fields.get("outcome_code", ""), + "next_action": fields.get("next_action", ""), + "idempotency_key": fields.get("idempotency_key", ""), + "snapshot_id": fields.get("snapshot_id", ""), + "knowledge_cutoff": fields.get("knowledge_cutoff", ""), + } + for entry_id, fields in entries + ] diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index c2f3994d5..5a5420f53 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -31,3 +31,8 @@ def test_rankweave_disabled_defaults_off(monkeypatch) -> None: def test_rankweave_disabled_flag_is_opt_in(monkeypatch) -> None: monkeypatch.setenv("RANKWEAVE_DISABLED", "1") assert load_settings().rankweave_disabled is True + + +def test_tepp_base_url_defaults_empty(monkeypatch) -> None: + monkeypatch.delenv("TEPP_BASE_URL", raising=False) + assert load_settings().tepp_base_url == "" diff --git a/docker-compose.yml b/docker-compose.yml index 5087366b4..dd3eb33f7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -105,6 +105,8 @@ services: ORCHESTRATOR_API_KEY: ${ORCHESTRATOR_API_KEY:-} VISION_MODEL: ${VISION_MODEL:-} SEARXNG_BASE_URL: http://searxng:8080 + # Empty = TEPP submit stays fail-closed (ADR 0022). Never invents a theta. + TEPP_BASE_URL: ${TEPP_BASE_URL:-} ports: - "${BACKEND_PORT:-18420}:8000" depends_on: diff --git a/docs/adr/0022-tepp-http-fail-closed.md b/docs/adr/0022-tepp-http-fail-closed.md new file mode 100644 index 000000000..fe8dcf8cc --- /dev/null +++ b/docs/adr/0022-tepp-http-fail-closed.md @@ -0,0 +1,37 @@ +# ADR 0022 — TEPP HTTP port stays fail-closed + +**Decision status:** Accepted +**Date:** 2026-08-17 + +## Context + +`TeppClient` already builds TEPP's published `AnalysisRunRequest` +(`schemas/analysis_run_request_v1.json`). Protected TEPP main is still +crate-only; the target HTTP resource is `POST /v1/analysis-runs` +(TEPP `docs/API_CONTRACT.md`). LineageWeave must not invent a theta +when that service is unset or unreachable, and must not treat IRT +period-report θ as a TEPP measurement. + +## Decision + +1. When `TEPP_BASE_URL` is set, POST the published request through + `http_client.post_json` to `{TEPP_BASE_URL}/v1/analysis-runs`. +2. When it is empty, keep the crate-only default transport. +3. Every caller uses `submit_fail_closed`, which returns a + `FailClosedEnvelope` (`accepted` / `tepp_not_available` / + `tepp_transport_failed`) with a customer-actionable `next_action`. + The envelope has no theta field; a provider `theta` key is stripped. +4. `POST /api/tepp/analysis-runs` is `post_admin`. Unavailable is a + 200 envelope, not a fabricated score. + +## Consequences + +- The buyer can request a TEPP run after `make seed` and read why + none exists yet. +- IRT `calibrate_period_report` thetas stay on the period-report + panel and are never copied into the TEPP envelope. + +## Related + +Outbox persistence is [ADR 0023](0023-tepp-valkey-outbox.md). +TEPP contract: ContextualWisdomLab/TEPP `docs/API_CONTRACT.md`. diff --git a/docs/adr/0023-tepp-valkey-outbox.md b/docs/adr/0023-tepp-valkey-outbox.md new file mode 100644 index 000000000..d6b861fbc --- /dev/null +++ b/docs/adr/0023-tepp-valkey-outbox.md @@ -0,0 +1,30 @@ +# ADR 0023 — Valkey outbox for TEPP envelopes + +**Decision status:** Accepted +**Date:** 2026-08-17 + +## Context + +A fail-closed TEPP submit (ADR 0022) is lost if it only lives in the +HTTP response. Period-report IRT rows are the wrong table: they store +LineageWeave's own GRM/GPCM θ, not a TEPP run. An analysis-run +registry (Milestone 2.1 on the stacked branch) is a different +authority and must not be reimplemented here. + +Valkey is already the product event queue (`activity:{post_id}`). + +## Decision + +1. Each TEPP submit `XADD`s onto `outbox:tepp` with outcome, next + action, and the published request identity (idempotency key, + snapshot, cutoff). No theta field. +2. `GET /api/tepp/outbox` returns the newest events for `post_read`. +3. `make seed` writes one crate-only fail-closed row so the home + panel names the next action after a fresh stack. + +## Consequences + +- The buyer sees the last TEPP attempt above the period-report + actions without opening a second product. +- A later analysis-run registry can drain this outbox; it must not + invent a score while doing so. diff --git a/frontend/package.json b/frontend/package.json index 575b7c586..cde22610e 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.75.0", + "version": "0.79.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index a32a26403..755f0dcb8 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -89,6 +89,25 @@ describe("App, authenticated", () => { let nextTicketId = 1; const events: { event_id: string; event_type: string; actor_account_id: string; summary: string }[] = []; let nextEventId = 1; + const teppEvents: { + event_id: string; + channel_code: string; + outcome_code: string; + next_action: string; + idempotency_key: string; + snapshot_id: string; + knowledge_cutoff: string; + }[] = [ + { + event_id: "tepp-1", + channel_code: "tepp", + outcome_code: "tepp_not_available", + next_action: "Configure TEPP_BASE_URL to submit a real analysis run. No TEPP score was invented.", + idempotency_key: "seed-tepp-2026-w02", + snapshot_id: "process_unit:2026-W02", + knowledge_cutoff: "2026-W02", + }, + ]; const fetchMock = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { const url = String(input); @@ -103,6 +122,28 @@ describe("App, authenticated", () => { }), ); } + if (url.endsWith("/api/tepp/outbox") && method === "GET") { + return Promise.resolve(jsonResponse({ events: teppEvents })); + } + if (url.endsWith("/api/tepp/analysis-runs") && method === "POST") { + teppEvents.unshift({ + event_id: "tepp-live", + channel_code: "tepp", + outcome_code: "tepp_not_available", + next_action: "Configure TEPP_BASE_URL to submit a real analysis run. No TEPP score was invented.", + idempotency_key: "live-1", + snapshot_id: "process_unit:2026-W02", + knowledge_cutoff: "2026-W02", + }); + return Promise.resolve( + jsonResponse({ + channel_code: "tepp", + outcome_code: "tepp_not_available", + next_action: teppEvents[0].next_action, + event_id: "tepp-live", + }), + ); + } if (url.endsWith("/api/lineage/rebuild") && method === "POST") { return Promise.resolve(jsonResponse({ edge_count: 4 })); } @@ -1339,6 +1380,9 @@ describe("App, authenticated", () => { expect(screen.getByText(/TEST-PU-REPORT/)).toBeInTheDocument(); expect(screen.getAllByText("shared metric").length).toBeGreaterThan(0); expect(screen.getAllByText(/CAT: sales-lead I=0\.70/).length).toBeGreaterThan(0); + const teppOutcome = screen.getByLabelText("TEPP submit outcome"); + expect(teppOutcome).toHaveTextContent("No TEPP score was invented"); + expect(teppOutcome).not.toHaveTextContent("θ"); expect(screen.getByRole("button", { name: /open report period 2026-W03/i })).toHaveTextContent( "vs 2026-W02: +0.92", ); @@ -1442,4 +1486,19 @@ describe("App, authenticated", () => { ), ); }); + + it("lets post_admin request a TEPP measurement without inventing a theta", async () => { + const fetchMock = stubBackend({ admin: true }); + render(); + + await userEvent.click(await screen.findByRole("button", { name: /request tepp measurement/i })); + await waitFor(() => + expect(fetchMock).toHaveBeenCalledWith( + expect.stringContaining("/api/tepp/analysis-runs"), + expect.objectContaining({ method: "POST" }), + ), + ); + expect(screen.getByLabelText("TEPP submit outcome")).toHaveTextContent("No TEPP score was invented"); + expect(screen.getByLabelText("TEPP submit outcome")).not.toHaveTextContent("θ"); + }); }); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 6056e5eb4..c5724bade 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -28,8 +28,10 @@ import { fetchRankings, fetchRelatedEntity, fetchRelatedKeymen, + fetchTeppOutbox, rebuildLineage, rebuildPeriodReports, + submitTeppAnalysisRun, updateTicketStatus, verifyPostRelations, type ActivityEvent, @@ -52,6 +54,7 @@ import { type PostSummary, type RankingList, type RelatedNode, + type TeppOutboxEvent, type VocEvidence, } from "./api"; import { LineageDag } from "./LineageDag"; @@ -1435,8 +1438,10 @@ function ReportsPanel({ const [payload, setPayload] = useState(null); const [index, setIndex] = useState(null); const [comparison, setComparison] = useState(null); + const [teppOutbox, setTeppOutbox] = useState([]); const [error, setError] = useState(null); const [rebuilding, setRebuilding] = useState(false); + const [requestingTepp, setRequestingTepp] = useState(false); const groupingLabels: Record = { process_unit: "Process unit", @@ -1450,11 +1455,13 @@ function ReportsPanel({ fetchPeriodReports(accessToken, grouping, period), fetchPeriodReportIndex(accessToken, grouping), fetchPeriodComparison(accessToken, period), + fetchTeppOutbox(accessToken), ]) - .then(([reports, periods, compared]) => { + .then(([reports, periods, compared, tepp]) => { setPayload(reports); setIndex(periods); setComparison(compared); + setTeppOutbox(tepp.events); }) .catch((err) => setError(String(err))); }, [accessToken, grouping, period]); @@ -1464,14 +1471,16 @@ function ReportsPanel({ setError(null); try { await rebuildPeriodReports(accessToken, grouping, period); - const [reports, periods, compared] = await Promise.all([ + const [reports, periods, compared, tepp] = await Promise.all([ fetchPeriodReports(accessToken, grouping, period), fetchPeriodReportIndex(accessToken, grouping), fetchPeriodComparison(accessToken, period), + fetchTeppOutbox(accessToken), ]); setPayload(reports); setIndex(periods); setComparison(compared); + setTeppOutbox(tepp.events); } catch (err) { setError(String(err)); } finally { @@ -1479,6 +1488,19 @@ function ReportsPanel({ } } + async function handleRequestTepp() { + setRequestingTepp(true); + setError(null); + try { + await submitTeppAnalysisRun(accessToken, `${grouping}:${period}`, period); + setTeppOutbox((await fetchTeppOutbox(accessToken)).events); + } catch (err) { + setError(String(err)); + } finally { + setRequestingTepp(false); + } + } + return (
@@ -1488,7 +1510,17 @@ function ReportsPanel({ {rebuilding ? "Calibrating..." : "Rebuild report"} )} + {canRebuild && ( + + )}
+ {teppOutbox[0] && ( +

+ TEPP {teppOutbox[0].outcome_code.replace(/_/g, " ")}: {teppOutbox[0].next_action} +

+ )}