API reference

The public API, generated from docstrings.

Agent and results

Reusable agent configuration; run() executes one traced session.

agent = Agent(provider=..., instructions="You are a helpful agent.") result = await agent.run("Summarize README.md")

Source code in src/regista/agent.py

class Agent:
    """Reusable agent configuration; ``run()`` executes one traced session.

    >>> agent = Agent(provider=..., instructions="You are a helpful agent.")
    >>> result = await agent.run("Summarize README.md")
    """

    def __init__(
        self,
        *,
        provider: Provider,
        instructions: str | Instructions,
        tools: list[Tool] | None = None,
        policy: PermissionPolicy | None = None,
        ask_handler: AskHandler | None = None,
        trace_dir: Path | str = "./.regista/traces",
        max_turns: int = 50,
        max_cost_usd: float | None = None,
        max_tokens: int = 8192,
        params: dict[str, Any] | None = None,
        price_overrides: dict[str, ModelPrice] | None = None,
        context: ContextConfig | None = None,
    ) -> None:
        if max_turns < 1:
            raise ConfigurationError(f"max_turns must be >= 1, got {max_turns}")
        self.instructions = Instructions.coerce(instructions)
        self.trace_dir = trace_dir
        self._config = LoopConfig(
            provider=provider,
            registry=ToolRegistry(tools),
            policy=policy if policy is not None else allow_all(),
            ask_handler=ask_handler,
            system=self.instructions.render(),
            max_turns=max_turns,
            max_cost_usd=max_cost_usd,
            max_tokens=max_tokens,
            params=params or {},
            price_overrides=price_overrides,
            context=context or ContextConfig(),
        )

    async def run(self, task: str) -> RunResult:
        """Run one task to completion in a fresh traced session."""
        return await Session(task, self._config, self.trace_dir).run()

    async def resume(self, trace_path: Path | str) -> RunResult:
        """Continue an interrupted session from its trace.

        The recorded prefix replays for $0 — every request hash-verified,
        recorded tool results served without re-running their effects — and
        the first request the recording can't answer falls through to this
        agent's live provider. From there new tool calls execute for real,
        gated by this agent's policy. A tool call the crash cut short (no
        recorded result) is re-executed.

        Assumes this Agent is configured like the recorded run; any
        difference makes the first request diverge, which just means the
        whole task re-runs live. Writes a new trace linked to the original
        via ``replay_of``.
        """
        return await resume_from_trace(trace_path, self._config, self.trace_dir)

    async def stream(self, task: str) -> AsyncIterator[StreamEvent]:
        """Run one task, yielding events as they happen.

        Yields TextDelta/ThinkingDelta/ToolCallStarted/ToolCallFinished/
        TurnCompleted as the session progresses, then RunCompleted with the
        same RunResult ``run()`` would return. The trace is identical to a
        blocking run — streaming changes when you see things, not what
        happened.
        """
        queue: asyncio.Queue[StreamEvent | None] = asyncio.Queue()
        session = Session(task, self._config, self.trace_dir)

        async def pump() -> RunResult:
            try:
                return await session.run(on_event=queue.put_nowait)
            finally:
                queue.put_nowait(None)  # sentinel: no more events

        runner = asyncio.ensure_future(pump())
        try:
            while True:
                event = await queue.get()
                if event is None:
                    break
                yield event
            yield RunCompleted(await runner)  # re-raises if the session raised
        finally:
            if not runner.done():
                runner.cancel()
                await asyncio.gather(runner, return_exceptions=True)

    def run_sync(self, task: str) -> RunResult:
        """``run()`` for synchronous callers (scripts, notebooks without top-level await)."""
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self.run(task))
        raise ConfigurationError(
            "run_sync() was called inside a running event loop; use `await agent.run(task)`"
        )

resume(trace_path) async

Continue an interrupted session from its trace.

The recorded prefix replays for $0 — every request hash-verified, recorded tool results served without re-running their effects — and the first request the recording can't answer falls through to this agent's live provider. From there new tool calls execute for real, gated by this agent's policy. A tool call the crash cut short (no recorded result) is re-executed.

Assumes this Agent is configured like the recorded run; any difference makes the first request diverge, which just means the whole task re-runs live. Writes a new trace linked to the original via replay_of.

Source code in src/regista/agent.py

async def resume(self, trace_path: Path | str) -> RunResult:
    """Continue an interrupted session from its trace.

    The recorded prefix replays for $0 — every request hash-verified,
    recorded tool results served without re-running their effects — and
    the first request the recording can't answer falls through to this
    agent's live provider. From there new tool calls execute for real,
    gated by this agent's policy. A tool call the crash cut short (no
    recorded result) is re-executed.

    Assumes this Agent is configured like the recorded run; any
    difference makes the first request diverge, which just means the
    whole task re-runs live. Writes a new trace linked to the original
    via ``replay_of``.
    """
    return await resume_from_trace(trace_path, self._config, self.trace_dir)

run(task) async

Run one task to completion in a fresh traced session.

Source code in src/regista/agent.py

async def run(self, task: str) -> RunResult:
    """Run one task to completion in a fresh traced session."""
    return await Session(task, self._config, self.trace_dir).run()

run_sync(task)

run() for synchronous callers (scripts, notebooks without top-level await).

Source code in src/regista/agent.py

