From ff14607590a983b389ca531c93972b519346e49e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:07:22 +0000 Subject: [PATCH 1/2] feat: start a pending lineage reconstruction (v0.87.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/analysis-runs/{id}/start runs ThreadWeave on the authorized cutoff bag and persists run-scoped edges. The home detail starts that row so a buyer can confirm the designed A-100 fork. TEPP start stays 422 — this path never invents a theta. Co-authored-by: Seongho Bae --- ARCHITECTURE.md | 9 +- CHANGELOG.d/0.87.0-analysis-run-start.md | 5 + CHANGELOG.md | 12 ++ CLAUDE.md | 5 +- backend/app/analysis_run_ingestion.py | 59 ++++- backend/app/analysis_run_start.py | 203 ++++++++++++++++++ backend/app/main.py | 30 +++ backend/tests/test_api.py | 99 +++++++++ docker/postgres-init/Dockerfile | 1 + .../0013-normalized-analysis-run-registry.md | 6 +- .../0017-authorized-analysis-run-create.md | 3 +- .../adr/0019-authorized-analysis-run-start.md | 95 ++++++++ .../ANALYSIS_RUN_REGISTRY_REFERENCES.md | 10 +- frontend/package.json | 2 +- frontend/src/App.test.tsx | 81 +++++++ frontend/src/App.tsx | 46 +++- frontend/src/api.ts | 19 ++ lineageweave/__init__.py | 2 +- .../0019_analysis_run_reconstruction.sql | 79 +++++++ .../0019_analysis_run_reconstruction.sql | 37 ++++ pyproject.toml | 2 +- scripts/seed_demo_data.py | 1 + ...test_analysis_run_reconstruction_schema.py | 127 +++++++++++ tests/test_analysis_run_start.py | 45 ++++ 24 files changed, 962 insertions(+), 16 deletions(-) create mode 100644 CHANGELOG.d/0.87.0-analysis-run-start.md create mode 100644 backend/app/analysis_run_start.py create mode 100644 docs/adr/0019-authorized-analysis-run-start.md create mode 100644 migrations/0019_analysis_run_reconstruction.sql create mode 100644 migrations/rollback/0019_analysis_run_reconstruction.sql create mode 100644 tests/test_analysis_run_reconstruction_schema.py create mode 100644 tests/test_analysis_run_start.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index aa557bd7a..ee11eb8d2 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -473,9 +473,12 @@ 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. +status in one transaction. `POST /api/analysis-runs/{id}/start` then +runs ThreadWeave on that cutoff bag and persists run-scoped edges +(ADR 0019). It does not invent a TEPP score. Request a lineage +reconstruction from the home list, open the Pending row, then start +reconstruction. 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. diff --git a/CHANGELOG.d/0.87.0-analysis-run-start.md b/CHANGELOG.d/0.87.0-analysis-run-start.md new file mode 100644 index 000000000..f20b39d08 --- /dev/null +++ b/CHANGELOG.d/0.87.0-analysis-run-start.md @@ -0,0 +1,5 @@ +# 0.87.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 434c4d63b..9a98d23ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ 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.87.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 0019). Open the Pending run, then start reconstruction. The + designed A-100 fork (revised quote and delivery question under the + pricing follow-up) is the acceptance tree. TEPP start is 422 — this + path does not invent a theta. A Succeeded retry returns the stored + digest. Live Event Lineage stays a separate rebuild. + ## [0.86.0] - 2026-08-16 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index 2f89bada7..d84602f08 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -3,7 +3,7 @@ Tool-specific pointer. Policy lives in [AGENTS.md](AGENTS.md) and the ADRs under `docs/adr/`. Do not fork those rules here. -## Analysis-run seed (v0.85.0) +## Analysis-run seed (v0.87.0) `make seed` writes a Demo Corp lineage run and a TEPP run on the same snapshot (ADR 0013). The TEPP path goes through `tepp_client`. A missing @@ -18,4 +18,5 @@ 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 cutoff bag (ADR 0019) and does not invent a theta. diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index d26eb6f6e..82f5be7b8 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -7,8 +7,9 @@ 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. +scope, and the first Pending event atomically. ``start_pending_analysis_run`` +(ADR 0019) later reconstructs lineage on that cutoff bag. Neither path +invents a TEPP score. """ from __future__ import annotations @@ -247,9 +248,63 @@ async def fetch_visible_analysis_run( affiliated_entity_ids, row["knowledge_cutoff"], ) + digest, edges = await fetch_reconstructed_edges(conn, analysis_run_id) + if digest is not None: + detail["reconstruction_result_sha256"] = digest + detail["reconstructed_edges"] = edges return detail +async def fetch_reconstructed_edges( + conn: asyncpg.Connection, + analysis_run_id: 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 0019 yet; treat that as no stored tree rather than 500. + """ + 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, + edge.child_post_id, + child_post.post_title as child_post_title, + 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 + ] + + async def fetch_visible_scope_posts( conn: asyncpg.Connection, scope_kind_code: str, diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py new file mode 100644 index 000000000..184d622b4 --- /dev/null +++ b/backend/app/analysis_run_start.py @@ -0,0 +1,203 @@ +"""Start a Pending lineage reconstruction without inventing a TEPP score. + +ADR 0019. ``POST /api/analysis-runs/{id}/start`` transitions Pending to +Running, runs ThreadWeave on the authorized cutoff bag, persists +run-scoped edges, then stamps Succeeded. TEPP stays a wire client. +""" + +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" +_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() + + +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 _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 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 is rejected so this path cannot invent a theta. A Succeeded + retry returns the stored reconstruction. Hidden runs 404. + """ + 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.") + if current["run_kind_code"] != _LINEAGE_KIND: + raise AnalysisRunStartError( + 422, + "Connect a TEPP transport from a Failed TEPP row. " + "This start path does not invent a measurement.", + ) + 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.", + ) + + now = datetime.now(timezone.utc) + await _append_status(conn, analysis_run_id, 2, _RUNNING, now) + + locked = await conn.fetchrow( + """ + select run.analysis_run_id, run.knowledge_cutoff, + 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, + ) + 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, 3, _SUCCEEDED, finished) + 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..120c8d33a 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,32 @@ 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 is 422 so this path + cannot invent a theta. A Succeeded retry returns the stored tree. + """ + _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 bff7f7c64..e447c6081 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -32,6 +32,9 @@ _REALM = "lineageweave-demo" _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" +_RECONSTRUCTION_MIGRATION = ( + Path(__file__).resolve().parents[2] / "migrations" / "0019_analysis_run_reconstruction.sql" +) def _postgres_available() -> bool: @@ -115,6 +118,7 @@ def seeded_db(demo_analyst_token): with conn.cursor() as cur: cur.execute(_MIGRATION_PATH.read_text()) cur.execute(_REGISTRY_MIGRATION.read_text()) + cur.execute(_RECONSTRUCTION_MIGRATION.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " @@ -573,6 +577,101 @@ 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"] + + 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 d95e2c917..b80b7b4d8 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -24,6 +24,7 @@ COPY migrations/0015_organization_name_resolution.sql /docker-entrypoint-initdb. COPY migrations/0016_cross_post_actor_identity.sql /docker-entrypoint-initdb.d/17-cross-post-actor-identity.sql COPY migrations/0017_prov_o_standard_relations.sql /docker-entrypoint-initdb.d/18-prov-o-standard-relations.sql COPY migrations/0018_analysis_run_registry.sql /docker-entrypoint-initdb.d/19-analysis-run-registry.sql +COPY migrations/0019_analysis_run_reconstruction.sql /docker-entrypoint-initdb.d/20-analysis-run-reconstruction.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 b60164dc6..f9e09e07f 100644 --- a/docs/adr/0013-normalized-analysis-run-registry.md +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -238,8 +238,10 @@ 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 (ADR 0019). 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..ec43f4ea4 100644 --- a/docs/adr/0017-authorized-analysis-run-create.md +++ b/docs/adr/0017-authorized-analysis-run-create.md @@ -1,6 +1,7 @@ # ADR 0017 — Operators request an analysis run through the product API -**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Decision status:** Accepted; merged to the #74 stack via #125. Not +protected-main truth until #74 lands. **Date:** 2026-08-16 **Depends on:** ADR 0013 normalized analysis-run registry; ADR 0014 authorized analysis-run read; ADR 0016 knowledge-cutoff posts diff --git a/docs/adr/0019-authorized-analysis-run-start.md b/docs/adr/0019-authorized-analysis-run-start.md new file mode 100644 index 000000000..c80e4a6ed --- /dev/null +++ b/docs/adr/0019-authorized-analysis-run-start.md @@ -0,0 +1,95 @@ +# ADR 0019 — 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. + +## 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 cannot invent a theta; +3. replays a Succeeded run; +4. accepts only Pending lineage; +5. appends Running, runs `lineage_edge_specs` / ThreadWeave on the + ABAC-visible cutoff bag, persists `analysis_run_reconstruction` plus + `analysis_run_lineage_edge`, then appends Succeeded. + +```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 other kind + API-->>Operator: 422 connect the measurement service + else already Succeeded + Registry-->>API: stored edges + API-->>Operator: 200 replay + else Pending lineage + Registry->>Registry: Running + API->>ThreadWeave: reconstruct cutoff records + 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. + +The home detail adds **Start reconstruction** on a Pending lineage row +and lists titled parent→child edges after Succeeded. + +## 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 b41b31c17..c4eaee918 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:** Migration 0018, ADR 0013, rollback, and real-PostgreSQL contract tests. +**Scope:** Migrations 0018–0019, ADR 0013 / 0017 / 0019, rollback, and +real-PostgreSQL contract tests. ## Standards mapped to implementation @@ -13,7 +14,8 @@ | ISO 8601-1:2019 | Use unambiguous timestamp representation and timezone-aware persistence. | PostgreSQL `timestamptz` for availability, capture, cutoff, request, occurrence, and record clocks; tests use explicit `Z` offsets. | | 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, and exclusion of raw source/provider payloads. | -| 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`; tests require the designed A-100 fork (revised quote + delivery question under the pricing follow-up). | ## Temporal reasoning @@ -76,9 +78,13 @@ 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 rollback with any registry rows and allow replay after explicit cleanup. | +| Start reconstruction recovers the designed tree | Persist edges from `lineage_edge_specs` on the A-100 fixture bag; the pricing follow-up must parent both the revised quote and the delivery question. A TEPP start must 422 without a theta. | ## APA 7th references +ContextualWisdomLab. (2026). *ThreadWeave* [Computer software]. +https://github.com/ContextualWisdomLab/ThreadWeave + 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). diff --git a/frontend/package.json b/frontend/package.json index dac241738..c93670d28 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.86.0", + "version": "0.87.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 934f150e4..0855a4ab7 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -269,6 +269,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", @@ -283,6 +338,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, @@ -1666,6 +1722,31 @@ 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 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 83fb88860..5987f7af5 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -4,6 +4,7 @@ import { askPostChat, BackendError, createAnalysisRun, + startAnalysisRun, createPostTicket, deriveCommitment, evaluatePost, @@ -1458,7 +1459,10 @@ function analysisRunCaption(run: AnalysisRun): string { */ function analysisRunNextAction(run: AnalysisRun): string | null { if (run.status_code === "analysis_status_pending") { - return "Open this run to confirm which posts it will use. Reconstruction has not started yet."; + if (run.run_kind_code === "analysis_run_tepp") { + return "Open this run to confirm which posts it will measure. Measurement has not started yet."; + } + return "Open this run, then start reconstruction. Reconstruction has not started yet."; } if (run.status_code !== "analysis_status_failed") { return null; @@ -1572,6 +1576,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) @@ -1597,6 +1602,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 { @@ -1675,6 +1696,29 @@ function AnalysisRunsPanel({ codeRevisionSha={selected.code_revision_sha} configurationSha256={selected.configuration_sha256} /> + {selected.status_code === "analysis_status_pending" && ( +

{analysisRunNextAction(selected)}

+ )} + {selected.run_kind_code === "analysis_run_lineage" && + selected.status_code === "analysis_status_pending" && ( + + )} + {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 f9bd4068e..208d0f0dc 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -522,6 +522,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: string; @@ -536,6 +544,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; } @@ -565,3 +575,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 5f70c6064..1950c39f8 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.86.0" +__version__ = "0.87.0" diff --git a/migrations/0019_analysis_run_reconstruction.sql b/migrations/0019_analysis_run_reconstruction.sql new file mode 100644 index 000000000..39d517760 --- /dev/null +++ b/migrations/0019_analysis_run_reconstruction.sql @@ -0,0 +1,79 @@ +-- Run-scoped lineage reconstruction result (ADR 0019). +-- +-- 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/rollback/0019_analysis_run_reconstruction.sql b/migrations/rollback/0019_analysis_run_reconstruction.sql new file mode 100644 index 000000000..ce0c73588 --- /dev/null +++ b/migrations/rollback/0019_analysis_run_reconstruction.sql @@ -0,0 +1,37 @@ +-- Fail-closed rollback for migration 0019. +-- +-- 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/pyproject.toml b/pyproject.toml index 0393b774a..ecfe24877 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.86.0" +version = "0.87.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 f6f575ccc..d4a109c11 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -120,6 +120,7 @@ def seed( cur.execute((migrations / "0015_organization_name_resolution.sql").read_text()) cur.execute((migrations / "0016_cross_post_actor_identity.sql").read_text()) cur.execute((migrations / "0018_analysis_run_registry.sql").read_text()) + cur.execute((migrations / "0019_analysis_run_reconstruction.sql").read_text()) cur.execute( """ insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values diff --git a/tests/test_analysis_run_reconstruction_schema.py b/tests/test_analysis_run_reconstruction_schema.py new file mode 100644 index 000000000..390a9c74b --- /dev/null +++ b/tests/test_analysis_run_reconstruction_schema.py @@ -0,0 +1,127 @@ +"""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" / "0019_analysis_run_reconstruction.sql" +_RECONSTRUCTION_ROLLBACK = _ROOT / "migrations" / "rollback" / "0019_analysis_run_reconstruction.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 "0019_analysis_run_reconstruction.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 + + 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")) + 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_start.py b/tests/test_analysis_run_start.py new file mode 100644 index 000000000..239c5b5bb --- /dev/null +++ b/tests/test_analysis_run_start.py @@ -0,0 +1,45 @@ +"""Start-reconstruction contracts: digest stability and designed-tree fidelity.""" + +from lineageweave.fixtures import sample_records +from lineageweave.lineage_persistence import lineage_edge_specs + +from backend.app.analysis_run_start import ( + AnalysisRunStartError, + reconstruction_result_digest, +) + + +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_error_carries_a_next_action() -> None: + """Operators get a next action, not an internal exception name.""" + error = AnalysisRunStartError( + 422, + "Connect a TEPP transport from a Failed TEPP row. " + "This start path does not invent a measurement.", + ) + assert error.status_code == 422 + assert "invent a measurement" in error.detail From 8ff0fee6a8e1b40b8b93085a0c488c7ec42b6d32 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:21:52 +0000 Subject: [PATCH 2/2] fix: freeze the cutoff bag before starting reconstruction (v0.87.1) Lock the Pending run before Running so a double-click is 409, persist snapshot membership at create, show the Result digest, and keep edge titles public-or-affiliated. CI now checks the A-100 fork through records_from_source_posts. This path still does not invent a theta. Co-authored-by: Seongho Bae --- ARCHITECTURE.md | 11 +- CHANGELOG.d/0.87.1-analysis-run-start-bag.md | 6 + CHANGELOG.md | 12 ++ CLAUDE.md | 5 +- backend/app/analysis_run_ingestion.py | 63 ++++++- backend/app/analysis_run_start.py | 170 ++++++++++++++---- backend/tests/test_api.py | 4 + docker/postgres-init/Dockerfile | 1 + .../0013-normalized-analysis-run-registry.md | 5 +- .../0017-authorized-analysis-run-create.md | 3 +- .../adr/0019-authorized-analysis-run-start.md | 14 +- .../ANALYSIS_RUN_REGISTRY_REFERENCES.md | 6 +- frontend/package.json | 2 +- frontend/src/App.test.tsx | 3 + frontend/src/App.tsx | 28 ++- lineageweave/__init__.py | 2 +- .../0020_analysis_source_snapshot_member.sql | 35 ++++ .../0020_analysis_source_snapshot_member.sql | 27 +++ pyproject.toml | 2 +- scripts/seed_demo_data.py | 32 ++++ ...test_analysis_run_reconstruction_schema.py | 17 ++ tests/test_analysis_run_start.py | 61 ++++++- 22 files changed, 444 insertions(+), 65 deletions(-) create mode 100644 CHANGELOG.d/0.87.1-analysis-run-start-bag.md create mode 100644 migrations/0020_analysis_source_snapshot_member.sql create mode 100644 migrations/rollback/0020_analysis_source_snapshot_member.sql diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ee11eb8d2..ebe0aae9b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -474,11 +474,12 @@ 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. `POST /api/analysis-runs/{id}/start` then -runs ThreadWeave on that cutoff bag and persists run-scoped edges -(ADR 0019). It does not invent a TEPP score. Request a lineage -reconstruction from the home list, open the Pending row, then start -reconstruction. Confirm the designed A-100 fork before treating the -live Event Lineage panel as that run's tree. +runs ThreadWeave on the frozen snapshot membership and persists +run-scoped edges (ADR 0019). 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. diff --git a/CHANGELOG.d/0.87.1-analysis-run-start-bag.md b/CHANGELOG.d/0.87.1-analysis-run-start-bag.md new file mode 100644 index 000000000..544989285 --- /dev/null +++ b/CHANGELOG.d/0.87.1-analysis-run-start-bag.md @@ -0,0 +1,6 @@ +# 0.87.1 start reconstructs the create-time cutoff bag + +Start locks the run before writing Running, persists snapshot membership +at create, and shows the Result digest prefix. Refresh after a double +start to see the stored tree. This path still does not invent a TEPP +measurement. diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a98d23ff..282ba8fca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ 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.87.1] - 2026-08-16 + +### Fixed + +- Start locks the Pending run before writing Running, so a double-click + is a next-action 409 instead of a 500. Create now freezes authorized + post ids on `analysis_source_snapshot_member`; start reconstructs that + bag instead of a later backfill that shares the cutoff clock. Hover + the Result digest prefix after Succeeded to verify the parent-choice + hash. Edge titles stay public-or-affiliated. Open the Pending Demo + Corp row, start reconstruction, and confirm the A-100 fork. + ## [0.87.0] - 2026-08-16 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index d84602f08..bf68e0c9d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -3,7 +3,7 @@ Tool-specific pointer. Policy lives in [AGENTS.md](AGENTS.md) and the ADRs under `docs/adr/`. Do not fork those rules here. -## Analysis-run seed (v0.87.0) +## Analysis-run seed (v0.87.1) `make seed` writes a Demo Corp lineage run and a TEPP run on the same snapshot (ADR 0013). The TEPP path goes through `tepp_client`. A missing @@ -19,4 +19,5 @@ 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). `POST /api/analysis-runs/{id}/start` -reconstructs that cutoff bag (ADR 0019) and does not invent a theta. +reconstructs that frozen cutoff bag (ADR 0019) 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 82f5be7b8..a45f12bf6 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -248,21 +248,48 @@ async def fetch_visible_analysis_run( affiliated_entity_ids, row["knowledge_cutoff"], ) - digest, edges = await fetch_reconstructed_edges(conn, analysis_run_id) + 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 0019 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( @@ -282,8 +309,12 @@ async def fetch_reconstructed_edges( 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 @@ -302,9 +333,38 @@ async def fetch_reconstructed_edges( "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, @@ -629,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 index 184d622b4..78f375f38 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -102,6 +102,70 @@ async def _append_status( ) +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 0020 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 _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 _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, *, @@ -112,7 +176,9 @@ async def start_pending_analysis_run( """Run ThreadWeave on a visible Pending lineage row. TEPP is rejected so this path cannot invent a theta. A Succeeded - retry returns the stored reconstruction. Hidden runs 404. + retry returns the stored reconstruction. 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) @@ -141,13 +207,10 @@ async def start_pending_analysis_run( "Open this run. Start is only for a Pending lineage reconstruction.", ) - now = datetime.now(timezone.utc) - await _append_status(conn, analysis_run_id, 2, _RUNNING, now) - locked = await conn.fetchrow( """ select run.analysis_run_id, run.knowledge_cutoff, - scope.corporate_entity_id + 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 @@ -155,43 +218,86 @@ async def start_pending_analysis_run( """, analysis_run_id, ) - 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( + locked_status = await conn.fetchval( """ - insert into analysis_run_reconstruction - (analysis_run_id, result_sha256, edge_count, reconstructed_at) - values ($1, $2, $3, $4) + select status_code + from analysis_run_current_status + where analysis_run_id = $1 """, analysis_run_id, - digest, - len(edges), - finished, ) - for edge in edges: + 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_lineage_edge - (analysis_run_id, child_post_id, parent_post_id, - fused_score, reconstructed_at) - values ($1, $2, $3, $4, $5) + insert into analysis_run_reconstruction + (analysis_run_id, result_sha256, edge_count, reconstructed_at) + values ($1, $2, $3, $4) """, analysis_run_id, - edge.child_id, - edge.parent_id, - edge.fused_score, + 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, ) - await _append_status(conn, analysis_run_id, 3, _SUCCEEDED, finished) + except asyncpg.UniqueViolationError as exc: + raise start_write_conflict_error() from exc started = await fetch_visible_analysis_run( conn, analysis_run_id, diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index e447c6081..3352e89db 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -35,6 +35,9 @@ _RECONSTRUCTION_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" / "0019_analysis_run_reconstruction.sql" ) +_SNAPSHOT_MEMBER_MIGRATION = ( + Path(__file__).resolve().parents[2] / "migrations" / "0020_analysis_source_snapshot_member.sql" +) def _postgres_available() -> bool: @@ -119,6 +122,7 @@ def seeded_db(demo_analyst_token): cur.execute(_MIGRATION_PATH.read_text()) cur.execute(_REGISTRY_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'), " diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index b80b7b4d8..e56fee0aa 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -25,6 +25,7 @@ COPY migrations/0016_cross_post_actor_identity.sql /docker-entrypoint-initdb.d/1 COPY migrations/0017_prov_o_standard_relations.sql /docker-entrypoint-initdb.d/18-prov-o-standard-relations.sql COPY migrations/0018_analysis_run_registry.sql /docker-entrypoint-initdb.d/19-analysis-run-registry.sql COPY migrations/0019_analysis_run_reconstruction.sql /docker-entrypoint-initdb.d/20-analysis-run-reconstruction.sql +COPY migrations/0020_analysis_source_snapshot_member.sql /docker-entrypoint-initdb.d/21-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 f9e09e07f..6f1a30b46 100644 --- a/docs/adr/0013-normalized-analysis-run-registry.md +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -240,8 +240,9 @@ Acceptance requires: first status atomically and compares request digests on idempotent retries. `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 (ADR 0019). A durable outbox / Valkey - worker and live TEPP execution remain later slices. + lineage cutoff bag in-process from frozen snapshot membership + (ADR 0019). 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 ec43f4ea4..30c9883d5 100644 --- a/docs/adr/0017-authorized-analysis-run-create.md +++ b/docs/adr/0017-authorized-analysis-run-create.md @@ -23,7 +23,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 diff --git a/docs/adr/0019-authorized-analysis-run-start.md b/docs/adr/0019-authorized-analysis-run-start.md index c80e4a6ed..b3d8dbcdc 100644 --- a/docs/adr/0019-authorized-analysis-run-start.md +++ b/docs/adr/0019-authorized-analysis-run-start.md @@ -29,9 +29,13 @@ transaction: 2. rejects non-lineage kinds so TEPP cannot invent a theta; 3. replays a Succeeded run; 4. accepts only Pending lineage; -5. appends Running, runs `lineage_edge_specs` / ThreadWeave on the - ABAC-visible cutoff bag, persists `analysis_run_reconstruction` plus - `analysis_run_lineage_edge`, then appends Succeeded. +5. locks the run row, re-reads status, appends Running, runs + `lineage_edge_specs` / ThreadWeave 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 @@ -65,7 +69,9 @@ Rules: - Failed TEPP remains a `tepp_client` transport problem. The home detail adds **Start reconstruction** on a Pending lineage row -and lists titled parent→child edges after Succeeded. +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. ## Consequences diff --git a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md index c4eaee918..2c7730d38 100644 --- a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md +++ b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md @@ -1,7 +1,7 @@ # Analysis-run registry standards and research traceability **Status:** Active PR evidence; not protected-main truth until merge. -**Scope:** Migrations 0018–0019, ADR 0013 / 0017 / 0019, rollback, and +**Scope:** Migrations 0018–0020, ADR 0013 / 0017 / 0019, rollback, and real-PostgreSQL contract tests. ## Standards mapped to implementation @@ -15,7 +15,7 @@ real-PostgreSQL contract tests. | 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, and exclusion of raw source/provider payloads. | | 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`; tests require the designed A-100 fork (revised quote + delivery question under the pricing follow-up). | +| 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 @@ -78,7 +78,7 @@ 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 rollback with any registry rows and allow replay after explicit cleanup. | -| Start reconstruction recovers the designed tree | Persist edges from `lineage_edge_specs` on the A-100 fixture bag; the pricing follow-up must parent both the revised quote and the delivery question. A TEPP start must 422 without a theta. | +| 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 start must 422 without a theta. Snapshot members exclude a later backfill. A concurrent start is 409 with a refresh next action. | ## APA 7th references diff --git a/frontend/package.json b/frontend/package.json index c93670d28..8153ec16f 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.87.0", + "version": "0.87.1", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 0855a4ab7..d65dc1a51 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -1741,6 +1741,9 @@ describe("App, authenticated", () => { 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"), ); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 5987f7af5..80f476fc0 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1538,11 +1538,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 ( @@ -1551,15 +1563,12 @@ 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} + ))}

    ); @@ -1695,6 +1704,7 @@ function AnalysisRunsPanel({ {selected.status_code === "analysis_status_pending" && (

    {analysisRunNextAction(selected)}

    diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 1950c39f8..6d22286fa 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.87.0" +__version__ = "0.87.1" diff --git a/migrations/0020_analysis_source_snapshot_member.sql b/migrations/0020_analysis_source_snapshot_member.sql new file mode 100644 index 000000000..955d111d8 --- /dev/null +++ b/migrations/0020_analysis_source_snapshot_member.sql @@ -0,0 +1,35 @@ +-- Create-time cutoff membership for an analysis source snapshot (ADR 0019). +-- +-- 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/0020_analysis_source_snapshot_member.sql b/migrations/rollback/0020_analysis_source_snapshot_member.sql new file mode 100644 index 000000000..c0bdd685a --- /dev/null +++ b/migrations/rollback/0020_analysis_source_snapshot_member.sql @@ -0,0 +1,27 @@ +-- Fail-closed rollback for migration 0020. +-- +-- 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..9dea9d3f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.87.0" +version = "0.87.1" 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 d4a109c11..0131cc33d 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -121,6 +121,7 @@ def seed( cur.execute((migrations / "0016_cross_post_actor_identity.sql").read_text()) cur.execute((migrations / "0018_analysis_run_registry.sql").read_text()) cur.execute((migrations / "0019_analysis_run_reconstruction.sql").read_text()) + cur.execute((migrations / "0020_analysis_source_snapshot_member.sql").read_text()) cur.execute( """ insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values @@ -1282,6 +1283,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. @@ -1291,6 +1321,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 @@ -1386,6 +1417,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 index 390a9c74b..0ad355028 100644 --- a/tests/test_analysis_run_reconstruction_schema.py +++ b/tests/test_analysis_run_reconstruction_schema.py @@ -15,6 +15,8 @@ _REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" _RECONSTRUCTION_MIGRATION = _ROOT / "migrations" / "0019_analysis_run_reconstruction.sql" _RECONSTRUCTION_ROLLBACK = _ROOT / "migrations" / "rollback" / "0019_analysis_run_reconstruction.sql" +_SNAPSHOT_MEMBER_MIGRATION = _ROOT / "migrations" / "0020_analysis_source_snapshot_member.sql" +_SNAPSHOT_MEMBER_ROLLBACK = _ROOT / "migrations" / "rollback" / "0020_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" @@ -38,9 +40,23 @@ def test_reconstruction_migration_is_normalized_and_wired() -> None: assert "metadata_payload" not in migration assert "theta" not in migration.casefold() assert "0019_analysis_run_reconstruction.sql" in dockerfile + assert "0020_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_]+)", @@ -91,6 +107,7 @@ def reconstruction_db(): 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() diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index 239c5b5bb..f2b6e53ab 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -1,12 +1,15 @@ """Start-reconstruction contracts: digest stability and designed-tree fidelity.""" -from lineageweave.fixtures import sample_records -from lineageweave.lineage_persistence import lineage_edge_specs - +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_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: @@ -34,6 +37,55 @@ def test_start_uses_the_same_parent_choices_as_library_reconstruct() -> None: 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_start_error_carries_a_next_action() -> None: """Operators get a next action, not an internal exception name.""" error = AnalysisRunStartError( @@ -43,3 +95,6 @@ def test_start_error_carries_a_next_action() -> None: ) assert error.status_code == 422 assert "invent a measurement" in error.detail + conflict = start_write_conflict_error() + assert conflict.status_code == 409 + assert "Refresh to see the stored tree" in conflict.detail