Alineo

import {
  Alineo,
  textOnly,
  type AgentSpec,
  type AgentStream,
  type AgentEvent,
  type SetupStep,
} from "alineo";
import { SQLiteAdapter } from "@alineo-labs/sqlite";

const adapter = new SQLiteAdapter("./.alineo/ledger.db");

alineo has no storage-adapter dependency of its own — every method below that takes opts.adapter requires it. Use SQLiteAdapter from @alineo-labs/sqlite for local dev or PostgresAdapter from @alineo-labs/postgres for production.

Alineo

A live AI coding agent running inside an OpenSandbox container. Wraps a Pi CLI process (pi --mode rpc --approve) in an HTTP bridge so the host can send prompts and receive streamed responses.

Alineo.start()

static async start(
  spec: AgentSpec | Record<string, unknown>,
  opts: {
    adapter: IStorageAdapter;
    rebuild?: boolean;
    spawnDepth?: number;
    maxAgents?: number;
    runId?: string;
    memory?: Memory;
    config?: AlineoAgentConfig;
    onEgressRequest?: EgressRequestHandler;
  },
): Promise<Alineo>

Validate spec, spin up a node:22 sandbox, install the Pi CLI and any setup steps, and return a ready Alineo. spec is an already-parsed object, not a file path — read one from disk yourself first (await Bun.file(path).json()), fetch it over HTTP, or build it programmatically. It's validated internally regardless (via validateAgentSpec()), so a raw JSON.parse()'d object works fine.

On first start the container is snapshotted after install — the setup snapshot. Subsequent calls restore from it, skipping the install and starting in ~3s instead of ~90s. See Setup snapshots.

config replaces alineo.config.json discovery, for embedded callers and tests that must not depend on the working directory. runId and memory tag the agent's ledger records and wire up @alineo-labs/memory; both are optional.

const spec = await Bun.file("./agents/my-agent.json").json();
const agent = await Alineo.start(spec, { adapter });

// Force a full reinstall, ignoring the cached snapshot:
const agent2 = await Alineo.start(spec, { adapter, rebuild: true });

spawnDepth/maxAgents override the spec's own fields — see Spawning child agents below.

onEgressRequest is required when the spec has any env credential binding with approval: "hold" — it decides each first outbound request to a held host. See Permission gate — holding network egress.

Alineo.resume()

static async resume(
  sandboxId: string,
  opts: {
    adapter: IStorageAdapter;
    spec?: AgentSpec | Record<string, unknown>;
    specPath?: string;
    runId?: string;
    memory?: Memory;
    config?: AlineoAgentConfig;
  },
): Promise<Alineo>

Reconnect to an existing sandbox after the host process has exited. Only the bridge process is restarted — Pi and the workspace are untouched. Pi resumes the most recent session via --continue.

If the container is gone — deleted, or left Exited by a machine reboot — resume() restores the agent instead of failing: a fresh container from the spec's setup snapshot, with the agent's latest turn checkpoint extracted into it. Without a checkpoint, the fresh agent starts a blank session. Check agent.sandboxId afterwards: it is the new container's.

Unlike load(), resume() still accepts a bare path (specPath) — or, if you already have the spec object in memory, pass it directly via spec and skip the read. If neither is set, the ledger is queried for the sandbox's name and the spec is read from ./agents/<name>.json.

// Save the sandbox ID from a previous run...
const agent = await Alineo.resume(savedSandboxId, { adapter });

// Or pass the spec path explicitly:
const agent2 = await Alineo.resume(savedSandboxId, { adapter, specPath: "./agents/my-agent.json" });

// Or pass an already-parsed spec object:
const agent3 = await Alineo.resume(savedSandboxId, { adapter, spec });

Alineo.reattach()

static async reattach(
  sandboxId: string,
  opts: {
    adapter: IStorageAdapter;
    spec?: AgentSpec | Record<string, unknown>;
    specPath?: string;
    runId?: string;
    memory?: Memory;
    config?: AlineoAgentConfig;
    skipReadyCheck?: boolean;
  },
): Promise<Alineo>

Reconnect to an existing sandbox without restarting its bridge. Where resume() kills and restarts the bridge process — dropping any turn in progress — reattach() only looks up the sandbox's proxy URL and checks that the bridge answers. The returned Alineo is fully functional: .prompt(), .bash(), and .steer() continue the same Pi process and conversation.

Use it when your own process restarted but the sandbox didn't — a control plane coming back up after a crash (this is how alineod recovers a swarm), or a CLI steering an agent that is busy mid-turn (alineo steer).

It fails fast (within 5 s) if the bridge doesn't answer — a container can be running while the bridge inside it has died. Fall back to resume() in that case. spec/specPath resolve the same way as for resume(), and a container that is gone is restored the same way too. skipReadyCheck reconnects to a paused sandbox, whose frozen bridge can't answer the probe.

let agent: Alineo;
try {
  agent = await Alineo.reattach(savedSandboxId, { adapter });
} catch {
  agent = await Alineo.resume(savedSandboxId, { adapter }); // bridge really is gone
}

Alineo.attach()

