diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ed005692..f766ef1af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,18 @@ All notable changes to this project are documented here. Format follows ## [2.23.1] - 2026-08-22 +### Added + +- Registered the `analysis_run_topic_lineage` analysis-run kind (migrations + 0131/0132, ADR 0147) and the exact LineageWeave consumer for TEPP's bounded + `tepp.trsl_topic_lineage.v1` artifact (TEPP ADR 0012). It submits through + `tepp_client`, verifies completion, digest, snapshot, cutoff, counts, and + predecessor/successor edges, and fails closed until that artifact is valid. + Project History derives its displayed topic counts only from authorized + artifact edges; the evidence DAG remains navigation evidence, not a fallback. + `make seed` now also writes a Demo Corp topic-lineage run alongside the + existing lineage/TEPP/period-report rows. + ### Changed - Related-node chips now show authorized business context: a unique diff --git a/CLAUDE.md b/CLAUDE.md index 677d16396..1d29356d0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -122,3 +122,21 @@ current and moves focus to the Keyman heading once Keyman rows have settled (ADR 0100). The report-member auto-land chain to related nodes and Ask is not used for GNB origins. A home-list open does not gain that focus. Do not invent a theta. + +## Analysis-run topic lineage (ADR 0147) + +`make seed` also writes a Demo Corp topic-lineage run on the same shared +snapshot, alongside the lineage, TEPP, and period-report rows. The +topic-lineage kind (`analysis_run_topic_lineage`, migration 0131) submits +through the same `tepp_client` boundary as TEPP (ADR 0022), requesting +TEPP's `trsl_tm_cpu_f64_v1` Temporal Relational Shared-Latent Topic +Measurement (TRSL-TM) result instead of calibrated psychometric measurement. +A missing transport or an unused accepted envelope is Failed +(`tepp_not_available` / `tepp_result_not_persisted`), the same as TEPP. Do +not invent a topic identity or a local topic model of any kind. `POST +/api/analysis-runs` still 422s this kind — Create +does not invent a Pending topic-lineage row; connect a TEPP transport from +a Failed topic-lineage row and re-run through +`POST /api/analysis-runs/{id}/start`, exactly like TEPP. A Succeeded +envelope persists into `analysis_run_topic_lineage_result` (migration 0132) +only after its exact `tepp.trsl_topic_lineage.v1` artifact validates. diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index fb76ad821..5f5ff0b1a 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""" @@ -611,11 +613,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 0147). 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( @@ -623,6 +626,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 180a660cf..e5c3f37a8 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -33,10 +33,17 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable +from lineageweave.topic_lineage_artifact import ( + TOPIC_LINEAGE_MODEL_CONTRACT_VERSION, + TOPIC_LINEAGE_OUTPUT_PROFILE, + TopicLineageUnavailable, + parse_topic_lineage_envelope, +) _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" @@ -69,11 +76,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 0147). 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( @@ -130,6 +137,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 0147). + + 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 + the bounded TRSL topic-lineage artifact instead of calibrated + psychometric measurement. + """ + cutoff = knowledge_cutoff + if cutoff.tzinfo is None: + cutoff = cutoff.replace(tzinfo=UTC) + return AnalysisRunRequest( + idempotency_key=idempotency_key, + tenant_workspace_id=str(corporate_entity_id), + snapshot_id=snapshot_sha256, + knowledge_cutoff=cutoff.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), + model_contract_version=TOPIC_LINEAGE_MODEL_CONTRACT_VERSION, + output_profile=TOPIC_LINEAGE_OUTPUT_PROFILE, + ) + + def _tepp_submission( client: TeppClient, request: AnalysisRunRequest, @@ -165,6 +200,37 @@ def tepp_submit_outcome( return status_code, failure_code +def _topic_lineage_submission( + client: TeppClient, + request: AnalysisRunRequest, +) -> tuple[str, str, dict[str, Any] | None]: + """Require TEPP's exact digest-bound topic-lineage artifact.""" + + try: + response = client.submit_analysis_run(request) + except TeppNotAvailable: + return _FAILED, "tepp_not_available", None + try: + parse_topic_lineage_envelope( + response, + expected_snapshot_id=request.snapshot_id, + expected_knowledge_cutoff=request.knowledge_cutoff, + ) + except TopicLineageUnavailable: + return _FAILED, "tepp_topic_contract_unavailable", None + return _SUCCEEDED, "", response + + +def topic_lineage_submit_outcome( + client: TeppClient, + request: AnalysisRunRequest, +) -> tuple[str, str]: + """Return the strict topic-lineage submission outcome.""" + + status_code, failure_code, _ = _topic_lineage_submission(client, request) + return status_code, failure_code + + async def _persist_tepp_result( conn: asyncpg.Connection, *, @@ -196,6 +262,40 @@ 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 TEPP's validated artifact envelope (ADR 0147).""" + + try: + artifact = parse_topic_lineage_envelope(envelope) + except TopicLineageUnavailable: + return False + remote_run_id = artifact["run_id"] + 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( @@ -628,6 +728,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, @@ -782,3 +889,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 0147). Never persists a locally + computed topic identity or substitutes evidence-DAG counts. + """ + now = datetime.now(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_submission(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(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/app/project_history.py b/backend/app/project_history.py index 1ac74422d..e7920cb2a 100644 --- a/backend/app/project_history.py +++ b/backend/app/project_history.py @@ -2,6 +2,8 @@ from __future__ import annotations +import hashlib +import json from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any, Protocol @@ -15,6 +17,11 @@ build_project_history_projection, normalize_project_key, ) +from lineageweave.topic_lineage_artifact import ( + TopicLineageUnavailable, + parse_topic_lineage_envelope, + project_topic_lineage_projection, +) PROJECT_HISTORY_DEFAULT_LIMIT = 64 PROJECT_HISTORY_MAXIMUM_LIMIT = 128 @@ -131,7 +138,7 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: 'semantic_project_key'::text, {_MENTION_KEY} as identity_key, {_MENTION_KEY} as matched_value, - mention.confidence, + mention.mention_confidence, mention.ontology_iri, 'post_project_mention.project_key'::text from post_project_mention mention @@ -143,7 +150,7 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: 'semantic_project_name'::text, coalesce({_MENTION_KEY}, {_MENTION_NAME}) as identity_key, {_MENTION_NAME} as matched_value, - mention.confidence, + mention.mention_confidence, mention.ontology_iri, 'post_project_mention.project_name'::text from post_project_mention mention @@ -178,7 +185,7 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: union all select role.post_id, role.actor_name, - role.responsibility, + role.responsibility_text as responsibility, role.actor_type_code, role.affiliated_organization_name, role.cataloged_person_id, @@ -197,6 +204,26 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: and edge.child_post_id = any($1::uuid[]) order by edge.child_post_id, edge.parent_post_id """ +_TOPIC_LINEAGE_SQL = """ +select distinct on (scope.corporate_entity_id) + result.result_json, + result.result_sha256, + result.remote_run_id, + snapshot.snapshot_sha256, + run.knowledge_cutoff + from analysis_run_topic_lineage_result result + join analysis_run run on run.analysis_run_id = result.analysis_run_id + join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id + join analysis_source_snapshot snapshot + on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id + join analysis_run_current_status status + on status.analysis_run_id = run.analysis_run_id + where run.run_kind_code = 'analysis_run_topic_lineage' + and status.status_code = 'analysis_status_succeeded' + and scope.corporate_entity_id::text = any($1::text[]) + and run.knowledge_cutoff <= $2 + order by scope.corporate_entity_id, run.requested_at desc, run.analysis_run_id desc +""" _INDEX_SQL = f""" with query_timeout as materialized ( select set_config( @@ -408,6 +435,12 @@ async def fetch_project_history_projection( visible_ids=visible_ids, normalized_key=normalized_key, ) + topic_lineage = await _fetch_topic_lineage_projection( + conn, + visible_ids=visible_ids, + corporate_entity_ids=corporate_entity_ids, + knowledge_cutoff=knowledge_cutoff, + ) projection = build_project_history_projection( project_key=project_key, focus_event_id=canonical_focus_id, @@ -415,6 +448,7 @@ async def fetch_project_history_projection( match_rows=match_rows, role_rows=role_rows, edge_rows=edge_rows, + topic_lineage=topic_lineage, truncated=truncated, ) projection["knowledge_cutoff"] = _as_utc(knowledge_cutoff) @@ -434,3 +468,40 @@ async def _fetch_project_children( roles = list(await conn.fetch(_ROLE_SQL, list(visible_ids))) edges = list(await conn.fetch(_EDGE_SQL, list(visible_ids))) return matches, roles, edges + + +async def _fetch_topic_lineage_projection( + conn: ProjectHistoryConnection, + *, + visible_ids: Sequence[str], + corporate_entity_ids: Sequence[str], + knowledge_cutoff: datetime, +) -> dict[str, Any]: + """Project only intact TEPP artifacts onto already-authorized posts.""" + + rows = await conn.fetch( + _TOPIC_LINEAGE_SQL, + list(corporate_entity_ids), + knowledge_cutoff, + ) + artifacts: list[Mapping[str, Any]] = [] + for row in rows: + envelope = row["result_json"] + try: + decoded = json.loads(envelope) if isinstance(envelope, str) else envelope + stored = json.dumps(decoded, separators=(",", ":"), sort_keys=True) + except (json.JSONDecodeError, TypeError, ValueError): + continue + if hashlib.sha256(stored.encode("utf-8")).hexdigest() != row["result_sha256"]: + continue + try: + artifact = parse_topic_lineage_envelope( + envelope, + expected_snapshot_id=str(row["snapshot_sha256"]), + expected_knowledge_cutoff=_as_utc(row["knowledge_cutoff"]), + expected_remote_run_id=str(row["remote_run_id"]), + ) + except TopicLineageUnavailable: + continue + artifacts.append(artifact) + return project_topic_lineage_projection(artifacts, visible_ids) diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index e68e51103..e333d12f0 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -138,6 +138,16 @@ / "migrations" / "0104_two_word_database_identifiers.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" +) def _postgres_available() -> bool: @@ -262,6 +272,8 @@ def seeded_db(demo_analyst_token): cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text()) cur.execute(_TENANT_SETTINGS_MIGRATION.read_text()) cur.execute(_IDENTIFIER_MIGRATION.read_text()) + cur.execute(_TOPIC_LINEAGE_KIND_MIGRATION.read_text()) + cur.execute(_TOPIC_LINEAGE_RESULT_MIGRATION.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index 712107b27..e4961aea5 100644 --- a/docker/postgres-init/migrate.sh +++ b/docker/postgres-init/migrate.sh @@ -19,7 +19,7 @@ for migration in /opt/lineageweave/migrations/*.sql; do case "$migration_name" in 0012_*|0013_*|0014_*|0015_*|0016_*|0017_*|0018_*|0019_*|0020_*|0021_*|0022_*|0023_*|0024_*|0025_*|0026_*|0027_*|0028_*|0029_*|0030_*|0031_*|0032_*|0033_*|0034_*|0035_*|0036_*|0037_*|0038_*|0039_*|0040_*|0041_*|0042_*|0043_*|0044_*|0045_*|0046_*|0047_*|0048_*|0049_*|0050_*) ;; 0051_*|0052_*|0053_*|0054_*) ;; - 0060_*|0100_*|0101_*|0102_*) ;; + 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*|0130_*|0131_*|0132_*) ;; *) continue ;; esac printf 'Applying %s\n' "$migration_name" diff --git a/docs/adr/0147-tepp-topic-lineage-project-history.md b/docs/adr/0147-tepp-topic-lineage-project-history.md new file mode 100644 index 000000000..8fbe26a60 --- /dev/null +++ b/docs/adr/0147-tepp-topic-lineage-project-history.md @@ -0,0 +1,83 @@ +# ADR 0147 — TEPP topic-lineage evidence in Project History + +**Decision status:** Accepted on this active product branch; not protected-main truth +**Implementation maturity:** active-PR +**Date:** 2026-08-23 +**Depends on:** ADR 0022, ADR 0084, ADR 0113, ADR 0127, ADR 0136, and TEPP ADR 0012 +**Figma File ID:** `SBpgot7uTvMxEaxUwvoc0S` +**Figma frames:** `308:2`, `309:2`, `309:50`, `310:2` + +## Context + +Project History already exposes one authorized, cutoff-safe timeline from a +post, post-scoped Ask, and Global Ask. Its `connected_post_count` and +`lineage_count` currently describe weak components in LineageWeave's fused +evidence DAG. Those values are useful navigation evidence, but they are not +topic-model lineages and cannot satisfy the TEPP TRSL-TM requirement. + +TEPP now defines a bounded CPU `f64` reference estimator and the completed +`tepp.trsl_topic_lineage.v1` artifact. The artifact carries explicit fitted +predecessor/successor associations, artifact-local topic indices, +connectable-post and lineage counts, snapshot/cutoff binding, and the fixed +claim boundary `fitted_topic_association_not_causation`. + +## Decision + +1. LineageWeave requests topic lineage with TEPP model contract + `trsl_tm_cpu_f64_v1` and output profile `trsl_topic_lineage_v1`. It does not + fit, select, rename, or repair topics locally. +2. A completed transport response is accepted only when its `result` is an + exact, bounded `tepp.trsl_topic_lineage.v1` object. Identifiers, RFC 3339 + cutoff, finite values, edge endpoints, topic indices, duplicate edges, + top-level counts, and the non-causal inference status are revalidated before + persistence and again before display. +3. Project History filters validated TEPP sequence edges to the already + authorized, cutoff-safe project post IDs. It recomputes the displayed + connectable-post count and lineage count from those filtered edges. Topic + identity is the pair `(TEPP run id, artifact-local topic index)` so indices + from different runs cannot collapse. +4. The existing fused `post_lineage_edge` remains the source for navigable + prior-history paths. It does not supply fallback topic counts. When no + validated, authorized artifact contributes an edge, topic counts are + unavailable rather than zero or substituted with DAG components. +5. The shared `ProjectHistoryTimeline` remains the only reader component for + dedicated Project History, each post, post-scoped Ask, and Global Ask. It + labels validated numbers as fitted topic association, never causation, and + renders an explicit unavailable state otherwise. No second timeline or DAG + overlay is introduced in this increment. +6. CHRONOS/TDT prediction status, topic birth/split/merge, accelerated backend + parity, production `K` selection, and causal claims remain outside this + artifact and require their own upstream contract and ADR change. + +## Consequences + +The count shown beside a project history has one scientific source and one +authorization boundary. A missing TEPP transport, accepted-only receipt, +non-converged estimator, invalid artifact, stale cutoff, or unrelated snapshot +cannot silently become a topic result. Existing temporal history and source +navigation remain readable while topic evidence is unavailable. + +## Verification + +- strict artifact parser round-trip, tamper, size, count, and scope tests; +- analysis-run request and completed-envelope failure tests; +- PostgreSQL projection tests proving unauthorized and out-of-project endpoints + do not affect counts; +- shared React component tests and Storybook scenes for validated and + unavailable states at desktop and phone widths; +- backend and frontend full checks, then exact-head hosted checks and an + independent review before protected merge. + +## References + +Blei, D. M., & Lafferty, J. D. (2006). Dynamic topic models. In *Proceedings +of the 23rd International Conference on Machine Learning* (pp. 113–120). +Association for Computing Machinery. https://doi.org/10.1145/1143844.1143859 + +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 + +ContextualWisdomLab. (2026). *ADR 0012: Temporal relational shared-latent +topic measurement* [ADR]. +https://github.com/ContextualWisdomLab/TEPP/blob/main/docs/adr/0012-temporal-relational-shared-latent-topic-measurement.md diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 6cbc26312..bb76f4466 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -129,6 +129,7 @@ claims that an unmerged PR or historical runtime observation is live behavior. | FR-12 | A hierarchy-enrichment timeout leaves the source-grounded summary readable and the actor unbound; it never creates a guessed catalog identity. | ADR 0101, ADR 0010, ADR 0026 | Commit `1c260f20` contains the boundary, ADR, and focused test; independent review, protected-main merge, and fresh runtime evidence remain pending | | FR-13 | Customer Master projects authorized corporate entities as a Group → Company → Plant tree. Real organization containment uses W3C ORG while Group/Company/Plant remain separate SKOS level concepts. Missing-parent, self-parent, and cyclic edges remain visible as unresolved roots; the UI owns nested `group` elements from their parent `treeitem`, supports Arrow/Home/End and Enter/Space operation, and opens source-backed evidence outside the tree. | ADR 0124, ADR 0004, ADR 0010 | Ontology/SHACL interoperability tests, `customerMasterTree.ts`, `CustomerMasterTree.tsx`, component tests, Storybook, and code commit `21074cf80cbfab3001bf18b6e1a618f75f4bed24` | | FR-14 | Global Ask presents a dedicated evidence workspace: semantic form submission, IME-safe keyboard behavior, explicit empty/loading/error/answer states, separated timeline and cited evidence, answer focus, responsive phone/tablet/PC layout, and the existing authorized cited-post → Event Lineage handoff. | ADR 0137, ADR 0002, ADR 0032, ADR 0090 | `AskAgentWorkspace.tsx`, focused component/token tests, Storybook state inventory, and existing App navigation regressions on #353 | +| FR-15 | Project History reports connectable posts and topic lineages only from a validated TEPP `tepp.trsl_topic_lineage.v1` artifact filtered to the already authorized project post set. Missing or invalid topic evidence is unavailable, never replaced by evidence-DAG components. | ADR 0147, TEPP ADR 0012 | Active product branch: TEPP reference estimator/artifact execution plus LineageWeave strict parser, analysis-run persistence, shared Project History projection, component tests, and Storybook states; protected merge and authenticated runtime evidence remain open | ## TRD @@ -333,20 +334,21 @@ runtime note into a shipped/live claim. ## Project-history lineage counts (2026-08-23) -- The shared timeline now distinguishes displayed authorized project posts from - posts incident to a persisted forward `post_lineage_edge`, and reports the - number of weakly connected evidence-lineage components in that bounded set. -- Isolated project posts remain visible but do not inflate the lineage count. - These counts describe LineageWeave's evidence-fusion DAG; they are not TEPP - topic identities and do not close the TRSL-TM topic-lineage dependency. +- The shared timeline reports connectable posts and topic lineages only from + digest-, snapshot-, cutoff-, and schema-validated TEPP + `tepp.trsl_topic_lineage.v1` predecessor/successor edges whose endpoints are + already in the authorized project post set. +- Missing, invalid, stale, or out-of-project artifacts render topic counts as + unavailable. `post_lineage_edge` still supplies prior-history navigation but + cannot substitute weak-component counts for TEPP topic identity. - The same projection supplies the dedicated Project history destination, each post's project-evidence action, post-scoped Ask, and Global Ask, so those surfaces cannot drift into competing count definitions. -- Stacked delivery evidence is PR #487 on PR #258. Local evidence is the full - 993-test backend suite, focused Project History UI tests, the complete App test - file, lint, production build, Storybook build, and rendered-story inspection. - Protected merge, authenticated runtime, and the TEPP topic-identity result - contract remain open and must not be represented as protected-main behavior. +- Active-branch evidence includes exact 100% statement/branch coverage for the + strict artifact consumer and PostgreSQL projection, the real Project History + API integration path, complete frontend tests/lint/build, and Storybook build. + Protected merge and authenticated rendered-runtime inspection remain open and + must not be represented as protected-main behavior. ## Ask-to-project-history integration (2026-08-21) diff --git a/docs/storybook-inventory.md b/docs/storybook-inventory.md index caa724ded..14da86bf0 100644 --- a/docs/storybook-inventory.md +++ b/docs/storybook-inventory.md @@ -9,7 +9,7 @@ reader-facing control you can click before changing product CSS. | `Evidence/CitationChip` | Click a cited title to open that source post. | `--color-chip-border`, `--radius-chip`, `CitationChip` | | `AnalysisRun/CutoffKnownBody` | Read the cutoff-known sentence, then compare it with the live body below. | `--color-accent-border`, `--space-panel-block`, `--radius-panel`, `CutoffKnownBody` | | `Analysis/LineageEntityPicker` | Choose which corp to reconstruct, then click Request a lineage reconstruction. | `--space-control-gap`, `--size-control-min`, `--radius-control`, `LineageEntityPicker` | -| `Evidence/ProjectHistoryTimeline` | Compare all project posts with connected-post and evidence-lineage counts, then open the selected source. | shared spacing, status, focus, and timeline tokens; `ProjectHistoryTimeline` | +| `Buyer/Project History Timeline` | Compare authorized project posts with validated TEPP topic-lineage counts or the explicit unavailable state, then open the selected source. | shared spacing, status, focus, and timeline tokens; `ProjectHistoryTimeline` | | `Chrome/PopupCloseButton` | Close the evidence panel or post popup. | `--space-close-inset`, `--font-size-close`, `PopupCloseButton` | | `Evidence/LineageDag` | Inspect a branching Event Lineage, then open a record or read its evidence trail. | `--color-primary`, `--color-accent-orange`, `LineageDag` | diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index d8a5d7693..294109141 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -2565,6 +2565,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: { @@ -2576,6 +2578,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 topic-lineage service 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": @@ -2608,6 +2612,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. " + @@ -2626,31 +2635,35 @@ 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 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 " + + `These posts are the cutoff corpus ${service} would ${verb}. Connect a ${service} ` + "transport, then re-run, to replace Failed with a calibrated 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. " + + `These posts are the cutoff corpus this ${service} run would have ${verbPast}. ` + "The run was cancelled before a calibrated 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; @@ -2767,21 +2780,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}$/; @@ -3063,14 +3086,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 topic-lineage 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 0f88d0ff2..33056672e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1029,11 +1029,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 = diff --git a/frontend/src/components/AskProjectHistoryLinks.test.tsx b/frontend/src/components/AskProjectHistoryLinks.test.tsx index c5973a316..4e3a91f14 100644 --- a/frontend/src/components/AskProjectHistoryLinks.test.tsx +++ b/frontend/src/components/AskProjectHistoryLinks.test.tsx @@ -23,8 +23,8 @@ const projection: ProjectHistoryProjection = { knowledge_cutoff: "2026-08-20T12:00:00Z", evidence_boundary_code: "authorized_visible_source_posts", event_count: 1, - connected_post_count: 0, - lineage_count: 0, + connected_post_count: null, + lineage_count: null, distinct_actor_count: 0, distinct_observed_actor_count: 0, truncated: false, diff --git a/frontend/src/components/ProjectHistoryTimeline.stories.tsx b/frontend/src/components/ProjectHistoryTimeline.stories.tsx index 8f33fc14a..90a26aee0 100644 --- a/frontend/src/components/ProjectHistoryTimeline.stories.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.stories.tsx @@ -62,6 +62,15 @@ const projection: ProjectHistoryProjection = { event_count: 5, connected_post_count: 5, lineage_count: 1, + topic_lineage: { + status: "validated", + schema_version: "tepp.trsl_topic_lineage.v1", + inference_status: "fitted_topic_association_not_causation", + artifact_count: 1, + connected_post_count: 5, + lineage_count: 1, + sequence_edges: [], + }, distinct_actor_count: 3, distinct_observed_actor_count: 2, truncated: false, @@ -150,3 +159,22 @@ export const ResponsibilityEvidenceGap: Story = { }, }, }; + +export const TopicLineageUnavailable: Story = { + args: { + projection: { + ...projection, + connected_post_count: null, + lineage_count: null, + topic_lineage: { + status: "unavailable", + schema_version: null, + inference_status: null, + artifact_count: 0, + connected_post_count: null, + lineage_count: null, + sequence_edges: [], + }, + }, + }, +}; diff --git a/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx b/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx index 0a9fdaef3..5f95c90e4 100644 --- a/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx @@ -14,8 +14,17 @@ const projection = { knowledge_cutoff: "2026-08-20T12:00:00Z", evidence_boundary_code: "authorized_visible_source_posts", event_count: 1, - connected_post_count: 0, - lineage_count: 0, + connected_post_count: null, + lineage_count: null, + topic_lineage: { + status: "unavailable", + schema_version: null, + inference_status: null, + artifact_count: 0, + connected_post_count: null, + lineage_count: null, + sequence_edges: [], + }, distinct_actor_count: 0, distinct_observed_actor_count: 0, truncated: false, @@ -64,6 +73,7 @@ describe("ProjectHistoryTimeline TEPP integration", () => { expect(screen.getByRole("heading", { name: /TEPP temporal validation/i })).toBeInTheDocument(); expect(screen.getByRole("heading", { name: /Project event timeline/i })).toBeInTheDocument(); + expect(screen.getByText(/TEPP topic-lineage counts unavailable/i)).toBeInTheDocument(); expect(screen.getAllByRole("tab")).toHaveLength(1); }); }); diff --git a/frontend/src/components/ProjectHistoryTimeline.test.tsx b/frontend/src/components/ProjectHistoryTimeline.test.tsx index 3ec742bb5..fc32b8d2b 100644 --- a/frontend/src/components/ProjectHistoryTimeline.test.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.test.tsx @@ -16,6 +16,15 @@ const projection: ProjectHistoryProjection = { event_count: 3, connected_post_count: 3, lineage_count: 1, + topic_lineage: { + status: "validated", + schema_version: "tepp.trsl_topic_lineage.v1", + inference_status: "fitted_topic_association_not_causation", + artifact_count: 1, + connected_post_count: 3, + lineage_count: 1, + sequence_edges: [], + }, distinct_actor_count: 2, distinct_observed_actor_count: 1, truncated: false, diff --git a/frontend/src/components/ProjectHistoryTimeline.tsx b/frontend/src/components/ProjectHistoryTimeline.tsx index 940028103..11e3512be 100644 --- a/frontend/src/components/ProjectHistoryTimeline.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.tsx @@ -52,6 +52,10 @@ export function ProjectHistoryTimeline({ const selectedResponsibilities = selectedEvent?.responsibility_evidence ?? selectedEvent?.observed_responsibilities ?? []; const actorCount = projection.distinct_actor_count ?? projection.distinct_observed_actor_count; + const topicCountsAvailable = + projection.topic_lineage?.status === "validated" && + projection.connected_post_count !== null && + projection.lineage_count !== null; const selectedIndex = selectedEvent ? projection.events.findIndex((event) => event.event_id === selectedEvent.event_id) : -1; @@ -101,12 +105,16 @@ export function ProjectHistoryTimeline({

