# The public API of contraagent. Generated by `uv run python tests/api/snapshot.py`; do not edit.

## contraagent

@dataclass(frozen=True, eq=False)
class AgentTask[Out]
    def __init__(
        self,
        kind: str,
        output: type | None,
        params: type[BaseModel] | None,
        plugin: Plugin | None,
        role: str | None,
        instructions: str | None,
        config: dict[str, Any],
        validators: tuple[Validator, ...],
        output_name: str,
        requires: tuple[str, ...] = (),
        claims: tuple[str, ...] = (),
        skills_from: tuple[Plugin, ...] = (),
        changes: tuple[str, ...] = (),
        output_from: str = 'file',
        unwrap: bool = False,
        tools: tuple[str, ...] | None = None,
        rerun_on_resume: bool = False,
        preserve: tuple[str, ...] = (),
        output_of: AgentTask[Any] | PythonTask[Any] | None = None,
        base_kind: str = '',
        outside_state: OutsideState | None = None,
    ) -> None
    def __call__(
        self,
        unit: str = '',
        *,
        key: str | None = None,
        after: Sequence[Pending[Any] | str] = (),
        priority: float = 0.0,
        credential: str | None = None,
        **params: Any,
    ) -> Pending[Out]
    base_kind: str = ''
    changes: tuple[str, ...] = ()
    claims: tuple[str, ...] = ()
    config: dict[str, Any]
    def differences(self, other: AgentTask[Any]) -> list[str]
    def digest(self) -> str
    executor = 'agent'
    instructions: str | None
    kind: str
    @property light: bool
    output: type | None
    @property output_format: str | None
    output_from: str = 'file'
    @property output_json_schema: dict[str, Any]
    output_name: str
    output_of: AgentTask[Any] | PythonTask[Any] | None = None
    @property output_schema: type[BaseModel] | None
    outside_state: OutsideState | None = None
    params: type[BaseModel] | None
    plugin: Plugin | None
    @property plugins: tuple[Plugin, ...]
    preserve: tuple[str, ...] = ()
    requires: tuple[str, ...] = ()
    rerun_on_resume: bool = False
    role: str | None
    skills_from: tuple[Plugin, ...] = ()
    def system_prompt(self, instructions: str | None = None) -> str | None
    tools: tuple[str, ...] | None = None
    unwrap: bool = False
    validators: tuple[Validator, ...]

class Changes(BaseModel)
    created: tuple[str, ...] = ()
    deleted: tuple[str, ...] = ()
    kept: tuple[str, ...] = ()
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    modified: tuple[str, ...] = ()
    removed: tuple[str, ...] = ()

@dataclass
class CheckReport
    def __init__(
        self,
        errors: list[str] = list(),
        warnings: list[str] = list(),
        kinds: dict[str, int] = dict(),
        topics: list[str] = list(),
        built: bool = True,
        routes: dict[str, str] = dict(),
        later_kinds: list[str] = list(),
    ) -> None
    def add(self, topic: str, errors: list[str]) -> None
    built: bool = True
    errors: list[str] = field(default_factory=list)
    kinds: dict[str, int] = field(default_factory=dict)
    later_kinds: list[str] = field(default_factory=list)
    @property ok: bool
    routes: dict[str, str] = field(default_factory=dict)
    topics: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

@dataclass(frozen=True)
class CommandValidator
    def __init__(
        self,
        argv: Sequence[str],
        name: str = '',
        timeout_s: float = 120.0,
        max_lines: int = 30,
    ) -> None
    def __call__(
        self,
        path: Path,
        value: Any = None,
        *,
        call: Path | None = None,
        cwd: Path | None = None,
    ) -> list[str]
    argv: tuple[str, ...]
    def command_for(self, path: Path, call: Path | None = None) -> list[str]
    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> CommandValidator
    max_lines: int = 30
    name: str = ''
    @property needs_call: bool
    timeout_s: float = 120.0
    def to_dict(self) -> dict[str, Any]

class ConfigError(ContraagentError, ValueError)

class ContraagentDeprecationWarning(DeprecationWarning)

class ContraagentError(Exception)

class CredentialError(ContraagentError)

class DeferralCause(BaseModel)
    cause: str
    dependency: str | None = Field(default=None, description='`dependency_deferred`: the key of the deferred task whose output it reads')
    flow_call: str | None = Field(default=None, description="`deadline`, `grace_window`, `cannot_finish`: the flow call whose time box it is; null for the run's deadline")
    left_s: float | None = Field(default=None, description='`cannot_finish`: the time left before the deadline, in seconds')
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    need_s: float | None = Field(default=None, description='`cannot_finish`: the work it needed, in seconds')
    rests: tuple[SkippedHop, ...] = Field(default=(), description='`rest`: each hop passed over and why its credential rests')
    def to_dict(self) -> dict[str, Any]

class Frame(BaseModel)
    file: str = Field(description="relative to the run's working directory when under it, else as recorded")
    function: str = Field(description='`module.qualname`')
    line: int
    loops: tuple[Loop, ...] = Field(default=(), description='the loops of a generator whose pass `line` was in, outermost first; empty outside a loop and for any other frame')
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True, serialize_by_alias=True)

class Hop(BaseModel)
    agent: str = ''
    cost_usd: float | None = None
    credential: str = ''
    model: str = ''
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    other_models: dict[str, OtherModel] = Field(default_factory=dict)
    route: str = ''
    stop: str = ''
    usage: TokenUsage | None = None
    wall_s: float = 0.0

class LaunchError(ContraagentError)

class Loop(BaseModel)
    line: int = Field(description="the loop's first line, which names it")
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True, serialize_by_alias=True)
    pass_: int | None = Field(validation_alias=AliasChoices('pass', 'pass_'), serialization_alias='pass', description='one iteration of this loop, numbered from 1 since the loop was last entered in that frame; null when not counted')

class McpServers(Protocol)
    def __call__(
        self,
        task: Task,
        params: dict[str, Any],
        servers: dict[str, dict[str, Any]],
        /,
    ) -> Mapping[str, Mapping[str, Any]] | None

class NoCurrentRun(ContraagentError)

@dataclass(frozen=True)
class Origin
    def __init__(
        self,
        flow: tuple[Frame, ...] = (),
        created_from: tuple[str, ...] = (),
        flow_call: str = '0',
    ) -> None
    created_from: tuple[str, ...] = ()
    flow: tuple[Frame, ...] = ()
    flow_call: str = '0'

class OtherModel(BaseModel)
    cache_creation: int = 0
    cache_read: int = 0
    cost_usd: float | None = None
    input: int = 0
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    output: int = 0
    def plus(self, other: OtherModel) -> OtherModel

class OutsideState(Protocol)
    def __call__(self, unit: str, params: dict[str, Any], /) -> Any

@dataclass(frozen=True)
class Overrides
    def __init__(self, agent: str = '', model: str = '', effort: str = '', credential: str = '') -> None
    agent: str = ''
    credential: str = ''
    effort: str = ''
    model: str = ''
    def run_defaults(self, defaults: Mapping[str, Any]) -> tuple[str, str, str]

@dataclass(frozen=True)
class Partial[Out]
    def __init__(self, value: Out, reason: str) -> None
    reason: str
    value: Out

class Pending[Out]
    def __init__(self) -> None
    @property key: str
    @property ok: bool
    @property origin: Origin | None
    @property output: Out
    @property path: Path | None
    @property result: TaskResult | None
    @property unit: str

@dataclass(frozen=True)
class PerTask
    def __init__(self, memory_mb: int | None = None, cores: int | None = None) -> None
    cores: int | None = None
    memory_mb: int | None = None

class Pipeline
    def __init__(
        self,
        config: PipelineConfig | str | Path | None = None,
        *,
        runs_dir: str | Path = 'runs',
        backends: Mapping[str, Backend] | None = None,
        overrides: Overrides = Overrides(),
        cwd: str | Path | None = None,
        max_concurrent: int = 4,
        preflight: PreflightPolicy | None = None,
        keep_transcripts: bool = True,
        env: Mapping[str, str] | None = None,
        log: Callable[[str], None] | None = None,
        max_passes: int | None = None,
        quota_rule: QuotaRule | None = None,
        scopes: Sequence[TaskScope] = (),
        enrichers: Sequence[ResultEnricher] = (),
        row_extras: Callable[[TaskResult], dict[str, Any]] | None = None,
        mcp_servers: McpServers | None = None,
        stub: Any = None,
        record: str | Path | None = None,
        replay: str | Path | None = None,
        execute: Sequence[str] = (),
        record_missing: bool = False,
        record_max_file_mb: float | None = None,
        deadline_s: float | None = None,
        grace_s: float = GRACE_S,
        slack_s: float | None = None,
        finishing_kinds: Sequence[str] = (),
        overdue_s: float = OVERDUE_S,
        claim_capacity: Mapping[str, int | PerTask] | None = None,
        retry_wait_s: float = RETRY_WAIT_S,
        throttle_s: float = THROTTLE_S,
        services: Mapping[str, Service] | None = None,
    ) -> None
    backends: dict[str, Backend]
    def check(self, flow: Callable[..., Any], /, *args: Any, **kwargs: Any) -> CheckReport
    config
    cwd: Path
    enrichers: tuple[ResultEnricher, ...]
    env: dict[str, str]
    keep_transcripts
    log
    max_concurrent
    max_passes: int
    mcp_servers: McpServers | None
    overrides
    preflight: PreflightPolicy
    queue_options: dict[str, Any]
    quota_rule
    record: Path | None
    record_max_file_mb: float
    replay: Replay | None
    row_extras
    def run(
        self,
        flow: Callable[..., Any],
        /,
        *args: Any,
        run_id: str | None = None,
        resume: bool = False,
        **kwargs: Any,
    ) -> RunReport
    runs_dir: Path
    scopes: tuple[TaskScope, ...]
    services: dict[str, Service]
    stub

@dataclass(frozen=True)
class PipelineConfig
    def __init__(self, raw: dict[str, Any] = dict(), path: Path | None = None) -> None
    def claim_capacity(self, errors: list[str] | None = None) -> dict[str, int | PerTask]
    def credentials(self) -> dict[str, Any]
    def defaults(self, errors: list[str] | None = None) -> dict[str, Any]
    def fallbacks(self) -> list[dict[str, Any]]
    @classmethod
    def load(cls, path: str | Path) -> PipelineConfig
    path: Path | None = None
    raw: dict[str, Any] = field(default_factory=dict)
    @property source: str
    def table(self, name: str) -> dict[str, Any]
    def task_tables(self, errors: list[str] | None = None) -> dict[str, dict[str, Any]]
    def unknown_tables(self, errors: list[str] | None = None) -> None

class PluginError(ContraagentError)

class PreflightStopped(ContraagentError)
    def __init__(self, record: PreflightRecord) -> None
    record

@dataclass(frozen=True, eq=False)
class PythonTask[Out]
    def __init__(
        self,
        fn: Callable[..., Out],
        name: str,
        output_schema: type[BaseModel] | None,
        validators: tuple[Validator, ...],
        claims: tuple[str, ...] = (),
        rerun_on_resume: bool = False,
        retries: int = 0,
        retry_on: tuple[type[Exception], ...] = (),
        output_name: str = DEFAULT_OUTPUT_NAME,
        outside_state: OutsideState | None = None,
    ) -> None
    def __call__(
        self,
        unit: str = '',
        *,
        key: str | None = None,
        after: Sequence[Pending[Any] | str] = (),
        priority: float = 0.0,
        **kwargs: Any,
    ) -> Pending[Out]
    def arguments(self, unit: str, kwargs: dict[str, Any]) -> dict[str, Any]
    claims: tuple[str, ...] = ()
    def definition(self, cwd: Path | None = None) -> dict[str, Any]
    def digest(self) -> str
    executor = 'python'
    fn: Callable[..., Out]
    def invoke(self, unit: str, kwargs: dict[str, Any]) -> Any
    @property kind: str
    name: str
    output_format = 'json'
    output_name: str = DEFAULT_OUTPUT_NAME
    output_schema: type[BaseModel] | None
    outside_state: OutsideState | None = None
    rerun_on_resume: bool = False
    retries: int = 0
    retry_on: tuple[type[Exception], ...] = ()
    validators: tuple[Validator, ...]

class QuotaRule(Protocol)
    def __call__(self, credential: str, reading: QuotaReading, /) -> str | tuple[str, datetime] | None

class RecordingError(ContraagentError)

@dataclass(frozen=True)
class RecordingImport
    def __init__(
        self,
        run_dir: Path,
        directory: Path,
        imported: tuple[str, ...] = (),
        unfinished: Mapping[str, str] = dict(),
        not_imported: Mapping[str, str] = dict(),
    ) -> None
    directory: Path
    imported: tuple[str, ...] = ()
    not_imported: Mapping[str, str] = field(default_factory=dict)
    @property ok: bool
    run_dir: Path
    unfinished: Mapping[str, str] = field(default_factory=dict)

class ResultEnricher(Protocol)
    def enrich(self, task: Task, before: Any, result: TaskResult) -> TaskResult | None
    def snapshot(self, task: Task) -> Any

class RoutingError(ContraagentError, ValueError)
    def __init__(self, *errors: str) -> None
    errors: list[str]

class Rows[Row: BaseModel](list[Row])

class RunExists(ContraagentError)
    def __init__(self, run_dir: Path, rows: int)
    rows: int
    run_dir: Path

class RunFormatError(ContraagentError)

class RunLocked(ContraagentError)

@dataclass
class RunReport
    def __init__(
        self,
        run_id: str,
        run_dir: Path,
        results: dict[str, TaskResult],
        summary: dict[str, Any] = dict(),
        value: Any = None,
        flow_errors: list[FlowCall] = list(),
    ) -> None
    @property deferred: list[TaskResult]
    @property failed: list[TaskResult]
    flow_errors: list[FlowCall] = field(default_factory=list)
    @property ok: bool
    @property partial: list[TaskResult]
    results: dict[str, TaskResult]
    run_dir: Path
    run_id: str
    summary: dict[str, Any] = field(default_factory=dict)
    value: Any = None

class Service(Protocol)
    def __call__(self, name: str, /) -> AbstractContextManager[str | None]

class SkippedHop(Rest)
    model: str = ''
    route: str = ''

@dataclass(frozen=True)
class Task
    def __init__(
        self,
        kind: str,
        unit: str = '',
        after: tuple[str, ...] = (),
        fn: Callable[[], object] | None = None,
        claims: tuple[str, ...] = (),
        time_boxes: tuple[TimeBox, ...] = (),
        rerun_on_resume: bool = False,
        estimate_s: float | None = None,
        tail_s: float | None = None,
        priority: float = 0.0,
        params: dict[str, Any] = dict(),
        attempt: int = 0,
        id: str = (lambda: uuid.uuid4().hex)(),
        key: str = '',
        payload: Any = None,
        origin: Origin | None = None,
        fingerprint: str = '',
        base_kind: str = '',
        fingerprint_at_start: bool = False,
        outside: str = '',
        credential: str = '',
    ) -> None
    after: tuple[str, ...] = ()
    attempt: int = 0
    base_kind: str = ''
    claims: tuple[str, ...] = ()
    credential: str = ''
    @property declared_unit: str
    estimate_s: float | None = None
    @property executor: Executor
    fingerprint: str = ''
    fingerprint_at_start: bool = False
    fn: Callable[[], object] | None = None
    id: str = field(default_factory=lambda: uuid.uuid4().hex)
    key: str = ''
    kind: str
    origin: Origin | None = None
    outside: str = ''
    params: dict[str, Any] = field(default_factory=dict)
    payload: Any = None
    priority: float = 0.0
    rerun_on_resume: bool = False
    tail_s: float | None = None
    time_boxes: tuple[TimeBox, ...] = ()
    unit: str = ''

