#Cluster Extension Points

A cluster host platform runs long-lived agent sessions across many nodes. The framework does not ship a scheduler or a placement engine. Instead it exposes a small set of seams: it defines the decision points, emits structured events, and lets the host supply the policy. Everything below is something you wire from outside the framework — you never fork it.

This page shows the equivalent native seams in Rust, Node.js, Python, and Go. The configuration shape follows each language's conventions, while the underlying capability and lifecycle contract stay the same.

#Identity labels

Every session can carry four opaque identity labels. The framework does not interpret them: it exposes them through session getters, persists them in SessionData, and restores them on resume. This is how a host attributes a session to a tenant, a principal, an agent template, and a wider correlation chain. The one exception is a durable-memory binding: when a session has one, tenant_id and principal default to the binding's namespace, and explicit values must match it exactly.

Pair identity labels with a sessionStore / session_store so the labels survive a process restart. On resume, caller-supplied options win, so you can relabel a session as you move it between nodes.

use a3s_code_core::{Agent, SessionOptions};

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let options = SessionOptions::new()
        .with_tenant_id("tenant-example")
        .with_principal("principal-example")
        .with_agent_template_id("agent-template-example")
        .with_correlation_id("trace-example")
        .with_file_session_store("./.a3s/sessions");
    let session = agent
        .session_builder("/path/to/project")
        .options(options)
        .build()
        .await?;

    println!("tenant: {:?}", session.tenant_id());
    println!("principal: {:?}", session.principal());
    println!("template: {:?}", session.agent_template_id());
    println!("correlation: {:?}", session.correlation_id());

    session.close().await;
    agent.close().await;
    Ok(())
}

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

