API Reference

Quick reference for the current public API.

For provider-level feature differences, see Provider Capabilities.

Which Entry Point?

If you want to...	Use	Learn the workflow in...
Ask one prompt about one source (or no source)	run() → Output	Sending Content to Models
Ask many prompts against shared sources	run_many() → OutputCollection	Analyzing Collections with Source Patterns
Run one explicit interaction (environment + input), incl. tools/continuation	interact()	Building an Agent Loop
Stream one explicit interaction as it arrives	stream() → Event timeline	Building an Agent Loop
Prepare a reusable environment (and front-load cache/upload I/O)	prepare_environment() → Environment	Reducing Costs with Context Caching
Submit non-urgent work and collect it later	defer() → DeferredHandle	Building With Deferred Delivery
Check deferred job status or collect terminal results	inspect_deferred() / collect_deferred() / cancel_deferred()	Submitting Work for Later Collection

2.0 cutover: run() / run_many() return the Output / OutputCollection model (named facets, not dict envelopes). continue_tool() is replaced by interact(environment, Input(continuation=..., tool_results=...)); persistent caching moves to prepare_environment() / Environment(cache=...); and defer() returns an OutputCollection from collect_deferred() (the single defer_many() entry point is removed).

Entry Points

The primary execution functions are exported from pollux:

Run a single prompt, optionally with sources for context.

The simple v2 facade: returns an :class:Output with named facets (text, structured, reasoning, usage, metrics, …). For stable environments, continuation, or tool loops, use :func:interact.

Parameters:

Name	Type	Description	Default
prompt	str | None	The prompt to run.	None
source	Source | None	A single source for context (convenience for sources=[source]).	None
sources	Sequence[Source]	Stable sources for context.	()
config	Config	Configuration specifying provider and model.	required
environment	Environment | None	Optional prepared :class:Environment (e.g. from :func:prepare_environment) carrying instructions, sources, tools, and cache preference. Cannot be combined with inline instructions/source/sources/tools.	None
instructions	str | None	Optional system-level instruction.	None
output	ResponseSchemaInput | None	Optional Pydantic model or JSON Schema for structured output.	None
temperature	float | None	Optional sampling temperature.	None
top_p	float | None	Optional nucleus-sampling probability.	None
max_tokens	int | None	Optional hard cap on output tokens.	None
seed	int | None	Optional sampling seed where supported.	None
reasoning_effort	str | None	Optional qualitative reasoning effort.	None
reasoning_budget_tokens	int | None	Optional explicit reasoning token budget.	None
tool_choice	ToolChoice | None	Optional tool-choice control.	None
tools	Sequence[ToolDeclaration] | None	Optional tool declarations.	None
provider_options	dict[str, dict[str, Any]] | None	Optional raw provider-scoped generation options.	None

Returns:

Type	Description
Output	The completed :class:Output.

Example

config = Config(provider="anthropic", model="claude-haiku-4-5") result = await run("Summarize.", source=Source.from_file("doc.pdf"), config=config) print(result.text)

Source code in src/pollux/__init__.py

async def run(
    prompt: str | None = None,
    *,
    source: Source | None = None,
    sources: Sequence[Source] = (),
    config: Config,
    environment: Environment | None = None,
    instructions: str | None = None,
    output: ResponseSchemaInput | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    reasoning_effort: str | None = None,
    reasoning_budget_tokens: int | None = None,
    tool_choice: ToolChoice | None = None,
    tools: Sequence[ToolDeclaration] | None = None,
    provider_options: dict[str, dict[str, Any]] | None = None,
) -> Output:
    """Run a single prompt, optionally with sources for context.

    The simple v2 facade: returns an :class:`Output` with named facets
    (``text``, ``structured``, ``reasoning``, ``usage``, ``metrics``, …). For
    stable environments, continuation, or tool loops, use :func:`interact`.

    Args:
        prompt: The prompt to run.
        source: A single source for context (convenience for ``sources=[source]``).
        sources: Stable sources for context.
        config: Configuration specifying provider and model.
        environment: Optional prepared :class:`Environment` (e.g. from
            :func:`prepare_environment`) carrying instructions, sources, tools,
            and cache preference. Cannot be combined with inline
            ``instructions``/``source``/``sources``/``tools``.
        instructions: Optional system-level instruction.
        output: Optional Pydantic model or JSON Schema for structured output.
        temperature: Optional sampling temperature.
        top_p: Optional nucleus-sampling probability.
        max_tokens: Optional hard cap on output tokens.
        seed: Optional sampling seed where supported.
        reasoning_effort: Optional qualitative reasoning effort.
        reasoning_budget_tokens: Optional explicit reasoning token budget.
        tool_choice: Optional tool-choice control.
        tools: Optional tool declarations.
        provider_options: Optional raw provider-scoped generation options.

    Returns:
        The completed :class:`Output`.

    Example:
        config = Config(provider="anthropic", model="claude-haiku-4-5")
        result = await run("Summarize.", source=Source.from_file("doc.pdf"), config=config)
        print(result.text)
    """
    if environment is not None and source is not None:
        raise ConfigurationError(
            "environment cannot be combined with inline source/sources",
            hint="Build the Environment with the sources, or drop the environment.",
        )
    all_sources = (source,) if source is not None else tuple(sources)
    collection = await run_many(
        prompt,
        sources=all_sources,
        config=config,
        environment=environment,
        instructions=instructions,
        output=output,
        temperature=temperature,
        top_p=top_p,
        max_tokens=max_tokens,
        seed=seed,
        reasoning_effort=reasoning_effort,
        reasoning_budget_tokens=reasoning_budget_tokens,
        tool_choice=tool_choice,
        tools=tools,
        provider_options=provider_options,
    )
    return collection.outputs[0]

Run multiple prompts with shared sources for source-pattern execution.

Returns an :class:OutputCollection whose outputs preserve input order; .answers / .structured give ergonomic per-prompt lists.

Parameters:

Name	Type	Description	Default
prompts	str | Sequence[str | None] | None	One or more prompts to run.	None
sources	Sequence[Source]	Stable sources shared across the prompts.	()
config	Config	Configuration specifying provider and model.	required
environment	Environment | None	Optional prepared :class:Environment (e.g. from :func:prepare_environment) carrying instructions, sources, tools, and cache preference. Cannot be combined with inline instructions/sources/tools.	None
instructions	str | None	Optional system-level instruction.	None
output	ResponseSchemaInput | None	Optional Pydantic model or JSON Schema for structured output.	None
temperature	float | None	Optional sampling temperature.	None
top_p	float | None	Optional nucleus-sampling probability.	None
max_tokens	int | None	Optional hard cap on output tokens.	None
seed	int | None	Optional sampling seed where supported.	None
reasoning_effort	str | None	Optional qualitative reasoning effort.	None
reasoning_budget_tokens	int | None	Optional explicit reasoning token budget.	None
tool_choice	ToolChoice | None	Optional tool-choice control.	None
tools	Sequence[ToolDeclaration] | None	Optional tool declarations.	None
provider_options	dict[str, dict[str, Any]] | None	Optional raw provider-scoped generation options.	None

Returns:

Name	Type	Description
The	OutputCollection	class:OutputCollection for the source pattern.

Example

config = Config(provider="anthropic", model="claude-haiku-4-5") results = await run_many( ["Question 1?", "Question 2?"], sources=[Source.from_text("Context...")], config=config, ) for answer in results.answers: print(answer)

Source code in src/pollux/__init__.py

async def run_many(
    prompts: str | Sequence[str | None] | None = None,
    *,
    sources: Sequence[Source] = (),
    config: Config,
    environment: Environment | None = None,
    instructions: str | None = None,
    output: ResponseSchemaInput | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    reasoning_effort: str | None = None,
    reasoning_budget_tokens: int | None = None,
    tool_choice: ToolChoice | None = None,
    tools: Sequence[ToolDeclaration] | None = None,
    provider_options: dict[str, dict[str, Any]] | None = None,
) -> OutputCollection:
    """Run multiple prompts with shared sources for source-pattern execution.

    Returns an :class:`OutputCollection` whose ``outputs`` preserve input order;
    ``.answers`` / ``.structured`` give ergonomic per-prompt lists.

    Args:
        prompts: One or more prompts to run.
        sources: Stable sources shared across the prompts.
        config: Configuration specifying provider and model.
        environment: Optional prepared :class:`Environment` (e.g. from
            :func:`prepare_environment`) carrying instructions, sources, tools,
            and cache preference. Cannot be combined with inline
            ``instructions``/``sources``/``tools``.
        instructions: Optional system-level instruction.
        output: Optional Pydantic model or JSON Schema for structured output.
        temperature: Optional sampling temperature.
        top_p: Optional nucleus-sampling probability.
        max_tokens: Optional hard cap on output tokens.
        seed: Optional sampling seed where supported.
        reasoning_effort: Optional qualitative reasoning effort.
        reasoning_budget_tokens: Optional explicit reasoning token budget.
        tool_choice: Optional tool-choice control.
        tools: Optional tool declarations.
        provider_options: Optional raw provider-scoped generation options.

    Returns:
        The :class:`OutputCollection` for the source pattern.

    Example:
        config = Config(provider="anthropic", model="claude-haiku-4-5")
        results = await run_many(
            ["Question 1?", "Question 2?"],
            sources=[Source.from_text("Context...")],
            config=config,
        )
        for answer in results.answers:
            print(answer)
    """
    async with Session(config) as session:
        return await session.run_many(
            prompts,
            sources=sources,
            environment=environment,
            instructions=instructions,
            output=output,
            temperature=temperature,
            top_p=top_p,
            max_tokens=max_tokens,
            seed=seed,
            reasoning_effort=reasoning_effort,
            reasoning_budget_tokens=reasoning_budget_tokens,
            tool_choice=tool_choice,
            tools=tools,
            provider_options=provider_options,
        )

Run one explicit v2 interaction over an environment and input.

The v2 interaction facade: build an :class:Environment once (instructions, sources, tools, cache preference), then send :class:Input turns. Returns an :class:Output with named facets (text, structured, reasoning, tool_calls, continuation, usage, metrics, diagnostics). Continue by passing the prior output's continuation and any tool_results in the next Input.

Parameters:

Name	Type	Description	Default
environment	Environment	Reusable model-facing setup for the interaction.	required
input	Input	The per-turn payload (user content, continuation, tool results).	required
config	Config	Configuration specifying provider and model.	required
output	ResponseSchemaInput | None	Optional Pydantic model or JSON Schema for structured output.	None
temperature	float | None	Optional sampling temperature.	None
top_p	float | None	Optional nucleus-sampling probability.	None
max_tokens	int | None	Optional hard cap on output tokens.	None
seed	int | None	Optional sampling seed where supported.	None
reasoning_effort	str | None	Optional qualitative reasoning effort.	None
reasoning_budget_tokens	int | None	Optional explicit reasoning token budget.	None
tool_choice	ToolChoice | None	Optional tool-choice control.	None
provider_options	dict[str, dict[str, Any]] | None	Optional raw provider-scoped generation options.	None

Returns:

Type	Description
Output	The completed :class:Output for the interaction.

Example

environment = Environment(instructions=system_prompt, tools=tool_decls) result = await interact(environment, Input("Inspect the repo."), config=cfg) while result.tool_calls: results = [ToolResult(call_id=c.id, content=run_tool(c)) for c in result.tool_calls] result = await interact( environment, Input(continuation=result.continuation, tool_results=results), config=cfg, )

Source code in src/pollux/__init__.py

async def interact(
    environment: Environment,
    input: Input,  # noqa: A002 - "input" is the canonical v2 primitive name
    *,
    config: Config,
    output: ResponseSchemaInput | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    reasoning_effort: str | None = None,
    reasoning_budget_tokens: int | None = None,
    tool_choice: ToolChoice | None = None,
    provider_options: dict[str, dict[str, Any]] | None = None,
) -> Output:
    """Run one explicit v2 interaction over an environment and input.

    The v2 interaction facade: build an :class:`Environment` once (instructions,
    sources, tools, cache preference), then send :class:`Input` turns. Returns an
    :class:`Output` with named facets (``text``, ``structured``, ``reasoning``,
    ``tool_calls``, ``continuation``, ``usage``, ``metrics``, ``diagnostics``).
    Continue by passing the prior output's ``continuation`` and any
    ``tool_results`` in the next ``Input``.

    Args:
        environment: Reusable model-facing setup for the interaction.
        input: The per-turn payload (user content, continuation, tool results).
        config: Configuration specifying provider and model.
        output: Optional Pydantic model or JSON Schema for structured output.
        temperature: Optional sampling temperature.
        top_p: Optional nucleus-sampling probability.
        max_tokens: Optional hard cap on output tokens.
        seed: Optional sampling seed where supported.
        reasoning_effort: Optional qualitative reasoning effort.
        reasoning_budget_tokens: Optional explicit reasoning token budget.
        tool_choice: Optional tool-choice control.
        provider_options: Optional raw provider-scoped generation options.

    Returns:
        The completed :class:`Output` for the interaction.

    Example:
        environment = Environment(instructions=system_prompt, tools=tool_decls)
        result = await interact(environment, Input("Inspect the repo."), config=cfg)
        while result.tool_calls:
            results = [ToolResult(call_id=c.id, content=run_tool(c)) for c in result.tool_calls]
            result = await interact(
                environment,
                Input(continuation=result.continuation, tool_results=results),
                config=cfg,
            )
    """
    async with Session(config) as session:
        return await session.interact(
            environment,
            input,
            output=output,
            temperature=temperature,
            top_p=top_p,
            max_tokens=max_tokens,
            seed=seed,
            reasoning_effort=reasoning_effort,
            reasoning_budget_tokens=reasoning_budget_tokens,
            tool_choice=tool_choice,
            provider_options=provider_options,
        )