class TaskDeferred(TaskFailed)

class TaskFailed(Exception)
    def __init__(self, result: TaskResult, item: Any = None) -> None
    item
    key: str
    result
    stop: str

@dataclass(frozen=True)
class TaskResult
    def __init__(
        self,
        task: Task,
        ok: bool,
        stop: str,
        launch_result: Any = None,
        wall_s: float = 0.0,
        reason: str = '',
        notes: dict[str, Any] = dict(),
        route: str = '',
        hops: tuple[Hop, ...] = (),
        rounds: int | None = None,
        round_errors: tuple[tuple[str, ...], ...] = (),
        changes: Changes | None = None,
        skipped_hops: tuple[SkippedHop, ...] = (),
        cause: DeferralCause | None = None,
        retryable: bool = True,
        replayed: bool = False,
        payload: Any = None,
        max_corrections: int | None = None,
    ) -> None
    cause: DeferralCause | None = None
    changes: Changes | None = None
    @classmethod
    def deferred(cls, task: Task, deferral: Deferral) -> TaskResult
    @classmethod
    def duplicate_key(cls, task: Task) -> TaskResult
    hops: tuple[Hop, ...] = ()
    @classmethod
    def launch_failed(cls, task: Task, reason: str) -> TaskResult
    launch_result: Any = None
    max_corrections: int | None = None
    notes: dict[str, Any] = field(default_factory=dict)
    ok: bool
    payload: Any = None
    reason: str = ''
    replayed: bool = False
    retryable: bool = True
    round_errors: tuple[tuple[str, ...], ...] = ()
    rounds: int | None = None
    route: str = ''
    @classmethod
    def skipped(cls, task: Task) -> TaskResult
    skipped_hops: tuple[SkippedHop, ...] = ()
    stop: str
    def summary(self) -> dict[str, Any]
    task: Task
    wall_s: float = 0.0

class TaskScope(Protocol)
    def __call__(self, task: Task, /) -> AbstractContextManager[None]

class TextOutput(str)

@dataclass(frozen=True)
class TimeBox
    def __init__(
        self,
        deadline_at: float,
        deadline_s: float,
        grace_s: float,
        slack_s: float = 0.0,
        finishing_kinds: frozenset[str] = frozenset(),
        flow_call: str | None = None,
    ) -> None
    def cause(self, name: str, **fields: Any) -> DeferralCause
    deadline_at: float
    deadline_s: float
    finishing_kinds: frozenset[str] = frozenset()
    flow_call: str | None = None
    grace_s: float
    slack_s: float = 0.0

class TokenUsage(BaseModel)
    cache_creation: int = 0
    cache_read: int = 0
    calls: int = 0
    context_window: int | None = None
    input: int = 0
    model_config: ConfigDict = ConfigDict(extra='allow')
    output: int = 0
    peak_context: int | None = None
    resident: int | None = None

class TooManyPasses(ContraagentError)

class Validator(Protocol)
    def __call__(self, *args: Any, **kwargs: Any) -> Iterable[str] | None

@overload
def agent_task(
    kind: str,
    *,
    output: type[Out] | Any | None = None,
    params: type[BaseModel] | None = None,
    plugin: str | Path | None = None,
    skills_from: Sequence[str | Path] = (),
    role: str | None = None,
    instructions: str | None = None,
    validators: Sequence[Validator] = (),
    output_name: str | None = None,
    output_from: Literal['file', 'message'] = 'file',
    unwrap: bool = False,
    requires: Sequence[str] = (),
    claims: Sequence[str] = (),
    changes: Sequence[str] = (),
    tools: Sequence[str] | None = None,
    rerun_on_resume: bool = False,
    preserve: Sequence[str] = (),
    output_of: AgentTask[Any] | PythonTask[Any] | None = None,
    implementations: None = None,
    outside_state: OutsideState | None = None,
    agent: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    credential: str | None = None,
    model_failover: bool | None = None,
    fallbacks: Sequence[Mapping[str, Any]] | None = None,
    disallowed_tools: Sequence[str] | None = None,
    max_turns: int | None = None,
    timeout_s: float | None = None,
    max_budget_usd: float | None = None,
    stall_s: float | None = None,
    context_budget: int | None = None,
    context_warn_pct: int | None = None,
    max_denials: int | None = None,
    max_concurrent: int | None = None,
    retries: int | None = None,
    retry_backoff_base: float | None = None,
    partial_on: Sequence[Literal['timeout', 'stalled']] | None = None,
    keep_failed_changes: bool | None = None,
    max_corrections: int | None = None,
) -> AgentTask[Out]
@overload
def agent_task(
    kind: str,
    *,
    output: type[Out] | Any | None = None,
    params: type[BaseModel] | None = None,
    plugin: str | Path | None = None,
    skills_from: Sequence[str | Path] = (),
    role: str | None = None,
    instructions: str | None = None,
    validators: Sequence[Validator] = (),
    output_name: str | None = None,
    output_from: Literal['file', 'message'] = 'file',
    unwrap: bool = False,
    requires: Sequence[str] = (),
    claims: Sequence[str] = (),
    changes: Sequence[str] = (),
    tools: Sequence[str] | None = None,
    rerun_on_resume: bool = False,
    preserve: Sequence[str] = (),
    output_of: AgentTask[Any] | PythonTask[Any] | None = None,
    implementations: Mapping[str, str | Path],
    outside_state: OutsideState | None = None,
    agent: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    credential: str | None = None,
    model_failover: bool | None = None,
    fallbacks: Sequence[Mapping[str, Any]] | None = None,
    disallowed_tools: Sequence[str] | None = None,
    max_turns: int | None = None,
    timeout_s: float | None = None,
    max_budget_usd: float | None = None,
    stall_s: float | None = None,
    context_budget: int | None = None,
    context_warn_pct: int | None = None,
    max_denials: int | None = None,
    max_concurrent: int | None = None,
    retries: int | None = None,
    retry_backoff_base: float | None = None,
    partial_on: Sequence[Literal['timeout', 'stalled']] | None = None,
    keep_failed_changes: bool | None = None,
    max_corrections: int | None = None,
) -> Mapping[str, AgentTask[Out]]
def agent_task(
    kind: str,
    *,
    output: type[Out] | Any | None = None,
    params: type[BaseModel] | None = None,
    plugin: str | Path | None = None,
    skills_from: Sequence[str | Path] = (),
    role: str | None = None,
    instructions: str | None = None,
    validators: Sequence[Validator] = (),
    output_name: str | None = None,
    output_from: Literal['file', 'message'] = 'file',
    unwrap: bool = False,
    requires: Sequence[str] = (),
    claims: Sequence[str] = (),
    changes: Sequence[str] = (),
    tools: Sequence[str] | None = None,
    rerun_on_resume: bool = False,
    preserve: Sequence[str] = (),
    output_of: AgentTask[Any] | PythonTask[Any] | None = None,
    implementations: Mapping[str, str | Path] | None = None,
    outside_state: OutsideState | None = None,
    agent: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    credential: str | None = None,
    model_failover: bool | None = None,
    fallbacks: Sequence[Mapping[str, Any]] | None = None,
    disallowed_tools: Sequence[str] | None = None,
    max_turns: int | None = None,
    timeout_s: float | None = None,
    max_budget_usd: float | None = None,
    stall_s: float | None = None,
    context_budget: int | None = None,
    context_warn_pct: int | None = None,
    max_denials: int | None = None,
    max_concurrent: int | None = None,
    retries: int | None = None,
    retry_backoff_base: float | None = None,
    partial_on: Sequence[Literal['timeout', 'stalled']] | None = None,
    keep_failed_changes: bool | None = None,
    max_corrections: int | None = None,
) -> AgentTask[Out] | Mapping[str, AgentTask[Out]]

def import_recording(
    run_dir: str | Path,
    directory: str | Path,
    *,
    record_max_file_mb: float = MAX_FILE_MB,
) -> RecordingImport

def log(name: str, row: Mapping[str, Any] | BaseModel) -> None

def max_passes(n: int, /) -> Callable[[F], F]

def pipeline_table(name: str) -> dict[str, Any]

@overload
def python_task(fn: Callable[..., Out], /) -> PythonTask[Out]
@overload
def python_task(
    *,
    name: str | None = None,
    validators: Sequence[Validator] = (),
    claims: Sequence[str] = (),
    rerun_on_resume: bool = False,
    retries: int = 0,
    retry_on: tuple[type[Exception], ...] = (),
    output_name: str | None = None,
    outside_state: OutsideState | None = None,
) -> Callable[[Callable[..., Out]], PythonTask[Out]]
def python_task(
    fn: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    validators: Sequence[Validator] = (),
    claims: Sequence[str] = (),
    rerun_on_resume: bool = False,
    retries: int = 0,
    retry_on: tuple[type[Exception], ...] = (),
    output_name: str | None = None,
    outside_state: OutsideState | None = None,
) -> PythonTask[Any] | Callable[[Callable[..., Any]], PythonTask[Any]]

def save(name: str, value: Any) -> None

def service_address(name: str) -> str | None

def time_box(
    flow: Generator[Y, S, R],
    *,
    deadline_s: float,
    grace_s: float = GRACE_S,
    finishing_kinds: Iterable[str] = (),
    slack_s: float | None = None,
) -> Generator[Y, S, R]

## contraagent.config

AMBIENT = Ambient(name='ambient')

class Ambient(Credential)
    isolate: bool = False
    type: Literal['ambient'] = 'ambient'

class ApiKey(Credential)
    key: SecretField
    provider: str = 'anthropic'
    type: Literal['api_key'] = 'api_key'

class Bedrock(Credential)
    aws_profile: str | None = None
    bearer: SecretField | None = None
    region: str
    type: Literal['bedrock'] = 'bedrock'

class Credential(BaseModel)
    def describe(self) -> dict[str, Any]
    isolate: bool = True
    @property label: str
    max_five_hour_pct: float | None = Field(default=None, gt=0, le=100)
    max_seven_day_pct: float | None = Field(default=None, gt=0, le=100)
    model_config: ConfigDict = ConfigDict(extra='forbid', arbitrary_types_allowed=True)
    name: str = ''
    def secret_sources(self) -> tuple[str, ...]
    traits: frozenset[str] = frozenset()
    type: str

class Gateway(Credential)
    auth: Literal['bearer', 'api_key'] = 'bearer'
    base_url: str
    def endpoint(self, env: dict[str, str] | None) -> str
    def header_lines(self) -> str
    headers: dict[str, str] = Field(default_factory=dict)
    provider: str = 'litellm'
    def secret_sources(self) -> tuple[str, ...]
    token: SecretField | None = None
    type: Literal['gateway'] = 'gateway'

@dataclass(frozen=True)
class KindConfig
    def __init__(
        self,
        agent: str = 'claude',
        model: str = '',
        effort: str | None = None,
        model_failover: bool = False,
        max_turns: int = 50,
        timeout_s: float = 600,
        max_budget_usd: float | None = None,
        stall_s: float | None = None,
        max_concurrent: int | None = None,
        settings: str = '',
        context_budget: int | None = None,
        context_warn_pct: int | None = None,
        max_denials: int | None = None,
        disallowed_tools: tuple[str, ...] = (),
        exclude_dynamic_sections: bool = False,
        credential: str = 'ambient',
        requires: tuple[str, ...] = (),
        light: bool = False,
        retries: int = 2,
        retry_backoff_base: float = 2.0,
        partial_on: tuple[str, ...] = (),
        keep_failed_changes: bool = False,
        max_corrections: int = DEFAULT_MAX_CORRECTIONS,
    ) -> None
    agent: str = 'claude'
    @classmethod
    def build(cls, defaults: dict[str, Any], overrides: dict[str, Any] | None) -> KindConfig
    context_budget: int | None = None
    context_warn_pct: int | None = None
    credential: str = 'ambient'
    disallowed_tools: tuple[str, ...] = ()
    effort: str | None = None
    exclude_dynamic_sections: bool = False
    keep_failed_changes: bool = False
    light: bool = False
    def limits(self) -> Limits
    max_budget_usd: float | None = None
    max_concurrent: int | None = None
    max_corrections: int = DEFAULT_MAX_CORRECTIONS
    max_denials: int | None = None
    max_turns: int = 50
    model: str = ''
    model_failover: bool = False
    partial_on: tuple[str, ...] = ()
    @property partial_stops: frozenset[Stop]
    requires: tuple[str, ...] = ()
    retries: int = 2
    retry_backoff_base: float = 2.0
    settings: str = ''
    stall_s: float | None = None
    timeout_s: float = 600
    def with_(self, **changes: Any) -> KindConfig

@dataclass(frozen=True)
class Limits
    def __init__(
        self,
        timeout_s: float = 600.0,
        max_turns: int | None = None,
        max_budget_usd: float | None = None,
        context_budget: int | None = None,
        context_warn_pct: int | None = None,
        stall_s: float | None = None,
        disallowed_tools: tuple[str, ...] = (),
        max_denials: int | None = None,
    ) -> None
    context_budget: int | None = None
    context_warn_pct: int | None = None
    def context_warned(self, peak_context: int | None) -> bool
    disallowed_tools: tuple[str, ...] = ()
    max_budget_usd: float | None = None
    max_denials: int | None = None
    max_turns: int | None = None
    stall_s: float | None = None
    timeout_s: float = 600.0

class OAuthToken(Credential)
    provider: str = 'anthropic'
    token: SecretField
    type: Literal['oauth_token'] = 'oauth_token'

class Pool(Credential)
    cooldown_s: float = 900.0
    members: tuple[str, ...]
    strategy: Literal['in_order', 'round_robin'] = 'in_order'
    type: Literal['pool'] = 'pool'

class QuotaReading(BaseModel)
    account: str = ''
    credential: str = ''
    extra: dict[str, Any] = Field(default_factory=dict)
    five_hour_pct: float | None = None
    five_hour_resets_at: str | None = None
    @classmethod
    def from_windows(cls, windows: Mapping[str, QuotaWindow], **fields: Any) -> QuotaReading
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    def named_windows(self) -> dict[str, QuotaWindow]
    observed_at: str | None = None
    overage_disabled_reason: str = ''
    overage_status: str = ''
    seven_day_pct: float | None = None
    seven_day_resets_at: str | None = None
    status: str = ''
    status_window: str = ''
    threshold_pct: float | None = None
    using_overage: bool | None = None
    windows: dict[str, QuotaWindow] = Field(default_factory=dict)

class QuotaWindow(BaseModel)
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    pct: float | None = None
    resets_at: str | None = None

