Sessions in depth

This page holds the full detail behind the Sessions guide. Read the guide first.

Identity

A session is one unit of agent work with a recorded lifecycle. Its ID is the conversation ID, so every conversation read in the SDK also finds the session. A session is addressed by its stream and its ID. The stream is the isolation boundary: a session and all of its child sessions live in one stream, and no link, derived ID, or read crosses streams. The same ID written into two streams is two unrelated sessions.

The factory starts a session in one of three ways:

• create(label) derives the ID from the stream name, the namespace, and the label. The same three always reach the same session. Two applications on one stream share a session only when they pick the same namespace and label.
• start() gives the session a fresh ID.
• .with_id(id) (TypeScript: withId) on either builder starts the session under an ID you choose.

.id() on a builder returns the ID it will start. For an unlabeled builder without an explicit ID, each call returns a new fresh ID. derive_session_id(stream, namespace, label) (TypeScript: deriveSessionId) computes a labeled ID without a builder.

Reopening a derived ID does not reset a session that already ended. Use a new ID for a new lifecycle. A derived ID holds hash bits where a fresh ID holds time, so order sessions by start time, never by ID.

Many agents can write in one session. Each handle writes as one agent. as_agent(id) (TypeScript: asAgent) returns a copy of a handle that writes as another agent.

Bootstrap a stream

sessions.bootstrap(partitions, retention) creates the agent topics in the stream. agent.sessions holds every session's records, so it needs an explicit retention. TopicRetention::expire_after(age) keeps records for that long with the server's default size bound. TopicRetention::new(expiry, max_size) sets both. A policy that never expires and has no size bound is refused with an invalid error.

Bootstrap uses a stream that already exists as it is, so an account with permission on the stream's topics can bootstrap a stream that was created for it. With the SinglePartition layout, bootstrap creates every agent topic with one partition, whatever count you pass.

When the server announces the sessions capability, bootstrap also registers the stream as a session source, then waits until session reads see the registration, up to the publish timeout. Registration needs session:admin on stream:<name>. A refused registration is logged and reported as registered: false, never as an error, so provisioning can register the stream for an account without that right. Turn registration off with register_source(false) in the session configuration.

laser.bootstrap(partitions, retention) creates the agent topics without the registration step. It knows no session configuration, so it never creates declared per-agent topics and never forces one partition.

Start a session

Both create(label) and start() return a builder. Chain its settings, then call begin().

begin() writes the start record on the session's partition of agent.sessions and returns the session with a lease. Rust returns (Session, SessionLease), TypeScript returns { session, lease }, and Python returns a (session, lease) tuple. A builder without an agent fails with an invalid error.

open(id) returns a session for an existing ID as a lens. It does no I/O, takes no lease, and writes nothing until as_agent(id) names an author.

End a session

end(), fail(error), and cancel() write the terminal record. Every copy of a session handle shares one terminal latch. The first call fixes the outcome and the record ID. Calling the same verb again resends that same record, which is safe after a failed publish. A different terminal verb fails with an invalid error. end() first writes a state snapshot when the session state changed since the last one.

fail(error) takes an AgentErrorBody with a code, an optional message, a retryable flag, and an optional detail map. Rust takes the struct, TypeScript an object, and Python a dict such as {"code": 7, "message": "timed out", "retryable": False}, where the code is the wire number (7 is Internal). A Python async with block that raises records code Internal with the exception text.

Each SDK has a guard that ends the session by the outcome of your work and releases the lease:

• In Rust, session.run(lease, |session| async move { .. }) ends the session as completed when the work returns Ok, and as failed on an error or a panic. A panic is recorded with panic: true in the error detail and then raised again.
• In TypeScript, session.run(lease, async (session) => { .. }) catches thrown errors and rejected promises, records the failure, and rethrows the original error. A thrown value that is not a LaserError is recorded with the panic flag. A promise you do not await stays yours, and the SDK adds no process-wide rejection listener.
• In Python, async with builder as session: begins the session, ends it when the block finishes, and fails it with the exception type and traceback when the block raises. await session.run(lease, work) does the same for a function that takes the session and may be async. Both raise the original exception.

When the terminal write also fails, the error from your work is the one you see. No guard can capture a dropped future, process death, or a Rust build with panic = "abort". Such a session shows as idle once its heartbeat stops.

Leases and liveness

A lease keeps the session listed in the client's heartbeat. Starting a session or picking up submitted work takes a lease. Opening a session as a lens or handling a record with ctx.session() does not. Drop or release the lease when the process stops working on the session. A terminal verb does not release leases held elsewhere in the process. In TypeScript, SessionLease also works with using.

