Wrap RunStore pending state, contextual event views and bounded live feeds in routes owned by your application.

Build your own read routes

Dendrux is a library. Your application owns HTTP, auth, tenant resolution, user-level run ownership, approval authority and response schemas. Use RunStore.get_pending to rebuild a form and RunStore.follow for an optional live feed. Neither requires FastAPI; install your web framework separately if you use the example below.

Reads validate their arguments before they query: a malformed id, tenant, cursor, filter or flag raises ValueError with a fixed message, which your route maps to 400 (see RunStore).

Pass an explicit tenant_id on every call: a nonempty string selects that tenant, None selects untenanted runs, and UNSCOPED is an explicit operator-only all-tenant view. Missing and foreign runs raise the same RunNotFoundError. Most older reads still default to all tenants; your routes must never rely on that default.

Pending state and sensitive arguments

In these snippets store and app are application-owned objects. Your current_tenant dependency authenticates a header credential and resolves its tenant. Your owned_run dependency additionally verifies that the authenticated actor may access the requested run, returning its ID or the same 404 as a missing run. Both dependencies are application policy, not Dendrux helpers.

from fastapi import Depends
 
@app.get("/runs/{run_id}/pending")
async def pending(
    run_id: str = Depends(owned_run),
    tenant: str = Depends(current_tenant),
):
    pause = await store.get_pending(run_id, tenant_id=tenant)
    return pause.to_dict() if pause is not None else None

The pause contains kind, status, pause_id, pending_tool_calls, the input question/options, and pause reason/message. Client-tool params use execution values and can contain real personal data. Approval and input calls use model-sent arguments (PII placeholders when applicable). If args_hidden=True, stop and request resolution on the server; never replace unavailable args with {} or approve blind.

Use the matching Agent submit method for each kind, and send the pause back with the answer. Send pause_id down with the form and pass it as expected_pause_id= when the actor decides. A decision aimed at an older pause then raises PauseStatusMismatchError (answer 409) instead of acting on the newer one, and approving without a pause id raises InputValidationError (answer 400). After a 409, call get_pending again before asking the actor again; never retry the old decision. See Pause ids and stale submits.

Optional live feed over SSE

follow has a shared poller per watched run, terminal drain, redaction, heartbeats, capacity (500 followers per store by default, and 100 per tenant scope) and a maximum iteration duration (default 900 seconds). The per-scope cap counts each tenant, all untenanted followers together and all UNSCOPED followers together, so a single-tenant app with more than 100 live readers per process builds EventFeed(store, max_followers_per_tenant=None) and calls feed.follow(...). EventFeed(store, max_followers=...) customizes the store cap. A feed yields a FollowEnd with a cursor on terminal drain, lag or duration expiry. Reconnect using Last-Event-ID; rerun authentication and ownership on every request.

The feed cannot bound a blocked network send. This application response wraps the whole ASGI response in an absolute deadline, including headers and body sends. It closes the follower even if no generator iteration begins or the client disconnects. The transport deadline may end before an end frame is delivered; reconnect using the last successfully received event ID.

import asyncio
import json
 
from fastapi import Depends, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from dendrux.errors import RunNotFoundError
from dendrux.store import FollowCapacityError, FollowEnd, FollowHeartbeat
 
class LiveResponse(StreamingResponse):
    def __init__(self, content, *, follow, deadline):
        super().__init__(content, media_type="text/event-stream", headers={
            "Cache-Control": "no-cache", "X-Accel-Buffering": "no",
        })
        self.follow = follow
        self.deadline = deadline
 
    async def __call__(self, scope, receive, send):
        try:
            async with asyncio.timeout_at(self.deadline):
                await super().__call__(scope, receive, send)
        except TimeoutError:
            pass  # The transport deadline ends the response without an error traceback.
        finally:
            await self.follow.aclose()
 
@app.get("/runs/{run_id}/live")
async def live(
    request: Request,
    run_id: str = Depends(owned_run),
    tenant: str = Depends(current_tenant),
):
    last = request.headers.get("last-event-id")
    if last is not None and (
        not last.isascii() or not last.isdigit() or len(last) > 10 or int(last) > 2**31 - 1
    ):
        raise HTTPException(400, "invalid Last-Event-ID")   # cursors must fit 32 bits
    cursor = int(last) if last is not None else None
    duration = 900
    deadline = asyncio.get_running_loop().time() + duration
    follow = None
    try:
        follow = store.follow(run_id, tenant_id=tenant,
                              after_sequence_index=cursor, max_duration_s=duration)
        await follow.start()
        if follow.drained_terminal:
            await follow.aclose()
            return Response(status_code=204)
    except ValueError:      # a malformed id or cursor is refused before any slot is taken
        raise HTTPException(400, "invalid request")
    except RunNotFoundError:
        await follow.aclose()
        raise HTTPException(404, "run not found")
    except FollowCapacityError:
        await follow.aclose()
        raise HTTPException(503, "busy", headers={"Retry-After": "5"})
    except BaseException:
        if follow is not None:
            await follow.aclose()
        raise
 
    async def frames():
        try:
            async for item in follow:
                if isinstance(item, FollowHeartbeat):
                    yield ": keepalive\n\n"
                elif isinstance(item, FollowEnd):
                    data = {"reason": item.reason, "cursor": item.cursor}
                    yield f"event: end\ndata: {json.dumps(data)}\n\n"
                else:
                    data = {"sequence_index": item.sequence_index,
                            "event_type": item.event_type, "data": item.data}
                    yield f"id: {item.sequence_index}\ndata: {json.dumps(data)}\n\n"
        finally:
            await follow.aclose()
 
    return LiveResponse(frames(), follow=follow, deadline=deadline)

Use streaming fetch with Authorization or your app's auth header; native EventSource cannot send custom headers. Never put credentials in URLs. Parse SSE incrementally, retain each event ID, handle 204 as finished, honor 503/Retry-After, and reconnect after a lag/duration end or a network failure with bounded backoff. Cancel the fetch reader on page teardown. Keep the same run ID across reconnects.

Do not pipe agent.stream() into browser SSE. Closing that stream cancels the run. Start work with agent.run under your application's execution lifecycle and follow the durable database log. A synchronous start can time out before it returns a run ID; persist an app request ID and a recovery association before starting long work.

Reading historical events

events = await store.get_events(run_id, tenant_id=tenant,
                                after_sequence_index=cursor, limit=100)
views = [await store.get_event_view(run_id, event, tenant_id=tenant)
         for event in events]

Each get_event_view call queries run status and can also read active-pause context. The loop above therefore performs at least one additional query per event; use a small page for inspection and follow for live delivery.

Page with the last sequence index. Use get_event_view for contextual approval arguments; it verifies the run is in scope. Pass only events read from that same run: StoredEvent has no run ID to validate their association. Standalone redact_event has no database context and hides unverified active approval/input args. follow already applies the contextual projection. Even the default view intentionally contains actionable client-tool params: authorize the actor before exposing it. Raw run details, traces and tool/provider records are sensitive; select the fields your audience needs instead of serializing dataclasses wholesale. Enabling payloads is reserved for trusted operators. Prompts, provider bodies and PII mappings must never be included in a general browser DTO.

For operator inspection and tracing, see Observability. For runnable request/response clients see Client tools.