(null);
const [requesting, setRequesting] = useState(false);
+ const [starting, setStarting] = useState(false);
useEffect(() => {
fetchAnalysisRuns(accessToken)
@@ -1665,6 +1679,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 {
@@ -1742,7 +1772,31 @@ function AnalysisRunsPanel({
+ {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 9eaaba8b8..dd8b657a4 100644
--- a/frontend/src/api.ts
+++ b/frontend/src/api.ts
@@ -543,6 +543,16 @@ export interface AnalysisRunVisiblePost {
live_after_cutoff?: boolean;
}
+export interface AnalysisRunReconstructedEdge {
+ parent_post_id: string;
+ parent_post_title: string;
+ parent_live_after_cutoff?: boolean;
+ child_post_id: string;
+ child_post_title: string;
+ child_live_after_cutoff?: boolean;
+ fused_score: number;
+}
+
export interface AnalysisRun {
analysis_run_id: string;
run_kind_code: AnalysisRunKindCode;
@@ -557,6 +567,8 @@ export interface AnalysisRun {
source_counts: AnalysisRunCount[];
status_history?: AnalysisRunStatusEvent[];
visible_posts?: AnalysisRunVisiblePost[];
+ reconstructed_edges?: AnalysisRunReconstructedEdge[];
+ reconstruction_result_sha256?: string;
code_revision_sha?: string;
configuration_sha256?: string;
}
@@ -586,3 +598,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 036dca1fa..a158575e0 100644
--- a/lineageweave/__init__.py
+++ b/lineageweave/__init__.py
@@ -55,4 +55,4 @@
"sentence_excerpts",
]
-__version__ = "0.88.0"
+__version__ = "0.89.0"
diff --git a/migrations/0020_analysis_run_reconstruction.sql b/migrations/0020_analysis_run_reconstruction.sql
new file mode 100644
index 000000000..b4cf16cab
--- /dev/null
+++ b/migrations/0020_analysis_run_reconstruction.sql
@@ -0,0 +1,110 @@
+-- Run-scoped lineage reconstruction result (ADR 0020).
+--
+-- Create freezes authorized post ids on the snapshot. Start reconstructs
+-- that bag, not a live source_post walk. 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_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),
+ recorded_at timestamptz not null default clock_timestamp(),
+ primary key (analysis_source_snapshot_id, source_post_id)
+);
+
+comment on table analysis_source_snapshot_member is
+ 'One authorized source post frozen into a snapshot; never a post body.';
+
+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 a frozen snapshot member.';
+
+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();
+
+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/0020_analysis_run_reconstruction.sql b/migrations/rollback/0020_analysis_run_reconstruction.sql
new file mode 100644
index 000000000..0e122755b
--- /dev/null
+++ b/migrations/rollback/0020_analysis_run_reconstruction.sql
@@ -0,0 +1,42 @@
+-- Fail-closed rollback for migration 0020.
+--
+-- 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',
+ 'analysis_source_snapshot_member'
+ ] 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 trigger if exists analysis_source_snapshot_member_update_reject
+ on analysis_source_snapshot_member;
+drop function if exists reject_analysis_run_lineage_edge_update();
+drop function if exists reject_analysis_run_reconstruction_update();
+drop function if exists reject_analysis_source_snapshot_member_update();
+drop table if exists analysis_run_lineage_edge;
+drop table if exists analysis_run_reconstruction;
+drop table if exists analysis_source_snapshot_member;
+
+commit;
diff --git a/pyproject.toml b/pyproject.toml
index 5a4aa12bc..fecebee1b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "lineageweave"
-version = "0.88.0"
+version = "0.89.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 5f69bbcde..3c2fa9bb4 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_role_catalog_identity.sql").read_text())
+ cur.execute((migrations / "0020_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..3c840e5d1
--- /dev/null
+++ b/tests/test_analysis_run_reconstruction_schema.py
@@ -0,0 +1,129 @@
+"""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" / "0020_analysis_run_reconstruction.sql"
+_RECONSTRUCTION_ROLLBACK = _ROOT / "migrations" / "rollback" / "0020_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",
+ "analysis_source_snapshot_member",
+}
+
+
+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 "0020_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
+ assert "reject_analysis_source_snapshot_member_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..7b6d77d54
--- /dev/null
+++ b/tests/test_analysis_run_start.py
@@ -0,0 +1,97 @@
+"""Start-reconstruction contracts: digest stability and designed-tree fidelity."""
+
+from pathlib import Path
+
+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,
+)
+from backend.app.lineage_ingestion import records_from_source_posts
+
+_START_SOURCE = Path(__file__).resolve().parents[1] / "backend" / "app" / "analysis_run_start.py"
+_INGESTION_SOURCE = (
+ Path(__file__).resolve().parents[1] / "backend" / "app" / "analysis_run_ingestion.py"
+)
+
+
+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_maps_source_post_rows_through_records_from_source_posts() -> None:
+ """CI-runnable A-100: start uses the same mapper as the product path.
+
+ The HTTP start test needs Keycloak. This check keeps the designed
+ fork on every pytest run: source_post-shaped rows through
+ ``records_from_source_posts`` must still place the revised quote and
+ the delivery question under rec-002.
+ """
+ rows = []
+ for rec in sample_records():
+ rows.append(
+ {
+ "post_id": rec.record_id,
+ "process_unit_id": "shared-pu",
+ "corporate_entity_id": "shared-corp",
+ "post_title": rec.label,
+ "thread_group_key": rec.group_key,
+ "secondary_grouping_key": rec.secondary_key,
+ "created_at": rec.occurred_at,
+ }
+ )
+ 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_start_locks_before_running_and_maps_unique_violation() -> None:
+ """Lock, then Running. A raced insert is 409, not 500."""
+ source = _START_SOURCE.read_text(encoding="utf-8")
+ lock_at = source.index("for update of run")
+ running_at = source.index(
+ "await _append_status(conn, analysis_run_id, running_ordinal, _RUNNING"
+ )
+ assert lock_at < running_at
+ assert "UniqueViolationError" in source
+ assert "fetch_snapshot_member_posts" in source
+ ingestion = _INGESTION_SOURCE.read_text(encoding="utf-8")
+ assert "persist_snapshot_members" in ingestion
+ assert "parent_visibility_code" in ingestion
+
+
+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
diff --git a/uv.lock b/uv.lock
index 03e15b410..0367bbb8d 100644
--- a/uv.lock
+++ b/uv.lock
@@ -454,7 +454,7 @@ wheels = [
[[package]]
name = "lineageweave"
-version = "0.88.0"
+version = "0.89.0"
source = { virtual = "." }
dependencies = [
{ name = "certifi" },