static async attach(
  sandboxId: string,
  opts: {
    adapter: IStorageAdapter;
    name: string;
    resources?: { cpu: string; memory: string; gpu?: string };
    memory?: Memory;
    config?: AlineoAgentConfig;
  },
): Promise<Alineo>

Connect to an already-running sandbox without touching its Pi bridge — unlike resume(), which kills and restarts the bridge process. Use this when you only need .spawn()/.sandbox, not .prompt()/.bash() (the returned Alineo has no bridge, so those throw).

The main caller is alineo spawn: it runs as a fresh CLI process started BY the very Pi bash-tool call it's attaching to (a session forking a child from inside its own turn) — going through resume() there would kill the bridge currently running the call itself.

const self = await Alineo.attach(process.env.ALINEO_SANDBOX_ID!, {
  adapter,
  name: "my-session",
});
const child = await self.spawn("./agents/worker.json");

Spawning child agents

agent.spawn()

spawn(childSpecPath: string, opts?: { spawnDepth?: number; maxAgents?: number }): Promise<Alineo>

Fork this agent's own live sandbox — filesystem, installed packages, checked-out state, everything currently on disk — into a brand-new independent sandbox running its own Pi bridge. Unlike Alineo.start() (always starts from a spec's own snapshot) or agent.branchSession()/agent.duplicateSession() (harness conversation branching — same container, same bridge, new session branch), this is sandbox-level forking: the child sees exactly what this agent's sandbox sees right now, including uncommitted work. No install/setup steps run — the child inherits whatever is already installed on this agent's sandbox.

const child = await agent.spawn("./agents/worker.json", { spawnDepth: 2, maxAgents: 5 });
try {
  for await (const chunk of textOnly(child.prompt("Handle the auth module"))) {
    process.stdout.write(chunk);
  }
} finally {
  await child.close();
}

Refuses immediately unless this agent's own spawn-depth budget (spawnDepth in the spec, or opts.spawnDepth to override) is a positive integer — 0 means no budget left, undefined means spawning was never enabled. Each spawn force-decrements the budget (current - 1) into the child's env, regardless of what the child's own spec says.

maxAgents (spec field or opts.maxAgents) is a separate, optional ceiling on total descendants for this lineage, independent of nesting depth. Unset means uncapped for this dimension — only spawnDepth gates whether spawning is allowed at all. Not coordinated across sibling branches spawned in parallel; it's a per-lineage counter.

Streaming

agent.prompt()

prompt(
  message: string,
  opts?: {
    streamingBehavior?: "steer" | "followUp";
    inactivityTimeoutMs?: number;
    onPermission?: (req: PermissionRequest) => PermissionDecision | Promise<PermissionDecision>;
  },
): AgentStream

Send a message to Pi and stream the response. Pi maintains its own conversation context across calls within a session.

onPermission auto-resolves each permission_request on the stream with the handler's decision, instead of calling resolvePermission() by hand — the events still flow through the stream. Only meaningful when the spec sets permissions. See Permission gate.

// Text only (most common):
for await (const chunk of textOnly(agent.prompt("Explain this repo"))) {
  process.stdout.write(chunk);
}

// Raw stream with tool events:
for await (const ev of agent.prompt("Run /workspace/script.py")) {
  if (ev.type === "text") process.stdout.write(ev.text);
  if (ev.type === "tool_start") console.log(`[tool] ${ev.toolName}`);
}

agent.bash()

bash(command: string): AgentStream

Run a shell command inside Pi's working context. Returns the same AgentStream type as prompt(), but not incrementally streamed — Pi returns bash output synchronously, so the full output arrives as a single text event once the command completes.

for await (const chunk of textOnly(agent.bash("ls -la /workspace"))) {
  process.stdout.write(chunk);
}

Mid-flight control

agent.steer()

async steer(message: string): Promise<void>

Redirect Pi's current response mid-flight. Pi acknowledges the instruction and adjusts its output. Best called after a short delay into a prompt() stream.

The message is delivered after the tool calls in the current turn finish and before Pi's next model call — it does not interrupt a tool call that is already running. Use abort() to stop immediately.

const stream = textOnly(agent.prompt("Write a very long essay..."));
setTimeout(() => agent.steer("Stop — give me 3 bullet points instead."), 1500);
for await (const chunk of stream) process.stdout.write(chunk);

agent.followUp()

async followUp(message: string): Promise<void>

Queue a message for Pi to process after it finishes the current task. Pi receives the message as a new turn in the same session once the active response completes.

agent.abort()

async abort(): Promise<void>

Cancel Pi's current operation. The in-flight prompt() stream ends with whatever was generated before the abort. Any pending permission requests are auto-rejected (steer() leaves them open).

Permissions & approvals

Active only when AgentSpec.permissions is set to something other than "auto". See Permission gate for the full model.

agent.resolvePermission()

async resolvePermission(requestId: string, decision: PermissionDecision): Promise<void>

Answer a permission_request event. PermissionDecision is { kind: "once" }, { kind: "always" }, or { kind: "reject"; feedback?: string }. always / reject also clear every other still-pending request for the same tool.

