Harness

Define the task environment and run the agent through it. Context managers decide what the policy sees, branch points decide where search may continue, and workspaces hold durable task data.

Agent

Run a policy in a task. Each decision prepares the context, generates an action, applies it to the environment, and records the turn with its exact prompt tokens and behavior log probabilities.

from focus import Agent

agent = Agent(
    policy=policy,
    env=env,
    system_prompt="Solve the task using the available tools.",
    max_steps=32,
)

async with agent:
    root = await agent.start("Your task")
    result = await agent.run(root)

This agent keeps the transcript with AppendOnly and makes every decision boundary available to search. After 32 steps, including compaction steps, it makes one final-answer attempt; set finish_prompt=None to stop without that attempt. Use context=Segments(256) for exact token continuation or context=Summarize(...) for summaries. branch_points selects candidate checkpoints, and execution configures shared capacity and persistence. The agent chooses an action parser from the policy format and the action kind declared by the context manager or environment. Pass parser= to override that choice. Derived agents share the same policy and execution services.

async with agent starts its execution services and closes them on exit, cancelling and awaiting outstanding requests. The caller closes policy clients and checkpoint storage separately. If several agents share services, use one outer lifecycle scope for all their work. Run derived agents inside that scope; use Collector(..., own=False) when a collector borrows it.

Definition

class Agent[S, A]

Agent.init

def __init__(self, *, policy: Policy[A], env: Env[S, A], system_prompt: str, context: ContextManager[S] | None=None, sampling_params: SamplingParams | None=None, max_steps: int=32, finish_prompt: str | Message | EllipsisType | None=..., tools: Sequence[dict[str, Any]] | None=None, branch_points: BranchPoints | None=None, context_limit: int | None=None, on_compaction_failure: Literal['raise', 'end']='raise', execution: Execution | None=None, actor: bool=True, parser: ActionParser[A] | None=None) -> None

Agent.aenter

async def __aenter__(self) -> Agent[S, A]

Agent.aexit

async def __aexit__(self, *exc: object) -> None

Agent.aclose

async def aclose(self) -> None

Close this agent’s execution services, cancelling and awaiting outstanding requests. This also affects every agent sharing those services. Policy clients and checkpoint storage remain open. The async context manager calls this on exit.

Agent.services

@services.setter
def services(self, services: ExecutionServices) -> None

The execution services that run this agent’s model requests. Agents built with the same Execution share capacity.

Agent.checkpointer

@property
def checkpointer(self) -> Checkpointer | None

The checkpointer from Execution, or None when states are kept only in memory.

Agent.generate_timeout

@property
def generate_timeout(self) -> float | None

Seconds to wait for one generation, from Execution.

Agent.save_checkpoint

def save_checkpoint(self, rs: RolloutState[S]) -> RolloutState[S]

Save a complete rollout state when a checkpointer is configured, whether or not search can branch there. Call this after editing a saved state by hand. If the state cannot be snapshotted, the previous recoverable checkpoint remains available.

Agent.restore

def restore(self, checkpoint_id: str) -> RolloutState[S]

Restore a saved state on the same branch, using compatible agent components. Stop the previous execution first. The saved resource allowance is restored with the state; service usage still counts failed attempts. Call fork() to start an independent branch.

Agent.restore_latest

def restore_latest(self, episode_id: str, branch_id: str, invocation_id: str='root') -> RolloutState[S]

Restore the latest intact snapshot for a branch and invocation.

Agent.env

@property
def env(self) -> Env[S, A]

The task environment.

Agent.policy

@property
def policy(self) -> Policy[A]

The policy used for task generation, also used by derived agents.

Agent.parser

@property
def parser(self) -> ActionParser[A]

The parser that turns this agent’s replies into actions: parser= if given, otherwise the policy format’s own, adapted to the declared action kind.

Agent.tools

@property
def tools(self) -> tuple[dict[str, Any], ...]

A copy of the tool schemas shown in the agent’s prompts, including any the context manager adds.

Agent.derive

def derive[T](self, *, env: Env[T, A], system_prompt: str, max_steps: int, finish_prompt: str | None, sampling_params: SamplingParams | None=None, tools: Sequence[dict[str, Any]] | None=None) -> Agent[T, A]