While it holds at least one lease, the client publishes one heartbeat per stream on agent.heartbeats, listing the sessions it holds leases on. It beats at the heartbeat interval or at one fifth of the shortest idle timeout among its leases, whichever is shorter, and never more often than every 10 ms. One heartbeat record lists at most 2,048 sessions, and a larger set splits across records. Heartbeats stop when the last lease is released. A lease is scoped to the stream name and the stream's creation time, so a lease from a deleted stream never reports liveness for a new stream with the same name.

Idle is never written to the log. A reader derives it from the later of the last event and the last heartbeat, plus the session's idle timeout. A session whose process died shows as idle, never as failed, because nothing proves it failed. The next record makes it active again. Idle never replaces submitted, paused, or a terminal status.

Session statuses

The spelling is canceled everywhere. A status the client does not know decodes as unrecognized. Records written after the terminal record still count, and the managed index flags them after_end. A session with no start record is implicit: active, shown idle after its timeout, and never completed.

Topics

Bootstrap never creates agent.control. Create it during provisioning and give send permission on it only to operator accounts, because that permission is what makes a control request authoritative.

Every record on agent.sessions names its addressee in the agdx.to header: an agent ID, or * for every agent. Records describe themselves, so the record kind and operation live in the record, not in the topic name.

Layouts

A stream picks one layout for its agents' work:

Routing follows these rules:

• Lifecycle and state always ride the session's partition of agent.sessions, in every layout. That partition is the session lane.
• The session's partition comes from the message key, which is the session ID.
• In the per-agent partition layout, a command lands on its addressee's partition and a reply on its requester's partition. Records for an agent the map does not name stay on the session's partition. Partition IDs start at zero. Declare the map, because hashing agent names can collide.
• In the per-agent topic layout, work addressed to a declared agent moves from agent.sessions to that agent's declared topic, keyed by session. See Declare a topic per agent.
• Order is exact within one partition. Across partitions, readers order records by broker append time and then by stream, topic, partition, and offset. The producer's clock is never used for order.

A declared layout holds for every later send on that stream through the same connection, so agents and session handles route the same way.

In every layout the SDK keeps agents from misreading each other's records. Only separate topics with separate permissions keep agents from reading each other's records.

Declare a topic per agent

The per-agent topic layout maps agent IDs to topic names. Each declared agent then reads its own topic, and per-topic permissions can keep agents from reading each other's work. Everyone still sends to agent.sessions. The SDK moves the record when it is sent:

• An AGDX command, response, error, or chunk whose target is a declared agent goes to that agent's topic.
• A plain record addressed to a declared agent goes there too. That covers plain requests and plain respond replies.
• Lifecycle, state, status, events, broadcast records (* or no target), and records for an agent the map does not name stay on agent.sessions.
• On a declared topic the partition is keyed by session, so one session's records keep their order there.

bootstrap(partitions, retention) creates each declared topic with the same partition count and retention as agent.sessions. A declared topic named agent.sessions or agent.control, or an invalid topic name, is an invalid error. An agent that listens on agent.sessions with this session configuration reads its declared topic plus agent.control. A declared caller waits for replies on its own topic, which covers request, contract, request_input, and MCP tool calls. Session source registration covers every topic of the stream, so session reads still see the moved records.

import {
  Agent,
  AgentId,
  AgentTopic,
  Laser,
  SessionConfig,
  TopicRetention,
  agentMessageBody,
  routeTo
} from "@laserdata/laser-sdk"

await using root = await Laser.connectEnv()
const laser = root.withDefaultStream("support")

const config = new SessionConfig().layout({
  kind: "perAgentTopic",
  topics: new Map([
    ["planner", "planner.inbox"],
    ["worker", "worker.inbox"]
  ])
})
await laser.sessions(config).bootstrap(1, TopicRetention.expireAfter(86_400_000))

// The worker listens on agent.sessions and reads worker.inbox.
await using worker = Agent.builder()
  .id(AgentId.new("worker"))
  .listenOn(AgentTopic.Sessions)
  .respondOn(AgentTopic.Sessions)
  .sessions(config)
  .handler({ handle: (message, ctx) => ctx.respond(agentMessageBody(message)) })
  .build()
  .spawn(laser)
await worker.ready()

// The command lands on worker.inbox and the reply on planner.inbox.
const outcome = await laser
  .contract(routeTo(AgentId.new("worker")))
  .from(AgentId.new("planner"))
  .payload(new TextEncoder().encode("plan the node-7 drain"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Sessions })
  .send()