Stream one explicit v2 interaction as :class:Event objects.

The streaming sibling of :func:interact: same environment/input/config and the same completed result, observed as a timeline. Iterate the events (text_delta, reasoning_delta, tool_call_delta, tool_call, usage, finish) and read the final assembled :class:Output from the terminal done event, whose output matches what :func:interact would return for the same interaction. Consumers never parse SSE or provider chunks.

A provider that does not support streaming raises ConfigurationError. A mid-stream provider failure raises from the iterator rather than emitting a done event, so a failed interaction never yields a final output.

Parameters:

Name	Type	Description	Default
environment	Environment	Reusable model-facing setup for the interaction.	required
input	Input	The per-turn payload (user content, continuation, tool results).	required
config	Config	Configuration specifying provider and model.	required
output	ResponseSchemaInput | None	Optional Pydantic model or JSON Schema for structured output.	None
temperature	float | None	Optional sampling temperature.	None
top_p	float | None	Optional nucleus-sampling probability.	None
max_tokens	int | None	Optional hard cap on output tokens.	None
seed	int | None	Optional sampling seed where supported.	None
reasoning_effort	str | None	Optional qualitative reasoning effort.	None
reasoning_budget_tokens	int | None	Optional explicit reasoning token budget.	None
tool_choice	ToolChoice | None	Optional tool-choice control.	None
provider_options	dict[str, dict[str, Any]] | None	Optional raw provider-scoped generation options.	None

Yields:

Name	Type	Description
Each	AsyncGenerator[Event, None]	class:Event in the interaction's timeline, ending in done.

Example

async for event in stream(environment, Input("Inspect the repo."), config=cfg): if event.type == "text_delta": print(event.text, end="") elif event.type == "done": result = event.output

Source code in src/pollux/__init__.py

async def stream(
    environment: Environment,
    input: Input,  # noqa: A002 - "input" is the canonical v2 primitive name
    *,
    config: Config,
    output: ResponseSchemaInput | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    reasoning_effort: str | None = None,
    reasoning_budget_tokens: int | None = None,
    tool_choice: ToolChoice | None = None,
    provider_options: dict[str, dict[str, Any]] | None = None,
) -> AsyncGenerator[Event, None]:
    """Stream one explicit v2 interaction as :class:`Event` objects.

    The streaming sibling of :func:`interact`: same environment/input/config and
    the same completed result, observed as a timeline. Iterate the events
    (``text_delta``, ``reasoning_delta``, ``tool_call_delta``, ``tool_call``,
    ``usage``, ``finish``) and read the final assembled :class:`Output` from the
    terminal ``done`` event, whose ``output`` matches what :func:`interact` would
    return for the same interaction. Consumers never parse SSE or provider chunks.

    A provider that does not support streaming raises ``ConfigurationError``. A
    mid-stream provider failure raises from the iterator rather than emitting a
    ``done`` event, so a failed interaction never yields a final output.

    Args:
        environment: Reusable model-facing setup for the interaction.
        input: The per-turn payload (user content, continuation, tool results).
        config: Configuration specifying provider and model.
        output: Optional Pydantic model or JSON Schema for structured output.
        temperature: Optional sampling temperature.
        top_p: Optional nucleus-sampling probability.
        max_tokens: Optional hard cap on output tokens.
        seed: Optional sampling seed where supported.
        reasoning_effort: Optional qualitative reasoning effort.
        reasoning_budget_tokens: Optional explicit reasoning token budget.
        tool_choice: Optional tool-choice control.
        provider_options: Optional raw provider-scoped generation options.

    Yields:
        Each :class:`Event` in the interaction's timeline, ending in ``done``.

    Example:
        async for event in stream(environment, Input("Inspect the repo."), config=cfg):
            if event.type == "text_delta":
                print(event.text, end="")
            elif event.type == "done":
                result = event.output
    """
    async with Session(config) as session:
        events = session.stream(
            environment,
            input,
            output=output,
            temperature=temperature,
            top_p=top_p,
            max_tokens=max_tokens,
            seed=seed,
            reasoning_effort=reasoning_effort,
            reasoning_budget_tokens=reasoning_budget_tokens,
            tool_choice=tool_choice,
            provider_options=provider_options,
        )
        try:
            async for event in events:
                yield event
        finally:
            await close_async_iterator(events)

Prepare a reusable :class:Environment, front-loading cache/upload I/O.

Build the stable model-facing setup once and reuse it across :func:interact, :func:run, and :func:run_many calls. When cache is a :class:CachePolicy, the persistent cache is created now (uploading sources and surfacing capability errors early); later interactions over the returned environment reuse it by identity.

Parameters:

Name	Type	Description	Default
sources	Sequence[Source]	Stable sources to bake into the environment (and its cache).	()
config	Config	Configuration specifying provider and model.	required
instructions	str | None	Optional system-level instruction.	None
tools	Sequence[ToolDeclaration] | None	Optional tool declarations.	None
cache	CacheSetting	Cache preference: a :class:CachePolicy for persistent caching, "auto" for provider-managed caching, "none" to disable, or None for the default.	None
metadata	dict[str, Any] | None	Optional provider-neutral metadata for planning.	None

Returns:

Type	Description
Environment	The prepared :class:Environment.

Example

environment = await prepare_environment( sources=[Source.from_file("paper.pdf")], instructions=system_prompt, cache=CachePolicy(ttl_seconds=3600), config=config, ) results = await run_many(prompts, environment=environment, config=config)

Source code in src/pollux/__init__.py

async def prepare_environment(
    *,
    sources: Sequence[Source] = (),
    config: Config,
    instructions: str | None = None,
    tools: Sequence[ToolDeclaration] | None = None,
    cache: CacheSetting = None,
    metadata: dict[str, Any] | None = None,
) -> Environment:
    """Prepare a reusable :class:`Environment`, front-loading cache/upload I/O.

    Build the stable model-facing setup once and reuse it across
    :func:`interact`, :func:`run`, and :func:`run_many` calls. When ``cache`` is
    a :class:`CachePolicy`, the persistent cache is created now (uploading
    sources and surfacing capability errors early); later interactions over the
    returned environment reuse it by identity.

    Args:
        sources: Stable sources to bake into the environment (and its cache).
        config: Configuration specifying provider and model.
        instructions: Optional system-level instruction.
        tools: Optional tool declarations.
        cache: Cache preference: a :class:`CachePolicy` for persistent caching,
            ``"auto"`` for provider-managed caching, ``"none"`` to disable, or
            ``None`` for the default.
        metadata: Optional provider-neutral metadata for planning.

    Returns:
        The prepared :class:`Environment`.

    Example:
        environment = await prepare_environment(
            sources=[Source.from_file("paper.pdf")],
            instructions=system_prompt,
            cache=CachePolicy(ttl_seconds=3600),
            config=config,
        )
        results = await run_many(prompts, environment=environment, config=config)
    """
    environment = Environment(
        instructions=instructions,
        sources=tuple(sources),
        tools=tuple(tools) if tools else (),
        cache=cache,
        metadata=metadata,
    )
    if isinstance(cache, CachePolicy):
        provider = _get_provider(config)
        try:
            snapshot = _EnvironmentSnapshot.from_environment(
                environment, provider=config.provider
            )
            await resolve_persistent_cache(snapshot, config, provider)
        finally:
            await _close_provider(provider)
    return environment

Submit one or more deferred requests and return a serializable handle.

The v2 deferred frontdoor: submit a single interaction or a source-pattern collection on a provider-side timeline, then :func:inspect_deferred and :func:collect_deferred later. Collection returns an :class:OutputCollection, the same shape as :func:run_many.

Tool calling, continuation, and persistent caching are out of scope for deferred delivery and are intentionally not part of this surface.

Parameters:

Name	Type	Description	Default
prompts	str | Sequence[str | None] | None	One or more prompts to submit.	None
sources	Sequence[Source]	Stable sources shared across the prompts.	()
config	Config	Configuration specifying provider and model.	required
instructions	str | None	Optional system-level instruction.	None
output	ResponseSchemaInput | None	Optional Pydantic model or JSON Schema for structured output. Pass the same schema to :func:collect_deferred to rehydrate.	None
temperature	float | None	Optional sampling temperature.	None
top_p	float | None	Optional nucleus-sampling probability.	None
max_tokens	int | None	Optional hard cap on output tokens.	None
seed	int | None	Optional sampling seed where supported.	None
reasoning_effort	str | None	Optional qualitative reasoning effort.	None
reasoning_budget_tokens	int | None	Optional explicit reasoning token budget.	None
provider_options	dict[str, dict[str, Any]] | None	Optional raw provider-scoped generation options.	None

Returns:

Type	Description
DeferredHandle	A serializable :class:DeferredHandle for the submitted job.

Source code in src/pollux/__init__.py

async def defer(
    prompts: str | Sequence[str | None] | None = None,
    *,
    sources: Sequence[Source] = (),
    config: Config,
    instructions: str | None = None,
    output: ResponseSchemaInput | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    reasoning_effort: str | None = None,
    reasoning_budget_tokens: int | None = None,
    provider_options: dict[str, dict[str, Any]] | None = None,
) -> DeferredHandle:
    """Submit one or more deferred requests and return a serializable handle.

    The v2 deferred frontdoor: submit a single interaction or a source-pattern
    collection on a provider-side timeline, then :func:`inspect_deferred` and
    :func:`collect_deferred` later. Collection returns an
    :class:`OutputCollection`, the same shape as :func:`run_many`.

    Tool calling, continuation, and persistent caching are out of scope for
    deferred delivery and are intentionally not part of this surface.

    Args:
        prompts: One or more prompts to submit.
        sources: Stable sources shared across the prompts.
        config: Configuration specifying provider and model.
        instructions: Optional system-level instruction.
        output: Optional Pydantic model or JSON Schema for structured output.
            Pass the same schema to :func:`collect_deferred` to rehydrate.
        temperature: Optional sampling temperature.
        top_p: Optional nucleus-sampling probability.
        max_tokens: Optional hard cap on output tokens.
        seed: Optional sampling seed where supported.
        reasoning_effort: Optional qualitative reasoning effort.
        reasoning_budget_tokens: Optional explicit reasoning token budget.
        provider_options: Optional raw provider-scoped generation options.

    Returns:
        A serializable :class:`DeferredHandle` for the submitted job.
    """
    prompt_tuple = (
        (prompts,) if isinstance(prompts, (str, type(None))) else tuple(prompts)
    )
    if not prompt_tuple:
        raise ConfigurationError(
            "defer() requires at least one prompt",
            hint="Pass one or more prompts to submit for deferred collection.",
        )
    environment = Environment(instructions=instructions, sources=tuple(sources))
    inputs = [Input(content=prompt) for prompt in prompt_tuple]
    requirements = _build_requirements(
        output=output,
        temperature=temperature,
        top_p=top_p,
        max_tokens=max_tokens,
        seed=seed,
        reasoning_effort=reasoning_effort,
        reasoning_budget_tokens=reasoning_budget_tokens,
        tool_choice=None,
        provider_options=provider_options,
    )
    provider = _get_provider(config)
    try:
        return await submit_deferred(
            environment, inputs, requirements, config, provider
        )
    finally:
        await _close_provider(provider)

Inspect the current state of a deferred job.

Source code in src/pollux/__init__.py

async def inspect_deferred(
    handle: DeferredHandle,
) -> DeferredSnapshot:
    """Inspect the current state of a deferred job."""
    provider = _resolve_deferred_provider(handle)
    try:
        return await inspect_deferred_handle(handle, provider)
    finally:
        await _close_provider(provider)

Collect a terminal deferred job into an :class:OutputCollection.