Create an agent for another environment and prompt that shares this agent’s policy and execution services. Summarize uses it to run the compaction task with the same model. The derived agent keeps the transcript, offers the new environment’s tools, and uses this agent’s sampling parameters unless overridden.

Agent.root

async def root(self, seed: Sequence[Message], *, workspace: Workspace | None=None) -> RolloutState[S]

Create a root checkpoint from a seed conversation. The environment validates the workspace and resets task state, and the context manager initializes before any generation.

Agent.start

async def start(self, user_prompt: str, *, workspace: Workspace | None=None) -> RolloutState[S]

Create a root checkpoint from a task prompt and optional workspace. Adds the system prompt and the workspace’s usage guidance. Continue with advance(), run(), or iter_run().

Agent.advance

async def advance(self, rs: RolloutState[S], *, sampling_params: SamplingParams | None=None) -> RolloutState[S]

Take one decision: honor an environment finish request, compact if the context manager requests it, then check whether a non-compacting context is near its limit. Otherwise generate and apply an action. The returned state records the generation; the input stays unchanged. finish_prompt=None ends a limited rollout without an extra generation.

Agent.finish

async def finish(self, rs: RolloutState[S], *, sampling_params: SamplingParams | None=None) -> RolloutState[S]

End the rollout with one final-answer turn. No tool runs; the environment’s finish() reads and scores the answer. With finish_prompt=None, mark the rollout truncated without generating.

Agent.compact

async def compact(self, rs: RolloutState[S]) -> RolloutState[S]

Fold the live context now, even if the context manager’s trigger has not fired, and record the fold turns and the new context version.

Agent.max_steps

@property
def max_steps(self) -> int

Decisions per continuation before the final-answer turn; -1 is unlimited.

Agent.iter_run

async def iter_run(self, rs: RolloutState[S], *, max_steps: int | None=None, sampling_params: SamplingParams | None=None) -> AsyncIterator[RolloutState[S]]

Run to completion, exposing selected continuations and the endpoint.

A new local decision consumes one step, including a fold-only advance. Finalization is outside that budget and happens once. Every yielded checkpoint has an independent snapshot.

Agent.run

async def run(self, rs: RolloutState[S], *, max_steps: int | None=None, sampling_params: SamplingParams | None=None) -> RolloutState[S]

Consume the shared loop and return its terminal or truncated endpoint.

Agent.run_from

async def run_from(self, prompt: str, *, workspace: Workspace | None=None, max_steps: int | None=None, sampling_params: SamplingParams | None=None) -> RolloutState[S]

Fresh episode: start then run. To branch, run(rs).

Env

Define the task’s rules. reset creates an initial state; step applies one parsed action and returns the next state, observations, rewards, and completion status. Return a new state so forked branches remain independent.

from focus import Env

Definition

class Env[S, A](ABC)

Fields

action: Literal['text', 'tool'] | None = 'text'

Env.reset

Required implementation.

@abstractmethod
async def reset(self) -> S

Return the initial state for a fresh episode.

Env.step

Required implementation.

@abstractmethod
async def step(self, state: S, action: A, *, workspace: Workspace | None=None) -> StepResult[S]

Apply the parsed action and return a StepResult with the next state, observations, and rewards. Return a new state to keep forks independent. Tools receive the current branch’s workspace through this call.

Env.validate_workspace

def validate_workspace(self, workspace: Workspace) -> None

Check task-specific workspace requirements before an episode starts. Raise an error for an incompatible workspace; the default accepts any workspace. The agent calls this before resetting the environment or generating tokens.

Env.finish

async def finish(self, state: S, action: A) -> StepResult[S]

Read or score a final action after the agent’s step budget is exhausted. This ends the episode without executing tools or other task operations. The default keeps the task state unchanged; override it to extract and score the final answer.

Env.should_finish

def should_finish(self, rollout: AnyRollout) -> bool

Whether to stop working and take the final-answer turn now. Called before each decision; the default never stops early. Override it for task limits such as a repeated identical request.

Env.finish_tools

def finish_tools(self, tools: Sequence[dict[str, Any]]) -> Sequence[dict[str, Any]]

The tool schemas offered on the final-answer turn. The default offers all of them; return only the submission tool to require an answer.