def run_sync(self, task: str) -> RunResult:
    """``run()`` for synchronous callers (scripts, notebooks without top-level await)."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(self.run(task))
    raise ConfigurationError(
        "run_sync() was called inside a running event loop; use `await agent.run(task)`"
    )

stream(task) async

Run one task, yielding events as they happen.

Yields TextDelta/ThinkingDelta/ToolCallStarted/ToolCallFinished/ TurnCompleted as the session progresses, then RunCompleted with the same RunResult run() would return. The trace is identical to a blocking run — streaming changes when you see things, not what happened.

Source code in src/regista/agent.py

async def stream(self, task: str) -> AsyncIterator[StreamEvent]:
    """Run one task, yielding events as they happen.

    Yields TextDelta/ThinkingDelta/ToolCallStarted/ToolCallFinished/
    TurnCompleted as the session progresses, then RunCompleted with the
    same RunResult ``run()`` would return. The trace is identical to a
    blocking run — streaming changes when you see things, not what
    happened.
    """
    queue: asyncio.Queue[StreamEvent | None] = asyncio.Queue()
    session = Session(task, self._config, self.trace_dir)

    async def pump() -> RunResult:
        try:
            return await session.run(on_event=queue.put_nowait)
        finally:
            queue.put_nowait(None)  # sentinel: no more events

    runner = asyncio.ensure_future(pump())
    try:
        while True:
            event = await queue.get()
            if event is None:
                break
            yield event
        yield RunCompleted(await runner)  # re-raises if the session raised
    finally:
        if not runner.done():
            runner.cancel()
            await asyncio.gather(runner, return_exceptions=True)

What await agent.run(task) returns. trace_path is the receipt: every detail this summary omits is in the trace.

Source code in src/regista/session.py

@dataclass(frozen=True)
class RunResult:
    """What ``await agent.run(task)`` returns. ``trace_path`` is the receipt:
    every detail this summary omits is in the trace."""

    session_id: str
    output: str
    stop_reason: StopReason
    usage: Usage
    cost_usd: float
    turns: int
    trace_path: Path
    error: str | None = None

Tools

Turn a typed function into a tool: @tool or @tool(name=..., ...).

The JSON Schema is generated from the signature; the docstring summary becomes the description and Args: lines become parameter descriptions. Unsupported signatures raise ConfigurationError here, at decoration time.

Source code in src/regista/tools/registry.py

def tool(
    fn: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    parallel_safe: bool = False,
) -> Tool | Callable[[Callable[..., Any]], Tool]:
    """Turn a typed function into a tool: ``@tool`` or ``@tool(name=..., ...)``.

    The JSON Schema is generated from the signature; the docstring summary
    becomes the description and ``Args:`` lines become parameter descriptions.
    Unsupported signatures raise ConfigurationError here, at decoration time.
    """

    def decorate(fn: Callable[..., Any]) -> Tool:
        summary, input_schema = build_input_schema(fn)
        tool_description = description or summary
        if not tool_description:
            raise ConfigurationError(
                f"tool '{fn.__name__}' needs a description: add a docstring "
                "or pass @tool(description=...)"
            )
        spec = ToolSpec(
            name=name or fn.__name__,
            description=tool_description,
            input_schema=input_schema,
            parallel_safe=parallel_safe,
        )
        return Tool(fn, spec)

    return decorate(fn) if fn is not None else decorate

Name → Tool, plus execution with timing and error capture.

Source code in src/regista/tools/registry.py

class ToolRegistry:
    """Name → Tool, plus execution with timing and error capture."""

    def __init__(self, tools: list[Tool] | None = None) -> None:
        self._tools: dict[str, Tool] = {}
        for t in tools or []:
            self.register(t)

    def register(self, tool: Tool) -> None:
        if tool.spec.name in self._tools:
            raise ConfigurationError(f"duplicate tool name '{tool.spec.name}'")
        self._tools[tool.spec.name] = tool

    def __contains__(self, name: str) -> bool:
        return name in self._tools

    def get(self, name: str) -> Tool:
        try:
            return self._tools[name]
        except KeyError:
            raise ToolError(f"unknown tool '{name}'") from None

    def specs(self) -> list[ToolSpec]:
        return [t.spec for t in self._tools.values()]

    async def execute(
        self, name: str, input: dict[str, Any], *, tool_use_id: str = ""
    ) -> ToolExecution:
        """Execute a tool call, capturing failures as data.

        A tool that raises produces ``is_error=True`` content the model can
        adapt to — only an unknown tool name is a harness fault (ToolError,
        raised by ``get``). ``tool_use_id`` is unused here; replay's stub
        registry keys recorded results on it.
        """
        tool = self.get(name)
        started = time.monotonic()
        try:
            content = await tool.invoke(input)
            is_error = False
        except Exception as exc:
            content = f"{type(exc).__name__}: {exc}"
            is_error = True
        duration_ms = int((time.monotonic() - started) * 1000)
        return ToolExecution(content=content, is_error=is_error, duration_ms=duration_ms)

execute(name, input, *, tool_use_id='') async

Execute a tool call, capturing failures as data.

A tool that raises produces is_error=True content the model can adapt to — only an unknown tool name is a harness fault (ToolError, raised by get). tool_use_id is unused here; replay's stub registry keys recorded results on it.

Source code in src/regista/tools/registry.py

async def execute(
    self, name: str, input: dict[str, Any], *, tool_use_id: str = ""
) -> ToolExecution:
    """Execute a tool call, capturing failures as data.

    A tool that raises produces ``is_error=True`` content the model can
    adapt to — only an unknown tool name is a harness fault (ToolError,
    raised by ``get``). ``tool_use_id`` is unused here; replay's stub
    registry keys recorded results on it.
    """
    tool = self.get(name)
    started = time.monotonic()
    try:
        content = await tool.invoke(input)
        is_error = False
    except Exception as exc:
        content = f"{type(exc).__name__}: {exc}"
        is_error = True
    duration_ms = int((time.monotonic() - started) * 1000)
    return ToolExecution(content=content, is_error=is_error, duration_ms=duration_ms)

The standard toolset, bound to one environment.

Returns read_file, write_file, list_dir, glob, search_files, shell, and fetch. Read-only tools are parallel_safe; write_file and shell are not. Pair with a policy — allow_all plus shell means the model can run anything your user account can.

Source code in src/regista/tools/builtin.py

def builtin_tools(
    environment: Environment,
    *,
    max_output_chars: int = 50_000,
    shell_timeout_s: float = 60.0,
    fetch_timeout_s: float = 30.0,
) -> list[Tool]:
    """The standard toolset, bound to one environment.

    Returns read_file, write_file, list_dir, glob, search_files, shell, and
    fetch. Read-only tools are ``parallel_safe``; write_file and shell are not.
    Pair with a policy — ``allow_all`` plus ``shell`` means the model can run
    anything your user account can.
    """

    @tool(parallel_safe=True)
    async def read_file(path: str) -> str:
        """Read a text file from the workspace.

        Args:
            path: Workspace-relative file path, e.g. "src/app.py".
        """
        return _truncate(await environment.read_file(path), max_output_chars)

    @tool
    async def write_file(path: str, content: str) -> str:
        """Write a text file in the workspace, creating parent directories
        and replacing any existing content.

        Args:
            path: Workspace-relative file path, e.g. "src/app.py".
            content: The full new file content.
        """
        await environment.write_file(path, content)
        return f"Wrote {len(content)} characters to {path}"

    @tool(parallel_safe=True)
    async def list_dir(path: str = ".") -> str:
        """List a workspace directory; names ending in "/" are directories.

        Args:
            path: Workspace-relative directory path; defaults to the root.
        """
        entries = await environment.list_dir(path)
        return "\n".join(entries) if entries else f"(empty directory: {path})"

    @tool(parallel_safe=True)
    async def glob(pattern: str) -> str:
        """Find workspace files matching a glob pattern.

        Args:
            pattern: A glob like "**/*.py" or "src/*.md", relative to the
                workspace root.
        """
        matches = await environment.glob(pattern)
        if not matches:
            return f"(no files match {pattern!r})"
        return _truncate("\n".join(matches), max_output_chars)

    @tool(parallel_safe=True)
    async def search_files(pattern: str, glob: str = "**/*") -> str:
        """Search workspace files for a regular expression; results are
        "path:line: text" lines.

        Args:
            pattern: A Python regular expression to search for.
            glob: Which files to search, as a glob pattern; defaults to all.
        """
        regex = re.compile(pattern)
        matches: list[str] = []
        for path in await environment.glob(glob):
            try:
                content = await environment.read_file(path)
            except (UnicodeDecodeError, OSError):
                continue  # binary or unreadable files are not searchable
            for line_no, line in enumerate(content.splitlines(), start=1):
                if regex.search(line):
                    matches.append(f"{path}:{line_no}: {line.strip()}")
                    if len(matches) >= _MAX_SEARCH_MATCHES:
                        matches.append(f"... [stopped at {_MAX_SEARCH_MATCHES} matches]")
                        return _truncate("\n".join(matches), max_output_chars)
        if not matches:
            return f"(no matches for {pattern!r} in {glob!r})"
        return _truncate("\n".join(matches), max_output_chars)

    @tool
    async def shell(command: str) -> str:
        """Run a shell command in the workspace root and return its output.

        Args:
            command: The command line to execute.
        """
        result = await environment.exec(command, timeout_s=shell_timeout_s)
        parts = []
        if result.timed_out:
            parts.append(f"[command timed out after {shell_timeout_s:g}s and was killed]")
        parts.append(f"exit code: {result.exit_code}")
        if result.stdout:
            parts.append(f"stdout:\n{result.stdout}")
        if result.stderr:
            parts.append(f"stderr:\n{result.stderr}")
        return _truncate("\n".join(parts), max_output_chars)

    @tool(parallel_safe=True)
    async def fetch(url: str) -> str:
        """Fetch a URL over HTTP(S) and return the response body as text.

        Args:
            url: An http:// or https:// URL.
        """
        if not url.startswith(("http://", "https://")):
            raise ValueError(f"fetch only supports http(s) URLs, got {url!r}")
        async with httpx.AsyncClient(follow_redirects=True, timeout=fetch_timeout_s) as client:
            response = await client.get(url)
        body = _truncate(response.text, max_output_chars)
        return f"HTTP {response.status_code}\n\n{body}"

    return [read_file, write_file, list_dir, glob, search_files, shell, fetch]

A connection to one MCP server; an async context manager.

async with MCPServer.stdio("uvx", ["mcp-server-fetch"]) as server: ... agent = Agent(provider=..., instructions=..., tools=await server.tools()) ... await agent.run(task)

The connection must stay open for as long as the agent may call the tools. prefix namespaces tool names ({prefix}__{name}) so two servers exporting the same tool name can coexist in one registry.

Source code in src/regista/tools/mcp.py

class MCPServer:
    """A connection to one MCP server; an async context manager.

    >>> async with MCPServer.stdio("uvx", ["mcp-server-fetch"]) as server:
    ...     agent = Agent(provider=..., instructions=..., tools=await server.tools())
    ...     await agent.run(task)

    The connection must stay open for as long as the agent may call the
    tools. ``prefix`` namespaces tool names (``{prefix}__{name}``) so two
    servers exporting the same tool name can coexist in one registry.
    """

    def __init__(
        self,
        transport_factory: Callable[[], Any],
        *,
        prefix: str | None = None,
        timeout_s: float | None = 30.0,
    ) -> None:
        self._transport_factory = transport_factory
        self._prefix = prefix
        self._timeout_s = timeout_s
        self._stack: AsyncExitStack | None = None
        self._session: Any = None

    @classmethod
    def stdio(
        cls,
        command: str,
        args: list[str] | None = None,
        *,
        env: dict[str, str] | None = None,
        prefix: str | None = None,
        timeout_s: float | None = 30.0,
    ) -> MCPServer:
        """A server spawned as a subprocess, spoken to over stdin/stdout."""
        mcp = _mcp()
        params = mcp.StdioServerParameters(command=command, args=args or [], env=env)
        return cls(
            lambda: mcp.client.stdio.stdio_client(params), prefix=prefix, timeout_s=timeout_s
        )

    @classmethod
    def http(
        cls,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        prefix: str | None = None,
        timeout_s: float | None = 30.0,
    ) -> MCPServer:
        """A remote server spoken to over streamable HTTP."""
        mcp = _mcp()
        return cls(
            lambda: mcp.client.streamable_http.streamablehttp_client(url, headers=headers),
            prefix=prefix,
            timeout_s=timeout_s,
        )

    async def __aenter__(self) -> MCPServer:
        mcp = _mcp()
        self._stack = AsyncExitStack()
        try:
            # stdio yields (read, write); streamable HTTP appends a session-id getter
            streams = await self._stack.enter_async_context(self._transport_factory())
            timeout = timedelta(seconds=self._timeout_s) if self._timeout_s else None
            self._session = await self._stack.enter_async_context(
                mcp.ClientSession(streams[0], streams[1], read_timeout_seconds=timeout)
            )
            await self._session.initialize()
        except BaseException:
            await self._stack.aclose()
            self._stack = None
            self._session = None
            raise
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        if self._stack is not None:
            await self._stack.aclose()
            self._stack = None
            self._session = None

    def _require_session(self) -> Any:
        if self._session is None:
            raise ConfigurationError(
                "MCPServer is not connected — use it as an async context manager"
            )
        return self._session

    async def tools(self) -> list[Tool]:
        """List the server's tools, each wrapped as a regista Tool."""
        listed = await self._require_session().list_tools()
        return [self._wrap(t) for t in listed.tools]

    def _wrap(self, mcp_tool: Any) -> Tool:
        mcp_name: str = mcp_tool.name
        name = f"{self._prefix}__{mcp_name}" if self._prefix else mcp_name
        spec = ToolSpec(
            name=name,
            description=mcp_tool.description or f"MCP tool '{mcp_name}'",
            input_schema=mcp_tool.inputSchema,
            parallel_safe=False,  # the protocol doesn't declare it; assume effects
        )

        async def call(**kwargs: Any) -> str:
            result = await self._require_session().call_tool(mcp_name, arguments=kwargs)
            text = _render_content(result.content)
            if result.isError:
                raise MCPToolError(text or f"MCP tool '{mcp_name}' failed")
            return text

        call.__name__ = name
        return Tool(call, spec)

http(url, *, headers=None, prefix=None, timeout_s=30.0) classmethod

A remote server spoken to over streamable HTTP.

Source code in src/regista/tools/mcp.py

@classmethod
def http(
    cls,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    prefix: str | None = None,
    timeout_s: float | None = 30.0,
) -> MCPServer:
    """A remote server spoken to over streamable HTTP."""
    mcp = _mcp()
    return cls(
        lambda: mcp.client.streamable_http.streamablehttp_client(url, headers=headers),
        prefix=prefix,
        timeout_s=timeout_s,
    )

stdio(command, args=None, *, env=None, prefix=None, timeout_s=30.0) classmethod

A server spawned as a subprocess, spoken to over stdin/stdout.

Source code in src/regista/tools/mcp.py

@classmethod
def stdio(
    cls,
    command: str,
    args: list[str] | None = None,
    *,
    env: dict[str, str] | None = None,
    prefix: str | None = None,
    timeout_s: float | None = 30.0,
) -> MCPServer:
    """A server spawned as a subprocess, spoken to over stdin/stdout."""
    mcp = _mcp()
    params = mcp.StdioServerParameters(command=command, args=args or [], env=env)
    return cls(
        lambda: mcp.client.stdio.stdio_client(params), prefix=prefix, timeout_s=timeout_s
    )

