#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 never interprets them — it propagates them to hooks, traces, and 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.

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/v6"
)

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)

# To clear: set opts.budget_guard = None and re-create the session.

package main

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

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

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:

Robustness is intentional but SDK-specific: a missing guard method is treated as the permissive default. Python callback errors fall back to Allow. Node callbacks must not throw. Go errors, timeouts, and malformed check* returns fail closed as deny.

#Cluster event vocabulary

The host emits cluster-level decisions as structured AgentEvent variants through its hook executor. In-session hooks subscribe to them uniformly — the same way they observe any other event — so policy authored at the host shows up to the agent's own hooks without special casing.

The cluster vocabulary is:

• BudgetThresholdHit { resource, kind, consumed, limit, message? } — a budget guard returned a soft decision (or the host crossed a threshold it tracks itself). kind distinguishes soft warnings from harder limits.
• PassivationRequested { reason, deadline_ms? } — 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 grace window before forced eviction.
• PeerInvocation { from_session_id, from_tenant_id?, correlation_id? } — another session invoked this one. The labels let the receiver attribute the call back to its origin tenant and correlation chain.

These are observed through the same verified hook API your in-session hooks already use — session.registerHook in Node, session.register_hook in Python (see Hooks). Treat the three variants above as the documented contract; the host is responsible for emitting them via its hook executor.

#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. Set both the ID prefix and fixed timestamp, 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/v6"
)

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 checkpoint after each completed tool round, keyed by run id. Any node that shares the same store can rehydrate the run and continue it.

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/v6"
)

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)
}

A new run id is allocated for the resumed work — the original run is left intact in the store. Two error paths are worth handling:

• resume_run requires a session_store — 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 it was pruned; retry later or treat the run as lost.

Because checkpoints are taken only between tool rounds, never mid-tool, a resumed run never replays a half-executed tool. See Persistence for store details.

#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. Omitted fields keep the framework's finite defaults; set unbounded: true only when unlimited retention is intentional. Eviction is strict FIFO, and running subagent tasks are never dropped — only terminal snapshots are evicted.

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