Support for `capture_run_messages` to store incomplete results

[psg2]

Description

When running Pydantic AI in the backend coupled with Vercel AI for the frontend but not exclusive to it, it's possible to interrupt the request stream in the frontend with a stop method, when this happens the request stream is interrupted with an exception asyncio.CancelledError but even wrapping the run with capture_run_messages the messages are still empty, unless the run fails with a model error from Pydantic AI itself apparently.

This leaves our app in a weird state because from the client's perspective it already received delta streams and is showing tool calls and text to the user, but we haven't persisted anything in the database yet. Given that we're running run_stream_events we could build the incomplete results and persist it, but I image that the package itself already has the code to manage the partial output.

It would be used to handle more exceptions gracefully to manage everything that was streamed from the provider prior to the error.

Slack thread: https://pydanticlogfire.slack.com/archives/C083V7PMHHA/p1760989459555679

References

No response

[DouweM]

@psg2 An include_incomplete flag on capture_run_messages makes sense to me. We may also want a incomplete field on the last message to make it's clear this is what happened.

So far we've discussed an interrupted streamed model response, but I think we should also include an incomplete model request if model response handling (i.e. tool call execution) was interrupted, so you can get the ToolReturnParts for tools that already did execute completely. That ModelRequest is a bit easier to construct manually as the FunctionToolResultEvent holds the full ToolReturnPart (unlike the model response events that contain deltas, not full parts), but this would still be much more convenient through capture_run_messages.

Note that an incomplete ModelResponse will usually give you a valid message history to use in a new agent run, unless the interruption happened in the middle of the model generating tool call arguments, which would then be incomplete/invalid JSON. How would you suggest we handle that?

However, an incomplete ModelRequest in the message history will typically not be a valid message history that can be used as-is in a new agent run, as the model would complain that not all tool calls from its most recent response have results in the request.
As a workaround, you can pass deferred_tool_results=DeferredToolResults(approvals={'<unexecuted_tool_call_id>': True, ...}) (see https://ai.pydantic.dev/deferred-tools/) to the new agent.run* call along with the message_history ending on the incomplete ModelRequest, which will result those unprocessed tool calls to be executed after all, while keeping the results that were already in the history (so you don't get duplicate side-effects).
We could change the default behavior when a ModelRequest with missing tool call results is passed in to automatically execute the unprocessed tool calls, so you don't have to manually construct that DeferredToolResults object.

---

Implementation wise, this could be a bit tricky as the messages list returned by capture_run_messages is currently the exact same object as state.message_history, and it would be unexpected if a flag to capture_run_messages affected that list (that's available on RunContext among other things) as well. So we may need to keep track of 2 lists, one with the completed history, and one with the incomplete message appended.

[psg2]

Note that an incomplete ModelResponse will usually give you a valid message history to use in a new agent run, unless the interruption happened in the middle of the model generating tool call arguments, which would then be incomplete/invalid JSON. How would you suggest we handle that?

Here I think it's fine to drop the latest Part with incomplete deltas as the model is likely to retry building that again in the following run given the message history.

However, an incomplete ModelRequest in the message history will typically not be a valid message history that can be used as-is in a new agent run, as the model would complain that not all tool calls from its most recent response have results in the request. As a workaround, you can pass deferred_tool_results=DeferredToolResults(approvals={'<unexecuted_tool_call_id>': True, ...}) (see ai.pydantic.dev/deferred-tools) to the new agent.run* call along with the message_history ending on the incomplete ModelRequest, which will result those unprocessed tool calls to be executed after all, while keeping the results that were already in the history (so you don't get duplicate side-effects).

I'm not confident in the correct approach here, in our case we already filter out unpaired tool call with tool return before calling the agent run whenever we fetch our history from the database.

We could change the default behavior when a ModelRequest with missing tool call results is passed in to automatically execute the unprocessed tool calls, so you don't have to manually construct that DeferredToolResults object.

This could be helpful, but not sure this should be the default behavior. It looks simple enough for clients to implement themselves and magic enough to end up in some weird scenarios.