Stage reference

Every stage: (records, **params) → records over plain list[dict], deterministic, non-mutating, auto-traced when a trace() is active. Contract, not implementation — inputs, outputs, and the invariants held.

select(records, fields)

Keeps only fields on each record, in keep-list order. Missing fields are simply absent (no error, no None padding). Lossless: the dropped tokens were fields you were never going to use. Also the stage that makes rows uniform — which downstream table encodings reward. Everywhere a stage takes fields, a bare string names one field (fields="snippet" ≡ fields=["snippet"]) — it is never iterated as characters.

dedupe(records, key=None)

Drops duplicates, first occurrence wins, order preserved. key=None compares whole records (key-order insensitive); a field name or list of names compares those fields. Fingerprints are type-qualified — 1, 1.0, True, "1" are four different values — with a canonical-JSON fallback for unhashable values.

allowlist(records, field, allowed)

Provenance gate. Keeps only records whose field value is in allowed; records missing the field, or carrying an unhashable value, are dropped — fail closed. The strongest injection control a shaper can offer is refusing content from sources you did not approve: apply this first, before any relevance logic, so untrusted records never compete for the budget. Dropped records appear in the trace/audit with this stage as the reason.

quarantine(records, fields=("snippet",), patterns=None, action="drop", into="quarantined")

Injection tripwire. A deterministic, case-insensitive pattern screen over the given text fields for the crudest prompt-injection markers — "ignore all previous instructions", role resets ("you are now"), system-prompt tags. action="drop" removes matches (audited); action="flag" keeps everything and writes a boolean to into so downstream policy decides.

rescore(records, query, fields=("snippet",), into="score", proximity=True, match="word")

Writes a deterministic lexical relevance score to into on a copy of each record, so ranking stops inheriting the search tool's score quality. BM25-style, computed inside the batch:

• per-term IDF ln((N−df+0.5)/(df+0.5)+1) — rare terms weigh more;
• saturating term frequency tf/(tf+1.5) — covering all the query's terms beats repeating one;
• match="word" (default) counts exact word occurrences the way BM25/Lucene tokenize — value ≠ values, underscores split snake_case; match="substring" is the forgiving, decoy-prone alternative;
• proximity=True: consecutive query terms in order within 80 characters earn a bonus — the source a query was phrased from usually carries its terms in sequence.

rank(records, by, desc=True)

Sorts stably by a field name or key function; records missing the field go last. Run it before trim_to_budget so the budget drops the least important records, not arbitrary ones. Values of by must be mutually comparable.

truncate_field(records, field, max_tokens, tokenizer=None, suffix="…")

Clips one long text field to the longest prefix that fits the cap, then appends the suffix. For the built-in estimator the cut is closed-form (ceil(len/4) is invertible: 4×budget characters); for custom tokenizers it binary-searches. Non-string values pass through untouched. Never exceeds the cap.

trim_to_budget(records, max_tokens, tokenizer=None, min_records=0)

Keeps the greedy prefix that fits the budget: each record costs the token count of its serialized text plus 2 for list framing; the walk stops before the budget is crossed. Assumes records are already ordered by importance. If even the first record exceeds the budget the result is empty — run truncate_field first for oversized records, or set min_records: the first that-many records (the best ones, post-rank) are kept even over budget, so a too-small budget can never return an empty list that reads as "nothing matched". The overrun stays visible in the trace/audit token counts.

merge(*sources, schema=None, dedupe_key=None)

Unifies differently-shaped tool outputs into one record shape. schema maps a unified field name to a source field name or a list of candidates tried in order; dedupe_key optionally dedupes across the merged result. With schema=None the sources are concatenated unchanged. A combiner rather than a single-collection transform — the one stage that is not auto-traced.

merge(web, docs, schema={"title": ["title", "name", "headline"],
                         "url":   ["url", "link", "href"]})

pipeline · stage · shaped · trace · tokens

• pipeline([...]) — plain left-to-right function composition; any records → records callable (even a lambda) is a valid stage, so you can drop to code for anything irregular and still breakpoint inside it.
• stage(fn, **params) — binds keyword args so a stage slots into a pipeline.
• @shaped(pipe) — runs a tool's record-list return value through a pipeline before it returns; the model never sees raw output. Async tools work too: a coroutine function is awaited, then shaped.
• pipeline([...]) exposes .fingerprint — a stable policy hash of stage names and bound parameters, recorded into any active trace so an audit can say which policy version shaped a context.
• trace(id_field=...) — context manager; every stage records records-in/out and tokens-before/after, and with id_field also which records each stage dropped (records lacking the field are untracked). t.audit() emits the JSON-able governance record. Inactive overhead is a single ContextVar lookup per stage — single-digit milliseconds (49.8 vs 44.4 ms over 10k records in RESULTS.md); active tracing token-counts every stage boundary and is priced accordingly.
• tokens — count, use_tiktoken() (exact where tiktoken covers the model), tiktoken_tokenizer(), set_tokenizer(fn), set_serializer(fn), serialize(value), and scoped(...) — a context-local overlay that isolates concurrent tenants. Strings always bypass the serializer.