Context

Choose what the policy sees at each step. ContextManager owns prompt preparation and context changes; this module supplies transcript retention, exact token continuation, summarization, and their triggers.

ContextManager

Decide what the policy sees. A context manager renders each prompt and, when its trigger fires, compacts the live context. The agent records each compaction’s turns, effect, and context version.

from focus.context import Compaction, CompactionResult, ContextManager, PromptTokens

class KeepRecent(ContextManager):
    trigger = PromptTokens(4096)

    async def compact(self, compaction: Compaction) -> CompactionResult:
        recent = compaction.messages[-4:]
        return CompactionResult((*compaction.seed, *recent), compaction.state)

context = KeepRecent()

This manager restarts from the system prompt and task, plus the last four messages, once the prompt exceeds 4,096 tokens. The agent records the fold and advances the context version. Context managers are shared across branches; keep branch-specific data in the rollout state. A manager can also set the agent’s defaults: max_tokens caps each request, final_answer=False ends a rollout out of steps without a final-answer turn, branch_points() chooses where search may continue, action declares the action kind its decisions need, and check_policy() refuses an incompatible policy at construction. Segments sets all five.

Definition

class ContextManager[S]

Fields

trigger: Trigger[S] | None = None
max_tokens: int | None = None
final_answer: bool = True
action: ActionKind | None = None

ContextManager.manages_context

@property
def manages_context(self) -> bool

Whether this manager ever changes the live context. AppendOnly returns False.

ContextManager.branch_points

def branch_points(self) -> BranchPoints

Where search may continue when the agent is given no branch_points. The default is SparseCheckpoints(); Segments returns EveryDecision().

ContextManager.check_policy

def check_policy(self, policy: Policy[Any]) -> None

Raise if this manager cannot render prompts for policy. The agent calls it once, at construction.

ContextManager.should_compact

def should_compact(self, rs: AnyRollout, prompt: Sequence[int]) -> bool

Whether to compact before the next decision. The default asks trigger.

ContextManager.compact

async def compact(self, compaction: Compaction[S]) -> CompactionResult[S]

Return the compacted context and any turns generated while producing it.

ContextManager.prepare

def prepare(self, rs: AnyRollout, agent: Agent[Any, Any], *, tools: Sequence[dict[str, Any]] | None=None) -> PreparedPrompt

Render the prompt for the next generation. The default calls the policy’s prepare() with the conversation and tool schemas.

ContextManager.record_reply

def record_reply(self, result: GenerateResult, messages: list[Message], agent: Agent[Any, Any]) -> None

Append a model reply to the branch’s messages. The default delegates to the policy.

ContextManager.schemas

def schemas(self) -> tuple[dict[str, Any], ...]

Schemas for context tools the model may call. The default offers none.

ContextManager.initialize

async def initialize(self, rs: AnyRollout, agent: Agent[Any, Any]) -> AnyRollout

Prepare a new rollout’s state for this manager, for example by checking that the task state carries the counters the trigger reads.

ContextManager.apply

async def apply(self, rs: AnyRollout, action: Any, agent: Agent[Any, Any], *, ending: bool) -> Applied

Apply a recorded action and return the updated rollout and its transition. The default delegates to the environment’s step(), or finish() on the final-answer turn.

AppendOnly

Keep the full context between requests: the policy sees the whole conversation, including tool results. This is the agent’s default context manager.

from focus import AppendOnly

context = AppendOnly()

The agent renders the accumulated conversation for every decision and never compacts it. Use Segments(tokens) to continue one assistant response in exact token segments, or Summarize(...) to fold a long transcript. The context manager determines what the policy sees; branch_points separately determines where search may continue.

Definition

class AppendOnly(ContextManager[Any])

AppendOnly.init

def __init__(self, *, continuation: bool=False) -> None

AppendOnly.check_policy

def check_policy(self, policy: Policy[Any]) -> None

AppendOnly.initialize

async def initialize(self, rs: AnyRollout, agent: Agent[Any, Any]) -> AnyRollout

In continuation mode, check that the policy accepts exact token input before the rollout starts.

AppendOnly.prepare

def prepare(self, rs: AnyRollout, agent: Agent[Any, Any], *, tools: Sequence[dict[str, Any]] | None=None) -> PreparedPrompt

