Workbench — the Orqest agent-runtime container

Every production Orqest agent eventually needs the same four pieces of infrastructure: a MemoryStore for durable facts, a Tracer for span capture, an EventBus for observability fan-out, and a ring-buffer of recent events so reconnecting clients can catch up on what they missed. Wiring these four together was repeated across the Orqest demo and every downstream consumer.

Workbench packages the quartet in one container. Construct it once per session (or once per process for single-tenant apps), reach through it to the underlying primitives, and pass it around instead of plumbing four objects through every function signature.

Minimal example

import asyncio

from orqest import Workbench
from orqest.memory import LocalMemoryStore, MemoryEntry
from orqest.observability import AgentEvent


async def main() -> None:
    wb = Workbench(memory=LocalMemoryStore(db_path="/tmp/memory.db"))

    # Memory — persisted facts
    await wb.memory.store(MemoryEntry(content="User prefers tetrahedral elements"))

    # Tracing — span tree
    span = wb.tracer.start_span("generate_mesh")
    # ... do work ...
    wb.tracer.end_span(span, status="ok")

    # Events — fan-out
    await wb.event_bus.emit(AgentEvent(event_type="plan.init", agent_name="demo"))

    # One snapshot for the UI sidecar
    snap = wb.snapshot()
    assert "trace" in snap and "events" in snap


asyncio.run(main())

Lifecycle patterns

• Process-level (demos, single-tenant CLIs) — build one workbench at startup, share it across requests.
• Per-session (chat apps) — build a workbench per session key; memory is a shared singleton injected in, tracer + bus are fresh.
• Per-request (multi-tenant backends) — fresh tracer + bus per turn, shared memory singleton.

Workbench doesn't force a pattern. Construct it however the consumer's lifecycle demands; pass memory= always, optionally share or instantiate the other three.

Reset semantics

wb.reset() clears the tracer and the recent-events buffer. Memory is not cleared — the point of memory is to outlive resets. If a test needs a clean slate, reset memory explicitly through whatever API the store exposes.

Snapshot shape

wb.snapshot() returns:

{
    "trace": [<span_dict>, ...],     # JSONTracer.export_json()
    "events": [<event_dict>, ...],    # bounded ring buffer
}

Memory is deliberately excluded — MemoryStore reads are async and each backend decides its own query semantics. Consumers composing a full sidecar response add memory alongside:

async def sidecar_state(wb: Workbench) -> dict:
    snap = wb.snapshot()
    snap["memories"] = await wb.memory.recall("", k=30)
    return snap

Runnable demo

notebooks/04_orchestrated_workflow.ipynb — Workbench bundling tracer + bus across a full Router → Parallel → Pipeline → RefinementLoop run.

Reference

Workbench

Workbench(
    *,
    memory: Any,
    tracer: Tracer | None = None,
    event_bus: EventBus | None = None,
    buffer_size: int = 200,
    ui_registry: Any = None,
    auto_register_first_party_ui: bool = True,
)

Runtime container for memory, tracing, events, and replay.

Instance attributes are public — reach through the workbench to talk to the underlying primitives directly when the canonical helpers aren't enough.

Attributes:

Name	Type	Description
memory		A :class:~orqest.memory.store.MemoryStore (protocol; typically :class:~orqest.memory.local.LocalMemoryStore).
tracer	Tracer	A :class:~orqest.observability.tracer.Tracer (defaults to a fresh JSONTracer).
event_bus	EventBus	A :class:~orqest.observability.events.EventBus (defaults to a fresh EventBus).
recent_events	deque[AgentEvent]	Ring buffer of the last buffer_size events seen on the bus. Populated automatically by a subscription wired at construction.

Wire the infrastructure pieces together.

Parameters:

