diff --git a/CHANGELOG.md b/CHANGELOG.md index 87383d244..83d301952 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,22 @@ All notable changes to this project are documented here. Format follows ### Added +- Registered the `analysis_run_topic_lineage` analysis-run kind (migrations + 0131/0132, ADR 0132), the LineageWeave-side consumption boundary for + TEPP's Temporal Relational Shared-Latent Topic Measurement (TRSL-TM, + TEPP ADR 0012) and CHRONOS/TDT event-intelligence status (TEPP ADR 0016). + It mirrors the existing TEPP measurement path exactly: submits through + `tepp_client`, fails closed (`tepp_not_available` / + `tepp_result_not_persisted`) until TEPP publishes a completed envelope, + and never computes a topic identity or event prediction locally. + `make seed` now also writes a Demo Corp topic-lineage run alongside the + existing lineage/TEPP/period-report rows. +- `EvidenceStatusMark`, a reusable evidence/inference/prediction status + badge (ADR 0132 decision 5, TEPP ADR 0016) distinguishing status by label + text and glyph shape, not color alone. Ships ahead of the Event Lineage + DAG topic-thread wiring it is designed for, so review and Storybook + coverage (`Analysis/EvidenceStatusMark`) are available now; it is + presentational only and never infers a status itself. - Opening a post with persisted image-region evidence now shows each region's bounding range beside its caption, OCR, and tags (ADR 0155). After `make seed`, a synthetic process-diagram region reads **Region location: diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index 5d074d88f..7335f6fb9 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -34,11 +34,13 @@ _LINEAGE_RUN_KIND = "analysis_run_lineage" _TEPP_RUN_KIND = "analysis_run_tepp" _REPORT_RUN_KIND = "analysis_run_report" +_TOPIC_LINEAGE_RUN_KIND = "analysis_run_topic_lineage" _CORPORATE_SCOPE = "analysis_scope_corporate_entity" _CAPTURE_CONTRACT_VERSION = "analysis-run-capture-v1" _KIND_SCHEMA_VERSION = { "analysis_run_lineage": "lineage-run-v1", "analysis_run_tepp": "tepp-run-v1", + "analysis_run_topic_lineage": "topic-lineage-run-v1", } _RUN_LIST_SQL = f""" @@ -405,6 +407,21 @@ async def fetch_visible_analysis_run( if digest is not None: detail["reconstruction_result_sha256"] = digest detail["reconstructed_edges"] = edges + if row["run_kind_code"] == _TOPIC_LINEAGE_RUN_KIND: + topic_result = await conn.fetchrow( + """ + select result_json, result_sha256 + from analysis_run_topic_lineage_result + where analysis_run_id = $1 + """, + analysis_run_id, + ) + if topic_result is not None: + envelope = topic_result["result_json"] + detail["topic_lineage_result"] = ( + json.loads(envelope) if isinstance(envelope, str) else envelope + ) + detail["topic_lineage_result_sha256"] = topic_result["result_sha256"] return detail @@ -613,11 +630,12 @@ def __init__(self, status_code: int, detail: str) -> None: def _require_lineage_create_kind(run_kind_code: str) -> None: - """Reject TEPP and report writes so this path cannot fake those products. + """Reject TEPP, topic-lineage, and report writes so this path cannot fake those products. - TEPP stays a ``tepp_client`` wire path. Period reports stay on the - Reports panel rebuild. A Pending TEPP row that never called the - transport is a fabricated measurement request. + TEPP and topic-lineage stay ``tepp_client`` wire paths (ADR 0022 / + ADR 0132). Period reports stay on the Reports panel rebuild. A Pending + TEPP or topic-lineage row that never called the transport is a + fabricated measurement request. """ if run_kind_code == _TEPP_RUN_KIND: raise AnalysisRunCreateError( @@ -625,6 +643,12 @@ def _require_lineage_create_kind(run_kind_code: str) -> None: "Connect a TEPP transport from a Failed TEPP row; this endpoint " "does not invent a measurement.", ) + if run_kind_code == _TOPIC_LINEAGE_RUN_KIND: + raise AnalysisRunCreateError( + 422, + "Connect a TEPP transport from a Failed topic-lineage row; this " + "endpoint does not invent a topic model.", + ) if run_kind_code == _REPORT_RUN_KIND: raise AnalysisRunCreateError( 422, diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 4b91a9990..c623fcdbc 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -37,12 +37,15 @@ _LINEAGE_KIND = "analysis_run_lineage" _TEPP_KIND = "analysis_run_tepp" _REPORT_KIND = "analysis_run_report" +_TOPIC_LINEAGE_KIND = "analysis_run_topic_lineage" _PENDING = "analysis_status_pending" _RUNNING = "analysis_status_running" _SUCCEEDED = "analysis_status_succeeded" _FAILED = "analysis_status_failed" _TEPP_MODEL_CONTRACT = "tepp-analysis-run-v1" _TEPP_OUTPUT_PROFILE = "calibrated_event_measurement" +_TOPIC_LINEAGE_MODEL_CONTRACT = "tepp-topic-lineage-v1" +_TOPIC_LINEAGE_OUTPUT_PROFILE = "topic_identity_lineage" class AnalysisRunStartError(AnalysisRunCreateError): @@ -69,11 +72,11 @@ def reconstruction_result_digest(edges: list[Edge]) -> str: def start_kind_rejection(run_kind_code: str) -> AnalysisRunStartError | None: """Return a 422 when start cannot run this kind. - Lineage reconstructs the frozen bag. TEPP submits through - ``tepp_client`` and never invents a theta. Period-report stays on - its own rebuild path. + Lineage reconstructs the frozen bag. TEPP and topic-lineage submit + through ``tepp_client`` and never invent a theta or a topic (ADR 0022 / + ADR 0132). Period-report stays on its own rebuild path. """ - if run_kind_code in {_LINEAGE_KIND, _TEPP_KIND}: + if run_kind_code in {_LINEAGE_KIND, _TEPP_KIND, _TOPIC_LINEAGE_KIND}: return None if run_kind_code == _REPORT_KIND: return AnalysisRunStartError( @@ -133,6 +136,34 @@ def tepp_run_request( ) +def topic_lineage_run_request( + *, + idempotency_key: str, + snapshot_sha256: str, + knowledge_cutoff: datetime, + corporate_entity_id: str, +) -> AnalysisRunRequest: + """Build TEPP's published request for a topic-lineage run (ADR 0132). + + Same wire shape as :func:`tepp_run_request` -- TEPP's + ``AnalysisRunRequest`` already carries no post body or fabricated + label -- only the model contract and output profile differ, selecting + TRSL-TM topic identity plus CHRONOS/TDT event-intelligence status + instead of calibrated psychometric measurement. + """ + cutoff = knowledge_cutoff + if cutoff.tzinfo is None: + cutoff = cutoff.replace(tzinfo=timezone.utc) + return AnalysisRunRequest( + idempotency_key=idempotency_key, + tenant_workspace_id=str(corporate_entity_id), + snapshot_id=snapshot_sha256, + knowledge_cutoff=cutoff.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + model_contract_version=_TOPIC_LINEAGE_MODEL_CONTRACT, + output_profile=_TOPIC_LINEAGE_OUTPUT_PROFILE, + ) + + def _tepp_submission( client: TeppClient, request: AnalysisRunRequest, @@ -168,6 +199,51 @@ def tepp_submit_outcome( return status_code, failure_code +def _topic_lineage_envelope_is_valid(envelope: dict[str, Any]) -> bool: + """Require TEPP's versioned topic-identity/CHRONOS-status contract (ADR 0132). + + ``_tepp_submission`` only checks that ``result`` is *a* dict -- a + ``completed`` envelope carrying the calibrated-measurement shape (or any + other unrelated payload) would pass it too, since both requests share the + same wire contract and differ only in ``model_contract_version`` / + ``output_profile``. This additionally requires TRSL-TM topic identity and + CHRONOS/TDT status, keyed by envelope version. + """ + result = envelope.get("result") + if not isinstance(result, dict): + return False + if type(result.get("envelope_version")) is not int: # bool is not a version + return False + if result["envelope_version"] != 1: + return False + topic_identity = result.get("topic_identity") + if not isinstance(topic_identity, (list, dict)) or not topic_identity: + return False + chronos_status = result.get("chronos_status") + if not isinstance(chronos_status, (list, dict, str)) or not chronos_status: + return False + return True + + +def topic_lineage_submit_outcome( + client: TeppClient, + request: AnalysisRunRequest, +) -> tuple[str, str, dict[str, Any] | None]: + """Submit through ``tepp_client`` and require the topic-lineage contract. + + Mirrors :func:`tepp_submit_outcome`, but a syntactically ``completed`` + envelope that omits the versioned topic-identity/CHRONOS-status contract + is also Failed (``tepp_topic_contract_unavailable``, ADR 0132 Decision + item 3), not silently persisted as a topic-lineage result. + """ + status_code, failure_code, envelope = _tepp_submission(client, request) + if status_code == _SUCCEEDED and not ( + envelope is not None and _topic_lineage_envelope_is_valid(envelope) + ): + return _FAILED, "tepp_topic_contract_unavailable", None + return status_code, failure_code, envelope + + async def _persist_tepp_result( conn: asyncpg.Connection, *, @@ -199,6 +275,42 @@ async def _persist_tepp_result( return True +async def _persist_topic_lineage_result( + conn: asyncpg.Connection, + *, + analysis_run_id: str, + envelope: dict[str, Any], +) -> bool: + """Persist only a validated, remote-completed topic-lineage envelope. + + Stores TEPP's TRSL-TM topic identity / CHRONOS status envelope + verbatim (ADR 0132); LineageWeave does not decompose or reinterpret + its evidence/inference/prediction fields here. + """ + remote_run_id = envelope.get("analysis_run_id") or envelope.get("run_id") + if not isinstance(remote_run_id, str) or not remote_run_id.strip(): + return False + result_json = json.dumps(envelope, separators=(",", ":"), sort_keys=True) + result_sha256 = hashlib.sha256(result_json.encode("utf-8")).hexdigest() + try: + async with conn.transaction(): + await conn.execute( + """ + insert into analysis_run_topic_lineage_result + (analysis_run_id, remote_run_id, result_json, result_sha256) + values ($1, $2, $3::jsonb, $4) + on conflict (analysis_run_id) do nothing + """, + analysis_run_id, + remote_run_id, + result_json, + result_sha256, + ) + except (asyncpg.PostgresError, TypeError, ValueError): + return False + return True + + def start_write_conflict_error() -> AnalysisRunStartError: """Next action when a concurrent start already wrote this run.""" return AnalysisRunStartError( @@ -631,6 +743,13 @@ async def deliver_queued_analysis_run( locked=outbox, tepp_client=tepp_client or TeppClient(), ) + elif outbox["work_kind_code"] == _TOPIC_LINEAGE_KIND: + await _deliver_topic_lineage_measurement( + conn, + analysis_run_id=analysis_run_id, + locked=outbox, + tepp_client=tepp_client or TeppClient(), + ) else: await _deliver_lineage_reconstruction( conn, @@ -789,3 +908,45 @@ async def _deliver_tepp_measurement( finished, failure_code, ) + + +async def _deliver_topic_lineage_measurement( + conn: asyncpg.Connection, + *, + analysis_run_id: str, + locked: asyncpg.Record, + tepp_client: TeppClient, +) -> None: + """Submit the frozen snapshot through ``tepp_client`` for topic-lineage. + + Mirrors :func:`_deliver_tepp_measurement` (ADR 0022) with the + topic-lineage model contract (ADR 0132). Never persists a locally + computed topic identity or CHRONOS/TDT event prediction. + """ + now = datetime.now(timezone.utc) + request = topic_lineage_run_request( + idempotency_key=str(locked["idempotency_key"]), + snapshot_sha256=str(locked["snapshot_sha256"]), + knowledge_cutoff=locked["knowledge_cutoff"], + corporate_entity_id=str(locked["corporate_entity_id"]), + ) + status_code, failure_code, envelope = topic_lineage_submit_outcome(tepp_client, request) + if status_code == _SUCCEEDED and envelope is not None: + if not await _persist_topic_lineage_result( + conn, + analysis_run_id=analysis_run_id, + envelope=envelope, + ): + status_code = _FAILED + failure_code = "tepp_result_not_persisted" + finished = datetime.now(timezone.utc) + if finished < now: + finished = now + await _append_status( + conn, + analysis_run_id, + await _next_status_ordinal(conn, analysis_run_id), + status_code, + finished, + failure_code, + ) diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 4cdf0729b..e9c66002d 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -117,6 +117,21 @@ _TENANT_SETTINGS_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" / "0103_tenant_settings.sql" ) +_TOPIC_LINEAGE_KIND_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0131_analysis_run_topic_lineage_kind.sql" +) +_TOPIC_LINEAGE_RESULT_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0132_analysis_run_topic_lineage_result.sql" +) +_TOPIC_LINEAGE_VALIDATE_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0204_validate_topic_lineage_kind.sql" +) _CHANNEL_WEIGHT_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" @@ -266,6 +281,9 @@ def seeded_db(demo_analyst_token): cur.execute(_PROJECT_BOUND_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text()) cur.execute(_TENANT_SETTINGS_MIGRATION.read_text()) + cur.execute(_TOPIC_LINEAGE_KIND_MIGRATION.read_text()) + cur.execute(_TOPIC_LINEAGE_RESULT_MIGRATION.read_text()) + cur.execute(_TOPIC_LINEAGE_VALIDATE_MIGRATION.read_text()) cur.execute(_CHANNEL_WEIGHT_MIGRATION.read_text()) cur.execute(_LEFTOVER_OBSERVED_EXPECTED_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_RANK_MIGRATION.read_text()) @@ -688,6 +706,82 @@ def test_analysis_runs_are_labeled_aggregates_and_hide_other_scopes( assert "postgresql://" not in str(body) assert "visible_posts" not in visible + +def test_topic_lineage_detail_returns_authoritative_envelope( + client, demo_analyst_token, seeded_db +) -> None: + """An authorized successful run exposes TEPP's opaque envelope and digest.""" + envelope = { + "status": "completed", + "analysis_run_id": "remote-topic-1", + "result": { + "envelope_version": 1, + "topic_identity": [{"topic_id": "synthetic-topic-1"}], + "chronos_status": "evidence", + }, + } + digest = "a" * 64 + with closing(psycopg2.connect(seeded_db["dsn"])) as conn, conn.cursor() as cur: + cur.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + select analysis_source_snapshot_id, 'analysis_run_topic_lineage', + 'synthetic-topic-detail', requested_by_account_id, + knowledge_cutoff, 'topic-lineage-run-v1', %s, %s, requested_at + from analysis_run where analysis_run_id = %s + returning analysis_run_id + """, + ("b" * 64, "c" * 40, seeded_db["visible_run_id"]), + ) + run_id = str(cur.fetchone()[0]) + cur.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values (%s, 'analysis_scope_corporate_entity', %s) + """, + (run_id, seeded_db["own_corp_id"]), + ) + for ordinal, status_code in enumerate( + ( + "analysis_status_pending", + "analysis_status_running", + "analysis_status_succeeded", + ), + start=1, + ): + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, %s, %s, + '2026-01-12T12:34:00Z'::timestamptz + interval '1 second' * %s) + """, + (run_id, ordinal, status_code, ordinal), + ) + cur.execute( + """ + insert into analysis_run_topic_lineage_result + (analysis_run_id, remote_run_id, result_json, result_sha256) + values (%s, 'remote-topic-1', %s::jsonb, %s) + """, + (run_id, __import__("json").dumps(envelope), digest), + ) + conn.commit() + + response = client.get( + f"/api/analysis-runs/{run_id}", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + + assert response.status_code == 200, response.text + assert response.json()["topic_lineage_result"] == envelope + assert response.json()["topic_lineage_result_sha256"] == digest + hidden = client.get( f"/api/analysis-runs/{seeded_db['hidden_run_id']}", headers={"Authorization": f"Bearer {demo_analyst_token}"}, diff --git a/docs/adr/0132-tepp-topic-lineage-consumption-boundary.md b/docs/adr/0132-tepp-topic-lineage-consumption-boundary.md new file mode 100644 index 000000000..f4fb00d28 --- /dev/null +++ b/docs/adr/0132-tepp-topic-lineage-consumption-boundary.md @@ -0,0 +1,187 @@ +# ADR 0132 — TEPP topic-lineage consumption boundary (TRSL-TM + CHRONOS/TDT) + +**Decision status:** Accepted +**Implementation maturity:** boundary-accepted; evidence/inference/prediction +mark primitive implemented (`frontend/src/components/EvidenceStatusMark.tsx`, +below); DAG topic-thread wiring and runtime remain open +**Date:** 2026-08-22 +**Depends on:** ADR 0022 (authorized TEPP start); ADR 0064 (lineage evidence +and tree assembly); ADR 0084 (research-grounded lineage and ontology policy) +**Refs:** TEPP ADR 0012 (Temporal Relational Shared-Latent Topic +Measurement); TEPP ADR 0016 (TDT, CHRONOS, and Event Ontology intelligence +boundary) + +## Context + +ADR 0084 already separates mention detection / lineage-instance +construction (LineageWeave's own evidence-fusion engine) from calibrated +measurement, which stays a TEPP wire-contract boundary that is never +reimplemented locally. That separation did not yet have a topic-identity +counterpart: `zcrht811_export_rows` board posts are scattered across time +with no thread connecting a post to the earlier/later posts that share its +underlying commercial topic (a competitor mention, a market trend, a sales +opportunity) the way a Git branch connects commits. + +TEPP's own accepted-target architecture already defines this problem +precisely: + +- TEPP ADR 0012 adopts **Temporal Relational Shared-Latent Topic + Measurement (TRSL-TM)**: one global topic identity set per modeled + period, where topics may be active, dormant, or reactivated over time + without losing identity, and topic birth/split/merge/retirement is an + explicit lineage extension — not an implicit side effect of fitting + unrelated time slices. +- TEPP ADR 0016 separates Event Ontology, TDT-style detection/tracking, and + CHRONOS-style reasoning (semantic/neural event-schema + extraction/prediction plus symbolic temporal-consistency reasoning). + Every output carries an evidence / inference / prediction status and + provenance; a prediction is never silently converted into historical + fact. + +TEPP is currently a foundation-slice (`crates/tepp_api` exposes only +`AnalysisRunRequest` / `AnalysisRunAccepted`; a completed-result contract is +still open — TEPP issue #156). Standing up a local topic model or event +predictor to fill that gap would repeat exactly the "invented psychometric +substitute" ADR 0084 and the tepp-readiness discipline already forbid. + +## Decision + +1. LineageWeave does not compute topic identity, topic + birth/split/merge/retirement, event-schema predictions, or temporal- + consistency verdicts locally. These remain TEPP's TRSL-TM (ADR 0012) and + CHRONOS/TDT (ADR 0016) computations, requested through `tepp_client` the + same way `analysis_run_start.py` requests a measurement run (ADR 0022). +2. A new analysis-run kind, topic-lineage, is added alongside the existing + lineage / TEPP / period-report kinds (ADR 0013 registry). Requesting it + builds a TEPP request payload from the run's idempotency key, snapshot + digest, knowledge cutoff, and corporate-entity workspace id — never post + bodies or a fabricated topic label — and submits through `TeppClient`. +3. An empty `TEPP_TRANSPORT_URL`, or a TEPP response that omits the + versioned topic-identity/CHRONOS-status envelope, appends Failed with a + machine-readable reason (`tepp_not_available` / + `tepp_topic_contract_unavailable`), mirroring ADR 0022. Failed is + terminal; the operator reconnects TEPP and retries the existing Failed row + through `POST /api/analysis-runs/{id}/start`. +4. When TEPP does publish the topic-identity and CHRONOS-status envelope, + LineageWeave persists it into a new topic-identity-thread projection + (3NF, two-word snake_case, partitioned by corporate-entity + observed + period to avoid a hot partition on the shared post table) that links + existing posts to TEPP topic ids and carries each edge's evidence / + inference / prediction status verbatim. This projection extends — it + does not replace — the existing evidence-fusion lineage tree (ADR + 0064/0084); a post can appear in both the fusion-based lineage DAG and a + TEPP topic thread, and the UI keeps the two visually distinct. +5. The frontend Event Lineage DAG gains a topic-thread overlay: nodes/edges + sourced from TEPP render with a distinct visual channel (color/pattern + token, not color alone, for accessibility) per evidence / inference / + prediction status, and render nothing (not a placeholder guess) for a + topic-lineage run that is Pending or Failed. Storybook stories cover + Pending, Failed (`tepp_not_available`, `tepp_topic_contract_unavailable`), + and each CHRONOS status; a Playwright e2e spec exercises the golden path + once a topic-identity envelope exists and the fail-closed path when it + does not. + +```mermaid +sequenceDiagram + participant Operator + participant API + participant TeppClient + participant Registry + Operator->>API: POST /api/analysis-runs (kind=topic_lineage) + API-->>Operator: 422; no Pending topic-lineage row is created + Note over Operator,Registry: Existing Failed rows come from the governed seed/import boundary + Operator->>API: Connect TEPP; POST /api/analysis-runs/{id}/start + Registry->>Registry: Running + API->>TeppClient: TopicLineageRequest v1 (TRSL-TM + CHRONOS/TDT) + alt TeppNotAvailable + Registry->>Registry: Failed tepp_not_available + else envelope lacks topic-identity/CHRONOS contract + Registry->>Registry: Failed tepp_topic_contract_unavailable + else versioned envelope present + Registry->>Registry: Succeeded; persist topic-identity threads + CHRONOS status + end + API-->>Operator: run status + evidence/inference/prediction detail +``` + +### Implementation note: the status-mark primitive ships ahead of the wiring + +Decision item 5's "distinct visual channel, not color alone" requirement is +implemented now as `EvidenceStatusMark` (`frontend/src/components/ +EvidenceStatusMark.tsx`, i18n in `evidenceStatusI18n.ts`, tokens in +`styles/tokens.css`): a reusable badge distinguishing evidence / inference / +prediction by label text and glyph shape (`●` / `◆` / `△`) in addition to +color, satisfying WCAG 1.4.1 with redundant, testable channels (see its +Storybook stories and `EvidenceStatusMark.test.tsx`). It is presentational +only — every call site must supply `status` from a real TEPP-sourced +envelope; the component never infers or invents one. Wiring it into +`LineageDag`'s topic-thread overlay is the remaining step once TEPP issue 156 +publishes the topic-identity/CHRONOS-status envelope this ADR's +decision 3-4 depend on; until then, no topic-lineage run reaches Succeeded, +so there is no envelope to source a `status` prop from. + +No Figma frame exists yet for this primitive (cf. ADR 0002's precedent for +recording that gap rather than fabricating a frame reference); add the file +ID here when a designer produces one. + +## Considered alternatives + +1. **Fit a local LDA/BERTopic-style model over `zcrht811_export_rows` now, + swap to TEPP later** — rejected: an ungrounded local topic model is + exactly the invented substitute ADR 0084/CLAUDE.md forbid, and its + outputs (unstable topic identity, no posterior uncertainty, no temporal- + identity contract) would not be swap-compatible with TRSL-TM's + logistic-normal, posterior-bearing topic coordinates. +2. **Treat the existing evidence-fusion lineage DAG (ADR 0064/0084) as + sufficient and skip a topic-identity dimension** — rejected: fusion-based + lineage links posts by lexical/embedding/temporal similarity, not by a + calibrated, longitudinally stable topic identity; it cannot express + topic birth/split/merge/retirement the way TRSL-TM can. +3. **Block all topic-lineage UI until TEPP ships the full contract** — + rejected: the fail-closed Pending/Failed states are themselves a + product (ADR 0022 precedent), and the UI/Storybook/e2e scaffolding can + and should be built and reviewable now so the feature activates the + moment TEPP's contract lands, instead of starting from zero then. + +## Consequences + +- LineageWeave's topic-lineage feature is fully specified and reviewable + before TEPP exposes its topic-identity/CHRONOS envelope; only the + transport needs to be connected once TEPP is ready (tracked in memory + `tepp-readiness-watch` and TEPP issue #156). +- The evidence / inference / prediction distinction from TEPP ADR 0016 + propagates end to end — API response, DB row, and UI rendering — so a + CHRONOS prediction can never be flattened into a fact anywhere in this + product. +- Coordination cost: this ADR's runtime activation depends on a + cross-repository contract change in TEPP; `docs/product-technical-gap- + baseline.md` must track that dependency explicitly rather than mark this + gap closed prematurely. + +## References — APA 7th + +ContextualWisdomLab. (2026). *TEPP* [Computer software]. GitHub. +https://github.com/ContextualWisdomLab/TEPP + +ContextualWisdomLab. (2026). *ADR 0012: Temporal relational shared-latent +topic measurement* [ADR]. GitHub. +https://github.com/ContextualWisdomLab/TEPP/blob/main/docs/adr/0012-temporal-relational-shared-latent-topic-measurement.md + +ContextualWisdomLab. (2026). *ADR 0016: TDT, CHRONOS, and Event Ontology +intelligence boundary* [ADR]. GitHub. +https://github.com/ContextualWisdomLab/TEPP/blob/main/docs/adr/0016-tdt-chronos-event-intelligence-boundary.md + +Roberts, M. E., Stewart, B. M., & Tingley, D. (2019). stm: An R package for +structural topic models. *Journal of Statistical Software, 91*(2), 1–40. +https://doi.org/10.18637/jss.v091.i02 + +Mimno, D., Wallach, H. M., Naradowsky, J., Smith, D. A., & McCallum, A. +(2009). Polylingual topic models. In *Proceedings of the 2009 Conference on +Empirical Methods in Natural Language Processing* (pp. 880–889). Association +for Computational Linguistics. https://aclanthology.org/D09-1092/ + +Allan, J. (Ed.). (2002). *Topic detection and tracking: Event-based +information organization*. Springer. https://doi.org/10.1007/978-1-4615-0933-2 + +Kalashnikov, D. V., Chen, Z., Mehrotra, S., & Nuray-Turan, R. (2007). CHRONOS: +Facilitating history discovery by linking temporal records. *Proceedings of +the VLDB Endowment*. https://doi.org/10.14778/2367502.2367559 diff --git a/frontend/src/App.css b/frontend/src/App.css index 01e491796..e5681c8c4 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -659,6 +659,37 @@ color: var(--badge-status-danger-text); } +.evidence-status-mark { + display: inline-flex; + align-items: center; + gap: 0.3rem; + padding: 0.1rem 0.6rem; + border-radius: var(--radius-chip); + font-size: var(--font-size-badge); + font-weight: 600; + white-space: nowrap; +} + +.evidence-status-glyph { + font-size: 0.7em; + line-height: 1; +} + +.evidence-status-evidence { + background: var(--badge-status-evidence-bg); + color: var(--badge-status-evidence-text); +} + +.evidence-status-inference { + background: var(--badge-status-inference-bg); + color: var(--badge-status-inference-text); +} + +.evidence-status-prediction { + background: var(--badge-status-prediction-bg); + color: var(--badge-status-prediction-text); +} + .keyman-select { background: none; border: none; diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 22f002775..4ba9b19d2 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -2476,6 +2476,8 @@ function analysisRunNextAction(run: AnalysisRun): string | null { return "Open this run, then start reconstruction. Reconstruction has not started yet."; case "analysis_run_tepp": return "Open this run to confirm which posts TEPP will measure. Measurement has not started yet — this is not a calibrated result."; + case "analysis_run_topic_lineage": + return "Open this run to confirm which posts TEPP will thread into topic lineage. Topic-lineage analysis has not started yet — this is not a calibrated topic result."; case "analysis_run_report": return "Open this run to confirm which posts the period report will use. The report has not been built yet."; default: { @@ -2487,6 +2489,8 @@ function analysisRunNextAction(run: AnalysisRun): string | null { switch (run.run_kind_code) { case "analysis_run_tepp": return "Open this run to see why it failed, then connect the measurement service and re-run."; + case "analysis_run_topic_lineage": + return "Open this run to see why it failed, then connect the TEPP transport and re-run."; case "analysis_run_lineage": return "Open this run to see why it failed, then retry reconstruction from a current snapshot."; case "analysis_run_report": @@ -2519,6 +2523,11 @@ function analysisRunEmptyPostsHint(run: AnalysisRun): string { "No posts were available at this cutoff for TEPP to measure. " + "Open a later run, or ask an administrator to capture a newer snapshot." ); + case "analysis_run_topic_lineage": + return ( + "No posts were available at this cutoff for topic-lineage analysis. " + + "Open a later run, or ask an administrator to capture a newer snapshot." + ); case "analysis_run_lineage": return ( "No posts were available at this cutoff for reconstruction. " + @@ -2537,31 +2546,36 @@ function analysisRunEmptyPostsHint(run: AnalysisRun): string { } /** - * Corpus copy for a TEPP run that already has cutoff posts. + * Corpus copy for a TEPP or topic-lineage run that already has cutoff posts. * * Those titles are the measurement bag, not a reconstruction result. - * Pending or running must not claim a calibrated measurement. + * Pending or running must not claim a calibrated measurement or topic. */ function analysisRunCorpusHint(run: AnalysisRun): string | null { - if (run.run_kind_code !== "analysis_run_tepp") return null; + const isTopicLineage = run.run_kind_code === "analysis_run_topic_lineage"; + if (run.run_kind_code !== "analysis_run_tepp" && !isTopicLineage) return null; + const service = isTopicLineage ? "topic-lineage" : "TEPP"; + const result = isTopicLineage ? "a topic-identity result" : "a calibrated result"; + const verb = isTopicLineage ? "thread" : "measure"; + const verbPast = isTopicLineage ? "threaded" : "measured"; switch (run.status_code) { case "analysis_status_failed": return ( - "These posts are the cutoff corpus TEPP would measure. Connect a TEPP " + - "transport, then re-run, to replace Failed with a calibrated result." + `These posts are the cutoff corpus ${service} would ${verb}. Connect a TEPP ` + + `transport, then re-run, to replace Failed with ${result}.` ); case "analysis_status_succeeded": - return "These posts are the cutoff corpus this TEPP run measured."; + return `These posts are the cutoff corpus this ${service} run ${verbPast}.`; case "analysis_status_pending": case "analysis_status_running": - return "These posts are the cutoff corpus TEPP will measure once this run finishes."; + return `These posts are the cutoff corpus ${service} will ${verb} once this run finishes.`; case "analysis_status_cancelled": return ( - "These posts are the cutoff corpus this TEPP run would have measured. " + - "The run was cancelled before a calibrated result." + `These posts are the cutoff corpus this ${service} run would have ${verbPast}. ` + + `The run was cancelled before ${result}.` ); case null: - return "These posts are the cutoff corpus attached to this TEPP run."; + return `These posts are the cutoff corpus attached to this ${service} run.`; default: { const unexpected: never = run.status_code; return unexpected; @@ -2671,21 +2685,31 @@ function AnalysisRunReproducibilityDigests({ */ function analysisRunCanStart(run: AnalysisRun): boolean { return ( - (run.run_kind_code === "analysis_run_lineage" || run.run_kind_code === "analysis_run_tepp") && + (run.run_kind_code === "analysis_run_lineage" || + run.run_kind_code === "analysis_run_tepp" || + run.run_kind_code === "analysis_run_topic_lineage") && (run.status_code === "analysis_status_pending" || run.status_code === "analysis_status_running") ); } function analysisRunStartLabel(run: AnalysisRun): string { - return run.run_kind_code === "analysis_run_tepp" - ? "Start TEPP measurement" - : "Start reconstruction"; + if (run.run_kind_code === "analysis_run_tepp") { + return "Start TEPP measurement"; + } + if (run.run_kind_code === "analysis_run_topic_lineage") { + return "Start topic lineage"; + } + return "Start reconstruction"; } -/** Failed TEPP is terminal. Create cannot invent a Pending TEPP row. */ +/** Failed TEPP/topic-lineage is terminal. Create cannot invent a Pending row. */ function analysisRunCanRequestTeppRetry(run: AnalysisRun): boolean { - return run.run_kind_code === "analysis_run_tepp" && run.status_code === "analysis_status_failed"; + return ( + (run.run_kind_code === "analysis_run_tepp" || + run.run_kind_code === "analysis_run_topic_lineage") && + run.status_code === "analysis_status_failed" + ); } const REPORT_PERIOD_KEY = /^\d{4}-W\d{2}$/; @@ -2964,14 +2988,19 @@ function AnalysisRunsPanel({ {starting ? selected.run_kind_code === "analysis_run_tepp" ? "Submitting the TEPP request..." - : "Reconstructing the cutoff bag..." + : selected.run_kind_code === "analysis_run_topic_lineage" + ? "Submitting the topic-lineage request..." + : "Reconstructing the cutoff bag..." : analysisRunStartLabel(selected)} )} {analysisRunCanRequestTeppRetry(selected) && (