@dataclass(frozen=True)
class Route
    def __init__(self, agent: str, model: str, credential: str = '') -> None
    agent: str
    credential: str = ''
    @property identity: tuple[str, str, str]
    def label(self, hop: int, total: int) -> str
    model: str

@dataclass(frozen=True)
class Secret
    def __init__(self, source: str) -> None
    def resolve(self, env: dict[str, str]) -> str
    source: str

class Subscription(Credential)
    config_dir: Path | None = None
    profile: str | None = None
    provider: str = 'anthropic'
    type: Literal['subscription'] = 'subscription'

class Vertex(Credential)
    credentials_file: Path | None = None
    project: str
    region: str
    type: Literal['vertex'] = 'vertex'

class WorkloadIdentity(Credential)
    federation_rule_id: str
    identity_token_file: Path
    organization_id: str
    provider: str = 'anthropic'
    service_account_id: str | None = None
    type: Literal['workload_identity'] = 'workload_identity'
    workspace_id: str | None = None

## contraagent.backends

class Backend(abc.ABC)
    def __init__(self, binary: str = '', base_env: dict[str, str] | None = None)
    def account(self, credential: Credential, model: str = '') -> str
    @abstractmethod
    def argv(self, launch: Launch) -> list[str]
    def attest(self, launch_result: LaunchResult, credential: Credential) -> str | None
    def base_path(self) -> str | None
    binary: str = ''
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    def catalogue(self) -> set[str] | None
    catalogue_timeout_s: float = 30.0
    def context_window(self, model: str) -> int | None
    cost_is_session_total: bool = False
    def discard_session(self, launch: Launch, session_id: str) -> None
    drivable_model_chain: bool = False
    effort_levels: tuple[str, ...] | None = None
    @abstractmethod
    def env(self, launch: Launch) -> dict[str, str]
    def exchange(
        self,
        launch: Launch,
        argv: list[str],
        env: dict[str, str],
        fold: EventFold,
        watch: LimitsWatch,
    ) -> tuple[bytes, str, int | None, Stop | None]
    def fold(self, launch: Launch) -> EventFold
    def launch_failed(
        self,
        launch: Launch,
        why: str,
        *,
        argv: list[str] | None = None,
        started_at: str = '',
        t0: float | None = None,
    ) -> LaunchResult
    light_launch: bool = False
    def mark_launch(self, launch: Launch) -> None
    def model_chains(self, cwd: Path | None = None) -> ModelChains | None
    def models(self) -> dict[str, int | None] | None
    name: str = ''
    native_model_chain: bool = False
    needs_inline_role_file: bool = False
    plaintext_stream: bool = False
    def prepare(self, launch: Launch) -> Launch
    def read_models(self) -> dict[str, int | None] | None
    def reclassify(self, launch_result: LaunchResult, stderr: str) -> None
    def resumable(self, launch: Launch) -> Launch
    def resumed(self, launch: Launch, prior: LaunchResult, prompt: str) -> Launch | None
    def run(self, launch: Launch) -> LaunchResult
    def run_catalogue(self, *args: str, env: dict[str, str] | None = None) -> str | None
    def scratch(self, label: str, which: str, create: bool = True) -> Path
    def scratch_parent(self) -> Path | None
    signals: Signals = Signals(stderr_stops=frozenset())
    stop_hooks_file: str | None = None
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None
    def usage_reading(self, launch: Launch) -> QuotaReading | None
    def version(self) -> str
    def watch_stderr(self, launch: Launch) -> Sink | None
    watched_limits: frozenset[str] = frozenset()
    wire: ModuleType | None = None

class CallFile(BaseModel)
    cwd: str = ''
    key: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    output: CallOutput | None = None
    params: dict[str, Any] = Field(default_factory=dict)
    @staticmethod
    def path_of(launch: Launch) -> Path | None
    @classmethod
    def read(cls, path: Path) -> CallFile
    run: str
    unit: str = ''
    version: int = 1
    def write(self, path: Path) -> Path

class CallOutput(BaseModel)
    commands: list[dict[str, Any]] = Field(default_factory=list)
    format: Literal['json', 'text', 'jsonl'] = 'json'
    from_: Literal['file', 'message'] = Field('file', validation_alias=AliasChoices('from', 'from_'), serialization_alias='from')
    json_schema: dict[str, Any] = Field(default_factory=dict)
    model_config: ConfigDict = ConfigDict(extra='forbid', serialize_by_alias=True)
    path: str
    preserve: list[str] = Field(default_factory=list)
    schema_ref: SchemaRef | None = None

@dataclass
class CallUsage
    def __init__(self, model: str, input: int, cache_read: int, cache_creation: int, output: int) -> None
    cache_creation: int
    cache_read: int
    @property context: int
    input: int
    model: str
    output: int

class ClaudeBackend(ResumeById, Backend)
    def __init__(self, binary: str = 'claude', base_env: dict[str, str] | None = None)
    def account(self, credential: Credential, model: str = '') -> str
    def argv(self, launch: Launch) -> list[str]
    def attest(self, launch_result: LaunchResult, credential: Credential) -> str | None
    binary = 'claude'
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    cost_is_session_total = True
    def discard_session(self, launch: Launch, session_id: str) -> None
    drivable_model_chain = True
    effort_levels: tuple[str, ...] | None = ('low', 'medium', 'high', 'xhigh', 'max')
    def env(self, launch: Launch) -> dict[str, str]
    def fold(self, launch: Launch) -> EventFold
    light_launch = True
    name = 'claude'
    native_model_chain = True
    def resumable(self, launch: Launch) -> Launch
    def resumed(self, launch: Launch, prior: LaunchResult, prompt: str) -> Launch | None
    signals: Signals = claude_wire.SIGNALS
    stop_hooks_file: str | None = 'hooks/hooks.json'
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None
    watched_limits: frozenset[str] = frozenset({'context_budget'})
    wire: ModuleType | None = claude_wire

class CodexBackend(ResumeById, Backend)
    def __init__(self, binary: str = 'codex', base_env: dict[str, str] | None = None)
    def argv(self, launch: Launch) -> list[str]
    binary = 'codex'
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    def codex_home(self, launch: Launch) -> Path
    def discard_session(self, launch: Launch, session_id: str) -> None
    effort_levels: tuple[str, ...] | None = ('low', 'medium', 'high', 'xhigh', 'max', 'ultra')
    def env(self, launch: Launch) -> dict[str, str]
    def exchange(
        self,
        launch: Launch,
        argv: list[str],
        env: dict[str, str],
        fold: EventFold,
        watch: LimitsWatch,
    ) -> tuple[bytes, str, int | None, Stop | None]
    def fold(self, launch: Launch) -> EventFold
    def hook_args(self, launch: Launch) -> list[str]
    def launch_tmp(self, launch: Launch) -> Path
    name = 'codex'
    needs_inline_role_file = True
    def prepare(self, launch: Launch) -> Launch
    def real_home(self) -> str
    def resumable(self, launch: Launch) -> Launch
    def resumed(self, launch: Launch, prior: LaunchResult, prompt: str) -> Launch | None
    def scratch_parent(self) -> Path
    signals: Signals = codex_wire.SIGNALS
    def skill_home(self, launch: Launch) -> Path
    stop_hooks_file: str | None = HOOKS_FILE
    def tmp_args(self, launch: Launch) -> list[str]
    watched_limits: frozenset[str] = frozenset({'max_turns'})
    wire: ModuleType | None = codex_wire

@dataclass(frozen=True)
class CredentialBinding
    def __init__(
        self,
        env: dict[str, str] = dict(),
        args: tuple[str, ...] = (),
        isolated: bool = False,
        isolation: str = '',
    ) -> None
    args: tuple[str, ...] = ()
    env: dict[str, str] = field(default_factory=dict)
    isolated: bool = False
    isolation: str = ''

@dataclass
class Ending
    def __init__(
        self,
        killed: Stop | None = None,
        exit_code: int | None = None,
        saw_any: bool = False,
        text: str = '',
        error: str = '',
        status: int | None = None,
        failed: bool = False,
        auth: bool = False,
        refused: bool = False,
        aborted: bool = False,
        capped: Stop | None = None,
        completed: bool = False,
    ) -> None
    aborted: bool = False
    auth: bool = False
    capped: Stop | None = None
    completed: bool = False
    error: str = ''
    exit_code: int | None = None
    failed: bool = False
    killed: Stop | None = None
    refused: bool = False
    saw_any: bool = False
    status: int | None = None
    text: str = ''

class EventFold
    def __init__(self) -> None
    def add_call(self, call: CallUsage, cost: float | None = None) -> None
    agent: str = ''
    calls: list[CallUsage]
    context_window: int | None
    cost: float | None
    def ending(self, *, exit_code: int | None, killed: Stop | None) -> Ending
    def feed(self, ev: dict[str, Any]) -> Stop | None
    def feed_line(self, raw: bytes | str) -> Stop | None
    known_events: frozenset[str] | None = None
    def launch_result(
        self,
        launch: Launch,
        agent: str,
        stop: Stop,
        *,
        exit_code: int | None,
        **fields: Any,
    ) -> LaunchResult
    def note_tool(self, name: str, args: object = None) -> None
    def note_web_query(self, query: object) -> None
    def read(self, data: bytes) -> Self
    def result(
        self,
        launch: Launch,
        agent: str,
        *,
        exit_code: int | None,
        killed: Stop | None,
    ) -> LaunchResult
    saw_any
    session_id
    signals: Signals = Signals()
    def stop(self, *, exit_code: int | None, killed: Stop | None) -> Stop
    tool_args: dict[str, list[str]]
    tool_uses: Counter[str]
    def unknown(self, typ: str) -> bool
    unread: Counter[str]
    def usage(self) -> Usage
    def warn_unread(self) -> None
    web_queries: list[str]

@dataclass(frozen=True)
class Launch
    def __init__(
        self,
        prompt: str,
        label: str = '',
        role: str | None = None,
        model: str = '',
        effort: str | None = None,
        system_prompt: str | None = None,
        append_system_prompt: str | None = None,
        tools: tuple[str, ...] | None = None,
        allowed_tools: tuple[str, ...] = (),
        json_schema: dict[str, Any] | None = None,
        settings: str | None = None,
        setting_sources: str | None = None,
        mcp_config: str | None = None,
        strict_mcp: bool = True,
        persist_session: bool = False,
        exclude_dynamic_sections: bool = False,
        limits: Limits = Limits(),
        credential: Credential = AMBIENT,
        cwd: Path = Path.cwd(),
        env: dict[str, str] = dict(),
        stream_path: Path | None = None,
        permission_prompts: str | None = 'none',
        permission_mode: str | None = None,
        prompt_via_stdin: bool | None = None,
        plugin_dirs: tuple[str, ...] = (),
        preamble: str | None = None,
        context_files: tuple[str, ...] = (),
        session_id: str | None = None,
        fallback_model: str | None = None,
        resume: str | None = None,
        session_dir: Path | None = None,
    ) -> None
    allowed_tools: tuple[str, ...] = ()
    append_system_prompt: str | None = None
    context_files: tuple[str, ...] = ()
    credential: Credential = AMBIENT
    cwd: Path = field(default_factory=Path.cwd)
    effort: str | None = None
    env: dict[str, str] = field(default_factory=dict)
    exclude_dynamic_sections: bool = False
    fallback_model: str | None = None
    json_schema: dict[str, Any] | None = None
    label: str = ''
    limits: Limits = field(default_factory=Limits)
    mcp_config: str | None = None
    model: str = ''
    permission_mode: str | None = None
    permission_prompts: str | None = 'none'
    persist_session: bool = False
    plugin_dirs: tuple[str, ...] = ()
    preamble: str | None = None
    prompt: str
    prompt_via_stdin: bool | None = None
    resume: str | None = None
    role: str | None = None
    session_dir: Path | None = None
    session_id: str | None = None
    setting_sources: str | None = None
    settings: str | None = None
    stream_path: Path | None = None
    strict_mcp: bool = True
    system_prompt: str | None = None
    tools: tuple[str, ...] | None = None

@dataclass
class LaunchResult
    def __init__(
        self,
        label: str,
        agent: str = '',
        stop: Stop = Stop.LAUNCH_FAILED,
        ok: bool = False,
        started_at: str = '',
        wall_s: float = 0.0,
        exit_code: int | None = None,
        cli_version: str = '',
        argv: list[str] = list(),
        model: str = '',
        credential: str = 'ambient',
        api_key_source: str = '',
        session_id: str = '',
        num_turns: int | None = None,
        duration_api_ms: int | None = None,
        result_text: str = '',
        structured_output: dict[str, Any] | list[Any] | None = None,
        cost_usd: float | None = None,
        usage: Usage = Usage(),
        other_models: dict[str, OtherModel] = dict(),
        quota: list[QuotaReading] = list(),
        quota_refusal: QuotaRefusal | None = None,
        permission_denials: list[dict[str, Any]] = list(),
        tool_uses: dict[str, int] = dict(),
        web_queries: list[str] = list(),
        hollow: bool = False,
        unread_events: dict[str, int] = dict(),
        init_model: str = '',
        init_tools: int = 0,
        init_errors: list[dict[str, Any]] = list(),
        api_retries: int = 0,
        retry_errors: dict[str, int] = dict(),
        stderr_tail: str = '',
        raw_result: dict[str, Any] = dict(),
        rounds: int | None = None,
        round_errors: list[list[str]] = list(),
        round_costs: list[float | None] = list(),
        unwrapped: bool = False,
        hook_errors: list[list[str]] = list(),
        context_budget: int | None = None,
        context_warned: bool = False,
        context_window: int | None = None,
        retryable: bool = True,
    ) -> None
    agent: str = ''
    api_key_source: str = ''
    api_retries: int = 0
    argv: list[str] = field(default_factory=list)
    def as_dict(self) -> dict[str, Any]
    cli_version: str = ''
    context_budget: int | None = None
    @property context_pct: float | None
    context_warned: bool = False
    context_window: int | None = None
    cost_usd: float | None = None
    credential: str = 'ambient'
    duration_api_ms: int | None = None
    exit_code: int | None = None
    hollow: bool = False
    hook_errors: list[list[str]] = field(default_factory=list)
    init_errors: list[dict[str, Any]] = field(default_factory=list)
    init_model: str = ''
    init_tools: int = 0
    label: str
    model: str = ''
    num_turns: int | None = None
    ok: bool = False
    other_models: dict[str, OtherModel] = field(default_factory=dict)
    @property peak_context: int | None
    permission_denials: list[dict[str, Any]] = field(default_factory=list)
    quota: list[QuotaReading] = field(default_factory=list)
    quota_refusal: QuotaRefusal | None = None
    raw_result: dict[str, Any] = field(default_factory=dict)
    def record(self, path: Path) -> None
    result_text: str = ''
    retry_errors: dict[str, int] = field(default_factory=dict)
    retryable: bool = True
    round_costs: list[float | None] = field(default_factory=list)
    round_errors: list[list[str]] = field(default_factory=list)
    rounds: int | None = None
    session_id: str = ''
    started_at: str = ''
    stderr_tail: str = ''
    stop: Stop = Stop.LAUNCH_FAILED
    structured_output: dict[str, Any] | list[Any] | None = None
    tool_uses: dict[str, int] = field(default_factory=dict)
    unread_events: dict[str, int] = field(default_factory=dict)
    unwrapped: bool = False
    usage: Usage = field(default_factory=Usage)
    wall_s: float = 0.0
    web_queries: list[str] = field(default_factory=list)

