Reliability

Reliability

Production reliability splits across the two pillars. Per-call behaviour — retries, fallback, escalation — lives on the Worker as a decorator. Whole-run budgets — page concurrency, error policy, cost caps, telemetry — live on the Orchestrator’s .options(...). Timeouts straddle: the per-call timeout is a Worker kwarg; the per-run cancel is an Orchestrator option.

The split is deliberate. A retry knows about one HTTP call; a cost cap knows about the whole run. Putting them on different pillars keeps each surface narrow.

Worker-side reliability

Every decorator returns a new Worker. The chain reads bottom-up: innermost runs first.

from scriva import Worker

worker = (
    Worker.openai("gpt-4o", timeout=30.0)        # innermost: per-call timeout
    .retry(times=3, backoff="expo")              # then: retry transient errors
    .fallback(Worker.tesseract())                # then: fallback Worker
    .escalate_to(                                # outermost: send unsure ones to a stronger model
        Worker.openai("gpt-4o-32k"),
        when=lambda r: (r.confidence or 0) < 0.7,
    )
)

Timeouts (timeout= on the engine factory)

Each engine factory accepts timeout= in seconds (default 30.0). The Worker wraps the engine call in asyncio.wait_for; on timeout the inner task is cancelled cooperatively, and the wrapper raises TimeoutError. Plays well with .retry(...) — a timed-out attempt is retryable.

worker = Worker.openai("gpt-4o", timeout=60.0)

.retry(times=, on=, backoff=)

Retry on transient errors.

worker = (
    Worker.openai("gpt-4o")
    .retry(
        times=3,
        on=(TransientError, RateLimitError, TimeoutError),
        backoff="expo",
    )
)

Kwarg	Type	Default	Meaning
times	int	3	Maximum attempts (including the first).
on	tuple[type[Exception], ...]	(TransientError, RateLimitError)	Exception classes that trigger a retry.
backoff	"expo" | "linear" | Callable	"expo"	Delay schedule. Callable receives the attempt index.
jitter	bool	True	Add random jitter to each delay.

The default catches TransientError (5xx, connection reset) and RateLimitError (429). It does not catch ConfigurationError or RecognitionError — those are deterministic, retrying won’t help.

.fallback(other)

Use other if this Worker raises after retries are exhausted. Chains: .fallback(a).fallback(b) tries self → a → b.

worker = Worker.openai("gpt-4o").fallback(Worker.tesseract())

For an explicit “try cheap first” ladder, prefer Worker.cascade(a, b, c, ...) — same behaviour, flatter chain. See Worker › Cascade.

.escalate_to(stronger, when=...)

Run the Worker, then re-run stronger when when(recognition) is true. The cheap shape for “model the easy crops cheaply, oracle the hard ones.”

worker = (
    Worker.openai("gpt-4o-mini")
    .score(method="rendering")
    .escalate_to(
        Worker.openai("gpt-4o"),
        when=lambda r: (r.confidence or 0) < 0.7,
    )
)

when runs on the primary Worker’s result. The oracle’s call is counted separately in cost accounting.

Per-Worker concurrency

Each Worker owns its concurrency budget; the Orchestrator enforces it with a semaphore.

worker = Worker.openai("gpt-4o").parallel(max=16)        # 16 concurrent calls

Default is 8. When .recognize.by_kind({...}) dispatches across multiple Workers, each Worker has its own semaphore — slow Workers don’t starve fast ones.

Orchestrator-side reliability

Whole-run knobs live on .options(...).

recipe = recipe.options(
    page_concurrency=4,
    error_policy="continue",
    default_language="ja",
    event_buffer=1024,
    cancel_on_signal=False,
)

Option	Type	Default	Meaning
page_concurrency	int	1	Parallel pages within one document run.
error_policy	str	"continue"	"continue", "page", or "abort".
default_language	str	None	ISO 639-1; passed to language-aware steps and Workers.
event_buffer	int	1024	Max queued events before back-pressure.
cancel_on_signal	bool	False	Install SIGINT handler tied to recipe.cancel().

Error policies

• "continue" — log the error, attach an empty Recognition with error set, keep going. Default; right for batch runs.
• "page" — abort the current page, continue with the next. Right for multi-page PDFs where one bad page should not poison the rest.
• "abort" — raise and stop. Right for CLI / one-shot.

Rate limits

Rate limits are a Worker concern in spirit (one engine, one bucket) but the Orchestrator exposes a per-run cap that aggregates across Workers:

recipe = recipe.options(
    rate_limit={"openai": RateLimit(rpm=600, tpm=200_000)},
)

The mapping key matches worker.engine.name. For per-tenant fairness (key=lambda page, region: page.attrs["tenant_id"]) the limit must live on the Worker — wrap with Worker.custom(...) or use scriva.reliability.with_rate_limit and pass the result as the recognise step.

Exactly one of rps/rpm/tpm must be set per bucket. The Worker blocks (with cancellation support) on starvation — it does not raise. Set on_starve="raise" for hard fail-fast.

Cost caps

Cost caps are a whole-run budget. They live on .options(...) and accumulate across every Worker call in the recipe:

recipe = recipe.options(
    cost_cap=CostCap(max_usd=5.0, on_exceed="raise"),
)

Kwarg	Type	Default	Meaning
max_usd	float	—	Hard budget for the run.
cost_fn	Callable[[Recognition], float] | None	None	Override; default reads scriva.pricing.
on_exceed	"raise" | "skip" | "warn"	"raise"	What to do when the cap trips.
accountant	CostAccountant | None	None	Share a budget across many recipes (e.g. a batch).

The default cost_fn reads per-model price tables in scriva.pricing and accumulates (input_tokens + output_tokens) × $/token. Override for engines whose pricing scriva doesn’t ship — or for non-USD billing.

