fix: lock start before Running and freeze snapshot members (v0.89.0)

ARCHITECTURE.md

@@ -475,7 +475,11 @@ revision and configuration digest prefixes.
 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.
+list, then open the Pending row and **Start reconstruction**
+(ADR 0020). That start locks the run, reconstructs the create-time
+snapshot members, and persists run-scoped edges; it does not replace
+live Event Lineage and does not invent a TEPP score. Hover **Result**
+to read the reconstruction digest.
 `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.

CHANGELOG.d/0.89.0-analysis-run-start.md

@@ -0,0 +1,5 @@
+# 0.89.0 Analysis-run start reconstruction
+
+Pending lineage rows can start ThreadWeave on the create-time snapshot
+members. The designed A-100 fork appears as titled edges. Hover Result
+to read the digest. TEPP start stays 422.

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.89.0] - 2026-08-17
+
+### Added
+
+- A Pending lineage run now has **Start reconstruction**. After
+  `make seed`, request a lineage reconstruction, open the Pending Demo
+  Corp row, and start it: the designed A-100 fork appears as titled
+  parent→child edges (revised quote and delivery question under the
+  pricing follow-up). The run is locked before Running; a raced start
+  is 409. Create freezes `analysis_source_snapshot_member`; start
+  reconstructs that bag. Hover **Result** to read the reconstruction
+  digest. TEPP start is 422 — this path does not invent a theta.
+  Edges stay on the run; live Event Lineage is unchanged (ADR 0020).
+
 ## [0.88.0] - 2026-08-17
 
 ### Added

CLAUDE.md

@@ -21,3 +21,7 @@ after cutoff were rewritten after the run; compare those bodies
 before treating them as reconstructed evidence (ADR 0016).
 `POST /api/analysis-runs` records Pending on an authorized
 cutoff capture (ADR 0017) and does not reconstruct lineage.
+`POST /api/analysis-runs/{id}/start` reconstructs a Pending
+lineage run from the create-time snapshot members (ADR 0020).
+TEPP start is 422. Hover the Result prefix to read the
+reconstruction digest.

backend/app/analysis_run_ingestion.py

@@ -7,8 +7,9 @@
 payloads never do.
 
 ``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, run,
-scope, and the first Pending event atomically. It does not reconstruct
-lineage or invent a TEPP score.
+scope, and the first Pending event atomically. ``start_pending_analysis_run``
+(ADR 0020) later reconstructs lineage on that cutoff bag. Neither path
+invents a TEPP score.
 """
 
 from __future__ import annotations
@@ -263,9 +264,226 @@ 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,
+        row["knowledge_cutoff"],
+    )
+    if digest is not None:
+        detail["reconstruction_result_sha256"] = digest
+    detail["reconstructed_edges"] = edges
     return detail
 
 
+def _affiliated_post_visible(
+    visibility_code: str,
+    corporate_entity_id: Any,
+    affiliated_entity_ids: set[str],
+) -> bool:
+    """Return whether this post is public or in the caller's walk."""
+    return visibility_code == "public" or str(corporate_entity_id) in affiliated_entity_ids
+
+
+async def fetch_reconstructed_edges(
+    conn: asyncpg.Connection,
+    analysis_run_id: str,
+    affiliated_entity_ids: list[str],
+    knowledge_cutoff: Any,
+) -> tuple[str | None, list[dict[str, Any]]]:
+    """Return the persisted digest and ABAC-visible titled edges.
+
+    Missing reconstruction tables mean this database has not applied
+    migration 0020 yet; treat that as no stored tree rather than 500.
+    Hidden parent or child titles are omitted; the digest stays on the
+    run. Titles rewritten after cutoff are marked ``live_after_cutoff``.
+    """
+    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,
+          parent_post.updated_at as parent_updated_at,
+          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,
+          child_post.updated_at as child_updated_at,
+          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,
+    )
+    affiliated = {str(entity_id) for entity_id in affiliated_entity_ids}
+    edges: list[dict[str, Any]] = []
+    for row in rows:
+        parent_visible = _affiliated_post_visible(
+            row["parent_visibility_code"],
+            row["parent_corporate_entity_id"],
+            affiliated,
+        )
+        child_visible = _affiliated_post_visible(
+            row["child_visibility_code"],
+            row["child_corporate_entity_id"],
+            affiliated,
+        )
+        if not parent_visible or not child_visible:
+            continue
+        edges.append(
+            {
+                "parent_post_id": str(row["parent_post_id"]),
+                "parent_post_title": row["parent_post_title"],
+                "parent_live_after_cutoff": live_write_after_cutoff(
+                    row["parent_updated_at"], knowledge_cutoff
+                ),
+                "child_post_id": str(row["child_post_id"]),
+                "child_post_title": row["child_post_title"],
+                "child_live_after_cutoff": live_write_after_cutoff(
+                    row["child_updated_at"], knowledge_cutoff
+                ),
+                "fused_score": float(row["fused_score"]),
+            }
+        )
+    return header["result_sha256"], edges
+
+
+async def persist_snapshot_members(
+    conn: asyncpg.Connection,
+    analysis_source_snapshot_id: Any,
+    post_ids: list[str],
+) -> None:
+    """Freeze authorized post ids onto the snapshot. Skip if 0020 is absent."""
+    try:
+        for post_id in post_ids:
+            await conn.execute(
+                """
+                insert into analysis_source_snapshot_member
+                    (analysis_source_snapshot_id, source_post_id)
+                values ($1, $2)
+                on conflict do nothing
+                """,
+                analysis_source_snapshot_id,
+                post_id,
+            )
+    except asyncpg.UndefinedTableError:
+        return
+
+
+async def fetch_snapshot_member_posts(
+    conn: asyncpg.Connection,
+    analysis_source_snapshot_id: Any,
+) -> list[asyncpg.Record] | None:
+    """Return frozen snapshot members, or ``None`` if migration 0020 is absent.
+
+    An empty list means the create-time bag was empty. Start must not
+    re-walk live ``source_post`` in that case.
+    """
+    try:
+        return 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
+            """,
+            analysis_source_snapshot_id,
+        )
+    except asyncpg.UndefinedTableError:
+        return None
+
+
+async def fetch_cutoff_reconstruct_posts(
+    conn: asyncpg.Connection,
+    scope_kind_code: str,
+    corporate_entity_id: Any,
+    process_unit_id: Any,
+    scope_key: str | None,
+    affiliated_entity_ids: list[str],
+    knowledge_cutoff: Any,
+) -> list[asyncpg.Record]:
+    """ABAC-visible cutoff rows with the grouping keys reconstruct needs.
+
+    Same scope branches as ``fetch_visible_scope_posts``. The list
+    payload stays titles-only; this bag is the start path only.
+    """
+    columns = (
+        "post_id, post_title, created_at, visibility_code, "
+        "corporate_entity_id, process_unit_id, "
+        "thread_group_key, secondary_grouping_key"
+    )
+    if scope_kind_code == "analysis_scope_corporate_entity" and corporate_entity_id:
+        rows = await conn.fetch(
+            f"select {columns} "
+            "from source_post where corporate_entity_id = $1 "
+            "and created_at <= $2 "
+            "order by created_at, post_title",
+            corporate_entity_id,
+            knowledge_cutoff,
+        )
+    elif scope_kind_code == "analysis_scope_process_unit" and process_unit_id:
+        rows = await conn.fetch(
+            f"select {columns} "
+            "from source_post where process_unit_id = $1 "
+            "and created_at <= $2 "
+            "order by created_at, post_title",
+            process_unit_id,
+            knowledge_cutoff,
+        )
+    elif scope_kind_code == "analysis_scope_thread_group" and scope_key:
+        rows = await conn.fetch(
+            f"select {columns} "
+            "from source_post where thread_group_key = $1 "
+            "and created_at <= $2 "
+            "order by created_at, post_title",
+            scope_key,
+            knowledge_cutoff,
+        )
+    elif scope_kind_code == "analysis_scope_all_visible":
+        rows = await conn.fetch(
+            f"select {columns} "
+            "from source_post where created_at <= $1 "
+            "order by created_at, post_title",
+            knowledge_cutoff,
+        )
+    else:
+        return []
+    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 fetch_visible_scope_posts(
     conn: asyncpg.Connection,
     scope_kind_code: str,
@@ -585,6 +803,7 @@ async def create_pending_analysis_run(
             """,
             capture.snapshot_sha256,
         )
+    await persist_snapshot_members(conn, snapshot_id, post_ids)
     count_exists = await conn.fetchval(
         """
         select 1 from analysis_source_count