feat: start a pending lineage reconstruction (v0.87.0)

ARCHITECTURE.md

@@ -473,9 +473,12 @@ without seeing later live rows or hidden bodies. Detail also returns
 revision and configuration digest prefixes.
 `POST /api/analysis-runs` records a Pending run on a new authorized
 cutoff capture (ADR 0017): snapshot, counts, run, scope, and the first
-status in one transaction. It does not reconstruct lineage and does not
-invent a TEPP score. Request a lineage reconstruction from the home
-list, then open the Pending row to confirm the cutoff corpus.
+status in one transaction. `POST /api/analysis-runs/{id}/start` then
+runs ThreadWeave on that cutoff bag and persists run-scoped edges
+(ADR 0019). It does not invent a TEPP score. Request a lineage
+reconstruction from the home list, open the Pending row, then start
+reconstruction. Confirm the designed A-100 fork before treating the
+live Event Lineage panel as that run's tree.
 `make seed` also records a TEPP measurement run through
 `tepp_client` on that same snapshot; the default transport is
 unavailable, so that run is Failed rather than a fabricated score.

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

@@ -0,0 +1,5 @@
+# 0.87.0 start a pending lineage reconstruction
+
+`POST /api/analysis-runs/{id}/start` runs ThreadWeave on a Pending
+lineage cutoff bag and persists run-scoped edges. Start reconstruction
+from the open run. This path does not invent a TEPP measurement.

CHANGELOG.md

@@ -4,6 +4,18 @@ All notable changes to this project are documented here. Format follows
 [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.87.0] - 2026-08-16
+
+### Added
+
+- `POST /api/analysis-runs/{id}/start` runs ThreadWeave on a visible
+  Pending lineage cutoff bag and persists run-scoped parent choices
+  (ADR 0019). Open the Pending run, then start reconstruction. The
+  designed A-100 fork (revised quote and delivery question under the
+  pricing follow-up) is the acceptance tree. TEPP start is 422 — this
+  path does not invent a theta. A Succeeded retry returns the stored
+  digest. Live Event Lineage stays a separate rebuild.
+
 ## [0.86.0] - 2026-08-16
 
 ### Added

CLAUDE.md

@@ -3,7 +3,7 @@
 Tool-specific pointer. Policy lives in [AGENTS.md](AGENTS.md) and the
 ADRs under `docs/adr/`. Do not fork those rules here.
 
-## Analysis-run seed (v0.85.0)
+## Analysis-run seed (v0.87.0)
 
 `make seed` writes a Demo Corp lineage run and a TEPP run on the same
 snapshot (ADR 0013). The TEPP path goes through `tepp_client`. A missing
@@ -18,4 +18,5 @@ Digest prefixes stay audible; hover a prefix to read the full digest.
 Opening a cutoff title shows the live post -- compare it with the
 cutoff before treating the body as reconstructed evidence (ADR 0016).
 `POST /api/analysis-runs` records Pending on an authorized
-cutoff capture (ADR 0017) and does not reconstruct lineage.
+cutoff capture (ADR 0017). `POST /api/analysis-runs/{id}/start`
+reconstructs that cutoff bag (ADR 0019) and does not invent a theta.

backend/app/analysis_run_ingestion.py

@@ -7,8 +7,9 @@
 payloads never do.
 
 ``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, run,
