`yggdrasil.loki.fleet`¶

fleet ¶

Loki fleet — delegate tasks to background process agents and monitor them.

Loki's autonomy multiplier: instead of doing every task itself, the agent can fan a set of independent tasks out to separate Loki processes — each a ygg loki do <task> --json subprocess running its own confined act loop — and watch them run to completion. A :class:Fleet spawns the agents (output streamed to temp files so a chatty agent never deadlocks on a full pipe), polls their status, and drives them to done; the CLI renders the live dashboard and the delegate skill exposes it programmatically.

Pure orchestration: :class:Fleet spawns whatever command it's given, so it's unit tested with a trivial python -c stand-in agent — no engine required. :meth:Fleet.do_command builds the real ygg loki do invocation.

FleetKPIs `dataclass` ¶

FleetKPIs(
    total: int = 0,
    running: int = 0,
    done: int = 0,
    failed: int = 0,
    queued: int = 0,
    validated: int = 0,
    mesh: int = 0,
    steps: int = 0,
    tokens: int = 0,
    cost: float = 0.0,
    elapsed: float = 0.0,
)

A fleet's live rollup as typed fields (not a loose dict).

AgentSummary `dataclass` ¶

AgentSummary(
    id: int,
    task: str,
    status: str,
    elapsed: float = 0.0,
    steps: int = 0,
    answer: str = "",
    files_changed: list[str] = list(),
    error: str = "",
)

Bases: DictResult

One delegated agent's outcome (mapping-compatible; to_dict for JSON).

AgentHandle ¶

AgentHandle(
    agent_id: int,
    task: str,
    cmd: list[str],
    proc: Popen,
    out_path: str,
    err_path: str,
    engine: Optional[str] = None,
)

One spawned process agent — its task, process, and (when done) result.

validated `property` ¶

validated: 'Optional[bool]'

Whether the agent ran a passing smoke checkpoint — True (all smokes green), False (one failed), or None (no smoke run). Lets the dashboard show which delegated work was actually self-validated.

Fleet ¶

Fleet(
    *,
    python: Optional[str] = None,
    max_parallel: Optional[int] = None,
    max_local: int = 1,
    cost_cap: Optional[float] = None,
    per_agent_budget: Optional[float] = None,
    mesh_dir: Optional[str] = None
)

Spawn and monitor a set of background process agents.

do_command ¶

do_command(
    task: str,
    *,
    root: str = ".",
    engine: Optional[str] = None,
    tier: Optional[str] = None,
    max_steps: int = 8,
    read_only: bool = False,
    allow_shell: bool = False,
    allow_web: bool = False
) -> list[str]

The ygg loki do argv for task — one isolated autonomous agent.

spawn ¶

spawn(
    task: str,
    *,
    cmd: Optional[list[str]] = None,
    env: Optional[dict[str, str]] = None,
    **kw: Any
) -> AgentHandle

Launch a process agent for task (or an explicit cmd) → its handle.

Output goes to temp files (not pipes), so a verbose agent can't deadlock the parent on a full OS pipe buffer while others are still running.

spawn_all ¶

spawn_all(tasks: list[str], **kw: Any) -> list[AgentHandle]

Queue tasks and launch up to max_parallel of them now.

Returns the handles started immediately; the rest launch from :meth:poll as running slots free.

mesh_keys ¶

mesh_keys() -> int

How many results peers have published to the shared mesh (0 if none).

poll ¶

poll() -> list[AgentHandle]

Refresh statuses: finalize exited agents, then fill freed slots.

monitor ¶

monitor(
    on_update: Optional[Callable[[list[AgentHandle]], None]] = None,
    *,
    interval: float = 0.2,
    timeout: Optional[float] = None,
    on_cap: Optional[Callable[[dict[str, Any]], Optional[float]]] = None
) -> list[AgentHandle]

Drive the fleet to completion, calling on_update each tick.

Returns when every agent has finished (or timeout elapses — survivors are cancelled and marked timeout). When the aggregate cost cap is hit with work still queued, on_cap(kpis) is asked for a new cap to go further; returning None drops the queue and finishes the running ones.

kpis ¶

kpis() -> FleetKPIs

Live rollup across the fleet — counts, steps, tokens, cost, wall time.

summary ¶

summary() -> list[AgentSummary]

One typed row per agent (mapping-compatible; to_dict for JSON).

yggdrasil.loki.fleet¶