Prototype transport-neutral durable runtime library

CHANGELOG.md

@@ -9,6 +9,7 @@
 
 ### Improvements
 - databricks-openai: `DatabricksOpenAI` and `AsyncDatabricksOpenAI` clients now follow HTTP redirects by default, configurable via the new `follow_redirects` parameter (#445)
+- databricks-ai-bridge: Add a transport-neutral `DatabricksDurableRuntime` with Lakebase request, response, event persistence, and stale-attempt recovery
 
 ### Bug Fixes
 - databricks-ai-bridge: Genie now returns the full answer text aggregated from all text attachments (#432)

README.md

@@ -24,6 +24,20 @@ For frameworks without dedicated integration packages:
 pip install databricks-ai-bridge
 ```
 
+## Durable Runtime
+
+[`DatabricksDurableRuntime`](./src/databricks_ai_bridge/durable_runtime/README.md)
+adds Lakebase-backed request, result, and event persistence with heartbeat-based
+stale-attempt recovery around a caller-owned async handler. The handler remains
+responsible for agent sessions and checkpoints.
+
+```sh
+pip install 'databricks-ai-bridge[memory]'
+```
+
+See the [transport-neutral library cookbook](./cookbooks/durable-runtime-library/README.md)
+for the developer-owned FastAPI wiring.
+
 ### Install from source
 
 With https:

cookbooks/durable-runtime-library/README.md

@@ -0,0 +1,136 @@
+# Durable Runtime Library
+
+This cookbook shows the transport-neutral library option.
+
+`openai_agent.py` contains an OpenAI Agents SDK approval flow. It uses the
+stable `session_id` for an `AsyncDatabricksSession` and emits SDK events into
+the durable event log. `agent.py` decides how recovered attempts should invoke
+the SDK by checking `context.is_recovery`.
+
+The developer-owned `server.py` must:
+
+- start and stop `DatabricksDurableRuntime`;
+- map the application's request into a stable `run_id`, `session_id`, and JSON payload;
+- expose submission and polling endpoints; and
+- convert persisted events into the application's streaming protocol.
+
+## Background, streaming, and client impact
+
+| Capability | Library contract | Developer and client contract |
+| --- | --- | --- |
+| Background | `runtime.submit()` persists and schedules work; `runtime.get()` returns stored status/result. | Developer chooses how the server returns `202`, exposes the run ID, and implements polling. |
+| Durable streaming | `context.emit()` stores ordered JSON events; `runtime.events()` reads them by cursor. | Developer defines the SSE/event format and reconnect route; the client must retain the chosen cursor. |
+
+The library does not mandate any HTTP contract. The cookbook chooses
+`POST /runs`, `GET /runs/{run_id}`, and `GET /runs/{run_id}/events?after=N` to
+make the missing server work visible.
+
+### OpenAI Agents SDK: before and after
+
+The OpenAI Agents SDK has no remote deployment client. A developer may preserve
+an existing client exactly, but must map that route to the runtime and decide
+how it expresses background and replay semantics. The cookbook instead chooses:
+
+```python
+async with http.stream(
+    "POST",
+    "/runs",
+    json={
+        "run_id": "run-1",
+        "session_id": "conversation-1",
+        "background": True,
+        "stream": True,
+        "payload": {"message": "hello"},
+    },
+) as response:
+    async for line in response.aiter_lines():
+        last_event_id = remember_sse_id(line, last_event_id)
+
+final = (await http.get("/runs/run-1")).json()
+```
+
+The server-side agent still calls `Runner.run_streamed()` and forwards each SDK
+event to `context.emit()`.
+
+### LangGraph SDK: before and after
+
+The native client starts and observes work through LangGraph's own protocol:
+
+```python
+thread = await langgraph.threads.create()
+run = await langgraph.runs.create(
+    thread["thread_id"], "agent", input={"messages": messages}
+)
+result = await langgraph.runs.join(thread["thread_id"], run["run_id"])
+
+async for event in langgraph.runs.stream(
+    thread["thread_id"], "agent", input={"messages": messages}
+):
+    consume(event)
+```
+
+The library is flexible enough for a developer to recreate those routes and
+keep `langgraph_sdk`, but the library does not provide that integration. With
+the cookbook server, the client changes to `/runs` and passes `thread_id` as
+`session_id`. This is the option with the least mandated client change and the
+most developer-owned protocol work.
+
+References: [OpenAI Agents SDK streaming](https://openai.github.io/openai-agents-python/streaming/),
+[LangGraph background runs](https://docs.langchain.com/langsmith/runs), and
+[LangGraph resumable streaming](https://docs.langchain.com/langsmith/streaming).
+
+Run it locally with:
+
+```bash
+pip install -r requirements.txt
+export OPENAI_API_KEY=...
+export LAKEBASE_AUTOSCALING_ENDPOINT=projects/.../endpoints/...
+uvicorn server:app --reload
+```
+
+## Durable HITL flow
+
+HITL is two durable runs. The first run returns `requires_action`; the second
+run carries the human decision and reuses the same SDK session.
+
+Start the proposal as a background stream:
+
+```bash
+curl -N -X POST localhost:8000/runs \
+  -H 'content-type: application/json' \
+  -d '{
+    "run_id": "proposal-1",
+    "session_id": "approval-session-1",
+    "background": true,
+    "stream": true,
+    "payload": {"action": "publish the release notes"}
+  }'
+```
+
+Poll `/runs/proposal-1`, review the persisted result, and submit approval:
+
+```bash
+curl -N -X POST localhost:8000/runs \
+  -H 'content-type: application/json' \
+  -d '{
+    "run_id": "approval-1",
+    "session_id": "approval-session-1",
+    "background": true,
+    "stream": true,
+    "payload": {
+      "action": "publish the release notes",
+      "decision": "approve",
+      "wait_seconds": 60
+    }
+  }'
+```
+
+Stop the process during the wait. On restart, the library reclaims the run and
+calls the same developer-owned executor with `context.is_recovery=True`. The
+developer is responsible for reconnecting the SDK session and choosing the
+recovery instruction. Replay from
+`/runs/approval-1/events?after=<last-event-id>` and poll `/runs/approval-1` for
+the final result.
+
+This option makes the ownership difference explicit: the library stores runs
+and events, while the developer implements all HTTP and recovery wiring.

cookbooks/durable-runtime-library/agent.py

@@ -0,0 +1,24 @@
+"""OpenAI Agents SDK adapter for the transport-neutral durable runtime."""
+
+from openai_agent import run_openai_agent
+
+from databricks_ai_bridge.durable_runtime import DurableExecutionContext, JsonObject
+
+
+async def run_agent(
+    request: JsonObject, context: DurableExecutionContext
+) -> JsonObject:
+    payload = request["payload"]
+    session_id = str(request["session_id"])
+    result = await run_openai_agent(
+        payload=payload,
+        session_id=session_id,
+        emit=context.emit,
+        is_recovery=context.is_recovery,
+    )
+
+    return {
+        "result": result,
+        "session_id": session_id,
+        "attempt": context.attempt,
+    }

cookbooks/durable-runtime-library/app.yaml

@@ -0,0 +1,3 @@
+command:
+  - python
+  - server.py

cookbooks/durable-runtime-library/openai_agent.py

@@ -0,0 +1,85 @@
+"""OpenAI Agents SDK loop for durable streaming and human approval."""
+
+import asyncio
+import os
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from agents import Agent, Runner, function_tool
+from databricks_openai.agents import AsyncDatabricksSession
+
+EventEmitter = Callable[[dict[str, Any]], Awaitable[int]]
+
+
+@function_tool
+async def complete_approved_action(action: str, wait_seconds: int) -> str:
+    """Simulate an approved side effect after a bounded delay."""
+    if wait_seconds < 0 or wait_seconds > 300:
+        raise ValueError("wait_seconds must be between 0 and 300")
+    await asyncio.sleep(wait_seconds)
+    return f"Completed approved action: {action}"
+
+
+def _create_agent() -> Agent:
+    return Agent(
+        name="Durable approval assistant",
+        model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"),
+        instructions=(
+            "For a PROPOSAL request, produce a short plan and do not call tools. "
+            "For an APPROVED request, always call complete_approved_action. "
+            "For a REJECTED request, acknowledge the rejection without calling tools."
+        ),
+        tools=[complete_approved_action],
+    )
+
+
+async def run_openai_agent(
+    payload: dict[str, Any],
+    session_id: str,
+    emit: EventEmitter,
+    *,
+    is_recovery: bool = False,
+) -> dict[str, Any]:
+    session = AsyncDatabricksSession(
+        session_id=session_id,
+        autoscaling_endpoint=os.environ["LAKEBASE_AUTOSCALING_ENDPOINT"],
+        schema=os.getenv(
+            "OPENAI_AGENT_SESSION_SCHEMA", "durable_openai_agent_sessions"
+        ),
+    )
+    action = str(payload["action"])
+    decision = payload.get("decision")
+    wait_seconds = int(payload.get("wait_seconds", 0))
+    if decision is None:
+        instruction = f"PROPOSAL: Draft a plan for this action: {action}"
+    elif decision == "approve":
+        instruction = (
+            f"APPROVED: Complete this action: {action}. "
+            f"Call complete_approved_action with wait_seconds={wait_seconds}."
+        )
+    elif decision == "reject":
+        instruction = f"REJECTED: Do not perform this action: {action}"
+    else:
+        raise ValueError("decision must be approve, reject, or omitted")
+
+    if is_recovery:
+        instruction = (
+            "RECOVERY: Continue the interrupted durable run using the existing session. "
+            + instruction
+        )
+
+    result = Runner.run_streamed(_create_agent(), input=instruction, session=session)
+
+    async for event in result.stream_events():
+        if event.type == "raw_response_event":
+            await emit(event.data.model_dump(mode="json"))
+
+    output = str(result.final_output or "")
+    if decision is None:
+        response = {"status": "requires_action", "action": action, "proposal": output}
+        await emit({"type": "agent.approval_required", "action": action})
+        return response
+
+    status = "completed" if decision == "approve" else "rejected"
+    await emit({"type": f"agent.{status}", "action": action})
+    return {"status": status, "action": action, "output": output}

cookbooks/durable-runtime-library/requirements.txt

@@ -0,0 +1,3 @@
+databricks-ai-bridge[agent-server]
+databricks-openai[memory]>=0.17.0
+openai-agents>=0.19.4,<0.20