on_exceed="raise" raises BudgetExceeded; "skip" returns an empty recognition for every subsequent region so the run completes cheaply; "warn" just logs and keeps going.

Inspect totals after the run:

result = recipe("scan.png")
result.accountant.total_usd          # -> 0.34
result.accountant.by_stage           # -> {"recognize": 0.31, "classify": 0.03}
result.accountant.by_model           # -> {"gpt-4o": 0.34}

Cancellation

recipe.cancel() is cooperative; it sets an asyncio.Event the Orchestrator and every standard Worker check at yield points.

import asyncio
task = asyncio.create_task(recipe.aio("scan.pdf"))
...
recipe.cancel()
await task          # raises CancelledError once steps yield

A long-running custom Worker that ignores the cancel flag will simply run to completion — it will not deadlock, but it will not honour the cancel either. Wrap any blocking subprocess or HTTP call in asyncio.wait_for(..., timeout=...) if you cannot insert your own checkpoint.

PII redaction

PII redaction runs before the Worker’s engine sees the crop. Wrap the Worker with scriva.reliability.with_redaction:

from scriva.reliability import with_redaction, RedactionPolicy

worker = with_redaction(
    Worker.openai("gpt-4o"),
    policy=RedactionPolicy(
        patterns=[
            r"\b\d{3}-\d{2}-\d{4}\b",         # US SSN
            r"\b\d{16}\b",                     # bare card numbers
        ],
        replace_with="<REDACTED>",
        scan_text=True,                       # mask hint text
        scan_image=True,                      # find and box-blur in the crop
    ),
    audit_log=Path("audit/redactions.jsonl"),
)

Policy	Covers
RedactionPolicy.standard()	SSNs, payment card numbers (Luhn), phone numbers, emails, IBANs.
RedactionPolicy.medical()	standard() + DOBs, MRNs, NHI numbers.
RedactionPolicy.identity()	standard() + passport-number formats per country.

The audit log is JSONL with one line per redaction:

{"ts": "2026-05-21T10:14:01Z", "stage": "recognize", "region_id": "p0-r14", "pattern": "ssn", "bbox": [120, 88, 240, 32]}

with_redaction returns a Worker, so it chains like any other decorator:

worker = (
    with_redaction(Worker.openai("gpt-4o"), policy=RedactionPolicy.medical())
    .cache(".scriva_cache")
    .retry(times=3)
)

Telemetry

Telemetry is a whole-run concern. Apply it on the Orchestrator and every step inherits it:

from scriva.reliability import sinks

recipe = recipe.options(telemetry=sinks.otel(endpoint="https://otel.internal:4317"))

sinks catalogue:

• sinks.otel(endpoint=...) — OpenTelemetry traces + metrics.
• sinks.datadog(api_key=...) — Datadog metrics.
• sinks.prometheus(port=9000) — Prometheus exposition endpoint.
• sinks.statsd(host, port) — StatsD UDP.
• sinks.json(path) — newline-delimited JSON to a file.
• Callable — any (Event) -> None works.

Metrics emitted: scriva.stage.duration_ms, scriva.stage.errors, scriva.recognize.tokens_in, scriva.recognize.tokens_out, scriva.recognize.cache_hits, scriva.cost.usd. Spans wrap each step with step.name and page.index attributes.

The event payloads themselves are documented in architecture.md › Observability — the telemetry sinks are thin adapters over that stream.

Idempotency

For batch ingestion, idempotency is at the boundary, not in the recipe. scriva.batch(...) accepts:

scriva.batch(
    sources,
    recipe=recipe,
    resume_file=".scriva_batch.json",    # ledger keyed by hash(source) + recipe id
    idempotency_key=lambda src: hash_file(src),
)

Re-running with the same resume_file skips already-completed sources. See scriva.batch.

Error hierarchy

ScrivaError
├── ConfigurationError    # wiring / capability mismatch — fix code, don't retry
├── EngineError           # external engine returned an error
│   ├── TransientError    # 5xx, connection reset — retryable
│   ├── RateLimitError    # 429 — retryable with backoff
│   └── PermanentError    # 4xx (non-429), auth — not retryable
├── RecognitionError      # Worker returned, but the result fails validation
├── BudgetExceeded        # cost cap hit
├── TimeoutError          # subclass of asyncio.TimeoutError
└── ResultMergeError      # incompatible result graphs

.retry(...) retries TransientError and RateLimitError by default; override with on=.

A production-shaped Worker

The shape you’ll write twice — once for OpenAI, once for Anthropic — then factor into your application:

from scriva import Worker
from scriva.reliability import with_redaction, RedactionPolicy

def safe(engine: Worker) -> Worker:
    return (
        with_redaction(engine, policy=RedactionPolicy.medical())
        .cache(".scriva_cache")
        .retry(times=4, backoff="expo")
        .parallel(max=20)
    )

worker = safe(Worker.openai("gpt-4o", timeout=30.0)).fallback(
    safe(Worker.anthropic("claude-opus-4-7", timeout=30.0)),
)

recipe = (
    Orchestrator()
    .split.grid()
    .recognize(worker)
    .reconstruct.grid()
    .export.xlsx("out.xlsx")
).options(
    page_concurrency=4,
    cost_cap=CostCap(max_usd=10.0),
    telemetry=sinks.otel(),
    error_policy="page",
)

scriva does not ship a one-size-fits-all “production Worker” — the combination depends on your model mix, tenancy model, and budget shape. The decorators above are the building blocks.

What to read next

• Architecture › Observability — full event payload table.
• Caching — the other lever for production cost control.
• Worker — the full decorator chain.
• Orchestrator — .options(...) reference.