tools() async

List the server's tools, each wrapped as a regista Tool.

Source code in src/regista/tools/mcp.py

async def tools(self) -> list[Tool]:
    """List the server's tools, each wrapped as a regista Tool."""
    listed = await self._require_session().list_tools()
    return [self._wrap(t) for t in listed.tools]

Replay

Re-run a recorded session; each live request is hash-verified.

mode: "strict" raises ReplayDivergence with a structural diff, "warn" warns and continues positionally, "hybrid" falls through to fallback (required) from the first divergence onward.

The replay's own trace lands next to the original unless trace_dir says otherwise. Returns the replay's RunResult — cost is $0 for every call served from the recording.

Source code in src/regista/replay.py

async def replay(
    trace_path: Path | str,
    *,
    mode: ReplayMode = "strict",
    fallback: Provider | None = None,
    trace_dir: Path | str | None = None,
    max_turns: int | None = None,
) -> RunResult:
    """Re-run a recorded session; each live request is hash-verified.

    ``mode``: "strict" raises ReplayDivergence with a structural diff,
    "warn" warns and continues positionally, "hybrid" falls through to
    ``fallback`` (required) from the first divergence onward.

    The replay's own trace lands next to the original unless ``trace_dir``
    says otherwise. Returns the replay's RunResult — cost is $0 for every
    call served from the recording.
    """
    trace_path = Path(trace_path)
    trace = Trace.load(trace_path)
    start = trace.start
    calls = trace.llm_calls()
    if not calls:
        raise ValueError(f"{trace_path} contains no recorded LLM calls to replay")
    first_request = ModelRequest.model_validate(calls[0][0].request)

    if max_turns is None:
        max_turns = trace.end.turns if trace.end else 50

    config = LoopConfig(
        provider=ReplayProvider(trace, mode=mode, fallback=fallback),
        registry=_StubToolRegistry(start.tool_schemas, trace.tool_results()),
        policy=_replay_stub_policy,
        ask_handler=None,
        system=start.instructions,
        max_turns=max_turns,
        max_cost_usd=None,
        max_tokens=first_request.max_tokens,
        params=dict(first_request.params),
        context=ContextConfig.model_validate(start.context),
    )
    session = Session(
        start.task, config, trace_dir or trace_path.parent, replay_of=start.session_id
    )
    return await session.run()

Serves one recorded trace's LLM responses, hash-verified call by call.

Source code in src/regista/providers/replay.py

class ReplayProvider:
    """Serves one recorded trace's LLM responses, hash-verified call by call."""

    name = "replay"

    def __init__(
        self,
        trace: Trace,
        *,
        mode: ReplayMode = "strict",
        fallback: Provider | None = None,
    ) -> None:
        if mode == "hybrid" and fallback is None:
            raise ConfigurationError("hybrid replay needs a fallback provider to fall through to")
        self.model = trace.start.model
        self._calls = trace.llm_calls()
        self._mode: ReplayMode = mode
        self._fallback = fallback
        self._index = 0
        self._live = False  # hybrid: set on first divergence, never unset

    async def complete(self, request: ModelRequest) -> ModelResponse:
        if self._live and self._fallback is not None:
            return await self._fallback.complete(request)

        if self._index >= len(self._calls):
            if self._mode == "hybrid" and self._fallback is not None:
                self._live = True
                return await self._fallback.complete(request)
            raise ReplayDivergence(
                f"recording exhausted: live run wants call {self._index + 1}, "
                f"only {len(self._calls)} were recorded",
                seq=self._calls[-1][0].seq if self._calls else -1,
            )

        recorded_request, recorded_response = self._calls[self._index]
        if request.request_hash() != recorded_request.request_hash:
            diff = diff_requests(recorded_request.request, request.model_dump(mode="json"))
            message = (
                f"request {self._index + 1} diverged from the recording "
                f"(trace seq {recorded_request.seq})"
            )
            if self._mode == "strict":
                raise ReplayDivergence(f"{message}\n{diff}", seq=recorded_request.seq, diff=diff)
            if self._mode == "hybrid" and self._fallback is not None:
                self._live = True
                return await self._fallback.complete(request)
            warnings.warn(f"{message}\n{diff}", ReplayDivergenceWarning, stacklevel=2)

        self._index += 1
        replayed = ModelResponse.model_validate(recorded_response.response)
        return replayed.model_copy(update={"replayed": True})

    async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
        """Synthetic deltas from the recorded response — a replay can drive a
        live UI exactly like the original run did."""
        response = await self.complete(request)
        for delta in synthetic_deltas(response.message):
            yield delta
        yield response

stream(request) async

Synthetic deltas from the recorded response — a replay can drive a live UI exactly like the original run did.

Source code in src/regista/providers/replay.py

async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
    """Synthetic deltas from the recorded response — a replay can drive a
    live UI exactly like the original run did."""
    response = await self.complete(request)
    for delta in synthetic_deltas(response.message):
        yield delta
    yield response

Evals

Eval/regression runner: assert on outcomes and trace shape, live or for $0.

An :class:EvalSuite is a list of tasks, each with checks — small functions that inspect the run's :class:~regista.session.RunResult and its :class:~regista.trace.reader.Trace and return None (pass) or a failure message. Three ways to run the same suite:

• await suite.run(agent) — live sessions, real cost. Development.
• await suite.record(agent) — live sessions whose traces are saved as fixtures (only when the task passes — a failing fixture is never committed).
• await suite.replay() — no agent, no keys, $0: each fixture is strictly replayed (any divergence fails the task — the regression signal) and the checks are judged against the recorded run, so cost/turn assertions mean what they meant when the fixture was recorded.

The intended CI shape is one pytest test::

report = await suite.replay()
assert report.passed, f"\n{report}"

Check = Callable[[RunResult, Trace], 'str | None'] module-attribute

A check inspects one run and returns None (pass) or a failure message.

Any function with this shape works; its __name__ labels it in reports.

EvalReport dataclass

The suite verdict. assert report.passed, f"\n{report}" reads well.

Source code in src/regista/evals.py

@dataclass(frozen=True)
class EvalReport:
    """The suite verdict. ``assert report.passed, f"\\n{report}"`` reads well."""

    tasks: list[TaskReport]

    @property
    def passed(self) -> bool:
        return all(t.passed for t in self.tasks)

    @property
    def total_cost_usd(self) -> float:
        return sum(t.result.cost_usd for t in self.tasks if t.result is not None)

    def __str__(self) -> str:
        done = sum(1 for t in self.tasks if t.passed)
        lines = [f"{done}/{len(self.tasks)} tasks passed, cost ${self.total_cost_usd:.4f}"]
        for t in self.tasks:
            lines.append(f"  {'PASS' if t.passed else 'FAIL'}  {t.name}")
            if t.error is not None:
                lines.append(f"        run failed: {_short(t.error, 200)}")
            for c in t.checks:
                if not c.passed:
                    lines.append(f"        {c.name}: {c.message}")
        return "\n".join(lines)

EvalSuite

A named set of tasks judged by the same machinery live or replayed.

Source code in src/regista/evals.py

class EvalSuite:
    """A named set of tasks judged by the same machinery live or replayed."""

    def __init__(self, tasks: Sequence[EvalTask]) -> None:
        names = [t.name for t in tasks]
        if len(set(names)) != len(names):
            duplicates = sorted({n for n in names if names.count(n) > 1})
            raise ValueError(f"duplicate task names: {duplicates}")
        self._tasks = list(tasks)

    async def run(self, agent: Agent) -> EvalReport:
        """Run every task live against ``agent`` and judge the checks."""
        return EvalReport([await self._run_one(task, agent) for task in self._tasks])

    async def record(self, agent: Agent) -> EvalReport:
        """Run live and save each *passing* task's trace to its fixture path.

        A failing task's fixture is never written — you can't accidentally
        commit a regression as the new baseline. Tasks without a ``trace``
        path just run.
        """
        reports = []
        for task in self._tasks:
            report = await self._run_one(task, agent)
            if report.passed and task.trace is not None and report.result is not None:
                destination = Path(task.trace)
                destination.parent.mkdir(parents=True, exist_ok=True)
                shutil.copyfile(report.result.trace_path, destination)
            reports.append(report)
        return EvalReport(reports)

    async def replay(self, *, trace_dir: Path | str | None = None) -> EvalReport:
        """Judge every task against its recorded fixture, for $0.

        Each fixture is strictly replayed first — a `ReplayDivergence` fails
        the task, because it means the harness/tool/prompt behavior no longer
        reproduces the recording. The checks are then judged against the
        *recorded* run, so cost and turn numbers are the original ones.
        """
        reports = []
        for task in self._tasks:
            reports.append(await self._replay_one(task, trace_dir))
        return EvalReport(reports)

    async def _run_one(self, task: EvalTask, agent: Agent) -> TaskReport:
        try:
            result = await agent.run(task.task)
        except Exception as exc:  # one broken task must not sink the suite
            return TaskReport(task.name, False, [], error=f"{type(exc).__name__}: {exc}")
        # a run that ended in an error outcome fails its task regardless of checks
        return _judge(task, result, Trace.load(result.trace_path), run_error=result.error)

    async def _replay_one(self, task: EvalTask, trace_dir: Path | str | None) -> TaskReport:
        if task.trace is None:
            return TaskReport(task.name, False, [], error="task has no trace fixture path")
        fixture = Path(task.trace)
        if not fixture.exists():
            return TaskReport(
                task.name, False, [], error=f"fixture {fixture} not found — run suite.record()"
            )
        recorded = Trace.load(fixture)
        result = _result_from_trace(recorded)
        if result is None:
            return TaskReport(
                task.name, False, [], error=f"fixture {fixture} has no session.end (crashed run?)"
            )
        try:
            await _replay(fixture, trace_dir=trace_dir)
            faithful = CheckOutcome("strict_replay", True)
        except ReplayDivergence as exc:
            faithful = CheckOutcome("strict_replay", False, _short(str(exc), 400))
        return _judge(task, result, recorded, extra=[faithful])

record(agent) async

Run live and save each passing task's trace to its fixture path.

A failing task's fixture is never written — you can't accidentally commit a regression as the new baseline. Tasks without a trace path just run.

Source code in src/regista/evals.py

