Plan 001: Make pipeline runs crash-safe — guard worker exceptions and checkpoint results periodically

[strickvl]

Plan 001: Make pipeline runs crash-safe — guard worker exceptions and checkpoint results periodically

Executor instructions: Follow this plan step by step. Run every
verification command and confirm the expected result before moving to the
next step. If anything in the "STOP conditions" section occurs, stop and
report — do not improvise. When done, update the status row for this plan
in plans/README.md.

Drift check (run first): git diff --stat 5b5c634..HEAD -- src/process_and_extract.py src/constants.py
If any in-scope file changed since this plan was written, compare the
"Current state" excerpts against the live code before proceeding; on a
mismatch, treat it as a STOP condition.

Status

• Priority: P1
• Effort: M
• Risk: LOW
• Depends on: none
• Category: bug
• Planned at: commit 5b5c634, 2026-06-12

Why this matters

The pipeline extracts entities from articles using a thread pool, merges the
results in memory on the main thread, and writes the merged entity tables to
disk once, at the very end of the run (write_results_and_statistics).
There is no try/except around the per-article result consumption: if any
worker raises an unexpected exception (a malformed article row, an LLM SDK
error that slips past internal handling), future.result() re-raises it on
the main thread, main() has no guard, and the process dies. Every merge
decision made up to that point — potentially hours of LLM calls — is lost.
(The extraction sidecar cache survives, so raw LLM outputs are not re-paid,
but all merge/dedup work is.) After this plan: one bad article is logged and
skipped instead of killing the run, and results are flushed to disk every N
articles so a crash loses at most one checkpoint interval of merge work.

Current state

• src/process_and_extract.py — CLI entry point and pipeline orchestration (874 lines).

  • process_articles_batch() (line 666) submits one extraction task per
    article to a ThreadPoolExecutor, then the main thread consumes futures
    in submission order and merges:

    # src/process_and_extract.py:748-762
            # --- Consume results in submission (article) order ---
            for future in futures:
                result = future.result()
                processed_rows.append(result.row)
    
                merge_and_finalize(
                    result,
                    entities=entities,
                    processor=processor,
                    args=args,
                    domain_config=domain_cfg,
                    status_tracker=status_tracker,
                    counters=counters,
                )
    
                progress.update(task_id, advance=1)

    futures is built (lines 733–745) by iterating
    enumerate(active_rows, 1), so futures[i] corresponds to
    active_rows[i]. Each row is a dict with an "id" key (see
    extract_single_article_only, line 532: article_id = article_info["id"]).

  • write_entities_to_files() (line 142) writes each entity type's full
    table via write_entities_table(entity_type, list(entity_dict.values()), base_dir).
    It is currently called only from write_results_and_statistics()
    (line 480), which main() calls once at line 860 after ALL articles finish.

  • status_tracker.flush() (atomic temp-file + os.replace, see
    src/utils/processing_status.py:75-86) is likewise only called once, at
    line 494 inside write_results_and_statistics().

  • Inside process_articles_batch, the resolved domain config is available
    as domain_cfg (line 699: domain_cfg = domain_config or DomainConfig(processor.domain)),
    and DomainConfig.get_output_dir() returns the domain's output directory
    (used the same way in src/frontend/data_access.py:58).

• src/constants.py — module of plain constants; env-var override pattern
  already in use, e.g. line 7:
  CLOUD_MODEL = os.getenv("HINBOX_CLOUD_MODEL", "gemini/gemini-2.0-flash").
• merge_and_finalize() (line 606) is the only writer to entities and
  status_tracker — it already wraps each entity-type merge in try/except
  (lines 301–325 of the file show the same pattern in merge_all_entities).
  The gap is exceptions raised before the result object exists, i.e. from
  the worker itself.
• Conventions: typing.Dict / typing.Tuple (not builtins) for hints;
  logging via the module's existing log(...) helper
  (from src.logging_config import ..., see existing log(f"Error merging {entity_type}", level="error", exception=e) at line 324).

Commands you will need

Purpose	Command	Expected on success
Tests	just test (or uv run pytest tests/ -v)	all pass
Single test file	uv run pytest tests/test_pipeline_crash_safety.py -v	all pass
Lint+format	just format && just lint	exit 0
Full CI parity	just ci	exit 0

Scope

In scope (the only files you should modify/create):

• src/process_and_extract.py
• src/constants.py
• tests/test_pipeline_crash_safety.py (create)

Out of scope (do NOT touch):

