diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 90c2ffbde..80021a527 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -733,9 +733,24 @@ async def _deliver_lineage_reconstruction( f"channels ({', '.join(sorted(active_channels))}). Run " "scripts/estimate_channel_weights.py, then start this run again.", ) - edges = lineage_edge_specs( - records_from_source_posts(rows), llm=adjudication_client, weights=weights - ) + try: + edges = lineage_edge_specs( + records_from_source_posts(rows), llm=adjudication_client, weights=weights + ) + except (HttpClientError, OSError, ValueError, TypeError) as exc: + # Only the llm channel talks to a provider mid-reconstruction; a + # deterministic run hitting this would be a code bug worth the + # crash. The enclosing transaction rolls back, so no partial + # graph persists (issue #289 RED 4/6) and the run stays + # startable once the provider recovers. + if "llm" not in active_channels: + raise + 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: diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index c2b1b6065..038ed0228 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -171,3 +171,91 @@ def test_running_restart_conflicts_and_succeeded_replay_is_documented() -> None: ) assert running.status_code == 409 assert "Pending" in running.detail + + +def test_provider_failure_mid_reconstruction_fails_closed_without_partial_graph() -> None: + """Issue #289 RED 4/6: an adjudication provider error during an + llm-inclusive delivery becomes an actionable 503 -- never a fabricated + score, never a partially persisted graph (no insert runs). + """ + import asyncio + + from backend.app.analysis_run_start import _deliver_lineage_reconstruction + from lineageweave.http_client import HttpClientError + + class _FailingJudge: + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + raise HttpClientError("HTTP 500 from orchestrator") + + class _Conn: + def __init__(self) -> None: + self.executed: list[str] = [] + + def transaction(self): + class _Tx: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback) -> bool: + return False + + return _Tx() + + async def fetch(self, query: str, *args: object): + if "lineage_channel_weight" in query: + return [ + {"channel_set_code": "channel_set_with_llm", "channel_code": "temporal", "weight_value": 0.4}, + {"channel_set_code": "channel_set_with_llm", "channel_code": "secondary_key", "weight_value": 0.1}, + {"channel_set_code": "channel_set_with_llm", "channel_code": "text", "weight_value": 0.1}, + {"channel_set_code": "channel_set_with_llm", "channel_code": "llm", "weight_value": 0.4}, + ] + if "analysis_source_snapshot_member" in query: + return [ + { + "post_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "post_title": "First post in thread", + "created_at": datetime(2026, 1, 1), + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "T-100", + "secondary_grouping_key": "", + }, + { + "post_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + "post_title": "Second post in thread", + "created_at": datetime(2026, 1, 2), + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "T-100", + "secondary_grouping_key": "", + }, + ] + raise AssertionError(f"unexpected query: {query[:80]}") + + async def execute(self, query: str, *args: object) -> str: + self.executed.append(query) + return "OK" + + conn = _Conn() + locked = { + "analysis_source_snapshot_id": "snapshot-1", + "corporate_entity_id": "corp", + "knowledge_cutoff": datetime(2026, 2, 1, tzinfo=timezone.utc), + } + with pytest.raises(AnalysisRunStartError) as raised: + asyncio.run( + _deliver_lineage_reconstruction( + conn, + analysis_run_id="00000000-0000-0000-0000-000000000009", + locked=locked, + affiliated_entity_ids=[], + adjudication_client=_FailingJudge(), + ) + ) + assert raised.value.status_code == 503 + assert "start this run again" in raised.value.detail + assert conn.executed == []