async def record(self, agent: Agent) -> EvalReport:
    """Run live and save each *passing* task's trace to its fixture path.

    A failing task's fixture is never written — you can't accidentally
    commit a regression as the new baseline. Tasks without a ``trace``
    path just run.
    """
    reports = []
    for task in self._tasks:
        report = await self._run_one(task, agent)
        if report.passed and task.trace is not None and report.result is not None:
            destination = Path(task.trace)
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(report.result.trace_path, destination)
        reports.append(report)
    return EvalReport(reports)

replay(*, trace_dir=None) async

Judge every task against its recorded fixture, for $0.

Each fixture is strictly replayed first — a ReplayDivergence fails the task, because it means the harness/tool/prompt behavior no longer reproduces the recording. The checks are then judged against the recorded run, so cost and turn numbers are the original ones.

Source code in src/regista/evals.py

async def replay(self, *, trace_dir: Path | str | None = None) -> EvalReport:
    """Judge every task against its recorded fixture, for $0.

    Each fixture is strictly replayed first — a `ReplayDivergence` fails
    the task, because it means the harness/tool/prompt behavior no longer
    reproduces the recording. The checks are then judged against the
    *recorded* run, so cost and turn numbers are the original ones.
    """
    reports = []
    for task in self._tasks:
        reports.append(await self._replay_one(task, trace_dir))
    return EvalReport(reports)

run(agent) async

Run every task live against agent and judge the checks.

Source code in src/regista/evals.py

async def run(self, agent: Agent) -> EvalReport:
    """Run every task live against ``agent`` and judge the checks."""
    return EvalReport([await self._run_one(task, agent) for task in self._tasks])

EvalTask dataclass

One task in a suite: a prompt, its checks, and (optionally) where its recorded fixture lives for record()/replay().

Source code in src/regista/evals.py

@dataclass(frozen=True)
class EvalTask:
    """One task in a suite: a prompt, its checks, and (optionally) where its
    recorded fixture lives for ``record()``/``replay()``."""

    name: str
    task: str
    checks: Sequence[Check] = ()
    trace: Path | str | None = None

max_cost_usd(limit)

The run cost at most limit dollars (recorded cost, under replay).

Source code in src/regista/evals.py

def max_cost_usd(limit: float) -> Check:
    """The run cost at most ``limit`` dollars (recorded cost, under replay)."""

    def check(result: RunResult, trace: Trace) -> str | None:
        if result.cost_usd <= limit:
            return None
        return f"cost ${result.cost_usd:.4f}, limit ${limit:.4f}"

    return _named(f"max_cost_usd({limit})", check)

max_turns_used(limit)

The run took at most limit turns.

Source code in src/regista/evals.py

def max_turns_used(limit: int) -> Check:
    """The run took at most ``limit`` turns."""

    def check(result: RunResult, trace: Trace) -> str | None:
        if result.turns <= limit:
            return None
        return f"used {result.turns} turns, limit {limit}"

    return _named(f"max_turns_used({limit})", check)

no_errors()

The trace has no error events and no is_error tool results.

Source code in src/regista/evals.py

def no_errors() -> Check:
    """The trace has no error events and no is_error tool results."""

    def check(result: RunResult, trace: Trace) -> str | None:
        errors = [e for e in trace if isinstance(e, ErrorEvent)]
        failed_tools = [e for e in trace if isinstance(e, ToolResult) and e.is_error]
        if not errors and not failed_tools:
            return None
        return f"{len(errors)} error event(s), {len(failed_tools)} failed tool result(s)" + (
            f"; first: {_short(errors[0].message)}" if errors else ""
        )

    return _named("no_errors", check)

output_contains(text)

The final output includes text.

Source code in src/regista/evals.py

def output_contains(text: str) -> Check:
    """The final output includes ``text``."""

    def check(result: RunResult, trace: Trace) -> str | None:
        if text in result.output:
            return None
        return f"output does not contain {text!r}: {_short(result.output)!r}"

    return _named(f"output_contains({text!r})", check)

stop_reason_is(expected)

The run stopped for expected ("completed", "max_turns", …).

Source code in src/regista/evals.py

def stop_reason_is(expected: str) -> Check:
    """The run stopped for ``expected`` ("completed", "max_turns", …)."""

    def check(result: RunResult, trace: Trace) -> str | None:
        if result.stop_reason == expected:
            return None
        return f"stop_reason is {result.stop_reason!r}, expected {expected!r}"

    return _named(f"stop_reason_is({expected!r})", check)

tool_never_called(name)

No tool.call for name appears in the trace.

Source code in src/regista/evals.py

def tool_never_called(name: str) -> Check:
    """No ``tool.call`` for ``name`` appears in the trace."""

    def check(result: RunResult, trace: Trace) -> str | None:
        count = sum(1 for e in trace if isinstance(e, ToolCall) and e.name == name)
        if count == 0:
            return None
        return f"tool {name!r} was called {count}x"

    return _named(f"tool_never_called({name!r})", check)

tool_was_called(name)

At least one tool.call for name appears in the trace.

Source code in src/regista/evals.py

def tool_was_called(name: str) -> Check:
    """At least one ``tool.call`` for ``name`` appears in the trace."""

    def check(result: RunResult, trace: Trace) -> str | None:
        if any(isinstance(e, ToolCall) and e.name == name for e in trace):
            return None
        return f"tool {name!r} was never called"

    return _named(f"tool_was_called({name!r})", check)

Policies

The permission gate: a pure decision function consulted before every tool call.

A policy maps a PermissionRequest to Allow, Deny, or Ask. Deny is data, not an exception — the model receives an error tool_result and adapts. Ask escalates to the Agent's ask_handler; with no handler configured it resolves to deny, so the harness never hangs waiting for input that can't arrive.

Presets beyond allow_all (read_only, workspace) land with the built-in tools.

PermissionRequest dataclass

Everything a policy (or a human, via Ask) needs to decide.

Source code in src/regista/policy/permissions.py

@dataclass(frozen=True)
class PermissionRequest:
    """Everything a policy (or a human, via Ask) needs to decide."""

    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    turn: int
    metadata: dict[str, Any] = field(default_factory=dict)

allow_all()

Permit every tool call. Fine for FakeProvider tests and trusted tools; prefer an asking policy for anything with side effects.

Source code in src/regista/policy/permissions.py

def allow_all() -> Callable[[PermissionRequest], PermissionDecision]:
    """Permit every tool call. Fine for FakeProvider tests and trusted tools;
    prefer an asking policy for anything with side effects."""

    def _allow_all(request: PermissionRequest) -> PermissionDecision:
        return Allow()

    _allow_all.policy_name = "allow_all"  # type: ignore[attr-defined]
    return _allow_all

policy_name(policy)

The name recorded in permission.decision events.

Source code in src/regista/policy/permissions.py

def policy_name(policy: object) -> str:
    """The name recorded in permission.decision events."""
    name = getattr(policy, "policy_name", None) or getattr(policy, "__name__", None)
    return str(name) if name else type(policy).__name__

Ready-made permission policies.

Presets decide by tool name — they pair with the built-in toolset and treat anything they don't recognize conservatively (deny or ask, never allow). Ask resolves to deny when no ask_handler is configured, so the safe default costs nothing to adopt.

For anything richer, write a function: a policy is just (PermissionRequest) -> Allow | Deny | Ask.

compose(*policies)

Consult policies in order; the first Deny or Ask decides.

Allow requires unanimity, so composing only ever tightens: stack a targeted deny in front of a permissive preset, e.g. compose(deny_shell, workspace()).

Source code in src/regista/policy/presets.py

def compose(*policies: PermissionPolicy) -> PermissionPolicy:
    """Consult policies in order; the first Deny or Ask decides.

    Allow requires unanimity, so composing only ever tightens: stack a
    targeted deny in front of a permissive preset, e.g.
    ``compose(deny_shell, workspace())``.
    """

    async def _composed(request: PermissionRequest) -> PermissionDecision:
        for policy in policies:
            decision = policy(request)
            if isawaitable(decision):
                decision = await decision
            if not isinstance(decision, Allow):
                return decision
        return Allow()

    names = ", ".join(policy_name(policy) for policy in policies)
    _composed.policy_name = f"compose({names})"  # type: ignore[attr-defined]
    return _composed

read_only(*, allow=())

Permit only known read-only built-ins (plus allow extras); deny the rest.

The strictest preset: right for exploration and summarization agents.

Source code in src/regista/policy/presets.py

def read_only(*, allow: Iterable[str] = ()) -> Callable[[PermissionRequest], PermissionDecision]:
    """Permit only known read-only built-ins (plus ``allow`` extras); deny the rest.

    The strictest preset: right for exploration and summarization agents.
    """
    allowed = _READ_ONLY_BUILTINS | set(allow)

    def _read_only(request: PermissionRequest) -> PermissionDecision:
        if request.tool_name in allowed:
            return Allow()
        return Deny(reason=f"read_only policy: '{request.tool_name}' is not a known read-only tool")

    _read_only.policy_name = "read_only"  # type: ignore[attr-defined]
    return _read_only

workspace()

Permit workspace-scoped file tools (reads and writes); ask for everything else — shell, fetch, and custom tools all escalate.

The environment already pins file effects to the workspace root, so this preset draws the line at effects that can leave it.

Source code in src/regista/policy/presets.py

def workspace() -> Callable[[PermissionRequest], PermissionDecision]:
    """Permit workspace-scoped file tools (reads and writes); ask for
    everything else — shell, fetch, and custom tools all escalate.

    The environment already pins file effects to the workspace root, so this
    preset draws the line at effects that can leave it.
    """

    def _workspace(request: PermissionRequest) -> PermissionDecision:
        if request.tool_name in _WORKSPACE_BUILTINS:
            return Allow()
        return Ask(prompt=f"Allow tool '{request.tool_name}' with input {request.tool_input!r}?")

    _workspace.policy_name = "workspace"  # type: ignore[attr-defined]
    return _workspace

Providers

The LLM boundary: one protocol, four implementations.

An adapter's whole job is normalization — wire format, tool-call shape, usage extraction, retries. Because ReplayProvider implements this same protocol, deterministic replay needs no special support anywhere else in the harness (ARCHITECTURE.md §6, §7).

Streaming lands in v0.1 step 12; the protocol grows a stream() method then.

ModelRequest

Bases: BaseModel

A fully-assembled, provider-neutral request — everything the model will see.