Name	Type	Description	Default
memory	Any	A :class:MemoryStore. Must be constructed ahead of time — Workbench doesn't assume a default backend because memory choices (local SQLite, Supabase, mock) depend on the consumer.	required
tracer	Tracer | None	Optional tracer; a fresh JSONTracer is used when not supplied.	None
event_bus	EventBus | None	Optional event bus; a fresh one is created when not supplied.	None
buffer_size	int	Max events retained in the recent_events ring buffer. Zero disables buffering.	200
ui_registry	Any	Optional :class:ComponentRegistry (lazy imported). When None and auto_register_first_party_ui is True, a fresh registry pre-loaded with the first-party component set (see :func:orqest.ui.default_registry) is constructed. Pass an explicit registry to skip auto-registration or to add custom components.	None
auto_register_first_party_ui	bool	When True (default) and ui_registry is None, the first-party :func:default_registry is used. Set False for a bare-bones registry the consumer populates manually.	True

Source code in orqest/workbench/workbench.py

def __init__(
    self,
    *,
    memory: Any,  # MemoryStore protocol; Any to avoid hard import cycle
    tracer: Tracer | None = None,
    event_bus: EventBus | None = None,
    buffer_size: int = 200,
    ui_registry: Any = None,
    auto_register_first_party_ui: bool = True,
) -> None:
    """Wire the infrastructure pieces together.

    Args:
        memory: A :class:`MemoryStore`. Must be constructed ahead
            of time — Workbench doesn't assume a default backend
            because memory choices (local SQLite, Supabase, mock)
            depend on the consumer.
        tracer: Optional tracer; a fresh ``JSONTracer`` is used
            when not supplied.
        event_bus: Optional event bus; a fresh one is created when
            not supplied.
        buffer_size: Max events retained in the ``recent_events``
            ring buffer. Zero disables buffering.
        ui_registry: Optional :class:`ComponentRegistry` (lazy
            imported). When ``None`` and
            ``auto_register_first_party_ui`` is True, a fresh
            registry pre-loaded with the first-party component set
            (see :func:`orqest.ui.default_registry`) is constructed.
            Pass an explicit registry to skip auto-registration or
            to add custom components.
        auto_register_first_party_ui: When True (default) and
            ``ui_registry`` is None, the first-party
            :func:`default_registry` is used. Set False for a
            bare-bones registry the consumer populates manually.
    """
    self.memory = memory
    self.tracer: Tracer = tracer if tracer is not None else JSONTracer()
    self.event_bus: EventBus = event_bus if event_bus is not None else EventBus()

    if ui_registry is None and auto_register_first_party_ui:
        from orqest.ui.registry import default_registry

        ui_registry = default_registry()
    elif ui_registry is None:
        from orqest.ui.registry import ComponentRegistry

        ui_registry = ComponentRegistry()
    self.ui_registry = ui_registry

    self.recent_events: deque[AgentEvent] = deque(
        maxlen=buffer_size if buffer_size > 0 else None
    )
    if buffer_size > 0:
        self.event_bus.subscribe_all(self._record_event)

reset

reset() -> None

Clear tracer state and the recent-events buffer.

Memory is NOT cleared — the whole point of memory is to outlive resets. Callers that want a full wipe (e.g. integration tests) should call memory.reset() or equivalent themselves.

Source code in orqest/workbench/workbench.py

def reset(self) -> None:
    """Clear tracer state and the recent-events buffer.

    Memory is NOT cleared — the whole point of memory is to
    outlive resets. Callers that want a full wipe (e.g. integration
    tests) should call ``memory.reset()`` or equivalent themselves.
    """
    clear = getattr(self.tracer, "clear", None)
    if callable(clear):
        clear()
    self.recent_events.clear()

snapshot

snapshot() -> dict[str, Any]

Return a JSON-safe snapshot of trace + events.

Memory is excluded because :class:MemoryStore access is async and consumer-specific. Callers compose the memory view themselves when serving a sidecar endpoint.

Source code in orqest/workbench/workbench.py

def snapshot(self) -> dict[str, Any]:
    """Return a JSON-safe snapshot of trace + events.

    Memory is excluded because :class:`MemoryStore` access is
    async and consumer-specific. Callers compose the memory view
    themselves when serving a sidecar endpoint.
    """
    export = getattr(self.tracer, "export_json", None)
    trace_payload: list[dict[str, Any]] = (
        export() if callable(export) else []
    )
    return {
        "trace": trace_payload,
        "events": [_event_to_dict(e) for e in self.recent_events],
    }

