​

Durable Runs

​

A run’s record

import asyncio
from omnicoreagent import OmniCoreAgent, ToolRegistry

tools = ToolRegistry()

@tools.register_tool("get_invoice", idempotent=True)
def get_invoice(invoice_id: str) -> dict:
    """Fetch an invoice."""
    return {"invoice_id": invoice_id, "amount": 120.0, "status": "unpaid"}

async def main():
    agent = OmniCoreAgent(
        name="billing",
        system_instruction="You look up invoices. Answer in plain text, in one sentence.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
    )
    result = await agent.run("Is invoice INV-7 paid?", session_id="acct-42")
    print(result["status"], "|", result["response"])

    run = await agent.get_run(result["run_id"])
    for key in ("run_id", "session_id", "status", "step", "attempt", "trace_ids", "usage"):
        print(f"{key}: {run[key]}")
    for call in run["tool_calls"]:
        print("call:", call["tool_name"], call["state"], call["outcome"])

    runs = await agent.list_runs(session_id="acct-42")
    print("runs in session:", [(r["run_id"], r["status"]) for r in runs])
    await agent.cleanup()

asyncio.run(main())

success | Invoice INV-7 is unpaid.
run_id: run_bcdec4a2ad314260b9b616e9334fc4fe
session_id: acct-42
status: completed
step: 2
attempt: 1
trace_ids: ['trace_69d42e3dac47424ab6b3e5340330fcde']
usage: {'requests': 2, 'request_tokens': 3026, 'response_tokens': 32, 'total_tokens': 3058}
call: get_invoice completed success
runs in session: [('run_bcdec4a2ad314260b9b616e9334fc4fe', 'completed')]

​

Survive a crash

# agent_def.py
import asyncio
import json
import os
from pathlib import Path

os.environ.setdefault("DATABASE_URL", "sqlite:///./runs.db")

from omnicoreagent import MemoryRouter, OmniCoreAgent, ToolRegistry

LEDGER = Path("charges.jsonl")  # stands in for the payment provider's records
tools = ToolRegistry()

@tools.register_tool("charge_card")
async def charge_card(order_id: str, amount: float) -> dict:
    """Charge the customer's card for an order."""
    with LEDGER.open("a") as ledger:
        ledger.write(json.dumps({"order_id": order_id, "amount": amount}) + "\n")
    await asyncio.sleep(30)  # a slow provider: the charge is made, the reply is late
    return {"order_id": order_id, "charged": amount}

@tools.register_tool("list_charges", idempotent=True)
def list_charges(order_id: str) -> dict:
    """List the charges already made for an order."""
    lines = LEDGER.read_text().splitlines() if LEDGER.exists() else []
    return {"charges": [c for c in map(json.loads, lines) if c["order_id"] == order_id]}

def build_agent() -> OmniCoreAgent:
    return OmniCoreAgent(
        name="payments",
        system_instruction=(
            "You take payments. Never charge an order twice. "
            "Answer in plain text, in one sentence."
        ),
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
        memory_router=MemoryRouter("sql"),
        agent_config={"run_lease_seconds": 15},
    )

# start.py
import asyncio
from agent_def import build_agent

async def main():
    agent = build_agent()
    await agent.run("Charge order 1042 for 42 dollars.", run_id="run_order_1042")

asyncio.run(main())

python start.py & PID=$!
until [ -f charges.jsonl ]; do sleep 0.5; done   # the charge is made...
kill -9 $PID                                      # ...and the process dies before the reply

# recover.py
import asyncio
from agent_def import build_agent
from omnicoreagent.core.runs import RunStateConflict

RUN_ID = "run_order_1042"

