Handle ingestion failures

Build a batch loader that keeps accepted documents and failed source attempts in one result. The supplied failure exercise genuinely loads 1 document and retains 3 failures. Its nonzero exit tells shell automation that ingestion is incomplete.

When you need this. The strict loader in the preceding lesson stops at the first bad source. A larger corpus may need a report that names every failed input so an operator can repair it. This lab reuses the same local loader and original Python API summaries. It adds per-source failure handling, without embedding or indexing a partial corpus.

Prerequisites. Run the first two ingestion labs. No model server or additional packages are needed.

Decide what stops the batch

An invalid manifest stops the batch before individual sources are attempted. Its entries define the source identities and paths; continuing with an ambiguous manifest would make the report hard to interpret.

A valid manifest can still name an unreadable or changed file. The batch catches the loader’s IngestionError for that source, retains its ID, path and error, and continues to the next listed entry. It does not catch every programming exception and disguise it as bad input.

Invalid manifest. Duplicate source identities or malformed manifest content must be repaired before rerunning. Per-file quarantine is for validly listed sources that failed to load, not for guessing what the manifest author intended.

Keep rejected attempts in the result

The result contains both documents and failures. Its status is complete when every source loads, partial when accepted and failed records coexist, and failed when no listed source loads. This status describes file ingestion, not whether a future answer will be correct.

Changed source. The exercise edits one temporary source without changing its expected hash. The loader reports hash_mismatch. Another source is removed and produces unreadable_source.

Invalid UTF-8. The exercise replaces a different temporary file with invalid UTF-8 and deliberately pins those new bytes in its temporary manifest. That lets the byte-hash check pass so the decoder’s invalid_utf8 boundary actually runs. The original snapshot stays unchanged.

The report acts as a local quarantine record. There is no queue, persistent quarantine service, retry scheduler or vector-store write. Repair the named inputs and run another batch with a new output path. Resume and incremental index updates belong to later modules.

Execute the controlled failure batch

Download the complete lab. The copied loader is identical to the preceding lesson’s implementation. The failure driver changes temporary copies only, after validating the supplied snapshot.

"""Quarantine bad local sources; explicitly report partial ingestion."""
import argparse
import hashlib
import json
from pathlib import Path
import tempfile
from load_documents import IngestionError, load_source, read_manifest

def ingest(manifest: Path) -> dict:
    # An invalid manifest fails the whole batch before any source is loaded.
    sources = read_manifest(manifest)
    documents, failures = [], []
    for source in sources:
        try:
            documents.append(load_source(manifest.parent, source))
        except IngestionError as error:
            failures.append({'source_id': source['source_id'], 'path': source['path'], 'error': str(error)})
    return {'schema_version': 1, 'execution_mode': 'offline-local-files',
            'status': 'partial' if failures and documents else 'failed' if failures else 'complete',
            'documents': documents, 'failures': failures}

def failure_exercise(manifest: Path) -> dict:
    """Mutate temporary copies only; leave the pinned source snapshot untouched."""
    sources = read_manifest(manifest)
    if len(sources) < 4:
        raise IngestionError('exercise_requires_four_sources')
    # Validate the entire snapshot before staging; never copy unchecked paths.
    for source in sources:
        load_source(manifest.parent, source)
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)
        for source in sources:
            target = root / source['path']; target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes((manifest.parent / source['path']).read_bytes())
        # Changed bytes are rejected before decoding, preserving the pinned contract.
        (root / sources[1]['path']).write_bytes(b'changed after snapshot')
        (root / sources[2]['path']).unlink()
        invalid = b'\xff'
        (root / sources[3]['path']).write_bytes(invalid)
        sources[3] = dict(sources[3], sha256=hashlib.sha256(invalid).hexdigest())
        copy = root / 'manifest.json'
        copy.write_text(json.dumps({'schema_version': 1, 'sources': sources}))
        return ingest(copy)

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--manifest', type=Path, default=Path(__file__).parent / 'fixtures/source-manifest.json')
    parser.add_argument('--output', type=Path, required=True)
    parser.add_argument('--exercise-failures', action='store_true')
    args = parser.parse_args()
    result = failure_exercise(args.manifest) if args.exercise_failures else ingest(args.manifest)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, indent=2) + '\n')
    print(json.dumps({'status': result['status'], 'loaded': len(result['documents']), 'failures': result['failures']}, sort_keys=True))
    # Partial completion is never advertised as full success to shell automation.
    raise SystemExit(0 if result['status'] == 'complete' else 2)

if __name__ == '__main__': main()

python3 ingest.py --output /tmp/ingestion-complete.json
python3 ingest.py --exercise-failures --output /tmp/ingestion-partial.json
python3 -m unittest -v test_load_documents.py test_ingest.py

{
  "failures": [
    {
      "error": "hash_mismatch",
      "path": "docs/path-read.md",
      "source_id": "path-read"
    },
    {
      "error": "unreadable_source",
      "path": "docs/json-text.md",
      "source_id": "json-text"
    },
    {
      "error": "invalid_utf8",
      "path": "docs/path-read-3-13.md",
      "source_id": "path-read-3-13"
    }
  ],
  "loaded": 1,
  "status": "partial"
}

The first command exits successfully with the unmodified fixture. The controlled failure command intentionally exits with status 2; run the tests as a separate command so that expected failure does not skip them in shell automation.

{
  "schema_version": 1,
  "execution_mode": "offline-local-files",
  "status": "partial",
  "documents": [
    {
      "source_id": "path-write",
      "path": "docs/path-write.md",
      "title": "path-write",
      "sha256": "a48b4b9f763cd6309ebe32af19c5a814adcf810c2259683da9f328ccee46948e",
      "version": "3.12",
      "url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text",
      "fact_status": "original summary; primary docs checked",
      "text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written."
    }
  ],
  "failures": [
    {
      "source_id": "path-read",
      "path": "docs/path-read.md",
      "error": "hash_mismatch"
    },
    {
      "source_id": "json-text",
      "path": "docs/json-text.md",
      "error": "unreadable_source"
    },
    {
      "source_id": "path-read-3-13",
      "path": "docs/path-read-3-13.md",
      "error": "invalid_utf8"
    }
  ]
}

This is actual deterministic local execution. It contains no model-generated answer or retrieval-quality measurement.

Refuse to promote incomplete ingestion

Partial batch. The CLI writes its result and returns exit 2 whenever the status is not complete. A caller must inspect that status before treating the accepted subset as the intended corpus. Successfully writing JSON does not mean every source loaded.

Try running once without --exercise-failures and once with it, using different output paths. Compare the status, source IDs in failures and process exit codes. Then inspect the tests for a batch where every source fails: its status must be failed, and its failures must remain visible.

The lab has no atomic corpus-promotion operation, retry policy or load benchmark. It reads small local files sequentially. Protecting concurrent writes and measuring large-corpus throughput require separate work.

What’s next

Continue with Version Documents and Chunks.

References

Implementation APIs were checked on 23 October 2026 against official Python pathlib, JSON and temporary-file documentation. Source summaries retain their original version-specific links. The implementation uses only the Python standard library.