Recipe: Session Hygiene

Goal

Keep a long-lived AgentSession honest: rewind it to an earlier turn, notice when a saved conversation is restored under different instructions or tools, find out which attachments a model never saw, count the provider requests a run really made, and make sure a message a user typed during a run is not lost.

Session Memory covers the basics of multi-turn sessions; this page covers the hygiene calls around them. The rules behind each one are in Sessions, Multimodal Content, Streaming and Hooks and Steering.

Prerequisites

• OPENAI_API_KEY for gpt-5.6-luna, the model used below. Every example works on any provider.
• The unsupported-part example uses accounts/fireworks/models/gpt-oss-120b (FIREWORKS_API_KEY), a text-only row; its on_unsupported_part="raise" half fails before any request is sent.

The snippets use top-level await; wrap them in an async def main() and asyncio.run(main()) in a script.

Rewind to an earlier turn

from doc_processing_toolkit import Agent

agent = Agent(model="gpt-5.6-luna", instructions="You are terse. Answer in one short sentence.")
session = agent.session()

await session.run("My name is Alice.")
await session.run("I work on invoices.")
await session.run("What is my name and what do I work on?")

print(session.run_boundaries)
# [{'run': 1, 'start': 0, 'end': 2}, {'run': 2, 'start': 2, 'end': 4}, {'run': 3, 'start': 4, 'end': 6}]

kept = session.truncate_to_turn(1)       # keep turn 0, drop turns 1 and 2
print(kept, session.run_boundaries)      # 2 [{'run': 1, 'start': 0, 'end': 2}]
print(session.metadata["run_count"])     # 3 — what was spent is not rewound

reply = await session.run("What do I work on? Say 'unknown' if I have not told you.")
print(reply.output)                      # unknown

truncate_to_turn(n) keeps the first n recorded turns and returns the number of messages kept. It is a pure prefix cut: nothing that survives is rewritten, so signed thinking blocks and Gemini thought signatures stay valid. If a cut would leave a tool call without its result, the cut point moves back instead of repairing messages. truncate_to(message_count) does the same by message count.

run_count, total_cost and the token totals are not rewound — the requests were made and billed. Only the run boundaries, which are message indices, move with the cut. See Editing history.

Restore after the prompt or tools changed

A saved envelope carries the conversation and a fingerprint of the system prompt and tool set — never the prompt itself. Restoring it into an Agent whose prefix differs is allowed, and reported:

import json
from doc_processing_toolkit.agent import AgentSession

envelope = json.loads(json.dumps(session.to_dict()))
print(envelope["prefix_fingerprint"])    # {'system': '<sha256>', 'tools': '<sha256>'}

same = AgentSession.from_dict(envelope, agent=agent)
print(same.prefix_changed_on_resume)     # False

changed_agent = Agent(model="gpt-5.6-luna", instructions="You are verbose.", tools=[lookup_invoice])
restored = AgentSession.from_dict(envelope, agent=changed_agent)

print(restored.prefix_changed_on_resume)             # True
print(restored.prefix_change_report["changed"])      # ['system', 'tools']
print(restored.restored_from_schema_version)         # 3

lookup_invoice is the @tool defined in the next-but-one section. prefix_change_report also carries the saved and current fingerprints. Only your application knows whether the change was deliberate, so the toolkit never refuses the restore; decide there whether to continue, start a fresh session, or warn the user. It matters most on Anthropic, where preserved thinking is bound to the prefix it was produced under. See Detecting a changed prefix on resume.

Find out which attachments were dropped

A part the model does not accept — here, an image sent to a text-only row — is dropped with a warning by default and recorded on result.dropped_parts. on_unsupported_part="raise" refuses the run before any request is sent:

import base64
from doc_processing_toolkit import Agent, UnsupportedPartError

png = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)
files = [{"data": png, "filename": "dot.png"}]
model = "accounts/fireworks/models/gpt-oss-120b"

strict = Agent(model=model, instructions="Be terse.", on_unsupported_part="raise")
try:
    await strict.run("Describe this image.", files=files)
except UnsupportedPartError as exc:
    print(exc.dropped_parts)             # the same records as below; nothing was billed