Source code in src/regista/providers/base.py

class ModelRequest(BaseModel):
    """A fully-assembled, provider-neutral request — everything the model will see."""

    model_config = ConfigDict(frozen=True)

    model: str
    system: str | None = None
    messages: list[Message]
    tools: list[ToolSpec] = []
    max_tokens: int = 8192
    params: dict[str, Any] = {}
    """Provider-specific passthrough, e.g. ``{"thinking": {"type": "adaptive"}}``."""

    def request_hash(self) -> str:
        """The replay divergence detector: hash of everything deterministic.

        The full request is included — if any input that shapes the request
        changes (a message, a tool schema, a sampling param), the hash changes.
        """
        return canonical_hash(self.model_dump(mode="json"))

params = {} class-attribute instance-attribute

Provider-specific passthrough, e.g. {"thinking": {"type": "adaptive"}}.

request_hash()

The replay divergence detector: hash of everything deterministic.

The full request is included — if any input that shapes the request changes (a message, a tool schema, a sampling param), the hash changes.

Source code in src/regista/providers/base.py

def request_hash(self) -> str:
    """The replay divergence detector: hash of everything deterministic.

    The full request is included — if any input that shapes the request
    changes (a message, a tool schema, a sampling param), the hash changes.
    """
    return canonical_hash(self.model_dump(mode="json"))

ModelResponse

Bases: BaseModel

A normalized response. raw is the provider's escape hatch and replayed marks responses served from a recording — both are excluded from trace serialization so a replay's payload stays byte-identical to the original's.

Source code in src/regista/providers/base.py

class ModelResponse(BaseModel):
    """A normalized response. ``raw`` is the provider's escape hatch and
    ``replayed`` marks responses served from a recording — both are excluded
    from trace serialization so a replay's payload stays byte-identical to
    the original's."""

    model_config = ConfigDict(frozen=True)

    message: Message
    stop_reason: StopReason
    usage: Usage = Usage()
    model: str = ""
    request_id: str | None = None
    raw: dict[str, Any] | None = None
    replayed: bool = False

    def model_dump_trace(self) -> dict[str, Any]:
        """The dict recorded in llm.response events (and served back by replay)."""
        return self.model_dump(mode="json", exclude={"raw", "replayed"})

model_dump_trace()

The dict recorded in llm.response events (and served back by replay).

Source code in src/regista/providers/base.py

def model_dump_trace(self) -> dict[str, Any]:
    """The dict recorded in llm.response events (and served back by replay)."""
    return self.model_dump(mode="json", exclude={"raw", "replayed"})

Provider

Bases: Protocol

Anything that can turn a ModelRequest into a ModelResponse.

model is part of the protocol because the model is chosen where the provider is constructed — the loop only reads it to assemble requests.

Source code in src/regista/providers/base.py

@runtime_checkable
class Provider(Protocol):
    """Anything that can turn a ModelRequest into a ModelResponse.

    ``model`` is part of the protocol because the model is chosen where the
    provider is constructed — the loop only reads it to assemble requests.
    """

    @property
    def name(self) -> str: ...

    @property
    def model(self) -> str: ...

    async def complete(self, request: ModelRequest) -> ModelResponse: ...

    def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
        """Yield deltas as they arrive, then the final ModelResponse last.

        Adapters without native streaming can wrap ``complete()`` with
        ``regista.streaming.synthetic_deltas``.
        """
        ...

stream(request)

Yield deltas as they arrive, then the final ModelResponse last.

Adapters without native streaming can wrap complete() with regista.streaming.synthetic_deltas.

Source code in src/regista/providers/base.py

def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
    """Yield deltas as they arrive, then the final ModelResponse last.

    Adapters without native streaming can wrap ``complete()`` with
    ``regista.streaming.synthetic_deltas``.
    """
    ...

Provider backed by the Anthropic Messages API.

The model is chosen here, explicitly — regista never defaults a model. Pass api_key or rely on ANTHROPIC_API_KEY; client injection exists for tests and custom transports.

Source code in src/regista/providers/anthropic.py

class AnthropicProvider:
    """``Provider`` backed by the Anthropic Messages API.

    The model is chosen here, explicitly — regista never defaults a model.
    Pass ``api_key`` or rely on ``ANTHROPIC_API_KEY``; ``client`` injection
    exists for tests and custom transports.
    """

    name = "anthropic"

    def __init__(
        self,
        model: str,
        *,
        api_key: str | None = None,
        base_url: str | None = None,
        timeout_s: float = 600.0,
        max_retries: int = 2,
        client: anthropic.AsyncAnthropic | None = None,
    ) -> None:
        self.model = model
        self._client = client or anthropic.AsyncAnthropic(
            api_key=api_key, base_url=base_url, timeout=timeout_s, max_retries=max_retries
        )

    def _build_kwargs(self, request: ModelRequest) -> dict[str, Any]:
        kwargs: dict[str, Any] = {
            "model": request.model,
            "max_tokens": request.max_tokens,
            "messages": _wire_messages(request.messages),
        }
        if request.system:
            kwargs["system"] = [
                {
                    "type": "text",
                    "text": request.system,
                    "cache_control": {"type": "ephemeral"},
                }
            ]
        if request.tools:
            # parallel_safe is harness metadata and never reaches the wire
            kwargs["tools"] = [
                {"name": t.name, "description": t.description, "input_schema": t.input_schema}
                for t in request.tools
            ]
        kwargs.update(request.params)
        return kwargs

    def _map_error(self, exc: anthropic.APIError) -> ProviderError:
        if isinstance(exc, anthropic.APIStatusError):
            retryable = exc.status_code in (408, 429) or exc.status_code >= 500
            return ProviderError(
                f"Anthropic API returned {exc.status_code}: {exc.message}",
                provider=self.name,
                retryable=retryable,
            )
        return ProviderError(
            f"connection to Anthropic failed: {exc}", provider=self.name, retryable=True
        )

    def _normalize(self, response: Any) -> ModelResponse:
        content = [b for b in map(_from_wire_block, response.content) if b is not None]
        stop_reason: StopReason = (
            cast("StopReason", response.stop_reason)
            if response.stop_reason in _KNOWN_STOP_REASONS
            else "other"
        )
        return ModelResponse(
            message=Message(role="assistant", content=content),
            stop_reason=stop_reason,
            usage=Usage(
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
                cache_read_tokens=response.usage.cache_read_input_tokens or 0,
                cache_write_tokens=response.usage.cache_creation_input_tokens or 0,
            ),
            model=response.model,
            request_id=getattr(response, "_request_id", None),
            raw=response.model_dump(mode="json"),
        )

    async def complete(self, request: ModelRequest) -> ModelResponse:
        try:
            response = await self._client.messages.create(**self._build_kwargs(request))
        except anthropic.APIConnectionError as exc:
            raise self._map_error(exc) from exc
        except anthropic.APIStatusError as exc:
            raise self._map_error(exc) from exc
        return self._normalize(response)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
        try:
            async with self._client.messages.stream(**self._build_kwargs(request)) as stream:
                async for event in stream:
                    if event.type == "content_block_delta":
                        if event.delta.type == "text_delta":
                            yield TextDelta(event.delta.text)
                        elif event.delta.type == "thinking_delta":
                            yield ThinkingDelta(event.delta.thinking)
                message = await stream.get_final_message()
        except anthropic.APIConnectionError as exc:
            raise self._map_error(exc) from exc
        except anthropic.APIStatusError as exc:
            raise self._map_error(exc) from exc
        yield self._normalize(message)

Provider for any Chat Completions-compatible endpoint.

api_key is optional because local servers (Ollama, vLLM) don't need one. The model is chosen here, explicitly — regista never defaults one.

Source code in src/regista/providers/openai_compat.py

