Long-running ingestion — interrupt, checkpoint, resume

The run lifecycle: converging over bounded windows

A concrete run: 112 segments, a corpus large enough that one pass needs several short-lived instances.

Three properties make this converge instead of just retrying forever: every work unit is enumerated before the first model call, so "pending" always has a concrete count; a success is committed the moment it lands, never batched up and lost together on the next failure; and a failure marks exactly the one unit that failed, so a rerun's cost is proportional to what's left, not to what already happened. None of this requires Taguru to know about "runs" or "jobs" at all — taguru extract's manifest and chunk checkpoints, plus taguru import's retract-then-apply idempotency, are what make each window's work durable on its own.

The split: what Taguru guarantees, what your runner decides

Every pattern on this page composes these guarantees; none of them require Taguru to be aware that a "long job" is happening.

Taguru ships no --max-minutes or --max-items flag, on extract or import, and none is planned — bounding a run's window is deliberately left to the runner. The reason it can be: because rerunning taguru extract/taguru import is free for anything already done, a bound needs no cooperation from Taguru to be safe. timeout(1), a cron/scheduler window, or a runner's own argument parsing are all sufficient — see below.

Work units, checkpoint keys, and deciding what is done

The default unit is one segment. One file under taguru extract's input directory becomes one source file, whose header names its own source id, which becomes one row in --out's manifest. The same one-segment, one-source-id granularity holds for the standard ingest connectors: one PDF, DOCX, PPTX file, or S3 object becomes one ConnectorDocument, one retract-then-apply unit — page, slide, and table boundaries surface as citation locators within that one segment, not as separate source ids or separate work units. For a corpus of a few large PDFs where that granularity is too coarse (per-section retries, per-section deletion), the natural unit shrinks to one section: split each PDF by heading yourself and feed each section through as its own segment with a source id that names its parent, e.g. manual.pdf#installation — a runner's choice to make when it wants finer granularity than a connector gives by default, not something Taguru requires. Each section is then an independent source file, its own manifest row, its own retract-then-apply unit — a failure in one section never touches another's.

Two distinct completion tests exist, one per stage — do not conflate them:

• Extract stage: a segment is done when its entry in .extract-manifest.json matches the current compute-input fingerprint (content hash × model × prompt version × context × output-shaping flags). taguru extract --dry-run reports this without calling the model. A connector-driven run has its own two-layer version of this same test — see ingest connectors: checkpoints.
• Import stage: querying a running server's GET /contexts/{id}/sources (or the SDK's iter_sources(), see below) tells you exactly which sources have landed. That test is sound only because a failed source now writes nothing — a partially-applied source can no longer masquerade as a completed one (the fix that closed this gap is issue #187; before it, "source exists" was not a safe checkpoint test for a resumable client). Offline, there is no per-source listing (taguru inspect reports a source count and any surviving import markers, not individual ids) — but you rarely need one there: retract-then-apply makes taguru import idempotent, so simply re-importing the whole --out directory after every window is cheap and correct without first figuring out what already landed.

Below segment granularity, chunk checkpoints key each unit by the hash of its own text, not chunk index — so a resumed run's chunk splits don't have to line up with a prior run's for reuse to be recognized. Combined with the manifest above, a rerun only ever pays for chunks whose content, model, or output-shaping settings actually changed. A connector-driven run gets the same property from a different key: its own checkpoint namespaces (connector:, s3-object:, s3-inventory:) sit in the same CheckpointStore alongside chunk checkpoints, never colliding with them — see ingest connectors: checkpoints.

Stopping: the first signal, the second, and what a rerun sees

taguru extract's Ctrl+C/SIGTERM handling is cooperative and already covered in full at the checkpoints reference: the first signal finishes the chunk in flight, checkpoints it, and exits 130; a second signal forces an immediate exit, also 130, for when the graceful path itself is stuck. The LangChain SDK exposes the same shape as an argument, not ambient signal handling: pass should_stop (a zero-argument callable, or a threading.Event) to ingest_text()/ingest_documents() — checked between chunks, and when it fires, IngestOutcome.interrupted is True and nothing imports. Wiring an actual SIGINT handler that flips the event (first signal) or exits the process outright (second) is the runner's job — see below for one shape of it.

Path one: a bounded loop around the built-in CLI

No custom code — timeout(1) and exit-code dispatch around the two shipped subcommands.

# 45m and the -k grace period are timeout(1)'s — taguru extract has no time flag of its own.
# -s INT sends the same signal Ctrl+C would (taguru's cooperative first-stage stop);
# -k 30 sends SIGKILL 30s later only if the process is still stuck on the graceful path.
timeout -k 30 -s INT 45m taguru extract --context cef2e28b-43f0-4b6c-8201-abab0785399f --name sake --description "酒蔵の知識" \
  --out sources/ docs/                # --name/--description are ignored once the context already exists
rc=$?
taguru import sources/            # apply whatever this window finished — safe even if extract above was cut short

# rc: 130 window's time ran out mid-run   → requeue the identical command
#      1   some segments failed, the rest completed → rerun retries only the failures
#           (the manifest records only successes, so a plain rerun already does this)
#      0   nothing left to extract → converged, the job is done
case "$rc" in
  130) echo "window expired, $(taguru extract --dry-run --out sources/ docs/ | grep -c 'would extract') unit(s) still not extracted" ;;
  1)   echo "some segments failed — see stderr above; rerun to retry them" ;;
  0)   echo "converged" ;;