- Connect a TEPP transport from this Failed row. Request a lineage - reconstruction does not invent a measurement. + {selected.run_kind_code === "analysis_run_topic_lineage" + ? "Connect a TEPP transport from this Failed row. Request a " + + "lineage reconstruction does not invent a topic model." + : "Connect a TEPP transport from this Failed row. Request a lineage " + + "reconstruction does not invent a measurement."}

)} {analysisRunReportPeriod(selected) && onSelectReportPeriod && ( diff --git a/frontend/src/api.ts b/frontend/src/api.ts index a951207b9..27f58f480 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1105,11 +1105,12 @@ export interface AnalysisRunCount { count_value: number; } -/** Registry kinds from `analysis_run.run_kind_code` (migration 0018). */ +/** Registry kinds from `analysis_run.run_kind_code` (migration 0018, extended 0131). */ export type AnalysisRunKindCode = | "analysis_run_lineage" | "analysis_run_report" - | "analysis_run_tepp"; + | "analysis_run_tepp" + | "analysis_run_topic_lineage"; /** Registry statuses from `analysis_run_status_event.status_code`. */ export type AnalysisRunStatusCode = @@ -1169,6 +1170,8 @@ export interface AnalysisRun { visible_posts?: AnalysisRunVisiblePost[]; reconstructed_edges?: AnalysisRunReconstructedEdge[]; reconstruction_result_sha256?: string; + topic_lineage_result?: Record; + topic_lineage_result_sha256?: string; code_revision_sha?: string; configuration_sha256?: string; } diff --git a/frontend/src/components/EvidenceStatusMark.stories.tsx b/frontend/src/components/EvidenceStatusMark.stories.tsx new file mode 100644 index 000000000..726bfb121 --- /dev/null +++ b/frontend/src/components/EvidenceStatusMark.stories.tsx @@ -0,0 +1,66 @@ +import type { Meta, StoryObj } from "@storybook/react-vite"; +import { expect, within } from "storybook/test"; +import { EvidenceStatusMark } from "./EvidenceStatusMark"; +import "../App.css"; + +const meta = { + title: "Analysis/EvidenceStatusMark", + component: EvidenceStatusMark, + parameters: { layout: "padded" }, +} satisfies Meta; + +export default meta; + +type Story = StoryObj; + +export const Evidence: Story = { + args: { status: "evidence" }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const status = canvas.getByRole("status"); + await expect(status).toHaveTextContent("Evidence"); + await expect(status.getAttribute("aria-label")).toMatch(/directly observed/i); + }, +}; + +export const Inference: Story = { + args: { status: "inference" }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const status = canvas.getByRole("status"); + await expect(status).toHaveTextContent("Inference"); + await expect(status.getAttribute("aria-label")).toMatch(/derived from observed evidence/i); + }, +}; + +export const Prediction: Story = { + args: { status: "prediction" }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const status = canvas.getByRole("status"); + await expect(status).toHaveTextContent("Prediction"); + // A prediction must never read as settled fact -- it's the whole point + // of carrying this status through from TEPP ADR 0016 to the UI. + await expect(status.getAttribute("aria-label")).toMatch(/unconfirmed/i); + }, +}; + +export const AllThreeSideBySide: Story = { + render: () => ( +
+ + + +
+ ), + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const marks = canvas.getAllByRole("status"); + await expect(marks).toHaveLength(3); + // Each mark's accessible name must differ -- the non-color + // distinction requirement (ADR 0132 decision 5) is testable, not + // just visual. + const labels = marks.map((mark) => mark.getAttribute("aria-label")); + await expect(new Set(labels).size).toBe(3); + }, +}; diff --git a/frontend/src/components/EvidenceStatusMark.test.tsx b/frontend/src/components/EvidenceStatusMark.test.tsx new file mode 100644 index 000000000..931142a5a --- /dev/null +++ b/frontend/src/components/EvidenceStatusMark.test.tsx @@ -0,0 +1,53 @@ +import { render, screen } from "@testing-library/react"; +import { afterEach, describe, expect, it } from "vitest"; +import { EvidenceStatusMark } from "./EvidenceStatusMark"; +import { setLocale } from "../i18n"; + +describe("EvidenceStatusMark", () => { + afterEach(() => { + setLocale("en"); + }); + + it("gives evidence, inference, and prediction each a distinct accessible label", () => { + const { rerender } = render(); + const evidenceLabel = screen.getByRole("img").getAttribute("aria-label"); + + rerender(); + const inferenceLabel = screen.getByRole("img").getAttribute("aria-label"); + + rerender(); + const predictionLabel = screen.getByRole("img").getAttribute("aria-label"); + + // The three statuses must never be indistinguishable by text alone + // (WCAG 1.4.1 -- this is the non-color channel, not a decoration). + expect(new Set([evidenceLabel, inferenceLabel, predictionLabel]).size).toBe(3); + expect(evidenceLabel).toMatch(/^Evidence:/); + expect(inferenceLabel).toMatch(/^Inference:/); + expect(predictionLabel).toMatch(/^Prediction:/); + }); + + it("never lets a prediction's copy claim it is confirmed fact", () => { + render(); + const label = screen.getByRole("img").getAttribute("aria-label") ?? ""; + expect(label).toMatch(/unconfirmed/i); + }); + + it("renders visible text, not an icon-only mark", () => { + render(); + expect(screen.getByRole("img")).toHaveTextContent("Evidence"); + }); + + it("hides the decorative glyph from assistive tech", () => { + render(); + const glyph = screen.getByRole("img").querySelector(".evidence-status-glyph"); + expect(glyph).toHaveAttribute("aria-hidden", "true"); + }); + + it("localizes the label and description together", () => { + setLocale("ko"); + render(); + const status = screen.getByRole("img"); + expect(status).toHaveTextContent("증거"); + expect(status.getAttribute("aria-label")).toContain("직접 관찰"); + }); +}); diff --git a/frontend/src/components/EvidenceStatusMark.tsx b/frontend/src/components/EvidenceStatusMark.tsx new file mode 100644 index 000000000..7bf3e252b --- /dev/null +++ b/frontend/src/components/EvidenceStatusMark.tsx @@ -0,0 +1,57 @@ +import { evidenceStatusText } from "../evidenceStatusI18n"; + +/** + * TEPP ADR 0016's three-tier status: directly observed, derived, or + * forecast. A prediction is never rendered or persisted as a fact + * (ADR 0132 decision 5). + */ +export type EvidenceStatus = "evidence" | "inference" | "prediction"; + +/** Glyph carries a second, non-color channel; the visible text label is the primary one. */ +const STATUS_GLYPH: Record = { + evidence: "●", + inference: "◆", + prediction: "△", +}; + +const STATUS_LABEL_KEY: Record = { + evidence: "Evidence", + inference: "Inference", + prediction: "Prediction", +}; + +const STATUS_DESCRIPTION_KEY: Record< + EvidenceStatus, + | "Directly observed in the source record." + | "Derived from observed evidence, not directly recorded." + | "A forecast. Treat as unconfirmed until later evidence arrives." +> = { + evidence: "Directly observed in the source record.", + inference: "Derived from observed evidence, not directly recorded.", + prediction: "A forecast. Treat as unconfirmed until later evidence arrives.", +}; + +/** + * Reusable evidence / inference / prediction status badge (ADR 0132 decision + * 5, TEPP ADR 0016). Distinguishes status by label text and glyph shape, not + * color alone, so it remains legible without color perception (WCAG 1.4.1). + * Presentational only -- callers supply `status` from a real TEPP-sourced + * envelope; this component never infers or invents one. + */ +export function EvidenceStatusMark({ status }: { status: EvidenceStatus }) { + const label = evidenceStatusText(STATUS_LABEL_KEY[status]); + const description = evidenceStatusText(STATUS_DESCRIPTION_KEY[status]); + return ( + + + {label} + + ); +} diff --git a/frontend/src/evidenceStatusI18n.ts b/frontend/src/evidenceStatusI18n.ts new file mode 100644 index 000000000..7c9e86bf1 --- /dev/null +++ b/frontend/src/evidenceStatusI18n.ts @@ -0,0 +1,61 @@ +import { getLocale, type Locale } from "./i18n"; + +const EVIDENCE_STATUS_COPY = { + en: { + Evidence: "Evidence", + Inference: "Inference", + Prediction: "Prediction", + "Directly observed in the source record.": "Directly observed in the source record.", + "Derived from observed evidence, not directly recorded.": + "Derived from observed evidence, not directly recorded.", + "A forecast. Treat as unconfirmed until later evidence arrives.": + "A forecast. Treat as unconfirmed until later evidence arrives.", + }, + ko: { + Evidence: "증거", + Inference: "추론", + Prediction: "예측", + "Directly observed in the source record.": "원본 기록에서 직접 관찰됨.", + "Derived from observed evidence, not directly recorded.": + "관찰된 증거로부터 도출됨, 직접 기록된 것이 아님.", + "A forecast. Treat as unconfirmed until later evidence arrives.": + "예측 결과입니다. 이후 증거가 확인되기 전까지는 미확정으로 취급하십시오.", + }, + zh: { + Evidence: "证据", + Inference: "推断", + Prediction: "预测", + "Directly observed in the source record.": "在原始记录中直接观察到。", + "Derived from observed evidence, not directly recorded.": + "从已观察的证据推导得出，并非直接记录。", + "A forecast. Treat as unconfirmed until later evidence arrives.": + "这是一项预测。在获得后续证据确认之前，请视为未确认。", + }, + ja: { + Evidence: "証拠", + Inference: "推論", + Prediction: "予測", + "Directly observed in the source record.": "元の記録で直接観測されました。", + "Derived from observed evidence, not directly recorded.": + "観測された証拠から導出されたもので、直接記録されたものではありません。", + "A forecast. Treat as unconfirmed until later evidence arrives.": + "これは予測です。後の証拠が届くまでは未確認として扱ってください。", + }, + vi: { + Evidence: "Bằng chứng", + Inference: "Suy luận", + Prediction: "Dự đoán", + "Directly observed in the source record.": "Được quan sát trực tiếp trong bản ghi nguồn.", + "Derived from observed evidence, not directly recorded.": + "Được suy ra từ bằng chứng đã quan sát, không được ghi nhận trực tiếp.", + "A forecast. Treat as unconfirmed until later evidence arrives.": + "Đây là một dự đoán. Hãy coi là chưa xác nhận cho đến khi có bằng chứng sau này.", + }, +} as const satisfies Record>; + +export type EvidenceStatusCopyKey = keyof (typeof EVIDENCE_STATUS_COPY)["en"]; + +/** Return reader-facing evidence/inference/prediction copy in the active product locale. */ +export function evidenceStatusText(key: EvidenceStatusCopyKey): string { + return EVIDENCE_STATUS_COPY[getLocale()][key]; +} diff --git a/frontend/src/styles/tokens.css b/frontend/src/styles/tokens.css index ae518b6e0..25eda5036 100644 --- a/frontend/src/styles/tokens.css +++ b/frontend/src/styles/tokens.css @@ -78,6 +78,14 @@ --badge-status-danger-bg: #f8d7da; --badge-status-danger-text: #721c24; + /* Evidence/Inference/Prediction Status Tokens (ADR 0132, TEPP ADR 0016) */ + --badge-status-evidence-bg: #e3f2fd; + --badge-status-evidence-text: #0d47a1; + --badge-status-inference-bg: #ede7f6; + --badge-status-inference-text: #4527a0; + --badge-status-prediction-bg: #fff3e0; + --badge-status-prediction-text: #e65100; + /* Spacing & Radius Tokens */ --space-chip-inline: 0.6rem; --space-chip-block: 0.1rem; @@ -191,6 +199,12 @@ --badge-status-success-text: #86e29b; --badge-status-danger-bg: rgba(114, 28, 36, 0.3); --badge-status-danger-text: #f5a3ab; + --badge-status-evidence-bg: rgba(96, 165, 250, 0.2); + --badge-status-evidence-text: #93c5fd; + --badge-status-inference-bg: rgba(167, 139, 250, 0.2); + --badge-status-inference-text: #c4b5fd; + --badge-status-prediction-bg: rgba(251, 146, 60, 0.2); + --badge-status-prediction-text: #fdba74; --color-drawer-bg: #16171d; --color-drawer-border: #2e303a; diff --git a/frontend/src/styles/tokens.test.ts b/frontend/src/styles/tokens.test.ts index 35836c74e..24ed9d4ed 100644 --- a/frontend/src/styles/tokens.test.ts +++ b/frontend/src/styles/tokens.test.ts @@ -28,6 +28,12 @@ const BADGE_AND_ACCENT_TOKENS = [ "--badge-status-success-text", "--badge-status-danger-bg", "--badge-status-danger-text", + "--badge-status-evidence-bg", + "--badge-status-evidence-text", + "--badge-status-inference-bg", + "--badge-status-inference-text", + "--badge-status-prediction-bg", + "--badge-status-prediction-text", ]; // Colors this file's dark-mode block replaced -- a regression here would diff --git a/migrations/0131_analysis_run_topic_lineage_kind.sql b/migrations/0131_analysis_run_topic_lineage_kind.sql new file mode 100644 index 000000000..ad284dda0 --- /dev/null +++ b/migrations/0131_analysis_run_topic_lineage_kind.sql @@ -0,0 +1,49 @@ +-- Adds the topic-lineage analysis-run kind (ADR 0132). +-- +-- Requesting/starting this kind submits through the same tepp_client +-- boundary as analysis_run_tepp (ADR 0022) -- it never computes a topic +-- identity or CHRONOS/TDT event-intelligence status locally. This +-- migration only registers the kind vocabulary and widens the existing +-- kind check constraints; it stores no post body and no fabricated +-- measurement. + +begin; + +insert into common_lookup_value + (lookup_category, lookup_code, lookup_label, display_order) +values + ('analysis_run_kind', 'analysis_run_topic_lineage', 'Topic lineage', 3) +on conflict (lookup_code) do nothing; + +alter table analysis_run + add constraint analysis_run_kind_check_topic_lineage + check (run_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp', + 'analysis_run_topic_lineage' + )) not valid; +alter table analysis_run drop constraint if exists analysis_run_kind_check; +alter table analysis_run rename constraint analysis_run_kind_check_topic_lineage + to analysis_run_kind_check; + +do $$ +begin + if to_regclass('public.analysis_run_outbox') is not null then + alter table analysis_run_outbox + add constraint analysis_run_outbox_kind_check_topic_lineage + check (work_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_tepp', + 'analysis_run_topic_lineage' + )) not valid; + alter table analysis_run_outbox + drop constraint if exists analysis_run_outbox_kind_check; + alter table analysis_run_outbox + rename constraint analysis_run_outbox_kind_check_topic_lineage + to analysis_run_outbox_kind_check; + end if; +end +$$; + +commit; diff --git a/migrations/0132_analysis_run_topic_lineage_result.sql b/migrations/0132_analysis_run_topic_lineage_result.sql new file mode 100644 index 000000000..2b0ad181d --- /dev/null +++ b/migrations/0132_analysis_run_topic_lineage_result.sql @@ -0,0 +1,15 @@ +-- Persist only a provider-authoritative completed TEPP topic-lineage +-- envelope (TRSL-TM topic identity + CHRONOS/TDT event-intelligence +-- status, ADR 0132). LineageWeave never computes or substitutes a topic +-- model or event prediction; result_json stores TEPP's versioned envelope +-- verbatim until its schema stabilizes into dedicated columns. +create table if not exists analysis_run_topic_lineage_result ( + analysis_run_id uuid primary key references analysis_run(analysis_run_id) on delete cascade, + remote_run_id text not null check (btrim(remote_run_id) <> ''), + result_json jsonb not null, + result_sha256 text not null check (result_sha256 ~ '^[0-9a-f]{64}$'), + persisted_at timestamptz not null default now() +); + +create index if not exists analysis_run_topic_lineage_result_remote_idx + on analysis_run_topic_lineage_result (remote_run_id); diff --git a/migrations/0204_validate_topic_lineage_kind.sql b/migrations/0204_validate_topic_lineage_kind.sql new file mode 100644 index 000000000..77f109882 --- /dev/null +++ b/migrations/0204_validate_topic_lineage_kind.sql @@ -0,0 +1,11 @@ +-- Validate widened checks without holding the short metadata lock during the scan. +alter table analysis_run validate constraint analysis_run_kind_check; + +do $$ +begin + if to_regclass('public.analysis_run_outbox') is not null then + alter table analysis_run_outbox + validate constraint analysis_run_outbox_kind_check; + end if; +end +$$; diff --git a/migrations/rollback/0204_validate_topic_lineage_kind.sql b/migrations/rollback/0204_validate_topic_lineage_kind.sql new file mode 100644 index 000000000..58dd2a788 --- /dev/null +++ b/migrations/rollback/0204_validate_topic_lineage_kind.sql @@ -0,0 +1,2 @@ +-- Validation changes only catalog state; migration 0131 owns restoration. +select 1; diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 0034b35a6..e2b29310f 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -46,6 +46,7 @@ DEMO_SOURCE_CONTRACT_VERSION = "demo-source-contract-v1" DEMO_LINEAGE_IDEMPOTENCY_KEY = "demo-lineage-seed-2026-w02" DEMO_TEPP_IDEMPOTENCY_KEY = "demo-tepp-seed-2026-w02" +DEMO_TOPIC_LINEAGE_IDEMPOTENCY_KEY = "demo-topic-lineage-seed-2026-w02" DEMO_REPORT_IDEMPOTENCY_KEY = "demo-report-seed-2026-w02" # (post_title, ticket_title, due_date) -- Event Lineage fixtures a report @@ -134,6 +135,7 @@ def seed( cur.execute((migrations / "0021_analysis_run_reconstruction.sql").read_text()) cur.execute((migrations / "0022_analysis_source_snapshot_member.sql").read_text()) cur.execute((migrations / "0023_analysis_run_outbox.sql").read_text()) + cur.execute((migrations / "0131_analysis_run_topic_lineage_kind.sql").read_text()) cur.execute((migrations / "0024_source_post_revision.sql").read_text()) cur.execute((migrations / "0025_role_person_catalog_identity.sql").read_text()) cur.execute( @@ -445,6 +447,11 @@ def seed( account_ids["demo.analyst"], corporate_entity_id, ) + _seed_demo_topic_lineage_run( + cur, + account_ids["demo.analyst"], + corporate_entity_id, + ) _seed_demo_report_run( cur, account_ids["demo.analyst"], @@ -1719,6 +1726,115 @@ def _seed_demo_tepp_run(cur, requested_by_account_id, corporate_entity_id) -> No _seed_demo_run_outbox(cur, run_id) +def topic_lineage_seed_request() -> AnalysisRunRequest: + """Build the Demo Corp topic-lineage request against the shared snapshot digest. + + Same wire shape as :func:`tepp_seed_request` (ADR 0132) -- only the + model contract and output profile select TRSL-TM topic identity plus + CHRONOS/TDT event-intelligence status instead of calibrated + psychometric measurement. + """ + return AnalysisRunRequest( + idempotency_key=DEMO_TOPIC_LINEAGE_IDEMPOTENCY_KEY, + tenant_workspace_id="demo-workspace", + snapshot_id=demo_source_snapshot_sha256(), + knowledge_cutoff="2026-01-12T12:00:00Z", + model_contract_version="tepp-topic-lineage-v1", + output_profile="topic_identity_lineage", + ) + + +def topic_lineage_seed_outcome(client: TeppClient | None = None) -> tuple[str, str | None]: + """Ask TEPP through the published client. A missing transport is Failed. + + Never invents a topic identity or CHRONOS/TDT event prediction. + ``tepp_not_available`` means the channel was dropped, not an abstained + measurement. A live envelope is also not yet a persistable result in + this seed, so the run is not stamped Succeeded. + """ + request = topic_lineage_seed_request() + try: + (client or TeppClient()).submit_analysis_run(request) + except TeppNotAvailable: + return "analysis_status_failed", "tepp_not_available" + return "analysis_status_failed", "tepp_result_not_persisted" + + +def _seed_demo_topic_lineage_run(cur, requested_by_account_id, corporate_entity_id) -> None: + """Insert one Demo-Corp topic-lineage run so the kind is visible without a live TEPP. + + Mirrors :func:`_seed_demo_tepp_run` (ADR 0132). Default transport is + unavailable, so the run ends Failed / ``tepp_not_available`` -- never + a fabricated topic model. + """ + snapshot_id = _ensure_demo_source_snapshot(cur) + _ensure_demo_source_counts(cur, snapshot_id) + _ensure_demo_source_snapshot_members(cur, snapshot_id, corporate_entity_id) + cur.execute( + """ + select analysis_run_id from analysis_run + where requested_by_account_id = %s + and idempotency_key = %s + """, + (requested_by_account_id, DEMO_TOPIC_LINEAGE_IDEMPOTENCY_KEY), + ) + run_row = cur.fetchone() + if run_row is None: + cur.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + values (%s, 'analysis_run_topic_lineage', %s, + %s, '2026-01-12T12:00:00Z', 'topic-lineage-run-v1', %s, %s, + '2026-01-12T12:34:00Z') + returning analysis_run_id + """, + ( + snapshot_id, + DEMO_TOPIC_LINEAGE_IDEMPOTENCY_KEY, + requested_by_account_id, + "d" * 64, + "e" * 40, + ), + ) + run_id = cur.fetchone()[0] + else: + run_id = run_row[0] + cur.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values (%s, 'analysis_scope_corporate_entity', %s) + on conflict (analysis_run_id) do nothing + """, + (run_id, corporate_entity_id), + ) + final_status, failure_code = topic_lineage_seed_outcome() + events = [ + (1, "analysis_status_pending", "2026-01-12T12:35:00Z", None), + (2, "analysis_status_running", "2026-01-12T12:36:00Z", None), + (3, final_status, "2026-01-12T12:37:00Z", failure_code), + ] + cur.execute( + "select 1 from analysis_run_status_event where analysis_run_id = %s limit 1", + (run_id,), + ) + if cur.fetchone() is None: + for ordinal, status, occurred, fail in events: + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at, failure_code) + values (%s, %s, %s, %s, %s) + """, + (run_id, ordinal, status, occurred, fail), + ) + _seed_demo_run_outbox(cur, run_id) + + def _seed_demo_report_run(cur, requested_by_account_id, corporate_entity_id) -> None: """Record the already-built Demo Corp period report on the shared snapshot. @@ -1829,7 +1945,7 @@ def _seed_demo_run_outbox(cur, analysis_run_id) -> None: snapshot_sha256=snapshot_sha256, knowledge_cutoff=knowledge_cutoff, ) - if work_kind_code == "analysis_run_tepp": + if work_kind_code in ("analysis_run_tepp", "analysis_run_topic_lineage"): claimed = datetime(2026, 1, 12, 12, 36, tzinfo=timezone.utc) delivered = datetime(2026, 1, 12, 12, 37, tzinfo=timezone.utc) else: diff --git a/tests/test_analysis_run_create.py b/tests/test_analysis_run_create.py index 664ecc830..ef9dc9f71 100644 --- a/tests/test_analysis_run_create.py +++ b/tests/test_analysis_run_create.py @@ -145,6 +145,10 @@ def test_create_rejects_tepp_and_report_kinds_without_a_fake_score() -> None: _require_lineage_create_kind("analysis_run_tepp") assert tepp.value.status_code == 422 assert "invent a measurement" in tepp.value.detail + with pytest.raises(AnalysisRunCreateError) as topic_lineage: + _require_lineage_create_kind("analysis_run_topic_lineage") + assert topic_lineage.value.status_code == 422 + assert "invent a topic model" in topic_lineage.value.detail with pytest.raises(AnalysisRunCreateError) as report: _require_lineage_create_kind("analysis_run_report") assert report.value.status_code == 422 diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index e4753f30f..ea82f9ac9 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -14,6 +14,8 @@ start_write_conflict_error, tepp_run_request, tepp_submit_outcome, + topic_lineage_run_request, + topic_lineage_submit_outcome, ) from backend.app.lineage_ingestion import records_from_source_posts from lineageweave.fixtures import sample_records @@ -95,7 +97,7 @@ def test_reconstructed_edge_hides_unaffiliated_private_titles() -> None: def test_period_report_start_is_unprocessable_and_tepp_is_allowed() -> None: - """Period-report stays 422. TEPP start is allowed so tepp_client can run.""" + """Period-report stays 422. TEPP/topic-lineage start is allowed so tepp_client can run.""" report = start_kind_rejection("analysis_run_report") assert report is not None assert report.status_code == 422 @@ -103,6 +105,7 @@ def test_period_report_start_is_unprocessable_and_tepp_is_allowed() -> None: assert "period report" in report.detail assert start_kind_rejection("analysis_run_lineage") is None assert start_kind_rejection("analysis_run_tepp") is None + assert start_kind_rejection("analysis_run_topic_lineage") is None def _tepp_request() -> AnalysisRunRequest: @@ -146,6 +149,132 @@ def __init__(self) -> None: assert failure == "tepp_result_not_persisted" +def _topic_lineage_request() -> AnalysisRunRequest: + return topic_lineage_run_request( + idempotency_key="run-topic-lineage-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", + ) + + +def test_topic_lineage_run_request_is_the_published_wire_shape() -> None: + """Start builds TEPP's seven-field request for topic lineage (ADR 0132).""" + request = _topic_lineage_request() + payload = request.to_json() + assert payload["contract_version"] == 1 + assert payload["idempotency_key"] == "run-topic-lineage-2026-w07" + assert payload["snapshot_id"] == "ab" * 32 + assert payload["knowledge_cutoff"] == "2026-01-12T12:00:00Z" + assert payload["model_contract_version"] == "tepp-topic-lineage-v1" + assert payload["output_profile"] == "topic_identity_lineage" + assert "theta" not in str(payload).casefold() + assert "chronos" not in str(payload).casefold() + + +def test_topic_lineage_submit_outcome_drops_a_missing_transport() -> None: + """A missing TEPP transport is Failed, never a fabricated topic model.""" + status, failure, envelope = topic_lineage_submit_outcome( + TeppClient(), _topic_lineage_request() + ) + assert status == "analysis_status_failed" + assert failure == "tepp_not_available" + assert envelope is None + + +def test_topic_lineage_submit_outcome_does_not_persist_an_empty_envelope() -> None: + """An accepted envelope is not yet a persistable topic-lineage result.""" + + class _Accepting(TeppClient): + def __init__(self) -> None: + super().__init__(transport=lambda _payload: {"status": "accepted"}) + + status, failure, envelope = topic_lineage_submit_outcome( + _Accepting(), _topic_lineage_request() + ) + assert status == "analysis_status_failed" + assert envelope is None + assert failure == "tepp_result_not_persisted" + + +def test_topic_lineage_submit_outcome_rejects_a_contentless_completed_envelope() -> None: + """A 'completed' envelope missing the topic-identity/CHRONOS contract is Failed. + + A syntactically valid envelope whose ``result`` lacks TRSL-TM topic + identity and CHRONOS/TDT status (e.g. it accidentally serves the + calibrated-measurement shape) must not be treated as a topic-lineage + success, per ADR 0132 Decision item 3. + """ + + class _EmptyResult(TeppClient): + def __init__(self) -> None: + super().__init__( + transport=lambda _payload: { + "status": "completed", + "analysis_run_id": "r-1", + "result": {}, + } + ) + + status, failure, envelope = topic_lineage_submit_outcome(_EmptyResult(), _topic_lineage_request()) + assert status == "analysis_status_failed" + assert failure == "tepp_topic_contract_unavailable" + assert envelope is None + + +def test_topic_lineage_submit_outcome_accepts_the_versioned_topic_envelope() -> None: + """A completed envelope carrying the versioned topic-identity/CHRONOS contract succeeds.""" + + class _Completed(TeppClient): + def __init__(self) -> None: + super().__init__( + transport=lambda _payload: { + "status": "completed", + "analysis_run_id": "r-1", + "result": { + "envelope_version": 1, + "topic_identity": [{"topic_id": "t-1"}], + "chronos_status": "evidence", + }, + } + ) + + status, failure, envelope = topic_lineage_submit_outcome(_Completed(), _topic_lineage_request()) + assert status == "analysis_status_succeeded" + assert failure == "" + assert envelope is not None + + +@pytest.mark.parametrize("invalid_version", [True, False, 0, 2, "1"]) +def test_topic_lineage_submit_outcome_rejects_unsupported_envelope_version( + invalid_version: object, +) -> None: + """Only integer envelope version 1 is the published contract.""" + + class _WrongVersion(TeppClient): + def __init__(self) -> None: + super().__init__( + transport=lambda _payload: { + "status": "completed", + "analysis_run_id": "r-1", + "result": { + "envelope_version": invalid_version, + "topic_identity": [{"topic_id": "t-1"}], + "chronos_status": "evidence", + }, + } + ) + + status, failure, envelope = topic_lineage_submit_outcome( + _WrongVersion(), _topic_lineage_request() + ) + assert (status, failure, envelope) == ( + "analysis_status_failed", + "tepp_topic_contract_unavailable", + None, + ) + + def test_configured_tepp_client_stays_unavailable_without_http() -> None: """Empty or non-http URLs keep the default dropped channel.""" assert isinstance(configured_tepp_client(""), TeppClient) diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 5216baf7e..871caa125 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -113,3 +113,47 @@ def test_channel_weight_migration_enforces_integrity_and_provenance() -> None: assert "weight_value > 0 and weight_value <= 1" in migration assert "sample_pair_count >= 200" in migration assert "source_snapshot_sha256 ~ '^[0-9a-f]{64}$'" in migration + + +def test_migrate_sh_replays_topic_lineage_migrations_on_existing_volumes() -> None: + """Existing Compose volumes must receive the topic-lineage kind and result table. + + ADR 0166's general four-digit filename boundary covers 0131/0132 without a + per-migration allowlist entry, so this asserts both migration files' own + names still match that boundary shape rather than a stale literal + `migrate.sh` no longer contains. + """ + for migration_name in ( + "0131_analysis_run_topic_lineage_kind.sql", + "0132_analysis_run_topic_lineage_result.sql", + "0204_validate_topic_lineage_kind.sql", + ): + migration_path = Path(__file__).resolve().parents[1] / "migrations" / migration_name + assert migration_path.exists() + assert re.fullmatch(r"[0-9]{4}_.+\.sql", migration_name) + assert int(migration_name[:4]) >= 12 + + +def test_topic_lineage_kind_migration_is_idempotent_for_replay() -> None: + """The kind-widening migration must not fail after a second apply.""" + migration = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0131_analysis_run_topic_lineage_kind.sql" + ).read_text(encoding="utf-8") + + assert "on conflict (lookup_code) do nothing" in migration + assert "drop constraint if exists analysis_run_kind_check" in migration + assert "analysis_run_topic_lineage" in migration + + +def test_topic_lineage_result_migration_is_idempotent_for_replay() -> None: + """The result-table migration must not fail after a second apply.""" + migration = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0132_analysis_run_topic_lineage_result.sql" + ).read_text(encoding="utf-8") + + assert "create table if not exists analysis_run_topic_lineage_result" in migration + assert "create index if not exists" in migration