async def main():
    agent = build_agent()
    run = await agent.get_run(RUN_ID)
    print("found:", run["status"], "| heartbeat:", run["heartbeat_at"])
    for call in run["tool_calls"]:
        print("  call:", call["tool_name"], call["state"])

    while True:
        try:
            result = await agent.resume(RUN_ID)
            break
        except (ValueError, RunStateConflict) as still_live:
            print(still_live)
            await asyncio.sleep(5)

    print(result["status"], "|", result["response"])
    run = await agent.get_run(RUN_ID)
    for call in run["tool_calls"]:
        print("  call:", call["tool_name"], call["state"], call["outcome"])
    print("charges made:", open("charges.jsonl").read().count("\n"))
    await agent.cleanup()

asyncio.run(main())

found: running | heartbeat: 2026-09-27T23:03:34.553926+00:00
  call: list_charges completed
  call: charge_card started
Run run_order_1042 is running in another process (its heartbeat is current)
Run run_order_1042 is running in another process (its heartbeat is current)
Run run_order_1042 is running in another process (its heartbeat is current)
success | Order 1042 has been charged $42.
  call: list_charges completed success
  call: charge_card completed unknown
  call: list_charges completed success
charges made: 1

• The dead process left the record running, with charge_card started.
• Until its heartbeat was 15 seconds old (run_lease_seconds), resume refused: from outside, a dead process and a busy one look the same.
• charge_card is not idempotent, so the runtime did not run it again, and never will: that is the guarantee. The model was told its outcome is unknown; on this run it checked with list_charges, found the charge and answered. One charge, not two. The call’s outcome is unknown.
• list_charges is idempotent, so the runtime may run it again by design: a call whose result was saved is never repeated, but one whose result was lost (it finished while a sibling call was still running) runs again on resume, because running it twice is harmless. The second list_charges above is a new call; how many a run makes depends on the model.
• The runtime never repeats a call whose outcome is unknown, but the model decides what to do next: told the outcome is unknown, a model can choose to call the tool again. That is a new call it made, recorded as one. Give tools that must not run twice a way to check (here list_charges), or make them idempotent (below).

​

How it works

​

Idempotent tools

@tools.register_tool("get_invoice", idempotent=True)
def get_invoice(invoice_id: str) -> dict:
    """Fetch an invoice."""
    ...

​

Your idempotency key

​

A call that runs past its time limit

​

What is guaranteed

• One owner. Saves are versioned, so two processes cannot both advance a run; the one that lost gets RunStateConflict. A live run refreshes its heartbeat even during a long call. If a process stops refreshing it without dying (its store unreachable, say), another may take the run over while that process’s tool is still running.
• Budgets. A model call’s worst case is held before it is sent and its real cost committed after. If the process stops in between, the hold stands until the run is resumed or ended, and is then released: that one call is counted as a call, but its tokens and cost are not. A tool call is counted each time it is dispatched, including once more when an approved call runs.
• Usage in the record (get_run(...)["usage"]) is written when a run ends; a segment whose process stopped adds none. The spend itself is in the budget counters.
• Sub-agents. A delegation is a tool call that is not idempotent, so the rows above apply to it. A child run left behind by a parent that stopped is not resumed, except a worker waiting for an approval.
• Background retries. A retry after a timeout resumes the same run, so completed calls are not repeated. A retry after a failure starts over only if the run made no non-idempotent call; otherwise the run ends failed, is not retried, and lists its side_effects for a person. A worker that died is resumed.

​

What each store keeps when the machine loses power

​

Pause and resume

​

Steer and interrupt a live run

import asyncio
from omnicoreagent import OmniCoreAgent, ToolRegistry

tools = ToolRegistry()

@tools.register_tool("fetch_report", idempotent=True)
async def fetch_report(quarter: str) -> dict:
    """Fetch the sales report for one quarter, such as Q2."""
    await asyncio.sleep(3)  # a slow report service
    return {"quarter": quarter, "revenue": {"Q2": 410, "Q3": 455}.get(quarter, 0)}

async def wait_for_call(agent, run_id, count):
    """Wait until the run has started `count` tool calls."""
    while True:
        run = await agent.get_run(run_id)
        if run and len(run["tool_calls"]) >= count:
            return
        await asyncio.sleep(0.2)

