Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions backend/app/analysis_run_start.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Comment on lines +736 to +739

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📝 Info: Rollback of partial graph is guaranteed by both callers

The no partial graph persists claim holds: the HTTP path (main.py:3617) and the worker (analysis_run_worker.py:56) both wrap delivery in conn.transaction() and handle AnalysisRunStartError, so the 503 raised before any insert rolls back.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

except (HttpClientError, OSError, ValueError, TypeError) as exc:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📝 Info: Catch list covers all provider-boundary errors

Every failure from llm.judge is caught: post_json raises HttpClientError/OSError/ValueError and chat_completion_content raises TypeError/ValueError. No malformed-response path escapes uncaught to a raw 500.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

# 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
Comment on lines +740 to +753

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Real reconstruction bugs disguised as retryable provider outages

On an llm-inclusive run, the except (HttpClientError, OSError, ValueError, TypeError) wraps the whole reconstruction pipeline, not just the provider call. A ValueError or TypeError from records_from_source_posts, RankWeave, or ThreadWeave becomes a 503 that tells operators the adjudication provider failed and to retry. A genuine data or code bug then hides as a transient outage that never surfaces.

Prompt for agents
In _deliver_lineage_reconstruction, the try block at backend/app/analysis_run_start.py:736-739 wraps lineage_edge_specs(records_from_source_posts(rows), ...) and the except at line 740 catches ValueError and TypeError in addition to HttpClientError and OSError. The intent (per the comment) is to map only a provider-boundary failure to a 503. However, on an llm-inclusive run the pure-computation parts of the pipeline (records_from_source_posts, rankweave.weighted_convex_fuse, threadweave.thread_messages) also run inside this try and can legitimately raise ValueError/TypeError for reasons unrelated to the provider. Those get misreported as a retryable provider 503, hiding real bugs. Consider narrowing what is wrapped so that only the provider call boundary is guarded: e.g. move records_from_source_posts(rows) outside the try, or restructure so that only HttpClientError/OSError (the exceptions the orchestrator transport actually raises) map to the 503, while ValueError/TypeError from adjudication response parsing are distinguished from ValueError/TypeError arising elsewhere in the pipeline.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — fixed in the follow-up PR: the conversion now happens only inside judge() via a _ProviderBoundaryAdjudication wrapper that raises a typed _AdjudicationProviderError; the delivery catches only that type, so a stray ValueError/TypeError from our own reconstruction code crashes loudly again on llm runs too.

digest = reconstruction_result_digest(edges)
finished = datetime.now(timezone.utc)
if finished < now:
Expand Down
88 changes: 88 additions & 0 deletions tests/test_analysis_run_start.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 == []