# 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

```{testcode}
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])
```

```{testoutput}
['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

```{testcode}
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"])
```

```{testoutput}
['ok']
quarantined
```

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

## Demote old conversation turns explicitly

```{testcode}
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()])
```

```{testoutput}
[('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.

```python
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 {doc}`overflow`.