for await (const ev of agent.prompt("Refactor auth")) {
  if (ev.type === "permission_request") {
    await agent.resolvePermission(ev.requestId, { kind: "once" });
  }
}

agent.listPendingPermissions()

async listPendingPermissions(): Promise<PendingPermission[]>

Tool calls currently paused awaiting a decision — each { requestId, tool, target, title, since }. Useful after reconnecting to a session to discover approvals still outstanding.

agent.pendingEgressRequests()

pendingEgressRequests(): EgressRequest[]

Outbound network requests to approval: "hold" hosts currently waiting for an onEgressRequest decision — each { host, since }. Empty unless the spec has held credential bindings.

agent.egressGate

readonly egressGate?: EgressApprovalGate

The host-side listener that holds egress to approval: "hold" hosts until onEgressRequest approves. Present only when the spec has held bindings; started on load(), stopped on close().

Environment and debugging

agent.setEnv()

async setEnv(vars: Record<string, string>): Promise<void>

Merge vars into the sandbox environment. Writes to /etc/alineo-env inside the container and restarts Pi so it picks up the new values. Awaits Pi's readiness before returning.

await agent.setEnv({ DATABASE_URL: "postgres://...", DEBUG: "1" });

agent.getLogs()

async getLogs(): Promise<string>

Retrieve the last 200 bridge log entries as a plain text string. Useful for debugging Pi startup or RPC issues.

agent.close()

async close(): Promise<void>

Delete the sandbox container and release all resources. Always call in a finally block.

try {
  // ...
} finally {
  await agent.close();
}

Properties

agent.sandbox

readonly sandbox: SandboxHandle

Direct access to the underlying SandboxHandle — the full Core SDK Sandbox API, bypassing Pi. Use this to read or write files, run shell commands, or inspect the container independently of Pi.

await agent.sandbox.writeFile("/workspace/input.csv", data);
const { stdout } = await agent.sandbox.exec("wc -l /workspace/input.csv");
const output = await agent.sandbox.readFile("/workspace/output.txt");

agent.sandboxId

readonly sandboxId: string

OpenSandbox container ID for this agent's sandbox.

agent.name

readonly name: string

Name from the agent spec.

agent.fromSnapshot

readonly fromSnapshot: boolean

true when this agent was restored from a cached snapshot (fast path). false on first load or after { rebuild: true }.

AgentStream

type AgentStream = AsyncIterable<AgentEvent>;

Returned by agent.prompt() and agent.bash(). Iterate with for await to receive events as Pi generates them:

for await (const ev of agent.prompt("...")) {
  // ev is AgentEvent
}

Use textOnly(stream) to filter to just the text chunks.

AgentEvent

type AgentEvent =
  | { type: "text"; text: string }
  | { type: "tool_start"; toolCallId: string; toolName: string; args: unknown }
  | { type: "tool_update"; toolCallId: string; toolName: string; partialResult: unknown }
  | { type: "tool_end"; toolCallId: string; toolName: string; result: unknown; isError: boolean }
  | { type: "extension_ui"; method: string; params: unknown; isDialog: boolean; requestId?: string }
  | { type: "permission_request"; requestId: string; tool: string; target: string; title: string }
  | { type: "permission_resolved"; requestId: string; decision: PermissionDecision }
  | {
      type: "auto_retry_start";
      attempt: number;
      maxAttempts: number;
      delayMs: number;
      errorMessage: string;
    }
  | { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string }
  | { type: "agent_start" }
  | { type: "agent_end"; messages: unknown[] }
  | { type: "turn_start"; turnIndex: number; timestamp: number }
  | { type: "turn_end"; turnIndex: number; message: unknown; toolResults: unknown[] }
  | { type: "message_start"; message: unknown }
  | { type: "message_update"; message: unknown; delta: unknown }
  | { type: "message_end"; message: unknown }
  | { type: "queue_update"; steering: string[]; followUp: string[] }
  | { type: "compaction_start"; reason: "manual" | "threshold" | "overflow" }
  | {
      type: "compaction_end";
      reason: "manual" | "threshold" | "overflow";
      result: {
        summary: string;
        firstKeptEntryId: string;
        tokensBefore: number;
        estimatedTokensAfter: number;
        details: unknown;
      } | null;
      aborted: boolean;
      willRetry: boolean;
    }
  | { type: "extension_error"; extensionPath: string; event: string; error: string };

See Streaming & tool events for usage examples.

textOnly()

function textOnly(stream: AgentStream): AsyncIterable<string>;

Filter an AgentStream to just text events, yielding the text string of each. Equivalent to the old PromptStream behavior.

import { textOnly } from "alineo";

for await (const chunk of textOnly(agent.prompt("Hello"))) {
  process.stdout.write(chunk);
}

alineo.config.json

alineo reads alineo.config.json from the current working directory to configure the client. All fields are optional.

{
  "serverUrl": "http://127.0.0.1:8080",
  "apiKey": "",
  "adapterPath": "./.alineo/ledger.db",
  "useServerProxy": true,
  "agentsDir": "./agents",
  "defaults": {
    "resources": { "cpu": "1000m", "memory": "1Gi" }
  }
}

alineo init writes this file automatically.