cadrumo.application.ledger.batch_ingest module

Per-item truth for a bounded batch ingestion run.

A batch is a bounded run over a known set of documents, not a queue and not a background job. What makes it safe to point at a folder is that one bad document cannot end the run: every item finishes in a typed row of its own — ingested, a no-op, refused with its reason, or held for review — and the run reports the list plus a summary. The exit status reflects “any item failed”, never “the first item failed”.

That distinction is not theoretical here. The adjacent statement-import folder path runs its per-file imports in a bare comprehension with no per-item guard, so the first file that raises discards every result already produced. This module exists so the evidence batch cannot acquire the same shape.

Two properties make a re-run safe rather than merely tolerable:

Ordering is by content address, so two runs over the same documents report in the same order no matter how the filesystem enumerated them. A report whose row order depended on directory iteration could not be diffed against the previous run, which is the first thing an operator does after a partial batch.

Identity is content address plus declared direction. The same bytes filed as issued and as received are genuinely two records — one is a sale, the other a purchase — so the direction has to be in the key. The address alone would collapse them; a clock or a filename would make a re-run mint duplicates. Because the key is derived rather than stored, resume after a crash is re-run: there is no journal format to invent, and no progress file to leave on disk.

See also

InvoiceKind

The declared direction that, with the content address, forms identity.

Attachment

Where the content address comes from: the SHA-256 of the stored bytes.

BatchItemStatus

How one batch item ended.

no_op is a success: the item was already ingested under this identity and was neither re-read nor re-written. pending_review is neither a success nor a failure — the item produced a draft an operator must adjudicate — so it does not fail the run while still being visible in the summary.

paused is the one status that means the work did NOT happen: the document needs a reading model and the machine could not admit one. It is not a failure of the document and carries no per-item refusal, because every paused item in a run shares one cause; that cause is stated once on the run.

alias of Literal[‘ingested’, ‘no_op’, ‘refused’, ‘pending_review’, ‘paused’]

BATCH_ITEM_STATUSES: frozenset[Literal['ingested', 'no_op', 'refused', 'pending_review', 'paused']]

Every status a batch item can end in. Derived from the type rather than restated, so a status added to the alias cannot go unreported in the summary. Typed as the alias rather than as bare str so iterating it yields the literal the consumers expect; each raw get_args member is revalidated through the alias’s own adapter rather than trusted as already-narrowed, since get_args itself only returns tuple[Any, ...].

FAILING_BATCH_ITEM_STATUSES: frozenset[Literal['ingested', 'no_op', 'refused', 'pending_review', 'paused']]

an item awaiting review has not failed, and a no-op is the idempotent success.

Type:

The statuses that make a run “any item failed”. Deliberately narrow

COMPLETED_BATCH_ITEM_STATUSES: frozenset[Literal['ingested', 'no_op', 'refused', 'pending_review', 'paused']]

The statuses in which the document’s work actually happened. Derived by SUBTRACTION from the full set rather than listed, so a status added to the alias must be classified as failing or paused to stay out of it – a new success-shaped status is counted as progress by construction rather than silently omitted from the deterministic-progress figure.

class BatchItemResult(**data)

Bases: BaseModel

What happened to exactly one document in the run.

Variables:

• content_address – SHA-256 of the document’s bytes, lowercase hex. The item’s position in the report and half of its identity.

• identity – Content address plus declared direction. Two runs over the same document with the same direction derive the same value, which is what makes a re-run a no-op rather than a duplicate.

• direction – The direction the operator declared for the run.

• source_name – The document’s own name, carried for the operator’s benefit only — never for identity, because a rename must not mint a second record.

• status – How the item ended.

• refusal_code – Machine-readable reason, present exactly when the item was refused.

• refusal_verdict – The application-owned failed condition and its machine facts, present exactly when the item was refused. Carries what was seen rather than only that something failed, without retaining the producer’s exception text.

• needed_inference – Whether this document required the inference lane at all. Carried per item because the run-level counts cannot be reconstructed afterwards: a completed row looks identical whether it was read deterministically or through a model, so a pooled “N completed” cannot say whether pacing worked or whether nothing was ever paced. Defaults to True, the conservative direction – a row built without stating it is not counted as deterministic progress.

Parameters:

data (Any)

content_address: ContentDigest

identity: str