{projectHistoryText(locale, "heading")}

- {projectHistoryText(locale, "summaryCounts", { - events: projection.event_count, - connected: projection.connected_post_count, - lineages: projection.lineage_count, - actors: actorCount, - })} + {projectHistoryText( + locale, + topicCountsAvailable ? "summaryCounts" : "summaryCountsUnavailable", + { + events: projection.event_count, + connected: projection.connected_post_count ?? 0, + lineages: projection.lineage_count ?? 0, + actors: actorCount, + }, + )}

diff --git a/frontend/src/projectHistory.ts b/frontend/src/projectHistory.ts index 1219baf4b..ab73645c3 100644 --- a/frontend/src/projectHistory.ts +++ b/frontend/src/projectHistory.ts @@ -109,8 +109,23 @@ export interface ProjectHistoryProjection { knowledge_cutoff?: string; evidence_boundary_code?: "authorized_visible_source_posts" | string; event_count: number; - connected_post_count: number; - lineage_count: number; + connected_post_count: number | null; + lineage_count: number | null; + topic_lineage?: { + status: "validated" | "unavailable"; + schema_version: "tepp.trsl_topic_lineage.v1" | null; + inference_status: "fitted_topic_association_not_causation" | null; + artifact_count: number; + connected_post_count: number | null; + lineage_count: number | null; + sequence_edges: Array<{ + artifact_run_id: string; + predecessor_post_id: string; + successor_post_id: string; + topic_index: number; + association_strength: number; + }>; + }; distinct_actor_count?: number; distinct_observed_actor_count: number; truncated: boolean; @@ -202,6 +217,7 @@ const MESSAGE_KEYS = [ "evidenceBoundary", "heading", "summaryCounts", + "summaryCountsUnavailable", "sourcePostTime", "documentTime", "truncated", @@ -256,6 +272,7 @@ const EN: Record = { evidenceBoundary: "Only source posts that pass the current permission, visibility, publication, and cutoff gates are included.", heading: "Project event timeline", summaryCounts: "{events} shown project posts · {connected} connected · lineage count {lineages} · {actors} actors in evidence", + summaryCountsUnavailable: "{events} shown project posts · TEPP topic-lineage counts unavailable · {actors} actors in evidence", sourcePostTime: "Dates use source-post creation time because a separate event clock is not recorded.", documentTime: "Dates use the document time recorded by the source.", truncated: "This bounded timeline is truncated. The selected event remains included.", @@ -307,6 +324,7 @@ const MESSAGES: Record> = { evidenceBoundary: "현재 권한·공개 범위·게시 상태·기준 시각을 통과한 원천 게시물만 포함합니다.", heading: "프로젝트 이벤트 타임라인", summaryCounts: "표시 중 프로젝트 글 {events}개 · 연결된 글 {connected}개 · Lineage {lineages}개 · 근거 담당자 {actors}명", + summaryCountsUnavailable: "표시 중 프로젝트 글 {events}개 · TEPP Topic Lineage 수치 이용 불가 · 근거 담당자 {actors}명", sourcePostTime: "별도 사건 시각이 없어 날짜는 원천 게시물 생성 시각을 사용합니다.", documentTime: "날짜는 기록된 문서 시각을 사용합니다.", truncated: "이 제한된 타임라인은 일부만 표시합니다. 선택한 이벤트는 계속 포함됩니다.", @@ -355,6 +373,7 @@ const MESSAGES: Record> = { evidenceBoundary: "仅包含通过当前权限、可见性、发布状态和截止时间检查的源帖子。", heading: "项目事件时间线", summaryCounts: "当前显示 {events} 条项目记录 · {connected} 条已连接 · {lineages} 条 Lineage · 依据中出现 {actors} 名责任人", + summaryCountsUnavailable: "当前显示 {events} 条项目记录 · TEPP 主题谱系计数不可用 · 依据中出现 {actors} 名责任人", sourcePostTime: "未记录独立事件时钟，因此日期采用源帖子创建时间。", documentTime: "日期采用记录的文档时间。", truncated: "此有界时间线已截断，但所选事件仍保留。", @@ -403,6 +422,7 @@ const MESSAGES: Record> = { evidenceBoundary: "現在の権限・可視性・公開状態・基準時刻を通過した原資料だけを含みます。", heading: "プロジェクトイベントのタイムライン", summaryCounts: "表示中のプロジェクト投稿 {events}件 · 接続済み {connected}件 · Lineage {lineages}件 · 根拠内の担当者 {actors}名", + summaryCountsUnavailable: "表示中のプロジェクト投稿 {events}件 · TEPP Topic Lineage 件数は利用不可 · 根拠内の担当者 {actors}名", sourcePostTime: "独立したイベント時刻がないため、原資料の作成時刻を使用します。", documentTime: "日付は記録された文書時刻を使用します。", truncated: "この上限付きタイムラインは省略されていますが、選択イベントは保持されます。", @@ -451,6 +471,7 @@ const MESSAGES: Record> = { evidenceBoundary: "Chỉ bao gồm bài nguồn vượt qua quyền, khả năng hiển thị, trạng thái xuất bản và mốc thời gian hiện tại.", heading: "Dòng thời gian sự kiện dự án", summaryCounts: "Đang hiển thị {events} bài dự án · {connected} bài đã kết nối · {lineages} lineage · {actors} người trong bằng chứng", + summaryCountsUnavailable: "Đang hiển thị {events} bài dự án · số liệu TEPP topic-lineage không khả dụng · {actors} người trong bằng chứng", sourcePostTime: "Không có đồng hồ sự kiện riêng, nên dùng thời gian tạo bài nguồn.", documentTime: "Ngày sử dụng thời gian tài liệu được ghi nhận.", truncated: "Dòng thời gian có giới hạn này đã bị rút gọn nhưng vẫn giữ sự kiện đang chọn.", diff --git a/lineageweave/project_history.py b/lineageweave/project_history.py index c485b585d..35bacc90a 100644 --- a/lineageweave/project_history.py +++ b/lineageweave/project_history.py @@ -315,6 +315,7 @@ def build_project_history_projection( match_rows: Sequence[Mapping[str, Any]], role_rows: Sequence[Mapping[str, Any]], edge_rows: Sequence[Mapping[str, Any]], + topic_lineage: Mapping[str, Any] | None = None, truncated: bool = False, maximum_depth: int = PROJECT_HISTORY_MAX_DEPTH, maximum_paths_per_event: int = PROJECT_HISTORY_MAX_PATHS_PER_EVENT, @@ -454,7 +455,21 @@ def build_project_history_projection( maximum_depth=maximum_depth, maximum_paths_per_event=maximum_paths_per_event, ) - connected_post_count, lineage_count = _lineage_counts(ordered_ids, edge_rows) + evidence_connected_post_count, evidence_lineage_count = _lineage_counts( + ordered_ids, edge_rows + ) + topic_lineage = dict( + topic_lineage + or { + "status": "unavailable", + "schema_version": None, + "inference_status": None, + "artifact_count": 0, + "connected_post_count": None, + "lineage_count": None, + "sequence_edges": [], + } + ) events: list[dict[str, Any]] = [] previous_actor_keys: Sequence[str] | None = None @@ -514,8 +529,11 @@ def build_project_history_projection( "focus_event_id": effective_focus, "time_basis_code": PROJECT_HISTORY_TIME_BASIS, "event_count": len(events), - "connected_post_count": connected_post_count, - "lineage_count": lineage_count, + "connected_post_count": topic_lineage["connected_post_count"], + "lineage_count": topic_lineage["lineage_count"], + "topic_lineage": topic_lineage, + "evidence_connected_post_count": evidence_connected_post_count, + "evidence_lineage_count": evidence_lineage_count, "distinct_actor_count": len(distinct_actor_keys), "distinct_observed_actor_count": len(distinct_observed_actor_keys), "truncated": bool(truncated), diff --git a/lineageweave/topic_lineage_artifact.py b/lineageweave/topic_lineage_artifact.py new file mode 100644 index 000000000..44023480c --- /dev/null +++ b/lineageweave/topic_lineage_artifact.py @@ -0,0 +1,303 @@ +"""Strict LineageWeave consumer for TEPP topic-lineage artifacts.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +import hashlib +import json +from typing import Any +from uuid import UUID + +TOPIC_LINEAGE_ARTIFACT_SCHEMA_VERSION = "tepp.trsl_topic_lineage.v1" +TOPIC_LINEAGE_MODEL_CONTRACT_VERSION = "trsl_tm_cpu_f64_v1" +TOPIC_LINEAGE_OUTPUT_PROFILE = "trsl_topic_lineage_v1" +TOPIC_LINEAGE_INFERENCE_STATUS = "fitted_topic_association_not_causation" +TOPIC_LINEAGE_ARTIFACT_BYTE_LIMIT = 256 * 1024 +TOPIC_LINEAGE_EDGE_LIMIT = 100_000 +_U64_MAXIMUM = 2**64 - 1 +_ARTIFACT_FIELDS = frozenset( + { + "schema_version", + "run_id", + "snapshot_id", + "knowledge_cutoff", + "selected_seed", + "iterations", + "objective", + "topic_count", + "evidence_count", + "connected_post_count", + "lineage_count", + "sequence_edges", + "inference_status", + } +) +_EDGE_FIELDS = frozenset( + { + "predecessor_document_id", + "successor_document_id", + "topic_index", + "association_strength", + } +) + + +class TopicLineageUnavailable(ValueError): + """TEPP topic-lineage evidence was absent or violated its contract.""" + + +def _text(value: Any, name: str, maximum: int = 256) -> str: + """Return bounded non-empty text without control characters.""" + + if not isinstance(value, str) or value != value.strip(): + raise TopicLineageUnavailable(f"{name} must be canonical text") + if ( + not value + or len(value.encode("utf-8")) > maximum + or any(ord(character) < 0x20 or ord(character) == 0x7F for character in value) + ): + raise TopicLineageUnavailable(f"{name} is outside its bound") + return value + + +def _u64(value: Any, name: str) -> int: + """Return one unsigned 64-bit integer without accepting booleans.""" + + if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= _U64_MAXIMUM: + raise TopicLineageUnavailable(f"{name} must be an unsigned 64-bit integer") + return value + + +def _rfc3339_utc(value: Any, name: str) -> str: + """Return one offset-aware timestamp in canonical UTC form.""" + + raw = _text(value, name, 64) + try: + parsed = datetime.fromisoformat(raw[:-1] + "+00:00" if raw.endswith("Z") else raw) + except ValueError as exc: + raise TopicLineageUnavailable(f"{name} must be RFC 3339") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise TopicLineageUnavailable(f"{name} must include an offset") + return parsed.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +def _uuid(value: Any, name: str) -> str: + """Return one lowercase canonical UUID.""" + + raw = _text(value, name, 36) + try: + parsed = UUID(raw) + except ValueError as exc: + raise TopicLineageUnavailable(f"{name} must be a UUID") from exc + if str(parsed) != raw: + raise TopicLineageUnavailable(f"{name} must be a canonical UUID") + return raw + + +def _json_object(value: Any, *, maximum_bytes: int) -> Mapping[str, Any]: + """Decode one bounded JSON object or validate an in-memory mapping.""" + + if isinstance(value, str): + if len(value.encode("utf-8")) > maximum_bytes: + raise TopicLineageUnavailable("topic-lineage JSON exceeds its byte limit") + try: + value = json.loads(value) + except json.JSONDecodeError as exc: + raise TopicLineageUnavailable("topic-lineage JSON is invalid") from exc + try: + encoded = json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise TopicLineageUnavailable("topic-lineage value is not JSON") from exc + if len(encoded) > maximum_bytes: + raise TopicLineageUnavailable("topic-lineage JSON exceeds its byte limit") + if not isinstance(value, Mapping): + raise TopicLineageUnavailable("topic-lineage JSON must be an object") + return value + + +def parse_topic_lineage_artifact(value: Any) -> dict[str, Any]: + """Validate and canonicalize one exact TEPP topic-lineage artifact.""" + + artifact = _json_object(value, maximum_bytes=TOPIC_LINEAGE_ARTIFACT_BYTE_LIMIT) + if frozenset(artifact) != _ARTIFACT_FIELDS: + raise TopicLineageUnavailable("topic-lineage artifact fields are invalid") + schema_version = _text(artifact["schema_version"], "schema_version", 64) + if schema_version != TOPIC_LINEAGE_ARTIFACT_SCHEMA_VERSION: + raise TopicLineageUnavailable("topic-lineage artifact schema is unsupported") + run_id = _text(artifact["run_id"], "run_id") + snapshot_id = _text(artifact["snapshot_id"], "snapshot_id") + knowledge_cutoff = _rfc3339_utc(artifact["knowledge_cutoff"], "knowledge_cutoff") + selected_seed = _u64(artifact["selected_seed"], "selected_seed") + iterations = _u64(artifact["iterations"], "iterations") + if iterations == 0: + raise TopicLineageUnavailable("iterations must be positive") + objective = artifact["objective"] + if isinstance(objective, bool) or not isinstance(objective, (int, float)): + raise TopicLineageUnavailable("objective must be numeric") + try: + objective = float(objective) + except OverflowError as exc: + raise TopicLineageUnavailable("objective must be finite") from exc + topic_count = _u64(artifact["topic_count"], "topic_count") + evidence_count = _u64(artifact["evidence_count"], "evidence_count") + connected_post_count = _u64(artifact["connected_post_count"], "connected_post_count") + lineage_count = _u64(artifact["lineage_count"], "lineage_count") + if topic_count < 2 or evidence_count < 2: + raise TopicLineageUnavailable("topic and evidence counts must be at least two") + if connected_post_count > evidence_count or lineage_count > topic_count: + raise TopicLineageUnavailable("topic-lineage counts exceed their dimensions") + raw_edges = artifact["sequence_edges"] + if not isinstance(raw_edges, list) or len(raw_edges) > TOPIC_LINEAGE_EDGE_LIMIT: + raise TopicLineageUnavailable("sequence_edges is outside its bound") + pairs: set[tuple[str, str]] = set() + connected: set[str] = set() + lineages: set[int] = set() + edges: list[dict[str, Any]] = [] + for raw_edge in raw_edges: + if not isinstance(raw_edge, Mapping) or frozenset(raw_edge) != _EDGE_FIELDS: + raise TopicLineageUnavailable("topic-lineage edge fields are invalid") + predecessor = _uuid(raw_edge["predecessor_document_id"], "predecessor_document_id") + successor = _uuid(raw_edge["successor_document_id"], "successor_document_id") + topic_index = _u64(raw_edge["topic_index"], "topic_index") + strength = raw_edge["association_strength"] + if isinstance(strength, bool) or not isinstance(strength, (int, float)): + raise TopicLineageUnavailable("association_strength must be numeric") + try: + strength = float(strength) + except OverflowError as exc: + raise TopicLineageUnavailable("association_strength must be finite") from exc + pair = (predecessor, successor) + if ( + predecessor == successor + or topic_index >= topic_count + or not 0.0 < strength <= 1.0 + or pair in pairs + ): + raise TopicLineageUnavailable("topic-lineage edge is invalid") + pairs.add(pair) + connected.update(pair) + lineages.add(topic_index) + edges.append( + { + "predecessor_document_id": predecessor, + "successor_document_id": successor, + "topic_index": topic_index, + "association_strength": strength, + } + ) + inference_status = _text(artifact["inference_status"], "inference_status", 64) + if inference_status != TOPIC_LINEAGE_INFERENCE_STATUS: + raise TopicLineageUnavailable("topic-lineage inference status is unsupported") + if connected_post_count != len(connected) or lineage_count != len(lineages): + raise TopicLineageUnavailable("topic-lineage counts do not match the edges") + return { + "schema_version": schema_version, + "run_id": run_id, + "snapshot_id": snapshot_id, + "knowledge_cutoff": knowledge_cutoff, + "selected_seed": selected_seed, + "iterations": iterations, + "objective": objective, + "topic_count": topic_count, + "evidence_count": evidence_count, + "connected_post_count": connected_post_count, + "lineage_count": lineage_count, + "sequence_edges": edges, + "inference_status": inference_status, + } + + +def topic_lineage_artifact_sha256(value: Any) -> str: + """Return TEPP's SHA-256 over canonical artifact field order.""" + + artifact = parse_topic_lineage_artifact(value) + wire = json.dumps(artifact, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(wire).hexdigest() + + +def parse_topic_lineage_envelope( + value: Any, + *, + expected_snapshot_id: str | None = None, + expected_knowledge_cutoff: str | None = None, + expected_remote_run_id: str | None = None, +) -> dict[str, Any]: + """Validate a completed, digest-bound transport envelope and its artifact.""" + + envelope = _json_object(value, maximum_bytes=TOPIC_LINEAGE_ARTIFACT_BYTE_LIMIT * 2) + if envelope.get("status") not in {"completed", "succeeded"}: + raise TopicLineageUnavailable("topic-lineage run is not completed") + remote_run_id = envelope.get("analysis_run_id") or envelope.get("run_id") + remote_run_id = _text(remote_run_id, "remote_run_id") + artifact = parse_topic_lineage_artifact(envelope.get("result")) + if envelope.get("result_schema_version") != TOPIC_LINEAGE_ARTIFACT_SCHEMA_VERSION: + raise TopicLineageUnavailable("topic-lineage result schema is unsupported") + digest = _text(envelope.get("result_sha256"), "result_sha256", 64) + if digest != topic_lineage_artifact_sha256(artifact): + raise TopicLineageUnavailable("topic-lineage result digest does not match") + if artifact["run_id"] != remote_run_id: + raise TopicLineageUnavailable("topic-lineage run identity does not match") + if expected_remote_run_id is not None and remote_run_id != expected_remote_run_id: + raise TopicLineageUnavailable("topic-lineage persisted run identity does not match") + if expected_snapshot_id is not None and artifact["snapshot_id"] != expected_snapshot_id: + raise TopicLineageUnavailable("topic-lineage snapshot identity does not match") + if expected_knowledge_cutoff is not None and artifact["knowledge_cutoff"] != _rfc3339_utc( + expected_knowledge_cutoff, "expected_knowledge_cutoff" + ): + raise TopicLineageUnavailable("topic-lineage knowledge cutoff does not match") + return artifact + + +def project_topic_lineage_projection( + artifacts: Sequence[Mapping[str, Any]], visible_post_ids: Sequence[str] +) -> dict[str, Any]: + """Filter validated TEPP edges to one authorized Project History post set.""" + + visible = set(visible_post_ids) + connected: set[str] = set() + lineages: set[tuple[str, int]] = set() + edges: list[dict[str, Any]] = [] + contributing_runs: set[str] = set() + for value in artifacts: + artifact = parse_topic_lineage_artifact(value) + for edge in artifact["sequence_edges"]: + predecessor = edge["predecessor_document_id"] + successor = edge["successor_document_id"] + if predecessor not in visible or successor not in visible: + continue + connected.update((predecessor, successor)) + lineages.add((artifact["run_id"], edge["topic_index"])) + contributing_runs.add(artifact["run_id"]) + edges.append( + { + "artifact_run_id": artifact["run_id"], + "predecessor_post_id": predecessor, + "successor_post_id": successor, + "topic_index": edge["topic_index"], + "association_strength": edge["association_strength"], + } + ) + edges.sort( + key=lambda edge: ( + edge["predecessor_post_id"], + edge["successor_post_id"], + edge["artifact_run_id"], + edge["topic_index"], + ) + ) + available = bool(edges) + return { + "status": "validated" if available else "unavailable", + "schema_version": TOPIC_LINEAGE_ARTIFACT_SCHEMA_VERSION if available else None, + "inference_status": TOPIC_LINEAGE_INFERENCE_STATUS if available else None, + "artifact_count": len(contributing_runs), + "connected_post_count": len(connected) if available else None, + "lineage_count": len(lineages) if available else None, + "sequence_edges": edges, + } 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..319dd4c88 --- /dev/null +++ b/migrations/0131_analysis_run_topic_lineage_kind.sql @@ -0,0 +1,45 @@ +-- Adds the topic-lineage analysis-run kind (ADR 0147). +-- +-- Requesting/starting this kind submits through the same tepp_client +-- boundary as analysis_run_tepp (ADR 0022) -- it never computes a topic +-- identity or predecessor/successor association 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 + drop constraint if exists analysis_run_kind_check; +alter table analysis_run + add constraint analysis_run_kind_check + check (run_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp', + 'analysis_run_topic_lineage' + )); + +do $$ +begin + if to_regclass('public.analysis_run_outbox') is not null then + alter table analysis_run_outbox + drop constraint if exists analysis_run_outbox_kind_check; + alter table analysis_run_outbox + add constraint analysis_run_outbox_kind_check + check (work_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_tepp', + 'analysis_run_topic_lineage' + )); + 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..34ad99cb1 --- /dev/null +++ b/migrations/0132_analysis_run_topic_lineage_result.sql @@ -0,0 +1,14 @@ +-- Persist only a provider-authoritative completed TEPP topic-lineage +-- `tepp.trsl_topic_lineage.v1` envelope (ADR 0147). LineageWeave never +-- computes or substitutes a topic model; result_json retains the exact +-- digest-bound TEPP artifact envelope. +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/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 3c84ba8b8..760a1439d 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -33,6 +33,10 @@ from lineageweave.http_client import get_json_list, post_form from lineageweave.post_summary import ACTOR_TYPE_PERSON, POST_SUMMARY_CONTRACT_VERSION from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable +from lineageweave.topic_lineage_artifact import ( + TOPIC_LINEAGE_MODEL_CONTRACT_VERSION, + TOPIC_LINEAGE_OUTPUT_PROFILE, +) REALM = "lineageweave-demo" DEFAULT_POSTGRES_DSN = "postgresql://lineageweave:lineageweave_dev_only@localhost:15432/lineageweave" @@ -45,6 +49,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 @@ -439,6 +444,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"], @@ -1670,6 +1680,113 @@ 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 0147); only the model + contract and output profile select the bounded topic-lineage artifact. + """ + 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=TOPIC_LINEAGE_MODEL_CONTRACT_VERSION, + output_profile=TOPIC_LINEAGE_OUTPUT_PROFILE, + ) + + +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 predecessor/successor association. + ``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 0147). 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. @@ -1780,7 +1897,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 c2b1b6065..7796ff919 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -14,11 +14,14 @@ 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 from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable +from lineageweave.topic_lineage_artifact import topic_lineage_artifact_sha256 def test_reconstruction_digest_is_stable_and_ignores_edge_order() -> None: @@ -94,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 @@ -102,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: @@ -145,6 +149,88 @@ 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 0147).""" + 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"] == "trsl_tm_cpu_f64_v1" + assert payload["output_profile"] == "trsl_topic_lineage_v1" + 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 = topic_lineage_submit_outcome(TeppClient(), _topic_lineage_request()) + assert status == "analysis_status_failed" + assert failure == "tepp_not_available" + + +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 = topic_lineage_submit_outcome(_Accepting(), _topic_lineage_request()) + assert status == "analysis_status_failed" + assert failure == "tepp_topic_contract_unavailable" + + +def test_topic_lineage_submit_outcome_accepts_only_the_bound_tepp_artifact() -> None: + """The exact completed artifact can cross the analysis-run boundary.""" + + artifact = { + "schema_version": "tepp.trsl_topic_lineage.v1", + "run_id": "tepp-run-1", + "snapshot_id": "ab" * 32, + "knowledge_cutoff": "2026-01-12T12:00:00Z", + "selected_seed": 7, + "iterations": 4, + "objective": 1.25, + "topic_count": 2, + "evidence_count": 2, + "connected_post_count": 2, + "lineage_count": 1, + "sequence_edges": [ + { + "predecessor_document_id": "00000000-0000-0000-0000-000000000001", + "successor_document_id": "00000000-0000-0000-0000-000000000002", + "topic_index": 0, + "association_strength": 0.8, + } + ], + "inference_status": "fitted_topic_association_not_causation", + } + envelope = { + "status": "completed", + "run_id": artifact["run_id"], + "result_schema_version": artifact["schema_version"], + "result_sha256": topic_lineage_artifact_sha256(artifact), + "result": artifact, + } + client = TeppClient(transport=lambda _payload: envelope) + + assert topic_lineage_submit_outcome(client, _topic_lineage_request()) == ( + "analysis_status_succeeded", + "", + ) + + 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 6bccf9d41..9550bb8d7 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -78,3 +78,51 @@ def test_migrate_sh_replays_project_history_lookup_indexes() -> None: ): assert f"create index if not exists {index_name}" in forward assert f"drop index if exists {index_name}" in rollback +def test_migrate_sh_replays_source_commercial_context_migration_on_existing_volumes() -> None: + """Existing Compose volumes must receive the source context columns.""" + script = ( + Path(__file__).resolve().parents[1] + / "docker" + / "postgres-init" + / "migrate.sh" + ).read_text(encoding="utf-8") + + assert "0130_*" in script + + +def test_migrate_sh_replays_topic_lineage_migrations_on_existing_volumes() -> None: + """Existing Compose volumes must receive the topic-lineage kind and result table.""" + script = ( + Path(__file__).resolve().parents[1] + / "docker" + / "postgres-init" + / "migrate.sh" + ).read_text(encoding="utf-8") + + assert "0131_*" in script + assert "0132_*" in script + + +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 diff --git a/tests/test_project_history.py b/tests/test_project_history.py index 693bc6544..9be07c4c7 100644 --- a/tests/test_project_history.py +++ b/tests/test_project_history.py @@ -4,15 +4,19 @@ import asyncio from datetime import datetime, timezone +import hashlib +import json import pytest from backend.app.project_history import ( ProjectHistoryNotFound, ProjectHistoryConnection, + _fetch_topic_lineage_projection, fetch_project_history_index, fetch_project_history_projection, ) +from lineageweave.topic_lineage_artifact import topic_lineage_artifact_sha256 from lineageweave.project_history import ( build_project_history_projection, classify_project_event, @@ -75,6 +79,113 @@ async def fetch(self, query: str, *args: object): return [] +def test_topic_lineage_repository_filters_to_authorized_post_ids() -> None: + """Persisted TEPP evidence is digest-bound before visible projection.""" + + artifact = { + "schema_version": "tepp.trsl_topic_lineage.v1", + "run_id": "tepp-run-1", + "snapshot_id": "ab" * 32, + "knowledge_cutoff": "2026-01-12T12:00:00Z", + "selected_seed": 7, + "iterations": 4, + "objective": 1.25, + "topic_count": 2, + "evidence_count": 2, + "connected_post_count": 2, + "lineage_count": 1, + "sequence_edges": [ + { + "predecessor_document_id": "00000000-0000-0000-0000-000000000001", + "successor_document_id": "00000000-0000-0000-0000-000000000002", + "topic_index": 0, + "association_strength": 0.8, + } + ], + "inference_status": "fitted_topic_association_not_causation", + } + envelope = { + "status": "completed", + "run_id": "tepp-run-1", + "result_schema_version": artifact["schema_version"], + "result_sha256": topic_lineage_artifact_sha256(artifact), + "result": artifact, + } + stored = json.dumps(envelope, separators=(",", ":"), sort_keys=True) + invalid_contract = {**envelope, "result_schema_version": "unknown"} + invalid_contract_stored = json.dumps( + invalid_contract, separators=(",", ":"), sort_keys=True + ) + + class Connection: + async def fetch(self, query: str, *args: object): + assert "analysis_run_topic_lineage_result" in query + return [ + { + "result_json": "{not-json", + "result_sha256": "0" * 64, + "remote_run_id": "tepp-run-1", + "snapshot_sha256": "ab" * 32, + "knowledge_cutoff": datetime(2026, 1, 12, 12, tzinfo=timezone.utc), + }, + { + "result_json": stored, + "result_sha256": "0" * 64, + "remote_run_id": "tepp-run-1", + "snapshot_sha256": "ab" * 32, + "knowledge_cutoff": datetime(2026, 1, 12, 12, tzinfo=timezone.utc), + }, + { + "result_json": invalid_contract_stored, + "result_sha256": hashlib.sha256(invalid_contract_stored.encode()).hexdigest(), + "remote_run_id": "tepp-run-1", + "snapshot_sha256": "ab" * 32, + "knowledge_cutoff": datetime(2026, 1, 12, 12, tzinfo=timezone.utc), + }, + { + "result_json": stored, + "result_sha256": hashlib.sha256(stored.encode()).hexdigest(), + "remote_run_id": "tepp-run-1", + "snapshot_sha256": "ab" * 32, + "knowledge_cutoff": datetime(2026, 1, 12, 12, tzinfo=timezone.utc), + } + ] + + projection = asyncio.run( + _fetch_topic_lineage_projection( + Connection(), + visible_ids=[ + "00000000-0000-0000-0000-000000000001", + "00000000-0000-0000-0000-000000000002", + ], + corporate_entity_ids=["11111111-1111-1111-1111-111111111111"], + knowledge_cutoff=datetime(2026, 8, 20, tzinfo=timezone.utc), + ) + ) + + assert projection["connected_post_count"] == 2 + assert projection["lineage_count"] == 1 + + +def test_project_history_keeps_an_already_visible_focus_without_a_second_lookup() -> None: + """A visible focus stays in the authorized page without a focus query.""" + + event = _event_row() + connection = _ProjectionConnection([event]) + result = asyncio.run( + fetch_project_history_projection( + connection, + project_key="P-100", + focus_post_id=str(event["post_id"]), + knowledge_cutoff=datetime(2026, 8, 20, tzinfo=timezone.utc), + corporate_entity_ids=[], + ) + ) + + assert result["focus_event_id"] == event["post_id"] + assert all("post_id = $4::uuid" not in query for query, _ in connection.calls) + + def test_project_identity_is_exact_but_unicode_compatible() -> None: """Compatibility forms may normalize; fuzzy project binding may not.""" @@ -206,8 +317,11 @@ def test_project_history_counts_connected_posts_and_distinct_lineages() -> None: ) assert projection["event_count"] == 6 - assert projection["connected_post_count"] == 5 - assert projection["lineage_count"] == 2 + assert projection["connected_post_count"] is None + assert projection["lineage_count"] is None + assert projection["evidence_connected_post_count"] == 5 + assert projection["evidence_lineage_count"] == 2 + assert projection["topic_lineage"]["status"] == "unavailable" def test_matching_observed_project_code_keeps_its_distinct_display_name() -> None: @@ -454,7 +568,7 @@ def test_project_history_projection_keeps_focus_and_authorization_bounds() -> No assert result["truncated"] is True assert result["focus_event_id"] == "00000000-0000-0000-0000-000000000099" assert result["knowledge_cutoff"] == "2026-08-20T00:00:00Z" - assert len(connection.calls) == 5 + assert len(connection.calls) == 6 assert connection.calls[0][1][-1] == 3 diff --git a/tests/test_topic_lineage_artifact.py b/tests/test_topic_lineage_artifact.py new file mode 100644 index 000000000..5a922a9df --- /dev/null +++ b/tests/test_topic_lineage_artifact.py @@ -0,0 +1,211 @@ +"""Exact TEPP topic-lineage artifact consumer contracts.""" + +from copy import deepcopy + +import pytest + +from lineageweave.topic_lineage_artifact import ( + TopicLineageUnavailable, + parse_topic_lineage_artifact, + parse_topic_lineage_envelope, + project_topic_lineage_projection, + topic_lineage_artifact_sha256, +) + + +def _artifact(run_id: str = "tepp-run-1") -> dict[str, object]: + """Return one synthetic, non-identifying TEPP artifact.""" + + return { + "schema_version": "tepp.trsl_topic_lineage.v1", + "run_id": run_id, + "snapshot_id": "ab" * 32, + "knowledge_cutoff": "2026-01-12T12:00:00Z", + "selected_seed": 7, + "iterations": 4, + "objective": 1.25, + "topic_count": 2, + "evidence_count": 3, + "connected_post_count": 3, + "lineage_count": 2, + "sequence_edges": [ + { + "predecessor_document_id": "00000000-0000-0000-0000-000000000001", + "successor_document_id": "00000000-0000-0000-0000-000000000002", + "topic_index": 0, + "association_strength": 0.8, + }, + { + "predecessor_document_id": "00000000-0000-0000-0000-000000000002", + "successor_document_id": "00000000-0000-0000-0000-000000000003", + "topic_index": 1, + "association_strength": 0.7, + }, + ], + "inference_status": "fitted_topic_association_not_causation", + } + + +def _envelope(artifact: dict[str, object] | None = None) -> dict[str, object]: + """Wrap one artifact in TEPP's completed digest-bound envelope.""" + + result = artifact or _artifact() + return { + "status": "completed", + "run_id": result["run_id"], + "result_schema_version": result["schema_version"], + "result_sha256": topic_lineage_artifact_sha256(result), + "result": result, + } + + +def test_exact_envelope_and_authorized_projection() -> None: + """Only validated edges whose endpoints are visible contribute counts.""" + + artifact = parse_topic_lineage_envelope( + _envelope(), + expected_snapshot_id="ab" * 32, + expected_knowledge_cutoff="2026-01-12T12:00:00+00:00", + expected_remote_run_id="tepp-run-1", + ) + projection = project_topic_lineage_projection( + [artifact], + [ + "00000000-0000-0000-0000-000000000001", + "00000000-0000-0000-0000-000000000002", + ], + ) + + assert projection["status"] == "validated" + assert projection["connected_post_count"] == 2 + assert projection["lineage_count"] == 1 + assert projection["artifact_count"] == 1 + assert len(projection["sequence_edges"]) == 1 + + +def test_projection_keeps_run_scoped_topic_identity_and_unavailable_state() -> None: + """Equal topic indexes from separate runs remain separate lineages.""" + + second = deepcopy(_artifact("tepp-run-2")) + second["sequence_edges"] = [deepcopy(_artifact()["sequence_edges"][0])] + second["connected_post_count"] = 2 + second["lineage_count"] = 1 + second["evidence_count"] = 2 + visible = [ + "00000000-0000-0000-0000-000000000001", + "00000000-0000-0000-0000-000000000002", + ] + + projection = project_topic_lineage_projection([_artifact(), second], visible) + assert projection["lineage_count"] == 2 + assert projection["artifact_count"] == 2 + assert project_topic_lineage_projection([_artifact()], [visible[0]])["status"] == "unavailable" + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.update(run_id=None), "canonical text"), + (lambda value: value.update(run_id=""), "outside"), + (lambda value: value.update(selected_seed=True), "unsigned"), + (lambda value: value.update(knowledge_cutoff="bad-date"), "RFC 3339"), + (lambda value: value.update(knowledge_cutoff="2026-01-12T12:00:00"), "offset"), + (lambda value: value.update(schema_version="unknown"), "schema"), + (lambda value: value.update(iterations=0), "iterations"), + (lambda value: value.update(objective="1.25"), "objective"), + (lambda value: value.update(objective=10**400), "finite"), + (lambda value: value.update(topic_count=1), "at least two"), + (lambda value: value.update(connected_post_count=4), "dimensions"), + (lambda value: value.update(connected_post_count=2), "counts"), + (lambda value: value.update(sequence_edges=()), "sequence_edges"), + (lambda value: value.update(inference_status="causal"), "inference"), + (lambda value: value.update(extra=True), "fields"), + (lambda value: value["sequence_edges"][0].update(extra=True), "edge fields"), + ( + lambda value: value["sequence_edges"][0].update( + predecessor_document_id="not-a-uuid" + ), + "UUID", + ), + ( + lambda value: value["sequence_edges"][0].update( + predecessor_document_id="AAAAAAAA-AAAA-AAAA-AAAA-AAAAAAAAAAAA" + ), + "canonical UUID", + ), + (lambda value: value["sequence_edges"][0].update(association_strength="0.8"), "numeric"), + (lambda value: value["sequence_edges"][0].update(association_strength=10**400), "finite"), + (lambda value: value["sequence_edges"][0].update(topic_index=2), "edge"), + ( + lambda value: value["sequence_edges"][0].update( + successor_document_id=value["sequence_edges"][0]["predecessor_document_id"] + ), + "edge", + ), + ], +) +def test_artifact_rejects_contract_drift(mutate, message: str) -> None: + """Schema, convergence, count, inference, and edge drift fail closed.""" + + artifact = _artifact() + mutate(artifact) + with pytest.raises(TopicLineageUnavailable, match=message): + parse_topic_lineage_artifact(artifact) + + +@pytest.mark.parametrize( + "value", + [ + "{not-json", + "x" * (256 * 1024 + 1), + ["not", "an", "object"], + {"oversized": "x" * (256 * 1024)}, + {"not_json": float("inf")}, + ], +) +def test_artifact_rejects_invalid_or_oversized_json(value) -> None: + """The JSON boundary is bounded and rejects non-objects and non-finite values.""" + + with pytest.raises(TopicLineageUnavailable): + parse_topic_lineage_artifact(value) + + +@pytest.mark.parametrize( + "mutation", + [ + lambda value: value.update(status="accepted"), + lambda value: value.update(result_sha256="0" * 64), + lambda value: value.update(run_id="another-run"), + ], +) +def test_envelope_rejects_incomplete_or_unbound_results(mutation) -> None: + """Completion, digest, and run identity are mandatory.""" + + envelope = _envelope() + mutation(envelope) + with pytest.raises(TopicLineageUnavailable): + parse_topic_lineage_envelope(envelope) + + +@pytest.mark.parametrize( + "kwargs", + [ + {"expected_remote_run_id": "another-run"}, + {"expected_snapshot_id": "different-snapshot"}, + {"expected_knowledge_cutoff": "2026-01-13T12:00:00Z"}, + ], +) +def test_envelope_rejects_persisted_identity_drift(kwargs) -> None: + """Stored run, snapshot, and cutoff bindings cannot drift.""" + + with pytest.raises(TopicLineageUnavailable): + parse_topic_lineage_envelope(_envelope(), **kwargs) + + +def test_envelope_rejects_an_unknown_result_schema() -> None: + """A completed result with another schema remains unavailable.""" + + envelope = _envelope() + envelope["result_schema_version"] = "unknown" + with pytest.raises(TopicLineageUnavailable, match="result schema"): + parse_topic_lineage_envelope(envelope)