feat(estimation): authorize the ADR 0200 anchor method — weights activate

backend/app/analysis_run_start.py

@@ -27,7 +27,10 @@
     latest_outbox_delivery_is_delivered,
     outbox_request_digest,
 )
-from backend.app.lineage_ingestion import records_from_source_posts
+from backend.app.lineage_ingestion import (
+    load_estimated_channel_weights,
+    records_from_source_posts,
+)
 from lineageweave.adjudication_client import AdjudicationClient
 from lineageweave.http_client import HttpClientError, post_json
 from lineageweave.lineage_persistence import lineage_edge_specs
@@ -49,6 +52,34 @@ class AnalysisRunStartError(AnalysisRunCreateError):
     """Fail-closed start: HTTP status plus a next-action detail string."""
 
 
+class _AdjudicationProviderError(RuntimeError):
+    """The adjudication provider (not our code) failed during a judge call."""
+
+
+class _ProviderBoundaryAdjudication:
+    """Convert judge-call provider failures into a typed boundary error.
+
+    ``post_json`` raises ``HttpClientError``/``OSError`` on transport and
+    HTTP failures and the content extraction raises
+    ``ValueError``/``TypeError`` on malformed provider envelopes; those
+    same builtin types raised by our reconstruction code would be real
+    bugs, so the conversion happens only inside ``judge`` -- the one
+    place a provider is actually on the line.
+    """
+
+    available = True
+
+    def __init__(self, inner) -> None:
+        self._inner = inner
+
+    def judge(self, candidate_label: str, record_label: str) -> float:
+        """Score one candidate pair, typing provider failures as such."""
+        try:
+            return self._inner.judge(candidate_label, record_label)
+        except (HttpClientError, OSError, ValueError, TypeError) as exc:
+            raise _AdjudicationProviderError(str(exc)) from exc
+
+
 def reconstruction_result_digest(edges: list[Edge]) -> str:
     """SHA-256 of the ordered parent choices. Never hashes a post body."""
     material = json.dumps(
@@ -716,7 +747,42 @@ async def _deliver_lineage_reconstruction(
             knowledge_cutoff=locked["knowledge_cutoff"],
             affiliated_entity_ids=affiliated_entity_ids,
         )
-    edges = lineage_edge_specs(records_from_source_posts(rows), llm=adjudication_client)
+    # ADR 0200 point 1: weights are measurement output only -- the
+    # activated fast-mlsirm set matching this run's active channels
+    # (four when the adjudication client is available, three
+    # deterministic otherwise). A missing set fails the start with a
+    # next-action message instead of reconstructing on constants.
+    active_channels = {"temporal", "secondary_key", "text"}
+    if adjudication_client is not None and getattr(adjudication_client, "available", False):
+        active_channels = active_channels | {"llm"}
+    weights = await load_estimated_channel_weights(conn, active_channels)
+    if weights is None:
+        raise AnalysisRunStartError(
+            503,
+            "Channel weights are not estimated yet for this run's active "
+            f"channels ({', '.join(sorted(active_channels))}). Run "
+            "scripts/estimate_channel_weights.py, then start this run again.",
+        )
+    # The provider boundary is exactly llm.judge() -- wrapping the whole
+    # pipeline would disguise a reconstruction code bug as a retryable
+    # provider outage. The enclosing transaction rolls back either way,
+    # so no partial graph persists (issue #289 RED 4/6).
+    guarded_client = (
+        _ProviderBoundaryAdjudication(adjudication_client)
+        if "llm" in active_channels
+        else adjudication_client
+    )
+    try:
+        edges = lineage_edge_specs(
+            records_from_source_posts(rows), llm=guarded_client, weights=weights
+        )
+    except _AdjudicationProviderError as exc:
+        raise AnalysisRunStartError(
+            503,
+            "The adjudication provider failed mid-reconstruction; nothing "
+            "was persisted. Check the contextual-orchestrator transport, "
+            "then start this run again.",
+        ) from exc
     digest = reconstruction_result_digest(edges)
     finished = datetime.now(timezone.utc)
     if finished < now:

backend/app/analysis_run_worker.py

@@ -7,16 +7,21 @@
 
 from __future__ import annotations
 
+import logging
+
 import asyncpg
 import redis.asyncio as redis
 from uuid import UUID
 
 from lineageweave.adjudication_client import AdjudicationClient
 from lineageweave.tepp_client import TeppClient
 
+from backend.app.analysis_run_ingestion import AnalysisRunCreateError
 from backend.app.analysis_run_outbox import OUTBOX_STREAM_KEY
 from backend.app.analysis_run_start import deliver_queued_analysis_run
 
+logger = logging.getLogger(__name__)
+
 
 async def consume_analysis_run_stream_once(
     client: redis.Redis,
@@ -40,26 +45,40 @@ async def consume_analysis_run_stream_once(
             except ValueError:
                 analysis_run_id = ""
             if analysis_run_id:
-                async with pool.acquire() as conn:
-                    async with conn.transaction():
-                        owner = await conn.fetchrow(
-                            """
-                            select requested_by_account_id
-                            from analysis_run
-                            where analysis_run_id = $1::uuid
-                            """,
-                            analysis_run_id,
-                        )
-                        if owner is not None:
-                            await deliver_queued_analysis_run(
-                                conn,
-                                analysis_run_id=analysis_run_id,
-                                account_id=str(owner["requested_by_account_id"]),
-                                affiliated_entity_ids=[],
-                                tepp_client=tepp_client,
-                                adjudication_client=adjudication_client,
-                                valkey_stream_entry_id=str(entry_id),
+                # One run's fail-closed refusal (404/409/503, e.g. channel
+                # weights not estimated yet, ADR 0145) must not end the
+                # worker task and halt every later run's delivery. The
+                # transaction rolls back, the durable outbox row stays
+                # available, and an explicit HTTP start retries the run
+                # once the operator resolves the named next action.
+                try:
+                    async with pool.acquire() as conn:
+                        async with conn.transaction():
+                            owner = await conn.fetchrow(
+                                """
+                                select requested_by_account_id
+                                from analysis_run
+                                where analysis_run_id = $1::uuid
+                                """,
+                                analysis_run_id,
                             )
+                            if owner is not None:
+                                await deliver_queued_analysis_run(
+                                    conn,
+                                    analysis_run_id=analysis_run_id,
+                                    account_id=str(owner["requested_by_account_id"]),
+                                    affiliated_entity_ids=[],
+                                    tepp_client=tepp_client,
+                                    adjudication_client=adjudication_client,
+                                    valkey_stream_entry_id=str(entry_id),
+                                )
+                except AnalysisRunCreateError as exc:
+                    logger.warning(
+                        "analysis-run %s delivery refused (%s): %s",
+                        analysis_run_id,
+                        exc.status_code,
+                        exc.detail,
+                    )
             last_id = str(entry_id)
     return last_id
 

backend/app/lineage_ingestion.py

@@ -22,9 +22,15 @@
 from lineageweave.lineage_persistence import lineage_edge_specs
 from lineageweave.models import Edge, Record
 
-# ADR 0145 rejected the unanchored estimator. A future accepted ADR must add
-# its independently validated method code here before persisted weights can run.
-_SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset()
+# Accepted ADR 0200 (points 2-3) authorizes exactly one anchor method:
+# expected-information estimates honestly labeled as validated by the
+# channels' internal response structure only, pending the TEPP
+# criterion-validity gate. When that gate exists, a set that fails it is
+# retired and this stays the only place an anchor method is ever added
+# -- ADR-first, per ADR 0145's original condition.
+_SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset(
+    {"unanchored_internal_structure"}
+)
 
 
 def _occurred_at(value: datetime) -> datetime:
@@ -186,8 +192,33 @@ async def load_estimated_channel_weights(
     return persisted
 
 
+class ChannelWeightsNotEstimated(RuntimeError):
+    """No activated estimated weight set exists for the active channels.
+
+    Product reconstruction treats fusion weights as measurement output
+    only (ADR 0200 point 1): estimated by fast-mlsirm, provenance-gated,
+    never hand-picked constants. No hand-picked default exists anywhere
+    -- the library demo estimates its weights from its declared design,
+    and unit tests inject synthetic weights explicitly.
+    """
+
+    def __init__(self, active_channels: set[str]) -> None:
+        super().__init__(
+            "no activated fast-mlsirm channel weight estimate exists for "
+            f"active channels {sorted(active_channels)}; run "
+            "scripts/estimate_channel_weights.py first -- product "
+            "reconstruction never falls back to hand-picked weights"
+        )
+        self.active_channels = active_channels
+
+
 async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]:
-    """Reconstruct lineage for every ``source_post`` and persist the edges."""
+    """Reconstruct lineage for every ``source_post`` and persist the edges.
+
+    Raises :class:`ChannelWeightsNotEstimated` when no activated
+    estimate matches this path's active channels -- run
+    ``scripts/estimate_channel_weights.py`` first (ADR 0200 point 1).
+    """
     rows = await conn.fetch(
         "select post_id, post_title, voc_type_code, created_at, corporate_entity_id, "
         "process_unit_id, thread_group_key, secondary_grouping_key "
@@ -196,9 +227,10 @@ async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]:
     # No adjudication client is wired on this path, so the active channel
     # set is the three deterministic channels (reconstruct drops llm when
     # unavailable rather than faking it).
-    weights = await load_estimated_channel_weights(
-        conn, {"temporal", "secondary_key", "text"}
-    )
+    active_channels = {"temporal", "secondary_key", "text"}
+    weights = await load_estimated_channel_weights(conn, active_channels)
+    if weights is None:
+        raise ChannelWeightsNotEstimated(active_channels)
     edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights)
     await persist_lineage_edges(conn, edges)
     return edges

backend/app/main.py

@@ -170,6 +170,7 @@
     visible_team_mention_post_ids,
 )
 from backend.app.lineage_ingestion import (
+    ChannelWeightsNotEstimated,
     rebuild_lineage,
     visible_lineage_graph,
 )
@@ -368,12 +369,11 @@ def _post_summary_client():
 def _adjudication_client():
     """Live orchestrator client when configured; otherwise the unavailable null.
 
-    reconstruct.py's DEFAULT_CHANNEL_WEIGHTS gives this channel the most
-    weight (0.40) of the four -- it is the only one that reasons about
-    content instead of approximating it (ADR 0064) -- but nothing ever
-    passed a real client through lineage_edge_specs() to reconstruct(),
-    so every lineage reconstruction had silently run on the weaker
-    3-channel fallback since the feature was built.
+    The llm channel is the only one that reasons about content instead
+    of approximating it (ADR 0064). Its fusion weight -- like every
+    channel's -- is a fast-mlsirm estimate (ADR 0200): with this client
+    available, a four-channel run uses the persisted
+    `channel_set_with_llm` estimate and fails closed until one exists.
     """
     settings = load_settings()
     if not (settings.orchestrator_base_url and settings.orchestrator_api_key):
@@ -1169,7 +1169,14 @@ async def rebuild_lineage_graph(
     _require_post_admin(account)
     async with pool.acquire() as conn:
         async with conn.transaction():
-            edges = await rebuild_lineage(conn)
+            try:
+                edges = await rebuild_lineage(conn)
+            except ChannelWeightsNotEstimated:
+                raise HTTPException(
+                    status.HTTP_503_SERVICE_UNAVAILABLE,
+                    "Channel weights are not estimated yet. Run "
+                    "scripts/estimate_channel_weights.py, then rebuild again.",
+                )
     return {"edge_count": len(edges)}
 
 

backend/tests/test_api.py

@@ -122,6 +122,16 @@
     / "migrations"
     / "0135_lineage_channel_weight.sql"
 )
+_CHANNEL_WEIGHT_UNION_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "0200_channel_weight_schema_union.sql"
+)
+_PAIR_JUDGMENT_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "0201_lineage_pair_judgment.sql"
+)
 _LEFTOVER_OBSERVED_EXPECTED_MIGRATION = (
     Path(__file__).resolve().parents[2]
     / "migrations"
@@ -267,6 +277,28 @@ def seeded_db(demo_analyst_token):
             cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text())
             cur.execute(_TENANT_SETTINGS_MIGRATION.read_text())
             cur.execute(_CHANNEL_WEIGHT_MIGRATION.read_text())
