fractal.impl package#

Submodules#

fractal.impl.claude module#

Claude Code CLI agent implementation.

class fractal.impl.claude.ClaudeParser(*, model=None)[source]#

Bases: StreamParser

Parser for claude --output-format stream-json output.

Parameters:

Initialize ClaudeParser.

Track the last-priced message id to skip per-block repeats.

Parameters:
feed(line)[source]#

Parse one claude stream line into normalized events.

Parameters:
Return type:

list[StreamEvent]

class fractal.impl.claude.ClaudeAgent(node, command=None, provider=None)[source]#

Bases: Agent

Claude Code CLI backend (cost-reporting, caller-minted sessions).

Parameters:

  • self (Agent)

  • node (Node)

  • command (str | None)

  • provider (str | None)

Initialize Agent.

Parameters:

  • node (Node) – The node this agent runs for (config dirs, worktree, ledger).

  • command (str | None) – Full agent command (e.g. 'claude --some-flag'); defaults to the node’s configured agent. Extra words are spliced into every invocation.

  • provider (str | None) – Provider-route override; None means native (Node.agent threads the node’s effective configured route).

  • self (Agent)

name: str = 'claude'#
config_file: str = 'settings.json'#
can_fork: bool = True#
mints_session: bool = False#
needs_pricing: bool = False#
cost_scope: str = 'call'#
enforces_budget: bool = True#
providers: tuple[str, ...] = ('openrouter',)#
parser(*, model=None)[source]#

Return the stream parser for the bound route.

Parameters:
Return type:

StreamParser

tracks_cost(model=None)[source]#

Return whether spend will be recorded.

Natively claude reports its own figures on the stream; through openrouter the close is chain-priced from the result usage, so only a priced model tracks.

Parameters:
Return type:

bool

fractal.impl.codex module#

Codex CLI agent implementation.

class fractal.impl.codex.CodexParser(*, model=None)[source]#

Bases: StreamParser

Parser for codex exec --json output.

Parameters:

Initialize CodexParser.

Bind the configured-model fallback and start the wall clock.

Parameters:
feed(line)[source]#

Parse one codex JSONL line into normalized events.

Parameters:
Return type:

list[StreamEvent]

turn_usage()[source]#

Return the wire usage of the stream’s one complete turn.

Raises:

ValueError – Unless the stream opened one thread that named its id, started one turn and completed it.

Parameters:

self (CodexParser)

Return type:

Any

finish()[source]#

Close the drained stream on its settled cost and wall time.

Parameters:

self (CodexParser)

Return type:

list[StreamEvent]

class fractal.impl.codex.CodexAgent(node, command=None, provider=None)[source]#

Bases: Agent

Codex CLI backend (token-priced, agent-minted threads, no fork).

Parameters:

  • self (Agent)

  • node (Node)

  • command (str | None)

  • provider (str | None)

Initialize Agent.

Parameters:

  • node (Node) – The node this agent runs for (config dirs, worktree, ledger).

  • command (str | None) – Full agent command (e.g. 'claude --some-flag'); defaults to the node’s configured agent. Extra words are spliced into every invocation.

  • provider (str | None) – Provider-route override; None means native (Node.agent threads the node’s effective configured route).

  • self (Agent)

name: str = 'codex'#
config_file: str = 'config.toml'#
can_fork: bool = False#
mints_session: bool = True#
needs_pricing: bool = True#
cost_scope: str = 'call'#
enforces_budget: bool = False#
providers: tuple[str, ...] = ('openrouter',)#
class fractal.impl.codex.UsageWindow(home, session, existing, baseline, path=None, identity=None, prefix_sha256=None, error=None, size=0)[source]#

Bases: object

The session-log state captured before a codex spawn.

A codex step is priced from the per-response token_usage_record rows codex appends to its session log (CODEX_HOME/sessions/.../rollout-*-<thread>.jsonl) while the process runs, bound to that process by this window: capture runs before the process starts and reconcile after it exits, so the records between the two are the invocation’s own.

Parameters:

  • home (Path)

  • session (str | None)

  • existing (dict[Path, int])

  • baseline (dict[str, int])

  • path (Path | None)

  • identity (tuple[int, int] | None)

  • prefix_sha256 (str | None)

  • error (str | None)

  • size (int)

home: Path#
session: str | None#
existing: dict[Path, int]#
baseline: dict[str, int]#
path: Path | None = None#
identity: tuple[int, int] | None = None#
prefix_sha256: str | None = None#
error: str | None = None#
size: int = 0#
classmethod capture(invocation)[source]#

Capture the rollout state before the spawn; a failure rides error.