console.log(outcome.kind)

use laser_sdk::prelude::full::*;
use std::time::Duration;

struct Worker;

impl AgentHandler for Worker {
    async fn handle(&self, message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        ctx.respond(message.body().to_vec()).await
    }
}

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_env().await?.with_default_stream("support");

    let config = SessionConfig::new().layout(SessionLayout::per_agent_topic([
        (AgentId::new("planner")?, "planner.inbox"),
        (AgentId::new("worker")?, "worker.inbox"),
    ]));
    laser
        .sessions_with(config.clone())
        .bootstrap(1, TopicRetention::expire_after(Duration::from_secs(86_400)))
        .await?;

    // The worker listens on agent.sessions and reads worker.inbox.
    let mut worker = Agent::builder()
        .id("worker".parse::<AgentId>()?)
        .listen_on(AgentTopic::Sessions)
        .respond_on(AgentTopic::Sessions)
        .sessions(config)
        .handler(Worker)
        .build()
        .spawn(laser.clone());
    worker.ready().await?;

    // The command lands on worker.inbox and the reply on planner.inbox.
    let outcome = laser
        .contract(Router::to("worker".parse::<AgentId>()?))
        .from("planner".parse::<AgentId>()?)
        .payload("plan the node-7 drain")
        .inbox_route(InboxRoute::Fixed(AgentTopic::Sessions))
        .send()
        .await?;
    println!("{}", matches!(outcome, Contract::Completed(_)));
    worker.shutdown().await
}

import asyncio

import laser_sdk as ls


async def work(ctx, message):
    await ctx.respond(bytes(message.body()))


async def main():
    async with await ls.Laser.connect_env() as root:
        laser = root.with_default_stream("support")

        layout = ls.SessionLayout.PerAgentTopic({"planner": "planner.inbox", "worker": "worker.inbox"})
        sessions = laser.sessions(layout=layout)
        await sessions.bootstrap(1, ls.TopicRetention.expire_after(86_400_000))

        # The worker listens on agent.sessions and reads worker.inbox.
        async with laser.spawn_agent(
            "worker",
            ls.AgentTopic.Sessions,
            work,
            respond_on=ls.AgentTopic.Sessions,
            sessions=sessions,
        ):
            # The command lands on worker.inbox and the reply on planner.inbox.
            outcome = await laser.contract(
                None,
                b"plan the node-7 drain",
                agent="worker",
                source="planner",
                fixed_inbox=ls.AgentTopic.Sessions,
            )
            print(isinstance(outcome, ls.Contract.Completed))


asyncio.run(main())

Declare the same layout in every process that writes to the stream, because each connection routes by its own declaration. Give each declared topic the grants of its agent only. An agent that the map does not name keeps reading agent.sessions.

Choose a layout

Agent topic names are the same in every stream, and nothing session related crosses a stream. Give each application, and each environment of an application, its own stream instead of sharing one.

• Keep the shared layout when a stream has a few roles.
• Move to per-agent topics when the roles are many or the lane is busy. On a shared lane the server examines the whole lane once for every filtered role, so each delivered record costs as many examined records as there are roles. Per-agent topics remove that cost, and per-topic permissions keep the roles apart.
• Use the per-agent partition layout when each declared agent's work should land on one known partition.
• Use a single partition when everything must keep one order, for example in tests or a small app.

Partitions are how one role scales. A consumer group gives each partition to one member at a time, so instances of a role beyond the partition count sit idle. Bootstrap agent.sessions with at least as many partitions as the busiest role has instances.

Agents and consumer groups

Each agent ID reads through its own consumer group, named after the agent ID unless the agent builder names another group. Every running instance of one agent joins that group, so the instances share its work and each record reaches one of them. Two different agent IDs never share a group.

On agent.sessions and agent.control, the runtime binds the agent's group to the addressee filter agdx.to In [<agent id>, "*"] before its first read. It does so when the server resolves group policies and serves filtered reads and the filter catalog, and the client was built from a connection string. The server then sends the agent only records addressed to it or to every agent. A group already bound to another filter is refused. In any other case, such as on plain Apache Iggy or with a client you built yourself, the group stays unbound. The agent reads every record, classifies it, and commits what is not its own work without calling the handler. Delivery is the same either way. The filter only saves reading, and it is not an access boundary.

Set the session configuration an agent runs with through its builder: Agent::builder().sessions(config) in Rust, Agent.builder().sessions(config) in TypeScript, and spawn_agent(..., sessions=laser.sessions(...)) in Python. Inside a handler, ctx.session() follows that configuration.

