From 79d3e307beb9a7eece258cc97d80b083393e6234 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:41:53 +0000 Subject: [PATCH] feat: start a pending lineage reconstruction (v0.88.0) POST /api/analysis-runs/{id}/start runs ThreadWeave on a frozen Pending lineage cutoff bag and persists run-scoped edges (ADR 0021). Succeeded retries replay the stored digest; Running is 409. TEPP and period-report start stay 422 so this path never invents a theta. Co-authored-by: Seongho Bae --- AGENTS.md | 4 + ARCHITECTURE.md | 15 +- CHANGELOG.d/0.88.0-analysis-run-start.md | 5 + CHANGELOG.md | 14 + CLAUDE.md | 4 +- backend/app/analysis_run_ingestion.py | 124 ++++++- backend/app/analysis_run_start.py | 336 ++++++++++++++++++ backend/app/main.py | 31 ++ backend/tests/test_api.py | 218 ++++++++++++ docker/postgres-init/Dockerfile | 2 + .../0013-normalized-analysis-run-registry.md | 7 +- .../0017-authorized-analysis-run-create.md | 10 +- .../adr/0021-authorized-analysis-run-start.md | 112 ++++++ .../ANALYSIS_RUN_REGISTRY_REFERENCES.md | 10 +- frontend/package.json | 2 +- frontend/src/App.test.tsx | 89 ++++- frontend/src/App.tsx | 79 +++- frontend/src/api.ts | 19 + lineageweave/__init__.py | 2 +- .../0021_analysis_run_reconstruction.sql | 79 ++++ .../0022_analysis_source_snapshot_member.sql | 35 ++ .../0021_analysis_run_reconstruction.sql | 37 ++ .../0022_analysis_source_snapshot_member.sql | 27 ++ pyproject.toml | 2 +- scripts/seed_demo_data.py | 33 ++ ...test_analysis_run_reconstruction_schema.py | 150 ++++++++ tests/test_analysis_run_registry_schema.py | 8 + tests/test_analysis_run_start.py | 119 +++++++ 28 files changed, 1541 insertions(+), 32 deletions(-) create mode 100644 CHANGELOG.d/0.88.0-analysis-run-start.md create mode 100644 backend/app/analysis_run_start.py create mode 100644 docs/adr/0021-authorized-analysis-run-start.md create mode 100644 migrations/0021_analysis_run_reconstruction.sql create mode 100644 migrations/0022_analysis_source_snapshot_member.sql create mode 100644 migrations/rollback/0021_analysis_run_reconstruction.sql create mode 100644 migrations/rollback/0022_analysis_source_snapshot_member.sql create mode 100644 tests/test_analysis_run_reconstruction_schema.py create mode 100644 tests/test_analysis_run_start.py diff --git a/AGENTS.md b/AGENTS.md index 47a71c8c3..cebc790a9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -91,6 +91,10 @@ A run-bearing analysis-run registry empties only after an unrevoked (ADR 0020 / v0.87.0). The documented phrase is not a secret. Do not expose purge on a public HTTP route. +`POST /api/analysis-runs/{id}/start` reconstructs a Pending lineage +cutoff bag through `reconstruct()` / `lineage_edge_specs` (ADR 0021 / +v0.88.0). TEPP and period-report start stay 422. Do not invent a theta. + ## CI gates `.github/workflows/tests.yml` runs the full suite on every PR to `main`. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bf19c5f77..96b775ba5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -472,10 +472,14 @@ run's scope whose `created_at` is at or before `knowledge_cutoff` without seeing later live rows or hidden bodies. Detail also returns revision and configuration digest prefixes. `POST /api/analysis-runs` records a Pending run on a new authorized -cutoff capture (ADR 0017): snapshot, counts, run, scope, and the first -status in one transaction. It does not reconstruct lineage and does not -invent a TEPP score. Request a lineage reconstruction from the home -list, then open the Pending row to confirm the cutoff corpus. +cutoff capture (ADR 0017): snapshot, counts, frozen membership, run, +scope, and the first status in one transaction. +`POST /api/analysis-runs/{id}/start` then runs ThreadWeave on that +frozen bag and persists run-scoped edges (ADR 0021). It does not invent +a TEPP score. Request a lineage reconstruction from the home list, open +the Pending row, then start reconstruction. Hover the Result digest +prefix, then confirm the designed A-100 fork before treating the live +Event Lineage panel as that run's tree. `make seed` also records a TEPP measurement run through `tepp_client` on that same snapshot; the default transport is unavailable, so that run is Failed rather than a fabricated score. @@ -492,7 +496,8 @@ calibrated negative result. A failed lineage row tells the operator to retry reconstruction, not to connect TEPP. A failed period-report row tells the operator to rebuild the report. A pending TEPP row does not claim a calibrated measurement. A pending lineage row -says reconstruction has not started yet. The +says reconstruction has not started yet; open it and start +reconstruction. The payload is lookup labels plus non-negative aggregate counts -- never source SQL, a DSN, a raw record, or a provider body. After `make seed`, Demo Analyst and Demo Admin see "Lineage reconstruction · Succeeded · diff --git a/CHANGELOG.d/0.88.0-analysis-run-start.md b/CHANGELOG.d/0.88.0-analysis-run-start.md new file mode 100644 index 000000000..a94209c26 --- /dev/null +++ b/CHANGELOG.d/0.88.0-analysis-run-start.md @@ -0,0 +1,5 @@ +# 0.88.0 start a pending lineage reconstruction + +`POST /api/analysis-runs/{id}/start` runs ThreadWeave on a Pending +lineage cutoff bag and persists run-scoped edges. Start reconstruction +from the open run. This path does not invent a TEPP measurement. diff --git a/CHANGELOG.md b/CHANGELOG.md index 00a19fe92..42239b071 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,20 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.88.0] - 2026-08-16 + +### Added + +- `POST /api/analysis-runs/{id}/start` runs ThreadWeave on a visible + Pending lineage cutoff bag and persists run-scoped parent choices + (ADR 0021). Open the Pending Demo Corp row, then start reconstruction. + The designed A-100 fork (revised quote and delivery question under the + pricing follow-up) is the acceptance tree. TEPP and period-report + start are 422 — this path does not invent a theta. A Succeeded retry + returns the stored digest. A Running restart is 409. Create freezes + authorized post ids so start cannot pick up a later backfill. Live + Event Lineage stays a separate rebuild. + ## [0.87.0] - 2026-08-16 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index 870c77f87..5096a8ea5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,4 +32,6 @@ Digest prefixes stay audible; hover a prefix to read the full digest. Opening a cutoff title shows the live post -- compare it with the cutoff before treating the body as reconstructed evidence (ADR 0016). `POST /api/analysis-runs` records Pending on an authorized -cutoff capture (ADR 0017) and does not reconstruct lineage. +cutoff capture (ADR 0017). `POST /api/analysis-runs/{id}/start` +reconstructs that frozen cutoff bag (ADR 0021) and does not invent a +theta. Hover the Result prefix to read the parent-choice digest. diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index d26eb6f6e..4fe53d760 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -6,9 +6,10 @@ lookup labels come back; source SQL, DSNs, raw records, and provider payloads never do. -``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, run, -scope, and the first Pending event atomically. It does not reconstruct -lineage or invent a TEPP score. +``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, frozen +membership, run, scope, and the first Pending event atomically. +``start_pending_analysis_run`` (ADR 0021) later reconstructs lineage on +that cutoff bag. Neither path invents a TEPP score. """ from __future__ import annotations @@ -247,9 +248,123 @@ async def fetch_visible_analysis_run( affiliated_entity_ids, row["knowledge_cutoff"], ) + digest, edges = await fetch_reconstructed_edges( + conn, + analysis_run_id, + affiliated_entity_ids, + ) + if digest is not None: + detail["reconstruction_result_sha256"] = digest + detail["reconstructed_edges"] = edges return detail +def reconstructed_edge_is_visible( + *, + parent_visibility_code: str, + parent_corporate_entity_id: Any, + child_visibility_code: str, + child_corporate_entity_id: Any, + affiliated_entity_ids: list[str], +) -> bool: + """Hide an edge when either endpoint is outside the caller's ABAC bag.""" + affiliated = {str(entity_id) for entity_id in affiliated_entity_ids} + parent_visible = ( + parent_visibility_code == "public" + or str(parent_corporate_entity_id) in affiliated + ) + child_visible = ( + child_visibility_code == "public" + or str(child_corporate_entity_id) in affiliated + ) + return parent_visible and child_visible + + +async def fetch_reconstructed_edges( + conn: asyncpg.Connection, + analysis_run_id: str, + affiliated_entity_ids: list[str], +) -> tuple[str | None, list[dict[str, Any]]]: + """Return the persisted digest and titled edges, or ``(None, [])``. + + Missing reconstruction tables mean this database has not applied + migration 0021 yet; treat that as no stored tree rather than 500. + Titles follow the same public-or-affiliated rule as ``visible_posts``. + """ + try: + header = await conn.fetchrow( + """ + select result_sha256 + from analysis_run_reconstruction + where analysis_run_id = $1 + """, + analysis_run_id, + ) + except asyncpg.UndefinedTableError: + return None, [] + if header is None: + return None, [] + rows = await conn.fetch( + """ + select + edge.parent_post_id, + parent_post.post_title as parent_post_title, + parent_post.visibility_code as parent_visibility_code, + parent_post.corporate_entity_id as parent_corporate_entity_id, + edge.child_post_id, + child_post.post_title as child_post_title, + child_post.visibility_code as child_visibility_code, + child_post.corporate_entity_id as child_corporate_entity_id, + edge.fused_score + from analysis_run_lineage_edge edge + join source_post parent_post on parent_post.post_id = edge.parent_post_id + join source_post child_post on child_post.post_id = edge.child_post_id + where edge.analysis_run_id = $1 + order by parent_post.post_title, child_post.post_title + """, + analysis_run_id, + ) + return header["result_sha256"], [ + { + "parent_post_id": str(row["parent_post_id"]), + "parent_post_title": row["parent_post_title"], + "child_post_id": str(row["child_post_id"]), + "child_post_title": row["child_post_title"], + "fused_score": float(row["fused_score"]), + } + for row in rows + if reconstructed_edge_is_visible( + parent_visibility_code=row["parent_visibility_code"], + parent_corporate_entity_id=row["parent_corporate_entity_id"], + child_visibility_code=row["child_visibility_code"], + child_corporate_entity_id=row["child_corporate_entity_id"], + affiliated_entity_ids=affiliated_entity_ids, + ) + ] + + +async def persist_snapshot_members( + conn: asyncpg.Connection, + snapshot_id: Any, + post_ids: list[str], +) -> None: + """Freeze authorized post ids on a new snapshot. Skip a legacy database.""" + if not post_ids: + return + try: + await conn.executemany( + """ + insert into analysis_source_snapshot_member + (analysis_source_snapshot_id, source_post_id) + values ($1, $2) + on conflict do nothing + """, + [(snapshot_id, post_id) for post_id in post_ids], + ) + except asyncpg.UndefinedTableError: + return + + async def fetch_visible_scope_posts( conn: asyncpg.Connection, scope_kind_code: str, @@ -441,7 +556,7 @@ async def create_pending_analysis_run( knowledge_cutoff: datetime | None, idempotency_key: str, ) -> dict[str, Any]: - """Insert snapshot, counts, run, scope, and Pending in one transaction. + """Insert snapshot, counts, frozen members, run, scope, and Pending. Does not reconstruct lineage and does not call TEPP. A missing measurement stays a later worker slice; this write only records the @@ -574,6 +689,7 @@ async def create_pending_analysis_run( capture.document_count, capture.thread_count, ) + await persist_snapshot_members(conn, snapshot_id, post_ids) try: run_id = await conn.fetchval( """ diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py new file mode 100644 index 000000000..0e9e3a2e7 --- /dev/null +++ b/backend/app/analysis_run_start.py @@ -0,0 +1,336 @@ +"""Start a Pending lineage reconstruction without inventing a TEPP score. + +ADR 0021. ``POST /api/analysis-runs/{id}/start`` transitions Pending to +Running, runs ThreadWeave on the frozen cutoff bag, persists run-scoped +edges, then stamps Succeeded. TEPP and period-report stay other paths. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +import asyncpg + +from backend.app.analysis_run_ingestion import ( + AnalysisRunCreateError, + fetch_visible_analysis_run, +) +from backend.app.lineage_ingestion import records_from_source_posts +from lineageweave.lineage_persistence import lineage_edge_specs +from lineageweave.models import Edge + +_LINEAGE_KIND = "analysis_run_lineage" +_TEPP_KIND = "analysis_run_tepp" +_REPORT_KIND = "analysis_run_report" +_PENDING = "analysis_status_pending" +_RUNNING = "analysis_status_running" +_SUCCEEDED = "analysis_status_succeeded" + + +class AnalysisRunStartError(AnalysisRunCreateError): + """Fail-closed start: HTTP status plus a next-action detail string.""" + + +def reconstruction_result_digest(edges: list[Edge]) -> str: + """SHA-256 of the ordered parent choices. Never hashes a post body.""" + material = json.dumps( + [ + { + "child_post_id": edge.child_id, + "fused_score": round(float(edge.fused_score), 6), + "parent_post_id": edge.parent_id, + } + for edge in sorted(edges, key=lambda item: (item.child_id, item.parent_id)) + ], + separators=(",", ":"), + sort_keys=True, + ) + return hashlib.sha256(material.encode()).hexdigest() + + +def start_kind_rejection(run_kind_code: str) -> AnalysisRunStartError | None: + """Return a 422 when start is not a lineage reconstruction. + + TEPP and period-report keep their own transports. This path must not + invent a theta or a calibrated report score. + """ + if run_kind_code == _LINEAGE_KIND: + return None + if run_kind_code == _TEPP_KIND: + return AnalysisRunStartError( + 422, + "Connect a TEPP transport from a Failed TEPP row. " + "This start path does not invent a measurement.", + ) + if run_kind_code == _REPORT_KIND: + return AnalysisRunStartError( + 422, + "Rebuild the period report from the reports panel. " + "This start path does not invent a measurement.", + ) + return AnalysisRunStartError( + 422, + "Start reconstructs a Pending lineage run only. " + "This start path does not invent a measurement.", + ) + + +def start_write_conflict_error() -> AnalysisRunStartError: + """Next action when a concurrent start already wrote this run.""" + return AnalysisRunStartError( + 409, + "Open this run. Refresh to see the stored tree if start already finished.", + ) + + +def reconstruction_member_ids( + snapshot_member_ids: list[str], + cutoff_post_ids: list[str], +) -> list[str]: + """Prefer create-time membership over a later cutoff re-query. + + An empty member list means this database has not frozen the bag yet + (migration 0022 missing, or a legacy snapshot). Start then uses the + live cutoff query so those rows still reconstruct. + """ + if snapshot_member_ids: + return list(snapshot_member_ids) + return list(cutoff_post_ids) + + +async def _cutoff_source_posts( + conn: asyncpg.Connection, + *, + corporate_entity_id: Any, + knowledge_cutoff: Any, + affiliated_entity_ids: list[str], +) -> list[asyncpg.Record]: + """ABAC-visible cutoff rows with the grouping keys reconstruct needs.""" + rows = await conn.fetch( + """ + select post_id, post_title, created_at, visibility_code, + corporate_entity_id, process_unit_id, + thread_group_key, secondary_grouping_key + from source_post + where corporate_entity_id = $1 and created_at <= $2 + order by created_at, post_title + """, + corporate_entity_id, + knowledge_cutoff, + ) + affiliated = {str(entity_id) for entity_id in affiliated_entity_ids} + return [ + row + for row in rows + if row["visibility_code"] == "public" + or str(row["corporate_entity_id"]) in affiliated + ] + + +async def _snapshot_member_posts( + conn: asyncpg.Connection, + snapshot_id: Any, +) -> list[asyncpg.Record]: + """Load frozen capture rows, or empty when the member table is absent.""" + try: + return list( + await conn.fetch( + """ + select post.post_id, post.post_title, post.created_at, + post.visibility_code, post.corporate_entity_id, + post.process_unit_id, post.thread_group_key, + post.secondary_grouping_key + from analysis_source_snapshot_member member + join source_post post on post.post_id = member.source_post_id + where member.analysis_source_snapshot_id = $1 + order by post.created_at, post.post_title + """, + snapshot_id, + ) + ) + except asyncpg.UndefinedTableError: + return [] + + +async def _append_status( + conn: asyncpg.Connection, + analysis_run_id: str, + status_ordinal: int, + status_code: str, + occurred_at: datetime, + failure_code: str | None = None, +) -> None: + """Append one legal lifecycle event. Failed rows carry a machine code.""" + await conn.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at, failure_code) + values ($1, $2, $3, $4, $5) + """, + analysis_run_id, + status_ordinal, + status_code, + occurred_at, + failure_code, + ) + + +async def _next_status_ordinal( + conn: asyncpg.Connection, + analysis_run_id: str, +) -> int: + """Return the next contiguous status ordinal for this run.""" + current_max = await conn.fetchval( + """ + select coalesce(max(status_ordinal), 0) + from analysis_run_status_event + where analysis_run_id = $1 + """, + analysis_run_id, + ) + return int(current_max) + 1 + + +async def start_pending_analysis_run( + conn: asyncpg.Connection, + *, + analysis_run_id: str, + account_id: str, + affiliated_entity_ids: list[str], +) -> dict[str, Any]: + """Run ThreadWeave on a visible Pending lineage row. + + TEPP and period-report are rejected so this path cannot invent a + theta. A Succeeded retry returns the stored reconstruction (documented + no-op replay). A Running or concurrent write is 409. Hidden runs 404. + The run row is locked before Running so a double-click is 409 or a + replay, never a 500. + """ + try: + UUID(analysis_run_id) + except ValueError as exc: + raise AnalysisRunStartError(404, "This analysis run is not visible.") from exc + + current = await fetch_visible_analysis_run( + conn, + analysis_run_id, + account_id, + affiliated_entity_ids, + ) + if current is None: + raise AnalysisRunStartError(404, "This analysis run is not visible.") + kind_error = start_kind_rejection(current["run_kind_code"]) + if kind_error is not None: + raise kind_error + if current["status_code"] == _SUCCEEDED: + return current + if current["status_code"] != _PENDING: + raise AnalysisRunStartError( + 409, + "Open this run. Start is only for a Pending lineage reconstruction.", + ) + + locked = await conn.fetchrow( + """ + select run.analysis_run_id, run.knowledge_cutoff, + run.analysis_source_snapshot_id, scope.corporate_entity_id + from analysis_run run + join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id + where run.analysis_run_id = $1 + for update of run + """, + analysis_run_id, + ) + locked_status = await conn.fetchval( + """ + select status_code + from analysis_run_current_status + where analysis_run_id = $1 + """, + analysis_run_id, + ) + if locked_status == _SUCCEEDED: + replayed = await fetch_visible_analysis_run( + conn, + analysis_run_id, + account_id, + affiliated_entity_ids, + ) + if replayed is None: + raise AnalysisRunStartError(404, "This analysis run is not visible.") + return replayed + if locked_status != _PENDING: + raise AnalysisRunStartError( + 409, + "Open this run. Start is only for a Pending lineage reconstruction.", + ) + + now = datetime.now(timezone.utc) + running_ordinal = await _next_status_ordinal(conn, analysis_run_id) + try: + await _append_status(conn, analysis_run_id, running_ordinal, _RUNNING, now) + member_rows = await _snapshot_member_posts( + conn, + locked["analysis_source_snapshot_id"], + ) + if member_rows: + rows = member_rows + else: + rows = await _cutoff_source_posts( + conn, + corporate_entity_id=locked["corporate_entity_id"], + knowledge_cutoff=locked["knowledge_cutoff"], + affiliated_entity_ids=affiliated_entity_ids, + ) + edges = lineage_edge_specs(records_from_source_posts(rows)) + digest = reconstruction_result_digest(edges) + finished = datetime.now(timezone.utc) + if finished < now: + finished = now + await conn.execute( + """ + insert into analysis_run_reconstruction + (analysis_run_id, result_sha256, edge_count, reconstructed_at) + values ($1, $2, $3, $4) + """, + analysis_run_id, + digest, + len(edges), + finished, + ) + for edge in edges: + await conn.execute( + """ + insert into analysis_run_lineage_edge + (analysis_run_id, child_post_id, parent_post_id, + fused_score, reconstructed_at) + values ($1, $2, $3, $4, $5) + """, + analysis_run_id, + edge.child_id, + edge.parent_id, + edge.fused_score, + finished, + ) + await _append_status( + conn, + analysis_run_id, + running_ordinal + 1, + _SUCCEEDED, + finished, + ) + except asyncpg.UniqueViolationError as exc: + raise start_write_conflict_error() from exc + started = await fetch_visible_analysis_run( + conn, + analysis_run_id, + account_id, + affiliated_entity_ids, + ) + if started is None: + raise AnalysisRunStartError(404, "This analysis run is not visible.") + return started diff --git a/backend/app/main.py b/backend/app/main.py index adb7a20a8..b3da2497f 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -72,6 +72,10 @@ fetch_visible_analysis_run, fetch_visible_analysis_runs, ) +from backend.app.analysis_run_start import ( + AnalysisRunStartError, + start_pending_analysis_run, +) from backend.app.activity_stream import ( create_valkey_client, get_valkey, @@ -1253,6 +1257,33 @@ async def create_analysis_run( return created +@app.post("/api/analysis-runs/{analysis_run_id}/start") +async def start_analysis_run( + analysis_run_id: str, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """Start ThreadWeave on a visible Pending lineage run. + + post_read is enough. Hidden runs 404. TEPP and period-report are 422 + so this path cannot invent a theta. A Succeeded retry returns the + stored tree. A Running restart is 409. + """ + _require_post_read(account) + async with pool.acquire() as conn: + async with conn.transaction(): + try: + started = await start_pending_analysis_run( + conn, + analysis_run_id=analysis_run_id, + account_id=account.user_account_id, + affiliated_entity_ids=list(account.corporate_entity_ids), + ) + except AnalysisRunStartError as exc: + raise HTTPException(exc.status_code, exc.detail) from exc + return started + + @app.get("/api/analysis-runs/{analysis_run_id}") async def read_analysis_run( analysis_run_id: str, diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 3b74c22a3..ef147e99f 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -33,6 +33,12 @@ _MIGRATION_PATH = Path(__file__).resolve().parents[2] / "migrations" / "0001_initial_schema.sql" _REGISTRY_MIGRATION = Path(__file__).resolve().parents[2] / "migrations" / "0018_analysis_run_registry.sql" _RETENTION_MIGRATION = Path(__file__).resolve().parents[2] / "migrations" / "0020_analysis_run_retention_purge.sql" +_RECONSTRUCTION_MIGRATION = ( + Path(__file__).resolve().parents[2] / "migrations" / "0021_analysis_run_reconstruction.sql" +) +_SNAPSHOT_MEMBER_MIGRATION = ( + Path(__file__).resolve().parents[2] / "migrations" / "0022_analysis_source_snapshot_member.sql" +) def _postgres_available() -> bool: @@ -117,6 +123,8 @@ def seeded_db(demo_analyst_token): cur.execute(_MIGRATION_PATH.read_text()) cur.execute(_REGISTRY_MIGRATION.read_text()) cur.execute(_RETENTION_MIGRATION.read_text()) + cur.execute(_RECONSTRUCTION_MIGRATION.read_text()) + cur.execute(_SNAPSHOT_MEMBER_MIGRATION.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " @@ -575,6 +583,216 @@ def test_create_analysis_run_records_pending_without_inventing_a_score( assert unauthenticated.status_code == 401 +def test_start_analysis_run_recovers_the_a100_fork( + client, demo_analyst_token, seeded_db +) -> None: + """Starting a Pending lineage run persists the designed fixture tree.""" + from scripts.seed_demo_data import insert_fixture_source_posts + + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) " + "values ('voc_type', 'vom', 'Voice of Market') " + "on conflict (lookup_code) do nothing" + ) + cur.execute( + "insert into process_unit (corporate_entity_id, process_unit_code, process_unit_name) " + "select corporate_entity_id, 'TEST-PU-START', 'Start reconstruction' " + "from source_post where post_id = %s returning process_unit_id", + (seeded_db["own_private_post_id"],), + ) + process_unit_id = cur.fetchone()[0] + cur.execute( + "select author_account_id, corporate_entity_id from source_post where post_id = %s", + (seeded_db["own_private_post_id"],), + ) + author_id, corp_id = cur.fetchone() + insert_fixture_source_posts(cur, author_id, corp_id, process_unit_id) + finally: + admin_conn.close() + + created = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_lineage", + "corporate_entity_id": seeded_db["own_corp_id"], + "knowledge_cutoff": "2026-02-15T00:00:00Z", + "idempotency_key": "buyer-start-2026-w07", + }, + ) + assert created.status_code == 201, created.text + run_id = created.json()["analysis_run_id"] + assert created.json()["status_label"] == "Pending" + + started = client.post( + f"/api/analysis-runs/{run_id}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert started.status_code == 200, started.text + body = started.json() + assert body["status_label"] == "Succeeded" + assert all(event["status_label"] != "Failed" for event in body["status_history"]) + assert body["reconstruction_result_sha256"] + children = { + edge["child_post_title"] + for edge in body["reconstructed_edges"] + if edge["parent_post_title"] == "Pricing renegotiation follow-up" + } + assert "Pricing renegotiation: revised quote sent" in children + assert "Delivery schedule question raised" in children + assert "theta" not in str(body).lower() + + replay = client.post( + f"/api/analysis-runs/{run_id}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert replay.status_code == 200 + assert replay.json()["reconstruction_result_sha256"] == body["reconstruction_result_sha256"] + + tepp = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_tepp", + "corporate_entity_id": seeded_db["own_corp_id"], + "knowledge_cutoff": "2026-02-15T00:00:00Z", + "idempotency_key": "buyer-start-tepp-2026-w07", + }, + ) + assert tepp.status_code == 201 + refused = client.post( + f"/api/analysis-runs/{tepp.json()['analysis_run_id']}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert refused.status_code == 422 + assert "invent a measurement" in refused.json()["detail"] + + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z') + returning analysis_source_snapshot_id + """, + ("9" * 64,), + ) + report_snapshot_id = cur.fetchone()[0] + cur.execute( + "select requested_by_account_id from analysis_run where analysis_run_id = %s", + (run_id,), + ) + requester_id = cur.fetchone()[0] + 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_report', 'buyer-start-report', + %s, '2026-01-12T12:00:00Z', 'lineage-run-v1', %s, %s, + '2026-01-12T12:30:00Z') + returning analysis_run_id + """, + (report_snapshot_id, requester_id, "8" * 64, "7" * 40), + ) + report_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) + """, + (report_run_id, seeded_db["own_corp_id"]), + ) + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 1, 'analysis_status_pending', '2026-01-12T12:31:00Z') + """, + (report_run_id,), + ) + cur.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z') + returning analysis_source_snapshot_id + """, + ("6" * 64,), + ) + running_snapshot_id = cur.fetchone()[0] + 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_lineage', 'buyer-start-running', + %s, '2026-01-12T12:00:00Z', 'lineage-run-v1', %s, %s, + '2026-01-12T12:30:00Z') + returning analysis_run_id + """, + (running_snapshot_id, requester_id, "5" * 64, "4" * 40), + ) + running_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) + """, + (running_run_id, seeded_db["own_corp_id"]), + ) + for ordinal, status, occurred in ( + (1, "analysis_status_pending", "2026-01-12T12:31:00Z"), + (2, "analysis_status_running", "2026-01-12T12:32:00Z"), + ): + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, %s, %s, %s) + """, + (running_run_id, ordinal, status, occurred), + ) + finally: + admin_conn.close() + + report_refused = client.post( + f"/api/analysis-runs/{report_run_id}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert report_refused.status_code == 422 + assert "invent a measurement" in report_refused.json()["detail"] + + running = client.post( + f"/api/analysis-runs/{running_run_id}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert running.status_code == 409 + + hidden = client.post( + f"/api/analysis-runs/{seeded_db['hidden_run_id']}/start", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert hidden.status_code == 404 + + def test_me_reflects_the_authenticated_account(client, demo_analyst_token) -> None: response = client.get("/api/me", headers={"Authorization": f"Bearer {demo_analyst_token}"}) assert response.status_code == 200 diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index ce2f0e6b5..71e9fc733 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -26,6 +26,8 @@ COPY migrations/0017_prov_o_standard_relations.sql /docker-entrypoint-initdb.d/1 COPY migrations/0018_analysis_run_registry.sql /docker-entrypoint-initdb.d/19-analysis-run-registry.sql COPY migrations/0019_role_catalog_identity.sql /docker-entrypoint-initdb.d/20-role-catalog-identity.sql COPY migrations/0020_analysis_run_retention_purge.sql /docker-entrypoint-initdb.d/21-analysis-run-retention-purge.sql +COPY migrations/0021_analysis_run_reconstruction.sql /docker-entrypoint-initdb.d/22-analysis-run-reconstruction.sql +COPY migrations/0022_analysis_source_snapshot_member.sql /docker-entrypoint-initdb.d/23-analysis-source-snapshot-member.sql # Official image already drops to this account at runtime; declare it so # the Dockerfile itself satisfies DS-0002 (explicit non-root USER). USER postgres diff --git a/docs/adr/0013-normalized-analysis-run-registry.md b/docs/adr/0013-normalized-analysis-run-registry.md index d13994bc5..afc601897 100644 --- a/docs/adr/0013-normalized-analysis-run-registry.md +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -239,8 +239,11 @@ Acceptance requires: 1. Add a transaction repository that creates snapshot, counts, run, scope, and first status atomically and compares request digests on idempotent retries. - `POST /api/analysis-runs` now records that Pending write (ADR 0017); - reconstruction and live TEPP execution remain later slices. + `POST /api/analysis-runs` now records that Pending write (ADR 0017). + `POST /api/analysis-runs/{id}/start` now reconstructs a Pending + lineage cutoff bag in-process from frozen snapshot membership + (ADR 0021). A durable outbox / Valkey worker and live TEPP execution + remain later slices. 2. Add RBAC/ABAC-protected run list/detail endpoints and the DB-grounded read-only administrator surface. 3. Add a normalized PostgreSQL outbox and Valkey delivery worker. diff --git a/docs/adr/0017-authorized-analysis-run-create.md b/docs/adr/0017-authorized-analysis-run-create.md index e3a535a18..be7328877 100644 --- a/docs/adr/0017-authorized-analysis-run-create.md +++ b/docs/adr/0017-authorized-analysis-run-create.md @@ -22,7 +22,8 @@ still owns reconstruction and live TEPP execution. they already walk. An unaffiliated corp is 404, not 403. - The capture digest hashes scope, entity, cutoff, and authorized post ids — never a post body, DSN, or source SQL. -- The write inserts snapshot, aggregate counts, `analysis_run`, +- The write inserts snapshot, aggregate counts, frozen + `analysis_source_snapshot_member` ids, `analysis_run`, `analysis_run_scope`, and `analysis_status_pending` in one transaction. - The first status is Pending. This slice does not reconstruct lineage and does not call TEPP. A missing measurement stays Failed only on the @@ -35,9 +36,10 @@ still owns reconstruction and live TEPP execution. ## Consequences The home panel's **Request a lineage reconstruction** button records a -Pending row the operator can open immediately. Reconstruction, TEPP -transport, and the outbox worker remain later slices. Do not stamp -Succeeded or invent a theta from this write. +Pending row the operator can open immediately. `POST +/api/analysis-runs/{id}/start` then reconstructs that frozen bag +(ADR 0021). TEPP transport and the outbox worker remain later slices. +Do not stamp Succeeded or invent a theta from this write. ## References — APA 7th diff --git a/docs/adr/0021-authorized-analysis-run-start.md b/docs/adr/0021-authorized-analysis-run-start.md new file mode 100644 index 000000000..61db36679 --- /dev/null +++ b/docs/adr/0021-authorized-analysis-run-start.md @@ -0,0 +1,112 @@ +# ADR 0021 — Operators start a pending lineage reconstruction + +**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Date:** 2026-08-16 +**Depends on:** ADR 0013 registry; ADR 0014 authorized read; ADR 0016 cutoff +posts; ADR 0017 authorized create +**Refs:** Issue #79 (Milestone 2 parent); ADR 0013 follow-up 3 (in-process +start; durable outbox remains later) + +## Context + +ADR 0017 let an operator record a Pending analysis run. The home button +said “Request a lineage reconstruction,” then the row stayed Pending. +Seed still owned the only Succeeded Demo Corp tree. A buyer cannot +treat a request they cannot start as a product. + +ADR 0013 follow-up 3 asked for a PostgreSQL outbox and Valkey worker. +That durable delivery path is still later. This slice starts +reconstruction in the authorized request so the operator can see the +cutoff tree immediately. A crash after Running and before Succeeded +rolls the transaction back to Pending. + +Landed #145 occupies ADR 0020 / package 0.87.0 for granted retention +purge. ADR 0019 binds R&R catalog identity. This decision is the next +free slot. + +## Decision + +`POST /api/analysis-runs/{id}/start` requires `post_read` and, in one +transaction: + +1. loads the authorized run (hidden scopes 404); +2. rejects non-lineage kinds so TEPP and period-report cannot invent a + theta or a calibrated score; +3. replays a Succeeded run (documented no-op; same stored digest); +4. accepts only Pending lineage — Running is 409; +5. locks the run row, re-reads status, appends Running, runs + `lineage_edge_specs` / `reconstruct()` on the frozen + `analysis_source_snapshot_member` bag (or the live cutoff query when + membership was never persisted), persists + `analysis_run_reconstruction` plus `analysis_run_lineage_edge`, then + appends Succeeded. A concurrent start is 409 with a refresh next + action, not a 500. + +```mermaid +sequenceDiagram + participant Operator + participant API + participant ThreadWeave + participant Registry + Operator->>API: POST /api/analysis-runs/{id}/start + API->>Registry: lock visible Pending lineage run + alt TEPP or period-report + API-->>Operator: 422 use the kind's own path + else already Succeeded + Registry-->>API: stored edges + API-->>Operator: 200 replay + else Running + API-->>Operator: 409 refresh + else Pending lineage + Registry->>Registry: Running + API->>ThreadWeave: reconstruct frozen bag + ThreadWeave-->>API: parent choices + Registry->>Registry: reconstruction + edges + Succeeded + API-->>Operator: 200 titled edges + end +``` + +Rules: + +- Edges are run-scoped. This write does not replace live + `post_lineage_edge` (the Event Lineage panel stays a later rebuild). +- The digest hashes parent id, child id, and rounded fused score — never + a post body, DSN, or image. +- Empty cutoff bags Succeed with zero edges. +- Failed TEPP remains a `tepp_client` transport problem. +- Create freezes authorized post ids on + `analysis_source_snapshot_member` so start cannot pick up a later + backfill that shares the cutoff clock. + +The home detail adds **Start reconstruction** on a Pending lineage row +and lists titled parent→child edges after Succeeded. The Result digest +prefix is audible next to Code and Config; hover it to verify the +parent-choice hash. Edge titles stay public-or-affiliated. TEPP and +period-report rows do not show the button. + +## Consequences + +Demo Analyst can request a run, start it, and confirm the designed A-100 +fork (revised quote and delivery question under the pricing follow-up) +without a seed-only Succeeded row. The durable outbox / Valkey worker +and live TEPP transport remain later slices. Do not stamp Succeeded +from a missing reconstruct library, and do not invent a theta. + +## References — APA 7th + +International Organization for Standardization. (2019). *ISO 8601-1:2019: +Date and time—Representations for information interchange—Part 1: Basic +rules* (confirmed 2024; Amendment 1:2022). + +Jensen, C. S., & Snodgrass, R. T. (1999). Temporal data management. +*IEEE Transactions on Knowledge and Data Engineering, 11*(1), 36–44. +https://doi.org/10.1109/69.755613 + +Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*. +World Wide Web Consortium. https://www.w3.org/TR/prov-dm/ + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C +Recommendation). https://www.w3.org/TR/prov-o/ + +World Wide Web Consortium. (2022). *Time ontology in OWL* (W3C +Recommendation). https://www.w3.org/TR/owl-time/ diff --git a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md index c776053b1..6d3427fa1 100644 --- a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md +++ b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md @@ -1,7 +1,8 @@ # Analysis-run registry standards and research traceability **Status:** Active PR evidence; not protected-main truth until merge. -**Scope:** Migrations 0018 and 0020, ADR 0013 / 0020, rollback, and real-PostgreSQL contract tests. +**Scope:** Migrations 0018–0022, ADR 0013 / 0017 / 0020 / 0021, rollback, and +real-PostgreSQL contract tests. ## Standards mapped to implementation @@ -14,7 +15,8 @@ | PostgreSQL 18 constraints and trigger contracts | Put integrity close to durable truth and use constraints for row shape while triggers enforce cross-row state and serialization. | Digest/check constraints, category allowlists, account-scoped uniqueness, shape constraints, immutable-row triggers, shared snapshot-row locking, and serialized status transitions. | | NIST SP 800-92 | Treat audit records as bounded, protected operational evidence rather than unstructured application logging. | Append-only status events, machine failure codes, actor identity, occurrence/record clocks, fail-closed rollback, `invoking_session_role` on each retention event, and exclusion of raw source/provider payloads. | | NIST SP 800-53 Rev. 5 AC-3 | Enforce least privilege on privileged procedures; a well-known procedure name is not an authorization secret. | `REVOKE ALL` on `purge_analysis_run_registry` from `PUBLIC`; `GRANT EXECUTE` only to `analysis_run_retention_admin`; unrevoked `analysis_run_retention_grant` required (ADR 0020). | -| OpenAPI 3.2.0 | Define explicit versioned API schemas rather than exposing database rows or implementation-specific payloads. | API intentionally deferred; ADR 0013 requires a source-redacting run list/detail contract before a product surface is claimed. | +| OpenAPI 3.2.0 | Define explicit versioned API schemas rather than exposing database rows or implementation-specific payloads. | `GET` / `POST /api/analysis-runs` and `POST /api/analysis-runs/{id}/start` return labels, clocks, aggregates, and titled reconstruction edges — never source SQL or a provider body. | +| ThreadWeave tree assembly | Persist the same parent choices the library reconstructs on the cutoff bag. | `start_pending_analysis_run` calls `lineage_edge_specs` on frozen `analysis_source_snapshot_member` rows (or the live cutoff query when membership is absent); tests require the designed A-100 fork through `records_from_source_posts` (revised quote + delivery question under the pricing follow-up). | ## Temporal reasoning @@ -77,12 +79,16 @@ provenance, retention, and immutable evidence rather than blanket masking. | Idempotency is actor-scoped | Permit identical opaque keys for two accounts and reject reuse by the same account. | | Lifecycle is ordered | Require pending first, contiguous ordinals, monotonic time, legal transitions, terminal finality, and append-only rows. | | Rollback does not erase audit data silently | Reject 0018 rollback with any registry rows. A run-bearing registry empties only through an unrevoked `analysis_run_retention_grant` plus `analysis_run_retention_admin`, then `purge_analysis_run_registry('approved-retention-purge')`; a wrong token, a raw `DELETE`, and a runtime role that only knows the public phrase stay rejected. Export then delete `analysis_run_retention_event` before 0020 rollback. | +| Start reconstruction recovers the designed tree | Persist edges from `lineage_edge_specs` on the A-100 fixture bag via `records_from_source_posts`; the pricing follow-up must parent both the revised quote and the delivery question. A TEPP or period-report start must 422 without a theta. Snapshot members exclude a later backfill. A concurrent or Running start is 409. A Succeeded retry returns the stored digest. | ## APA 7th references American Institute of Certified Public Accountants. (2017). *SOC 2®: SOC for Service Organizations: Trust Services Criteria*. +ContextualWisdomLab. (2026). *ThreadWeave* [Computer software]. +https://github.com/ContextualWisdomLab/ThreadWeave + International Organization for Standardization. (2016). *ISO 15489-1:2016: Information and documentation—Records management—Part 1: Concepts and principles*. diff --git a/frontend/package.json b/frontend/package.json index 0d43d9fa2..4c66c7205 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.87.0", + "version": "0.88.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index fd8a15146..b06693250 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -329,6 +329,61 @@ describe("App, authenticated", () => { }), ); } + if (url.endsWith("/api/analysis-runs/run-demo-lineage-pending/start") && method === "POST") { + return Promise.resolve( + jsonResponse({ + analysis_run_id: "run-demo-lineage-pending", + run_kind_code: "analysis_run_lineage", + run_kind_label: "Lineage reconstruction", + scope_kind_code: "analysis_scope_corporate_entity", + scope_kind_label: "Corporate entity", + scope_entity_name: "Demo Corp", + status_code: "analysis_status_succeeded", + status_label: "Succeeded", + knowledge_cutoff: "2026-01-12T12:00:00Z", + requested_at: "2026-01-12T12:35:00Z", + source_counts: [], + visible_posts: [{ post_id: "post-1", post_title: "Public post" }], + reconstructed_edges: [ + { + parent_post_id: "post-follow-up", + parent_post_title: "Pricing renegotiation follow-up", + child_post_id: "post-quote", + child_post_title: "Pricing renegotiation: revised quote sent", + fused_score: 0.72, + }, + { + parent_post_id: "post-follow-up", + parent_post_title: "Pricing renegotiation follow-up", + child_post_id: "post-delivery", + child_post_title: "Delivery schedule question raised", + fused_score: 0.68, + }, + ], + reconstruction_result_sha256: "aa".repeat(32), + status_history: [ + { + status_ordinal: 1, + status_code: "analysis_status_pending", + status_label: "Pending", + occurred_at: "2026-01-12T12:35:00Z", + }, + { + status_ordinal: 2, + status_code: "analysis_status_running", + status_label: "Running", + occurred_at: "2026-01-12T12:36:00Z", + }, + { + status_ordinal: 3, + status_code: "analysis_status_succeeded", + status_label: "Succeeded", + occurred_at: "2026-01-12T12:37:00Z", + }, + ], + }), + ); + } if (url.endsWith("/api/analysis-runs") && method === "POST") { const created = { analysis_run_id: "run-demo-lineage-pending", @@ -343,6 +398,7 @@ describe("App, authenticated", () => { requested_at: "2026-01-12T12:35:00Z", source_counts: [], visible_posts: [{ post_id: "post-1", post_title: "Public post" }], + reconstructed_edges: [], status_history: [ { status_ordinal: 1, @@ -1754,6 +1810,7 @@ describe("App, authenticated", () => { expect(reportButton).not.toHaveTextContent("reconstruction"); await userEvent.click(reportButton); + expect(screen.queryByRole("button", { name: "Start reconstruction" })).not.toBeInTheDocument(); expect( await screen.findByText( "No posts were available at this cutoff for the period report. Open a later run, or ask an administrator to capture a newer snapshot.", @@ -1776,6 +1833,7 @@ describe("App, authenticated", () => { expect(screen.queryByText(/replace Failed/i)).not.toBeInTheDocument(); expect(screen.queryByText(/this TEPP run measured/i)).not.toBeInTheDocument(); expect(screen.queryByText(/Reconstruction has not started yet/)).not.toBeInTheDocument(); + expect(screen.queryByRole("button", { name: "Start reconstruction" })).not.toBeInTheDocument(); }); it("does not tell a succeeded TEPP run to replace Failed", async () => { @@ -1810,9 +1868,10 @@ describe("App, authenticated", () => { ).toBeInTheDocument(); expect( screen.getAllByText( - "Open this run to confirm which posts it will use. Reconstruction has not started yet.", + "Open this run, then start reconstruction. Reconstruction has not started yet.", ), ).toHaveLength(2); + expect(screen.getByRole("button", { name: "Start reconstruction" })).toBeInTheDocument(); const postCall = fetchMock.mock.calls.find( (call) => String(call[0]).endsWith("/api/analysis-runs") && call[1]?.method === "POST", ); @@ -1824,6 +1883,34 @@ describe("App, authenticated", () => { ); }); + it("starts reconstruction and shows the designed A-100 fork", async () => { + const fetchMock = stubBackend(); + render(); + + await userEvent.click( + await screen.findByRole("button", { name: "Request a lineage reconstruction" }), + ); + await userEvent.click(await screen.findByRole("button", { name: "Start reconstruction" })); + expect( + await screen.findByRole("heading", { name: "Lineage reconstruction · Succeeded · Demo Corp" }), + ).toBeInTheDocument(); + expect( + screen.getByText( + "Pricing renegotiation: revised quote sent follows Pricing renegotiation follow-up", + ), + ).toBeInTheDocument(); + expect( + screen.getByText("Delivery schedule question raised follows Pricing renegotiation follow-up"), + ).toBeInTheDocument(); + const digests = screen.getByLabelText("Analysis run reproducibility digests"); + expect(digests).toHaveTextContent("Result aaaaaaaaaaaa"); + expect(screen.getByTitle("aa".repeat(32))).toHaveTextContent("Result aaaaaaaaaaaa"); + const startCall = fetchMock.mock.calls.find((call) => + String(call[0]).endsWith("/api/analysis-runs/run-demo-lineage-pending/start"), + ); + expect(startCall?.[1]?.method).toBe("POST"); + }); + it("shows the calibrated period-report mean theta on the home page", async () => { stubBackend(); render(); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 07088e9d4..0296fb6ba 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -4,6 +4,7 @@ import { askPostChat, BackendError, createAnalysisRun, + startAnalysisRun, createPostTicket, deriveCommitment, evaluatePost, @@ -1459,7 +1460,7 @@ function analysisRunNextAction(run: AnalysisRun): string | null { case "analysis_status_pending": switch (run.run_kind_code) { case "analysis_run_lineage": - return "Open this run to confirm which posts it will use. Reconstruction has not started yet."; + 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_report": @@ -1583,11 +1584,23 @@ function analysisRunLivePostButtonLabel(postTitle: string): string { function AnalysisRunReproducibilityDigests({ codeRevisionSha, configurationSha256, + reconstructionResultSha256, }: { codeRevisionSha?: string; configurationSha256?: string; + reconstructionResultSha256?: string; }) { - if (!codeRevisionSha && !configurationSha256) { + const parts: { label: string; digest: string }[] = []; + if (codeRevisionSha) { + parts.push({ label: "Code", digest: codeRevisionSha }); + } + if (configurationSha256) { + parts.push({ label: "Config", digest: configurationSha256 }); + } + if (reconstructionResultSha256) { + parts.push({ label: "Result", digest: reconstructionResultSha256 }); + } + if (parts.length === 0) { return null; } return ( @@ -1596,20 +1609,29 @@ function AnalysisRunReproducibilityDigests({ Hover a prefix to read the full digest for verification.{" "} - {codeRevisionSha ? ( - {`Code ${analysisRunDigestPrefix(codeRevisionSha)}`} - ) : null} - {codeRevisionSha && configurationSha256 ? " · " : null} - {configurationSha256 ? ( - - {`Config ${analysisRunDigestPrefix(configurationSha256)}`} + {parts.map((part, index) => ( + + {index > 0 ? " · " : null} + {`${part.label} ${analysisRunDigestPrefix(part.digest)}`} - ) : null} + ))}

); } +/** + * Start is only for a Pending Demo Corp lineage row after Request. + * + * TEPP and period-report keep their own transports. This button must + * not appear on those kinds. + */ +function analysisRunCanStartReconstruction(run: AnalysisRun): boolean { + return ( + run.run_kind_code === "analysis_run_lineage" && run.status_code === "analysis_status_pending" + ); +} + function AnalysisRunsPanel({ accessToken, onSelectPost, @@ -1621,6 +1643,7 @@ function AnalysisRunsPanel({ const [selected, setSelected] = useState(null); const [error, setError] = useState(null); const [requesting, setRequesting] = useState(false); + const [starting, setStarting] = useState(false); useEffect(() => { fetchAnalysisRuns(accessToken) @@ -1646,6 +1669,22 @@ function AnalysisRunsPanel({ } } + async function handleStartReconstruction() { + if (!selected) return; + setError(null); + setStarting(true); + try { + const started = await startAnalysisRun(accessToken, selected.analysis_run_id); + const listed = await fetchAnalysisRuns(accessToken); + setRuns(listed.analysis_runs); + setSelected(started); + } catch (err) { + setError(err instanceof BackendError ? err.message : String(err)); + } finally { + setStarting(false); + } + } + async function handleOpen(runId: string) { setError(null); try { @@ -1725,7 +1764,27 @@ function AnalysisRunsPanel({ + {analysisRunCanStartReconstruction(selected) && ( + + )} + {selected.reconstructed_edges && selected.reconstructed_edges.length > 0 && ( +
    + {selected.reconstructed_edges.map((edge) => ( +
  • + {edge.child_post_title} follows {edge.parent_post_title} +
    • + ))} +
+ )}
    {selected.source_counts.map((count) => (
  • diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 3385d5179..0213c3fcd 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -536,6 +536,14 @@ export interface AnalysisRunStatusEvent { failure_code?: string; } +export interface AnalysisRunReconstructedEdge { + parent_post_id: string; + parent_post_title: string; + child_post_id: string; + child_post_title: string; + fused_score: number; +} + export interface AnalysisRun { analysis_run_id: string; run_kind_code: AnalysisRunKindCode; @@ -550,6 +558,8 @@ export interface AnalysisRun { source_counts: AnalysisRunCount[]; status_history?: AnalysisRunStatusEvent[]; visible_posts?: { post_id: string; post_title: string }[]; + reconstructed_edges?: AnalysisRunReconstructedEdge[]; + reconstruction_result_sha256?: string; code_revision_sha?: string; configuration_sha256?: string; } @@ -579,3 +589,12 @@ export function createAnalysisRun( body: JSON.stringify(request), }); } + +export function startAnalysisRun( + accessToken: string, + analysisRunId: string, +): Promise { + return backendFetch(`/api/analysis-runs/${analysisRunId}/start`, accessToken, { + method: "POST", + }); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 1950c39f8..036dca1fa 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.87.0" +__version__ = "0.88.0" diff --git a/migrations/0021_analysis_run_reconstruction.sql b/migrations/0021_analysis_run_reconstruction.sql new file mode 100644 index 000000000..8c66a584f --- /dev/null +++ b/migrations/0021_analysis_run_reconstruction.sql @@ -0,0 +1,79 @@ +-- Run-scoped lineage reconstruction result (ADR 0021). +-- +-- A Pending analysis run may later persist the ThreadWeave parent choices +-- for its cutoff bag. Edges belong to the run, not the live Event Lineage +-- panel. No post body, DSN, or fabricated measurement is stored. + +create table if not exists analysis_run_reconstruction ( + analysis_run_id uuid primary key + references analysis_run (analysis_run_id), + result_sha256 text not null, + edge_count integer not null, + reconstructed_at timestamptz not null, + recorded_at timestamptz not null default clock_timestamp(), + constraint analysis_run_reconstruction_digest_check + check (result_sha256 ~ '^[0-9a-f]{64}$'), + constraint analysis_run_reconstruction_edge_count_check + check (edge_count >= 0), + constraint analysis_run_reconstruction_time_check + check (reconstructed_at <= recorded_at) +); + +comment on table analysis_run_reconstruction is + 'One immutable reconstruction digest per analysis run; never a post body ' + 'or a fabricated psychometric score.'; + +create table if not exists analysis_run_lineage_edge ( + analysis_run_id uuid not null + references analysis_run_reconstruction (analysis_run_id), + child_post_id uuid not null + references source_post (post_id), + parent_post_id uuid not null + references source_post (post_id), + fused_score double precision not null, + reconstructed_at timestamptz not null, + primary key (analysis_run_id, child_post_id), + constraint analysis_run_lineage_edge_distinct_check + check (child_post_id <> parent_post_id), + constraint analysis_run_lineage_edge_score_check + check (fused_score >= 0 and fused_score <= 1) +); + +comment on table analysis_run_lineage_edge is + 'One reconstructed parent choice per child post inside one analysis run.'; + +create or replace function reject_analysis_run_reconstruction_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_reconstruction_is_immutable'; +end +$$; + +comment on function reject_analysis_run_reconstruction_update() is + 'Rejects mutation of a persisted reconstruction digest.'; + +drop trigger if exists analysis_run_reconstruction_update_reject + on analysis_run_reconstruction; +create trigger analysis_run_reconstruction_update_reject +before update or delete on analysis_run_reconstruction +for each row execute function reject_analysis_run_reconstruction_update(); + +create or replace function reject_analysis_run_lineage_edge_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_lineage_edge_is_immutable'; +end +$$; + +comment on function reject_analysis_run_lineage_edge_update() is + 'Rejects mutation of a persisted run-scoped lineage edge.'; + +drop trigger if exists analysis_run_lineage_edge_update_reject + on analysis_run_lineage_edge; +create trigger analysis_run_lineage_edge_update_reject +before update or delete on analysis_run_lineage_edge +for each row execute function reject_analysis_run_lineage_edge_update(); diff --git a/migrations/0022_analysis_source_snapshot_member.sql b/migrations/0022_analysis_source_snapshot_member.sql new file mode 100644 index 000000000..1ea20fc3d --- /dev/null +++ b/migrations/0022_analysis_source_snapshot_member.sql @@ -0,0 +1,35 @@ +-- Create-time cutoff membership for an analysis source snapshot (ADR 0021). +-- +-- The snapshot digest already hashes authorized post ids. This relation +-- stores those ids so start reconstructs the same bag, not a later +-- backfill that shares the cutoff clock. No post body is stored. + +create table if not exists analysis_source_snapshot_member ( + analysis_source_snapshot_id uuid not null + references analysis_source_snapshot (analysis_source_snapshot_id), + source_post_id uuid not null + references source_post (post_id), + primary key (analysis_source_snapshot_id, source_post_id) +); + +comment on table analysis_source_snapshot_member is + 'Authorized post ids frozen at snapshot capture; start reconstructs ' + 'these rows and never a later backfill.'; + +create or replace function reject_analysis_source_snapshot_member_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_source_snapshot_member_is_immutable'; +end +$$; + +comment on function reject_analysis_source_snapshot_member_update() is + 'Rejects mutation of frozen snapshot membership.'; + +drop trigger if exists analysis_source_snapshot_member_update_reject + on analysis_source_snapshot_member; +create trigger analysis_source_snapshot_member_update_reject +before update or delete on analysis_source_snapshot_member +for each row execute function reject_analysis_source_snapshot_member_update(); diff --git a/migrations/rollback/0021_analysis_run_reconstruction.sql b/migrations/rollback/0021_analysis_run_reconstruction.sql new file mode 100644 index 000000000..3042eecff --- /dev/null +++ b/migrations/rollback/0021_analysis_run_reconstruction.sql @@ -0,0 +1,37 @@ +-- Fail-closed rollback for migration 0021. +-- +-- Reconstruction evidence must be exported or explicitly deleted under an +-- approved retention procedure before these objects can be removed. + +begin; + +do $$ +declare + relation_name text; + relation_has_rows boolean; +begin + foreach relation_name in array array[ + 'analysis_run_lineage_edge', + 'analysis_run_reconstruction' + ] loop + if to_regclass('public.' || relation_name) is not null then + execute format('select exists (select 1 from %I)', relation_name) + into relation_has_rows; + if relation_has_rows then + raise exception 'analysis_run_reconstruction_not_empty'; + end if; + end if; + end loop; +end +$$; + +drop trigger if exists analysis_run_lineage_edge_update_reject + on analysis_run_lineage_edge; +drop trigger if exists analysis_run_reconstruction_update_reject + on analysis_run_reconstruction; +drop function if exists reject_analysis_run_lineage_edge_update(); +drop function if exists reject_analysis_run_reconstruction_update(); +drop table if exists analysis_run_lineage_edge; +drop table if exists analysis_run_reconstruction; + +commit; diff --git a/migrations/rollback/0022_analysis_source_snapshot_member.sql b/migrations/rollback/0022_analysis_source_snapshot_member.sql new file mode 100644 index 000000000..c51a55d54 --- /dev/null +++ b/migrations/rollback/0022_analysis_source_snapshot_member.sql @@ -0,0 +1,27 @@ +-- Fail-closed rollback for migration 0022. +-- +-- Snapshot membership must be exported or explicitly deleted under an +-- approved retention procedure before these objects can be removed. + +begin; + +do $$ +declare + relation_has_rows boolean; +begin + if to_regclass('public.analysis_source_snapshot_member') is not null then + execute 'select exists (select 1 from analysis_source_snapshot_member)' + into relation_has_rows; + if relation_has_rows then + raise exception 'analysis_source_snapshot_member_not_empty'; + end if; + end if; +end +$$; + +drop trigger if exists analysis_source_snapshot_member_update_reject + on analysis_source_snapshot_member; +drop function if exists reject_analysis_source_snapshot_member_update(); +drop table if exists analysis_source_snapshot_member; + +commit; diff --git a/pyproject.toml b/pyproject.toml index ecfe24877..5a4aa12bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.87.0" +version = "0.88.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 2f3c66c45..8cb1f0ea1 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -122,6 +122,8 @@ def seed( cur.execute((migrations / "0018_analysis_run_registry.sql").read_text()) cur.execute((migrations / "0019_role_catalog_identity.sql").read_text()) cur.execute((migrations / "0020_analysis_run_retention_purge.sql").read_text()) + cur.execute((migrations / "0021_analysis_run_reconstruction.sql").read_text()) + cur.execute((migrations / "0022_analysis_source_snapshot_member.sql").read_text()) cur.execute( """ insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values @@ -1283,6 +1285,35 @@ def _ensure_demo_source_counts(cur, snapshot_id) -> None: ) +def _ensure_demo_source_snapshot_members(cur, snapshot_id, corporate_entity_id) -> None: + """Freeze Demo Corp post ids on the shared snapshot when the table exists.""" + cur.execute( + "select 1 from information_schema.tables " + "where table_schema = 'public' " + "and table_name = 'analysis_source_snapshot_member'" + ) + if cur.fetchone() is None: + return + cur.execute( + "select 1 from analysis_source_snapshot_member " + "where analysis_source_snapshot_id = %s limit 1", + (snapshot_id,), + ) + if cur.fetchone() is not None: + return + cur.execute( + """ + insert into analysis_source_snapshot_member + (analysis_source_snapshot_id, source_post_id) + select %s, post_id from source_post + where corporate_entity_id = %s + and created_at <= '2026-01-12T00:00:00Z' + on conflict do nothing + """, + (snapshot_id, corporate_entity_id), + ) + + def _seed_demo_analysis_run(cur, requested_by_account_id, corporate_entity_id) -> None: """Insert one Demo-Corp lineage run so Analysis runs is not empty. @@ -1292,6 +1323,7 @@ def _seed_demo_analysis_run(cur, requested_by_account_id, corporate_entity_id) - """ 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 @@ -1387,6 +1419,7 @@ def _seed_demo_tepp_run(cur, requested_by_account_id, corporate_entity_id) -> No """ 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 diff --git a/tests/test_analysis_run_reconstruction_schema.py b/tests/test_analysis_run_reconstruction_schema.py new file mode 100644 index 000000000..8ca265f37 --- /dev/null +++ b/tests/test_analysis_run_reconstruction_schema.py @@ -0,0 +1,150 @@ +"""Static and optional PostgreSQL contracts for run-scoped reconstruction.""" + +from __future__ import annotations + +import os +import re +import uuid +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import pytest + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_RECONSTRUCTION_MIGRATION = _ROOT / "migrations" / "0021_analysis_run_reconstruction.sql" +_RECONSTRUCTION_ROLLBACK = ( + _ROOT / "migrations" / "rollback" / "0021_analysis_run_reconstruction.sql" +) +_SNAPSHOT_MEMBER_MIGRATION = ( + _ROOT / "migrations" / "0022_analysis_source_snapshot_member.sql" +) +_SNAPSHOT_MEMBER_ROLLBACK = ( + _ROOT / "migrations" / "rollback" / "0022_analysis_source_snapshot_member.sql" +) +_POSTGRES_IMAGE = _ROOT / "docker" / "postgres-init" / "Dockerfile" +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_REQUIRED_TABLES = { + "analysis_run_reconstruction", + "analysis_run_lineage_edge", +} + + +def test_reconstruction_migration_is_normalized_and_wired() -> None: + """Static contract: 3NF names, no payload JSON, Dockerfile copy, rollback.""" + migration = _RECONSTRUCTION_MIGRATION.read_text(encoding="utf-8") + rollback = _RECONSTRUCTION_ROLLBACK.read_text(encoding="utf-8") + dockerfile = _POSTGRES_IMAGE.read_text(encoding="utf-8") + created_tables = set( + re.findall(r"create table if not exists\s+([a-z0-9_]+)", migration, re.I) + ) + assert _REQUIRED_TABLES <= created_tables + assert "jsonb" not in migration.casefold() + assert "metadata_payload" not in migration + assert "theta" not in migration.casefold() + assert "0021_analysis_run_reconstruction.sql" in dockerfile + assert "0022_analysis_source_snapshot_member.sql" in dockerfile + assert "analysis_run_reconstruction_not_empty" in rollback + assert "reject_analysis_run_reconstruction_update" in migration + assert "reject_analysis_run_lineage_edge_update" in migration + member_migration = _SNAPSHOT_MEMBER_MIGRATION.read_text(encoding="utf-8") + member_rollback = _SNAPSHOT_MEMBER_ROLLBACK.read_text(encoding="utf-8") + assert "analysis_source_snapshot_member" in member_migration + assert "jsonb" not in member_migration.casefold() + assert "theta" not in member_migration.casefold() + assert "analysis_source_snapshot_member_not_empty" in member_rollback + assert "reject_analysis_source_snapshot_member_update" in member_migration + for object_name in re.findall( + r"create table if not exists\s+([a-z0-9_]+)", + member_migration, + re.I, + ): + assert len(object_name.split("_")) >= 2, object_name + + object_patterns = ( + r"create table if not exists\s+([a-z0-9_]+)", + r"create or replace function\s+([a-z0-9_]+)", + r"create trigger\s+([a-z0-9_]+)", + ) + for pattern in object_patterns: + for object_name in re.findall(pattern, migration, re.I): + assert len(object_name.split("_")) >= 2, object_name + + +def _postgres_available() -> bool: + """Return whether the configured administrator DSN is reachable.""" + try: + import psycopg2 + + psycopg2.connect(_ADMIN_DSN, connect_timeout=2).close() + return True + except Exception: + return False + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while preserving DSN query options.""" + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +@pytest.fixture +def reconstruction_db(): + """Yield a throwaway registry+reconstruction database.""" + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + import psycopg2 + + database_name = f"lineageweave_recon_{uuid.uuid4().hex[:12]}" + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + try: + with admin.cursor() as cursor: + cursor.execute(f'create database "{database_name}"') + finally: + admin.close() + conn = psycopg2.connect(_database_dsn(database_name)) + conn.autocommit = True + try: + with conn.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_RECONSTRUCTION_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_SNAPSHOT_MEMBER_MIGRATION.read_text(encoding="utf-8")) + yield conn + finally: + conn.close() + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + try: + with admin.cursor() as cursor: + cursor.execute( + "select pg_terminate_backend(pid) from pg_stat_activity " + "where datname = %s and pid <> pg_backend_pid()", + (database_name,), + ) + cursor.execute(f'drop database "{database_name}"') + finally: + admin.close() + + +def test_empty_reconstruction_rollback_is_replayable(reconstruction_db) -> None: + """An empty reconstruction schema can be rolled back and removed.""" + with reconstruction_db.cursor() as cursor: + cursor.execute( + "select table_name from information_schema.tables " + "where table_schema = 'public' and table_name = any(%s)", + (list(_REQUIRED_TABLES),), + ) + assert {row[0] for row in cursor.fetchall()} == _REQUIRED_TABLES + cursor.execute(_RECONSTRUCTION_ROLLBACK.read_text(encoding="utf-8")) + cursor.execute( + "select table_name from information_schema.tables " + "where table_schema = 'public' and table_name = any(%s)", + (list(_REQUIRED_TABLES),), + ) + assert cursor.fetchall() == [] + cursor.execute(_RECONSTRUCTION_ROLLBACK.read_text(encoding="utf-8")) diff --git a/tests/test_analysis_run_registry_schema.py b/tests/test_analysis_run_registry_schema.py index 3d185dbed..6041d3090 100644 --- a/tests/test_analysis_run_registry_schema.py +++ b/tests/test_analysis_run_registry_schema.py @@ -276,10 +276,18 @@ def test_registry_contract_is_normalized_and_has_one_temporal_authority() -> Non assert "0018_analysis_run_registry.sql" in dockerfile assert "0019_role_catalog_identity.sql" in dockerfile assert "0020_analysis_run_retention_purge.sql" in dockerfile + assert "0021_analysis_run_reconstruction.sql" in dockerfile + assert "0022_analysis_source_snapshot_member.sql" in dockerfile seed = (_ROOT / "scripts" / "seed_demo_data.py").read_text(encoding="utf-8") assert seed.index("0019_role_catalog_identity.sql") < seed.index( "0020_analysis_run_retention_purge.sql" ) + assert seed.index("0020_analysis_run_retention_purge.sql") < seed.index( + "0021_analysis_run_reconstruction.sql" + ) + assert seed.index("0021_analysis_run_reconstruction.sql") < seed.index( + "0022_analysis_source_snapshot_member.sql" + ) assert "analysis_run_registry_not_empty" in rollback retention = _RETENTION_MIGRATION.read_text(encoding="utf-8") retention_rollback = _RETENTION_ROLLBACK.read_text(encoding="utf-8") diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py new file mode 100644 index 000000000..e7752b350 --- /dev/null +++ b/tests/test_analysis_run_start.py @@ -0,0 +1,119 @@ +"""Start-reconstruction contracts: digest, freeze, 422/409, designed tree.""" + +from backend.app.analysis_run_ingestion import reconstructed_edge_is_visible +from backend.app.analysis_run_start import ( + AnalysisRunStartError, + reconstruction_member_ids, + reconstruction_result_digest, + start_kind_rejection, + start_write_conflict_error, +) +from backend.app.lineage_ingestion import records_from_source_posts +from lineageweave.fixtures import sample_records +from lineageweave.lineage_persistence import lineage_edge_specs + + +def test_reconstruction_digest_is_stable_and_ignores_edge_order() -> None: + """The same parent choices hash the same way regardless of insert order.""" + edges = lineage_edge_specs(sample_records()) + reversed_edges = list(reversed(edges)) + assert reconstruction_result_digest(edges) == reconstruction_result_digest(reversed_edges) + assert reconstruction_result_digest([]) == reconstruction_result_digest([]) + assert reconstruction_result_digest(edges) != reconstruction_result_digest([]) + + +def test_start_uses_the_same_parent_choices_as_library_reconstruct() -> None: + """The product start path must recover the designed A-100 fork. + + fixtures.sample_records() is the synthetic gold tree: rec-002 is the + branch point for the revised quote and the delivery question. A start + that dropped an edge or invented a parent would fail this check. + """ + edges = lineage_edge_specs(sample_records()) + children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"} + assert children >= {"rec-003", "rec-004"} + assert all(0.0 <= edge.fused_score <= 1.0 for edge in edges) + assert "theta" not in reconstruction_result_digest(edges) + + +def test_start_wiring_recovers_a100_from_source_post_rows() -> None: + """CI must exercise records_from_source_posts, not only library reconstruct.""" + rows = [ + { + "post_id": record.record_id, + "post_title": record.label, + "created_at": record.occurred_at, + "thread_group_key": record.group_key, + "secondary_grouping_key": record.secondary_key, + "process_unit_id": None, + "corporate_entity_id": "corp-demo", + } + for record in sample_records() + ] + edges = lineage_edge_specs(records_from_source_posts(rows)) + children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"} + assert children >= {"rec-003", "rec-004"} + assert reconstruction_result_digest(edges) == reconstruction_result_digest( + lineage_edge_specs(sample_records()) + ) + + +def test_snapshot_members_exclude_a_later_backfill() -> None: + """Start reconstructs the create-time bag, not a later cutoff re-query.""" + captured = ["rec-001", "rec-002", "rec-003", "rec-004"] + cutoff_with_backfill = [*captured, "rec-backfill"] + assert reconstruction_member_ids(captured, cutoff_with_backfill) == captured + assert reconstruction_member_ids([], cutoff_with_backfill) == cutoff_with_backfill + + +def test_reconstructed_edge_hides_unaffiliated_private_titles() -> None: + """Edge titles use the same public-or-affiliated rule as cutoff posts.""" + affiliated = ["corp-demo"] + assert reconstructed_edge_is_visible( + parent_visibility_code="public", + parent_corporate_entity_id="corp-other", + child_visibility_code="public", + child_corporate_entity_id="corp-other", + affiliated_entity_ids=affiliated, + ) + assert not reconstructed_edge_is_visible( + parent_visibility_code="private", + parent_corporate_entity_id="corp-other", + child_visibility_code="public", + child_corporate_entity_id="corp-demo", + affiliated_entity_ids=affiliated, + ) + + +def test_tepp_and_period_report_start_are_unprocessable() -> None: + """TEPP and period-report start stay 422 so this path cannot invent a score.""" + tepp = start_kind_rejection("analysis_run_tepp") + assert tepp is not None + assert tepp.status_code == 422 + assert "invent a measurement" in tepp.detail + report = start_kind_rejection("analysis_run_report") + assert report is not None + assert report.status_code == 422 + assert "invent a measurement" in report.detail + assert "period report" in report.detail + assert start_kind_rejection("analysis_run_lineage") is None + + +def test_hidden_run_start_is_not_found() -> None: + """Operators get a 404 next action, not an internal exception name.""" + error = AnalysisRunStartError(404, "This analysis run is not visible.") + assert error.status_code == 404 + assert "not visible" in error.detail + + +def test_running_restart_conflicts_and_succeeded_replay_is_documented() -> None: + """Running is 409. Succeeded replay is a documented no-op (200 in the API).""" + conflict = start_write_conflict_error() + assert conflict.status_code == 409 + assert "Refresh to see the stored tree" in conflict.detail + running = AnalysisRunStartError( + 409, + "Open this run. Start is only for a Pending lineage reconstruction.", + ) + assert running.status_code == 409 + assert "Pending" in running.detail