class LimitsWatch
    def __init__(self, limits: Limits, watched: frozenset[str], fold: EventFold)
    def __call__(self) -> Stop | None
    budget: float | None
    context: int | None
    fold
    turns: int | None

class ModelChains(Protocol)
    def describe(self, model: str) -> str

class OmpBackend(Backend)
    def __init__(
        self,
        binary: str = '',
        base_env: dict[str, str] | None = None,
        thinking: str = 'medium',
        trace: bool = True,
    )
    def account(self, credential: Credential, model: str = '') -> str
    def argv(self, launch: Launch) -> list[str]
    binary = 'omp'
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    drivable_model_chain = False
    effort_levels: tuple[str, ...] | None = ('off', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max', 'auto')
    def env(self, launch: Launch) -> dict[str, str]
    def fold(self, launch: Launch) -> EventFold
    light_launch = True
    def model_chains(self, cwd: Path | None = None) -> ModelChains
    name = 'omp'
    native_model_chain = True
    needs_inline_role_file = True
    plaintext_stream = True
    def read_models(self) -> dict[str, int | None] | None
    signals: Signals = omp_wire.SIGNALS
    stop_hooks_file: str | None = HOOKS_FILE
    thinking: str = thinking
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None
    trace: bool = trace
    def usage_reading(self, launch: Launch) -> QuotaReading | None
    watched_limits: frozenset[str] = frozenset({'max_budget_usd', 'max_turns'})
    wire: ModuleType | None = omp_wire

class OmpRpcBackend(OmpBackend)
    def argv(self, launch: Launch) -> list[str]
    def exchange(
        self,
        launch: Launch,
        argv: list[str],
        env: dict[str, str],
        fold: EventFold,
        watch: LimitsWatch,
    ) -> tuple[bytes, str, int | None, Stop | None]
    def fold(self, launch: Launch) -> EventFold
    name = 'omp-rpc'
    plaintext_stream = False
    signals: Signals = dataclasses.replace(omp_wire.SIGNALS, stderr_stops=FAILED_STOPS | {Stop.LAUNCH_FAILED})

class OpencodeBackend(OpencodeKeyAccount, Backend)
    PERMISSION_OF: Mapping[str, str] = MappingProxyType({'Bash': 'bash', 'Read': 'read', 'Write': 'edit', 'Edit': 'edit', 'MultiEdit': 'edit', 'NotebookEdit': 'edit', 'Glob': 'glob', 'Grep': 'grep', 'LS': 'list', 'WebFetch': 'webfetch', 'WebSearch': 'websearch', 'Skill': 'skill', 'Agent': 'task', 'Task': 'task', 'TodoWrite': 'todowrite'})
    def account(self, credential: Credential, model: str = '') -> str
    def argv(self, launch: Launch) -> list[str]
    def billed_key(self, credential: Credential, model: str) -> tuple[str, str]
    binary = 'opencode'
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    catalogue_timeout_s: float = 60.0
    def config(self, launch: Launch) -> dict[str, Any]
    effort_levels: tuple[str, ...] | None = None
    def env(self, launch: Launch) -> dict[str, str]
    def fold(self, launch: Launch) -> EventFold
    light_launch = True
    name = 'opencode'
    def read_models(self) -> dict[str, int | None] | None
    def reclassify(self, launch_result: LaunchResult, stderr: str) -> None
    signals: Signals = opencode_wire.SIGNALS
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None
    def usage_reading(self, launch: Launch) -> QuotaReading | None
    def watch_stderr(self, launch: Launch) -> Sink | None
    watched_limits: frozenset[str] = frozenset({'max_budget_usd', 'context_budget'})
    wire: ModuleType | None = opencode_wire

class PiBackend(OpencodeKeyAccount, Backend)
    def __init__(
        self,
        binary: str = '',
        base_env: dict[str, str] | None = None,
        thinking: str = 'medium',
        agent_dirs: Path | None = None,
    )
    def account(self, credential: Credential, model: str = '') -> str
    agent_dirs: Path
    def argv(self, launch: Launch) -> list[str]
    def billed_key(self, credential: Credential, model: str) -> tuple[str, str]
    binary = 'pi'
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    effort_levels: tuple[str, ...] | None = ('off', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max')
    def env(self, launch: Launch) -> dict[str, str]
    def fold(self, launch: Launch) -> EventFold
    light_launch = True
    name = 'pi'
    needs_inline_role_file = True
    def read_models(self) -> dict[str, int | None] | None
    signals: Signals = pi_wire.SIGNALS
    thinking: str = thinking
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None
    def usage_reading(self, launch: Launch) -> QuotaReading | None
    watched_limits: frozenset[str] = frozenset({'max_budget_usd'})
    wire: ModuleType | None = pi_wire

@dataclass(frozen=True)
class QuotaRefusal
    def __init__(self, retry_after_s: float | None = None) -> None
    retry_after_s: float | None = None

class SchemaRef(BaseModel)
    module: str
    qualname: str
    search_path: str | None = None

@dataclass(frozen=True)
class Signals
    def __init__(
        self,
        quota: re.Pattern[str] = QUOTA_RE,
        auth: re.Pattern[str] | None = None,
        refusal: re.Pattern[str] | None = None,
        status: re.Pattern[str] | None = None,
        status_decides: bool = False,
        silence_is_startup: bool = False,
        nonzero_exit_is_error: bool = True,
        stderr_stops: frozenset[Stop] = FAILED_STOPS,
    ) -> None
    auth: re.Pattern[str] | None = None
    def auth_in(self, text: str) -> bool
    nonzero_exit_is_error: bool = True
    quota: re.Pattern[str] = QUOTA_RE
    def quota_in(self, text: str) -> bool
    refusal: re.Pattern[str] | None = None
    def refusal_in(self, text: str) -> bool
    silence_is_startup: bool = False
    status: re.Pattern[str] | None = None
    status_decides: bool = False
    def status_in(self, text: str) -> int | None
    stderr_stops: frozenset[Stop] = field(default=FAILED_STOPS)

class Sink(Protocol)
    def __call__(self, line: bytes, /) -> Stop | None

class Stop(str, enum.Enum)
    ABORTED = 'aborted'
    API_ERROR = 'api_error'
    AUTH_FAILED = 'auth_failed'
    BUDGET = 'budget'
    COMPLETED = 'completed'
    CONTEXT = 'context_budget'
    CREDENTIAL_MISMATCH = 'credential_mismatch'
    DENIED = 'denied'
    ERROR = 'error'
    INVALID_INPUT = 'invalid_input'
    INVALID_OUTPUT = 'invalid_output'
    LAUNCH_FAILED = 'launch_failed'
    MAX_TURNS = 'max_turns'
    NO_OUTPUT = 'no_output'
    RATE_LIMITED = 'rate_limited'
    REFUSED = 'refused'
    STALLED = 'stalled'
    TIMEOUT = 'timeout'

@dataclass
class Usage
    def __init__(self, calls: list[CallUsage] = list(), output_total: int = 0) -> None
    def as_dict(self) -> dict[str, Any]
    calls: list[CallUsage] = field(default_factory=list)
    @property context_total: int
    output_total: int = 0
    @property peak_context: int
    @property resident: int
    def totals(self, context_window: int | None = None) -> TokenUsage | None

def create_backend(name: str, **kwargs: Any) -> Backend

def known_agents() -> tuple[str, ...]

def normalize_stream_line(format: str, raw_line: str) -> list[dict[str, Any]]

def register_backend(name: str, cls: type[Backend]) -> None

def register_normalizer(
    format: str,
    normalize_event: Callable[[dict[str, Any]], dict[str, Any] | list[dict[str, Any]] | None],
    plaintext: Callable[[str], dict[str, Any] | None] | None = None,
) -> None

## contraagent.engine

class AccountOf(Protocol)
    def __call__(self, credential: str, /) -> str

class AgentRunner
    def __init__(
        self,
        backends: Backend | Mapping[str, Backend],
        configs: Mapping[str, KindConfig],
        *,
        run_dir: Path | str,
        run_id: str = '',
        adapter: TaskAdapter | None = None,
        routers: Mapping[str, Router] | None = None,
        model: str = '',
        model_failover: bool = False,
        credentials: Mapping[str, Credential] | None = None,
        env: Mapping[str, str] | None = None,
        keep_transcripts: bool = False,
        cwd: Path | str | None = None,
        scopes: Sequence[TaskScope] = (),
        enrichers: Sequence[ResultEnricher] = (),
        rests: QuotaRests | None = None,
    )
    def account_of(self, backend: Backend, credential: Credential, model: str) -> str
    adapter: TaskAdapter
    def backend_for_route(self, task: Task, route: Route | None) -> Backend
    backends: dict[str, Backend]
    def config(self, task: Task) -> KindConfig
    configs: Mapping[str, KindConfig]
    def credential_for(self, task: Task, route: Route | None) -> Credential
    credentials: dict[str, Credential]
    cwd: Path
    def deferral(self, task: Task) -> Deferral | None
    enrichers: Sequence[ResultEnricher] = tuple(enrichers)
    env: dict[str, str]
    keep_transcripts
    def launch(
        self,
        task: Task,
        route: Route | None = None,
        backend: Backend | None = None,
        credential: Credential | None = None,
    ) -> Launch
    model
    model_failover
    pools: dict[str, CredentialPool]
    rests
    def router_for(self, task: Task) -> Router | None
    routers: dict[str, Router]
    def run(self, task: Task) -> TaskResult
    run_dir: Path
    run_id
    scopes: Sequence[TaskScope] = tuple(scopes)

class Answer(Protocol)
    def __call__(self, item: Any, result: TaskResult, /) -> Any

@dataclass(frozen=True)
class Before
    def __init__(
        self,
        stamp: tuple[int, int] | None,
        preserved: Mapping[str, Any] = dict(),
        text: str | None = None,
    ) -> None
    preserved: Mapping[str, Any] = field(default_factory=dict)
    def restore(self, path: Path) -> None
    stamp: tuple[int, int] | None
    text: str | None = None
    def value(self, schema: type[BaseModel] | None) -> Any

@dataclass
class Check
    def __init__(
        self,
        path: Path,
        ok: bool,
        errors: list[str],
        value: Any = None,
        schema_ok: bool = False,
    ) -> None
    errors: list[str]
    ok: bool
    path: Path
    schema_ok: bool = False
    value: Any = None

class ClaimCapacity(Protocol)
    def allowed(self, held: int) -> int
    def started(self) -> None

class Created(Protocol)
    def __call__(self) -> Iterable[Task]

class Cursor(BaseModel)
    completed_stages: list[str] = Field(default_factory=list)
    history: list[HistoryEntry] = Field(default_factory=list)
    markers: dict[str, list[dict[str, Any]]] = Field(default_factory=dict)
    model_config: ConfigDict = ConfigDict(extra='forbid', validate_assignment=True)
    run: str
    schema_version: str
    stage: str | None = None
    units: dict[str, UnitCursor] = Field(default_factory=dict)
    updated_at: datetime

class CursorStore
    def __init__(
        self,
        path: str | Path,
        run: str = '',
        *,
        stages: Iterable[object] | None = None,
        steps: Iterable[object] | None = None,
        skip_when_completed: Iterable[object] = (),
    ) -> None
    def advance_stage(self, name: str, at: datetime | None = None) -> Cursor
    def advance_step(
        self,
        unit: str,
        step: str,
        at: datetime | None = None,
        loop_pass: int | None = None,
    ) -> Cursor
    def clear_inflight(self, unit: str) -> Cursor
    def load(self) -> Cursor
    path: Path
    def read(self) -> Cursor | None
    def record(self, marker_key: str, entry: Mapping[str, Any]) -> Cursor
    def resume_view(self) -> dict[str, Any]
    run
    def set_inflight(self, unit: str, item_ids: list[str], at: datetime | None = None) -> Cursor
    def steps_to_run(self, unit: str) -> list[str]

@dataclass(frozen=True)
class Deferral
    def __init__(self, reason: str, cause: DeferralCause) -> None
    cause: DeferralCause
    reason: str

@dataclass(eq=False)
class FlowCall
    def __init__(
        self,
        id: str,
        gen: Generator[Any, Any, Any],
        parent: FlowCall | None = None,
        function: str = '',
        site: tuple[str, int] | None = None,
        state: FlowState = 'running',
        frontier: tuple[str, ...] = (),
        value: Any = None,
        error: BaseException | None = None,
        time_box: TimeBox | None = None,
        started_at: float = time.time(),
        _children: itertools.count = (lambda: itertools.count(1))(),
    ) -> None
    error: BaseException | None = None
    frontier: tuple[str, ...] = ()
    function: str = ''
    gen: Generator[Any, Any, Any] = field(repr=False)
    id: str
    parent: FlowCall | None = field(default=None, repr=False)
    site: tuple[str, int] | None = None
    started_at: float = field(default_factory=time.time)
    state: FlowState = 'running'
    time_box: TimeBox | None = None
    value: Any = field(default=None, repr=False)

class FlowDriver
    def __init__(
        self,
        *,
        waits_for: WaitsFor | None = None,
        answer: Answer | None = None,
        created: Created | None = None,
        on_flow_call: OnFlowCall | None = None,
        past_grace: Callable[[], bool] | None = None,
        slow_step_s: float = 1.0,
    ) -> None
    calls: dict[str, FlowCall]
    current: FlowCall | None
    done: dict[str, TaskResult]
    def finish(self) -> None
    def follow_up(self, result: TaskResult) -> list[Task]
    past_grace: Callable[[], bool] | None
    def run(self, queue: WorkQueue, *flows: Generator[Any, Any, Any]) -> dict[str, Any]
    slow_step_s
    def start(self, flow: Generator[Any, Any, Any]) -> list[Task]
    def time_boxes(self, call: FlowCall | None) -> tuple[TimeBox, ...]
    @property waiting: list[FlowCall]

class FollowUp(Protocol)
    def __call__(self, result: TaskResult, /) -> list[Task]

class HistoryEntry(BaseModel)
    at: datetime
    model_config: ConfigDict = ConfigDict(extra='forbid', validate_assignment=True)
    name: str

class Inflight(BaseModel)
    item_ids: list[str] = Field(default_factory=list)
    model_config: ConfigDict = ConfigDict(extra='forbid', validate_assignment=True)
    pid: int | None = None
    since: datetime
    start_ticks: int | None = None

@dataclass(frozen=True)
class InputContract(_FileContract)
    def __init__(
        self,
        path: Path,
        schema: type[BaseModel] | None,
        required: bool = True,
        producer: str = '',
        producer_failure: Callable[[], str] | None = None,
    ) -> None
    path: Path
    producer: str = ''
    producer_failure: Callable[[], str] | None = field(default=None, compare=False)
    required: bool = True
    def resolve(self, cwd: Path) -> Path
    schema: type[BaseModel] | None

class JournalEstimates
    def __init__(
        self,
        runs_dir: str | Path = 'runs',
        *,
        recent: int = RECENT_RUNS,
        loop_cap: int = LOOP_CAP,
    ) -> None
    def estimate(self, task: Task) -> float | None
    def flow_call_ended(self, call_id: str, state: str) -> None
    loop_cap
    def observe_attempt(self, row: dict[str, Any]) -> None
    def observe_plan(self, row: dict[str, Any]) -> None
    def tail(self, task: Task) -> float | None

class OnEvent(Protocol)
    def __call__(self, event: str, task: Task | None, /, **fields: Any) -> None

class OnFlowCall(Protocol)
    def __call__(self, event: str, call: FlowCall, /) -> None

class OnResult(Protocol)
    def __call__(self, result: TaskResult, attempt: int, /) -> None

@dataclass(frozen=True)
class OutputContract(_FileContract)
    def __init__(
        self,
        path: Path,
        schema: type[BaseModel] | None,
        fresh: bool = True,
        validators: tuple[Validator, ...] = (),
        params: Callable[[], Mapping[str, Any]] | None = None,
        call: Path | None = None,
        from_message: bool = False,
        unwrap: bool = False,
        preserve: tuple[str, ...] = (),
    ) -> None
    def before(self, cwd: Path) -> Before
    call: Path | None = None
    def check(self, cwd: Path, before: Before | None = None) -> Check
    @property compares: bool
    fresh: bool = True
    from_message: bool = False
    def message_path(self, cwd: Path) -> Path
    params: Callable[[], Mapping[str, Any]] | None = None
    path: Path
    preserve: tuple[str, ...] = ()
    def resolve(self, cwd: Path) -> Path
    schema: type[BaseModel] | None
    unwrap: bool = False
    validators: tuple[Validator, ...] = ()

class PythonTaskRunner
    def __init__(
        self,
        inner: TaskRunner | None = None,
        *,
        adapter: TaskAdapter | None = None,
        run_dir: str | Path | None = None,
        cwd: str | Path | None = None,
        on_failed_attempt: Callable[[TaskResult], None] | None = None,
        scopes: Sequence[TaskScope] = (),
        enrichers: Sequence[ResultEnricher] = (),
        config: PipelineConfig | None = None,
    ) -> None
    def config(self, task: Task) -> RunnerConfig
    def contracts(self, task: Task) -> tuple[list[Any], list[Any]]
    def deferral(self, task: Task) -> Deferral | None
    def run(self, task: Task) -> TaskResult

class QuotaRests
    def __init__(
        self,
        credentials: Mapping[str, Credential],
        rule: QuotaRule | None = None,
        *,
        finishing_kinds: Iterable[str] = (),
        grace_s: float = GRACE_S,
        on_change: Callable[[str, Rest], None] | None = None,
        now: Callable[[], datetime] = lambda: datetime.now(UTC),
        clock: Callable[[], float] = time.monotonic,
    ) -> None
    credentials: dict[str, Credential]
    finishing_kinds: frozenset[str]
    grace_s
    def latest(self, name: str, account: str | None = None) -> QuotaReading | None
    def observe(self, reading: QuotaReading) -> None
    on_change
    def reason(self, name: str, kind: str = '', account: AccountOf | None = None) -> str
    def resting_members(self, pool: str, kind: str = '', account: AccountOf | None = None) -> set[str]
    def rests(self, name: str, kind: str = '', account: AccountOf | None = None) -> list[Rest]
    rule

class Rest(BaseModel)
    account: str = ''
    cap: float | None = None
    credential: str
    model_config: ConfigDict = ConfigDict(extra='allow', frozen=True)
    pct: float | None = None
    reading: dict[str, Any] | None = None
    rule: str = ''
    def skipped(self, route: str, model: str) -> SkippedHop
    def text(self) -> str
    def to_dict(self) -> dict[str, Any]
    until: str = ''
    window: Literal['', Window] = ''

class RetryingRunner
    def __init__(
        self,
        runner: AgentRunner,
        max_concurrent_overrides: dict[str, Any] | None = None,
        on_attempt: Callable[[TaskResult], None] | None = None,
        on_exhausted: Callable[[Task, TaskResult], None] | None = None,
    )
    backoff: Callable[[float, int], float] = staticmethod(backoff_s)
    def config(self, task: Task) -> KindConfig
    def deferral(self, task: Task) -> Deferral | None
    def run(self, task: Task) -> TaskResult

class Router
    def __init__(
        self,
        primary: Route,
        fallbacks: tuple[Route, ...] = (),
        probe_interval_s: float = 900.0,
        max_probe_interval_s: float = 7200.0,
        clock: Callable[[], float] = time.monotonic,
    ) -> None
    def active(self) -> Route | None
    def chain(self) -> str
    def describe(self) -> str
    @property exhausted: bool
    def history(self) -> list[tuple[str, str]]
    def hop_of(self, route: Route) -> int
    def label_of(self, route: Route) -> str
    max_probe_interval_s
    def next_route_after(self, route: Route) -> Route | None
    def now(self) -> float
    def on_credential(self, credential: str) -> Router
    def other_agents_after(self, route: Route) -> list[Route]
    def pick(self) -> Route | None
    primary
    probe_interval_s
    def rate_limited(self, route: Route | None, retry_after_s: float | None = None) -> str
    routes: tuple[Route, ...]
    def same_agent_after(self, route: Route) -> list[Route]
    def succeeded(self, route: Route | None, started: float | None = None) -> None

@dataclass
class RoutingPlan
    def __init__(
        self,
        config: PipelineConfig,
        backends: Mapping[str, Backend],
        configs: dict[str, KindConfig] = dict(),
        routers: dict[str, Router] = dict(),
        overrides: Overrides = Overrides(),
        credentials: dict[str, Credential] = dict(),
    ) -> None
    backends: Mapping[str, Backend]
    @classmethod
    def build(
        cls,
        config: PipelineConfig,
        kinds: Iterable[Any],
        backends: Mapping[str, Backend],
        overrides: Overrides = Overrides(),
        *,
        declared: Mapping[str, Mapping[str, Any]] | None = None,
        bases: Mapping[str, str] | None = None,
        requirements: Mapping[str, Iterable[str]] | None = None,
        light: Iterable[str] = (),
        budgets_required: bool = True,
        all_kinds: bool = True,
    ) -> RoutingPlan
    def call_problems(self, kind: str, credential: str) -> list[str]
    config: PipelineConfig
    configs: dict[str, KindConfig] = field(default_factory=dict)
    credentials: dict[str, Credential] = field(default_factory=dict)
    def log(self) -> None
    overrides: Overrides = Overrides()
    def preflight(self, policy: PreflightPolicy) -> PreflightRecord
    routers: dict[str, Router] = field(default_factory=dict)
    def routes(self) -> dict[str, str]

class Run
    def __init__(
        self,
        run_id: str,
        runs_dir: str | Path = 'runs',
        *,
        run_group: str = '',
        lock: bool = True,
        fresh: bool = False,
        row_extras: Callable[[TaskResult], dict[str, Any]] | None = None,
        unit_of: Callable[[Task], str] | None = None,
        log: Callable[[str], None] | None = None,
        cwd: str | Path | None = None,
        config: PipelineConfig | None = None,
    )
    def agent_runner(
        self,
        backends: Backend | Mapping[str, Backend],
        configs: Mapping[str, KindConfig],
        *,
        adapter: TaskAdapter | None = None,
        routers: Mapping[str, Router] | None = None,
        model: str = '',
        model_failover: bool = False,
        credentials: Mapping[str, Credential] | None = None,
        env: Mapping[str, str] | None = None,
        keep_transcripts: bool = False,
        cwd: Path | str | None = None,
        scopes: Sequence[TaskScope] = (),
        enrichers: Sequence[ResultEnricher] = (),
        rests: QuotaRests | None = None,
    ) -> AgentRunner
    def close(self) -> None
    config
    def cursor(
        self,
        *,
        stages: Iterable[object] | None = None,
        steps: Iterable[object] | None = None,
        skip_when_completed: Iterable[object] = (),
    ) -> CursorStore
    dir: Path
    def estimator(self) -> JournalEstimates
    def finish(self) -> None
    def install_signal_handlers(self) -> None
    journal: Journal
    def on_teardown(self, callback: Callable[[], None]) -> None
    def progress(self, line: str, *, warning: bool = False) -> None
    def python_runner(
        self,
        inner: TaskRunner | None = None,
        adapter: TaskAdapter | None = None,
        cwd: str | Path | None = None,
        *,
        scopes: Sequence[TaskScope] = (),
        enrichers: Sequence[ResultEnricher] = (),
    ) -> PythonTaskRunner
    def queue(
        self,
        runner: TaskRunner,
        *,
        follow_up: FollowUp | None = None,
        resume: bool = False,
        max_concurrent: int = MAX_CONCURRENT,
        deadline_s: float | None = None,
        retry_wait_s: float = RETRY_WAIT_S,
        throttle_s: float = THROTTLE_S,
        skip: Callable[[Task], bool] | None = None,
        grace_s: float = GRACE_S,
        retry_on: frozenset[str] = RETRY_ON,
        throttle_on: frozenset[str] = THROTTLE_ON,
        finishing_kinds: Iterable[str] = (),
        claim_capacity: Mapping[str, int | PerTask] | None = None,
        estimate: Callable[[Task], float | None] | None = None,
        overdue_s: float = OVERDUE_S,
        slack_s: float | None = None,
    ) -> WorkQueue
    def quota_rests(
        self,
        credentials: Mapping[str, Credential],
        rule: QuotaRule | None = None,
        *,
        finishing_kinds: Iterable[str] = (),
        grace_s: float = GRACE_S,
    ) -> QuotaRests
    def record_attempt(self, result: TaskResult) -> None
    def record_flow_call(
        self,
        event: str,
        call_id: str,
        *,
        parent: str | None = None,
        function: str = '',
        flow: Sequence[Frame] = (),
        state: FlowState | None = None,
        error: BaseException | None = None,
        time_box: TimeBox | None = None,
        started_at: float | None = None,
    ) -> None
    def rows(self) -> list[dict[str, Any]]
    run_group: str
    run_id: str
    def run_python_steps(self, steps: list[tuple[str, Callable[[], None]]]) -> None
    runs_dir: Path
    def stop_on_signals(self) -> None
    def teardown(self) -> None
    def update_info(self, **fields: Any) -> None

class RunnerConfig(Protocol)
    @property max_concurrent: int | None

class TaskAdapter
    def changes(self, task: Task) -> Sequence[str]
    def context_files(self, task: Task, backend: Backend) -> tuple[str, ...]
    def definition(self, task: Task) -> dict[str, Any] | None
    def env(self, task: Task) -> dict[str, str]
    def input_contracts(self, task: Task) -> Sequence[InputContract]
    def mcp_config_path(self, task: Task) -> str | None
    def missing_params(self, task: Task) -> list[str]
    def output_contracts(self, task: Task) -> Sequence[OutputContract]
    def plugin_dirs(self, task: Task) -> tuple[str, ...]
    def prompt(self, task: Task, backend: Backend) -> str
    def role(self, task: Task) -> str | None
    def settings_path(self, task: Task) -> str | None
    def system_prompt(self, task: Task) -> str | None
    def tools(self, task: Task) -> tuple[str, ...] | None

class TaskRunner(Protocol)
    def config(self, task: Task) -> RunnerConfig
    def run(self, task: Task) -> TaskResult

class UnitCursor(BaseModel)
    completed: list[str] = Field(default_factory=list)
    history: list[HistoryEntry] = Field(default_factory=list)
    inflight: Inflight | None = None
    loop_pass: int = 0
    model_config: ConfigDict = ConfigDict(extra='forbid', validate_assignment=True)
    stage: str | None = None
    step: str | None = None

class WaitsFor(Protocol)
    def __call__(self, item: Any, /) -> Task

class WorkQueue
    def __init__(
        self,
        runner: TaskRunner,
        *,
        follow_up: FollowUp | None = None,
        on_event: OnEvent | None = None,
        on_result: OnResult | None = None,
        max_concurrent: int = MAX_CONCURRENT,
        deadline_s: float | None = None,
        retry_wait_s: float = RETRY_WAIT_S,
        throttle_s: float = THROTTLE_S,
        skip: Callable[[Task], bool] | None = None,
        grace_s: float = GRACE_S,
        retry_on: frozenset[str] = frozenset(),
        throttle_on: frozenset[str] = frozenset(),
        finishing_kinds: tuple[str, ...] = (),
        claim_capacity: Mapping[str, int | ClaimCapacity] | None = None,
        estimate: Callable[[Task], float | None] | None = None,
        overdue_s: float = OVERDUE_S,
        slack_s: float | None = None,
    )
    def add(self, task: Task) -> bool
    def capacity(self, name: str) -> int
    claim_capacity: dict[str, int | ClaimCapacity]
    deadline_at: float | None
    deferred: list[Task]
    def extend(self, tasks: Iterable[Task]) -> None
    finishing_kinds: tuple[str, ...]
    follow_up: FollowUp
    grace_s
    hard_stop_at: float | None
    def in_endgame(self) -> bool
    max_concurrent: int
    on_event: OnEvent
    overdue_s
    def past_deadline(self) -> bool
    def past_hard_stop(self) -> bool
    results: list[TaskResult]
    retry_on: frozenset[str]
    retry_wait_s
    def run(self) -> dict[str, Any]
    runner
    slack_s: float
    def slots(self) -> int
    throttle_on: frozenset[str]
    throttle_s
    time_box: TimeBox | None

def run_until_valid(
    backend: Backend,
    launch: Launch,
    contracts: Sequence[OutputContract],
    max_corrections: int = DEFAULT_MAX_CORRECTIONS,
) -> LaunchResult

## contraagent.plugins

@dataclass(frozen=True)
class Plugin
    def __init__(
        self,
        root: Path,
        name: str,
        hooks: tuple[dict[str, Any], ...] = (),
        agents: tuple[str, ...] = (),
    ) -> None
    agents: tuple[str, ...] = ()
    def besides_skills(self) -> list[str]
    def definition(self, role: str | None, limit: int = 40000) -> dict[str, Any]
    hooks: tuple[dict[str, Any], ...] = ()
    def lint(self) -> list[str]
    @classmethod
    def load(cls, root: str | Path) -> Plugin
    name: str
    def render_leftovers(self) -> list[str]
    def resolve_role(self, role: str | None) -> str | None
    def role_file(self, role: str) -> Path
    @property roles: tuple[str, ...]
    root: Path
    def skill_files(self) -> list[tuple[str, str]]
    @property skills_only: bool

def new_plugin(directory: str | Path, name: str = '') -> list[Path]

## contraagent.monitor

class BlocksBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    blocks: list[PanelBody]
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['blocks'] = 'blocks'

CODE_CHARACTERS = 500000

CODE_LINES = 5000

class ChartSeries(Closed)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str
    values: list[float | None]

class CodeBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    highlight: list[int | tuple[int, int]] = Field(default_factory=list, description='lines marked: a line, or a range, both ends included')
    language: str = Field(default='', description="a name of `LANGUAGES` or a file extension; '': plain text")
    model_config: ConfigDict = ConfigDict(extra='forbid')
    path: str | None = Field(default=None, description='the file the text is, named over it')
    start: int = Field(default=1, ge=0, description='the number of the first line')
    text: str
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['code'] = 'code'
    wrap: bool = Field(default=False, description='long lines wrap; otherwise they scroll sideways')

class ColumnsBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    series: list[ChartSeries]
    stacked: bool = False
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['columns'] = 'columns'
    unit: ValueFormat | None = None
    x: list[str]

class DirectorySource
    def __init__(self, root: str | Path) -> None
    def activity(self, run: str) -> str | None
    def dir(self, run: str) -> Path | None
    def document(self, run: str, name: str) -> str | None
    def documents(self, run: str) -> list[str]
    def exists(self, run: str, name: str) -> bool
    def file(self, run: str, path: str) -> RunFile | None
    def file_stat(self, run: str, path: str) -> Stat | None
    def folder(self, run: str, folder: str) -> list[str]
    def info(self, run: str) -> dict[str, Any] | None
    def keep(self, run: str, text: str) -> None
    def kept(self, run: str) -> str | None
    def lines(
        self,
        run: str,
        log: str,
        since: int = 0,
        limit: int | None = None,
        max_bytes: int | None = None,
    ) -> tuple[list[str], int]
    def lines_at(
        self,
        run: str,
        log: str,
        at: LogPosition,
        limit: int | None = None,
        max_bytes: int | None = None,
    ) -> tuple[list[str], LogPosition]
    def logs(self, run: str) -> list[str]
    def path(self, run: str, name: str = '') -> Path
    def position(self, run: str, log: str, line: int) -> LogPosition
    root: Path
    def runs(self) -> list[str]
    def size(self, run: str, log: str) -> int
    def stat(self, run: str, name: str = '') -> Stat | None
    async def wait(self, run: str, log: str, beyond: int, timeout: float) -> int

@dataclass
class ExportReport
    def __init__(self, out: Path, runs: list[str], bytes_by_run: dict[str, int], total_bytes: int) -> None
    bytes_by_run: dict[str, int]
    out: Path
    runs: list[str]
    total_bytes: int

class GridCell(Closed)
    label: str = Field(min_length=1)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tip: str | None = None
    tone: Tone

class HBar(Closed)
    label: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tone: Tone | None = None
    value: float

class HBarsBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    bars: list[HBar]
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['hbars'] = 'hbars'
    unit: ValueFormat | None = None

class HeatmapBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['heatmap'] = 'heatmap'
    unit: ValueFormat | None = None
    values: list[list[float | None]]
    x: list[str]
    y: list[str]

LANGUAGES = ('python', 'json', 'javascript', 'typescript', 'c', 'cpp', 'java', 'go', 'rust', 'shell', 'toml', 'yaml', 'sql', 'xml', 'diff')

class LinesBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    series: list[ChartSeries]
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['lines'] = 'lines'
    unit: ValueFormat | None = None
    x: list[str]

class Liveness(Protocol)
    def hosts(self) -> list[dict[str, Any]]
    def live(self) -> set[str]
    def snapshot(self, run: str) -> dict[str, Any]
    def unresponsive(self) -> set[str]

class LogPosition(NamedTuple)
    def after(self, lines: Sequence[str]) -> LogPosition
    line: int
    offset: int

class Monitor
    def __init__(
        self,
        runs_dir: str | Path = 'runs',
        *,
        source: RunSource | None = None,
        liveness: Liveness | None = None,
        many_hosts: bool = False,
        title: str = 'contraagent',
        panels: Mapping[str, Panel | PanelFn] | None = None,
        task_detail: TaskDetailFn | None = None,
        static_dir: str | Path | None = None,
        stream_formats: dict[str, str] | None = None,
        groups: list[dict[str, Any]] | None = None,
        keep_digests: bool = True,
    )
    def check_run(self, run: str) -> str
    def cursor(self, run: str) -> dict[str, Any] | None
    def definitions(self, run: str) -> list[dict[str, Any]]
    def digest(self, run: str) -> dict[str, Any]
    group_defs: list[dict[str, Any]]
    def hosts(self) -> dict[str, Any]
    def journal(self, run: str, since: int = 0) -> tuple[list[dict[str, Any]], int]
    keep_digests
    liveness: Liveness
    many_hosts
    def outputs(self, run: str) -> dict[str, Any]
    def panel(self, name: str, run: str) -> dict[str, Any]
    def panel_info(self) -> list[dict[str, Any]]
    panels: dict[str, Panel]
    def plan(self, run: str, live: dict[str, Any] | None = None) -> dict[str, Any]
    def plan_rows(self, run: str) -> list[dict[str, Any]]
    def procs(self, run: str) -> dict[str, Any]
    def run_panels(self) -> list[str]
    def runs(self) -> list[dict[str, Any]]
    runs_dir: Path
    source: RunSource
    static_dir: Path | None
    def stream(self, run: str, key: str, since: int, limit: int = 1500) -> tuple[dict[str, Any], int]
    stream_formats: dict[str, str]
    def stream_from(
        self,
        run: str,
        key: str,
        since: int | None,
        limit: int = 1500,
    ) -> tuple[dict[str, Any], int]
    def stream_size(self, run: str, key: str) -> int
    def task_agent(self, run: str, key: str) -> str
    def task_detail(self, run: str, key: str) -> dict[str, Any]
    task_detail_fn
    def task_io(
        self,
        run: str,
        key: str,
        *,
        outputs: dict[str, Any] | None = None,
        plan_rows: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]
    title
    def trends(self, period: Period, *, run_group: str = '', stub: bool = False) -> dict[str, Any]
    def trends_panel(self, name: str, period: Period, *, run_group: str = '') -> dict[str, Any]
    async def wait_stream(self, run: str, key: str, beyond: int, timeout: float) -> int

class NoteBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    text: str
    title: str | None = Field(default=None, description='a heading over the body')
    tone: Tone | None = None
    type: Literal['note'] = 'note'

PANEL_TYPES = ('tiles', 'table', 'columns', 'lines', 'heatmap', 'hbars', 'status_grid', 'note', 'code', 'tree', 'blocks')

@dataclass
class Panel
    def __init__(
        self,
        fn: PanelFn | TrendsPanelFn,
        title: str = '',
        description: str = '',
        icon: str = '▤',
        scope: Literal['run', 'trends'] = 'run',
        tab: str = '',
    ) -> None
    def __call__(self, *args: Any) -> dict[str, Any]
    description: str = ''
    fn: PanelFn | TrendsPanelFn
    icon: str = '▤'
    def info(self, name: str) -> dict[str, Any]
    scope: Literal['run', 'trends'] = 'run'
    tab: str = ''
    title: str = ''

PanelBody = Annotated[Annotated[TilesBody, Tag('tiles')] | Annotated[TableBody, Tag('table')] | Annotated[ColumnsBody, Tag('columns')] | Annotated[LinesBody, Tag('lines')] | Annotated[HeatmapBody, Tag('heatmap')] | Annotated[HBarsBody, Tag('hbars')] | Annotated[StatusGridBody, Tag('status_grid')] | Annotated[NoteBody, Tag('note')] | Annotated[CodeBody, Tag('code')] | Annotated[TreeBody, Tag('tree')] | Annotated[BlocksBody, Tag('blocks')] | Annotated[dict[str, Any], Tag('untyped')], Discriminator(_panel_kind)]

class PanelFn(Protocol)
    def __call__(self, run: RunRecords, /) -> PanelBody | BaseModel | Mapping[str, Any]

class PanelTile(Closed)
    id: str
    label: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    sub: list[list[str]] = Field(default_factory=list)
    tip: str | None = None
    tone: Tone | None = None
    value: int | float | str = Field(description='a number as given (an integer stays one), or text already formatted')

@dataclass(frozen=True)
class Period
    def __init__(self, start: date, end: date, tz: str = 'UTC') -> None
    def as_dict(self) -> dict[str, Any]
    @property closes: datetime
    def day_of(self, ts: object) -> date | None
    @property days: list[date]
    end: date
    @property opens: datetime
    @classmethod
    def parse(cls, start: str = '', end: str = '', tz: str = '', *, today: date | None = None) -> Period
    start: date
    tz: str = 'UTC'
    @property zone: ZoneInfo

class PeriodError(ValueError)

class PeriodRuns(list[str])
    def __init__(self, runs: Iterable[str], source: RunSource) -> None
    def dir(self, run: str) -> Path | None
    def document(self, run: str, name: str) -> Any
    def info(self, run: str) -> dict[str, Any]
    def journal(self, run: str) -> list[dict[str, Any]]
    def log(self, run: str, name: str) -> list[dict[str, Any]]
    def output(self, run: str, key: str) -> RunFile | None
    def records(self, run: str, log: str) -> list[dict[str, Any]]
    def run(self, run: str) -> RunRecords

class RunFile(NamedTuple)
    data: bytes
    mtime: str
    @property size: int
    @property text: str | None

class RunRecords
    def __init__(self, run: str, source: RunSource) -> None
    def dir(self) -> Path | None
    def document(self, name: str) -> Any
    id
    def info(self) -> dict[str, Any]
    def journal(self) -> list[dict[str, Any]]
    def log(self, name: str) -> list[dict[str, Any]]
    def output(self, key: str) -> RunFile | None
    def records(self, log: str) -> list[dict[str, Any]]

class RunSource(Protocol)
    def activity(self, run: str) -> str | None
    def dir(self, run: str) -> Path | None
    def document(self, run: str, name: str) -> str | None
    def documents(self, run: str) -> list[str]
    def exists(self, run: str, name: str) -> bool
    def file(self, run: str, path: str) -> RunFile | None
    def file_stat(self, run: str, path: str) -> Stat | None
    def folder(self, run: str, folder: str) -> list[str]
    def info(self, run: str) -> dict[str, Any] | None
    def keep(self, run: str, text: str) -> None
    def kept(self, run: str) -> str | None
    def lines(
        self,
        run: str,
        log: str,
        since: int = 0,
        limit: int | None = None,
        max_bytes: int | None = None,
    ) -> tuple[list[str], int]
    def lines_at(
        self,
        run: str,
        log: str,
        at: LogPosition,
        limit: int | None = None,
        max_bytes: int | None = None,
    ) -> tuple[list[str], LogPosition]
    def logs(self, run: str) -> list[str]
    def position(self, run: str, log: str, line: int) -> LogPosition
    def runs(self) -> list[str]
    def size(self, run: str, log: str) -> int
    def stat(self, run: str, name: str = '') -> Stat | None
    async def wait(self, run: str, log: str, beyond: int, timeout: float) -> int

class Stat(NamedTuple)
    @property mtime: str
    mtime_ns: int
    size: int

class StatusGridBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    cells: list[list[GridCell | None]]
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['status_grid'] = 'status_grid'
    x: list[str]
    y: list[str]

class TableBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    columns: list[TableColumn]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    rows: list[dict[str, Any]]
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['table'] = 'table'

class TableColumn(Closed)
    align: Literal['start', 'end'] | None = None
    bar: bool = False
    format: ValueFormat | None = None
    id: str
    label: str
    link: str | None = Field(default=None, min_length=1, description="the key of the row that holds the cell's URL")
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tip: str | None = None

class TaskDetailFn(Protocol)
    def __call__(self, run: str, key: str, rows: list[dict[str, Any]], /) -> Mapping[str, Any] | None

class TilesBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tiles: list[PanelTile]
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['tiles'] = 'tiles'

class TreeBody(_Body)
    anchor: str | None = Field(default=None, min_length=1, description="a name a tree node's link leads to: the body is scrolled into view (`TreeLink.anchor`)")
    collapsed: bool = Field(default=False, description='shown folded, under its title')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    nodes: list[TreeNode]
    selected: str | None = None
    title: str | None = Field(default=None, description='a heading over the body')
    type: Literal['tree'] = 'tree'

class TreeLink(Closed)
    anchor: str | None = Field(default=None, min_length=1)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    panel: str | None = Field(default=None, min_length=1)
    run: str | None = Field(default=None, min_length=1, description='the run of `panel` or `task`; a trends panel names it')
    task: str | None = Field(default=None, min_length=1, description="a task's key")

class TreeNode(Closed)
    badge: str | None = None
    children: list[TreeNode] = Field(default_factory=list)
    code: CodeBody | None = None
    icon: Literal['file', 'directory'] | None = None
    id: str | None = Field(default=None, min_length=1)
    link: TreeLink | None = None
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str = Field(min_length=1)
    open: bool = True
    tone: Tone | None = None

class TrendsPanelFn(Protocol)
    def __call__(self, runs: PeriodRuns, period: Period, /) -> PanelBody | BaseModel | Mapping[str, Any]

def create_app(
    monitor: Monitor,
    *,
    allowed_hosts: Iterable[str] = (),
    cors_origins: Iterable[str] = (),
    docs: bool = False,
    base_path: str = '',
) -> FastAPI

def export(
    monitor: Monitor,
    out: str | Path,
    *,
    runs: int | None = 10,
    since: str | None = None,
    keep: Iterable[str] = (),
    exclude: Iterable[str] = (),
    period: Period | None = None,
    live: bool = False,
    now: Callable[[], datetime] = lambda: datetime.now(UTC),
) -> ExportReport

def serve(
    monitor: Monitor,
    host: str = '127.0.0.1',
    port: int = 8787,
    *,
    allowed_hosts: Iterable[str] = (),
    cors_origins: Iterable[str] = (),
    docs: bool = False,
    base_path: str = '',
) -> None

## contraagent.monitor.api

API_VERSION = '3.5.0'

class AccountQuota(Closed)
    account: str = Field('', description='the account the readings are of (`QuotaReading.account`); empty when not known')
    credential: str
    first: QuotaReading
    last: QuotaReading
    model_config: ConfigDict = ConfigDict(extra='forbid')
    rests: list[CredentialRest] = Field(default_factory=list, description='its rests now (`/api/hosts`): those a live run began and that have not ended, nor passed their `until`')

class AgentProcess(_Process)
    age_s: int
    children: list[ProcNode] = Field(default_factory=list)
    cmd: str = Field(description='the first three words of the command line')
    full_cmd: str
    key: str
    kind: str
    model: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pid: int
    plugin: str
    role: str = Field(description="the role it runs as, from the prompt's TELEMETRY line; empty when none")
    unit: str

AnyTranscriptEvent = Annotated[Annotated[LaunchEvent, Tag('LaunchEvent')] | Annotated[MessageUpdateEvent, Tag('MessageUpdateEvent')] | Annotated[ToolStartEvent, Tag('ToolStartEvent')] | Annotated[ToolEndEvent, Tag('ToolEndEvent')] | Annotated[ToolCallEchoEvent, Tag('ToolCallEchoEvent')] | Annotated[ToolResultEchoEvent, Tag('ToolResultEchoEvent')] | Annotated[MessageEvent, Tag('MessageEvent')] | Annotated[NoticeEvent, Tag('NoticeEvent')] | Annotated[RetryEvent, Tag('RetryEvent')] | Annotated[OtherEvent, Tag('other')], Discriminator(_event_tag), WrapValidator(_lenient)]

class ContractField(Closed)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str
    required: bool
    type: str = Field(description='JSON-schema type, `list[...]`, a `$ref` name, or `a|b` for a union')

class ContractRecord(Open)
    executor: Executor = 'agent'
    inputs: list[ContractSpec] = Field(default_factory=list)
    key: str
    kind: str = ''
    model_config: ConfigDict = ConfigDict(extra='allow')
    outputs: list[ContractSpec] = Field(default_factory=list)
    params: dict[str, str | int | float | bool] = Field(default_factory=dict, description="the task's params that are strings, numbers or booleans; an earlier task's output is one of `inputs`")
    unit: str = ''

class ContractSpec(Open)
    errors: list[str] = Field(default_factory=list)
    fields: list[ContractField]
    file: FileView
    format: Literal['json', 'text', 'jsonl'] | None = Field(None, description='what the file holds: one JSON document, any text, or rows (JSON Lines, `fields` those of one line)')
    fresh: bool | None = None
    io: Literal['output', 'input']
    model_config: ConfigDict = ConfigDict(extra='allow')
    path: str
    preserve: list[str] = Field(default_factory=list, description='output: JSON pointers to the parts of the file the task keeps as they were (`*`: every member)')
    producer: str | None = Field(None, description='input: the key of the task whose output it is')
    required: bool | None = None
    rounds: int | None = Field(None, description='rounds it took to pass, when the journal recorded it')
    schema_name: str
    schema_ref: str = ''
    status: ContractStatus | None = None
    validators: list[str] = Field(default_factory=list)

class CredentialRest(Rest)
    ended: str = Field('', description='when it was found not resting again, ISO 8601; empty while it rests (or the run ended first)')
    since: str = Field(description='when it began resting, ISO 8601')

class CursorMessage(Closed)
    data: CursorView | None
    event: Literal['cursor']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class CursorView(Closed)
    completed_stages: list[str] = Field(default_factory=list)
    history: list[StageEntry] = Field(default_factory=list)
    markers: dict[str, list[dict[str, Any]]] = Field(default_factory=dict, description='free-form event lists by name')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    stage: str | None = None
    units: dict[str, UnitPosition] = Field(default_factory=dict)
    updated_at: str = ''

class DefinitionFile(Open)
    model_config: ConfigDict = ConfigDict(extra='allow')
    part: str
    path: str
    text: str

class DownstreamTask(Closed)
    executor: Executor
    key: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    unit: str

class DroppedRoute(BaseModel)
    agent: str
    credential: str
    kinds: tuple[str, ...] = Field(description='the kinds whose chain held it')
    model: str = Field(description="empty: the coding agent's default model")
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    reason: str = Field(description="not in its coding agent's model list, or how its probe ended (`rate limited`, ...)")

class ExportInfo(Closed)
    exported_at: str = Field(description='when it was written, ISO 8601')
    left_out: list[str] = Field(description='the views it does not hold: `transcripts`, `processes`')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    redacted: list[str] = Field(description='what was taken out of the views: `paths` (made relative, else `<path>`), `host`, and each key named with --exclude')
    runs: list[str] = Field(description='the runs it holds, newest activity first')

class FeedReady(Closed)
    journal: int = Field(description='journal rows sent before this message')
    model_config: ConfigDict = ConfigDict(extra='forbid')

class FileView(Closed)
    content: Any = None
    exists: bool
    model_config: ConfigDict = ConfigDict(extra='forbid')
    mtime: str = ''
    size: int = 0
    truncated: bool = False

class FlowCallNode(Closed)
    error: str | None = Field(None, description='`<type>: <message>` of what it raised (or what closing it raised); else null')
    flow: list[Frame] = Field(default_factory=list, description="the flow path where the parent yielded it, with its loops' passes; empty for a top flow")
    function: str = Field(description='`module.qualname` of the generator function (or of a flow that does not yield)')
    id: str = Field(description='`0` for the top flow, `<parent>.<n>` for a child')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    parent: str | None = Field(description="the parent flow call's id; null for a top flow")
    state: FlowState
    time_box: TimeBoxRecord | None = Field(None, description='its own deadline, for a flow call yielded through `time_box`; else null')

class HostSummary(Closed)
    host_id: str
    interval_s: float | None = Field(description='seconds between its reports; null when it does not report')
    last_seen: str = Field(description='when it last reported, ISO 8601; empty when it does not report (one host)')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    running: int = Field(description='its runs whose process is alive')
    runs: int = Field(description='its runs')
    unresponsive: int = Field(description='its runs last reported alive since it stopped reporting')
    up: bool = Field(description='its reports keep arriving (a hub: a heartbeat within the TTL; one host: always)')

class Hosts(Closed)
    hosts: list[HostSummary]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    quota: list[AccountQuota] = Field(default_factory=list, description='per credential (account): its first and last utilization reading over every run; highest weekly use first')

class InflightPythonTask(Closed)
    age_s: int
    key: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pid: int
    unit: str

class Info(Closed)
    api_version: str = Field(description='Version of this API contract (semver)')
    export: ExportInfo | None = Field(None, description='set only in an export: what it holds; a server never sends it')
    hosts: bool = Field(False, description='the runs come from many hosts (a hub): `/api/hosts` lists them')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    panels: list[PanelInfo]
    title: str

class JournalMessage(Closed)
    data: JournalRow
    event: Literal['journal']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class JournalPage(Closed)
    entries: list[JournalRow]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    total: int

class JournalRow(BaseModel)
    agent: str = ''
    attempt: int = 0
    changes: Changes | None = None
    context_budget: int | None = None
    context_warned: bool = False
    cost_usd: float | None = None
    credential: str = ''
    executor: Executor = 'agent'
    exit_code: int | None = None
    fingerprint: str = ''
    flow_call: str | None = None
    hook_errors: list[list[str]] = Field(default_factory=list)
    hops: list[Hop] = Field(default_factory=list)
    key: str
    kind: str
    max_corrections: int | None = None
    model: str = ''
    model_config: ConfigDict = ConfigDict(extra='allow')
    ok: bool
    outside: str = ''
    quota: list[QuotaReading] = Field(default_factory=list)
    reason: str = ''
    replayed: bool = False
    rerun: str = ''
    round_errors: list[list[str]] = Field(default_factory=list)
    rounds: int | None = None
    route: str = ''
    run_group: str = ''
    run_id: str = ''
    signal: int | None = None
    skipped_hops: list[SkippedHop] = Field(default_factory=list)
    stop: str = Field(json_schema_extra={'enum': list(JOURNAL_STOPS)})
    ts: str
    unit: str = ''
    unread_events: dict[str, int] = Field(default_factory=dict)
    usage: TokenUsage | None = None
    wall_s: float = 0.0
    web_queries: list[str] = Field(default_factory=list)

class KeptKind(BaseModel)
    agent: str
    kind: str
    model: str
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    reason: str

class LaunchEvent(TranscriptEvent)
    argv: list[str]
    cwd: str
    model_config: ConfigDict = ConfigDict(extra='allow')
    prompt: str = Field(description='cut at 200 000 characters')
    prompt_truncated: int = Field(0, description='characters of the prompt not shown')
    ts: float | None = Field(None, description='Unix time the launch started')
    type: Literal['launch']

class MessageEvent(TranscriptEvent)
    model_config: ConfigDict = ConfigDict(extra='allow')
    role: str | None = None
    text: str | None = None
    thinking: str | None = None
    type: Literal['message_start', 'message_end', 'turn_end']

class MessageUpdateEvent(TranscriptEvent)
    delta: str
    ev: str
    model_config: ConfigDict = ConfigDict(extra='allow')
    type: Literal['message_update']

class ModelUsage(Closed)
    agent: str
    cache_creation: int = 0
    cache_read: int = 0
    calls: int = Field(0, description='API calls')
    context_window: int | None = Field(None, description="its context window (tokens of one API call's input) as the coding agent reported it; null when none did")
    cost_usd: float | None = Field(None, description='what the coding agent reported; null when no session on it reported a cost')
    input: int = 0
    model: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    output: int = 0
    peak_context: int | None = Field(None, description='peak context: the largest input of one API call of a session on it (tokens); null when none reported one')
    sessions: int = Field(description='hops: agent sessions launched on it (pool members tried on one hop count once); 0 for an other model only')
    wall_s: float = 0.0

class NoticeEvent(TranscriptEvent)
    model_config: ConfigDict = ConfigDict(extra='allow')
    text: str
    type: Literal['notice']

class OtherEvent(TranscriptEvent)
    model_config: ConfigDict = ConfigDict(extra='allow')
    type: str

class OutputTally(Closed)
    failed: int
    invalid: int
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pending: int
    refused: int
    running: int
    valid: int

class Outputs(Closed)
    contracts: list[ContractRecord]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tally: OutputTally = Field(description='output contracts per status')

class OutputsMessage(Closed)
    data: Outputs
    event: Literal['outputs']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class PanelInfo(Closed)
    description: str
    icon: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str
    scope: Literal['run', 'trends'] = 'run'
    tab: str = Field('', description="a trends panel's tab of the Trends page, with the panels of the same `tab`; '': on the first tab")
    title: str

class PanelMessage(Closed)
    data: PanelBody
    event: str = Field(pattern='^panel:.+$', description='`panel:<name>`, one per run panel in /api/info `panels`')
    model_config: ConfigDict = ConfigDict(extra='forbid')

class Plan(Closed)
    flow_calls: list[FlowCallNode] = Field(default_factory=list, description='in start order')
    groups: list[PlanGroup]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    models: list[ModelUsage] = Field(default_factory=list, description="each model the run's agent sessions ran on, in order of first use")
    nodes: list[PlanNode]
    preflight: PreflightRecord | None = Field(None, description="what the run's preflight did (`run.json`); null when its run.json records none")
    quota_caps: list[QuotaCap] = Field(default_factory=list, description="the run's credentials that have quota caps (`run.json`)")
    quota_rule: str = Field('', description="`module.qualname` of the run's quota rule (`Pipeline(quota_rule=)`); empty for none")
    rests: list[CredentialRest] = Field(default_factory=list, description='each rest of a credential in the run (`rests.jsonl`), in start order')
    services: list[ServiceSpan] = Field(default_factory=list, description='each start of a service of the run (`services.jsonl`), with its stop, in start order')

class PlanGroup(Closed)
    blocked: int
    done: int
    failed: int
    id: str
    label: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    partial: int = 0
    pending: int
    running: int
    skipped: int
    total: int

class PlanMessage(Closed)
    data: Plan
    event: Literal['plan']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class PlanNode(PlanRow)
    after: list[str] = Field(default_factory=list, description='task keys it waits for')
    attempts: int = 1
    base_kind: str = Field('', description="an implementation kind's base kind (`agent_task(implementations=)`), its group unless `groups` lists it; else empty")
    claims: list[str] = Field(default_factory=list, description="the names it holds while it runs, its kind's `claims` filled in")
    created_from: list[str] = Field(default_factory=list, description='keys whose results its flow call had received when it created it')
    credential: str = Field('', description='the credential its task call names (`credential=`), which every route of its kind ran it as; empty when it names none, or a run-wide override outranks it')
    deferred: str = Field('', description='why the queue deferred it, never launched (`deferred.jsonl`): a deadline, a resting credential; empty when it was not')
    deferred_cause: DeferralCause | None = Field(None, description='the same as data; null when it was not deferred, or the run recorded no cause')
    executor: Executor = 'agent'
    flow: list[Frame] = Field(default_factory=list, description='the flow path: frames of flow code that created it, outermost first')
    flow_call: str | None = Field(None, description='the id of the flow call that created it; null for a task with no origin')
    group: str
    key: str
    kind: str
    level: int = Field(description='depth in the dependency graph, 0 for a task that waits for nothing')
    model_config: ConfigDict = ConfigDict(extra='allow', serialize_by_alias=True)
    priority: float = Field(0.0, description="its task call's `priority`: a tie-breaker in the queue, higher first")
    rerun: str = Field('', description='why a resume ran it again although an earlier attempt of the run had finished it (its outside state changed, ...), from its last journal row; empty when no earlier attempt had')
    route: str = ''
    state: TaskState
    stop: str = ''
    def to_dict(self) -> dict[str, Any]
    ts: str = Field('', description='when it was queued (absent for a node rebuilt from the journal)')
    unit: str = ''
    wall_s: float = 0.0

class PreflightRecord(BaseModel)
    dropped: list[DroppedRoute] = Field(default_factory=list)
    kept: list[KeptKind] = Field(default_factory=list)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    policy: PreflightPolicy = Field(description='`Pipeline(preflight=)`; `off`: no route was probed')
    stopped: list[str] = Field(default_factory=list, description='the kinds with no route left that stopped the run (policy `stop`)')
    substituted: list[Substitution] = Field(default_factory=list)

class ProcNode(_Process)
    age_s: int
    children: list[ProcNode] = Field(default_factory=list)
    cmd: str = Field(description='the first three words of the command line')
    full_cmd: str
    key: str | None = None
    kind: str | None = None
    model: str | None = None
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pid: int
    plugin: str | None = None
    role: str | None = Field(None, description="the role it runs as, from the prompt's TELEMETRY line")
    unit: str | None = None

class Procs(Closed)
    alive: bool
    argv: list[str]
    launches: list[AgentProcess]
    max_workers: int | None
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pid: int | None
    python_tasks: list[InflightPythonTask]
    run: str
    run_group: str
    started_at: str
    tree: list[ProcNode]

class ProcsMessage(Closed)
    data: Procs
    event: Literal['procs']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class QuotaCap(BaseModel)
    credential: str
    five_hour_pct: float | None = None
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    seven_day_pct: float | None = None

class ReadyMessage(Closed)
    data: FeedReady
    event: Literal['ready']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class RetryEvent(TranscriptEvent)
    attempt: int | None = None
    err: str = ''
    model_config: ConfigDict = ConfigDict(extra='allow')
    type: Literal['auto_retry_start']

RunEvent = JournalMessage | ProcsMessage | PlanMessage | OutputsMessage | CursorMessage | PanelMessage | ReadyMessage

class RunSummary(Closed)
    attempts: int = Field(description='journal rows: finished task attempts')
    cost_usd: float | None = Field(None, description='what the coding agents reported the run cost, in USD; null when none reported')
    host_id: str = Field('', description='the host that drives it; empty when its run.json does not say')
    kinds: dict[str, int] = Field(description='journal rows per task kind')
    last_ts: str = Field(description="ISO time of the last journal row, else the directory's mtime")
    model_config: ConfigDict = ConfigDict(extra='forbid')
    not_run: int = Field(0, description='journal rows of tasks that never ran: refused for an input (`invalid_input`) or skipped')
    ok: int = Field(description='journal rows that passed, not counting `partial` ones')
    partial: int = Field(0, description='journal rows with stop `partial`')
    quota: list[AccountQuota] = Field(default_factory=list, description="per credential (account): the run's first and last utilization reading; highest weekly use first")
    replay: str = Field('', description="a replay: the directory of the recordings it played (`Pipeline(replay=)`), and with `record_missing` recorded into (then `stub` is false, and each played attempt's journal row says `replayed`); empty otherwise")
    row_extras: list[str] = Field(default_factory=list, description="the journal columns that are the pipeline's own (`Run(row_extras=)`), in the order they first appear")
    run: str
    run_group: str = Field(description='the label grouping runs (`Run(run_group=)`); the run id when none was given')
    run_uid: str = Field('', description='the id no other run has, on any host; empty when its run.json does not say')
    running: bool = Field(description='the process that runs it is alive')
    started_at: str
    stub: bool = Field(False, description="a stub run: the stub agent, or a replay's recordings, played its tasks (`Pipeline(stub=)`, `contraagent run --stub`)")
    tasks: TaskCounts
    tokens: TokenUsage = Field(description='tokens of every attempt, summed')
    unresponsive: bool = Field(False, description='its host last reported it alive and has stopped reporting (a hub only)')
    wall_s: int = Field(description="sum of the rows' wall time, rounded to seconds")

class ServiceSpan(Closed)
    address: str | None = Field(None, description='what it gave its tasks to reach it; null for none')
    key: str = Field('', description='the task whose claim started it')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str = Field(description='the claim name the service stands for')
    start_error: str = Field('', description='the exception of a start that failed (the service did not start); empty when it started')
    start_s: float = Field(0.0, description='how long its start took, in seconds')
    started_at: str = Field(description='when its start ended, ISO 8601')
    stop_cause: ServiceStopCause | None = Field(None, description='why it stopped; null while it is up')
    stop_error: str = Field('', description='the exception its stop raised; empty when the stop went well')
    stopped_at: str = Field('', description='when it stopped, ISO 8601; empty while it is up, or when the run ended with no stop recorded')

class StageEntry(Closed)
    at: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    name: str

class StreamMessage(Closed)
    data: StreamPage
    event: Literal['stream']
    model_config: ConfigDict = ConfigDict(extra='forbid')

class StreamPage(Closed)
    behind: int
    events: list[AnyTranscriptEvent] = Field(json_schema_extra=_any_of_items)
    model_config: ConfigDict = ConfigDict(extra='forbid')
    total: int

class Substitution(BaseModel)
    from_agent: str
    from_model: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    to_agent: str
    to_model: str

class TaskCounts(Closed)
    done: int
    failed: int
    model_config: ConfigDict = ConfigDict(extra='forbid')
    not_run: int = 0
    not_started: int
    partial: int = Field(0, description='tasks whose last result was `partial`: ok, the output a placeholder; not in `done`')
    running: int = 0
    skipped: int
    total: int
    waiting: int

class TaskDefinition(Open)
    base_kind: str | None = Field(None, description="agent task: an implementation kind's base kind (`agent_task(implementations=)`)")
    doc: str | None = Field(None, description="Python task: the function's docstring")
    executor: Executor | None = None
    file: str | None = Field(None, description="Python task: the source file, relative to the run's working directory when inside it")
    files: list[DefinitionFile] = Field(default_factory=list)
    function: str | None = Field(None, description='Python task: `module.qualname` of its function')
    kind: str
    line: int | None = Field(None, description="Python task: the function's first line in `file`")
    model_config: ConfigDict = ConfigDict(extra='allow')
    name: str | None = None
    output: str | None = None
    output_from: Literal['file', 'message'] | None = Field(None, description='agent task: `file`, the agent writes its output; `message`, its final message is the output')
    output_of: str | None = Field(None, description='agent task: the kind whose output file it writes (`agent_task(output_of=)`)')
    params: str | None = None
    plugin: bool | None = None
    plugin_dirs: list[str] = Field(default_factory=list)
    preserve: list[str] = Field(default_factory=list, description='agent task: JSON pointers to the parts of its output file it keeps (`agent_task(preserve=)`)')
    requires: list[str] = Field(default_factory=list)
    retries: int | None = Field(None, description='Python task: the fresh attempts after the first, each after an exception of `retry_on` (`@python_task(retries=)`)')
    retry_on: list[str] | None = Field(None, description='Python task: the exception classes that give an attempt a fresh one')
    role: str | None = None
    root: str | None = None
    signature: str | None = Field(None, description="Python task: the function's signature")
    tools: list[str] | None = Field(None, description='agent task: `[]`, a light launch with no tools (`agent_task(tools=())`); null, the tools its role allows')
    validators: list[str] = Field(default_factory=list)

class TaskDetail(Open)
    definition: TaskDefinition | None
    io: TaskIO
    key: str = Field(description='the task key asked for')
    launches: list[TaskLaunch] = Field(description='its launches running now; empty once it has finished (finished attempts are `rows`)')
    model_config: ConfigDict = ConfigDict(extra='allow')
    row_extras: list[str] = Field(default_factory=list, description="the columns of `rows` that are the pipeline's own (`Run(row_extras=)`), in the order they first appear")
    rows: list[JournalRow]
    stream: str | None = Field(description="the transcript's task key for /stream and /events/stream, null when it has none")
    task_detail_error: str | None = None

class TaskIO(Closed)
    downstream: list[DownstreamTask]
    executor: Executor
    inputs: list[ContractSpec]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    outputs: list[ContractSpec]
    params: dict[str, Any]
    upstream: list[UpstreamTask]

class TaskLaunch(Closed)
    age_s: int | None
    cmd: str = Field(description='the full command line, cut at 600 characters')
    key: str | None
    kind: str | None
    model: str | None
    model_config: ConfigDict = ConfigDict(extra='forbid')
    pid: int | None
    plugin: str | None
    role: str | None
    unit: str | None

class TimeBoxRecord(BaseModel)
    deadline_s: float = Field(description='seconds from `started_at` to the deadline')
    finishing_kinds: tuple[str, ...] = Field(default=(), description='the kinds that still launch past the deadline')
    grace_s: float = Field(description='seconds past the deadline its `finishing_kinds` still launch')
    model_config: ConfigDict = ConfigDict(extra='forbid', frozen=True)
    slack_s: float = Field(description='seconds before the deadline from which its tasks are ordered for it')
    started_at: str = Field(description='when the flow call started, where its deadline counts from (UTC ISO 8601)')

class ToolCallEchoEvent(TranscriptEvent)
    call: str
    cmd: str
    intent: str
    model_config: ConfigDict = ConfigDict(extra='allow')
    tool: str | None
    type: Literal['acall']
    web_fetch: str | None = Field(None, description='the URL, when the call reads a web page')
    web_query: str | None = Field(None, description='the query, when the call is a web search')

class ToolEndEvent(TranscriptEvent)
    call: str
    err: bool
    model_config: ConfigDict = ConfigDict(extra='allow')
    out: str
    tool: str | None
    type: Literal['tool_execution_end']

class ToolResultEchoEvent(TranscriptEvent)
    call: str
    err: bool
    model_config: ConfigDict = ConfigDict(extra='allow')
    out: str
    tool: str | None
    type: Literal['aresult']

class ToolStartEvent(TranscriptEvent)
    call: str
    cmd: str = Field(description='a one-line preview: the command, pattern, path, URL or query')
    intent: str
    model_config: ConfigDict = ConfigDict(extra='allow')
    tool: str | None
    type: Literal['tool_execution_start', 'tool_execution_update']
    web_fetch: str | None = Field(None, description='the URL, when the call reads a web page')
    web_query: str | None = Field(None, description='the query, when the call is a web search')

class TrendDay(Closed)
    attempts: int
    cost_usd: dict[str, float] = Field(description='group -> cost, USD; a group with no cost reported is absent')
    day: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    ok: int = Field(description='attempts that passed, not counting `partial` ones')
    partial: int = Field(0, description='attempts with stop `partial`: ok, counted apart from `ok`')

class TrendDeferral(Closed)
    cause: str
    key: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    reason: str
    run: str
    ts: str

class TrendDenial(Closed)
    count: int
    kinds: dict[str, int]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    tool: str

class TrendKind(Closed)
    attempts: int
    cost_usd: float | None
    executor: Executor
    group: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    ok: int = Field(description='attempts that passed, not counting `partial` ones')
    partial: int = Field(0, description='attempts with stop `partial`: ok, counted apart from `ok`')
    tasks: int
    tasks_ok: int = Field(description='tasks whose last attempt passed, not counting `partial` ones')
    tasks_partial: int = Field(0, description='tasks whose last attempt was `partial`')
    tokens: TokenUsage
    wall_s: float = Field(description="every attempt's wall time, summed")
    wall_s_median: float = Field(description="the median attempt's wall time")

class TrendPeriod(Closed)
    days: list[str] = Field(description='every day of the period, in order')
    from_: str = Field(alias='from', description='the first day, YYYY-MM-DD')
    model_config: ConfigDict = ConfigDict(extra='forbid', populate_by_name=True)
    to: str = Field(description='the last day, YYYY-MM-DD')
    tz: str

class TrendQuota(Closed)
    account: str = Field('', description='the account the readings are of (`QuotaReading.account`); empty when not known')
    credential: str
    daily_peak: list[float | None] = Field(description='per day of `period.days`; null when no reading arrived that day')
    hourly_peak: list[float | None] = Field(description="per hour from the period's first instant (the first day's midnight in `tz`); null when no reading arrived in it")
    model_config: ConfigDict = ConfigDict(extra='forbid')
    window: str = Field(description='`five_hour`, `seven_day`, or another window a reading names (`seven_day_opus`)')

class TrendRest(RestEvent)
    event: Literal['start', 'end']
    run: str
    ts: str

class TrendRun(Closed)
    attempts: int
    cost_usd: float | None
    finished_at: str
    host_id: str
    last_ts: str = Field(description='when its last attempt in the period ended; empty when it has none')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    ok: int = Field(description='attempts that passed, not counting `partial` ones')
    partial: int = Field(0, description='attempts with stop `partial`: ok, counted apart from `ok`')
    run: str
    run_group: str
    started_at: str
    stub: bool
    wall_s: float

class TrendStarts(Closed)
    counts: list[list[int]]
    hours: list[int]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    weekdays: list[str]

class TrendStop(Closed)
    attempts: int
    kinds: dict[str, int]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    stop: str

class TrendTotals(Closed)
    attempts: int
    cost_usd: float | None = Field(description='what the coding agents reported; null when no attempt reported a cost')
    model_config: ConfigDict = ConfigDict(extra='forbid')
    ok: int = Field(description='attempts that passed, not counting `partial` ones')
    partial: int = Field(0, description='attempts with stop `partial`: ok, counted apart from `ok`')
    runs: int = Field(description='runs with an attempt in the period, or started in it')
    tokens: TokenUsage
    wall_s: float

class Trends(Closed)
    bodies: dict[str, dict[str, Any]]
    daily: list[TrendDay]
    deferred: list[TrendDeferral]
    deferred_causes: dict[str, int]
    denials: list[TrendDenial]
    kinds: list[TrendKind]
    model_config: ConfigDict = ConfigDict(extra='forbid')
    models: list[ModelUsage]
    panels: list[PanelInfo]
    period: TrendPeriod
    quota: list[TrendQuota]
    rests: list[TrendRest]
    runs: list[TrendRun]
    starts: TrendStarts
    stops: list[TrendStop]
    totals: TrendTotals

class UnitPosition(Closed)
    completed: list[str]
    inflight: list[str]
    loop_pass: int
    model_config: ConfigDict = ConfigDict(extra='forbid')
    stage: str | None
    step: str | None

class UpstreamTask(Closed)
    executor: Executor
    key: str
    kind: str
    model_config: ConfigDict = ConfigDict(extra='forbid')
    outputs: list[ContractSpec]
    unit: str

## contraagent.testing

@dataclass(frozen=True)
class Reply
    def __init__(
        self,
        output: Any = _UNSET,
        stop: Stop = Stop.COMPLETED,
        reason: str = '',
        quota: QuotaReading | None = None,
        quota_refusal: QuotaRefusal | None = None,
        cost_usd: float | None = None,
        peak_context: int | None = None,
        context_window: int | None = None,
    ) -> None
    context_window: int | None = None
    cost_usd: float | None = None
    output: Any = _UNSET
    peak_context: int | None = None
    quota: QuotaReading | None = None
    quota_refusal: QuotaRefusal | None = None
    reason: str = ''
    stop: Stop = Stop.COMPLETED

class StubAgent(Backend)
    def __init__(
        self,
        script: Mapping[str, Any] | None = None,
        *,
        default: Any = _UNSET,
        execute: Iterable[str] = (),
        delay_s: float = 0.0,
    ) -> None
    def argv(self, launch: Launch) -> list[str]
    def backends(self, names: Iterable[str] | None = None) -> dict[str, StubAgent]
    binary = ''
    def bind_credential(
        self,
        credential: Credential,
        model: str = '',
        *,
        env: dict[str, str] | None = None,
        label: str = '',
    ) -> CredentialBinding
    default
    delay_s
    def env(self, launch: Launch) -> dict[str, str]
    execute: frozenset[str]
    launches: list[StubLaunch]
    def launches_of(self, key: str, run: str | None = None) -> list[StubLaunch]
    light_launch = True
    name = 'stub'
    def play_python(
        self,
        run: str,
        key: str,
        kind: str,
        unit: str,
        params: dict[str, Any],
        json_schema: dict[str, Any] | None,
        *,
        rows: bool = False,
    ) -> tuple[bool, Any]
    played: list[StubTask]
    def plays(self, kind: str) -> bool
    def resumable(self, launch: Launch) -> Launch
    def resumed(self, launch: Launch, prior: LaunchResult, prompt: str) -> Launch | None
    def run(self, launch: Launch) -> LaunchResult
    script: dict[str, Any]
    @classmethod
    def tool_deny_problem(cls, denied: tuple[str, ...]) -> str | None

@dataclass(frozen=True)
class StubLaunch
    def __init__(
        self,
        run: str,
        key: str,
        kind: str,
        unit: str,
        params: dict[str, Any],
        model: str,
        role: str | None,
        n: int,
        resumed: bool,
        prompt: str,
        call: CallFile,
        launch: Launch,
        in_turn: bool = False,
    ) -> None
    call: CallFile = field(repr=False)
    in_turn: bool = False
    key: str
    kind: str
    launch: Launch = field(repr=False)
    model: str
    n: int
    params: dict[str, Any]
    prompt: str
    resumed: bool
    role: str | None
    run: str
    unit: str

@dataclass(frozen=True)
class StubTask
    def __init__(self, run: str, key: str, kind: str, unit: str, params: dict[str, Any], n: int) -> None
    key: str
    kind: str
    n: int
    params: dict[str, Any] = field(repr=False)
    run: str
    unit: str

def fails(stop: Stop = Stop.ERROR, reason: str = 'stub: scripted failure') -> Reply

def rate_limited(retry_after_s: float | None = None) -> Reply

def writes(
    value: Any,
    *,
    stop: Stop = Stop.COMPLETED,
    cost_usd: float | None = None,
    quota: QuotaReading | None = None,
    peak_context: int | None = None,
    context_window: int | None = None,
) -> Reply

def writes_example() -> Reply

def writes_nothing() -> Reply