Record model and tool calls

The SDK never calls a model. Your code calls its provider, and the session records the call.

• assemble(policy) reads the session lane, applies a context policy, and returns the selected records with a manifest. It writes nothing. text() on the result joins the records' payloads, one per line.
• model(request, assembled) records the model request and, when you pass the assembled context, its manifest. It returns a call. Its complete(response) records the answer with its usage, the answering model, the finish reason, and the duration, and its fail(error) records a failed call. When the response carries no duration, the SDK measures it from the request.
• tool(name, args) records a tool call. Its complete(result) and fail(error) record the outcome with the measured duration.
• record_model_call(request, response, assembled) (TypeScript: recordModelCall) records a call that already happened. It records the response's duration, or zero when the response carries none.

One call keeps one correlation across its request, manifest, and result. correlation() on a call returns it (Python: the correlation property). ModelRequest takes a model and a prompt body, with an optional provider and an optional operation such as text_completion, chat by default. A manifest lists at most 1,024 fragments.

Tool arguments and JSON model request bodies pass a redactor before they are published. The default replaces the values of authorization, api_key, token, password, secret, and cookie with [redacted] at any depth, matching keys without regard to case. redact(fn) returns a handle with your own redactor. In Python, a redactor that raises publishes [redacted]. default_redact (TypeScript: defaultRedact) is exported so your redactor can extend it. Redaction is a convenience, not a guarantee: a secret under another key is published as it is.

Session state

state() returns one JSON document per session, built from patches on the session lane.

• set(key, value) adds or replaces one top-level key.
• patch(ops) appends a JSON Patch delta.
• replace(document) replaces the whole document as one patch that removes the keys it drops and sets every key it holds.
• snapshot() writes the whole document. The SDK also writes a snapshot after every 64 deltas and before end().
• get() folds the retained lane from the newest snapshot. It reports complete: false when the records the document starts from are no longer retained.

A handle that has not written yet starts from the folded lane, so a lens such as ctx.session() writes against the current revision. A reader applies each patch as a whole: if one operation fails, the whole patch is rejected and the document stays unchanged. A repeated delta applies once. A snapshot replaces the document only when it was written against the current revision, so a delayed snapshot from one agent cannot overwrite a later change from another. A successful append does not prove that a patch applied. Read the state at or after your record to learn the outcome.

Session state is separate from Key-value state.

Context, memory, and linked writes

context() returns the last 50 records of the session lane, trimmed to about 4,000 estimated tokens. Change the defaults with context_turns and context_tokens in the session configuration. Python's context(last_n=, token_budget=) also takes them per call. context_with(policy) (TypeScript: contextWith) takes your own policy. Every SDK estimates tokens the same way: the byte count divided by four, rounded up.

record_retrieval(query, items) records which memory items entered the context, with their scores. record_compaction(compaction) records that a summary replaced a range of records.

memory() returns memory for this session in the agent.session namespace. Change it with memory_namespace in the session configuration. Rust and Python memory_in(namespace), TypeScript memory(namespace), and Python memory(handle) pick another namespace or handle. linked_memory() also stamps each remembered item with the record that motivated it and the agent that wrote it. See Memory.

kv(namespace) and linked_graph(name) return key-value and graph handles that link every write to this session. graph(name) returns the shared graph without a link. reference() returns the session's stream and ID, for a write that lands outside the session's stream. acting_on(source) (TypeScript: actingOn) returns a handle that stamps a source record on graph writes and on remembered items. Inside a handler, ctx.session() already acts on the handled record. In Rust, kv needs the kv feature and graph and linked_graph need the graph feature.

scope() (TypeScript: the scope property) returns the underlying context scope for a topic outside the session lane. append(envelope) writes one typed envelope on the session lane. The envelope must name this session as its conversation.

Hand work to an agent

submit(agent, input) starts a session on behalf of another agent. Name yourself with .from(id) (Python: from_), which is required. You can add .label(..), .namespace(..), .budget(..), .tag(..), and .operation(..), which is invoke_agent by default. send() writes a submitted session and a command addressed to the agent, and returns the session ID and the command's correlation.

The agent's reliable consumer marks the session active when it picks the command up, and holds a lease while the handler runs. Inside the handler, ctx.session() returns that session, writing as the handling agent and acting on the handled record.

Budgets

A budget states the tokens and cost a session may use. Set it with .budget(..) on a session builder or a submission, and it rides the session's start record. Cost is in integer micro-units of the deployment's currency.