Env.on_compaction_failure

def on_compaction_failure(self, state: S, reasons: tuple[str, ...]) -> S

The terminal task state when a rejected compaction ends the episode, which happens when the agent is built with on_compaction_failure="end". The episode ends with zero outcome reward; override this to clear a partial answer or record the reasons.

Execution

How an agent’s model calls run, persist, and recover: the shared execution services, an optional checkpointer, retry rules, a resource budget, and a generation timeout.

from focus import Agent, Execution

execution = Execution(retries=3, generate_timeout=120.0)
agent = Agent(
    policy=policy,
    env=env,
    system_prompt="Solve the task.",
    execution=execution,
)

This agent retries a failed generation up to three times and abandons a generation after two minutes. Pass the same Execution to several agents, or to agents built for different policy snapshots, so their requests share one capacity limit. Add a checkpointer to persist rollout states for recovery.

Definition

@dataclass
class Execution

Fields

services: ExecutionServices = field(default_factory=ExecutionServices)
checkpointer: Checkpointer | None = None
retries: int = 2
retry_on: tuple[type[Exception], ...] = (ConnectionError, TimeoutError)
resource_budget: ResourceBudget = field(default_factory=ResourceBudget)
generate_timeout: float | None = None

StepResult

Describe the result of one environment action: the next task state, reply messages, reward components, and whether the episode ended.

from focus import StepResult

result = StepResult(
    next_state=state, reward_outcome=1.0, done=True
)

This result ends the episode with outcome reward 1.0 and the supplied final state. Add messages when an action produces observations for the agent.

Definition

@dataclass(frozen=True)
class StepResult[S_co]

Fields

next_state: S_co
messages: tuple[Message, ...] = field(default_factory=tuple)
reward_outcome: float = 0.0
reward_step: float = 0.0
done: bool = False

StepResult.reward

@property
def reward(self) -> float

The raw sum of reward_outcome and reward_step. Training applies the weights in RewardConfig to the separate components.

ToolEnv

Build an environment that dispatches tool calls. Implement initial_state and the task’s reward or submission rules. Each tool defines its arguments, behavior, and workspace requirements.

from focus import ToolEnv

Definition

class ToolEnv(Env[S, ToolTurn], ABC)

ToolEnv.init

def __init__(self, tools: Sequence[Tool], *, submit: SubmitTool | None=SUBMIT) -> None

ToolEnv.validate_workspace

def validate_workspace(self, workspace: Workspace) -> None

Run each tool’s workspace validation hook when present. Structural third-party tools without this hook remain supported. Override this method and call super() to add environment-wide checks.

ToolEnv.reset

@final
async def reset(self) -> S

Create the initial state and validate the fields that tool dispatch updates.

ToolEnv.initial_state

Required implementation.

@abstractmethod
async def initial_state(self) -> S

Build fresh task state.

ToolEnv.tools

@property
def tools(self) -> tuple[Tool, ...]

The available tools, including the submission tool when configured.

ToolEnv.schemas

@property
def schemas(self) -> list[dict[str, Any]]

The tool schemas supplied to the policy’s chat template or native API.

ToolEnv.step

async def step(self, state: S, turn: ToolTurn, *, workspace: Workspace | None=None) -> StepResult[S]

ToolEnv.finish

async def finish(self, state: S, turn: ToolTurn) -> StepResult[S]

Read the final submission without executing tools or retrying errors.

ToolEnv.reward

def reward(self, state: S, turn: ToolTurn) -> float

Score a valid submission (or a text ending when submit is disabled).

ToolEnv.answer_field

@property
def answer_field(self) -> str

The argument name used for the final answer in the submission schema.

ToolEnv.on_submit

def on_submit(self, state: S, turn: ToolTurn, payload: dict[str, Any]) -> S

Record a valid answer. Text is used only with submit disabled.

ToolEnv.on_tool_calls

def on_tool_calls(self, state: S, turn: ToolTurn) -> S

Decrease the optional round allowance by the number of dispatched tool calls.

ToolError

Signal an expected tool failure that can be reported to the agent as an observation, such as invalid arguments or an unavailable resource.

from focus import ToolError

Definition

class ToolError(Exception)

