Recipes

These recipes use only behavior implemented in the 0.1 series. Every record has a stable ID, timestamp, and source so its manifest is reproducible.

Select one retrieved document under a budget

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "refund", "payload": "refund policy is 30 days", "sources": ["db://kb/refund"],
     "created_at": "2026-01-01T00:00:00Z"},
    {"id": "office", "payload": "office address is Tokyo", "sources": ["db://kb/office"],
     "created_at": "2026-01-01T00:01:00Z"},
])

pack = frame.rank("refund policy").pack(
    budget=8,
    token_profile=cx.profiles.approximate(),
    boundary=cx.Boundary.external("provider"),
)
print([item.id for item in pack.items])
print([(item["id"], item["reason"]) for item in pack.manifest.excluded])

['refund']
[('office', 'budget')]

The built-in ranker is lexical. For embedding or model-based scores, supply a deterministic batched Ranker; Cognoxium does not make the network request.

Quarantine failed tool results

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "ok", "payload": "tests passed", "role": "tool", "sources": ["tool://ci/1"],
     "metadata": {"exit_code": 0}, "created_at": "2026-01-01T00:00:00Z"},
    {"id": "failed", "payload": "unverified output", "role": "tool",
     "sources": ["tool://ci/2"], "metadata": {"exit_code": 1},
     "created_at": "2026-01-01T00:01:00Z"},
])

pack = frame.quarantine(
    lambda row: row["metadata"].get("exit_code") != 0
).pack(budget=100, boundary=cx.Boundary.external("provider"))
print([item.id for item in pack.items])
print(pack.manifest.excluded[0]["reason"])

['ok']
quarantined

Quarantine records an application decision; Cognoxium does not determine whether a tool result is correct.

Demote old conversation turns explicitly

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "policy", "payload": "Follow policy", "sources": ["app://policy"],
     "retention": "required", "trust": "trusted", "priority": 10,
     "created_at": "2026-01-01T00:00:00Z"},
    {"id": "old-turn", "payload": "Earlier discussion", "sources": ["chat://turn/1"],
     "retention": "required", "created_at": "2026-01-01T00:01:00Z"},
])

authorized = frame.demote(
    cx.col("id") == "old-turn",
    to=cx.Retention.OPTIONAL,
)
print([(item.id, item.retention.value) for item in authorized.collect()])

[('policy', 'required'), ('old-turn', 'optional')]

The application performs the authorization decision before calling demote(); packing never demotes required content automatically.

Persist an auditable frame

Use write_jsonl() for reviewable fixtures, write_ipc() for fast local interchange, and write_parquet() for analytical storage. Arrow formats require the arrow extra.

frame.write_jsonl("context.jsonl")
restored = cx.CognitionFrame.read_jsonl("context.jsonl")
assert restored.collect() == frame.collect()

Persist pack.manifest.to_dict() separately when audit requirements permit. A manifest can itself contain sensitive identifiers, source URIs, and conflict values.

Recover from budget overflow

Catch BudgetExceeded, log its structured fields, and make an application decision. Never branch on translated exception text. See the executable recovery in Overflow recovery.