• src/utils/processing_status.py — its flush is already atomic; no changes needed.
• src/utils/file_ops.py — write_entities_table already does batched writes.
• src/engine/** — merge internals are unrelated to this plan.
• Any change to the extraction sidecar cache.

Git workflow

• Branch: advisor/001-crash-safe-pipeline
• Commit style: imperative subject, backticks around code identifiers
  (matches repo history, e.g. Fix Parquet write amplification: batch writes per entity type).
• Do NOT push or open a PR unless the operator instructed it.

Steps

Step 1: Add a checkpoint-interval constant

In src/constants.py, near the other pipeline constants (after the
MAX_ITERATIONS = 3 block), add:

# Crash-safety: flush entity tables + processing status every N merged articles.
# 0 disables periodic checkpointing (final write still happens).
CHECKPOINT_INTERVAL = int(os.getenv("HINBOX_CHECKPOINT_INTERVAL", "25"))

Verify: uv run python -c "from src.constants import CHECKPOINT_INTERVAL; print(CHECKPOINT_INTERVAL)" → 25

Step 2: Guard worker exceptions in the consume loop

In process_articles_batch() (src/process_and_extract.py:748), change the
consume loop so a raising future is recorded and skipped instead of crashing
the run. Pair each future with its source row so the article id is known even
when the result object never materializes:

            for (row_index, row), future in zip(
                enumerate(active_rows, 1), futures
            ):
                try:
                    result = future.result()
                except Exception as e:
                    article_id = str(row.get("id", f"row-{row_index}"))
                    log(
                        f"Worker failed for article {article_id}; skipping",
                        level="error",
                        exception=e,
                    )
                    if status_tracker:
                        status_tracker.mark_skipped(
                            article_id, f"worker_error: {type(e).__name__}: {e}"
                        )
                    processed_rows.append(row)
                    progress.update(task_id, advance=1)
                    continue

                processed_rows.append(result.row)
                merge_and_finalize(...)   # unchanged existing call
                progress.update(task_id, advance=1)

Match the existing log(...) call signature used at line 324
(log(msg, level="error", exception=e)). Keep the merge_and_finalize
arguments exactly as they are today.

Verify: just lint → exit 0 (tests come in Step 4)

Step 3: Checkpoint entities + status every N articles

Still in process_articles_batch():

1. Import CHECKPOINT_INTERVAL from src.constants (top of file, with the
   existing constants imports).
2. Before the with Progress(...) block, compute
   checkpoint_dir = domain_cfg.get_output_dir().
3. In the consume loop, maintain a counter of successfully merged articles.
   After each successful merge_and_finalize(...), when
   CHECKPOINT_INTERVAL > 0 and merged_count % CHECKPOINT_INTERVAL == 0:

                    if status_tracker:
                        status_tracker.flush()
                    write_entities_to_files(entities, checkpoint_dir)
                    log(
                        f"Checkpoint: wrote entity tables after {merged_count} articles",
                        level="info",
                    )

write_entities_to_files is defined in the same module (line 142) — no new
import needed. The final write in write_results_and_statistics() stays as
is (it is idempotent: each call rewrites the full tables).

Verify: just lint → exit 0

Step 4: Tests

Create tests/test_pipeline_crash_safety.py. Use unittest.mock.patch /
MagicMock; never call real LLMs or embeddings. Patch
src.process_and_extract.extract_single_article_only,
src.process_and_extract.merge_and_finalize, and
src.process_and_extract.write_entities_to_files in the
src.process_and_extract namespace (they are defined there).

Build inputs as:

import argparse
from unittest.mock import MagicMock, patch

def _args(limit: int) -> argparse.Namespace:
    return argparse.Namespace(limit=limit)

def _domain_config(tmp_path):
    cfg = MagicMock()
    cfg.get_concurrency_config.return_value = {
        "extract_workers": 2, "extract_per_article": 1,
    }
    cfg.get_cache_config.return_value = {"enabled": False}
    cfg.get_output_dir.return_value = str(tmp_path)
    return cfg

Cases (see Test plan below for the full list). For the worker-failure case,
make the patched extract_single_article_only raise RuntimeError("boom")
when called with the second row and return a MagicMock(row=row) otherwise
(use side_effect with a function inspecting the row kwarg/arg).

Verify: uv run pytest tests/test_pipeline_crash_safety.py -v → all pass

Step 5: Full suite + CI parity

Verify: just ci → exit 0 (289+ tests pass; no lint/format errors)

Test plan

New tests in tests/test_pipeline_crash_safety.py (model the mock style on
tests/test_extraction_retry.py):

1. Worker exception does not kill the batch: 3 rows, worker raises on
   row 2 → process_articles_batch returns 3 processed rows,
   merge_and_finalize called exactly 2 times, and
   status_tracker.mark_skipped called once with an id containing row 2's id
   and a reason starting worker_error:.
2. Checkpoint cadence: 5 rows, CHECKPOINT_INTERVAL patched to 2
   (patch src.process_and_extract.CHECKPOINT_INTERVAL if imported as a
   name, or monkeypatch accordingly) → write_entities_to_files called 2
   times by the loop (after articles 2 and 4) and status_tracker.flush
   called at least 2 times.
3. Checkpointing disabled: CHECKPOINT_INTERVAL = 0 → the loop calls
   write_entities_to_files 0 times.

Done criteria

• just ci exits 0
• uv run pytest tests/test_pipeline_crash_safety.py -v → 3+ new tests pass
• grep -n "future.result()" src/process_and_extract.py shows the call inside a try: block
• grep -n "CHECKPOINT_INTERVAL" src/constants.py src/process_and_extract.py → present in both
• No files outside the in-scope list modified (git status)
• plans/README.md status row updated

STOP conditions

Stop and report back (do not improvise) if:

• The consume loop at src/process_and_extract.py:748 no longer matches the
  excerpt (e.g. it already has a try/except, or futures are no longer
  ordered/zipped with active_rows).
• DomainConfig.get_output_dir() does not exist or requires arguments.
• Calling write_entities_to_files mid-run fails because entities contains
  non-serializable in-progress state — that would mean merge mutates entities
  in a way that is only valid at end-of-run, which changes the design.
• Test 1 reveals that worker exceptions are already swallowed somewhere
  inside extract_single_article_only's callees — then the guard is dead
  code and the plan's premise is wrong.

Maintenance notes

• If parallel merge (not just extraction) is ever introduced, the
  checkpoint write must move behind whatever lock protects entities.
• Reviewer should scrutinize: the zip of futures with active_rows
  (ordering assumption), and that a skipped article still appends its
  original row to processed_rows (downstream statistics count on row
  parity).
• Deferred: retrying a failed article within the same run (currently it is
  just marked skipped; rerunning the pipeline picks it up again because it
  was never marked processed).