class OpenAICompatProvider:
    """``Provider`` for any Chat Completions-compatible endpoint.

    ``api_key`` is optional because local servers (Ollama, vLLM) don't need
    one. The model is chosen here, explicitly — regista never defaults one.
    """

    name = "openai_compat"

    def __init__(
        self,
        model: str,
        *,
        base_url: str = "https://api.openai.com/v1",
        api_key: str | None = None,
        timeout_s: float = 600.0,
        max_retries: int = 2,
        backoff_base_s: float = 0.5,
    ) -> None:
        self.model = model
        self._max_retries = max_retries
        self._backoff_base_s = backoff_base_s
        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
        self._client = httpx.AsyncClient(
            base_url=base_url.rstrip("/"), headers=headers, timeout=timeout_s
        )

    async def aclose(self) -> None:
        await self._client.aclose()

    def _build_payload(self, request: ModelRequest) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "model": request.model,
            "max_tokens": request.max_tokens,
            "messages": _wire_messages(request.system, request.messages),
        }
        if request.tools:
            payload["tools"] = [
                {
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.input_schema,
                    },
                }
                for t in request.tools
            ]
        payload.update(request.params)
        return payload

    def _status_error(self, status_code: int, body: str) -> ProviderError:
        retryable = status_code in (408, 429) or status_code >= 500
        return ProviderError(
            f"endpoint returned {status_code}: {body[:500]}",
            provider=self.name,
            retryable=retryable,
        )

    @staticmethod
    def _usage_from(usage: dict[str, Any]) -> Usage:
        cached = (usage.get("prompt_tokens_details") or {}).get("cached_tokens") or 0
        return Usage(
            input_tokens=usage.get("prompt_tokens") or 0,
            output_tokens=usage.get("completion_tokens") or 0,
            cache_read_tokens=cached,
        )

    async def complete(self, request: ModelRequest) -> ModelResponse:
        payload = self._build_payload(request)
        last_error = ProviderError("no attempts made", provider=self.name)
        for attempt in range(self._max_retries + 1):
            try:
                response = await self._client.post("/chat/completions", json=payload)
            except httpx.TransportError as exc:
                last_error = ProviderError(
                    f"connection to {self._client.base_url} failed: {exc}",
                    provider=self.name,
                    retryable=True,
                )
            else:
                if response.status_code == 200:
                    return self._normalize(response.json())
                last_error = self._status_error(response.status_code, response.text)
                if not last_error.retryable:
                    raise last_error
            if attempt < self._max_retries:
                await asyncio.sleep(self._backoff_base_s * 2**attempt)
        raise last_error

    def _normalize(self, body: dict[str, Any]) -> ModelResponse:
        try:
            choice = body["choices"][0]
        except (KeyError, IndexError) as exc:
            raise ProviderError(
                f"malformed response: no choices in {json.dumps(body)[:500]}",
                provider=self.name,
            ) from exc
        wire_message = choice.get("message") or {}

        content: list[ContentBlock] = []
        if wire_message.get("content"):
            content.append(TextBlock(text=wire_message["content"]))
        for call in wire_message.get("tool_calls") or []:
            content.append(
                ToolUseBlock(
                    id=call["id"],
                    name=call["function"]["name"],
                    input=_parse_arguments(call["function"].get("arguments") or ""),
                )
            )

        return ModelResponse(
            message=Message(role="assistant", content=content),
            stop_reason=_STOP_REASONS.get(choice.get("finish_reason"), "other"),
            usage=self._usage_from(body.get("usage") or {}),
            model=body.get("model") or self.model,
            request_id=body.get("id"),
            raw=body,
        )

    async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
        """SSE streaming: one attempt, no retry loop (a broken stream can't be
        resumed mid-response). Tool-call arguments arrive in fragments and are
        assembled per index before parsing."""
        payload = self._build_payload(request)
        payload["stream"] = True
        payload["stream_options"] = {"include_usage": True}

        text_parts: list[str] = []
        calls: dict[int, dict[str, str]] = {}
        finish_reason = ""
        usage: dict[str, Any] = {}
        response_model = ""
        request_id: str | None = None

        try:
            async with self._client.stream("POST", "/chat/completions", json=payload) as response:
                if response.status_code != 200:
                    body = (await response.aread()).decode(errors="replace")
                    raise self._status_error(response.status_code, body)
                async for line in response.aiter_lines():
                    if not line.startswith("data:"):
                        continue
                    data = line[len("data:") :].strip()
                    if data == "[DONE]":
                        break
                    chunk = json.loads(data)
                    response_model = chunk.get("model") or response_model
                    request_id = chunk.get("id") or request_id
                    if chunk.get("usage"):
                        usage = chunk["usage"]
                    for choice in chunk.get("choices") or []:
                        if choice.get("finish_reason"):
                            finish_reason = choice["finish_reason"]
                        delta = choice.get("delta") or {}
                        if delta.get("content"):
                            text_parts.append(delta["content"])
                            yield TextDelta(delta["content"])
                        for fragment in delta.get("tool_calls") or []:
                            slot = calls.setdefault(
                                fragment.get("index") or 0,
                                {"id": "", "name": "", "arguments": ""},
                            )
                            if fragment.get("id"):
                                slot["id"] = fragment["id"]
                            function = fragment.get("function") or {}
                            if function.get("name"):
                                slot["name"] = function["name"]
                            if function.get("arguments"):
                                slot["arguments"] += function["arguments"]
        except httpx.TransportError as exc:
            raise ProviderError(
                f"stream from {self._client.base_url} failed: {exc}",
                provider=self.name,
                retryable=True,
            ) from exc

        content: list[ContentBlock] = []
        if text_parts:
            content.append(TextBlock(text="".join(text_parts)))
        for index in sorted(calls):
            slot = calls[index]
            content.append(
                ToolUseBlock(
                    id=slot["id"] or f"call_{index}",
                    name=slot["name"],
                    input=_parse_arguments(slot["arguments"]),
                )
            )
        yield ModelResponse(
            message=Message(role="assistant", content=content),
            stop_reason=_STOP_REASONS.get(finish_reason, "other"),
            usage=self._usage_from(usage),
            model=response_model or self.model,
            request_id=request_id,
        )

stream(request) async

SSE streaming: one attempt, no retry loop (a broken stream can't be resumed mid-response). Tool-call arguments arrive in fragments and are assembled per index before parsing.

Source code in src/regista/providers/openai_compat.py

async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
    """SSE streaming: one attempt, no retry loop (a broken stream can't be
    resumed mid-response). Tool-call arguments arrive in fragments and are
    assembled per index before parsing."""
    payload = self._build_payload(request)
    payload["stream"] = True
    payload["stream_options"] = {"include_usage": True}

    text_parts: list[str] = []
    calls: dict[int, dict[str, str]] = {}
    finish_reason = ""
    usage: dict[str, Any] = {}
    response_model = ""
    request_id: str | None = None

    try:
        async with self._client.stream("POST", "/chat/completions", json=payload) as response:
            if response.status_code != 200:
                body = (await response.aread()).decode(errors="replace")
                raise self._status_error(response.status_code, body)
            async for line in response.aiter_lines():
                if not line.startswith("data:"):
                    continue
                data = line[len("data:") :].strip()
                if data == "[DONE]":
                    break
                chunk = json.loads(data)
                response_model = chunk.get("model") or response_model
                request_id = chunk.get("id") or request_id
                if chunk.get("usage"):
                    usage = chunk["usage"]
                for choice in chunk.get("choices") or []:
                    if choice.get("finish_reason"):
                        finish_reason = choice["finish_reason"]
                    delta = choice.get("delta") or {}
                    if delta.get("content"):
                        text_parts.append(delta["content"])
                        yield TextDelta(delta["content"])
                    for fragment in delta.get("tool_calls") or []:
                        slot = calls.setdefault(
                            fragment.get("index") or 0,
                            {"id": "", "name": "", "arguments": ""},
                        )
                        if fragment.get("id"):
                            slot["id"] = fragment["id"]
                        function = fragment.get("function") or {}
                        if function.get("name"):
                            slot["name"] = function["name"]
                        if function.get("arguments"):
                            slot["arguments"] += function["arguments"]
    except httpx.TransportError as exc:
        raise ProviderError(
            f"stream from {self._client.base_url} failed: {exc}",
            provider=self.name,
            retryable=True,
        ) from exc

    content: list[ContentBlock] = []
    if text_parts:
        content.append(TextBlock(text="".join(text_parts)))
    for index in sorted(calls):
        slot = calls[index]
        content.append(
            ToolUseBlock(
                id=slot["id"] or f"call_{index}",
                name=slot["name"],
                input=_parse_arguments(slot["arguments"]),
            )
        )
    yield ModelResponse(
        message=Message(role="assistant", content=content),
        stop_reason=_STOP_REASONS.get(finish_reason, "other"),
        usage=self._usage_from(usage),
        model=response_model or self.model,
        request_id=request_id,
    )

A scripted provider for tests — public API, not a test-only hack.

Users unit-test their agents with it; contributors test loop changes with it; ~90% of regista's own suite runs on it. Zero network, zero keys, zero cost.

FakeProvider

Serves a scripted sequence of responses and records every request.

Source code in src/regista/providers/fake.py

class FakeProvider:
    """Serves a scripted sequence of responses and records every request."""

    name = "fake"

    def __init__(self, responses: list[ModelResponse], *, model: str = "fake-model") -> None:
        self.model = model
        self._responses = list(responses)
        self._served = 0
        self.requests: list[ModelRequest] = []

    async def complete(self, request: ModelRequest) -> ModelResponse:
        self.requests.append(request)
        if self._served >= len(self._responses):
            raise ProviderError(
                f"FakeProvider script exhausted after {len(self._responses)} responses",
                provider=self.name,
            )
        response = self._responses[self._served]
        self._served += 1
        return response

    async def stream(self, request: ModelRequest) -> AsyncIterator[ProviderDelta | ModelResponse]:
        response = await self.complete(request)
        for delta in synthetic_deltas(response.message):
            yield delta
        yield response

text_response(text, *, usage=None)

A final assistant message ending the turn.

Source code in src/regista/providers/fake.py

def text_response(text: str, *, usage: Usage | None = None) -> ModelResponse:
    """A final assistant message ending the turn."""
    return ModelResponse(
        message=Message(role="assistant", content=[TextBlock(text=text)]),
        stop_reason="end_turn",
        usage=usage or Usage(input_tokens=10, output_tokens=10),
        model="fake-model",
    )

tool_use_response(*calls, text='', usage=None)

An assistant message requesting tool calls.

Each call is (tool_use_id, tool_name, input).

Source code in src/regista/providers/fake.py

def tool_use_response(
    *calls: tuple[str, str, dict[str, Any]],
    text: str = "",
    usage: Usage | None = None,
) -> ModelResponse:
    """An assistant message requesting tool calls.

    Each call is ``(tool_use_id, tool_name, input)``.
    """
    blocks: list[TextBlock | ToolUseBlock] = []
    if text:
        blocks.append(TextBlock(text=text))
    blocks.extend(ToolUseBlock(id=id_, name=name, input=input_) for id_, name, input_ in calls)
    return ModelResponse(
        message=Message(role="assistant", content=list(blocks)),
        stop_reason="tool_use",
        usage=usage or Usage(input_tokens=10, output_tokens=10),
        model="fake-model",
    )

Streaming

The streaming event vocabulary for agent.stream().

Streaming changes when you see things, never what happened: the trace records only final requests and responses, so a streamed session produces the same trace — and replays with the same hashes — as a blocking one. Providers yield TextDelta/ThinkingDelta items followed by their final ModelResponse; the loop adds the tool and turn events around them.

synthetic_deltas chunks a completed message into deltas — how FakeProvider and ReplayProvider stream, and the reason replayed sessions can still drive a live UI.

ProviderDelta = TextDelta | ThinkingDelta module-attribute

What providers yield before their final ModelResponse.

StreamEvent = TextDelta | ThinkingDelta | ToolCallStarted | ToolCallFinished | TurnCompleted | RunCompleted module-attribute

Everything agent.stream() can yield.

RunCompleted dataclass

Always the final event: the same RunResult run() would return.

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class RunCompleted:
    """Always the final event: the same RunResult ``run()`` would return."""

    result: RunResult

TextDelta dataclass

A fragment of assistant text, in order.

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class TextDelta:
    """A fragment of assistant text, in order."""

    text: str

ThinkingDelta dataclass

A fragment of extended-thinking content.

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class ThinkingDelta:
    """A fragment of extended-thinking content."""

    thinking: str

ToolCallFinished dataclass

A tool call resolved — executed, failed, or denied (is_error).

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class ToolCallFinished:
    """A tool call resolved — executed, failed, or denied (``is_error``)."""

    tool_use_id: str
    name: str
    content: str
    is_error: bool

ToolCallStarted dataclass

The model requested a tool call; the gate and execution come next.

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class ToolCallStarted:
    """The model requested a tool call; the gate and execution come next."""

    tool_use_id: str
    name: str
    input: dict[str, Any]

TurnCompleted dataclass

One llm round-trip finished; usage and cost are for this turn only.

Source code in src/regista/streaming.py

@dataclass(frozen=True)
class TurnCompleted:
    """One llm round-trip finished; usage and cost are for this turn only."""

    turn: int
    stop_reason: StopReason
    usage: Usage
    cost_usd: float | None

synthetic_deltas(message, *, chunk_size=8)

Chunk a completed message into deltas, for providers that don't (or can't) stream natively — FakeProvider and ReplayProvider.