over_budget() (TypeScript: overBudget) tells you whether the summed input and output tokens of the session's records passed the token ceiling, or their summed cost passed the cost ceiling. A deployment that indexes sessions answers from its index. On plain Apache Iggy, and for a session the index does not know yet, the SDK folds the retained lane by the same rule. A session without a budget is never over it.

On a deployment that indexes sessions, an agent with an ID checks each session before its handler runs, once per session per poll batch. When the session is over its budget, the agent fails it once with reason budget and an error naming the ceiling, then commits the work without calling the handler. Later work for that session is committed the same way. On plain Apache Iggy the runtime does not enforce budgets, because folding the lane for every record would cost too much. A workflow checks its run session at every step boundary, stops with a budget exceeded error, and ends the run session failed with reason budget. Budget reads are eventually consistent, so a budget is a cooperative limit, not a hard spending cap.

Dead letters

A dead-lettered record never fails its session by itself. It counts as an error, shows as dead_letter on the timeline, and keeps the session's ID in the dead-letter record. To fail the session instead, turn on fail_on_dead_letter in the session configuration and pass it to the agent: SessionConfig::new().fail_on_dead_letter(true) in Rust, new SessionConfig().failOnDeadLetter(true) in TypeScript, and laser.sessions(fail_on_dead_letter=True) in Python.

Operator control

control(stream, id).as_operator(op) returns a control handle that writes requests on agent.control. Only accounts with send permission on that topic can write them.

The same requests found on any other topic are never applied, and a timeline shows them as unauthorized_control. Control is cooperative. A running agent follows agent.control and records every pause and cancel request addressed to it or to every agent, including requests sent before it started. The follower reads every partition of agent.control directly, outside any consumer group, so every instance of an agent sees every request.

Inside a handler, session.pending_control() (TypeScript: pendingControl) returns the recorded requests as pause_requested and cancel_requested. The runtime never interrupts a handler, so the handler decides when to stop. cancel_requested() reads the session's control records directly, so it also answers on plain Apache Iggy. Workflows check it between steps.

signed_by(key) (TypeScript: signedBy) on the control handle signs each control record, so a verifying reader can prove which operator sent it. signed_by(key) on a session handle signs its terminal record. The managed session index verifies signed records against the stream's key registry and reports the signer as verified_actor on the event. Without a signature, the operator ID in a control record is a claim. In Rust, signing needs the sign feature.

Pause and resume

Pause means no new actions in the session until it resumes. An action already in flight finishes and is recorded. Pause is cooperative, so a client that ignores it can still append.

1. pause() writes a request that names its participants, the agents whose acknowledgments complete the pause. Set them with participants(..), or let the SDK read them from the session lane: the agents its work was sent to and the agents that picked it up.
2. Each named participant acknowledges with a session.paused record that names the exact request. An agent outside the set acknowledges when it first receives work for the paused session.
3. An agent holds work that arrives while the session is paused. It writes a session.parked record on the lane before it commits the source, and it never treats held work as handled.
4. resume() lifts the pause. Each agent acknowledges with session.resumed, handles every held record at least once before new work, and writes session.unparked after each one. A crash between the effect and that record can repeat the effect, so effects still need an idempotency key or a fenced write.
5. After a restart, a rebalance, or a reconnect, an agent rebuilds its held work from the lane. A bounded read that cannot prove the list complete reports it as incomplete instead of dropping work.

A cancel while paused ends the session as canceled, and held records are not handled. parked() lists the held records that no agent reported handled, with a complete flag. The flag is false when the read did not reach the oldest retained record of the session's partition, or when a held record can no longer be read back. The managed index counts held records as held on the session summary. No deployment advertises the pause runtime as a capability yet.

const control = laser
  .sessions()
  .control("support", session.conversation)
  .asOperator(AgentId.new("ops"))
  .participants([AgentId.new("worker")])
await control.pause()
// ... review ...
await control.resume()

const held = await laser.sessions().open(session.conversation).parked()
console.log(held.records.length, held.complete)

let control = laser
    .sessions()
    .control("support", session.conversation())
    .as_operator("ops".parse::<AgentId>()?)
    .participants(["worker".parse::<AgentId>()?]);
control.pause().await?;
// ... review ...
control.resume().await?;

let held = laser.sessions().open(session.conversation()).parked().await?;
println!("{} {}", held.records.len(), held.complete);

control = (
    laser.sessions()
    .control("support", session.conversation)
    .as_operator("ops")
    .participants(["worker"])
)
await control.pause()
# ... review ...
await control.resume()