-scope, and the first Pending event atomically. It does not reconstruct
-lineage or invent a TEPP score.
+scope, and the first Pending event atomically. ``start_pending_analysis_run``
+(ADR 0019) later reconstructs lineage on that cutoff bag. Neither path
+invents a TEPP score.
 """
 
 from __future__ import annotations
@@ -247,9 +248,63 @@ async def fetch_visible_analysis_run(
         affiliated_entity_ids,
         row["knowledge_cutoff"],
     )
+    digest, edges = await fetch_reconstructed_edges(conn, analysis_run_id)
+    if digest is not None:
+        detail["reconstruction_result_sha256"] = digest
+    detail["reconstructed_edges"] = edges
     return detail
 
 
+async def fetch_reconstructed_edges(
+    conn: asyncpg.Connection,
+    analysis_run_id: str,
+) -> tuple[str | None, list[dict[str, Any]]]:
+    """Return the persisted digest and titled edges, or ``(None, [])``.
+
+    Missing reconstruction tables mean this database has not applied
+    migration 0019 yet; treat that as no stored tree rather than 500.
+    """
+    try:
+        header = await conn.fetchrow(
+            """
+            select result_sha256
+            from analysis_run_reconstruction
+            where analysis_run_id = $1
+            """,
+            analysis_run_id,
+        )
+    except asyncpg.UndefinedTableError:
+        return None, []
+    if header is None:
+        return None, []
+    rows = await conn.fetch(
+        """
+        select
+          edge.parent_post_id,
+          parent_post.post_title as parent_post_title,
+          edge.child_post_id,
+          child_post.post_title as child_post_title,
+          edge.fused_score
+        from analysis_run_lineage_edge edge
+        join source_post parent_post on parent_post.post_id = edge.parent_post_id
+        join source_post child_post on child_post.post_id = edge.child_post_id
+        where edge.analysis_run_id = $1
+        order by parent_post.post_title, child_post.post_title
+        """,
+        analysis_run_id,
+    )
+    return header["result_sha256"], [
+        {
+            "parent_post_id": str(row["parent_post_id"]),
+            "parent_post_title": row["parent_post_title"],
+            "child_post_id": str(row["child_post_id"]),
+            "child_post_title": row["child_post_title"],
+            "fused_score": float(row["fused_score"]),
+        }
+        for row in rows
+    ]
+
+
 async def fetch_visible_scope_posts(
     conn: asyncpg.Connection,
     scope_kind_code: str,

backend/app/analysis_run_start.py

@@ -0,0 +1,203 @@
+"""Start a Pending lineage reconstruction without inventing a TEPP score.
+
+ADR 0019. ``POST /api/analysis-runs/{id}/start`` transitions Pending to
+Running, runs ThreadWeave on the authorized cutoff bag, persists
+run-scoped edges, then stamps Succeeded. TEPP stays a wire client.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID
+
+import asyncpg
+
+from backend.app.analysis_run_ingestion import (
+    AnalysisRunCreateError,
+    fetch_visible_analysis_run,
+)
+from backend.app.lineage_ingestion import records_from_source_posts
+from lineageweave.lineage_persistence import lineage_edge_specs
+from lineageweave.models import Edge
+
+_LINEAGE_KIND = "analysis_run_lineage"
+_PENDING = "analysis_status_pending"
+_RUNNING = "analysis_status_running"
+_SUCCEEDED = "analysis_status_succeeded"
+
+
+class AnalysisRunStartError(AnalysisRunCreateError):
+    """Fail-closed start: HTTP status plus a next-action detail string."""
+
+
+def reconstruction_result_digest(edges: list[Edge]) -> str:
+    """SHA-256 of the ordered parent choices. Never hashes a post body."""
+    material = json.dumps(
+        [
+            {
+                "child_post_id": edge.child_id,
+                "fused_score": round(float(edge.fused_score), 6),
+                "parent_post_id": edge.parent_id,
+            }
+            for edge in sorted(edges, key=lambda item: (item.child_id, item.parent_id))
+        ],
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+    return hashlib.sha256(material.encode()).hexdigest()
+
+
+async def _cutoff_source_posts(
+    conn: asyncpg.Connection,
+    *,
+    corporate_entity_id: Any,
+    knowledge_cutoff: Any,
+    affiliated_entity_ids: list[str],
+) -> list[asyncpg.Record]:
+    """ABAC-visible cutoff rows with the grouping keys reconstruct needs."""
+    rows = await conn.fetch(
+        """
+        select post_id, post_title, created_at, visibility_code,
+               corporate_entity_id, process_unit_id,
+               thread_group_key, secondary_grouping_key
+        from source_post
+        where corporate_entity_id = $1 and created_at <= $2
+        order by created_at, post_title
+        """,
+        corporate_entity_id,
+        knowledge_cutoff,
+    )
+    affiliated = {str(entity_id) for entity_id in affiliated_entity_ids}
+    return [
+        row
+        for row in rows
+        if row["visibility_code"] == "public"
+        or str(row["corporate_entity_id"]) in affiliated
+    ]
+
+
+async def _append_status(
+    conn: asyncpg.Connection,
+    analysis_run_id: str,
+    status_ordinal: int,
+    status_code: str,
+    occurred_at: datetime,
+    failure_code: str | None = None,
+) -> None:
+    """Append one legal lifecycle event. Failed rows carry a machine code."""
+    await conn.execute(
+        """
+        insert into analysis_run_status_event
+            (analysis_run_id, status_ordinal, status_code, occurred_at, failure_code)
+        values ($1, $2, $3, $4, $5)
+        """,
+        analysis_run_id,
+        status_ordinal,
+        status_code,
+        occurred_at,
+        failure_code,
+    )
+
+
+async def start_pending_analysis_run(
+    conn: asyncpg.Connection,
+    *,
+    analysis_run_id: str,
+    account_id: str,
+    affiliated_entity_ids: list[str],
+) -> dict[str, Any]:
+    """Run ThreadWeave on a visible Pending lineage row.
+
+    TEPP is rejected so this path cannot invent a theta. A Succeeded
+    retry returns the stored reconstruction. Hidden runs 404.
+    """
+    try:
+        UUID(analysis_run_id)
+    except ValueError as exc:
+        raise AnalysisRunStartError(404, "This analysis run is not visible.") from exc
+
+    current = await fetch_visible_analysis_run(
+        conn,
+        analysis_run_id,
+        account_id,
+        affiliated_entity_ids,
+    )
+    if current is None:
+        raise AnalysisRunStartError(404, "This analysis run is not visible.")
+    if current["run_kind_code"] != _LINEAGE_KIND:
+        raise AnalysisRunStartError(
+            422,
+            "Connect a TEPP transport from a Failed TEPP row. "
+            "This start path does not invent a measurement.",
+        )
+    if current["status_code"] == _SUCCEEDED:
+        return current
+    if current["status_code"] != _PENDING:
+        raise AnalysisRunStartError(
+            409,
+            "Open this run. Start is only for a Pending lineage reconstruction.",
+        )
+
+    now = datetime.now(timezone.utc)
+    await _append_status(conn, analysis_run_id, 2, _RUNNING, now)
+
+    locked = await conn.fetchrow(
+        """
+        select run.analysis_run_id, run.knowledge_cutoff,
+               scope.corporate_entity_id
+        from analysis_run run
+        join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id
+        where run.analysis_run_id = $1
+        for update of run
+        """,
+        analysis_run_id,
+    )
+    rows = await _cutoff_source_posts(
+        conn,
+        corporate_entity_id=locked["corporate_entity_id"],
+        knowledge_cutoff=locked["knowledge_cutoff"],
+        affiliated_entity_ids=affiliated_entity_ids,
+    )
+    edges = lineage_edge_specs(records_from_source_posts(rows))
+    digest = reconstruction_result_digest(edges)
+    finished = datetime.now(timezone.utc)
+    if finished < now:
+        finished = now
+    await conn.execute(
+        """
+        insert into analysis_run_reconstruction
+            (analysis_run_id, result_sha256, edge_count, reconstructed_at)
+        values ($1, $2, $3, $4)
+        """,
+        analysis_run_id,
+        digest,
+        len(edges),
+        finished,
+    )
+    for edge in edges:
+        await conn.execute(
+            """
+            insert into analysis_run_lineage_edge
+                (analysis_run_id, child_post_id, parent_post_id,
+                 fused_score, reconstructed_at)
+            values ($1, $2, $3, $4, $5)
+            """,
+            analysis_run_id,
+            edge.child_id,
+            edge.parent_id,
+            edge.fused_score,
+            finished,
+        )
+    await _append_status(conn, analysis_run_id, 3, _SUCCEEDED, finished)
+    started = await fetch_visible_analysis_run(
+        conn,
+        analysis_run_id,
+        account_id,
+        affiliated_entity_ids,
+    )
+    if started is None:
+        raise AnalysisRunStartError(404, "This analysis run is not visible.")
+    return started

backend/app/main.py

@@ -72,6 +72,10 @@
     fetch_visible_analysis_run,
     fetch_visible_analysis_runs,
 )