async def main():
    agent = OmniCoreAgent(
        name="analyst",
        system_instruction="You report sales. Fetch one quarter per call. Answer in plain text, in one sentence.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
    )
    run_id = "run_sales_summary"
    task = asyncio.create_task(agent.run("What was revenue in Q2?", run_id=run_id))

    await wait_for_call(agent, run_id, 1)
    print(await agent.steer(run_id, "Also give me Q3, and the change from Q2.", sender="alice"))

    await wait_for_call(agent, run_id, 2)
    print(await agent.interrupt(run_id))

    result = await task
    print("run:", result["status"])
    run = await agent.get_run(run_id)
    print("record:", run["status"], "| step", run["step"], "|",
          [(c["tool_name"], c["outcome"]) for c in run["tool_calls"]])

    result = await agent.resume(run_id)
    print("resume:", result["status"], "|", result["response"])
    await agent.cleanup()

asyncio.run(main())

{'status': 'queued', 'message_id': 'steer_5d842f853a3d4e7d81792978166e9286'}
{'status': 'interrupt_requested', 'run_id': 'run_sales_summary'}
run: interrupted
record: interrupted | step 2 | [('fetch_report', 'success'), ('fetch_report', 'success')]
resume: success | Q2 revenue was 410, Q3 revenue was 455, and the change from Q2 was +45.

• steer(run_id, message, sender=None) queues a message on the record. It arrives as a user message at the run’s next step boundary, never in the middle of a model or tool call; here the model then fetched Q3 as well. A message for a run that is waiting or interrupted arrives when it resumes. It is user input: the injection guardrail checks it first, and a blocked message returns {"status": "blocked", ...} and is never queued. It is recorded in the run’s history and trace (run_steered). Only your application can steer a run; tool output cannot.
• interrupt(run_id) asks a running run to stop at its next step boundary. The Q3 call already running finished first. The run returns status: "interrupted", keeps everything it did, and resume continues it. A run that finishes before it reaches another boundary simply finishes.

​

Runs a dead process left behind

for claim in await agent.claim_orphaned_runs():
    await agent.resume_claimed(claim)

• An orphan is a run of this agent that is running and whose heartbeat is older than its lease. A background run is left to its supervisor.
• A decided run is one in awaiting_approval or awaiting_budget whose every approval is approved or denied (or whose budget request is granted or denied), that no live owner holds, and that nobody has resumed for OMNICOREAGENT_SERVE_ORPHAN_SWEEP_DECIDED_GRACE_SECONDS (30 by default) after the last decision. The sweep resumes it exactly as resume would, so a person’s decision leads somewhere even when the client that should have called resume was refused (a 503 from the admission limit) or is gone. The support desk ramp at 100 users left 119 approved runs waiting like this. A run with a decision still pending is left waiting, and so is a decided run inside its grace period. Pass decided_grace_seconds= to claim_orphaned_runs when you sweep yourself.
• If a client’s resume and the sweep arrive together, the record’s version check lets exactly one win. The other is told the run was taken (409 over HTTP, ValueError from resume), and a call runs once.
• A run is claimed through its record’s own version check, so two replicas never resume the same run, and a live run is never taken from its owner.
• A claim keeps the run’s lease alive with heartbeats until the resumed run’s own heartbeat takes over, so a slow start (a cold model client can take over a minute) never lets a second sweeper claim the run. If you claim runs yourself and do not resume one, call agent.release_claim(claim).
• A resume removes the sandboxes a dead attempt left only after it has saved the run as its own, so a process that lost the claim never removes the winner’s sandbox.
• A recovered run continues under the rules above: completed calls never run again, and a call that did not finish becomes unknown_outcome unless its tool is idempotent.
• Its trace says why: the run_resumed event has cause recovered_after_lapsed_lease, trigger orphan_sweep, the dead previous_owner, and orphaned_seconds. A decided run has the cause an explicit resume gives (approval, budget_grant or budget_denied), the same trigger, and decision (approved, denied, granted) with decided_at.
• A run recovered OMNICOREAGENT_SERVE_ORPHAN_SWEEP_MAX_RECOVERIES times (3) and still not finished is ended failed: it is probably what stops its process. Resuming a decided run does not count as a recovery.
• At most OMNICOREAGENT_SERVE_ORPHAN_SWEEP_MAX_CONCURRENT (2) recovered runs work at once. When a sweep fills every slot, the next one follows as soon as a slot frees, so a backlog (a hundred decided runs) is not resumed two per interval.

