Compaction

Compaction

Bases: Capability

Source code in src/agents/sandbox/capabilities/compaction.py

class Compaction(Capability):
    type: Literal["compaction"] = "compaction"
    policy: CompactionPolicy | None = Field(default=None)

    @field_validator("policy", mode="before")
    @classmethod
    def _validate_policy(cls, value: object) -> object | None:
        if value is None:
            return None
        if isinstance(value, CompactionPolicy):
            return value
        if isinstance(value, Mapping):
            policy_type = value.get("type")
            if policy_type == "static":
                return StaticCompactionPolicy.model_validate(dict(value))
            if policy_type == "dynamic":
                return DynamicCompactionPolicy.model_validate(dict(value))
            raise ValueError(f"Unsupported compaction policy type: {policy_type!r}")
        return value

    @field_serializer("policy", when_used="always", return_type=dict[str, Any])
    def _serialize_policy(self, policy: CompactionPolicy | None) -> dict[str, Any] | None:
        if policy is None:
            return None
        return policy.model_dump()

    def sampling_params(self, sampling_params: dict[str, Any]) -> dict[str, Any]:
        policy = self.policy
        if policy is None:
            model = sampling_params.get("model")
            if isinstance(model, str) and model:
                model_info = CompactionModelInfo.maybe_for_model(model)
                if model_info is None:
                    policy = StaticCompactionPolicy()
                else:
                    policy = DynamicCompactionPolicy(model_info=model_info)
            else:
                policy = StaticCompactionPolicy()

        return {
            "context_management": [
                {
                    "type": "compaction",
                    "compact_threshold": policy.compaction_threshold(sampling_params),
                }
            ]
        }

    def process_context(self, context: list[TResponseInputItem]) -> list[TResponseInputItem]:
        """When a compaction item is received, truncate the context before it."""
        last_compaction_index: int | None = None
        for index in range(len(context) - 1, -1, -1):
            item = context[index]
            item_type = (
                item.get("type") if isinstance(item, Mapping) else getattr(item, "type", None)
            )
            if item_type == "compaction":
                last_compaction_index = index
                break

        if last_compaction_index is not None:
            return context[last_compaction_index:]

        return context

process_context

process_context(
    context: list[TResponseInputItem],
) -> list[TResponseInputItem]

When a compaction item is received, truncate the context before it.

Source code in src/agents/sandbox/capabilities/compaction.py

def process_context(self, context: list[TResponseInputItem]) -> list[TResponseInputItem]:
    """When a compaction item is received, truncate the context before it."""
    last_compaction_index: int | None = None
    for index in range(len(context) - 1, -1, -1):
        item = context[index]
        item_type = (
            item.get("type") if isinstance(item, Mapping) else getattr(item, "type", None)
        )
        if item_type == "compaction":
            last_compaction_index = index
            break

    if last_compaction_index is not None:
        return context[last_compaction_index:]

    return context

clone

clone() -> Capability

Return a per-run copy of this capability.

Source code in src/agents/sandbox/capabilities/capability.py

def clone(self) -> "Capability":
    """Return a per-run copy of this capability."""
    cloned = self.model_copy(deep=False)
    for name, value in self.__dict__.items():
        cloned.__dict__[name] = _clone_capability_value(value)
    return cloned

bind

bind(session: BaseSandboxSession) -> None

Bind a live session to this plugin (default no-op).

Source code in src/agents/sandbox/capabilities/capability.py

def bind(self, session: BaseSandboxSession) -> None:
    """Bind a live session to this plugin (default no-op)."""
    self.session = session

bind_run_as

bind_run_as(user: User | None) -> None

Bind the sandbox user identity for model-facing operations.

Source code in src/agents/sandbox/capabilities/capability.py

def bind_run_as(self, user: User | None) -> None:
    """Bind the sandbox user identity for model-facing operations."""
    self.run_as = user

required_capability_types

required_capability_types() -> set[str]

Return capability types that must be present alongside this capability.

Source code in src/agents/sandbox/capabilities/capability.py

def required_capability_types(self) -> set[str]:
    """Return capability types that must be present alongside this capability."""
    return set()

instructions async

instructions(manifest: Manifest) -> str | None

Return a deterministic instruction fragment appended during run preparation.

Source code in src/agents/sandbox/capabilities/capability.py

async def instructions(self, manifest: Manifest) -> str | None:
    """Return a deterministic instruction fragment appended during run preparation."""
    _ = manifest
    return None

CompactionModelInfo

Bases: BaseModel

Source code in src/agents/sandbox/capabilities/compaction.py

class CompactionModelInfo(BaseModel):
    context_window: int

    @classmethod
    def maybe_for_model(cls, model: str) -> CompactionModelInfo | None:
        context_window = _MODEL_CONTEXT_WINDOWS.get(_model_lookup_key(model))
        if context_window is None:
            return None
        return cls(context_window=context_window)

    @classmethod
    def for_model(cls, model: str) -> CompactionModelInfo:
        model_info = cls.maybe_for_model(model)
        if model_info is not None:
            return model_info
        raise ValueError(f"Unknown context window for model: {model!r}")

CompactionPolicy

Bases: BaseModel, ABC

Source code in src/agents/sandbox/capabilities/compaction.py

class CompactionPolicy(BaseModel, abc.ABC):
    type: str

    @abc.abstractmethod
    def compaction_threshold(self, sampling_params: dict[str, Any]) -> int: ...

DynamicCompactionPolicy

Bases: CompactionPolicy

Source code in src/agents/sandbox/capabilities/compaction.py

class DynamicCompactionPolicy(CompactionPolicy):
    type: Literal["dynamic"] = "dynamic"
    model_info: CompactionModelInfo
    threshold: float = Field(ge=0, le=1, default=0.9)

    def compaction_threshold(self, sampling_params: dict[str, Any]) -> int:
        _ = sampling_params
        return int(self.model_info.context_window * self.threshold)

StaticCompactionPolicy

Bases: CompactionPolicy

Source code in src/agents/sandbox/capabilities/compaction.py

class StaticCompactionPolicy(CompactionPolicy):
    type: Literal["static"] = "static"
    threshold: int = Field(default=_DEFAULT_COMPACT_THRESHOLD)

    def compaction_threshold(self, sampling_params: dict[str, Any]) -> int:
        _ = sampling_params
        return self.threshold