direction: InvoiceKind

source_name: str

status: BatchItemStatus

refusal_code: str | None

refusal_verdict: PreconditionVerdict | None

needed_inference: bool

class UnresolvedBatchSource(**data)

Bases: BaseModel

A submitted source whose bytes could not be read at all.

Deliberately not a BatchItemResult. An item’s identity IS its content address, and a file that cannot be read has no content address to derive one from — giving it a placeholder would put a value into the field that ordering and idempotency both key on, and a second unreadable file would then collide with the first. Reporting it in its own channel keeps the item rows identity-bearing while still putting the failure in front of the operator, which dropping it silently would not.

Variables:

• source_name – The source as the operator named it.

• refusal_code – Machine-readable reason the bytes could not be read.

• refusal_verdict – The application-owned failed condition and its machine facts for that same reason.

Parameters:

data (Any)

source_name: str

refusal_code: str

refusal_verdict: PreconditionVerdict

class InferencePause(**data)

Bases: BaseModel

The typed refusal that closed the run’s inference lane.

One record rather than a refusal stamped on every affected document. Every paused item in a run shares one cause — the machine could not admit a model — so N copies of it would be N identical messages an operator learns to scroll past, which is how a real refusal goes unread.

facts and precondition_verdict are the application-owned provisioning outcome. This record deliberately does not restate a reason, rendered explanation, or recovery command: its consumer must resolve the exact verdict against the live operator surface.

Parameters:

data (Any)

facts: Mapping[str, str | int | bool]

precondition_verdict: PreconditionVerdict

class BatchRunResult(**data)

Bases: BaseModel

The whole run: every item’s row, plus what they add up to.

Variables:

• items – One row per document, ordered by content address.

• unresolved – Sources whose bytes could not be read, so they never became items. Counted as failures.

• inference_pause – Present exactly when the inference lane closed.

• any_failed – Whether any item was refused. The exit status reads this, so a run that refused its first document and ingested the rest is a failed run that nonetheless did its work.

• any_deferred – Whether work remains that this run did not attempt.

Parameters:

data (Any)

items: tuple[BatchItemResult, ...]

unresolved: tuple[UnresolvedBatchSource, ...]

inference_pause: InferencePause | None

property any_failed: bool

Return whether any item was refused or any source was unreadable.

property any_deferred: bool

Return whether the run left work it did not attempt.

Kept off any_failed rather than folded into it. A paused item is not a failure — nothing went wrong with the document and a re-run costs nothing, because completed items are no-ops. But it is not success either, and a caller automating this needs to tell “everything is done” from “the deterministic half is done”, which one boolean cannot say.

count_of(status)

Return how many items ended in status.

Return type:

int

Parameters:

status (Literal['ingested', 'no_op', 'refused', 'pending_review', 'paused'])

property deterministic_completed: int

Return how many items completed WITHOUT needing the inference lane.

Reported separately from the paced count because together they are the evidence that pacing worked: deterministic progress continuing while inference-bearing work parks is the whole property. Pooled into one “completed” figure, a run that paced everything and a run that paced nothing read identically.

“Completed” here is every ending that DID the work – newly ingested, held for review, or already present from an earlier run. A held item was read; it is the finding that needs attention, not the reading.

property paced: int

Return how many items were parked because the inference lane closed.

A zero here is only meaningful beside a non-zero elsewhere: a count that has never been shown able to rise is not a measurement.

property summary: dict[str, int]

Return the per-status tally, every status present even at zero.

A status missing from the summary reads as “not applicable” rather than “none occurred”, and the two are different things to an operator looking for the refusals.

batch_item_identity(*, content_address, direction)

Return the identity of one batch item.

Derived from the document’s bytes and the direction declared for it, and from nothing else. No clock, no filename, no run identifier: each of those would differ between two runs over the same document and turn an idempotent re-run into a duplicate write.

The direction is part of the key rather than an attribute beside it because the same bytes filed as issued and as received are two genuinely different records — a sale and a purchase — and collapsing them would silently drop one.

Parameters:

• content_address (str) – Lowercase hex SHA-256 of the document’s bytes.

• direction (InvoiceKind) – The direction declared for this run.

Return type:

str

Returns:

The identity string.

order_batch_items(items)

Return items ordered by content address.