lenient = Agent(model=model, instructions="Be terse.")
result = await lenient.run("Say OK, then say whether you received an image.", files=files)
print(result.output)                     # OK. I did not receive an image.
print(result.dropped_parts)
# [{'kind': 'base64-image/png', 'index': 1,
#   'reason': "input type 'base64-image/png' is not in the model's input_types ['text']",
#   'provider': 'fireworks', 'model': 'accounts/fireworks/models/gpt-oss-120b'}]

index is the part's position in the prompt, so you can name the file back to the user. UnsupportedPartError is a ValueError subclass exported from the package root, and it lists every offending part, not only the first. The policy is also a YAML key and a parameter of the agent factory's create_agent / create_agent_session.

Count real requests with request_start

Agent(emit_request_start=True) adds one request_start event immediately before every provider send — first request, tool follow-up, retry, continuation or compaction call. It is off by default:

from doc_processing_toolkit import Agent, tool
from doc_processing_toolkit.agent import HookDirective

@tool
def lookup_invoice(invoice_id: str) -> str:
    """Return the total of an invoice.

    Args:
        invoice_id: The invoice number.
    """
    return f"Invoice {invoice_id} totals $412.00"

seen = []

async def hook(event):
    if event.event_type == "request_start":
        m = event.metadata
        seen.append((m["request_index"], m["purpose"], m["iteration"]))
    if event.event_type == "tool_call_result":
        # a turn boundary follows (the tool result goes back to the model): delivered
        return HookDirective(action="inject_message", message="Also name the currency.")
    if event.event_type == "run_end":
        # the model has already answered: no boundary is left to deliver this into
        return HookDirective(action="inject_message", message="Now repeat the total in words.")
    return None

agent = Agent(
    model="gpt-5.6-luna",
    instructions="Use lookup_invoice to answer. Be terse.",
    tools=[lookup_invoice],
    hooks=hook,
    emit_request_start=True,
)
session = agent.session()
result = await session.run("What is the total of invoice 7?")

print(seen)                              # [(1, 'generation', 1), (2, 'tool_followup', 1)]
print(len(result.request_cost_records))  # 2 — one event per billed request on a clean run

request_index is 1-based and strictly increasing across the run — correlate on it. purpose is one of agent.hooks.REQUEST_PURPOSES. The event is observation-only: a directive returned from it is ignored with a warning. Hooks run on the tool-calling path only; on a run without tools the event still reaches an Agent.stream() consumer. See request_start is opt-in.

Never lose a late message

A queued inject_message is delivered at the next turn boundary. When the run ends first, it is handed back instead of being dropped. The run above queued one message of each kind:

print(result.output)                     # $412.00 USD
print(result.injected_messages)          # ['Also name the currency.']         — the model saw it
print(result.undelivered_injections)     # ['Now repeat the total in words.']  — it did not

if result.undelivered_injections:
    follow_up = await session.run("\n\n".join(result.undelivered_injections))
    print(follow_up.output)              # Four hundred twelve US dollars.

The two lists are disjoint, so sending the undelivered text as the next turn never counts it twice. In a real application the late message is one a user typed while the model was writing its final answer — a window that is often tens of seconds long. If the run raised instead, the same list is on the exception as exc.undelivered_injections.

To have the session send it for you, create it with agent.session(late_injection_policy="new_turn", max_follow_up_turns=1); each follow-up is still an ordinary turn with its own result and cost records, collected in session.last_turn_results. See Late hook messages.

What to validate

• after truncate_to_turn(n), len(session.run_boundaries) == n and metadata["run_count"] is unchanged
• the model no longer knows facts from the dropped turns
• from_dict() into an unchanged Agent reports prefix_changed_on_resume is False; into a changed one it names what changed
• dropped_parts is populated under both policies, and the "raise" policy bills nothing
• the number of request_start events equals len(result.request_cost_records) on a clean run
• no text appears in both injected_messages and undelivered_injections

Test sources

• tests/unit/test_session_truncation.py
• tests/unit/test_session_envelope_version.py
• tests/unit/test_unsupported_parts_policy.py
• tests/unit/test_request_start_events.py
• tests/unit/test_hooks.py, tests/unit/test_late_injection_policy.py

Related pages

• Sessions — history, run boundaries, envelopes and the prefix fingerprint
• Hooks and Steering — the three outcomes of an injected message
• Session Memory — history modes and compaction in a session