tool

Turn a function into a tool. The tool’s name, description, and argument schema come from the function’s signature and docstring.

from focus import tool

@tool
async def lookup(term: str, limit: int = 5) -> str:
    """Look up a term in the glossary.

    Args:
        term: The word to define.
        limit: How many definitions to return.
    """
    return term

The model sees a lookup tool with a required string term and an optional integer limit. Arguments are validated before the call; a wrong type or unknown argument becomes a ToolError the model can read. Annotate a parameter workspace to receive the branch’s workspace; it is hidden from the model. Return a string or any JSON-serializable value.

Definition

def tool(fn: Callable[..., Any] | None=None, /, *, name: str | None=None, description: str | None=None) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]

BaseTool

Define a tool the agent can call. Subclass it with a name, description, parameter schema, and call implementation; ToolEnv handles dispatch.

from focus.extend import BaseTool

Definition

class BaseTool(ABC)

Fields

name: str = ''
description: str = ''
parameters: dict[str, Any] = {}

BaseTool.function

@property
def function(self) -> dict[str, Any]

BaseTool.validate_workspace

def validate_workspace(self, workspace: Workspace) -> None

Check the workspace capabilities this tool needs. ToolEnv calls this hook before generation; the default accepts any workspace. Keep branch-specific data in the supplied workspace.

BaseTool.verify_args

def verify_args(self, params: str | dict[str, Any]) -> dict[str, Any]

Parse JSON arguments, check required fields, and convert values to their declared types. Validate value ranges and task-specific constraints in your tool.

BaseTool.call

Required implementation.

@abstractmethod
def call(self, params: str | dict[str, Any], **kwargs: Any) -> str | Awaitable[str]

Checkpointer

Define persistence for complete rollout states. Saving records a snapshot; loading returns an isolated state suitable for recovery or branching.

from focus.extend import Checkpointer

from focus.extend import MemoryCheckpointer

checkpointer = MemoryCheckpointer()

Use the in-memory implementation to save and recover isolated states within a process. A custom store implements _put(), _get(), info(), and list(); the base class handles serialization and validation. Pass it through Execution(checkpointer=...). BranchPoints separately selects where search may continue.

Definition

class Checkpointer(ABC)

Checkpointer._put

Required implementation.

@abstractmethod
def _put(self, stored: _Stored) -> None

Atomically store this snapshot and its metadata. The base class serializes the rollout and computes the digest before calling this backend hook.

Checkpointer._get

Required implementation.

@abstractmethod
def _get(self, checkpoint_id: str) -> _Stored

Retrieve the stored snapshot by checkpoint ID. The base class verifies its digest and deserializes an isolated rollout after this hook returns.

Checkpointer.info

Required implementation.

@abstractmethod
def info(self, checkpoint_id: str) -> CheckpointInfo

Read checkpoint metadata without decoding the rollout payload.

Checkpointer.list

Required implementation.

@abstractmethod
def list(self, episode_id: str, branch_id: str, invocation_id: str='root') -> tuple[CheckpointInfo, ...]

Return checkpoint metadata in reverse commit order for the requested episode, branch, and invocation.

Checkpointer.put

def put[S](self, rollout: RolloutState[S]) -> RolloutState[S]

Atomically save a new snapshot; the input’s checkpoint ID is its parent.

Checkpointer.get

def get(self, checkpoint_id: str) -> AnyRollout

Load an independent snapshot. Explicit IDs never silently fall back.

Checkpointer.latest

def latest(self, episode_id: str, branch_id: str, invocation_id: str='root') -> AnyRollout

Load the last intact snapshot, skipping corrupt records in this scope.

DEFAULT_FINISH_PROMPT

The agent’s default instruction for the final-answer turn when a rollout runs out of steps.

from focus.extend import DEFAULT_FINISH_PROMPT

Definition

DEFAULT_FINISH_PROMPT = 'You are out of steps: this is your last turn, and no further work tools will run. Give your best-supported final answer from what you have already gathered, in the form the task requires. If the task provides a final-answer tool, call that tool; otherwise answer directly in text.'

FunctionTool

The tool @tool creates: a function with a JSON schema derived from its signature and docstring, whose arguments are validated before each call.