​

End a run nobody will resume

# The run was interrupted, as above, and nobody will resume it.
ended = await agent.abandon_run("run_stale", status="cancelled", reason="customer left")
print("abandoned:", ended["status"], ended["error"])
try:
    await agent.resume("run_stale")
except ValueError as exc:
    print("resume:", exc)

abandoned: cancelled {'type': 'RunEndedOutside', 'message': 'customer left'}
resume: Run run_stale is cancelled; only a waiting or stopped run can resume

​

Run again with the same run ID

again = await agent.run("Name a larger prime number.", run_id="run_prime")
run = await agent.get_run("run_prime")
print("again:", again["status"], "| attempt", run["attempt"],
      "| earlier:", [(a["attempt"], a["status"]) for a in run["previous_attempts"]])

again: success | attempt 2 | earlier: [(1, 'completed')]

​

Run statuses

​

A run that ends after side effects

run = await agent.get_run("run_support_7")
print(run["status"], run["error"]["type"])
print(run["side_effects"])

timeout TimeoutError
[{'tool_name': 'issue_refund', 'tool_call_id': 'call_1', 'outcome': 'success'}]

​

One story per run

story = await agent.get_run_trajectory("run_order_1042")
print("status:", story["status"], "| attempt:", story["attempt"])
for segment in story["segments"]:
    print(" ", segment["trace_id"], segment["status"], segment["trace_kept"])
print("tool calls:", story["totals"]["tool_calls"])

status: completed | attempt: 1
  trace_d80546f3f35e4811af5bf2092e041301 interrupted True
  trace_ec36eb4aa0b0423c94df381b016c24cb completed True
tool calls: {'total': 3, 'by_outcome': {'success': 2, 'error': 0, 'rejected': 0, 'timeout': 0, 'cancelled': 1, 'denied': 0, 'awaiting_approval': 0}}

​

Sandboxes across a pause

​

Background runs

​

How long a run’s evidence is kept

from omnicoreagent import OmniCoreAgent

agent = OmniCoreAgent(
    ...,
    agent_config={"run_retention_days": 90},
    telemetry_config={"retention_days": 90, "payload_retention_days": 90},
)

print(await agent.prune_runs())
print(await agent.prune_telemetry())
status = agent.telemetry_retention_status()
print(status["trace_store"]["retention_days"], status["runs"]["retention_days"])

{'trigger': 'explicit', 'retention_days': 30, 'runs_removed': 0, 'error': None, 'before': '2026-08-28T23:12:41.977697+00:00', 'at': '2026-09-27T23:12:42.021941+00:00'}
{'trigger': 'explicit', 'traces_removed': 0, 'payloads_removed': 0, 'payloads_retained_by_reference': 0, 'error': None, 'at': '2026-09-27T23:12:42.645372+00:00'}
7 30

​

Options

​

When things go wrong

ValueError: Run run_order_1042 is running in another process (its heartbeat is current)

ValueError: Run run_prime is completed; only a waiting or stopped run can resume

LookupError: No run run_nope

ValueError: Run run_prime is completed; only a running run can be interrupted
ValueError: Run run_prime is completed; it cannot be steered

ValueError: status must be cancelled, failed or timeout

RunRequestLost: Run run_order_1042 cannot be resumed: its request was never stored

​

Next