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