Source code in src/regista/streaming.py

def synthetic_deltas(message: Message, *, chunk_size: int = 8) -> Iterator[ProviderDelta]:
    """Chunk a completed message into deltas, for providers that don't
    (or can't) stream natively — FakeProvider and ReplayProvider."""
    for block in message.content:
        if isinstance(block, TextBlock):
            for i in range(0, len(block.text), chunk_size):
                yield TextDelta(block.text[i : i + chunk_size])
        elif isinstance(block, ThinkingBlock):
            for i in range(0, len(block.thinking), chunk_size):
                yield ThinkingDelta(block.thinking[i : i + chunk_size])

Context

Context management: keep the conversation inside the model's budget.

The trigger is provider-reported usage (never a local token estimate): when a turn's observed input tokens cross max_input_tokens, the oldest messages are summarized — by the session's own provider — and replaced with a single summary message, keeping the most recent messages verbatim.

Compaction is itself traced and replayable: the summarization call is a regular llm.request/llm.response pair (with its own request_hash), followed by a context.compaction event. A replayed session re-runs this exact logic and is served the recorded summary, so post-compaction request hashes still match. That is why compaction lives behind the same provider seam as everything else.

ContextConfig

Bases: BaseModel

When and how to compact. Recorded in session.start so replay can reproduce the same compaction points.

Source code in src/regista/context/__init__.py

class ContextConfig(BaseModel):
    """When and how to compact. Recorded in session.start so replay can
    reproduce the same compaction points."""

    model_config = ConfigDict(frozen=True)

    max_input_tokens: int | None = None
    """Compact when a turn's observed input tokens reach this; None disables."""

    keep_recent_messages: int = 4
    """Messages preserved verbatim at the end of the history."""

    summary_max_tokens: int = 2048

keep_recent_messages = 4 class-attribute instance-attribute

Messages preserved verbatim at the end of the history.

max_input_tokens = None class-attribute instance-attribute

Compact when a turn's observed input tokens reach this; None disables.

compact_history(history, *, provider, context, writer, turn, observed_input_tokens, price_overrides=None) async

Summarize the old history; returns (new_history, usage, cost) or None if there is nothing safe to drop. Provider errors propagate to the loop.

Source code in src/regista/context/__init__.py

async def compact_history(
    history: list[Message],
    *,
    provider: Provider,
    context: ContextConfig,
    writer: TraceWriter,
    turn: int,
    observed_input_tokens: int,
    price_overrides: dict[str, ModelPrice] | None = None,
) -> tuple[list[Message], Usage, float | None] | None:
    """Summarize the old history; returns (new_history, usage, cost) or None
    if there is nothing safe to drop. Provider errors propagate to the loop."""
    split = _split_point(history, context.keep_recent_messages)
    if split < 1:
        return None

    request = ModelRequest(
        model=provider.model,
        system=COMPACTION_SYSTEM,
        messages=[
            Message.user(
                "Summarize this agent conversation so far:\n\n" + render_transcript(history[:split])
            )
        ],
        max_tokens=context.summary_max_tokens,
    )
    writer.emit(
        ev.LlmRequest(
            turn=turn, request=request.model_dump(mode="json"), request_hash=request.request_hash()
        )
    )
    started = time.monotonic()
    response = await provider.complete(request)
    latency_ms = int((time.monotonic() - started) * 1000)

    if response.replayed:
        cost: float | None = 0.0
    else:
        cost = cost_usd(response.model, response.usage, price_overrides)
    writer.emit(
        ev.LlmResponse(
            turn=turn,
            response=response.model_dump_trace(),
            usage=response.usage,
            cost_usd=cost,
            latency_ms=latency_ms,
            replayed=response.replayed,
        )
    )

    summary = response.message.text()
    new_history = [
        Message.user(f"[Conversation so far, compacted by the harness]\n\n{summary}"),
        *history[split:],
    ]
    writer.emit(
        ev.ContextCompaction(
            tokens_before=observed_input_tokens,
            # a chars/4 estimate; the next llm.response reports the real count
            tokens_after=len(render_transcript(new_history)) // 4,
            summary=summary,
            dropped_messages=split,
        )
    )
    return new_history, response.usage, cost

render_transcript(history)

A deterministic plain-text transcript for the summarizer.

Thinking blocks are excluded — private reasoning is not context the summary should carry forward.

Source code in src/regista/context/__init__.py

def render_transcript(history: list[Message]) -> str:
    """A deterministic plain-text transcript for the summarizer.

    Thinking blocks are excluded — private reasoning is not context the
    summary should carry forward.
    """
    lines: list[str] = []
    for message in history:
        parts: list[str] = []
        for block in message.content:
            if isinstance(block, TextBlock):
                parts.append(block.text)
            elif isinstance(block, ToolUseBlock):
                parts.append(f"[tool_use {block.name} {json.dumps(block.input, sort_keys=True)}]")
            elif isinstance(block, ToolResultBlock):
                content = block.content
                if len(content) > _TRANSCRIPT_RESULT_CAP:
                    content = f"{content[:_TRANSCRIPT_RESULT_CAP]}... [truncated]"
                status = "error" if block.is_error else "ok"
                parts.append(f"[tool_result {status}: {content}]")
            elif isinstance(block, ThinkingBlock):
                continue
        lines.append(f"{message.role}: " + "\n".join(parts))
    return "\n\n".join(lines)

Types

Provider-neutral message vocabulary shared by every module.

The shapes here are a superset of the Anthropic Messages API (role plus a list of typed content blocks). Adapters translate their wire formats into this model, never out of it: rich-to-flat translation (for OpenAI-style APIs) is easy, flat-to-rich is lossy. See ARCHITECTURE.md §7.

This module depends on nothing else in regista.

Message

Bases: BaseModel

One conversation entry: a role and a list of typed content blocks.

Source code in src/regista/types.py

class Message(BaseModel):
    """One conversation entry: a role and a list of typed content blocks."""

    model_config = ConfigDict(frozen=True)

    role: Role
    content: list[ContentBlock]

    @classmethod
    def user(cls, text: str) -> Message:
        return cls(role="user", content=[TextBlock(text=text)])

    @classmethod
    def assistant(cls, text: str) -> Message:
        return cls(role="assistant", content=[TextBlock(text=text)])

    def text(self) -> str:
        """Concatenated text of all TextBlocks (ignores tool/thinking blocks)."""
        return "".join(block.text for block in self.content if isinstance(block, TextBlock))

    def tool_uses(self) -> list[ToolUseBlock]:
        return [block for block in self.content if isinstance(block, ToolUseBlock)]

text()

Concatenated text of all TextBlocks (ignores tool/thinking blocks).

Source code in src/regista/types.py

def text(self) -> str:
    """Concatenated text of all TextBlocks (ignores tool/thinking blocks)."""
    return "".join(block.text for block in self.content if isinstance(block, TextBlock))

TextBlock

Bases: BaseModel

Plain assistant or user text.

Source code in src/regista/types.py

class TextBlock(BaseModel):
    """Plain assistant or user text."""

    model_config = ConfigDict(frozen=True)

    type: Literal["text"] = "text"
    text: str

ThinkingBlock

Bases: BaseModel

Extended-thinking content.

signature is preserved verbatim: the Anthropic API requires it when thinking blocks are sent back in multi-turn history.

Source code in src/regista/types.py

class ThinkingBlock(BaseModel):
    """Extended-thinking content.

    ``signature`` is preserved verbatim: the Anthropic API requires it when
    thinking blocks are sent back in multi-turn history.
    """

    model_config = ConfigDict(frozen=True)

    type: Literal["thinking"] = "thinking"
    thinking: str
    signature: str | None = None

ToolResultBlock

Bases: BaseModel

The harness reporting a tool's outcome back to the model.

A permission denial or tool failure is carried as is_error=True — it is data the model can adapt to, not an exception.

Source code in src/regista/types.py

class ToolResultBlock(BaseModel):
    """The harness reporting a tool's outcome back to the model.

    A permission denial or tool failure is carried as ``is_error=True`` —
    it is data the model can adapt to, not an exception.
    """

    model_config = ConfigDict(frozen=True)

    type: Literal["tool_result"] = "tool_result"
    tool_use_id: str
    content: str
    is_error: bool = False

ToolSpec

Bases: BaseModel

What the model sees of a tool: name, description, and JSON Schema input.

parallel_safe is harness metadata (read-only tools opt in to concurrent execution); it is never sent to the model.

Source code in src/regista/types.py

class ToolSpec(BaseModel):
    """What the model sees of a tool: name, description, and JSON Schema input.

    ``parallel_safe`` is harness metadata (read-only tools opt in to concurrent
    execution); it is never sent to the model.
    """

    model_config = ConfigDict(frozen=True)

    name: str
    description: str
    input_schema: dict[str, Any]
    parallel_safe: bool = False

ToolUseBlock

Bases: BaseModel

The model requesting a tool call. The harness executes; the model only asks.

Source code in src/regista/types.py

class ToolUseBlock(BaseModel):
    """The model requesting a tool call. The harness executes; the model only asks."""

    model_config = ConfigDict(frozen=True)

    type: Literal["tool_use"] = "tool_use"
    id: str
    name: str
    input: dict[str, Any]

Usage

Bases: BaseModel

Token accounting as reported by the provider (never locally estimated).

Cache fields map from Anthropic's cache_read_input_tokens / cache_creation_input_tokens; providers without caching leave them 0.

Source code in src/regista/types.py

class Usage(BaseModel):
    """Token accounting as reported by the provider (never locally estimated).

    Cache fields map from Anthropic's ``cache_read_input_tokens`` /
    ``cache_creation_input_tokens``; providers without caching leave them 0.
    """

    model_config = ConfigDict(frozen=True)

    input_tokens: int = 0
    output_tokens: int = 0
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0

    def __add__(self, other: Usage) -> Usage:
        return Usage(
            input_tokens=self.input_tokens + other.input_tokens,
            output_tokens=self.output_tokens + other.output_tokens,
            cache_read_tokens=self.cache_read_tokens + other.cache_read_tokens,
            cache_write_tokens=self.cache_write_tokens + other.cache_write_tokens,
        )