+from backend.app.analysis_run_start import (
+    AnalysisRunStartError,
+    start_pending_analysis_run,
+)
 from backend.app.activity_stream import (
     create_valkey_client,
     get_valkey,
@@ -1253,6 +1257,32 @@ async def create_analysis_run(
     return created
 
 
+@app.post("/api/analysis-runs/{analysis_run_id}/start")
+async def start_analysis_run(
+    analysis_run_id: str,
+    account: CurrentAccount = Depends(get_current_account),
+    pool: asyncpg.Pool = Depends(get_pool),
+) -> dict[str, Any]:
+    """Start ThreadWeave on a visible Pending lineage run.
+
+    post_read is enough. Hidden runs 404. TEPP is 422 so this path
+    cannot invent a theta. A Succeeded retry returns the stored tree.
+    """
+    _require_post_read(account)
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            try:
+                started = await start_pending_analysis_run(
+                    conn,
+                    analysis_run_id=analysis_run_id,
+                    account_id=account.user_account_id,
+                    affiliated_entity_ids=list(account.corporate_entity_ids),
+                )
+            except AnalysisRunStartError as exc:
+                raise HTTPException(exc.status_code, exc.detail) from exc
+    return started
+
+
 @app.get("/api/analysis-runs/{analysis_run_id}")
 async def read_analysis_run(
     analysis_run_id: str,