Returns the same shape as :func:run_many: one :class:Output per submitted request, in submission order. Each output's diagnostics.raw["deferred"] carries the job id and per-item status.

Parameters:

Name	Type	Description	Default
handle	DeferredHandle	The deferred handle returned by :func:defer.	required
response_schema	type[Any] | dict[str, Any] | None	Optional Pydantic model or JSON Schema for structured output rehydration. Must match the schema used at submission time.	None

Source code in src/pollux/__init__.py

async def collect_deferred(
    handle: DeferredHandle,
    *,
    response_schema: type[Any] | dict[str, Any] | None = None,
) -> OutputCollection:
    """Collect a terminal deferred job into an :class:`OutputCollection`.

    Returns the same shape as :func:`run_many`: one :class:`Output` per
    submitted request, in submission order. Each output's
    ``diagnostics.raw["deferred"]`` carries the job id and per-item status.

    Args:
        handle: The deferred handle returned by :func:`defer`.
        response_schema: Optional Pydantic model or JSON Schema for structured
            output rehydration. Must match the schema used at submission time.
    """
    provider = _resolve_deferred_provider(handle)
    try:
        return await collect_deferred_handle(
            handle,
            provider,
            response_schema=response_schema,
        )
    finally:
        await _close_provider(provider)

Request provider-side cancellation for a deferred job.

Source code in src/pollux/__init__.py

async def cancel_deferred(
    handle: DeferredHandle,
) -> None:
    """Request provider-side cancellation for a deferred job."""
    provider = _resolve_deferred_provider(handle)
    try:
        await cancel_deferred_handle(handle, provider)
    finally:
        await _close_provider(provider)

Core Types

Source includes both the generic source constructors and narrow provider-specific helpers such as Source.with_gemini_video_settings(...) and Source.with_gemini_url_context().

A structured representation of a single input source.

Source code in src/pollux/source.py

@dataclass(frozen=True, slots=True)
class Source:
    """A structured representation of a single input source."""

    source_type: SourceType
    identifier: str
    mime_type: str
    size_bytes: int
    content_loader: Callable[[], bytes]
    provider_hints: tuple[ProviderHint, ...] = ()

    @classmethod
    def from_text(cls, text: str, *, identifier: str | None = None) -> Source:
        """Create a Source from text content.

        Args:
            text: The text content.
            identifier: Display label. Defaults to the first 50 characters of *text*.
        """
        content = text.encode("utf-8")
        ident = identifier or text[:50]
        return cls(
            source_type="text",
            identifier=ident,
            mime_type="text/plain",
            size_bytes=len(content),
            content_loader=lambda: content,
        )

    @classmethod
    def from_json(
        cls, data: dict[str, Any] | Any, *, identifier: str | None = None
    ) -> Source:
        """Create a Source from a JSON-serializable object.

        Args:
            data: A dictionary or object to serialize into a JSON string. If the object
                has a `model_dump()` method (like Pydantic models), it will be used.
            identifier: Display label. Defaults to "json-data".
        """
        if hasattr(data, "model_dump") and callable(data.model_dump):
            data = data.model_dump()

        try:
            content = json.dumps(data).encode("utf-8")
        except TypeError as exc:
            raise SourceError(f"Data is not JSON serializable: {exc}") from exc

        ident = identifier or "json-data"
        return cls(
            source_type="json",
            identifier=ident,
            mime_type="application/json",
            size_bytes=len(content),
            content_loader=lambda: content,
        )

    @classmethod
    def from_file(cls, path: str | Path, *, mime_type: str | None = None) -> Source:
        """Create a Source from a local file.

        Args:
            path: Path to the file. Must exist or ``SourceError`` is raised.
            mime_type: MIME type override. Auto-detected from extension when *None*.
        """
        p = Path(path)
        if not p.exists():
            raise SourceError(f"File not found: {p}")

        mt = mime_type or mimetypes.guess_type(str(p))[0] or "application/octet-stream"
        size = p.stat().st_size

        def loader() -> bytes:
            return p.read_bytes()

        return cls(
            source_type="file",
            identifier=str(p),
            mime_type=mt,
            size_bytes=size,
            content_loader=loader,
        )

    @classmethod
    def from_youtube(cls, url: str) -> Source:
        """Create a Source from a YouTube URL reference (no download)."""
        encoded = f"youtube:{url}".encode()
        return cls(
            source_type="youtube",
            identifier=url,
            mime_type="video/mp4",
            size_bytes=0,
            content_loader=lambda: encoded,
        )

    @classmethod
    def from_uri(
        cls, uri: str, *, mime_type: str = "application/octet-stream"
    ) -> Source:
        """Create a Source from a URI.

        Args:
            uri: Remote URI (e.g. ``gs://`` or ``https://``).
            mime_type: MIME type. Defaults to ``application/octet-stream``.
        """
        encoded = f"uri:{mime_type}:{uri}".encode()
        return cls(
            source_type="uri",
            identifier=uri,
            mime_type=mime_type,
            size_bytes=0,
            content_loader=lambda: encoded,
        )

    @classmethod
    def from_arxiv(cls, ref: str) -> Source:
        """Create an arXiv PDF Source from an arXiv ID or URL.

        Args:
            ref: An arXiv ID (e.g. ``"2301.07041"``) or full arXiv URL.
        """
        if not isinstance(ref, str):
            raise TypeError("ref must be a str")

        normalized_url = cls._normalize_arxiv_to_pdf_url(ref.strip())
        encoded = normalized_url.encode("utf-8")
        return cls(
            source_type="arxiv",
            identifier=normalized_url,
            mime_type="application/pdf",
            size_bytes=0,
            content_loader=lambda: encoded,
        )

    @staticmethod
    def _normalize_arxiv_to_pdf_url(ref: str) -> str:
        """Normalize arXiv id or URL to canonical PDF URL."""
        if not ref:
            raise SourceError("arXiv reference cannot be empty")

        arxiv_id = ref
        if ref.startswith(("http://", "https://")):
            parsed = urlparse(ref)
            host = parsed.netloc.lower()
            if host not in _ARXIV_HOSTS:
                raise SourceError(f"Expected arxiv.org URL, got: {parsed.netloc}")

            path = parsed.path.strip("/")
            if path.startswith("abs/"):
                arxiv_id = path[len("abs/") :]
            elif path.startswith("pdf/"):
                arxiv_id = path[len("pdf/") :]
            else:
                raise SourceError(f"Unsupported arXiv URL path: {parsed.path}")

        if arxiv_id.endswith(".pdf"):
            arxiv_id = arxiv_id[:-4]

        arxiv_id = arxiv_id.strip("/")
        if not _ARXIV_ID_RE.match(arxiv_id):
            raise SourceError(f"Invalid arXiv id: {arxiv_id}")

        return f"https://arxiv.org/pdf/{arxiv_id}.pdf"

    def _content_hash(self) -> str:
        """Compute SHA256 hash of raw content bytes."""
        content = self.content_loader()
        return hashlib.sha256(content).hexdigest()

    def gemini_video_settings_for(
        self, provider: str | None
    ) -> dict[str, str | float] | None:
        """Return Gemini video settings when the active provider can use them."""
        provider_hints = self.provider_hints_for(provider)
        if provider_hints is None:
            return None
        return provider_hints.get("video_metadata")

    def provider_hints_for(
        self, provider: str | None
    ) -> dict[str, dict[str, str | float]] | None:
        """Return immutable provider hints as plain dictionaries for transport."""
        if provider is None:
            return None

        hints = {
            hint.name: hint.payload_dict()
            for hint in self.provider_hints
            if hint.provider == provider
        }
        return hints or None

    def cache_identity_hash(self, *, provider: str | None = None) -> str:
        """Compute SHA256 hash for cache identity.

        Includes provider-visible source semantics such as Gemini video settings.
        Falls back to raw content hash when no provider-specific settings apply,
        preserving backward-compatible cache keys.
        """
        provider_hints = self.provider_hints_for(provider)
        if provider_hints is None:
            return self._content_hash()
        combined = (
            self._content_hash()
            + "|"
            + json.dumps(
                provider_hints,
                sort_keys=True,
                separators=(",", ":"),
            )
        )
        return hashlib.sha256(combined.encode("utf-8")).hexdigest()

    def _environment_identity(self, *, provider: str) -> dict[str, Any]:
        """Return provider-visible source identity for environment hashing."""
        identity: dict[str, Any] = {
            "source_type": self.source_type,
            "mime_type": self.mime_type,
            "content_hash": self._content_hash(),
        }
        if self.source_type == "file":
            identity["identifier"] = Path(self.identifier).name
        elif self.source_type not in {"text", "json"}:
            identity["identifier"] = self.identifier
        provider_hints = self.provider_hints_for(provider)
        if provider_hints is not None:
            identity["provider_hints"] = provider_hints
        return identity

    def with_gemini_video_settings(
        self,
        *,
        start_offset: str | None = None,
        end_offset: str | None = None,
        fps: float | None = None,
    ) -> Source:
        """Return a copy with validated Gemini video controls attached.

        Pollux keeps this API provider-specific and stable even if Google's
        underlying wire fields evolve. The Gemini adapter maps these settings
        to the current SDK request shape. These settings only affect Gemini
        requests and Gemini explicit-cache identity.
        """
        if not self._is_video_source():
            raise SourceError(
                "Gemini video settings require a video source "
                "(local video file, video URI, or YouTube URL)"
            )

        validated_start_offset: str | None = None
        validated_end_offset: str | None = None
        validated_fps: float | None = None

        if start_offset is not None:
            if not isinstance(start_offset, str) or not start_offset.strip():
                raise SourceError("start_offset must be a non-empty string")
            validated_start_offset = start_offset

        if end_offset is not None:
            if not isinstance(end_offset, str) or not end_offset.strip():
                raise SourceError("end_offset must be a non-empty string")
            validated_end_offset = end_offset

        if fps is not None:
            if isinstance(fps, bool) or not isinstance(fps, (int, float)):
                raise SourceError("fps must be a number")
            fps_value = float(fps)
            if fps_value <= 0 or fps_value > 24:
                raise SourceError("fps must be > 0 and <= 24")
            validated_fps = fps_value

        if (
            validated_start_offset is None
            and validated_end_offset is None
            and validated_fps is None
        ):
            raise SourceError(
                "Provide at least one Gemini video setting: "
                "start_offset, end_offset, or fps"
            )

        return replace(
            self,
            provider_hints=self._with_provider_hint(
                provider="gemini",
                name="video_metadata",
                payload={
                    k: v
                    for k, v in (
                        ("start_offset", validated_start_offset),
                        ("end_offset", validated_end_offset),
                        ("fps", validated_fps),
                    )
                    if v is not None
                },
            ),
        )

    def with_gemini_url_context(self) -> Source:
        """Return a copy that asks Gemini to retrieve this HTTP(S) URI at request time."""
        if self.source_type != "uri":
            raise SourceError("Gemini URL Context requires Source.from_uri(...)")

        parsed = urlparse(self.identifier)
        if parsed.scheme not in {"http", "https"}:
            raise SourceError("Gemini URL Context requires an HTTP(S) URI")

        return replace(
            self,
            provider_hints=self._with_provider_hint(
                provider="gemini",
                name="url_context",
                payload={},
            ),
        )

    def _is_video_source(self) -> bool:
        """Return True when Gemini video controls can apply to this source."""
        return self.source_type == "youtube" or self.mime_type.startswith("video/")

    def _with_provider_hint(
        self,
        *,
        provider: str,
        name: str,
        payload: dict[str, str | float],
    ) -> tuple[ProviderHint, ...]:
        """Return provider hints with one named hint replaced or added."""
        hint = ProviderHint(
            provider=provider,
            name=name,
            payload=tuple(sorted(payload.items())),
        )
        existing = tuple(
            item
            for item in self.provider_hints
            if not (item.provider == provider and item.name == name)
        )
        return (*existing, hint)

Functions

from_text classmethod

from_text(text, *, identifier=None)

Create a Source from text content.

Parameters:

Name	Type	Description	Default
text	str	The text content.	required
identifier	str | None	Display label. Defaults to the first 50 characters of text.	None

Source code in src/pollux/source.py

@classmethod
def from_text(cls, text: str, *, identifier: str | None = None) -> Source:
    """Create a Source from text content.

    Args:
        text: The text content.
        identifier: Display label. Defaults to the first 50 characters of *text*.
    """
    content = text.encode("utf-8")
    ident = identifier or text[:50]
    return cls(
        source_type="text",
        identifier=ident,
        mime_type="text/plain",
        size_bytes=len(content),
        content_loader=lambda: content,
    )

from_json classmethod

from_json(data, *, identifier=None)

