#Sessions

Agent owns configuration and provider state. Session binds that agent to one workspace and one conversation lifecycle.

This is where UI integration usually begins: subscribe to AgentEvent values from the session and map them to progress, tool calls, confirmations, and results.

import { Agent } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');
const session = agent.session('/repo', {
  model: 'provider/model-id',
  planningMode: 'enabled',
  goalTracking: true,
  autoDelegation: { enabled: true, maxTasks: 4 },
  autoParallel: false,
});

#Rust Construction

Rust session construction is async-first because default memory, file-backed stores, queues, trajectory recording, and MCP discovery can require I/O:

let session = agent
    .session_builder("/repo")
    .options(options)
    .build()
    .await?;

session_async, resume_session_async, session_for_agent_async, and session_for_worker_async are direct async alternatives. Node and Python keep their existing factory names and delegate to this async construction kernel inside the native binding.

The synchronous Rust Agent::session factory is a strict compatibility path. It accepts explicitly pre-initialized resources and never starts or blocks an async runtime. A default/file memory store, file session store, queue, trajectory recorder, or any host MCP manager supplied in SessionOptions returns CodeError::AsyncSessionBuildRequired; session-option MCP capability discovery is always async. Use the builder instead of catching that error and silently changing backends. The sync path can only inherit agent-global MCP tools already cached during agent initialization.

planningMode is explicit: use 'auto' for the default structured pre-analysis path, 'enabled' to force planning, and 'disabled' to turn planning off for latency-sensitive calls. Node.js and Python also accept a boolean planning option, and Rust has with_planning(bool): omitted means 'auto', true means 'enabled', and false means 'disabled'. When both are set, planningMode wins. Go exposes only PlanningMode.

Planning emits run-scoped state. Hosts can render that state as a TaskList and update each item as the runtime records progress instead of trying to infer progress from text tokens.

#Single-Flight Operations

A session admits only one transcript-affecting operation at a time. send, stream, their attachment variants, slash commands, and resumeRun share the same fail-fast admission gate. An overlapping call returns CodeError::SessionBusy before reading history or dispatching a command; it is not queued behind the active operation.

Wait for the active result, consume the stream to completion, or cancel it before starting another conversation operation. The runtime retains a stream's lease until its producer has actually stopped, even if the public stream handle is dropped or aborted. Direct host tool helpers do not mutate the transcript and do not claim this lease.

Node and Python stream iterators wait for that lifecycle cleanup at the terminal boundary. Once a fully consumed iterator reports completion, an immediate next conversation operation will not inherit a stale busy state from that stream.

#How a Run Advances

Each session has one fact log at <workspace>/.a3s/effect-log/<thread>.jsonl. A send or stream appends the prompt as a user.message fact and folds the log to choose each next step: call the model, run a tool, compact, park, or finish. A run takes at most 32 steps before it stops with a StepLimit error. See Fact-log control for the full model.

What this means for a host:

• Confirmations park. A tool call that needs approval emits confirmation_required and waits. Answer it with confirmToolUse (Rust and Python confirm_tool_use, Go ConfirmToolUse). The answer is recorded as a fact, so it also works after a restart. The log has no timeout of its own.
• Questions park. An ask_user call emits user_question and waits for the answer.
• Tool work is bounded per turn. The stock tool budget ends a turn after 8 successful tool results. The tool-round cap (maxToolRounds, Python max_tool_rounds, Go MaxToolRounds) makes the next model call see no tools, so it must answer in text.
• Steer is a fact. An applied steer is another user.message, so it also resets both counters.

#Send

let result = session
    .send(
        "Review this repository and list release blockers",
        None,
    )
    .await?;

println!("{}", result.text);
println!("{}", result.usage.total_tokens);
println!("{:?}", result.verification_summary().status);

const result = await session.send(
  'Review this repository and list release blockers',
);

console.log(result.text);
console.log(result.totalTokens);
console.log(result.verificationStatus);

result = session.send(
    "Review this repository and list release blockers",
)

print(result.text)
print(result.total_tokens)
print(result.verification_status)

result, err := session.Run(
	ctx,
	"Review this repository and list release blockers",
)
if err != nil {
	return err
}

fmt.Println(result.Text)
fmt.Println(result.Usage.TotalTokens)
fmt.Println(result.VerificationSummary.Status)

#Stream

use a3s_code_core::{AgentEvent, CodeError};

let (mut events, lifecycle) = session
    .stream("Run the focused tests and explain failures", None)
    .await?;

while let Some(event) = events.recv().await {
    match event {
        AgentEvent::TextDelta { text } => print!("{text}"),
        AgentEvent::ToolStart { name, .. } => println!("\ntool: {name}"),
        AgentEvent::End { .. } => break,
        AgentEvent::Error { message } => return Err(CodeError::Llm(message)),
        _ => {}
    }
}
lifecycle
    .await
    .map_err(|error| CodeError::Internal(error.into()))?;

const stream = await session.stream(
  'Run the focused tests and explain failures',
);