Deterministic by construction, so two runs over the same documents report in the same order regardless of how the filesystem enumerated them. Direction breaks a tie, which can only arise when the same bytes were submitted under both directions in one run.

Return type:

tuple[BatchItemResult, ...]

Parameters:

items (Iterable[BatchItemResult])

order_batch_sources(sources)

Return (content_address, source_name) pairs ordered by content address.

The same ordering rule applied before any work happens, so the run PROCESSES in report order too. Ordering only the finished rows would still leave the processing order — and therefore which items a interrupted run completed — dependent on directory enumeration.

Return type:

tuple[tuple[str, str], ...]

Parameters:

sources (Iterable[tuple[str, str]])

summarise_batch(items, unresolved=(), inference_pause=None)

Return the run result for items, ordered by content address.

Return type:

BatchRunResult

Parameters:

• items (Sequence[BatchItemResult])

• unresolved (Sequence[UnresolvedBatchSource])

• inference_pause (InferencePause | None)

BATCH_DRAFT_EXTRACTOR

Names the function that produced a batch draft, not the reader that read it. Which reader ran is a PER-FIELD fact and is already carried by the draft’s own provenance envelopes (FieldOrigin); a document-level extractor label claiming one reader for a draft assembled from several would be the laundering those envelopes exist to prevent.

BATCH_READ_TRANSPORTS: Final[tuple[str, ...]]

Transports that carried a batch reading, recorded beside the draft. On-host is not assumed here, it is DERIVED from an enforced refusal: an evidence-derived request may only reach an off-host provider carrying a per-invocation consent token, the dispatch refuses without one, and this path mints none and accepts none. So no batch read can have left the host, and recording that is an observation about the gate rather than a guess about configuration.

The moment this path gains a consent token, this constant stops being the right answer and must become a function of that token – which is why it is named for what it asserts rather than inlined at the write site.

run_evidence_batch(*, bucket_id, sources, direction, evidence_ports, extraction_ports, operation, legends, settings=None, on_item=None, profile=None)

Run the ingestion pipeline over every source, one typed row each.

No item can end the run. Every per-item failure is caught and becomes that item’s refusal row, because the alternative — the shape the adjacent statement-import folder path still has — discards every result already produced when the first document raises.

A re-run is the resume. Each item’s identity is its content address plus the declared direction, handed to the evidence store’s idempotency guard, so a second run over the same folder re-attaches nothing and emits no second lifecycle event. An item counts as complete only when its evidence AND its draft are both present: evidence-without-draft is exactly the state an interrupted or environment-refused first run leaves behind, and treating it as done would strand that document permanently, never retried and reported as an untroubled no-op.

Nothing is written outside secure storage. The two writes are the encrypted evidence record and the encrypted draft. There is no spool, no journal and no progress file — which is what makes “resume is re-run” a property rather than a slogan, since there is no local state to go stale (sensitive-financial-data-secure-storage-only).

Parameters:

• bucket_id (str) – Ledger bucket every record is written into.

• sources (Iterable[Path | str]) – Files and directories to ingest.

• direction (InvoiceKind) – The direction declared for the whole run. Part of each item’s identity, so the same document filed both ways is two records rather than one.

• evidence_ports (LedgerEvidencePorts) – Required evidence catalogue, attachment, and event capabilities for this bucket.

• extraction_ports (InvoiceDraftExtractionPorts) – Required invoice-draft extraction capabilities, including the application-owned shape probe, structured reader, and text-layer reader.

• operation (PinnedAuthorityOperation) – The caller-owned pinned authority operation used for all governed fact and catalogue resolution during this run. The batch never opens or replaces the operation.

• legends (tuple[RegimeLegend, ...]) – Regime legends resolved by the same pinned operation and date coordinate used by the extraction caller.

• settings (Settings | None) – Resolved Settings; load_settings() when omitted, so override_settings() is honoured.

• on_item (Callable[[BatchItemResult], None] | None) – Called with each row as it completes, for progress reporting. A raising callback must not lose the run, so it is guarded like any other per-item failure.

• profile (HardwareProfile | None) – Measured hardware the admission check judges against; probed when omitted. The same injection point the admission primitive itself exposes, so a contended machine can be exercised from real measurements rather than by substituting the decision.

Returns:

Every row, ordered by content address.

Return type:

BatchRunResult