// Getters return string | null
console.log(session.tenantId); // 'tenant-example'
console.log(session.principal); // 'principal-example'
console.log(session.agentTemplateId); // 'agent-template-example'
console.log(session.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('/path/to/project', opts)

# Getters are properties, return str | None
print(session.tenant_id)          # 'tenant-example'
print(session.principal)          # 'principal-example'
print(session.agent_template_id)  # 'agent-template-example'
print(session.correlation_id)     # 'trace-example'

package main

import (
	"context"
	"fmt"
	"log"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func value(value *string) string {
	if value == nil {
		return ""
	}
	return *value
}

func main() {
	ctx := context.Background()
	agent, err := code.Create(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(ctx)

	session, err := agent.Session(ctx, "/path/to/project", &code.SessionOptions{
		TenantID:           "tenant-example",
		Principal:          "principal-example",
		AgentTemplateID:    "agent-template-example",
		CorrelationID:      "trace-example",
		FileSessionStoreDir: "./.a3s/sessions",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(ctx)

	info, err := session.Info(ctx)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("tenant:", value(info.TenantID))
	fmt.Println("principal:", value(info.Principal))
	fmt.Println("template:", value(info.AgentTemplateID))
	fmt.Println("correlation:", value(info.CorrelationID))
}

#Budget / cost guard

A budget guard lets the host gate every LLM call against a cost or token budget. The framework calls your guard before each LLM request and after it returns. The guard is policy you own; the framework only enforces the decision you hand back.

use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};

use a3s_code_core::{
    budget::{BudgetDecision, BudgetGuard},
    llm::TokenUsage,
    Agent, SessionOptions,
};

#[derive(Debug)]
struct TokenGuard {
    limit: usize,
    used: AtomicUsize,
}

#[async_trait::async_trait]
impl BudgetGuard for TokenGuard {
    async fn check_before_llm(
        &self,
        _session_id: &str,
        estimated_prompt_tokens: usize,
    ) -> BudgetDecision {
        let used = self.used.load(Ordering::Relaxed);
        if used.saturating_add(estimated_prompt_tokens) > self.limit {
            BudgetDecision::Deny {
                resource: "tokens".into(),
                reason: "monthly cap reached".into(),
            }
        } else {
            BudgetDecision::Allow
        }
    }

    async fn record_after_llm(&self, _session_id: &str, usage: &TokenUsage) {
        self.used.fetch_add(usage.total_tokens, Ordering::Relaxed);
    }
}

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let guard = Arc::new(TokenGuard {
        limit: 50_000,
        used: AtomicUsize::new(0),
    });
    let session = agent
        .session_builder(".")
        .options(SessionOptions::new().with_budget_guard(guard))
        .build()
        .await?;

    let result = session.send("Summarize this repository.", None).await?;
    println!("{}", result.text);

    session.close().await;
    agent.close().await;
    Ok(())
}

session.setBudgetGuard({
  checkBeforeLlm(ctx) {
    if (overLimit(ctx.sessionId, ctx.estimatedTokens)) {
      return {
        decision: 'deny',
        resource: 'tokens',
        reason: 'monthly cap reached',
      };
    }
    return { decision: 'allow' };
  },
  recordAfterLlm(ctx) {
    meter(ctx.sessionId, ctx.usage);
  },
});

// Clear the guard
session.setBudgetGuard(null);

class MyGuard:
    def check_before_llm(self, session_id, estimated_tokens):
        if over_limit(session_id, estimated_tokens):
            return {'decision': 'deny', 'resource': 'tokens', 'reason': 'monthly cap reached'}
        return {'decision': 'allow'}

    def record_after_llm(self, session_id, usage):
        meter(session_id, usage)

opts = SessionOptions()
opts.budget_guard = MyGuard()
session = agent.session('/path/to/project', opts)

# Clear the guard
session.set_budget_guard(None)

package main

import (
	"context"
	"fmt"
	"log"
	"sync/atomic"
	"time"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func main() {
	ctx := context.Background()
	agent, err := code.Create(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(ctx)

	session, err := agent.Session(ctx, ".", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(ctx)

	const limit uint64 = 50_000
	var used atomic.Uint64
	err = session.SetBudgetGuard(ctx, &code.BudgetGuardHandlers{
		CheckBeforeLLM: func(
			_ context.Context,
			call code.BudgetLLMContext,
		) (*code.BudgetDecision, error) {
			if used.Load()+uint64(call.EstimatedTokens) > limit {
				return &code.BudgetDecision{
					Decision: "deny",
					Resource: "tokens",
					Reason:   "monthly cap reached",
				}, nil
			}
			return &code.BudgetDecision{Decision: "allow"}, nil
		},
		RecordAfterLLM: func(
			_ context.Context,
			call code.BudgetUsageContext,
		) error {
			used.Add(uint64(call.Usage.TotalTokens))
			return nil
		},
		Timeout: 2 * time.Second,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer session.SetBudgetGuard(context.Background(), nil)

	result, err := session.Run(ctx, "Summarize this repository.")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("%s\nspent %d / %d tokens\n", result.Text, used.Load(), limit)
}

The native guard decision shape is equivalent across all four SDKs:

A denied call raises RuntimeError in Python, rejects in Node, and returns an error in Go; the session stays open. A missing guard method is treated as the permissive default in every SDK. In every SDK, callback errors, timeouts, and malformed check* returns fail closed as deny.

#Cluster event vocabulary

AgentEvent reserves three cluster-level variants with stable JSON tags, so a host and external producers can target one schema:

• BudgetThresholdHit { resource, kind, consumed, limit, message? } (budget_threshold_hit) — the agent loop emits it on the session event stream when a budget guard returns soft (kind soft) or deny (kind hard, with the reason in message).
• PassivationRequested { reason, deadline_ms? } (passivation_requested) — the host is asking the session to reach a safe, persistable state so it can be evicted from this node. deadline_ms, when present, is the Unix-epoch deadline in milliseconds before forced close.
• PeerInvocation { from_session_id, from_tenant_id?, correlation_id? } (peer_invocation) — another session invoked this one. The labels let the receiver attribute the call back to its origin tenant and correlation chain.

The framework produces only BudgetThresholdHit. It never emits PassivationRequested or PeerInvocation; a host that uses them emits them itself through the transport it owns. These variants are not hook event types, so registerHook / register_hook handlers do not receive them — observe them on the event stream (see Hooks for the hook event list).

#Deterministic IDs and time (replay)

A cluster that wants bit-identical replay of a run on a different node must remove the two sources of nondeterminism in a normal run: random IDs and the wall clock. The Rust core models both behind a HostEnv { id_generator, clock }. The default pairs a UUID generator with the system clock; replay tooling swaps in a SequentialIdGenerator and a FixedClock so that re-executing the same inputs produces the same IDs and timestamps, and therefore the same output, on any node.

All four SDKs expose the same deterministic configuration. The ID prefix and fixed timestamp are independent — an omitted field keeps its system-backed default — so set both for replay, and recreate the configuration from the same values at the start of each replay.

use std::sync::Arc;

use a3s_code_core::{
    host_env::{FixedClock, HostEnv, SequentialIdGenerator},
    Agent, SessionOptions,
};

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let host_env = Arc::new(HostEnv::new(
        Arc::new(SequentialIdGenerator::new("replay")),
        Arc::new(FixedClock::new(1_700_000_000_000)),
    ));
    let session = agent
        .session_builder(".")
        .options(SessionOptions::new().with_host_env(host_env))
        .build()
        .await?;

    println!("{}", session.session_id());
    session.close().await;
    agent.close().await;
    Ok(())
}

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

const agent = await Agent.create('agent.acl');
const session = await agent.sessionAsync('.', {
  hostEnv: {
    sequentialIdPrefix: 'replay',
    fixedTimeMs: 1_700_000_000_000,
  },
});

console.log(session.sessionId);
await session.closeAsync();
await agent.close();

from a3s_code import Agent, HostEnvConfig, SessionOptions

agent = Agent.create('agent.acl')
options = SessionOptions()
options.host_env = HostEnvConfig(
    sequential_id_prefix='replay',
    fixed_time_ms=1_700_000_000_000,
)
session = agent.session('.', options)

print(session.session_id)
session.close()
agent.close()

package main

import (
	"context"
	"fmt"
	"log"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func main() {
	ctx := context.Background()
	agent, err := code.NewAgent(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(ctx)

	prefix := "replay"
	fixedTime := uint64(1_700_000_000_000)
	session, err := agent.Session(ctx, ".", &code.SessionOptions{
		HostEnv: &code.HostEnvConfig{
			SequentialIDPrefix: &prefix,
			FixedTimeMS:        &fixedTime,
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(ctx)

	fmt.Println(session.SessionID())
}

#Loop checkpoints and run resumption

With a sessionStore / session_store configured, the agent loop persists a loop checkpoint, keyed by run id, at the end of each turn once that turn's tool calls have settled. The checkpoint is deleted when the run finishes in-process. Any node that shares the same store can recover the run. The checkpoint supplies facts, not decisions: the recovering node seeds a fresh fact-log thread from it and folds that log, which chooses the next step.

use a3s_code_core::{Agent, SessionOptions};

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let session = agent
        .session_builder(".")
        .options(
            SessionOptions::new()
                .with_file_session_store("./.a3s/sessions")
                .with_session_id("session-from-node-a"),
        )
        .build()
        .await?;

    let result = session.resume_run("run-id-from-node-a").await?;
    println!("{}", result.text);

    session.close().await;
    agent.close().await;
    Ok(())
}

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

const session = agent.session(workspace, {
  sessionStore: new FileSessionStore('./.a3s/sessions'),
  sessionId: 'session-from-node-a',
});

const result = await session.resumeRun('run-id-from-node-a');

from a3s_code import FileSessionStore

opts = SessionOptions()
opts.session_store = FileSessionStore('./.a3s/sessions')
opts.session_id = 'session-from-node-a'
session = agent.session(workspace, opts)

result = session.resume_run('run-id-from-node-a')

package main

import (
	"context"
	"fmt"
	"log"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func main() {
	ctx := context.Background()
	agent, err := code.Create(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(ctx)

	options := &code.SessionOptions{
		SessionID:           "session-from-node-a",
		FileSessionStoreDir: "./.a3s/sessions",
	}
	session, err := agent.Session(ctx, ".", options)
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(ctx)

	result, err := session.ResumeRun(ctx, "run-id-from-node-a")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(result.Text)
}

The resumed work is recorded as a new run; the checkpoint's run is left intact in the store. When no checkpoint exists but the session's fact log already has facts, resume_run folds that log instead. The two errors below fire only when the fact log is also empty:

• resume_run requires a session_store on this session: no store was configured; fall back to a fresh session.
• no loop checkpoint found for run 'X': the run never reached its first checkpoint, or the checkpoint was deleted when the run ended; treat the run as settled or lost.

A tool call whose result is missing from the log runs once on resume. Hosts that need an exact, replay-safe recovery id use spawnRecoveryWithRunId. See Persistence for the full rules.

#Retention caps for long-running sessions

A session that runs for hours or days accumulates state in four in-memory stores: run records, per-run event buffers, trace events, and terminal subagent task snapshots. Left unbounded, these grow with session age — fine for short-lived sessions, a real leak for long-lived ones.

SessionRetentionLimits caps each store. The defaults are:

Omitted fields keep these defaults; set unbounded: true (Rust: SessionRetentionLimits::unbounded()) only when unlimited retention is intentional. Caps are soft: when a store reaches its cap, the oldest entry is dropped on insert and no error is returned. Running subagent tasks are never dropped — only terminal snapshots are evicted, oldest completion first.

Use retentionLimits in Node, opts.retention_limits in Python, or SessionOptions.RetentionLimits in Go. Rust hosts use SessionOptions::with_retention_limits(...). See Limits.

See also: Multi-machine · Persistence · Limits · Hooks