while (true) {
  const { value: event, done } = await stream.next();
  if (done) break;
  if (!event) continue;

  if (event.text) process.stdout.write(event.text);
  if (event.toolName) console.log('tool:', event.toolName);
}

for event in session.stream("Run the focused tests and explain failures"):
    if event.type == "text_delta" and event.text:
        print(event.text, end="", flush=True)
    elif event.type == "tool_start":
        print(f"\ntool: {event.tool_name or 'unknown'}")
    elif event.type == "error":
        raise RuntimeError(event.error or "stream error")

stream, err := session.Stream(
	ctx,
	"Run the focused tests and explain failures",
	nil,
)
if err != nil {
	return err
}

for event := range stream.Events {
	if event.Type != code.EventTextDelta {
		continue
	}
	var payload struct {
		Text string `json:"text"`
	}
	if err := event.DecodePayload(&payload); err != nil {
		return err
	}
	fmt.Print(payload.Text)
}
if err := <-stream.Done; err != nil {
	return err
}

Every SDK event is an EventEnvelopeV1 projection with version === 1, an open type string, complete payload, and optional metadata. Convenience fields such as text and toolName are derived from that envelope. Keep a default branch and retain the payload for future event types.

#Steer and Interrupt an Active Run

Run control changes an operation already in progress; it does not start a second transcript turn. steer queues a newer user direction; at the next provider call it is appended to the fact log as a user.message and sent with that request. interrupt cancels current provider and Tool work cooperatively, then lets supervised cleanup settle before the Run becomes cancelled.

use a3s_code_core::{InterruptRequest, SteerRequest};

let state = session.run_control_snapshot().await;
let receipt = session
    .steer(SteerRequest::new("Prioritize the failing test"))
    .await?;
println!("{:?}", receipt.state);

session
    .interrupt(InterruptRequest::new().with_reason("User stopped the run"))
    .await?;

const state = await session.runControlSnapshot();
const receipt = await session.steer('Prioritize the failing test', {
  runId: state?.runId,
  expectedTurnId: state?.turnId,
  expectedTurnRevision: state?.turnRevision,
});
console.log(receipt.state);

await session.interrupt({ reason: 'User stopped the run' });

state = await session.run_control_snapshot_async()
options = {}
if state:
    options["run_id"] = state["run_id"]
    if state.get("turn_id") is not None:
        options["expected_turn_id"] = state["turn_id"]
    options["expected_turn_revision"] = state["turn_revision"]
receipt = await session.steer_async(
    "Prioritize the failing test",
    options,
)
print(receipt["state"])

await session.interrupt_async({"reason": "User stopped the run"})

state, err := session.RunControlSnapshot(ctx)
if err != nil {
	return err
}
options := &code.SteerOptions{}
if state != nil {
	options.RunID = &state.RunID
	options.ExpectedTurnID = state.TurnID
	options.ExpectedTurnRevision = &state.TurnRevision
}
receipt, err := session.Steer(ctx, "Prioritize the failing test", options)
if err != nil {
	return err
}
fmt.Println(receipt.State)
_, err = session.Interrupt(ctx, &code.InterruptOptions{})

Requests are idempotent when a caller reuses the same request ID with the same payload. Optional Run and expected-Turn fields make stale UI actions fail closed. Accepted control is not yet applied control: observe run_control_applied or inspect persisted Run events when the distinction matters. Neither operation changes the selected model, permissions, sandbox, budgets, or approval policy.

#Side Questions

The SDK does not expose a dedicated ephemeral-question helper. Passing explicit history to send or stream leaves session.history() unchanged, but the prompt and answer are still appended to the session's fact log, so later turns in the same session can see them. Explicit history only seeds that log when it is still empty. For a question that must not affect the session, ask it from a separate session.

const snapshot = session.history();
const answer = await session.send(
  'What files has this session already inspected?',
  snapshot,
);

console.log(answer.text);
console.log(session.history().length === snapshot.length);

#Resume

use a3s_code_core::{Agent, SessionOptions};

let session = agent
    .resume_session_async(
        "release-review",
        SessionOptions::new().with_file_session_store("./.a3s/sessions"),
    )
    .await?;

import { Agent, FileSessionStore } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');
const session = agent.resumeSession('release-review', {
  sessionStore: new FileSessionStore('./.a3s/sessions'),
});

from a3s_code import FileSessionStore, SessionOptions

opts = SessionOptions()
opts.session_store = FileSessionStore("./.a3s/sessions")
session = agent.resume_session("release-review", opts)

options := &code.SessionOptions{
	FileSessionStoreDir: ".a3s/sessions",
}
session, err := agent.ResumeSession(ctx, "release-review", options)

Set autoSave: true or call await session.save() when using a session store. This resumes a saved session snapshot; interrupted run checkpoints use session.resumeRun(runId) instead. Go uses Save, ResumeSession, and ResumeRun(ctx, runID) for the same two persistence levels. See Persistence.