from focus.extend import FunctionTool

Definition

class FunctionTool(BaseTool)

FunctionTool.init

def __init__(self, fn: Callable[..., Any], *, name: str | None=None, description: str | None=None) -> None

FunctionTool.call

async def call(self, params: str | dict[str, Any], *, workspace: Any=None, **kwargs: Any) -> str

Validate the arguments, pass the branch’s workspace if the function asks for it, and return the result as text.

InMemoryWorkspace

Store named text entries for each branch. The entries survive context compaction, and forks receive independent copies. The agent reads and writes them through the storage tool.

from focus.extend import InMemoryWorkspace

workspace = InMemoryWorkspace()
workspace.write(
    "notes/evidence", "Source doc-17 identifies the museum founder."
)

This example saves a note for later use. Pass workspace to Collector.collect() or Agent.start(). To recover after a restart, use a checkpointer to save the complete rollout.

Definition

class InMemoryWorkspace(Workspace)

InMemoryWorkspace.init

def __init__(self) -> None

InMemoryWorkspace.tools

def tools(self) -> Sequence[Tool]

InMemoryWorkspace.guidance

def guidance(self) -> str

InMemoryWorkspace.compaction_guidance

def compaction_guidance(self) -> str

InMemoryWorkspace.fork

def fork(self) -> InMemoryWorkspace

InMemoryWorkspace.read

def read(self, path: str) -> str

InMemoryWorkspace.write

def write(self, path: str, content: str) -> None

InMemoryWorkspace.delete

def delete(self, path: str) -> None

InMemoryWorkspace.paths

def paths(self) -> tuple[str, ...]

InMemoryWorkspace.exists

def exists(self, path: str) -> bool

InMemoryWorkspace.clear

def clear(self) -> None

InMemoryWorkspace.update

def update(self, entries: Mapping[str, str]) -> None

InMemoryWorkspace.init

def init(self, entries: Mapping[str, str] | None=None) -> None

InMemoryWorkspace.snapshot

def snapshot(self) -> Mapping[str, str]

InMemoryWorkspace.serialize

def serialize(self) -> dict[str, str]

InMemoryWorkspace.deserialize

@classmethod
def deserialize(cls, state: Mapping[str, str]) -> InMemoryWorkspace

InMemoryWorkspace.contains

def __contains__(self, path: object) -> bool

InMemoryWorkspace.len

def __len__(self) -> int

MemoryCheckpointer

Store serialized rollout snapshots in memory and load isolated copies within the process.

from focus.extend import MemoryCheckpointer

Definition

class MemoryCheckpointer(Checkpointer)

MemoryCheckpointer.init

def __init__(self) -> None

MemoryCheckpointer.info

def info(self, checkpoint_id: str) -> CheckpointInfo

MemoryCheckpointer.list

def list(self, episode_id: str, branch_id: str, invocation_id: str='root') -> tuple[CheckpointInfo, ...]

NullWorkspace

Run without a scratch store or storage tools. Information carried through compaction lives in the retained context.

from focus.extend import NullWorkspace

workspace = NullWorkspace()

Use for tasks that need only the conversation and environment state.

Definition

class NullWorkspace(Workspace)

NullWorkspace.fork

def fork(self) -> NullWorkspace

NullWorkspace.folded_guidance

def folded_guidance(self) -> str

NullWorkspace.compaction_guidance

def compaction_guidance(self) -> str

PackedSequence

Hold a prefix-stable token sequence, its training mask, and the behavior log probability of each generated token.

from focus.extend import PackedSequence

Definition

@dataclass(frozen=True)
class PackedSequence

Fields

tokens: tuple[TokenId, ...]
trainable: tuple[bool, ...]
logprobs: tuple[float, ...] | None = None
exact: bool = True
tag: TurnTag = 'task'
spans: tuple[tuple[int, int, int], ...] = ()
retain_for_training: bool = False
actor: bool = True
invocation_id: str = 'root'
format_id: str = ''

RolloutState

Hold a resumable agent state: live conversation, task state, workspace, generation history, and remaining resources. Agent creates and advances it; fork creates an independent continuation.

from focus.extend import RolloutState

Definition

@dataclass(frozen=True)
class RolloutState[S]