Create a Source from a JSON-serializable object.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | Any	A dictionary or object to serialize into a JSON string. If the object has a model_dump() method (like Pydantic models), it will be used.	required
identifier	str | None	Display label. Defaults to "json-data".	None

Source code in src/pollux/source.py

@classmethod
def from_json(
    cls, data: dict[str, Any] | Any, *, identifier: str | None = None
) -> Source:
    """Create a Source from a JSON-serializable object.

    Args:
        data: A dictionary or object to serialize into a JSON string. If the object
            has a `model_dump()` method (like Pydantic models), it will be used.
        identifier: Display label. Defaults to "json-data".
    """
    if hasattr(data, "model_dump") and callable(data.model_dump):
        data = data.model_dump()

    try:
        content = json.dumps(data).encode("utf-8")
    except TypeError as exc:
        raise SourceError(f"Data is not JSON serializable: {exc}") from exc

    ident = identifier or "json-data"
    return cls(
        source_type="json",
        identifier=ident,
        mime_type="application/json",
        size_bytes=len(content),
        content_loader=lambda: content,
    )

from_file classmethod

from_file(path, *, mime_type=None)

Create a Source from a local file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the file. Must exist or SourceError is raised.	required
mime_type	str | None	MIME type override. Auto-detected from extension when None.	None

Source code in src/pollux/source.py

@classmethod
def from_file(cls, path: str | Path, *, mime_type: str | None = None) -> Source:
    """Create a Source from a local file.

    Args:
        path: Path to the file. Must exist or ``SourceError`` is raised.
        mime_type: MIME type override. Auto-detected from extension when *None*.
    """
    p = Path(path)
    if not p.exists():
        raise SourceError(f"File not found: {p}")

    mt = mime_type or mimetypes.guess_type(str(p))[0] or "application/octet-stream"
    size = p.stat().st_size

    def loader() -> bytes:
        return p.read_bytes()

    return cls(
        source_type="file",
        identifier=str(p),
        mime_type=mt,
        size_bytes=size,
        content_loader=loader,
    )

from_youtube classmethod

from_youtube(url)

Create a Source from a YouTube URL reference (no download).

Source code in src/pollux/source.py

@classmethod
def from_youtube(cls, url: str) -> Source:
    """Create a Source from a YouTube URL reference (no download)."""
    encoded = f"youtube:{url}".encode()
    return cls(
        source_type="youtube",
        identifier=url,
        mime_type="video/mp4",
        size_bytes=0,
        content_loader=lambda: encoded,
    )

from_uri classmethod

from_uri(uri, *, mime_type='application/octet-stream')

Create a Source from a URI.

Parameters:

Name	Type	Description	Default
uri	str	Remote URI (e.g. gs:// or https://).	required
mime_type	str	MIME type. Defaults to application/octet-stream.	'application/octet-stream'

Source code in src/pollux/source.py

@classmethod
def from_uri(
    cls, uri: str, *, mime_type: str = "application/octet-stream"
) -> Source:
    """Create a Source from a URI.

    Args:
        uri: Remote URI (e.g. ``gs://`` or ``https://``).
        mime_type: MIME type. Defaults to ``application/octet-stream``.
    """
    encoded = f"uri:{mime_type}:{uri}".encode()
    return cls(
        source_type="uri",
        identifier=uri,
        mime_type=mime_type,
        size_bytes=0,
        content_loader=lambda: encoded,
    )

from_arxiv classmethod

from_arxiv(ref)

Create an arXiv PDF Source from an arXiv ID or URL.

Parameters:

Name	Type	Description	Default
ref	str	An arXiv ID (e.g. "2301.07041") or full arXiv URL.	required

Source code in src/pollux/source.py

@classmethod
def from_arxiv(cls, ref: str) -> Source:
    """Create an arXiv PDF Source from an arXiv ID or URL.

    Args:
        ref: An arXiv ID (e.g. ``"2301.07041"``) or full arXiv URL.
    """
    if not isinstance(ref, str):
        raise TypeError("ref must be a str")

    normalized_url = cls._normalize_arxiv_to_pdf_url(ref.strip())
    encoded = normalized_url.encode("utf-8")
    return cls(
        source_type="arxiv",
        identifier=normalized_url,
        mime_type="application/pdf",
        size_bytes=0,
        content_loader=lambda: encoded,
    )

gemini_video_settings_for

gemini_video_settings_for(provider)

Return Gemini video settings when the active provider can use them.

Source code in src/pollux/source.py

def gemini_video_settings_for(
    self, provider: str | None
) -> dict[str, str | float] | None:
    """Return Gemini video settings when the active provider can use them."""
    provider_hints = self.provider_hints_for(provider)
    if provider_hints is None:
        return None
    return provider_hints.get("video_metadata")

provider_hints_for

provider_hints_for(provider)

Return immutable provider hints as plain dictionaries for transport.

Source code in src/pollux/source.py

def provider_hints_for(
    self, provider: str | None
) -> dict[str, dict[str, str | float]] | None:
    """Return immutable provider hints as plain dictionaries for transport."""
    if provider is None:
        return None

    hints = {
        hint.name: hint.payload_dict()
        for hint in self.provider_hints
        if hint.provider == provider
    }
    return hints or None

cache_identity_hash

cache_identity_hash(*, provider=None)

Compute SHA256 hash for cache identity.

Includes provider-visible source semantics such as Gemini video settings. Falls back to raw content hash when no provider-specific settings apply, preserving backward-compatible cache keys.

Source code in src/pollux/source.py

def cache_identity_hash(self, *, provider: str | None = None) -> str:
    """Compute SHA256 hash for cache identity.

    Includes provider-visible source semantics such as Gemini video settings.
    Falls back to raw content hash when no provider-specific settings apply,
    preserving backward-compatible cache keys.
    """
    provider_hints = self.provider_hints_for(provider)
    if provider_hints is None:
        return self._content_hash()
    combined = (
        self._content_hash()
        + "|"
        + json.dumps(
            provider_hints,
            sort_keys=True,
            separators=(",", ":"),
        )
    )
    return hashlib.sha256(combined.encode("utf-8")).hexdigest()

with_gemini_video_settings

with_gemini_video_settings(
    *, start_offset=None, end_offset=None, fps=None
)

Return a copy with validated Gemini video controls attached.

Pollux keeps this API provider-specific and stable even if Google's underlying wire fields evolve. The Gemini adapter maps these settings to the current SDK request shape. These settings only affect Gemini requests and Gemini explicit-cache identity.

Source code in src/pollux/source.py

def with_gemini_video_settings(
    self,
    *,
    start_offset: str | None = None,
    end_offset: str | None = None,
    fps: float | None = None,
) -> Source:
    """Return a copy with validated Gemini video controls attached.

    Pollux keeps this API provider-specific and stable even if Google's
    underlying wire fields evolve. The Gemini adapter maps these settings
    to the current SDK request shape. These settings only affect Gemini
    requests and Gemini explicit-cache identity.
    """
    if not self._is_video_source():
        raise SourceError(
            "Gemini video settings require a video source "
            "(local video file, video URI, or YouTube URL)"
        )

    validated_start_offset: str | None = None
    validated_end_offset: str | None = None
    validated_fps: float | None = None

    if start_offset is not None:
        if not isinstance(start_offset, str) or not start_offset.strip():
            raise SourceError("start_offset must be a non-empty string")
        validated_start_offset = start_offset

    if end_offset is not None:
        if not isinstance(end_offset, str) or not end_offset.strip():
            raise SourceError("end_offset must be a non-empty string")
        validated_end_offset = end_offset

    if fps is not None:
        if isinstance(fps, bool) or not isinstance(fps, (int, float)):
            raise SourceError("fps must be a number")
        fps_value = float(fps)
        if fps_value <= 0 or fps_value > 24:
            raise SourceError("fps must be > 0 and <= 24")
        validated_fps = fps_value

    if (
        validated_start_offset is None
        and validated_end_offset is None
        and validated_fps is None
    ):
        raise SourceError(
            "Provide at least one Gemini video setting: "
            "start_offset, end_offset, or fps"
        )

    return replace(
        self,
        provider_hints=self._with_provider_hint(
            provider="gemini",
            name="video_metadata",
            payload={
                k: v
                for k, v in (
                    ("start_offset", validated_start_offset),
                    ("end_offset", validated_end_offset),
                    ("fps", validated_fps),
                )
                if v is not None
            },
        ),
    )

with_gemini_url_context

with_gemini_url_context()

Return a copy that asks Gemini to retrieve this HTTP(S) URI at request time.

Source code in src/pollux/source.py

def with_gemini_url_context(self) -> Source:
    """Return a copy that asks Gemini to retrieve this HTTP(S) URI at request time."""
    if self.source_type != "uri":
        raise SourceError("Gemini URL Context requires Source.from_uri(...)")

    parsed = urlparse(self.identifier)
    if parsed.scheme not in {"http", "https"}:
        raise SourceError("Gemini URL Context requires an HTTP(S) URI")

    return replace(
        self,
        provider_hints=self._with_provider_hint(
            provider="gemini",
            name="url_context",
            payload={},
        ),
    )

Immutable configuration for Pollux execution.

Provider and model are required for cloud providers—Pollux does not guess what you want. Local OpenAI-compatible servers may omit model when the server has a single configured model or otherwise does not require it. API keys are auto-resolved from standard environment variables.

Example

config = Config(provider="gemini", model="gemini-2.0-flash")

API key is automatically resolved from GEMINI_API_KEY

For self-hosted OpenAI-compatible servers, use provider="local" with a base_url (or set POLLUX_LOCAL_BASE_URL). No API key is required.

Source code in src/pollux/config.py

@dataclass(frozen=True)
class Config:
    """Immutable configuration for Pollux execution.

    Provider and model are required for cloud providers—Pollux does not guess
    what you want. Local OpenAI-compatible servers may omit ``model`` when the
    server has a single configured model or otherwise does not require it.
    API keys are auto-resolved from standard environment variables.

    Example:
        config = Config(provider="gemini", model="gemini-2.0-flash")
        # API key is automatically resolved from GEMINI_API_KEY

    For self-hosted OpenAI-compatible servers, use ``provider="local"`` with
    a ``base_url`` (or set ``POLLUX_LOCAL_BASE_URL``). No API key is required.
    """

    provider: ProviderName
    model: str | None = None
    #: Auto-resolved from the provider-specific API key env var when *None*.
    #: Optional for ``provider="local"``.
    api_key: str | None = None
    #: Required for ``provider="local"``; rejected for cloud providers.
    #: Falls back to ``POLLUX_LOCAL_BASE_URL`` when unset.
    base_url: str | None = None
    use_mock: bool = False
    request_concurrency: int = 6
    #: HTTP request timeout in seconds for providers that own their transport.
    #: Currently applied by the local OpenAI-compatible provider.
    request_timeout_s: float = 300.0
    retry: RetryPolicy = field(default_factory=RetryPolicy)
    #: Optional capability declarations that override the provider's static
    #: capabilities for this config (v2 interaction path). A declared capability
    #: wins over the provider's static value; undeclared capabilities fall back to
    #: it. Useful for local OpenAI-compatible servers whose support varies.
    capabilities: Mapping[str, bool] | None = None

    def __post_init__(self) -> None:
        """Auto-resolve credentials and validate configuration."""
        # Validate provider
        if self.provider not in _SUPPORTED_PROVIDERS:
            raise ConfigurationError(
                f"Unknown provider: {self.provider!r}",
                hint=(
                    "Supported providers: "
                    + ", ".join(repr(p) for p in _SUPPORTED_PROVIDERS)
                ),
            )

        # Validate numeric fields
        if not isinstance(self.request_concurrency, int):
            raise ConfigurationError(
                f"request_concurrency must be an integer, got {type(self.request_concurrency).__name__}",
                hint="Pass a whole number ≥ 1 for request_concurrency.",
            )
        if self.request_concurrency < 1:
            raise ConfigurationError(
                f"request_concurrency must be ≥ 1, got {self.request_concurrency}",
                hint="This controls how many API calls run in parallel.",
            )
        if not isinstance(self.request_timeout_s, int | float):
            raise ConfigurationError(
                f"request_timeout_s must be numeric, got {type(self.request_timeout_s).__name__}",
                hint="Pass a timeout in seconds, for example request_timeout_s=600.",
            )
        if self.request_timeout_s <= 0:
            raise ConfigurationError(
                f"request_timeout_s must be > 0, got {self.request_timeout_s}",
                hint="Pass a positive timeout in seconds.",
            )

        if self.provider == "local":
            # Local: resolve base_url for real calls, skip API-key resolution entirely.
            if self.base_url is None and not self.use_mock:
                env_url = _resolve_local_base_url()
                if env_url:
                    object.__setattr__(self, "base_url", env_url)
            if not self.use_mock and not self.base_url:
                raise ConfigurationError(
                    "base_url required for provider='local'",
                    hint=(
                        "Pass base_url='http://localhost:...' or set "
                        f"{_LOCAL_BASE_URL_ENV_VAR}."
                    ),
                )
            return

        if not isinstance(self.model, str) or not self.model:
            raise ConfigurationError(
                f"model required for provider={self.provider!r}",
                hint="Pass the provider model name, for example model='gpt-5-nano'.",
            )

        # Cloud providers: base_url is not a meaningful override here.
        if self.base_url is not None:
            raise ConfigurationError(
                f"base_url is only valid for provider='local', not {self.provider!r}",
                hint="Remove base_url, or switch to provider='local'.",
            )

        # Auto-resolve API key from environment if not provided
        if self.api_key is None and not self.use_mock:
            object.__setattr__(self, "api_key", resolve_api_key(self.provider))

        # Validate: real API calls need a key
        if not self.use_mock and not self.api_key:
            env_var = _API_KEY_ENV_VARS[self.provider]
            raise ConfigurationError(
                f"API key required for {self.provider}",
                hint=f"Set {env_var} environment variable or pass api_key=...",
            )

    def __str__(self) -> str:
        """Return a redacted, developer-friendly representation."""
        return (
            f"Config(provider={self.provider!r}, model={self.model!r}, "
            f"api_key={'[REDACTED]' if self.api_key else None}, "
            f"base_url={self.base_url!r}, use_mock={self.use_mock})"
        )

    __repr__ = __str__

Serializable Pollux handle for a deferred job.

Source code in src/pollux/deferred.py

@dataclass(frozen=True)
class DeferredHandle:
    """Serializable Pollux handle for a deferred job."""

    job_id: str
    provider: str
    model: str
    request_count: int
    submitted_at: float
    schema_hash: str | None = None
    provider_state: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize the handle for persistence."""
        return dict(asdict(self))

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> DeferredHandle:
        """Rebuild a handle from serialized data."""
        return cls(
            job_id=str(data["job_id"]),
            provider=str(data["provider"]),
            model=str(data["model"]),
            request_count=int(data["request_count"]),
            submitted_at=float(data["submitted_at"]),
            schema_hash=(
                None
                if data.get("schema_hash") is None
                else str(data.get("schema_hash"))
            ),
            provider_state=(
                dict(data["provider_state"])
                if isinstance(data.get("provider_state"), dict)
                else None
            ),
        )

Functions

to_dict

to_dict()

Serialize the handle for persistence.

Source code in src/pollux/deferred.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the handle for persistence."""
    return dict(asdict(self))

from_dict classmethod

from_dict(data)

Rebuild a handle from serialized data.

Source code in src/pollux/deferred.py

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> DeferredHandle:
    """Rebuild a handle from serialized data."""
    return cls(
        job_id=str(data["job_id"]),
        provider=str(data["provider"]),
        model=str(data["model"]),
        request_count=int(data["request_count"]),
        submitted_at=float(data["submitted_at"]),
        schema_hash=(
            None
            if data.get("schema_hash") is None
            else str(data.get("schema_hash"))
        ),
        provider_state=(
            dict(data["provider_state"])
            if isinstance(data.get("provider_state"), dict)
            else None
        ),
    )

Normalized snapshot of a deferred job lifecycle.

Source code in src/pollux/deferred.py

@dataclass(frozen=True)
class DeferredSnapshot:
    """Normalized snapshot of a deferred job lifecycle."""

    job_id: str
    provider: str
    model: str
    status: DeferredStatus
    provider_status: str
    request_count: int
    succeeded: int
    failed: int
    pending: int
    submitted_at: float
    completed_at: float | None = None
    expires_at: float | None = None

    @property
    def is_terminal(self) -> bool:
        """Return True when the job is ready to collect or permanently done."""
        return self.status in _TERMINAL_STATUSES

Attributes

is_terminal property

is_terminal

Return True when the job is ready to collect or permanently done.

Bounded retry policy with exponential backoff and optional jitter.

Source code in src/pollux/retry.py

@dataclass(frozen=True)
class RetryPolicy:
    """Bounded retry policy with exponential backoff and optional jitter."""

    # Defaults are intentionally conservative: retries should help without
    # surprising tail-latency.
    #: Total attempts including the initial call (``2`` = one retry).
    max_attempts: int = 2
    initial_delay_s: float = 0.5
    backoff_multiplier: float = 2.0
    max_delay_s: float = 5.0
    jitter: bool = True  # "full jitter" when enabled
    #: Wall-clock deadline across all attempts; *None* disables the deadline.
    max_elapsed_s: float | None = 15.0

    def __post_init__(self) -> None:
        """Validate invariants to keep retry behavior predictable."""
        if self.max_attempts < 1:
            raise ValueError("RetryPolicy.max_attempts must be >= 1")
        if self.initial_delay_s < 0:
            raise ValueError("RetryPolicy.initial_delay_s must be >= 0")
        if self.backoff_multiplier <= 0:
            raise ValueError("RetryPolicy.backoff_multiplier must be > 0")
        if self.max_delay_s < 0:
            raise ValueError("RetryPolicy.max_delay_s must be >= 0")
        if self.max_elapsed_s is not None and self.max_elapsed_s < 0:
            raise ValueError("RetryPolicy.max_elapsed_s must be >= 0 or None")

Interaction Types (2.0)

The canonical v2 interaction model. interact() takes an Environment and an Input and returns an Output; OutputCollection is the source-pattern aggregate that run_many() and collect_deferred() return. The 1.x Options and ResultEnvelope types are no longer part of the public API.

The reusable, stable model-facing setup around interactions.

sources and tools accept any ordered sequence and are frozen to tuples. Tool declarations must be :class:ToolDeclaration objects; build one from a raw dict schema with :meth:ToolDeclaration.from_dict.

Source code in src/pollux/interaction/environment.py

@dataclass(frozen=True, slots=True)
class Environment:
    """The reusable, stable model-facing setup around interactions.

    ``sources`` and ``tools`` accept any ordered sequence and are frozen to
    tuples. Tool declarations must be :class:`ToolDeclaration` objects; build one
    from a raw dict schema with :meth:`ToolDeclaration.from_dict`.
    """

    instructions: str | None = None
    sources: Sequence[Source] = ()
    tools: Sequence[ToolDeclaration] = ()
    cache: CacheSetting = None
    metadata: dict[str, Any] | None = None

    def __post_init__(self) -> None:
        """Freeze the source and tool sequences to immutable tuples."""
        object.__setattr__(self, "sources", tuple(self.sources))
        object.__setattr__(self, "tools", tuple(self.tools))

    def fingerprint(self, *, provider: str) -> str:
        """Return stable identity for the provider-visible environment.

        The fingerprint covers instructions, ordered sources, ordered tools,
        and the active provider. Model identity belongs to :class:`Config` and
        must be composed separately by durable runtimes. Cache preferences and
        metadata are deliberately excluded because they do not change the
        model-visible environment.

        Fingerprints remain stable across compatible Pollux releases. A future
        semantic change will increment the embedded fingerprint version.
        """
        if not isinstance(provider, str) or not provider:
            from pollux.errors import ConfigurationError

            raise ConfigurationError(
                "Environment fingerprint requires a non-empty provider",
                hint="Pass config.provider when creating durable identity.",
            )
        payload = {
            "version": _FINGERPRINT_VERSION,
            "provider": provider,
            "instructions": self.instructions,
            "sources": [
                source._environment_identity(provider=provider)
                for source in self.sources
            ],
            "tools": [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters,
                    "strict": tool.strict,
                }
                for tool in self.tools
            ],
        }
        try:
            encoded = json.dumps(
                payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
            ).encode("utf-8")
        except (TypeError, ValueError) as exc:
            from pollux.errors import ConfigurationError

            raise ConfigurationError(
                "Environment identity is not JSON serializable",
                hint="Use JSON-compatible tool schemas and provider hints.",
            ) from exc
        return hashlib.sha256(encoded).hexdigest()