held = await laser.sessions().open(session.conversation).parked()
print(len(held.records), held.complete)

Child sessions

Work with its own lifecycle is a child session: a new session with a parent and a root, in the same stream. A root session has neither. Start one with .parent(parent, root) on a session builder, where root is the parent itself when the parent has no parent. A child's history stays in the child. Nothing copies it into the parent, so collect child results in your application and record them on the parent.

The SDK also starts child sessions for you:

• A workflow run is a root session, and each step and compensation is a child session whose ID derives from the run ID and the step label.
• A contract runs as a child with .parent(parent, root) on the contract builder.
• The A2A and MCP bridges run calls as children with submit_in and call_tool_in.

Read a session back

Each record read back from a session carries its display type, derived from the record:

A Rust SessionTurn holds display and message, and text() returns the body as UTF-8. Python has the same display, message, and text(). TypeScript has display and message, and sessionTurnText(turn) returns the text.

checkpoint() records where the session lane ends now. It is a client bookmark that serializes, never a record on the log. turns_at(checkpoint) reads the records before it, and turns_since(checkpoint) the records after it. state_at(checkpoint, init, fold) and replay(checkpoint, init, fold) fold those records into your own state.

const state = session.state()
await state.set("tasks", ["triage"])
await state.patch([{ op: "add", path: "/tasks/-", value: "diagnose" }])
const view = await state.get()
console.log(view.revision, view.document, view.complete)

const checkpoint = await session.checkpoint()
const saved = JSON.stringify(checkpoint)
const toolCalls = (count: number, turn: SessionTurn) =>
  turn.display === "tool.call" ? count + 1 : count
const later = await session.replay(Checkpoint.fromJSON(JSON.parse(saved)), 0, toolCalls)

let state = session.state();
state.set("tasks", json!(["triage"])).await?;
let patch = serde_json::from_value(json!([
    { "op": "add", "path": "/tasks/-", "value": "diagnose" }
]))
.map_err(|error| LaserError::Codec(error.to_string()))?;
state.patch(patch).await?;
let view = state.get().await?;
println!("{} {} {}", view.revision, view.document, view.complete);

let checkpoint = session.checkpoint().await?;
let saved = serde_json::to_string(&checkpoint)
    .map_err(|error| LaserError::Codec(error.to_string()))?;
let restored: Checkpoint =
    serde_json::from_str(&saved).map_err(|error| LaserError::Codec(error.to_string()))?;
let tool_calls = |count: usize, turn: &SessionTurn| {
    if turn.display == DisplayType::ToolCall { count + 1 } else { count }
};
let later = session.replay(restored, 0, tool_calls).await?;

state = session.state()
await state.set("tasks", ["triage"])
await state.patch([{"op": "add", "path": "/tasks/-", "value": "diagnose"}])
view = await state.get()
print(view["revision"], view["document"], view["complete"])

checkpoint = await session.checkpoint()
saved = checkpoint.to_json()


def tool_calls(count, turn):
    return count + 1 if turn.display == "tool.call" else count


later = await session.replay(ls.Checkpoint.from_json(saved), 0, tool_calls)

In Rust, DisplayType is part of laser_sdk::prelude::full.

Managed session reads

A deployment can keep a session index for a registered stream. The server sets the sessions capability only when it serves the index, and a client starts with it off. The session factory reads the index of its stream:

• get(id) returns one session's summary: status, label, namespace, owning agent, parent and root, start and end times, first and last event and heartbeat times, counts of events, model calls, tool calls, and errors, input and output tokens and cost, the budget, the SDK, the derived idle, over_budget, pause_requested, and cancel_requested flags, and held, the count of records held while paused and not yet handled. session.status() reads the same summary for one handle.
• list() pages the stream's sessions, newest first. Filter by status, agent, text (a label or ID substring), root (the tree under one session), or label_prefix (TypeScript: labelPrefix). Continue with cursor, set the page size with limit, and ask for a count of every match with total. A zero limit leaves the page size to the server.
• events(id) pages one session's timeline in broker time order, without payloads. Each event names its position, display type, agent, addressee, correlation, cause, tool, usage, and a value-free summary. A signed record names its verified signer as verified_actor. fixed_frontier pins the first page's frontier for a consistent historical walk.
• state(id, history_limit) returns the folded state document with its change history. Each history row records the revision, the operation ID, the outcome (applied, rejected, stale, or duplicate), the source position, and the reason. A zero limit leaves the history size to the server.
• links(id, surface) lists the memory items, keys, graph nodes and edges, projections, and child sessions the session wrote, recalled, or touched. Pass a surface to narrow the list.
• sources(id) lists the source partitions that hold the session's records, with how far the index has folded each one.
• changes(after, limit) reads the stream's change feed after a sequence number. watch(poll_every) follows it from now and reports the changed session IDs of each poll, or a resync when it fell behind the retained feed. List the sessions again after a resync.