With a checkpoint for that run, resumeRun rebuilds the session's fact log from the checkpoint transcript. Prior model turns are recorded as facts and are not sent to the model again. Without a checkpoint, it folds the existing log. Either way the fold chooses the next step: a tool call whose result is missing runs once, and a quiescent log resumes with zero steps. With neither a checkpoint nor a log, it fails and names the missing session store or checkpoint.

#Lifecycle and Close

session.close() is a full graceful stop. The first call flips the session into the closed state — further send/stream calls fast-fail with CodeError::SessionClosed instead of starting a new run — then cancels the active run, every in-flight delegated subagent task, and all pending human-in-the-loop confirmations. Subsequent calls are no-ops and never panic. Check the closed state with session.isClosed() (Node) / session.is_closed() (Python) / session.IsClosed(ctx) (Go).

session.close();
if (session.isClosed()) {
  // send/stream now reject with CodeError::SessionClosed
}

session.close()
if session.is_closed():
    # send/stream now reject with CodeError::SessionClosed
    pass

if err := session.Close(ctx); err != nil {
	return err
}
closed, err := session.IsClosed(ctx)

#Cancellation token

Every run derives its per-operation cancellation token from a single session-level parent via child_token(), so close() cascades to all in-flight work in one shot. Embedders that need the raw token — for example to wire it into a host-side select! or to abort the session without the run-store and hook side effects of close() — can clone it through AgentSession::session_cancel_token().

#Agent-side registry

The owning Agent tracks its live sessions by Weak reference (pruned lazily) so a control plane can drive lifecycle without holding a session handle:

• Agent::list_sessions() returns the live session IDs (sorted, stable).
• Agent::close_session(id) closes one session by ID — the same cleanup as AgentSession::close(), invoked out-of-band.
• Agent::close() closes every live session and tears down agent-owned background resources (it also disconnects the global MCP connections). After it returns, new session / resumeSession calls fail fast with CodeError::SessionClosed.
• Agent::is_closed() reports whether the agent itself has been closed.

const ids = await agent.listSessions();
await agent.closeSession(ids[0]);
await agent.close(); // closes all remaining sessions + global MCP
console.log(agent.isClosed());

ids = agent.list_sessions()
agent.close_session(ids[0])
agent.close()  # closes all remaining sessions + global MCP
print(agent.is_closed())

ids, err := agent.ListSessions(ctx)
if err == nil && len(ids) > 0 {
	_, err = agent.CloseSession(ctx, ids[0])
}
err = agent.Close(ctx)

#Host Identity Labels

SessionOptions carries four opaque identity fields that the host can attach at session creation. The framework only transports them — it never interprets or enforces them. They are propagated into SessionData, hooks, and traces, and restored on resume, so the host can drive multi-tenant aggregation, billing, and distributed tracing:

const session = agent.session('/repo', {
  tenantId: 'tenant-example',
  principal: 'principal-example',
  agentTemplateId: 'agent-template-example',
  correlationId: 'trace-example',
});

opts = SessionOptions()
opts.tenant_id = 'tenant-example'
opts.principal = 'principal-example'
opts.agent_template_id = 'agent-template-example'
opts.correlation_id = 'trace-example'
session = agent.session('/repo', opts)
print(session.tenant_id, session.principal)

session, err := agent.Session(ctx, "/repo", &code.SessionOptions{
	TenantID:        "tenant-example",
	Principal:       "principal-example",
	AgentTemplateID: "agent-template-example",
	CorrelationID:   "trace-example",
})

#Run Replay

Each send() or stream() creates a run record. Applications can inspect those records for UI state, audit, replay, cancellation, and tests:

const runs = await session.runs();
const latest = runs.at(-1);

if (latest) {
  console.log(await session.runSnapshot(latest.id));
  console.log(await session.runEvents(latest.id));
}

runs, err := session.Runs(ctx)
if err == nil && len(runs) > 0 {
	latest := runs[len(runs)-1]
	snapshot, snapshotErr := session.RunSnapshot(ctx, latest.ID)
	events, eventsErr := session.RunEvents(ctx, latest.ID)
	_, _, _ = snapshot, snapshotErr, eventsErr
	_ = events
}

currentRun() is for the operation that is current at the time of the call. During an active send() or stream(), pass its id to cancelRun(id) to request cancellation. When idle, currentRun() may return null or a retained run snapshot depending on the preceding control flow, so use runs() for completed history and inspect status before treating a snapshot as cancellable:

const terminal = new Set(['completed', 'failed', 'cancelled']);
const current = await session.currentRun();
if (current?.id && !terminal.has(current.status)) {
  await session.cancelRun(current.id);
}

#Agent Definitions

sessionForAgent() applies a named agent definition. It looks the name up in the built-in agents and in the agentDirs you pass; it does not scan .a3s/agents on its own.

const session = agent.sessionForAgent('/repo', 'explore', ['./agents'], {
  planningMode: 'auto',
});

session, err := agent.SessionForAgent(
	ctx,
	"/repo",
	"explore",
	[]string{"./agents"},
	&code.SessionOptions{PlanningMode: code.PlanningAuto},
)

For disposable value-defined workers, use SessionForWorker; both methods return the regular Go Session API.