Kitaru vs DBOS: different layers, not competing runtimes

DBOS is a polyglot durable workflow library on Postgres. Code crashes, and DBOS makes sure you don’t start over — workflow state sits alongside your OLTP data in the same database, and Conductor gives ops a UI for scheduled workflows and durable queues. It’s a serious answer to “how do I keep this running,” and it’s not trying to be anything else.

Kitaru isn’t trying to keep anything running. It answers a different question: once an agent run happened, can you re-run it? Kitaru records runs as sessions — through a framework adapter or by importing the traces you already collect — and replays them against your real code, so a model or prompt change gets tested against what actually happened in production.

The two stack cleanly. An agent that DBOS keeps alive in production can still be wrapped by a Kitaru adapter, or have its traces imported, and its sessions become the regression suite that catches the next bad deploy before it ships.

What makes Kitaru unique

How the two surfaces map

from pydantic_ai import Agent
from kitaru_pydantic_ai import KitaruAgent

reviewer = KitaruAgent(
  Agent("openai:gpt-5.4", system_prompt="You're a compliance reviewer."),
  agent_id=AGENT_ID,
)

# DBOS keeps this durable in production.
# Every run also lands as a Kitaru session.
result = reviewer.run_sync(case)

# Later: freeze the sessions that matter, test a change.
import kitaru

client = kitaru.KitaruClient()
cohort = client.cohorts.create("hard-cases", sessions=session_ids)
experiment = client.experiments.create(
  "cheaper-model",
  model="gpt-5-mini",
  tool_policy=History(scope="cohort", on_miss="fail"),
)

before = experiment.run(cohort=cohort, version="v1")
after = experiment.run(cohort=cohort, version="pr-311")
client.compare(before, after)

from dbos import DBOS
import openai

client = openai.OpenAI()

@DBOS.step()
def research(topic: str) -> str:
  resp = client.chat.completions.create(
      model="gpt-4o-mini",
      messages=[{"role": "user",
                 "content": f"Research: {topic}"}],
  )
  return resp.choices[0].message.content

@DBOS.step()
def draft(brief: str) -> str:
  resp = client.chat.completions.create(
      model="gpt-4o",
      messages=[{"role": "user",
                 "content": f"Write a draft:\n{brief}"}],
  )
  return resp.choices[0].message.content

@DBOS.workflow()
def review_flow(topic: str) -> str:
  brief = research(topic)
  text = draft(brief)

  # Human approval via DBOS.send / recv messaging.
  approved = DBOS.recv(timeout_seconds=86400)
  return text if approved else "Rejected"

# Plus: start the DBOS runtime, send approval
# from another process via DBOS.send.