Fields

messages: tuple[Message, ...]
seed: tuple[Message, ...]
state: S
workspace: Workspace
turns: tuple[Turn[S], ...] = field(default_factory=tuple)
truncated: bool = False
checkpoint_version: int = field(default=CHECKPOINT_SCHEMA, kw_only=True)
checkpoint_at: int = 0
context_version: int = 0
effects: tuple[Effect, ...] = ()
invocation_id: str = 'root'
episode_id: str = field(default_factory=lambda: uuid4().hex)
branch_id: str = field(default_factory=lambda: uuid4().hex)
checkpoint_id: str | None = None
execution: dict[str, Any] = field(default_factory=dict)
history: tuple[Message, ...] = ()
context_refs: tuple[int | None, ...] = ()
advances: int = 0
steps_remaining: int | None = None
snapshot_available: bool = True
budget: ResourceBudget = field(default_factory=ResourceBudget)

RolloutState.forkable

@property
def forkable(self) -> bool

Whether this state is live, restorable, and eligible for branching.

RolloutState.tagged

def tagged(self, tags: Collection[TurnTag] | None=TASK) -> tuple[Turn[S], ...]

The turns carrying any of tags (None means every turn).

RolloutState.transitions

@cached_property
def transitions(self) -> tuple[StepResult[S], ...]

The environment transitions recorded so far, in order. Compaction turns have no environment transition.

RolloutState.folds

@cached_property
def folds(self) -> int

The number of consecutive groups of compaction turns in the history. Two compactions with no task turn between them count as one group. A compaction that generates no tokens still records a marker.

RolloutState.boundaries

def boundaries(self) -> tuple[int, ...]

Indices of the first turns after each consecutive group of compaction turns. These mark where generation resumes from a rewritten context.

RolloutState.reward

@property
def reward(self) -> float

The sum of outcome and step rewards across environment transitions. This raw total is used for ranking; RewardConfig.configured_return() applies the training weights and discount.

RolloutState.reward_outcome

@property
def reward_outcome(self) -> float

The sum of task-outcome rewards recorded at episode endings.

RolloutState.reward_step

@property
def reward_step(self) -> float

The sum of step rewards across environment transitions.

RolloutState.done

@property
def done(self) -> bool

Whether the latest environment transition ended the episode. A subsequent compaction turn leaves this status unchanged.

RolloutState.turns_taken

@property
def turns_taken(self) -> int

The number of environment transitions so far. Search uses changes in this count to charge task-transition budgets. Compaction consumes an agent step but does not increase this count.

RolloutState.edge

def edge(self, parent: RolloutState[S]) -> tuple[Turn[S], ...]

Return the turns added after parent. For example, after resuming a checkpoint and generating two segments, state.edge(checkpoint) returns those two turns with their token records and environment transitions. Raises ValueError if the parent’s turns are not a prefix of this state’s history.

RolloutState.generated_since

def generated_since(self, parent: RolloutState[S], *, tags: Collection[TurnTag] | None=TASK) -> list[float]

Return the behavior log probabilities of tokens generated after parent, in order, filtered by turn tags. Use task tags to inspect task-generation uncertainty; turns without recorded probabilities contribute no values.

RolloutState.owner

@property
def owner(self) -> Owner

The episode, branch, and invocation scope used to account for this rollout’s requests and resource budget.

RolloutState.check_continuation

def check_continuation(self) -> None

Validate the checkpoint format required to resume generation. Raises ValueError if the checkpoint does not satisfy that format.

RolloutState.snapshot

def snapshot(self) -> RolloutState[S]

Copy the branch’s mutable state and workspace while sharing immutable generation records. Use a snapshot to preserve the current checkpoint before advancing it. This method can copy states that are not eligible for branching.

RolloutState.fork

def fork(self) -> RolloutState[S]

Create an independent branch with a new identity and a copy of the remaining resource allowance. Raises ValueError when the state is not forkable.

RoundBudget

Add per-round tool-call counters to a task state. ToolEnv spends calls_remaining as tools run; StateCounter refills it after compaction.

from focus.extend import RoundBudget

budget = RoundBudget(calls_remaining=5, calls_max=5)