Index summaries never hold values. A tool call keeps its tool name, argument key names, size, and content hash. A model record keeps model, provider, usage, and finish reason, never the prompt or completion text. Raw values stay on the log under its native read permission.

Every reply that comes from the index carries the fold frontier of each source, so you can tell a settled view from one that is still catching up. An events page also lists gaps, each a missing range with its reason: expired_before_fold, pruned, truncated, or rebuilding. The liveness_unknown flag says the server's heartbeat view has not caught up, so idle is not meaningful yet. Other flags mark a partial view: label_truncated, overflow, events_truncated, lane_conflict, and rebuilding.

A read fails with a session error: LaserError::Session in Rust and SessionError in Python and TypeScript. Its kinds are unsupported, not_found, not_registered, invalid, unauthorized, stale, backend, and unavailable. Without the capability, the reads fail with an unsupported error before sending. Access needs session:read on stream:<name> and read permission on the whole stream, so a reader limited to some topics of the stream gets unauthorized.

laser.read_at(source) (TypeScript: readAt) fetches the one record a source reference names, such as a timeline row's position. It checks that the topic was not recreated and that the record sits at that offset, and returns nothing when the record is gone. It uses a standard poll, so it works on plain Apache Iggy too.

const sessions = laser.sessions()
const page = await sessions.list().status("active").limit(20).fetch()
for (const info of page.items) {
  console.log(info.id.toString(), info.status, info.idle, info.overBudget)
}

const events = await sessions.events(session.conversation).limit(100).fetch()
for (const event of events.items) {
  const record = await laser.readAt(event.at)
}

const watch = await sessions.watch(1_000)
const change = await watch.next()

let sessions = laser.sessions();
let page = sessions
    .list()
    .status(SessionStatus::Active)
    .limit(20)
    .fetch()
    .await?;
for info in &page.items {
    println!("{} {:?} {} {}", info.id, info.status, info.idle, info.over_budget);
}

let events = sessions.events(session.conversation()).limit(100).fetch().await?;
for event in &events.items {
    let record = laser.read_at(&event.at).await?;
}

let mut watch = sessions.watch(Duration::from_secs(1)).await?;
let change = watch.next().await?;

sessions = laser.sessions()
page = await sessions.list(status="active", limit=20)
for info in page["items"]:
    print(info["id"], info["status"], info["idle"], info["over_budget"])

events = await sessions.events(session.conversation, limit=100)
for event in events["items"]:
    record = await laser.read_at(event["at"])

watch = await sessions.watch(1_000)
change = await watch.next()

Python returns the replies as dicts and takes the list and event filters as keywords. In Rust, SessionStatus is part of laser_sdk::prelude::full. The AGDX page describes the read surface on the wire and over HTTP.

Read limits

context(), context_with, assemble, state().get(), cancel_requested(), and parked() examine at most CONTEXT_READ_WINDOW (10,000) raw records of each partition, the newest ones, then keep this session's records. Many sessions share a partition, so older records of a quiet session on a busy partition can fall outside that window.

turns_at, turns_since, state_at, and replay read the whole range their checkpoint selects. For a long session, save your state with a checkpoint at regular points, then replay reads only the records after the last one.

Recreated streams and topics

A session factory remembers the stream's creation time, the creation time of agent.sessions, and its partition count when it first touches the lane. If the stream or agent.sessions is deleted and created again, or the partition count changes, every later write through that factory and its handles fails with a stale session error before anything is published. Build a new factory with laser.sessions() to write to the new topic. On a managed deployment, a new factory also checks that the lane matches the registered session source, and a stale error means the registration must be recovered first.

Submit and control example

await using worker = Agent.builder()
  .id(AgentId.new("worker"))
  .listenOn(AgentTopic.Sessions)
  .handler({
    handle: async (_message, ctx) => {
      const session = ctx.session()
      await session.state().set("seen", true)
      await session.end()
    }
  })
  .build()
  .spawn(laser)
await worker.ready()

const submitted = await laser
  .sessions()
  .submit(AgentId.new("worker"), new TextEncoder().encode('{"ticket": 42}'))
  .from(AgentId.new("intake"))
  .label("ticket-42")
  .budget({ tokens: 4_000n })
  .send()

