DBOSRuntime: expose max_recovery_attempts on DBOSRuntimeConfig (#607)

## Summary

- Adds `max_recovery_attempts` to `DBOSRuntimeConfig`. When set, it is
forwarded to the `@DBOS.workflow` decorator wrapping the runtime's
control loop, capping how many times a workflow will be replayed after a
crash before being marked `MAX_RECOVERY_ATTEMPTS_EXCEEDED`.
- Only forwarded when explicitly set, so DBOS's own default (currently
100) keeps applying for unset configs.
- Motivated by OOM crash-loops: a workflow that exhausts pod memory gets
re-recovered on the new pod, OOMs again, and burns through ~100
recoveries before DBOS gives up. A user-tunable cap turns that into a
bounded failure.

.changeset/dbos-max-recovery-attempts.md

@@ -0,0 +1,5 @@
+---
+"llama-agents-dbos": minor
+---
+
+Add `max_recovery_attempts` to `DBOSRuntimeConfig`. When set, it is forwarded to the `@DBOS.workflow` decorator wrapping the runtime's control loop.

packages/llama-agents-dbos/src/llama_agents/dbos/runtime.py

@@ -205,6 +205,7 @@ class DBOSRuntimeConfig(TypedDict, total=False):
     journal_table_name: str
     pool_size: int
     pool_min_size: int
+    max_recovery_attempts: int
     _experimental_executor_lease: ExecutorLeaseConfig | None
 
 
@@ -286,6 +287,10 @@ class DBOSRuntime(Runtime):
                     is unavailable.
                 pool_min_size: Minimum size of the asyncpg pool. Defaults to
                     ``pool_size``.
+                max_recovery_attempts: Forwarded to ``@DBOS.workflow``.
+                    Caps how many times a workflow is replayed after a crash
+                    before being marked ``MAX_RECOVERY_ATTEMPTS_EXCEEDED``.
+                    Defaults to DBOS's own default when unset.
                 _experimental_executor_lease: Lease-based executor identity.
                     When set, the runtime acquires a named slot from a
                     Postgres-backed pool on launch and uses it as the DBOS
@@ -377,7 +382,11 @@ class DBOSRuntime(Runtime):
         name = workflow.workflow_name
 
         # Create DBOS-wrapped control loop with stable name
-        @DBOS.workflow(name=f"{name}.control_loop")
+        wf_kwargs: dict[str, Any] = {"name": f"{name}.control_loop"}
+        if "max_recovery_attempts" in self.config:
+            wf_kwargs["max_recovery_attempts"] = self.config["max_recovery_attempts"]
+
+        @DBOS.workflow(**wf_kwargs)
         async def _dbos_control_loop(
             init_state: BrokerState,
             start_event: StartEvent | None = None,

packages/llama-agents-dbos/tests/test_dbos_runtime.py

@@ -447,3 +447,23 @@ def test_resolve_pool_sizes_falls_back_to_constant_when_dbos_unavailable() -> No
         min_size, max_size = runtime._resolve_pool_sizes()
     assert max_size == 10
     assert min_size == 10
+
+
+def test_register_forwards_max_recovery_attempts() -> None:
+    """When set, max_recovery_attempts is forwarded to @DBOS.workflow."""
+
+    class _W(Workflow):
+        @step
+        async def go(self, ctx: Context, ev: StartEvent) -> StopEvent:
+            return StopEvent(result="ok")
+
+    runtime = DBOSRuntime(max_recovery_attempts=3)
+    captured: dict[str, Any] = {}
+
+    def _capture(**kwargs: Any) -> Any:
+        captured.update(kwargs)
+        return lambda fn: fn
+
+    with patch("llama_agents.dbos.runtime.DBOS.workflow", _capture):
+        runtime.register(_W())
+    assert captured["max_recovery_attempts"] == 3