Add support for durable execution with DBOS

[qianl15]

Summary

Closes #2629

This PR integrates DBOS with Agent.run and related methods to provide out-of-the-box durable execution and checkpointing.

Changes

• Workflows:

  • Agent.run and Agent.run_sync are now decorated as DBOS workflows.
  • Users no longer need to manually wrap these functions in DBOS workflows.

• Steps:

  • Model requests and MCP calls (get_tools, call_tool) are decorated as DBOS steps.
  • Outside DBOS, these remain ordinary functions.
  • Inside DBOS workflows, outputs are automatically checkpointed in Postgres.

• Tooling:

  • Users may pass in DBOS-decorated functions (workflows, steps, transactions) as tools.
  • The integration does not automatically wrap tools in DBOS decorators, so users retain full control. For example, they can pass in a workflow which will be invoked as a child workflow; or they can pass in a step.
  • Tools behave as normal functions when not used with DBOS.

TODO

• Add documentation
• Update to V1 API.
• Make human-in-the-loop work with DBOSAgent.

[qianl15]

@qianl15 Thanks Qian, we're getting there :) If you could port the HITL tool approval test we have for Temporal, as you suggested in the description, that'd be great!

Thanks @DouweM for the review! I addressed all your comments and added the HITL tests (both async and sync work as expected).

[DouweM]

@qianl15 Thanks Qian, nothing more to add :) We're going to launch v1 out of beta tomorrow, and this feels a bit too big to add during the beta stage, so I'll keep the PR open for a few more days to merge it for v1.0.1/v1.1.

[qianl15]

@qianl15 Thanks Qian, nothing more to add :) We're going to launch v1 out of beta tomorrow, and this feels a bit too big to add during the beta stage, so I'll keep the PR open for a few more days to merge it for v1.0.1/v1.1.

@DouweM Sounds good! Thanks for the update. All comments have been addressed on my end. Looking forward to the v1 launch 🚀

[DouweM]

The goal of this change to TemporalAgent was to make sure that the event_stream_handler is only every called from inside an activity, where non-determinism/IO is available. I believe that to get the same behavior here, we'd have to create a new step, call that from here, and then inside the step call the self.event_stream_handler

[qianl15]

Thanks for raising this! I'm open to discussion here.

The difference is intentional: in DBOSAgent, tools and event stream handlers are not automatically wrapped. Instead, users choose whether to decorate them with @DBOS.step or invoke DBOS workflows/queues inside them. This provides flexibility:

• Use @DBOS.step if the function requires non-determinism or I/O.
• Skip the decorator if durability isn't needed, so you avoid the extra DB checkpoint write.
• If the function needs to enqueue tasks, it should run inside the agent's main workflow (not in a step).

This lets you control the right level of durability for your application.

Note: If you call a @DBOS.step inside another step or outside a workflow, it just runs as a normal function (no extra wrapping).

Here's the updated test case showing the intended usage:

# Wrap event_stream_handler as a DBOS step since it uses logfire (non-deterministic)
@DBOS.step()
async def event_stream_handler(
    ctx: RunContext[Deps],
    stream: AsyncIterable[AgentStreamEvent],
):
    logfire.info(f'{ctx.run_step=}')
    async for event in stream:
        logfire.info('event', event=event)

And the docs now read:

Custom tool functions and event stream handlers are **not automatically wrapped** by DBOS.  
If they involve non-deterministic behavior or perform I/O, explicitly decorate them with `@DBOS.step`.  

[DouweM]

@qianl15 Ah right, I sort of knew that :) So wrapping the event_stream_handler in a @DBOS.step works as expected, even though it's being passed an iterable that items are yielded into asynchronously? That's great.

Then all that's left before I merge are the 2 docs comments below.

[qianl15]

Yeah! DBOS steps don't require their input parameters to be serializable (that constraint only applies to DBOS workflows for recovery). So you can safely pass in an async iterable to a @DBOS.step, and it'll work as expected.

[DouweM]

As it says above, "if you add optional groups, please update docs/install.md" :)

[qianl15]

Updated docs/install.md!