Background Tasks

from timbal import Tool

async def slow_operation() -> str:
    """Long-running work."""
    return "done"

long_running_tool = Tool(
    handler=slow_operation,
    background_mode="always",
)

from timbal import Agent
from timbal.tools import Bash

agent = Agent(
    name="task_manager",
    model="openai/gpt-4o-mini",
    tools=[Bash("*")],
)

# Agent starts the command in the background and returns immediately
result = await agent(prompt="Install project dependencies in the background").collect()

{"task_id": "k4d9x2mq7c1a", "status": "running"}

Background task(s) finished since your last turn. ...

<background_task_completed>
task_id: k4d9x2mq7c1a
name: build_site
title: Build the marketing site
status: completed
result: ...
summary: ...
</background_task_completed>

{"source": "runtime", "kind": "background_task_completed"}

from timbal.state import get_background_task

get_background_task("k4d9x2mq7c1a")

{
    "status": "running",           # running | completed | error | cancelled | timed_out | stalled | not_found
    "task_id": "k4d9x2mq7c1a",
    "name": "build_site",
    "title": "Build the marketing site",
    "input": {"prompt": "Build the marketing site"},
    "started_at": 1755385200000,   # Unix ms
    "summary": {
        "text": "...",             # trailing assistant text, capped at 1000 chars
        "phase": "processing",     # last stage label, heuristic phase, or terminal status
        "pct": 50,                 # 0–100 when the child reports progress (else null)
        "last_tool": "write_file", # most recent tool_use name, if any
        "tools_in_flight": ["write_file", "bash"],  # open tool calls not yet finished
        "event_count": 42,
    },
    "transcript_cursor": 42,       # pass as `after` to read raw events
}

from contextlib import aclosing

stream = agent.stream_until_idle(
    prompt="Prepare the report in the background, then tell me the result.",
    max_idle_secs=300,
    max_continuations=10,
)
async with aclosing(stream):
    async for event in stream:
        print(event)

print(stream.stop_reason, stream.last_run_id)

{"kind": "background_completion", "continuation_index": 1, "task_ids": ["..."]}
# with posted events: "kind": "session_event" and "event_ids": [...]

from timbal import Tool

def refresh_cache() -> str:
    """Refresh a cached value."""
    return "refreshed"

Tool(handler=refresh_cache, background_mode="always", background_announce="silent")

from timbal.core.continuation import AgentContinuation, continuation_for_run

async def wait_until_the_line_is_quiet() -> None:
    return None

stream = AgentContinuation(
    agent,
    admit=wait_until_the_line_is_quiet,  # awaited before each follow-up starts
    prompt="Prepare the report in the background, then tell me the result.",
)

# Elsewhere, with only a run id (a request handler, say):
run_id = "not-a-live-session"
driver = continuation_for_run(run_id)
if driver is not None:
    driver.stop()

from timbal.state import post_session_event

# create=True registers a store when this run does not have one yet.
run_id = "session-run"
event = post_session_event(
    run_id,
    "The nightly export finished: 1,204 rows.",
    key="export",
    create=True,
)

from timbal.state import report_progress


async def build_site(repo: str) -> str:
    """Build and deploy the site. Takes a few minutes."""
    for phase in ("install", "build", "deploy"):
        await run_phase(repo, phase)
        report_progress(f"{phase} finished")
    return "Deployed."

from timbal import Tool

async def slow_operation() -> str:
    """Long-running work."""
    return "done"

long_running_tool = Tool(
    handler=slow_operation,
    background_mode="always",
    background_timeout=600.0,  # 10 minutes
)

from timbal import Tool

async def slow_operation() -> str:
    """Long-running work."""
    return "done"

long_running_tool = Tool(
    handler=slow_operation,
    background_mode="always",
    background_stall_timeout=120.0,  # no events for 2 minutes → stalled
)

from timbal import Agent
from timbal.tools import Bash

agent = Agent(
    name="orchestrator",
    model="openai/gpt-5-mini",
    tools=[Bash("*")],
    max_background_concurrent=4,
    max_background_depth=1,
)

from timbal import Agent
from timbal.tools import Bash

agent = Agent(
    name="debugger",
    model="openai/gpt-5-mini",
    tools=[Bash("*")],
    background_transcript_tool=True,
)

from timbal.state import read_background_transcript

page = read_background_transcript("k4d9x2mq7c1a")
# {"status": "running", "task_id": ..., "events": [...], "cursor": 42,
#  "gapped": false, "forgotten_through": 0}

# Later, pick up where you left off
page = read_background_transcript("k4d9x2mq7c1a", after=page["cursor"])

from timbal.state import wait_for_background

cursor = 0

# Block until terminal (or timeout)
snap = await wait_for_background("k4d9x2mq7c1a", timeout=30.0)
if snap["status"] == "running":
    pass  # timeout — poll again

# Long-poll until new events land past a cursor (or terminal / timeout)
snap = await wait_for_background("k4d9x2mq7c1a", after=cursor, timeout=5.0)

from timbal import Agent

agent = Agent(
    name="logger",
    model="openai/gpt-5-mini",
    max_background_log_events=10_000,
    max_background_log_bytes=None,  # unlimited bytes
    background_task_retention_secs=600,
)

from timbal.state.background import current_background_store

store = current_background_store()
record = store.get("k4d9x2mq7c1a") if store is not None else None
if record is not None:
    async for event in record.log.subscribe():
        print(event)

from timbal.state import cancel_background_task, list_background_tasks

list_background_tasks()
# [{"task_id": "k4d9x2mq7c1a", "name": "build_site", "status": "running",
#   "started_at": 1755385200000, "title": "Build the marketing site"}]

cancel_background_task("k4d9x2mq7c1a")   # returns the final snapshot

from timbal import Tool

def start_remote_build() -> str:
    """Start a build outside this process."""
    return "started"

class RemoteApi:
    def stop(self, run_id: str | None) -> None:
        return None

remote_api = RemoteApi()

def stop_remote_build(record):
    remote_api.stop(record.metadata.get("run_id"))

Tool(
    name="build_site",
    handler=start_remote_build,
    background_mode="always",
    on_background_cancel=stop_remote_build,
)