+            cur.execute(_CHANNEL_WEIGHT_UNION_MIGRATION.read_text())
+            cur.execute(_PAIR_JUDGMENT_MIGRATION.read_text())
+            # Product reconstruction fails closed without an ACTIVATED
+            # estimate (ADR 0200 points 1+3); this synthetic fixture set
+            # under the authorized anchor stands in for a fast-mlsirm
+            # estimate in unit tests.
+            cur.execute(
+                "insert into lineage_channel_weight "
+                "(channel_set_code, channel_code, weight_value, "
+                " estimation_run_id, estimation_method_code, estimator_version, "
+                " anchor_method_code, source_snapshot_sha256, sample_pair_count, "
+                " knowledge_cutoff) values "
+                "('channel_set_deterministic', 'temporal', 0.5, "
+                " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', "
+                " 'unanchored_internal_structure', repeat('a', 64), 600, now()), "
+                "('channel_set_deterministic', 'secondary_key', 0.34, "
+                " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', "
+                " 'unanchored_internal_structure', repeat('a', 64), 600, now()), "
+                "('channel_set_deterministic', 'text', 0.16, "
+                " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', "
+                " 'unanchored_internal_structure', repeat('a', 64), 600, now())"
+            )
             cur.execute(_LEFTOVER_OBSERVED_EXPECTED_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_RANK_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_COVERAGE_MIGRATION.read_text())