This example starts a round with five tool calls. Include these fields in the state returned by Env.reset() when compaction is triggered by the tool-call allowance.

Definition

@dataclass(frozen=True)
class RoundBudget

Fields

calls_remaining: int = 0
calls_max: int = 0

SQLiteCheckpointer

Store complete rollout snapshots in a local SQLite database for recovery across processes.

from focus.extend import SQLiteCheckpointer

Definition

class SQLiteCheckpointer(Checkpointer)

SQLiteCheckpointer.init

def __init__(self, path: str | Path) -> None

SQLiteCheckpointer.info

def info(self, checkpoint_id: str) -> CheckpointInfo

SQLiteCheckpointer.list

def list(self, episode_id: str, branch_id: str, invocation_id: str='root') -> tuple[CheckpointInfo, ...]

SUBMIT

The built-in submit tool instance used by ToolEnv to accept a final answer and end the task. Customize the environment’s submission handling to score that answer.

from focus.extend import SUBMIT

Definition

SUBMIT = SubmitTool()

StorageTool

Expose workspace file operations to the agent. Its workspace must support reading, writing, deleting, and listing paths.

from focus.extend import StorageTool

Definition

class StorageTool(BaseTool)

StorageTool.init

def __init__(self, workspace: InMemoryWorkspace | None=None) -> None

StorageTool.validate_workspace

def validate_workspace(self, workspace: Workspace) -> None

Require the key-value operations used by the storage tool.

StorageTool.call

def call(self, params: str | dict[str, Any], *, workspace: InMemoryWorkspace | None=None, **_: object) -> str

SubmitFields

Add an optional answer field to a task state dataclass. Tool environments can store the agent’s submitted answer here for final scoring.

from focus.extend import SubmitFields

Definition

@dataclass(frozen=True)
class SubmitFields

Fields

answer: str | None = None

SubmitTool

Describe the terminal submission action and its argument schema; the environment intercepts it to finish the task.

from focus.extend import SubmitTool

Definition

class SubmitTool(BaseTool)

SubmitTool.init

def __init__(self, *, name: str='submit', description: str | None=None, parameters: dict[str, Any] | None=None, field: str='answer') -> None

SubmitTool.verify_args

def verify_args(self, params: str | dict[str, Any]) -> dict[str, Any]

Validate declared string fields as well as required keys.

SubmitTool.call

def call(self, params: str | dict[str, Any], **kwargs: object) -> str

Tool

Specify the name, schema, and callable behavior required of an agent tool. Structural implementations can satisfy this protocol without subclassing BaseTool.

from focus.extend import Tool

Definition

class Tool(Protocol)

Fields

name: str
description: str
parameters: dict[str, Any]

Tool.call

def call(self, params: str | dict[str, Any], **kwargs: Any) -> str | Awaitable[str]

Turn

Record one generation and its associated task transition or context-management operation. Rollout histories use these records to recover token spans and rewards.

from focus.extend import Turn

Definition

@dataclass(frozen=True)
class Turn[S_co]

Fields

prefix: tuple[TokenId, ...]
tokens: tuple[TokenId, ...] = ()
logprobs: tuple[float, ...] | None = None
exact: bool = True
transition: StepResult[S_co] | None = None
tag: TurnTag = 'task'
compaction_penalty: float = 0.0
compaction_issue: str = ''
invocation_id: str = 'root'
format_id: str = ''
context_version: int = 0
actor: bool = True
sampling_params: SamplingParams | None = None
reward_components: dict[str, float] = field(default_factory=dict)

Workspace

Define the agent’s branch-local scratch storage, its tools, and how it is copied or restored. Implement this interface to add a storage backend.

from focus.extend import Workspace

Definition

class Workspace(ABC)

Workspace.tools

def tools(self) -> Sequence[Tool]

Return the tools the agent can use to access this workspace. The default returns none.

Workspace.compaction_tools

def compaction_tools(self) -> Sequence[Tool]

Return the workspace tools available during compaction. Defaults to the same tools used by the agent.

Workspace.guidance

def guidance(self) -> str

Explain how the agent should use the workspace. The agent adds this text to the system prompt; an empty string adds no guidance.

Workspace.folded_guidance

def folded_guidance(self) -> str

