fix(passthrough): stop reconcile worker from re-selecting permanently-failed transactions forever

[sjawhar]

Problem

PassthroughReconcileWorker (gated by PASSTHROUGH_MATERIALIZE_BACKFILL_ENABLED, running in the luthien-admin ECS service) runs reconcile_passthrough every 300s forever. Each sweep:

1. Runs _ELIGIBLE_UNMATERIALIZED_TRANSACTIONS_SQL to find the oldest 200 unmaterialized passthrough transactions. request_logs had no index supporting the query's filter, so this ran a Parallel Seq Scan discarding essentially the whole table every cycle.
2. Attempts to materialize all 200. When a transaction is permanently unparseable (missing_required_field, unsupported_variant, etc.), it fails.
3. reconcile.py did failed += 1; continue and never persisted the failure anywhere the eligibility query's NOT EXISTS(conversation_events) check could see.

This PR stacks on #796 (feat/passthrough-multiprovider-capture) because that's the branch that introduces passthrough_materialize/ -- none of this code exists on main yet. GitHub cannot target a PR's base at a branch that only exists in a fork (not upstream), so this PR's base is main and its diff necessarily includes #796's full changeset until #796 merges. The commits unique to this fix are the single top commit, fix(passthrough): stop reconcile worker from re-selecting permanently-failed transactions forever -- everything else is #796's. This PR should not merge before #796 does; once #796 merges, this branch will be rebased onto main and its diff will shrink to just that one commit.

Fix

1. Dead-letter permanent failures (reconcile.py)

Every MaterializationFailed reason except missing_request_logs is raised only after read_raw_transaction/parse_captured_transaction successfully fetched request_logs bytes and then found them unparseable/invalid for the matched endpoint (the PassthroughNormalizeReason enum values, missing_endpoint, and the _InvalidRequestLog family of invalid_<column>/missing_<column> reasons). request_logs rows are never updated after insert, so retrying one of those without a code change reproduces the identical outcome forever -- they're permanent.

missing_request_logs (zero rows found for the transaction_id) is the one reason that reflects an absence rather than malformed content already read. Treating it as permanent risks permanently blacklisting a transaction whose row simply hadn't landed yet, so it's the only reason left retryable. Raw DB exceptions (timeouts, connection errors) were already handled separately in reconcile_passthrough's try/except before ever becoming a MaterializationFailed -- untouched by this change, and correctly never dead-lettered.

Permanent failures are now recorded in a new passthrough_materialization_dead_letters table (transaction_id PK, reason, failed_at) and excluded via NOT EXISTS in the eligibility query.

2. Index (migrations/023_add_passthrough_materialization_dead_letters.sql)

A partial index whose WHERE clause mirrors the eligibility query's predicate verbatim (direction = 'inbound' AND the four passthrough endpoint patterns), so Postgres can use it as a covering index-only scan restricted to just the rows that could ever be eligible, rather than indexing everything that's merely direction = 'inbound'.

EXPLAIN evidence (measured locally against a seeded Postgres 16 instance: 2.4M request_logs rows -- 1.2M outbound, 1.2M inbound non-passthrough, 3,000 already-materialized passthrough, 5,265 permanently-broken passthrough matching the production backlog's date range and size):

	Before	After
Plan	Parallel Seq Scan on request_logs	Index Only Scan using idx_request_logs_passthrough_eligible
Rows removed by filter (per worker)	800,000	0 (fully covered by the partial predicate)
Planner cost	70408.58	309.86 (~227x)
Execution time	194ms (cold)	7ms
Buffers	1204 hit + 45295 read	102 hit, 0 read

A second sweep, after dead-lettering the 5,265-row backlog, returns 0 rows in 6ms via the same index -- confirming convergence.

CREATE INDEX CONCURRENTLY is used given the table's size (multi-GB, multi-million-row, live production traffic); verified psql -f commits the CREATE TABLE and CREATE INDEX CONCURRENTLY in this one file as two independent statements (not wrapped in an implicit transaction), and confirmed indisvalid = true after the build.

3. One-shot semantics (worker.py)

PASSTHROUGH_MATERIALIZE_BACKFILL_ENABLED's own config docstring says "Enable one-shot passthrough materialization backfill for historical request_logs" -- but the worker looped on a fixed interval forever regardless of backlog state, which is what turned "permanently broken rows never converging" into "runs forever burning ACU." PassthroughReconcileWorker now stops itself once a sweep selects zero eligible rows at all (distinct from "nothing succeeded" -- a sweep that only hit failures must keep running to dead-letter/retry them). This makes the worker's actual behavior match its documented one-shot contract, and means a drained backlog stops all further DB load until the next app restart (one more sweep, immediately terminates).

Tests

13 new/updated tests in tests/luthien_proxy/unit_tests/passthrough_materialize/{test_reconcile,test_worker}.py:

• test_reconcile_materializes_only_eligible_unmaterialized_transactions_and_converges_next_sweep -- updated from the pre-fix version, which asserted the bug (second_pass == ReconcileStats(failed=1), i.e. the same broken row re-selected forever); now asserts second_pass == ReconcileStats() and that the dead-letter row was written with reason="missing_required_field".
• test_reconcile_converges_when_the_entire_backlog_is_permanently_broken -- reproduces the production bug shape directly (an all-broken backlog) and shows sweep 2 selects nothing.
• test_reconcile_leaves_a_transient_failure_reason_eligible_for_retry -- missing_request_logs is re-selected and re-attempted on both sweeps, never dead-lettered.
• test_migration_creates_the_dead_letter_table_and_supporting_index -- SQLite side of migration 023 applies cleanly (exercised via check_migrations in every fixture in this file already; this asserts the resulting schema directly). Postgres side verified manually (see EXPLAIN evidence above; a live Postgres instance isn't available in the default dev_checks.sh/CI gate -- tests/luthien_proxy/integration_tests/test_migration_sync.py is @pytest.mark.integration and pre-existing, not selected by default, per fix(migrations): stop swallowing failed migrations in run-migrations.sh #812's notes).
• test_reconcile_worker_stops_once_a_sweep_selects_nothing / test_reconcile_worker_keeps_sweeping_while_failures_remain -- the new one-shot termination behavior, and that it doesn't fire prematurely while there's still failed work to process.

Full suite: 3177 tests, 3174 passed, 3 failed. The 3 failures (test_onboard.py::test_find_docker_ports_respects_env_vars, test_config_registry.py::TestResolveDefaults::test_default_value_when_no_overrides, test_config_registry.py::TestResolvePriority::test_db_ignored_for_non_db_settable) are pre-existing and unrelated -- reproduced byte-for-byte on a clean checkout of this PR's own base branch (feat/passthrough-multiprovider-capture tip, before any of this PR's changes) in this sandbox, and pass individually in isolation (environment-dependent test-isolation flakiness in POSTGRES_PORT/REDIS_PORT env var resolution, unrelated to passthrough_materialize). ruff format, ruff check, and pyright are all clean (0 errors/warnings).

Non-goals

• Not touching scripts/backfill_passthrough_materialization.py / drain_passthrough_backfill (the manual one-shot CLI script) -- it already terminates correctly (sweep.materialized == 0); it benefits from the index and dead-letter table for free.
• Not adding an operator UI/endpoint to clear dead-lettered rows for re-materialization after a future parser fix ships -- out of scope for this concern; the table is a plain transaction_id-keyed row, trivial to clear later.

CI: expect claude-review to fail (ANTHROPIC_API_KEY not exposed to fork-triggered runs, per repo convention) -- everything else should be green.