esac

--dry-run's per-segment line is "{source}: would extract (…) → …" for anything not yet in the manifest, and "{source}: unchanged, skipped (--force re-extracts)" for the rest — counting the former is the honest "how much is left" signal, since neither line literally says "pending".

Bounding by count instead of time means passing a slice of the not-yet-extracted files as explicit arguments — taguru extract accepts individual file paths alongside directories, so taguru extract --out sources/ $(taguru extract --dry-run --out sources/ docs/ | sed -n 's/: would extract.*//p' | head -20) processes at most 20 segments this invocation; run lifecycle's relation-label convergence (extract.html's prompt section) happens within one invocation, so slicing this way is safe, just less able to converge vocabulary across the whole corpus in one shot. Avoid set -e in a wrapping shell script around this block — exit 130 is an expected, not exceptional, outcome here.

The manifest skip above only ever answers "unchanged, redo nothing" or "changed, redo everything" — a run that has already spent every model call on a segment but wants to retry a later step alone (a stricter validation rule, a rendering fix) has no cheaper path through it. --resume-from STEP is the other tool for that: it satisfies whatever a prior run's attempts log recorded (today, --out/.extract-trace/*.attempts.jsonl's prompt/call steps) instead of a live call, so downstream-only changes need not pay for the model again.

Path two: a custom runner on the LangChain SDK

When the work unit is finer than "one segment" (e.g. sections split from a handful of PDFs) or the deployment needs its own progress UI, a small runner around TaguruIngester gets the same properties. The flags below are this sample's own — nothing here is a Taguru API.

# runner.py — one process invocation, bounded by --max-minutes / --max-items.
# Both flags belong to this script's own argparse, not to taguru_langchain.
import argparse, signal, threading, time
from langchain_openai import ChatOpenAI
from taguru import Taguru
from taguru_langchain import TaguruIngester, FilesystemCheckpointStore

parser = argparse.ArgumentParser()
parser.add_argument("--max-minutes", type=float, default=45.0)
parser.add_argument("--max-items", type=int, default=None)
args = parser.parse_args()

stop = threading.Event()
def on_signal(signum, frame):
    # first SIGINT/SIGTERM: stop cooperatively between chunks;
    # second: fall through to the default handler and exit immediately.
    if stop.is_set():
        signal.signal(signum, signal.SIG_DFL)
        raise KeyboardInterrupt
    stop.set()
signal.signal(signal.SIGINT, on_signal)
signal.signal(signal.SIGTERM, on_signal)

client = Taguru()  # $TAGURU_URL / $TAGURU_API_TOKEN
done = set(client.context("sake").iter_sources())  # the completion test — import stage

ingester = TaguruIngester(
    context="sake",
    llm=ChatOpenAI(model="gpt-4.1", temperature=0),
    checkpoint_store=FilesystemCheckpointStore(".taguru-checkpoints"),
    on_event=lambda e: print(e.kind, getattr(e, "source", "")),
)

deadline = time.monotonic() + args.max_minutes * 60
failures = []
corpus = load_corpus()  # yours — a list of LangChain Document, one per section/segment
pending = [doc for doc in corpus if doc.metadata["source"] not in done]
if args.max_items:
    pending = pending[: args.max_items]   # this run's own count bound