Trace

The versioned trace event schema — regista's most important contract.

Every subsystem writes these events; replay, resume, and OTel export are all consumers of them. The rule (ARCHITECTURE.md §2): if a behavior isn't in the trace, it's a bug — and any schema change requires a SCHEMA_VERSION bump plus a migration note.

LLM requests/responses are stored as plain dicts (the normalized ModelRequest/ ModelResponse dumps) rather than typed models: trace/ sits below providers/ in the dependency graph and must not import from it.

ContextCompaction

Bases: _Event

Compaction changes every subsequent request, so it must be traced.

Source code in src/regista/trace/events.py

class ContextCompaction(_Event):
    """Compaction changes every subsequent request, so it must be traced."""

    type: Literal["context.compaction"] = "context.compaction"
    tokens_before: int
    tokens_after: int
    summary: str
    dropped_messages: int

LlmResponse

Bases: _Event

The replay payload: the full normalized response, verbatim.

Source code in src/regista/trace/events.py

class LlmResponse(_Event):
    """The replay payload: the full normalized response, verbatim."""

    type: Literal["llm.response"] = "llm.response"
    turn: int
    response: dict[str, Any]
    usage: Usage
    cost_usd: float | None
    latency_ms: int
    replayed: bool = False

PermissionDecision

Bases: _Event

ask decisions record their resolution so deny paths replay identically.

Source code in src/regista/trace/events.py

class PermissionDecision(_Event):
    """``ask`` decisions record their resolution so deny paths replay identically."""

    type: Literal["permission.decision"] = "permission.decision"
    tool_use_id: str
    decision: Literal["allow", "deny", "ask"]
    resolution: Literal["allowed", "denied"] | None = None
    policy: str = ""
    reason: str = ""

SessionEnd

Bases: _Event

Summary only — replay never needs it, humans always read it first.

Source code in src/regista/trace/events.py

class SessionEnd(_Event):
    """Summary only — replay never needs it, humans always read it first."""

    type: Literal["session.end"] = "session.end"
    stop_reason: str
    turns: int
    usage: Usage
    cost_usd: float
    wall_time_ms: int
    final_output: str = ""

SessionStart

Bases: _Event

Everything needed to validate that a replay is configured like the original.

Source code in src/regista/trace/events.py

class SessionStart(_Event):
    """Everything needed to validate that a replay is configured like the original."""

    type: Literal["session.start"] = "session.start"
    task: str
    instructions: str
    model: str
    provider: str
    tool_schemas: list[dict[str, Any]]
    policy: str
    context: dict[str, Any] = {}
    """ContextConfig dump — replay needs it to compact at the same points."""
    regista_version: str
    replay_of: str | None = None

context = {} class-attribute instance-attribute

ContextConfig dump — replay needs it to compact at the same points.

canonical_hash(payload)

SHA-256 over a canonical JSON serialization (sorted keys, no whitespace).

This is the replay divergence detector: two requests are "the same" iff their canonical hashes match. Callers must exclude nondeterministic fields before hashing.

Source code in src/regista/trace/events.py

def canonical_hash(payload: Any) -> str:
    """SHA-256 over a canonical JSON serialization (sorted keys, no whitespace).

    This is the replay divergence detector: two requests are "the same" iff
    their canonical hashes match. Callers must exclude nondeterministic fields
    before hashing.
    """
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(canonical.encode()).hexdigest()

Load and inspect recorded traces.

Trace is the read-side twin of TraceWriter and the foundation replay builds on: it indexes LLM request/response pairs in seq order.

Trace

An in-memory view of one session's JSONL trace file.

Source code in src/regista/trace/reader.py

class Trace:
    """An in-memory view of one session's JSONL trace file."""

    def __init__(self, events: list[TraceEvent], path: Path | None = None) -> None:
        if not events:
            raise ValueError("trace contains no events")
        self.events = events
        self.path = path

    @classmethod
    def load(cls, path: Path | str) -> Trace:
        path = Path(path)
        events: list[TraceEvent] = []
        with path.open(encoding="utf-8") as f:
            for line_no, line in enumerate(f, start=1):
                if not line.strip():
                    continue
                event = event_adapter.validate_json(line)
                if event.schema_version > SCHEMA_VERSION:
                    raise ValueError(
                        f"{path}:{line_no}: trace schema v{event.schema_version} is newer "
                        f"than this regista (v{SCHEMA_VERSION}) — upgrade regista to read it"
                    )
                events.append(event)
        return cls(events, path=path)

    def __iter__(self) -> Iterator[TraceEvent]:
        return iter(self.events)

    def __len__(self) -> int:
        return len(self.events)

    @property
    def start(self) -> SessionStart:
        first = self.events[0]
        if not isinstance(first, SessionStart):
            raise ValueError("trace does not begin with session.start")
        return first

    @property
    def end(self) -> SessionEnd | None:
        """None if the session crashed before session.end was written."""
        last = self.events[-1]
        return last if isinstance(last, SessionEnd) else None

    def llm_calls(self) -> list[tuple[LlmRequest, LlmResponse]]:
        """Request/response pairs in seq order — the replay index."""
        requests = [e for e in self.events if isinstance(e, LlmRequest)]
        responses = [e for e in self.events if isinstance(e, LlmResponse)]
        # strict=False: a crash between request and response leaves one unpaired
        # trailing request, which is expected in a crashed trace
        return list(zip(requests, responses, strict=False))

    def tool_results(self) -> dict[str, ToolResult]:
        """tool_use_id → result, for stubbed-tool replay."""
        return {e.tool_use_id: e for e in self.events if isinstance(e, ToolResult)}

    def summary(self) -> TraceSummary:
        start, end = self.start, self.end
        return TraceSummary(
            session_id=start.session_id,
            task=start.task,
            model=start.model,
            events=len(self.events),
            turns=sum(1 for e in self.events if isinstance(e, LlmResponse)),
            tool_calls=len(self.tool_results()),
            cost_usd=end.cost_usd if end else None,
            stop_reason=end.stop_reason if end else None,
            replay_of=start.replay_of,
        )

end property

None if the session crashed before session.end was written.

llm_calls()

Request/response pairs in seq order — the replay index.

Source code in src/regista/trace/reader.py

def llm_calls(self) -> list[tuple[LlmRequest, LlmResponse]]:
    """Request/response pairs in seq order — the replay index."""
    requests = [e for e in self.events if isinstance(e, LlmRequest)]
    responses = [e for e in self.events if isinstance(e, LlmResponse)]
    # strict=False: a crash between request and response leaves one unpaired
    # trailing request, which is expected in a crashed trace
    return list(zip(requests, responses, strict=False))

tool_results()

tool_use_id → result, for stubbed-tool replay.

Source code in src/regista/trace/reader.py

def tool_results(self) -> dict[str, ToolResult]:
    """tool_use_id → result, for stubbed-tool replay."""
    return {e.tool_use_id: e for e in self.events if isinstance(e, ToolResult)}

OpenTelemetry export: one more consumer of the one trace.

Post-hoc by design — spans are built from recorded events with their recorded timestamps, so exporting yesterday's session (or a $0 replay of it) paints the same picture in Jaeger/Grafana as a live run would have. There is no second instrumentation path to drift out of sync with the JSONL log.

Span hierarchy: session → turn → llm / tool. LLM spans carry gen_ai.* semantic-convention attributes plus regista extras (request_hash, cost, replayed). Requires the otel extra: pip install "regista-harness[otel]".

export_trace(trace, *, tracer_provider=None)

Emit one recorded session as a span tree.

Spans go to tracer_provider (or the global one), so any configured exporter — OTLP, Jaeger, console — receives them.

Source code in src/regista/trace/otel.py

def export_trace(trace: Trace, *, tracer_provider: Any = None) -> None:
    """Emit one recorded session as a span tree.

    Spans go to ``tracer_provider`` (or the global one), so any configured
    exporter — OTLP, Jaeger, console — receives them.
    """
    otel_trace = _otel()
    provider = tracer_provider or otel_trace.get_tracer_provider()
    tracer = provider.get_tracer("regista")

    start, end = trace.start, trace.end
    session_span = tracer.start_span(
        f"agent.session {start.session_id}",
        start_time=_ns(trace.events[0].ts),
        attributes={
            "regista.session_id": start.session_id,
            "regista.task": start.task,
            "regista.policy": start.policy,
            "regista.version": start.regista_version,
            "regista.replay_of": start.replay_of or "",
            "gen_ai.system": start.provider,
            "gen_ai.request.model": start.model,
        },
    )
    if end is not None:
        session_span.set_attributes(
            {
                "regista.stop_reason": end.stop_reason,
                "regista.turns": end.turns,
                "regista.cost_usd": end.cost_usd,
                "gen_ai.usage.input_tokens": end.usage.input_tokens,
                "gen_ai.usage.output_tokens": end.usage.output_tokens,
            }
        )
    if end is None or end.stop_reason == "error":
        session_span.set_status(otel_trace.StatusCode.ERROR)
    session_ctx = otel_trace.set_span_in_context(session_span)

    # group events by turn: llm events carry their turn; tool/permission/error
    # events belong to the most recent llm turn seen
    turns: dict[int, list[Any]] = {}
    current_turn = 0
    for event in trace.events:
        if isinstance(event, (LlmRequest, LlmResponse)):
            current_turn = event.turn
        span_worthy = (
            LlmRequest,
            LlmResponse,
            ToolCall,
            ToolResult,
            PermissionDecision,
            ErrorEvent,
            ContextCompaction,
        )
        if isinstance(event, span_worthy):
            turns.setdefault(current_turn, []).append(event)

    for turn_number, events in sorted(turns.items()):
        turn_start, turn_end = _ns(events[0].ts), _ns(events[-1].ts)
        turn_span = tracer.start_span(
            f"turn {turn_number}", context=session_ctx, start_time=turn_start
        )
        turn_ctx = otel_trace.set_span_in_context(turn_span)
        _export_turn_events(otel_trace, tracer, turn_ctx, events, start.model)
        turn_span.end(end_time=turn_end)

    session_span.end(end_time=_ns(trace.events[-1].ts))