Functions

fingerprint

fingerprint(*, provider)

Return stable identity for the provider-visible environment.

The fingerprint covers instructions, ordered sources, ordered tools, and the active provider. Model identity belongs to :class:Config and must be composed separately by durable runtimes. Cache preferences and metadata are deliberately excluded because they do not change the model-visible environment.

Fingerprints remain stable across compatible Pollux releases. A future semantic change will increment the embedded fingerprint version.

Source code in src/pollux/interaction/environment.py

def fingerprint(self, *, provider: str) -> str:
    """Return stable identity for the provider-visible environment.

    The fingerprint covers instructions, ordered sources, ordered tools,
    and the active provider. Model identity belongs to :class:`Config` and
    must be composed separately by durable runtimes. Cache preferences and
    metadata are deliberately excluded because they do not change the
    model-visible environment.

    Fingerprints remain stable across compatible Pollux releases. A future
    semantic change will increment the embedded fingerprint version.
    """
    if not isinstance(provider, str) or not provider:
        from pollux.errors import ConfigurationError

        raise ConfigurationError(
            "Environment fingerprint requires a non-empty provider",
            hint="Pass config.provider when creating durable identity.",
        )
    payload = {
        "version": _FINGERPRINT_VERSION,
        "provider": provider,
        "instructions": self.instructions,
        "sources": [
            source._environment_identity(provider=provider)
            for source in self.sources
        ],
        "tools": [
            {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters,
                "strict": tool.strict,
            }
            for tool in self.tools
        ],
    }
    try:
        encoded = json.dumps(
            payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        from pollux.errors import ConfigurationError

        raise ConfigurationError(
            "Environment identity is not JSON serializable",
            hint="Use JSON-compatible tool schemas and provider hints.",
        ) from exc
    return hashlib.sha256(encoded).hexdigest()

One model turn: user content plus optional prior state and tool results.

history and tool_results accept any ordered sequence and are frozen to tuples.

Source code in src/pollux/interaction/input.py

@dataclass(frozen=True, slots=True)
class Input:
    """One model turn: user content plus optional prior state and tool results.

    ``history`` and ``tool_results`` accept any ordered sequence and are frozen
    to tuples.
    """

    content: str | None = None
    history: Sequence[Message] | None = None
    continuation: Continuation | None = None
    tool_results: Sequence[ToolResult] = ()

    def __post_init__(self) -> None:
        """Coerce sequences to tuples and validate the turn is well formed."""
        if self.history is not None:
            object.__setattr__(self, "history", tuple(self.history))
        object.__setattr__(self, "tool_results", tuple(self.tool_results))

        if self.history is not None and self.continuation is not None:
            raise ConfigurationError(
                "history and continuation are mutually exclusive",
                hint="Use exactly one prior-turn state source per interaction.",
            )

        has_content = bool(self.content and self.content.strip())
        if not has_content and not self.tool_results:
            raise ConfigurationError(
                "Input has no user content and no tool results",
                hint="Provide content for a new turn, or tool_results to "
                "continue from prior tool calls.",
            )

The completed result of one model interaction, as named facets.

Source code in src/pollux/interaction/output.py

@dataclass(frozen=True, slots=True)
class Output:
    """The completed result of one model interaction, as named facets."""

    text: str = ""
    structured: Any = None
    reasoning: str | None = None
    tool_calls: tuple[ToolCall, ...] = ()
    continuation: Continuation | None = None
    usage: Usage = field(default_factory=Usage)
    metrics: Metrics = field(default_factory=Metrics)
    diagnostics: Diagnostics = field(default_factory=Diagnostics)

    def to_jsonable(self) -> dict[str, Any]:
        """Serialize to a JSON-compatible dict (optional facets omitted)."""
        payload: dict[str, Any] = {"text": self.text}
        if self.structured is not None:
            payload["structured"] = _jsonable_structured(self.structured)
        if self.reasoning is not None:
            payload["reasoning"] = self.reasoning
        if self.tool_calls:
            payload["tool_calls"] = [tc.to_jsonable() for tc in self.tool_calls]
        if self.continuation is not None:
            payload["continuation"] = self.continuation.to_jsonable()
        payload["usage"] = self.usage.to_jsonable()
        payload["metrics"] = self.metrics.to_jsonable()
        diagnostics = self.diagnostics.to_jsonable()
        if diagnostics:
            payload["diagnostics"] = diagnostics
        return payload

Functions

to_jsonable

to_jsonable()

Serialize to a JSON-compatible dict (optional facets omitted).

Source code in src/pollux/interaction/output.py

def to_jsonable(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict (optional facets omitted)."""
    payload: dict[str, Any] = {"text": self.text}
    if self.structured is not None:
        payload["structured"] = _jsonable_structured(self.structured)
    if self.reasoning is not None:
        payload["reasoning"] = self.reasoning
    if self.tool_calls:
        payload["tool_calls"] = [tc.to_jsonable() for tc in self.tool_calls]
    if self.continuation is not None:
        payload["continuation"] = self.continuation.to_jsonable()
    payload["usage"] = self.usage.to_jsonable()
    payload["metrics"] = self.metrics.to_jsonable()
    diagnostics = self.diagnostics.to_jsonable()
    if diagnostics:
        payload["diagnostics"] = diagnostics
    return payload

One event in the timeline of a streamed interaction.

Only the facet relevant to type is set: text for text_delta / reasoning_delta, delta for tool_call_delta, tool_call for tool_call, usage for usage, finish_reason for finish, and output for done.

Source code in src/pollux/interaction/event.py

@dataclass(frozen=True, slots=True)
class Event:
    """One event in the timeline of a streamed interaction.

    Only the facet relevant to ``type`` is set: ``text`` for ``text_delta`` /
    ``reasoning_delta``, ``delta`` for ``tool_call_delta``, ``tool_call`` for
    ``tool_call``, ``usage`` for ``usage``, ``finish_reason`` for ``finish``, and
    ``output`` for ``done``.
    """

    type: EventType
    text: str = ""
    delta: ToolCallDelta | None = None
    tool_call: ToolCall | None = None
    usage: Usage | None = None
    finish_reason: str | None = None
    output: Output | None = None

Aggregate result for a multi-call source pattern.

Source code in src/pollux/interaction/collection.py

@dataclass(frozen=True, slots=True)
class OutputCollection:
    """Aggregate result for a multi-call source pattern."""

    outputs: tuple[Output, ...] = ()
    prompt_indexes: tuple[int, ...] | None = None
    source_indexes: tuple[int, ...] | None = None

    def __post_init__(self) -> None:
        """Coerce the outputs sequence to an immutable tuple."""
        object.__setattr__(self, "outputs", tuple(self.outputs))

    @property
    def answers(self) -> list[str]:
        """Per-interaction primary text, in input order."""
        return [output.text for output in self.outputs]

    @property
    def structured(self) -> list[Any]:
        """Per-interaction structured payloads, in input order."""
        return [output.structured for output in self.outputs]

    @property
    def usage(self) -> Usage:
        """Token usage summed across interactions."""
        input_tokens = sum(o.usage.input_tokens for o in self.outputs)
        output_tokens = sum(o.usage.output_tokens for o in self.outputs)
        total_tokens = sum(o.usage.total_tokens for o in self.outputs)
        reasoning = [
            o.usage.reasoning_tokens
            for o in self.outputs
            if o.usage.reasoning_tokens is not None
        ]
        cached = [
            o.usage.cached_tokens
            for o in self.outputs
            if o.usage.cached_tokens is not None
        ]
        return Usage(
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=total_tokens,
            reasoning_tokens=sum(reasoning) if reasoning else None,
            cached_tokens=sum(cached) if cached else None,
        )

    @property
    def status(self) -> CollectionStatus:
        """Partial-completion status based on answer presence."""
        if not self.outputs:
            return "ok"
        empty = sum(1 for output in self.outputs if not output.text.strip())
        if empty == len(self.outputs):
            return "error"
        if empty > 0:
            return "partial"
        return "ok"

    def to_jsonable(self) -> dict[str, Any]:
        """Serialize to a JSON-compatible dict with per-output detail and aggregates."""
        payload: dict[str, Any] = {
            "status": self.status,
            "outputs": [output.to_jsonable() for output in self.outputs],
            "usage": self.usage.to_jsonable(),
        }
        if self.prompt_indexes is not None:
            payload["prompt_indexes"] = list(self.prompt_indexes)
        if self.source_indexes is not None:
            payload["source_indexes"] = list(self.source_indexes)
        return payload

Attributes

answers property

answers

Per-interaction primary text, in input order.

structured property

structured

Per-interaction structured payloads, in input order.

usage property

usage

Token usage summed across interactions.

status property

status

Partial-completion status based on answer presence.

Functions

to_jsonable

to_jsonable()

Serialize to a JSON-compatible dict with per-output detail and aggregates.

Source code in src/pollux/interaction/collection.py

def to_jsonable(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict with per-output detail and aggregates."""
    payload: dict[str, Any] = {
        "status": self.status,
        "outputs": [output.to_jsonable() for output in self.outputs],
        "usage": self.usage.to_jsonable(),
    }
    if self.prompt_indexes is not None:
        payload["prompt_indexes"] = list(self.prompt_indexes)
    if self.source_indexes is not None:
        payload["source_indexes"] = list(self.source_indexes)
    return payload

Per-interaction controls for the response Pollux asks the model to produce.

Source code in src/pollux/interaction/requirements.py

@dataclass(frozen=True, slots=True)
class OutputRequirements:
    """Per-interaction controls for the response Pollux asks the model to produce."""

    output_schema: ResponseSchemaInput | None = None
    temperature: float | None = None
    top_p: float | None = None
    max_tokens: int | None = None
    seed: int | None = None
    reasoning_effort: str | None = None
    reasoning_budget_tokens: int | None = None
    tool_choice: ToolChoice | None = None
    #: Raw provider-scoped generation options keyed by provider name. Passed
    #: through without normalization at the active provider boundary.
    provider_options: dict[str, dict[str, Any]] | None = None

    def __post_init__(self) -> None:
        """Validate requirement shapes early for clear errors."""
        if self.output_schema is not None and not (
            isinstance(self.output_schema, dict)
            or (
                isinstance(self.output_schema, type)
                and _is_base_model(self.output_schema)
            )
        ):
            raise ConfigurationError(
                "output_schema must be a Pydantic model class or JSON schema dict",
                hint="Pass a BaseModel subclass or a dict following JSON Schema.",
            )
        if self.max_tokens is not None and (
            isinstance(self.max_tokens, bool)
            or not isinstance(self.max_tokens, int)
            or self.max_tokens <= 0
        ):
            raise ConfigurationError(
                "max_tokens must be a positive integer",
                hint="Pass max_tokens=16384 or greater for thinking models.",
            )
        if self.reasoning_budget_tokens is not None and (
            isinstance(self.reasoning_budget_tokens, bool)
            or not isinstance(self.reasoning_budget_tokens, int)
            or self.reasoning_budget_tokens < 0
        ):
            raise ConfigurationError(
                "reasoning_budget_tokens must be a non-negative integer",
                hint="Pass reasoning_budget_tokens=0 or a larger integer.",
            )
        if (
            self.reasoning_effort is not None
            and self.reasoning_budget_tokens is not None
        ):
            raise ConfigurationError(
                "reasoning_effort and reasoning_budget_tokens are mutually exclusive",
                hint="Choose either qualitative effort or an explicit token budget.",
            )
        if self.seed is not None and (
            isinstance(self.seed, bool) or not isinstance(self.seed, int)
        ):
            raise ConfigurationError(
                "seed must be an integer",
                hint="Pass seed=42 for reproducible sampling where supported.",
            )
        if self.provider_options is not None:
            _validate_provider_options(self.provider_options)

    def output_schema_json(self) -> dict[str, Any] | None:
        """Return JSON Schema for provider APIs."""
        return response_schema_json(self.output_schema)

    def output_schema_model(self) -> type[BaseModel] | None:
        """Return the Pydantic model class when one was provided."""
        return response_schema_model(self.output_schema)

    def output_schema_hash(self) -> str | None:
        """Return a stable hash of the JSON Schema."""
        return response_schema_hash(self.output_schema)

    def provider_options_for(self, provider: str) -> dict[str, Any] | None:
        """Return raw generation options for the active provider."""
        if self.provider_options is None:
            return None
        payload = self.provider_options.get(provider)
        return dict(payload) if payload is not None else None

Functions

output_schema_json

output_schema_json()

Return JSON Schema for provider APIs.

Source code in src/pollux/interaction/requirements.py

def output_schema_json(self) -> dict[str, Any] | None:
    """Return JSON Schema for provider APIs."""
    return response_schema_json(self.output_schema)

output_schema_model

output_schema_model()

Return the Pydantic model class when one was provided.

Source code in src/pollux/interaction/requirements.py

def output_schema_model(self) -> type[BaseModel] | None:
    """Return the Pydantic model class when one was provided."""
    return response_schema_model(self.output_schema)

output_schema_hash

output_schema_hash()

Return a stable hash of the JSON Schema.

Source code in src/pollux/interaction/requirements.py

def output_schema_hash(self) -> str | None:
    """Return a stable hash of the JSON Schema."""
    return response_schema_hash(self.output_schema)

provider_options_for

provider_options_for(provider)

Return raw generation options for the active provider.

Source code in src/pollux/interaction/requirements.py

def provider_options_for(self, provider: str) -> dict[str, Any] | None:
    """Return raw generation options for the active provider."""
    if self.provider_options is None:
        return None
    payload = self.provider_options.get(provider)
    return dict(payload) if payload is not None else None

Opaque, serializable state for provider-correct replay.

Applications receive this value from :attr:Output.continuation, persist it through :meth:to_jsonable / :meth:from_jsonable, and pass it back through Input(continuation=...). Replay fields are intentionally not public.

Source code in src/pollux/interaction/continuation.py

@dataclass(frozen=True, slots=True, init=False, repr=False)
class Continuation:
    """Opaque, serializable state for provider-correct replay.

    Applications receive this value from :attr:`Output.continuation`, persist it
    through :meth:`to_jsonable` / :meth:`from_jsonable`, and pass it back through
    ``Input(continuation=...)``. Replay fields are intentionally not public.
    """

    SCHEMA_VERSION: ClassVar[int] = SCHEMA_VERSION
    __state: _ContinuationState

    def __init__(self) -> None:
        raise TypeError(
            "Continuation values are created by Pollux outputs or "
            "Continuation.from_jsonable()"
        )

    def __repr__(self) -> str:
        """Return a representation that does not reveal replay state."""
        return f"Continuation(version={SCHEMA_VERSION})"

    def to_jsonable(self) -> dict[str, Any]:
        """Return a defensive JSON-compatible serialization of this handle."""
        state = self.__state
        payload: dict[str, Any] = {
            "version": SCHEMA_VERSION,
            "provider": state.provider,
            "messages": [message.to_jsonable() for message in state.messages],
        }
        if state.response_id is not None:
            payload["response_id"] = state.response_id
        if state.provider_state is not None:
            payload["provider_state"] = deepcopy(state.provider_state)
        return payload

    @classmethod
    def from_jsonable(
        cls,
        data: Mapping[str, Any],
        *,
        expected_provider: str | None = None,
    ) -> Continuation:
        """Restore a versioned artifact and optionally verify its provider."""
        version = data.get("version")
        if version != SCHEMA_VERSION:
            raise PolluxError(
                f"Incompatible continuation: expected schema version "
                f"{SCHEMA_VERSION}, got {version!r}",
                hint="This continuation was produced by a different Pollux "
                "version. Start a new interaction instead of reusing it.",
            )
        provider = data.get("provider")
        if not isinstance(provider, str) or not provider:
            raise PolluxError(
                "Incompatible continuation: missing provider identity",
                hint="Start a new interaction instead of editing continuation state.",
            )
        if expected_provider is not None and provider != expected_provider:
            raise PolluxError(
                f"Continuation provider {provider!r} does not match the active "
                f"provider {expected_provider!r}",
                hint="Reuse a continuation only with the provider that produced it.",
            )
        raw_messages = data.get("messages")
        if not isinstance(raw_messages, list):
            raise PolluxError("Incompatible continuation: messages must be a list")
        if not all(isinstance(message, Mapping) for message in raw_messages):
            raise PolluxError(
                "Incompatible continuation: every message must be an object"
            )
        response_id = data.get("response_id")
        provider_state = data.get("provider_state")
        return _new_continuation(
            messages=tuple(
                _ReplayMessage.from_jsonable(message) for message in raw_messages
            ),
            response_id=response_id if isinstance(response_id, str) else None,
            provider=provider,
            provider_state=deepcopy(provider_state)
            if isinstance(provider_state, dict)
            else None,
        )

Functions

to_jsonable

to_jsonable()

Return a defensive JSON-compatible serialization of this handle.

Source code in src/pollux/interaction/continuation.py

def to_jsonable(self) -> dict[str, Any]:
    """Return a defensive JSON-compatible serialization of this handle."""
    state = self.__state
    payload: dict[str, Any] = {
        "version": SCHEMA_VERSION,
        "provider": state.provider,
        "messages": [message.to_jsonable() for message in state.messages],
    }
    if state.response_id is not None:
        payload["response_id"] = state.response_id
    if state.provider_state is not None:
        payload["provider_state"] = deepcopy(state.provider_state)
    return payload

from_jsonable classmethod

from_jsonable(data, *, expected_provider=None)

Restore a versioned artifact and optionally verify its provider.

Source code in src/pollux/interaction/continuation.py

@classmethod
def from_jsonable(
    cls,
    data: Mapping[str, Any],
    *,
    expected_provider: str | None = None,
) -> Continuation:
    """Restore a versioned artifact and optionally verify its provider."""
    version = data.get("version")
    if version != SCHEMA_VERSION:
        raise PolluxError(
            f"Incompatible continuation: expected schema version "
            f"{SCHEMA_VERSION}, got {version!r}",
            hint="This continuation was produced by a different Pollux "
            "version. Start a new interaction instead of reusing it.",
        )
    provider = data.get("provider")
    if not isinstance(provider, str) or not provider:
        raise PolluxError(
            "Incompatible continuation: missing provider identity",
            hint="Start a new interaction instead of editing continuation state.",
        )
    if expected_provider is not None and provider != expected_provider:
        raise PolluxError(
            f"Continuation provider {provider!r} does not match the active "
            f"provider {expected_provider!r}",
            hint="Reuse a continuation only with the provider that produced it.",
        )
    raw_messages = data.get("messages")
    if not isinstance(raw_messages, list):
        raise PolluxError("Incompatible continuation: messages must be a list")
    if not all(isinstance(message, Mapping) for message in raw_messages):
        raise PolluxError(
            "Incompatible continuation: every message must be an object"
        )
    response_id = data.get("response_id")
    provider_state = data.get("provider_state")
    return _new_continuation(
        messages=tuple(
            _ReplayMessage.from_jsonable(message) for message in raw_messages
        ),
        response_id=response_id if isinstance(response_id, str) else None,
        provider=provider,
        provider_state=deepcopy(provider_state)
        if isinstance(provider_state, dict)
        else None,
    )

A portable, application-authored text or tool transcript message.

Source code in src/pollux/interaction/continuation.py

@dataclass(frozen=True, slots=True)
class Message:
    """A portable, application-authored text or tool transcript message."""

    role: MessageRole
    content: str = ""
    tool_calls: Sequence[ToolCall] = ()
    tool_call_id: str | None = None

    def __post_init__(self) -> None:
        """Freeze tool calls and reject shapes adapters cannot portably replay."""
        if self.role not in _MESSAGE_ROLES:
            raise ConfigurationError(
                f"Unsupported history message role: {self.role!r}",
                hint="Use 'user', 'assistant', or 'tool'. Put system instructions "
                "on Environment.instructions.",
            )
        if not isinstance(self.content, str):
            raise ConfigurationError(
                "Message content must be text",
                hint="Keep media in Source values or the current Input.",
            )
        calls = tuple(self.tool_calls)
        if not all(isinstance(call, ToolCall) for call in calls):
            raise ConfigurationError(
                "Message tool_calls must contain ToolCall values",
                hint="Normalize provider tool calls with ToolCall.from_text(...).",
            )
        if any(call.provider_state is not None for call in calls):
            raise ConfigurationError(
                "Portable history tool calls cannot contain provider state",
                hint="Rebuild transcript calls with ToolCall.from_text(...).",
            )
        object.__setattr__(self, "tool_calls", calls)

        if self.role == "user":
            if not self.content.strip():
                raise ConfigurationError("User history messages require non-empty text")
            if calls or self.tool_call_id is not None:
                raise ConfigurationError(
                    "User history messages cannot contain tool-call fields"
                )
        elif self.role == "assistant":
            if not self.content and not calls:
                raise ConfigurationError(
                    "Assistant history messages require text or tool calls"
                )
            if self.tool_call_id is not None:
                raise ConfigurationError(
                    "Assistant history messages cannot have tool_call_id"
                )
        else:
            if not isinstance(self.tool_call_id, str) or not self.tool_call_id.strip():
                raise ConfigurationError(
                    "Tool history messages require a non-empty tool_call_id"
                )
            if calls:
                raise ConfigurationError(
                    "Tool history messages cannot contain nested tool calls"
                )

    def to_jsonable(self) -> dict[str, Any]:
        """Serialize this portable transcript message."""
        payload: dict[str, Any] = {"role": self.role, "content": self.content}
        if self.tool_calls:
            payload["tool_calls"] = [call.to_jsonable() for call in self.tool_calls]
        if self.tool_call_id is not None:
            payload["tool_call_id"] = self.tool_call_id
        return payload

    @classmethod
    def from_jsonable(cls, data: Mapping[str, Any]) -> Message:
        """Parse and validate a portable transcript message."""
        role = data.get("role")
        if role not in _MESSAGE_ROLES:
            raise ConfigurationError(
                f"Unsupported history message role: {role!r}",
                hint="Use 'user', 'assistant', or 'tool'. Put system instructions "
                "on Environment.instructions.",
            )
        raw_calls = data.get("tool_calls")
        calls = _tool_calls_from_jsonable(raw_calls)
        content = data.get("content", "")
        if not isinstance(content, str):
            raise ConfigurationError("Message content must be text")
        tool_call_id = data.get("tool_call_id")
        return cls(
            role=cast("MessageRole", role),
            content=content,
            tool_calls=calls,
            tool_call_id=tool_call_id if isinstance(tool_call_id, str) else None,
        )

    @classmethod
    def from_openai(cls, data: Mapping[str, Any]) -> Message:
        """Build a portable message from one OpenAI Chat Completions message.

        System messages are intentionally rejected: stable system context belongs
        on :attr:`Environment.instructions`, not in portable turn history.
        """
        role = data.get("role")
        if role not in _MESSAGE_ROLES:
            raise ConfigurationError(
                f"Unsupported OpenAI history role: {role!r}; move system "
                "messages to Environment.instructions",
                hint="Move system messages to Environment.instructions.",
            )
        raw_calls = data.get("tool_calls")
        imported_calls = (
            tuple(
                ToolCall.from_openai(call)
                for call in raw_calls
                if isinstance(call, dict)
            )
            if isinstance(raw_calls, list)
            else ()
        )
        calls = tuple(
            ToolCall.from_text(
                id=call.id,
                name=call.name,
                arguments_text=call.arguments_text,
                index=call.index,
            )
            for call in imported_calls
        )
        tool_call_id = data.get("tool_call_id")
        return cls(
            role=cast("MessageRole", role),
            content=_openai_text_content(data.get("content")),
            tool_calls=calls,
            tool_call_id=tool_call_id if isinstance(tool_call_id, str) else None,
        )

    def to_openai(self) -> dict[str, Any]:
        """Serialize as one OpenAI Chat Completions transcript message."""
        payload: dict[str, Any] = {"role": self.role, "content": self.content}
        if self.tool_calls:
            payload["tool_calls"] = [call.to_openai() for call in self.tool_calls]
        if self.tool_call_id is not None:
            payload["tool_call_id"] = self.tool_call_id
        return payload

Functions

to_jsonable

to_jsonable()

Serialize this portable transcript message.

Source code in src/pollux/interaction/continuation.py

def to_jsonable(self) -> dict[str, Any]:
    """Serialize this portable transcript message."""
    payload: dict[str, Any] = {"role": self.role, "content": self.content}
    if self.tool_calls:
        payload["tool_calls"] = [call.to_jsonable() for call in self.tool_calls]
    if self.tool_call_id is not None:
        payload["tool_call_id"] = self.tool_call_id
    return payload

from_jsonable classmethod

from_jsonable(data)

Parse and validate a portable transcript message.

Source code in src/pollux/interaction/continuation.py

@classmethod
def from_jsonable(cls, data: Mapping[str, Any]) -> Message:
    """Parse and validate a portable transcript message."""
    role = data.get("role")
    if role not in _MESSAGE_ROLES:
        raise ConfigurationError(
            f"Unsupported history message role: {role!r}",
            hint="Use 'user', 'assistant', or 'tool'. Put system instructions "
            "on Environment.instructions.",
        )
    raw_calls = data.get("tool_calls")
    calls = _tool_calls_from_jsonable(raw_calls)
    content = data.get("content", "")
    if not isinstance(content, str):
        raise ConfigurationError("Message content must be text")
    tool_call_id = data.get("tool_call_id")
    return cls(
        role=cast("MessageRole", role),
        content=content,
        tool_calls=calls,
        tool_call_id=tool_call_id if isinstance(tool_call_id, str) else None,
    )

from_openai classmethod

from_openai(data)

Build a portable message from one OpenAI Chat Completions message.

System messages are intentionally rejected: stable system context belongs on :attr:Environment.instructions, not in portable turn history.

Source code in src/pollux/interaction/continuation.py

@classmethod
def from_openai(cls, data: Mapping[str, Any]) -> Message:
    """Build a portable message from one OpenAI Chat Completions message.

    System messages are intentionally rejected: stable system context belongs
    on :attr:`Environment.instructions`, not in portable turn history.
    """
    role = data.get("role")
    if role not in _MESSAGE_ROLES:
        raise ConfigurationError(
            f"Unsupported OpenAI history role: {role!r}; move system "
            "messages to Environment.instructions",
            hint="Move system messages to Environment.instructions.",
        )
    raw_calls = data.get("tool_calls")
    imported_calls = (
        tuple(
            ToolCall.from_openai(call)
            for call in raw_calls
            if isinstance(call, dict)
        )
        if isinstance(raw_calls, list)
        else ()
    )
    calls = tuple(
        ToolCall.from_text(
            id=call.id,
            name=call.name,
            arguments_text=call.arguments_text,
            index=call.index,
        )
        for call in imported_calls
    )
    tool_call_id = data.get("tool_call_id")
    return cls(
        role=cast("MessageRole", role),
        content=_openai_text_content(data.get("content")),
        tool_calls=calls,
        tool_call_id=tool_call_id if isinstance(tool_call_id, str) else None,
    )

to_openai

to_openai()

Serialize as one OpenAI Chat Completions transcript message.

Source code in src/pollux/interaction/continuation.py

def to_openai(self) -> dict[str, Any]:
    """Serialize as one OpenAI Chat Completions transcript message."""
    payload: dict[str, Any] = {"role": self.role, "content": self.content}
    if self.tool_calls:
        payload["tool_calls"] = [call.to_openai() for call in self.tool_calls]
    if self.tool_call_id is not None:
        payload["tool_call_id"] = self.tool_call_id
    return payload

A provider-neutral description of a tool the model may request.

Source code in src/pollux/interaction/tools.py

@dataclass(frozen=True, slots=True)
class ToolDeclaration:
    """A provider-neutral description of a tool the model may request."""

    name: str
    description: str = ""
    parameters: dict[str, Any] = field(default_factory=dict)
    strict: bool = True

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> ToolDeclaration:
        """Build a declaration from a flat or OpenAI-style ``function`` dict."""
        payload: Mapping[str, Any] = data
        nested = data.get("function")
        if isinstance(nested, dict):
            payload = nested
        name = payload.get("name")
        if not isinstance(name, str) or not name:
            raise ConfigurationError(
                "tool declaration requires a non-empty 'name'",
                hint="Pass {'name': 'get_weather', 'description': ..., "
                "'parameters': {...}}.",
            )
        description = payload.get("description", "")
        parameters = payload.get("parameters", {})
        strict = payload.get("strict", True)
        return cls(
            name=name,
            description=str(description) if description is not None else "",
            parameters=dict(parameters) if isinstance(parameters, dict) else {},
            strict=strict if isinstance(strict, bool) else True,
        )

Functions

from_dict classmethod

from_dict(data)

Build a declaration from a flat or OpenAI-style function dict.

Source code in src/pollux/interaction/tools.py

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ToolDeclaration:
    """Build a declaration from a flat or OpenAI-style ``function`` dict."""
    payload: Mapping[str, Any] = data
    nested = data.get("function")
    if isinstance(nested, dict):
        payload = nested
    name = payload.get("name")
    if not isinstance(name, str) or not name:
        raise ConfigurationError(
            "tool declaration requires a non-empty 'name'",
            hint="Pass {'name': 'get_weather', 'description': ..., "
            "'parameters': {...}}.",
        )
    description = payload.get("description", "")
    parameters = payload.get("parameters", {})
    strict = payload.get("strict", True)
    return cls(
        name=name,
        description=str(description) if description is not None else "",
        parameters=dict(parameters) if isinstance(parameters, dict) else {},
        strict=strict if isinstance(strict, bool) else True,
    )

A normalized tool request emitted by the model.

arguments_text is the exact raw provider argument string (after any stream fragments are assembled). arguments is the parsed JSON; when the text is not valid JSON it is None and arguments_error explains why. This preserves both convenience and recoverability.

Source code in src/pollux/interaction/tools.py

@dataclass(frozen=True, slots=True)
class ToolCall:
    """A normalized tool request emitted by the model.

    ``arguments_text`` is the exact raw provider argument string (after any
    stream fragments are assembled). ``arguments`` is the parsed JSON; when the
    text is not valid JSON it is ``None`` and ``arguments_error`` explains why.
    This preserves both convenience and recoverability.
    """

    id: str
    name: str
    arguments_text: str = ""
    arguments: JSONValue = None
    arguments_error: str | None = None
    index: int | None = None
    provider_state: dict[str, Any] | None = None

    @classmethod
    def from_text(
        cls,
        *,
        id: str,  # noqa: A002 - mirrors the public ToolCall.id field name
        name: str,
        arguments_text: str = "",
        index: int | None = None,
        provider_state: dict[str, Any] | None = None,
    ) -> ToolCall:
        """Build a call from raw provider fields, parsing arguments once."""
        arguments, arguments_error = _parse_arguments(arguments_text)
        return cls(
            id=id,
            name=name,
            arguments_text=arguments_text,
            arguments=arguments,
            arguments_error=arguments_error,
            index=index,
            provider_state=provider_state,
        )

    @classmethod
    def from_openai(cls, data: Mapping[str, Any]) -> ToolCall:
        """Build a tool call from an OpenAI Chat Completions ``tool_calls`` item."""
        function = data.get("function")
        function = function if isinstance(function, dict) else {}
        arguments = function.get("arguments", "")
        index = data.get("index")
        return cls.from_text(
            id=str(data.get("id", "")),
            name=str(function.get("name", "")),
            arguments_text=arguments if isinstance(arguments, str) else str(arguments),
            index=index if isinstance(index, int) else None,
            provider_state={"openai": dict(data)},
        )

    def arguments_dict(self) -> dict[str, Any]:
        """Return parsed JSON-object arguments or raise an explicit config error.

        Agent loops usually dispatch tools from object-shaped JSON arguments.
        This helper keeps that path strict: malformed JSON and non-object JSON
        are rejected instead of being silently treated as an empty dict.
        """
        if self.arguments_error is not None:
            raise ToolCallParseError(
                f"Tool call {self.name!r} has invalid JSON arguments",
                hint=self.arguments_error,
                tool_name=self.name,
                tool_call_id=self.id,
                arguments_text=self.arguments_text,
            )
        if self.arguments is None:
            return {}
        if not isinstance(self.arguments, dict):
            raise ToolCallParseError(
                f"Tool call {self.name!r} arguments must be a JSON object",
                hint="Define tool parameters as an object schema and retry the turn.",
                tool_name=self.name,
                tool_call_id=self.id,
                arguments_text=self.arguments_text,
            )
        return dict(self.arguments)

    def to_jsonable(self) -> dict[str, Any]:
        """Serialize to a compact JSON-compatible dict (optional facets omitted)."""
        payload: dict[str, Any] = {
            "id": self.id,
            "name": self.name,
            "arguments_text": self.arguments_text,
        }
        if self.arguments_error is not None:
            payload["arguments_error"] = self.arguments_error
        if self.index is not None:
            payload["index"] = self.index
        if self.provider_state is not None:
            payload["provider_state"] = self.provider_state
        return payload

    def to_openai(self) -> dict[str, Any]:
        """Serialize as an OpenAI Chat Completions ``tool_calls`` item."""
        payload: dict[str, Any] = {
            "id": self.id,
            "type": "function",
            "function": {
                "name": self.name,
                "arguments": self.arguments_text,
            },
        }
        if self.index is not None:
            payload["index"] = self.index
        return payload

Functions

from_text classmethod

from_text(
    *,
    id,
    name,
    arguments_text="",
    index=None,
    provider_state=None,
)

Build a call from raw provider fields, parsing arguments once.

Source code in src/pollux/interaction/tools.py

@classmethod
def from_text(
    cls,
    *,
    id: str,  # noqa: A002 - mirrors the public ToolCall.id field name
    name: str,
    arguments_text: str = "",
    index: int | None = None,
    provider_state: dict[str, Any] | None = None,
) -> ToolCall:
    """Build a call from raw provider fields, parsing arguments once."""
    arguments, arguments_error = _parse_arguments(arguments_text)
    return cls(
        id=id,
        name=name,
        arguments_text=arguments_text,
        arguments=arguments,
        arguments_error=arguments_error,
        index=index,
        provider_state=provider_state,
    )

from_openai classmethod

from_openai(data)

Build a tool call from an OpenAI Chat Completions tool_calls item.

Source code in src/pollux/interaction/tools.py

@classmethod
def from_openai(cls, data: Mapping[str, Any]) -> ToolCall:
    """Build a tool call from an OpenAI Chat Completions ``tool_calls`` item."""
    function = data.get("function")
    function = function if isinstance(function, dict) else {}
    arguments = function.get("arguments", "")
    index = data.get("index")
    return cls.from_text(
        id=str(data.get("id", "")),
        name=str(function.get("name", "")),
        arguments_text=arguments if isinstance(arguments, str) else str(arguments),
        index=index if isinstance(index, int) else None,
        provider_state={"openai": dict(data)},
    )

arguments_dict

arguments_dict()

Return parsed JSON-object arguments or raise an explicit config error.

Agent loops usually dispatch tools from object-shaped JSON arguments. This helper keeps that path strict: malformed JSON and non-object JSON are rejected instead of being silently treated as an empty dict.

Source code in src/pollux/interaction/tools.py

def arguments_dict(self) -> dict[str, Any]:
    """Return parsed JSON-object arguments or raise an explicit config error.

    Agent loops usually dispatch tools from object-shaped JSON arguments.
    This helper keeps that path strict: malformed JSON and non-object JSON
    are rejected instead of being silently treated as an empty dict.
    """
    if self.arguments_error is not None:
        raise ToolCallParseError(
            f"Tool call {self.name!r} has invalid JSON arguments",
            hint=self.arguments_error,
            tool_name=self.name,
            tool_call_id=self.id,
            arguments_text=self.arguments_text,
        )
    if self.arguments is None:
        return {}
    if not isinstance(self.arguments, dict):
        raise ToolCallParseError(
            f"Tool call {self.name!r} arguments must be a JSON object",
            hint="Define tool parameters as an object schema and retry the turn.",
            tool_name=self.name,
            tool_call_id=self.id,
            arguments_text=self.arguments_text,
        )
    return dict(self.arguments)

to_jsonable

to_jsonable()

Serialize to a compact JSON-compatible dict (optional facets omitted).

Source code in src/pollux/interaction/tools.py

def to_jsonable(self) -> dict[str, Any]:
    """Serialize to a compact JSON-compatible dict (optional facets omitted)."""
    payload: dict[str, Any] = {
        "id": self.id,
        "name": self.name,
        "arguments_text": self.arguments_text,
    }
    if self.arguments_error is not None:
        payload["arguments_error"] = self.arguments_error
    if self.index is not None:
        payload["index"] = self.index
    if self.provider_state is not None:
        payload["provider_state"] = self.provider_state
    return payload

to_openai

to_openai()

Serialize as an OpenAI Chat Completions tool_calls item.

Source code in src/pollux/interaction/tools.py

def to_openai(self) -> dict[str, Any]:
    """Serialize as an OpenAI Chat Completions ``tool_calls`` item."""
    payload: dict[str, Any] = {
        "id": self.id,
        "type": "function",
        "function": {
            "name": self.name,
            "arguments": self.arguments_text,
        },
    }
    if self.index is not None:
        payload["index"] = self.index
    return payload

Application-produced output returned to the model for a prior tool call.

Source code in src/pollux/interaction/tools.py

@dataclass(frozen=True, slots=True)
class ToolResult:
    """Application-produced output returned to the model for a prior tool call."""

    call_id: str
    content: str
    is_error: bool = False

    @classmethod
    def from_value(
        cls,
        *,
        call_id: str,
        value: JSONValue,
        is_error: bool = False,
    ) -> ToolResult:
        """Build a tool result from a JSON value, serializing non-strings.

        Strings are preserved as-is. Other JSON values are encoded compactly so
        agent loops can return structured tool output without repeating
        ``json.dumps`` at every dispatch site.
        """
        if isinstance(value, str):
            content = value
        else:
            try:
                content = json.dumps(value, ensure_ascii=False, sort_keys=True)
            except (TypeError, ValueError) as exc:
                raise ConfigurationError(
                    "ToolResult.from_value() requires a JSON-serializable value",
                    hint="Return a string, dict, list, number, boolean, or None.",
                ) from exc
        return cls(call_id=call_id, content=content, is_error=is_error)

    def to_jsonable(self) -> dict[str, Any]:
        """Serialize to a JSON-compatible dict."""
        payload: dict[str, Any] = {"call_id": self.call_id, "content": self.content}
        if self.is_error:
            payload["is_error"] = True
        return payload

Functions

from_value classmethod

from_value(*, call_id, value, is_error=False)

Build a tool result from a JSON value, serializing non-strings.

Strings are preserved as-is. Other JSON values are encoded compactly so agent loops can return structured tool output without repeating json.dumps at every dispatch site.

Source code in src/pollux/interaction/tools.py

@classmethod
def from_value(
    cls,
    *,
    call_id: str,
    value: JSONValue,
    is_error: bool = False,
) -> ToolResult:
    """Build a tool result from a JSON value, serializing non-strings.

    Strings are preserved as-is. Other JSON values are encoded compactly so
    agent loops can return structured tool output without repeating
    ``json.dumps`` at every dispatch site.
    """
    if isinstance(value, str):
        content = value
    else:
        try:
            content = json.dumps(value, ensure_ascii=False, sort_keys=True)
        except (TypeError, ValueError) as exc:
            raise ConfigurationError(
                "ToolResult.from_value() requires a JSON-serializable value",
                hint="Return a string, dict, list, number, boolean, or None.",
            ) from exc
    return cls(call_id=call_id, content=content, is_error=is_error)

to_jsonable

to_jsonable()

Serialize to a JSON-compatible dict.

Source code in src/pollux/interaction/tools.py

def to_jsonable(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    payload: dict[str, Any] = {"call_id": self.call_id, "content": self.content}
    if self.is_error:
        payload["is_error"] = True
    return payload

Error Types

Bases: Exception

Base exception for all Pollux errors.

Source code in src/pollux/errors.py

class PolluxError(Exception):
    """Base exception for all Pollux errors."""

    def __init__(self, message: str, *, hint: str | None = None) -> None:
        super().__init__(message)
        self.hint = hint

Bases: PolluxError

Configuration validation or resolution failed.

Source code in src/pollux/errors.py

class ConfigurationError(PolluxError):
    """Configuration validation or resolution failed."""

Bases: PolluxError

Source validation or loading failed.

Source code in src/pollux/errors.py

class SourceError(PolluxError):
    """Source validation or loading failed."""

Bases: PolluxError

Execution planning failed.

Source code in src/pollux/errors.py

class PlanningError(PolluxError):
    """Execution planning failed."""

Bases: PolluxError

A Pollux internal error (bug) or invariant violation.

Source code in src/pollux/errors.py

class InternalError(PolluxError):
    """A Pollux internal error (bug) or invariant violation."""

Bases: PolluxError

API call failed.

Providers attach retry metadata so core execution can perform bounded retries without brittle substring matching.

Source code in src/pollux/errors.py

class APIError(PolluxError):
    """API call failed.

    Providers attach retry metadata so core execution can perform bounded
    retries without brittle substring matching.
    """

    def __init__(
        self,
        message: str,
        *,
        hint: str | None = None,
        retryable: bool | None = None,
        status_code: int | None = None,
        retry_after_s: float | None = None,
        provider: str | None = None,
        phase: str | None = None,
        call_idx: int | None = None,
        error_category: str | None = None,
    ) -> None:
        super().__init__(message, hint=hint)
        self.retryable = retryable
        self.status_code = status_code
        self.retry_after_s = retry_after_s
        self.provider = provider
        self.phase = phase
        self.call_idx = call_idx
        self.error_category = error_category

Bases: APIError

Rate limit exceeded (HTTP 429).

Source code in src/pollux/errors.py

class RateLimitError(APIError):
    """Rate limit exceeded (HTTP 429)."""

Bases: APIError

Cache operation failed.

Source code in src/pollux/errors.py

class CacheError(APIError):
    """Cache operation failed."""

Bases: PolluxError

Deferred job is not yet in a terminal state.

Source code in src/pollux/errors.py

class DeferredNotReadyError(PolluxError):
    """Deferred job is not yet in a terminal state."""

    def __init__(self, snapshot: Any) -> None:
        super().__init__(
            "Deferred job is not ready to collect",
            hint="Inspect the attached snapshot and retry after the job reaches a terminal state.",
        )
        self.snapshot = snapshot