for doc in pending:
    if stop.is_set() or time.monotonic() >= deadline:
        break
    try:
        outcome = ingester.ingest_text(doc.page_content, source=doc.metadata["source"], should_stop=stop)
        if outcome.interrupted:
            break   # chunk checkpoints already saved what completed; exit and requeue
    except Exception as exc:
        failures.append((doc.metadata["source"], exc))   # one failure never aborts the job

print(f"{len(pending) - len(failures)} succeeded, {len(failures)} failed, "
      f"{len(corpus) - len(done) - len(pending)} not reached this window")
for source, exc in failures:
    print(f"  FAILED {source}: {exc}")

Everything durable here is Taguru's: checkpoint_store persists each accepted chunk before the next one starts, and a segment that ultimately fails keeps its checkpoint for the next attempt (see the SDK's own checkpoint/resume section). Everything bounding the window — --max-minutes, --max-items, the signal handler, the per-segment try/except, the failure summary — is this sample's own policy, freely rewritten to fit a different scheduler or retry budget.

Progress you can watch: attempts, tokens, retries

Both producers expose the same shape of per-attempt visibility, deliberately mirrored field for field (see the diagnostics reference for the full schema). The CLI's --diagnostics-out FILE writes one JSONL record per LLM attempt, flushed incrementally — tail -f diagnostics.jsonl | jq against a running job shows each attempt's state, elapsed_seconds, and provider_metadata.finish_reason/output_tokens as it happens, including the corrective-retry attempts a single silent ingest_text() call would otherwise hide.

The SDK's on_event callback (issue #177) emits the same information as typed events: document_started, chunk_started, attempt_started, attempt_failed (with parse_error, elapsed_seconds, provider_metadata), chunk_completed (reused=True when a checkpoint satisfied it — no LLM call at all), import_started/import_completed, and the embedding-refresh events. It must stay synchronous and non-blocking; a callback exception is caught and reported via warnings.warn rather than corrupting the ingest it was observing.

Failures, torn imports, and re-submission

Extract-side failure is per-segment: the run exits 1, the failing segment's stderr line (or diagnostics record) names why, every other segment continues, and the failed segment keeps its chunk checkpoints — a rerun resumes it rather than starting over.

Import-side failure is where "torn" becomes possible in principle: validate-then-apply already rejects a malformed file before anything is written, but a fault reaching mid-source (capacity, disk) between the four durable steps (retract → passage → associations → aliases) is caught by a per-source header line, written before the first step and removed only after the last. A surviving marker is named at the server's next boot and by taguru inspect, verbatim:

sake: WARNING — the import of source 'docs/aomine.md' never completed; its truth
may be half-applied — re-import its source file or retract the source

Both repairs are exact, because retract-then-apply already makes them so: re-import the original (or corrected) source file, or retract the source outright and leave it absent. Over HTTP, POST /import reports the same fault as 409 (conflict) or 507 (capacity) with a message naming what landed — the retraction still makes the corrected retry exact. The one operational rule that keeps this from happening under a concurrent runner: never race two imports of the same source — one writer at a time per source, enforced by the runner, not by Taguru.

When sources change or disappear

A segment edited between runs is just a manifest mismatch: its content hash no longer matches the recorded compute inputs, so it re-extracts. The resulting source file's retract-then-apply then replaces the old facts wholesale on import — no separate "diff" step exists or is needed.

A segment removed from the corpus is different: nothing in Taguru garbage-collects a segment that stops being extracted. The manifest only ever grows entries; it has no notion of "no longer part of this corpus." Detecting and retiring stale membership is the runner's job — diff the corpus's current file list (or section list) against iter_sources()/GET /contexts/{id}/sources, and for anything present in Taguru but absent from the corpus, emit a header-only pure retraction source file (a "type": "source" header with no operation lines) naming that source — see "one file = one source's complete truth".

Where each guarantee is specified

This page is a composition, not a new contract — the normative source for each piece:

• Extract manifest — extract.html#manifest; chunk checkpoints and cooperative stop — extract.html#checkpoints; diagnostics sidecar — extract.html#diagnostics; replaying a recorded run and --resume-from — extract.html#replay.
• One-source-one-file idempotency — import.html#one-file-one-source; validate-then-apply and the header line — import.html#validate; HTTP status semantics — import.html#http.
• LangChain SDK progress events (issue #177) and checkpoint/resume (issue #179) — Python SDK README · TypeScript SDK README.
• Atomic per-source import (the fix that makes source listing a sound completion test) — issue #187.