with_healing

with_healing(
    config: Any, *, api_key: Any | None = None
) -> Any

Construct a :class:HealingRunner wired to this workbench's bus.

Convenience factory. Lazy-imports :mod:orqest.healing so this module stays import-light for consumers that don't use healing.

Construction alone does not start healing. The runner only subscribes its watchdogs and starts the poll loop when entered as an async context manager. Forgetting the async with leaves you with a silent no-op — no detections, no recoveries, no error. Always use the form below:

.. code-block:: python

async with workbench.with_healing(config) as runner:
    ...  # agent work happens inside this block

Parameters:

Name	Type	Description	Default
config	Any	A :class:HealingConfig instance.	required
api_key	Any | None	Single key or per-provider map for the fallback model chain. Required only if config.fallback_models is non-empty.	None

Returns:

Type	Description
Any	An unstarted :class:HealingRunner. Enter it via ``async
Any	with`` to actually wire the watchdogs.

Source code in orqest/workbench/workbench.py

def with_healing(
    self,
    config: Any,
    *,
    api_key: Any | None = None,
) -> Any:
    """Construct a :class:`HealingRunner` wired to this workbench's bus.

    Convenience factory. Lazy-imports :mod:`orqest.healing` so this
    module stays import-light for consumers that don't use healing.

    **Construction alone does not start healing.** The runner only
    subscribes its watchdogs and starts the poll loop when entered
    as an async context manager. Forgetting the ``async with``
    leaves you with a silent no-op — no detections, no recoveries,
    no error. Always use the form below:

    .. code-block:: python

        async with workbench.with_healing(config) as runner:
            ...  # agent work happens inside this block

    Args:
        config: A :class:`HealingConfig` instance.
        api_key: Single key or per-provider map for the fallback
            model chain. Required only if ``config.fallback_models``
            is non-empty.

    Returns:
        An unstarted :class:`HealingRunner`. Enter it via ``async
        with`` to actually wire the watchdogs.
    """
    from orqest.healing import HealingRunner

    return HealingRunner(config, bus=self.event_bus, api_key=api_key)

with_docker_sandbox

with_docker_sandbox(
    *,
    user_id: str,
    session_id: str,
    image: str = "orqest/agent-runtime:latest",
    allowed_packages: set[str] | None = None,
    memory_mb: int = 2048,
    cpus: float = 2.0,
    pids_limit: int = 512,
    promotion_policy: str = "threshold",
    promotion_threshold: int = 3,
    host_port: int | None = None,
    hmac_secret: bytes | str | None = None,
    **kwargs: Any,
) -> Any

Construct a :class:DockerSandbox wired to this workbench's bus.

user_id and session_id are required positional-keyword args — strict by design. They are framework-controlled (not LLM-emitted) and never appear in any LLM-visible context window: they're carried in the MCP transport's HMAC-signed JWT bearer token, validated by middleware inside the container.

user_id is the persistence boundary — the per-user SQLite tool library mounted at /data inside the container is keyed by orqest-user-<user_id>. Tools persisted by user A's sessions survive into user A's future sessions but never cross to user B.