Parameters:

invocation (Invocation)

Return type:

UsageWindow

reconcile(session)[source]#

Return the usage and served model of the records the invocation appended.

Raises:

ValueError – When the rollout for session cannot be bound to the invocation, the invocation spawned sub-agent threads, or the records do not describe one counted turn.

Parameters:
Return type:

tuple[dict[str, int], str]

fractal.impl.grok module#

Grok Build CLI agent implementation.

class fractal.impl.grok.GrokParser(*, model=None)[source]#

Bases: StreamParser

Parser for grok --output-format streaming-json output.

Parameters:

Initialize GrokParser.

Bind the configured-model fallback and start the wall clock.

Parameters:
feed(line)[source]#

Parse one grok NDJSON event line into normalized events.

Parameters:
Return type:

list[StreamEvent]

class fractal.impl.grok.GrokAgent(node, command=None, provider=None)[source]#

Bases: Agent

Grok Build CLI backend (cost-reporting, caller-minted sessions).

Parameters:

  • self (Agent)

  • node (Node)

  • command (str | None)

  • provider (str | None)

Initialize Agent.

Parameters:

  • node (Node) – The node this agent runs for (config dirs, worktree, ledger).

  • command (str | None) – Full agent command (e.g. 'claude --some-flag'); defaults to the node’s configured agent. Extra words are spliced into every invocation.

  • provider (str | None) – Provider-route override; None means native (Node.agent threads the node’s effective configured route).

  • self (Agent)

name: str = 'grok'#
config_file: str = 'config.toml'#
can_fork: bool = True#
mints_session: bool = False#
needs_pricing: bool = False#
cost_scope: str = 'call'#
enforces_budget: bool = False#
providers: tuple[str, ...] = ()#

fractal.impl.omp module#

Oh My Pi (omp) CLI agent implementation.

class fractal.impl.omp.OmpParser(*, model=None)[source]#

Bases: StreamParser

Parser for omp -p --mode json output.

Parameters:

Initialize OmpParser.

Bind the configured-model fallback and start the wall clock.

Parameters:
feed(line)[source]#

Parse one omp AgentSessionEvent line into normalized events.

Parameters:
Return type:

list[StreamEvent]

class fractal.impl.omp.OmpAgent(node, command=None, provider=None)[source]#

Bases: Agent

Oh My Pi CLI backend (cost-reporting, agent-minted sessions).

Parameters:

  • self (Agent)

  • node (Node)

  • command (str | None)

  • provider (str | None)

Initialize Agent.

Parameters:

  • node (Node) – The node this agent runs for (config dirs, worktree, ledger).

  • command (str | None) – Full agent command (e.g. 'claude --some-flag'); defaults to the node’s configured agent. Extra words are spliced into every invocation.

  • provider (str | None) – Provider-route override; None means native (Node.agent threads the node’s effective configured route).

  • self (Agent)

name: str = 'omp'#
config_file: str = 'config.yml'#
can_fork: bool = True#
mints_session: bool = True#
needs_pricing: bool = False#
cost_scope: str = 'call'#
enforces_budget: bool = False#
providers: tuple[str, ...] = ()#

fractal.impl.opencode module#

opencode CLI agent implementation.

class fractal.impl.opencode.OpencodeParser(*, model=None)[source]#

Bases: StreamParser

Parser for opencode run --format json output.

Parameters:

Initialize OpencodeParser.

Bind the configured-model fallback and start the wall clock.

Parameters:
feed(line)[source]#

Parse one opencode JSON event line into normalized events.

Parameters:
Return type:

list[StreamEvent]

class fractal.impl.opencode.OpencodeAgent(node, command=None, provider=None)[source]#

Bases: Agent

opencode CLI backend (cost-reporting, agent-minted sessions).

Parameters:

  • self (Agent)

  • node (Node)

  • command (str | None)

  • provider (str | None)

Initialize Agent.

Parameters:

  • node (Node) – The node this agent runs for (config dirs, worktree, ledger).

  • command (str | None) – Full agent command (e.g. 'claude --some-flag'); defaults to the node’s configured agent. Extra words are spliced into every invocation.

  • provider (str | None) – Provider-route override; None means native (Node.agent threads the node’s effective configured route).

  • self (Agent)

name: str = 'opencode'#
config_file: str = 'opencode.json'#
can_fork: bool = True#
mints_session: bool = True#
needs_pricing: bool = False#
cost_scope: str = 'call'#
enforces_budget: bool = False#
results_per_step: bool = True#
providers: tuple[str, ...] = ()#

Module contents#

Implementations for fractal.

On this page
Show Source