lineageweave/channel_weight_estimation.py

@@ -57,6 +57,29 @@
 _FIXTURE_SIMULATION_SEED = 20260824
 
 
+def fixture_design_digest() -> str:
+    """Reproducible SHA-256 of the demo's declared generative design.
+
+    Plays the role the corpus snapshot digest plays for production
+    estimates: the provenance row names exactly which design supported
+    the demo estimate. Deterministic by construction.
+    """
+    import hashlib
+
+    material = "\n".join(
+        [
+            *(
+                f"{channel}\t{probability}"
+                for channel, probability in sorted(_FIXTURE_FOLLOW_PROBABILITY.items())
+            ),
+            f"groups\t{_FIXTURE_GROUP_COUNT}",
+            f"pairs\t{_FIXTURE_PAIR_COUNT}",
+            f"seed\t{_FIXTURE_SIMULATION_SEED}",
+        ]
+    )
+    return hashlib.sha256(material.encode("utf-8")).hexdigest()
+
+
 def simulate_fixture_pair_scores() -> tuple[list[dict[str, float]], list[int]]:
     """Simulate the demo design's channel responses, deterministically.
 

lineageweave/lineage_persistence.py

@@ -20,7 +20,7 @@ def lineage_edge_specs(
     records: Sequence[Record],
     *,
     llm: AdjudicationClient | None = None,
-    weights: dict[str, float] | None = None,
+    weights: dict[str, float],
 ) -> list[Edge]:
     """Run reconstruct and return every resulting parent→child edge.
 
@@ -35,12 +35,11 @@ def lineage_edge_specs(
     faked) -- callers that want the highest-weighted reasoning channel
     actually contributing to real reconstructions must pass a real one.
 
-    ``weights`` defaults to ``None``, keeping ``reconstruct()``'s
-    documented fallback constants; callers with a persisted
-    psychometric estimate (ADR 0145) pass it here.
+    ``weights`` is required and always a psychometric estimate (ADR
+    0145, second amendment): the persisted fast-mlsirm corpus estimate
+    on product paths, or the demo-design estimate from
+    :func:`~lineageweave.channel_weight_estimation.estimate_fixture_channel_weights`.
+    No hand-picked default exists anywhere.
     """
-    if weights is None:
-        trees = reconstruct(list(records), llm=llm)
-    else:
-        trees = reconstruct(list(records), llm=llm, weights=weights)
+    trees = reconstruct(list(records), llm=llm, weights=weights)
     return [edge for tree in trees for edge in tree.edges]