// An operator with send permission on agent.control
await laser
  .sessions()
  .control("support", submitted.session)
  .asOperator(AgentId.new("ops"))
  .cancel()

struct Worker;

impl AgentHandler for Worker {
    async fn handle(&self, _message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        let session = ctx.session();
        session.state().set("seen", json!(true)).await?;
        session.end().await
    }
}

let mut worker = Agent::builder()
    .id("worker".parse()?)
    .listen_on(AgentTopic::Sessions)
    .handler(Worker)
    .build()
    .spawn(laser.clone());
worker.ready().await?;

let submitted = laser
    .sessions()
    .submit("worker".parse::<AgentId>()?, br#"{"ticket": 42}"#.to_vec())
    .from("intake".parse::<AgentId>()?)
    .label("ticket-42")
    .budget(Budget { tokens: Some(4_000), cost_micros: None })
    .send()
    .await?;

// An operator with send permission on agent.control
laser
    .sessions()
    .control("support", submitted.session)
    .as_operator("ops".parse::<AgentId>()?)
    .cancel()
    .await?;

async def handle(ctx, message):
    session = ctx.session()
    await session.state().set("seen", True)
    await session.end()


worker = laser.spawn_agent("worker", ls.AgentTopic.Sessions, handle)
await worker.ready()

submitted = await (
    laser.sessions()
    .submit("worker", b'{"ticket": 42}')
    .from_("intake")
    .label("ticket-42")
    .budget(ls.Budget(tokens=4_000))
    .send()
)

# An operator with send permission on agent.control
await laser.sessions().control("support", submitted.session).as_operator("ops").cancel()

The submitted session reads session.submitted and agent.handoff, then session.resumed when the worker picks the command up, then state.updated for the delta and for the snapshot end() writes, and session.completed. In Rust, Budget is part of laser_sdk::prelude::full.

Budget example

const session = laser.sessions().open(submitted.session)
if (await session.overBudget()) {
  console.log("the session passed its budget")
}

let session = laser.sessions().open(submitted.session);
if session.over_budget().await? {
    println!("the session passed its budget");
}

session = laser.sessions().open(submitted.session)
if await session.over_budget():
    print("the session passed its budget")

Custom layout example

const sessions = laser.sessions(
  new SessionConfig()
    .stream("support")
    .layout({
      kind: "perAgentPartition",
      partitions: new Map([
        ["planner", 0],
        ["worker", 1]
      ])
    })
    .idleTimeout(600_000)
    .memoryNamespace("support.sessions")
)
await sessions.bootstrap(2, TopicRetention.expireAfter(86_400_000))

let sessions = laser.sessions_with(
    SessionConfig::new()
        .stream("support")
        .layout(SessionLayout::per_agent_partition([
            (AgentId::new("planner")?, 0),
            (AgentId::new("worker")?, 1),
        ]))
        .idle_timeout(Duration::from_secs(600))
        .memory_namespace("support.sessions"),
);
sessions
    .bootstrap(2, TopicRetention::expire_after(Duration::from_secs(86_400)))
    .await?;

sessions = laser.sessions(
    stream="support",
    layout=ls.SessionLayout.PerAgentPartition({"planner": 0, "worker": 1}),
    idle_timeout_ms=600_000,
    memory_namespace="support.sessions",
)
await sessions.bootstrap(2, ls.TopicRetention.expire_after(86_400_000))

The session factory takes its configuration differently in each language: Rust laser.sessions_with(SessionConfig), TypeScript laser.sessions(new SessionConfig()), and Python keywords on laser.sessions(...). laser.sessions() alone uses the defaults on the connection's default stream.

Where each feature runs

Starting and ending sessions, model and tool records, state, context, checkpoints, replay, over_budget(), read_at, submission, control, and pause work on plain Apache Iggy, on Laser Stack, and on LaserData Cloud. These need the managed session index on Laser Stack or LaserData Cloud:

• get, list, events, state, links, sources, changes, watch, and status().
• Source registration during bootstrap.
• Runtime budget enforcement before a handler runs.
• Verified signers on timeline events.

kv(namespace) and linked_graph(name) need the managed key-value and graph views. Default memory recall reads the managed key-value view, and folded recall works on plain Apache Iggy. In Rust, sessions are part of the agent feature. Python and TypeScript include them.

Key operations

TypeScript uses camelCase names, for example contextWith, recordModelCall, linkedGraph, cancelRequested, asOperator, and readAt.