Prepare this branch’s next model input. The default renders the full conversation; with continuation=True, as Segments uses, it appends the original generated tokens to the initial prompt and marks the request for exact-continuation validation.

AppendOnly.record_reply

def record_reply(self, result: GenerateResult, messages: list[Message], agent: Agent[Any, Any]) -> None

Record the assistant reply. In continuation mode, keep its original token IDs in branch-local message metadata.

Segments

Continue one assistant response in exact segments of at most tokens tokens. Each segment is a decision, so search can branch in the middle of a long answer.

from focus import Segments

context = Segments(256)

Segments(256) caps each request at 256 tokens, exposes each segment boundary for branching, and disables the extra final-answer request by default. It declares token-chunk actions, so the agent normally selects TokenChunkParser; an explicit Agent(parser=...) takes precedence. The policy must accept and return exact token IDs.

Definition

class Segments(AppendOnly)

Segments.init

def __init__(self, tokens: int) -> None

Segments.branch_points

def branch_points(self) -> BranchPoints

Summarize

Ask the policy to summarize the transcript, then continue the task from its notes. Summarization runs as a short sub-rollout with the same model and can use workspace tools.

from focus import Summarize

context = Summarize(
    max_prompt_tokens=8192,
    max_reply_tokens=1024,
    max_interactions=4,
)

This manager requests a summary when the rendered prompt exceeds 8,192 tokens. The summary task allows four interactions, with at most 1,024 generated tokens per reply. Use trigger= for another condition, such as StateCounter() after a round of tool calls. Task state and workspace survive compaction. Compaction consumes one outer agent step; its generated turns remain in the history for training. penalty_cap bounds the episode’s total penalty for malformed structured summaries.

Definition

class Summarize(ContextManager[S])

Summarize.init

def __init__(self, *, trigger: Trigger[S] | None=None, max_interactions: int=8, max_prompt_tokens: int=8192, max_reply_tokens: int=2048, summary_target_tokens: int | None=None, require_complete_reply: bool=False, structured_summary: bool=False, sampling_params: SamplingParams | None=None, instruction: str=DEFAULT_INSTRUCTION, fold_workspace: Workspace | None=None, transcript_template: str='{transcript}', memory_role: Literal['user', 'assistant']='user', state_renderer: Callable[[S], str] | None=None, resume_instruction: str=_ASSISTANT_RESUME, include_folded_guidance: bool=True, penalty_cap: float | None=None) -> None

Summarize.compact

async def compact(self, compaction: Compaction[S]) -> CompactionResult[S]

Run the fold sub-rollout and return the new messages and its generated turns. Raises IncompleteCompactionError when the model does not produce usable notes.

AnyOf

Request compaction when any of the supplied triggers fires. After compaction, reset the state tracked by each trigger.

from focus.context import AnyOf, PromptTokens, StateCounter

trigger = AnyOf(PromptTokens(8192), StateCounter())

This example compacts when the prompt exceeds 8,192 tokens or the round’s tool-call allowance is spent. The task state must contain calls_remaining and a positive calls_max.

Definition

class AnyOf(Trigger[Any])

AnyOf.init

def __init__(self, *triggers: Trigger[Any]) -> None

AnyOf.of

@classmethod
def of(cls, triggers: Sequence[Trigger[Any]]) -> AnyOf

Construct the combined trigger from a sequence. Configuration mappings use this factory to build the entries in their triggers list.

AnyOf.fires

def fires(self, prompt: Sequence[TokenId], state: Any) -> bool

AnyOf.check

def check(self, state: Any) -> None

Validate the task-state fields required by every trigger.

AnyOf.relieve

def relieve(self, state: Any) -> Any

Compaction

What a context manager’s compact() receives: the live messages, the seed conversation, the task state, the workspace, and the agent being folded.

from focus.context import Compaction

Definition

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

Fields

messages: tuple[Message, ...]
seed: tuple[Message, ...]
state: S
workspace: Workspace
agent: Agent[S, Any]
checkpoint: Any = None

CompactionResult

Return the retained messages and state produced by one compaction operation.

from focus.context import CompactionResult

Definition

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

Fields

messages: tuple[Message, ...]
state: S
turns: tuple[Turn[Any], ...] = ()

