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

ARCHITECTURE.md

@@ -475,7 +475,10 @@ 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 runs ThreadWeave on the cutoff bag and persists
+run-scoped edges; it does not replace live Event Lineage and does not
+invent a TEPP score.
 `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,4 @@
+# 0.89.0 Analysis-run start reconstruction
+
+Pending lineage rows can start ThreadWeave on the cutoff bag. The
+designed A-100 fork appears as titled edges. TEPP start stays 422.

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.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). 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

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,127 @@ 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 0020 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_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,

backend/app/analysis_run_start.py

@@ -0,0 +1,189 @@
+"""Start a Pending lineage reconstruction without inventing a TEPP score.
+
+ADR 0020. ``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_cutoff_reconstruct_posts,
+    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 _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,
+    )
+
+
+def _next_status_ordinal(current: dict[str, Any]) -> int:
+    """Continue the append-only lifecycle after the last visible event."""
+    history = current.get("status_history") or []
+    ordinals = [int(event["status_ordinal"]) for event in history]
+    return (max(ordinals) if ordinals else 0) + 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 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)
+    running_ordinal = _next_status_ordinal(current)
+    await _append_status(conn, analysis_run_id, running_ordinal, _RUNNING, now)
+
+    locked = await conn.fetchrow(
+        """
+        select run.analysis_run_id, run.knowledge_cutoff,
+               scope.scope_kind_code, scope.corporate_entity_id,
+               scope.process_unit_id, scope.scope_key
+        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,
+    )
+    if locked is None:
+        raise AnalysisRunStartError(404, "This analysis run is not visible.")
+    rows = await fetch_cutoff_reconstruct_posts(
+        conn,
+        locked["scope_kind_code"],
+        locked["corporate_entity_id"],
+        locked["process_unit_id"],
+        locked["scope_key"],
+        affiliated_entity_ids,
+        locked["knowledge_cutoff"],
+    )
+    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)
+    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