session_id is the per-container lifecycle key — mint a fresh UUID per session. Reusing one across containers is undefined behavior (the JWT will fail validation against the new container's env-injected ORQEST_SESSION_ID).

Construction alone does not start the container. The sandbox is an async context manager; docker run happens at __aenter__, docker rm -f at __aexit__. Always use:

.. code-block:: python

async with workbench.with_docker_sandbox(
    user_id="alice",
    session_id=str(uuid4()),
) as sandbox:
    ...

Parameters:

Name	Type	Description	Default
user_id	str	Strict per-user identifier. The persistence boundary.	required
session_id	str	Strict per-session identifier. Typically str(uuid4()).	required
image	str	Docker image to run; defaults to orqest/agent-runtime:latest.	'orqest/agent-runtime:latest'
allowed_packages	set[str] | None	Allowlist of pip packages the LLM can declare in GeneratedToolSpec.dependencies. Default-deny — empty set blocks all installs.	None
memory_mb	int	Container --memory cap.	2048
cpus	float	Container --cpus cap.	2.0
pids_limit	int	Container --pids-limit cap (fork-bomb defense).	512
promotion_policy	str	"threshold" (default), "eager", or "operator_approval".	'threshold'
promotion_threshold	int	For "threshold" policy, N successful invocations before persistence.	3
host_port	int | None	Host port to publish; None lets Docker pick.	None
hmac_secret	bytes | str | None	Secret for signing JWTs. None mints a fresh random secret per sandbox (most secure; doesn't survive process restart).	None

Returns:

Type	Description
Any	An unstarted :class:DockerSandbox (async context manager).

Source code in orqest/workbench/workbench.py

def with_docker_sandbox(
    self,
    *,
    user_id: str,
    session_id: str,
    image: str = "orqest/agent-runtime:latest",
    allowed_packages: set[str] | None = None,
    memory_mb: int = 2048,
    cpus: float = 2.0,
    pids_limit: int = 512,
    promotion_policy: str = "threshold",
    promotion_threshold: int = 3,
    host_port: int | None = None,
    hmac_secret: bytes | str | None = None,
    **kwargs: Any,
) -> Any:
    """Construct a :class:`DockerSandbox` wired to this workbench's bus.

    ``user_id`` and ``session_id`` are **required positional-keyword args**
    — strict by design. They are framework-controlled (not LLM-emitted)
    and never appear in any LLM-visible context window: they're carried
    in the MCP transport's HMAC-signed JWT bearer token, validated by
    middleware inside the container.

    ``user_id`` is the persistence boundary — the per-user SQLite tool
    library mounted at ``/data`` inside the container is keyed by
    ``orqest-user-<user_id>``. Tools persisted by user A's sessions
    survive into user A's future sessions but never cross to user B.

    ``session_id`` is the per-container lifecycle key — mint a fresh
    UUID per session. Reusing one across containers is undefined
    behavior (the JWT will fail validation against the new container's
    env-injected ``ORQEST_SESSION_ID``).

    **Construction alone does not start the container.** The sandbox
    is an async context manager; ``docker run`` happens at
    ``__aenter__``, ``docker rm -f`` at ``__aexit__``. Always use:

    .. code-block:: python

        async with workbench.with_docker_sandbox(
            user_id="alice",
            session_id=str(uuid4()),
        ) as sandbox:
            ...

    Args:
        user_id: Strict per-user identifier. The persistence boundary.
        session_id: Strict per-session identifier. Typically ``str(uuid4())``.
        image: Docker image to run; defaults to ``orqest/agent-runtime:latest``.
        allowed_packages: Allowlist of pip packages the LLM can declare in
            ``GeneratedToolSpec.dependencies``. Default-deny — empty set
            blocks all installs.
        memory_mb: Container ``--memory`` cap.
        cpus: Container ``--cpus`` cap.
        pids_limit: Container ``--pids-limit`` cap (fork-bomb defense).
        promotion_policy: ``"threshold"`` (default), ``"eager"``, or
            ``"operator_approval"``.
        promotion_threshold: For ``"threshold"`` policy, N successful
            invocations before persistence.
        host_port: Host port to publish; ``None`` lets Docker pick.
        hmac_secret: Secret for signing JWTs. ``None`` mints a fresh
            random secret per sandbox (most secure; doesn't survive
            process restart).

    Returns:
        An unstarted :class:`DockerSandbox` (async context manager).

    """
    # Lazy import to keep workbench module import-light for users who
    # don't use Docker tier (and to surface a friendly error when the
    # `docker` SDK isn't installed).
    from orqest.sandbox.docker import DockerSandbox

    return DockerSandbox(
        user_id=user_id,
        session_id=session_id,
        image=image,
        allowed_packages=allowed_packages,
        memory_mb=memory_mb,
        cpus=cpus,
        pids_limit=pids_limit,
        promotion_policy=promotion_policy,
        promotion_threshold=promotion_threshold,
        host_port=host_port,
        hmac_secret=hmac_secret,
        bus=self.event_bus,
        **kwargs,
    )