ContextError

An invalid context operation. The message is returned to the model as corrective feedback.

from focus.context import ContextError

Definition

class ContextError(ValueError)

ContextTools

Context operations the model calls as tools, such as dropping or summarizing earlier messages. A reply that mixes context operations with task actions is refused.

from focus.context import ContextTools

Definition

class ContextTools(ContextManager[Any])

ContextTools.manages_context

@property
def manages_context(self) -> bool

Context tools always manage the live context.

ContextTools.operation

async def operation(self, rs: AnyRollout, name: str, arguments: dict[str, Any], agent: Agent[Any, Any]) -> tuple[AnyRollout, str, str]

Apply one named context operation to the rollout. Raise ContextError with a message the model can act on.

ContextTools.apply

async def apply(self, rs: AnyRollout, action: Any, agent: Agent[Any, Any], *, ending: bool) -> Applied

Route context tool calls to operation() and other actions to the environment.

IncompleteCompactionError

Report rejected compaction output while retaining the generations that produced it.

from focus.context import IncompleteCompactionError

Definition

class IncompleteCompactionError(ValueError)

IncompleteCompactionError.init

def __init__(self, *, turns: tuple[Turn[Any], ...], reasons: tuple[str, ...]) -> None

PromptTokens

Request compaction when the rendered prompt grows beyond a token threshold.

from focus.context import PromptTokens

trigger = PromptTokens(max_prompt_tokens=8192)

Fires once the prompt contains more than 8,192 tokens. Pass it as Summarize(trigger=trigger), or combine triggers with |.

Definition

class PromptTokens(Trigger[Any])

PromptTokens.init

def __init__(self, max_prompt_tokens: int=8192) -> None

PromptTokens.fires

def fires(self, prompt: Sequence[TokenId], state: Any) -> bool

StateCounter

Request compaction when a task-state counter is spent, then refill that counter from its configured maximum.

from focus.context import StateCounter

trigger = StateCounter(
    remaining="calls_remaining", maximum="calls_max"
)

Uses the fields supplied by RoundBudget. Initialize calls_max to a positive round allowance and charge calls_remaining as tool calls run.

Definition

class StateCounter(Trigger[Any])

StateCounter.init

def __init__(self, remaining: str='calls_remaining', maximum: str='calls_max') -> None

StateCounter.fires

def fires(self, prompt: Sequence[TokenId], state: Any) -> bool

StateCounter.check

def check(self, state: Any) -> None

Check that the state contains both counter fields and that its refill allowance is positive. The agent performs this validation before generation.

StateCounter.relieve

def relieve(self, state: Any) -> Any

Trigger

Define when context compaction should happen. Implement fires and, for state-based budgets, relieve so the task can make progress after compaction.

from focus.context import Trigger

Definition

class Trigger[S](ABC)

Trigger.fires

Required implementation.

@abstractmethod
def fires(self, prompt: Sequence[TokenId], state: S) -> bool

Return whether the prompt or task state requires compaction.

Trigger.check

def check(self, state: S) -> None

Validate the initial task state before generation. Override this hook when the trigger requires particular state fields. The default performs no checks.

Trigger.relieve

def relieve(self, state: S) -> S

Return the task state to use after compaction. The default leaves it unchanged. Counter-based triggers reset their counters here so the same condition does not immediately trigger another compaction.

Applied

A context manager’s updated rollout and optional task transition after applying an action. Context operations can update the rollout without advancing the task.

from focus.context.base import Applied

Definition

@dataclass(frozen=True)
class Applied

Fields

rollout: AnyRollout
transition: StepResult[Any] | None = None

DEFAULT_INSTRUCTION

Provide the default instruction given to the agent during context compaction.

from focus.context.compaction import DEFAULT_INSTRUCTION

Definition

DEFAULT_INSTRUCTION = "You are compacting the working context of an agent that is part-way through a task. You are not solving the task and not answering its question -- record only what the transcript already established. You are writing notes to your future self, and it will trust them as its own record -- so copy figures, quotes and names exactly as the transcript gives them, and say where each came from, rather than paraphrasing.\n\n{workspace}\n\nWhen you are done, reply with no tool call: that reply replaces the whole transcript as the agent's working context.{budget}{limit}"