Explain how to use memory after compaction. Defaults to guidance(); override it when the available memory changes after a summary.

Workspace.compaction_guidance

def compaction_guidance(self) -> str

Provide instructions for using the workspace during compaction. An empty string adds no instructions.

Workspace.snapshot

def snapshot(self) -> Mapping[str, str]

Return inspectable workspace contents for metrics and debugging. The default is empty. The model accesses stored data through the workspace tools.

Workspace.fork

Required implementation.

@abstractmethod
def fork(self) -> Workspace

Create an independent copy of the workspace with the same initial contents.

AnyRollout

Refer to a RolloutState without fixing its environment-state and action type parameters.

from focus.harness import AnyRollout

Definition

type AnyRollout = 'RolloutState[Any]'

CheckpointInfo

Describe a stored checkpoint’s identity, ordering, and snapshot metadata.

from focus.harness import CheckpointInfo

Definition

@dataclass(frozen=True)
class CheckpointInfo

Fields

id: str
parent_id: str | None
episode_id: str
branch_id: str
invocation_id: str
format_version: int = 1

CorruptCheckpoint

Report a saved snapshot that failed integrity verification or decoding.

from focus.harness import CorruptCheckpoint

Definition

class CorruptCheckpoint(ValueError)

TASK

Identify ordinary task-generation turns in a rollout.

from focus.harness import TASK

Definition

TASK: tuple[TurnTag, ...] = ('task',)

TurnTag

Classify a generation turn by its role in the task or context-management process.

from focus.harness import TurnTag

Definition

type TurnTag = Literal['task', 'fold']

WorkspaceError

Report a failed operation on a branch’s workspace.

from focus.harness import WorkspaceError

Definition

class WorkspaceError(Exception)

dispatch

Execute one parsed tool call and return the text observation supplied to the model.

from focus.harness import dispatch

Definition

async def dispatch(tools: Mapping[str, Tool], call: ToolCall, **context: object) -> str

flatten_tool_calls

Render tool-call blocks as inert text for transcript inspection and summarization.

from focus.harness import flatten_tool_calls

Definition

def flatten_tool_calls(content: str) -> str

format_transcript

Render messages as a role-labeled transcript with tool calls flattened to text.

from focus.harness import format_transcript

Definition

def format_transcript(messages: Sequence[Message], *, sep: str='\n') -> str

logprob

Sum the behavior log probabilities of the generated tokens in a sequence of turns. Raises an error if a turn has no recorded probabilities.

from focus.harness import logprob

Definition

def logprob(turns: Sequence[Turn[object]]) -> float

_Stored

Carry the serialized checkpoint bytes, integrity digest, and CheckpointInfo passed between Checkpointer and its storage backend. This record is the input to _put() and the result of _get().

from focus.harness.persistence import _Stored

Definition

@dataclass(frozen=True)
class _Stored

Fields

info: CheckpointInfo
payload: bytes
digest: str

regions

Pack a sequence of turns into training token regions. Context rewrites create new regions because the next prompt starts from a different token prefix.

from focus.harness import regions

Definition

def regions(turns: Sequence[Turn[object]]) -> tuple[PackedSequence, ...]

tool_schema

Build the function schema that a chat template exposes to the model for a tool.

from focus.harness import tool_schema

Definition

def tool_schema(tool: Tool) -> dict[str, Any]

DEFAULT_DESCRIPTION

Provide the default model-facing description of the final submission tool.

from focus.harness.tools.submission import DEFAULT_DESCRIPTION

Definition

DEFAULT_DESCRIPTION = 'Submit your final answer. This ends the task: no tool call after it will run, so ' + 'call it only once the answer is settled.'

trainable_logprobs

Read every generated token’s behavior log probability, in generation order.

from focus.harness import trainable_logprobs

Definition

def trainable_logprobs(turns: Sequence[Turn[object]]) -> list[float]

Effect

Record a committed task or context operation and the generation that requested it. Checkpoint policies inspect these effects to choose resumable boundaries.

from focus.harness.turn import Effect

Definition

@dataclass(frozen=True)
class Effect

Fields

kind: str
turn_index: int
context_before: int = 0
context_after: int = 0
detail: str = ''
invocation_id: str = 'root'
turn_end: int = -1