集群扩展点

集群宿主平台在多个节点上运行长时运行的智能体会话。框架本身并不附带调度器或放置引擎，而是暴露一小组接缝：它定义决策点、发出结构化事件，并把策略交给宿主来提供。下文的所有内容都是你从框架外部接入的——你永远不需要分叉框架。

本页给出 Rust、Node.js、Python 和 Go 中等价的原生接缝。配置形式遵循各语言习惯，底层能力和生命周期契约保持一致。

身份标签

每个会话都可以携带四个不透明的身份标签。框架不解释它们：它通过会话读取器暴露它们，把它们持久化到 SessionData，并在恢复时还原它们。宿主正是借此把一个会话归属到租户、主体、智能体模板以及更广的关联链。唯一的例外是持久记忆（durable-memory）绑定：会话带有该绑定时，tenant_id 与 principal 默认取绑定的命名空间，显式值必须与之完全一致。

请将身份标签与 sessionStore / session_store 搭配使用，使标签在进程重启后依然保留。恢复时，由调用方提供的选项优先生效，因此你可以在节点之间迁移会话时为其重新打标签。

Rust

Node.js

Python

Rust

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

TypeScript

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'

Python

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

预算 / 成本守卫

预算守卫让宿主针对成本或令牌预算对每一次 LLM 调用进行把关。框架会在每次 LLM 请求之前调用你的守卫，并在请求返回之后再次调用。守卫是你自己拥有的策略；框架只负责执行你返回的决策。

Rust

Node.js

Python

Rust

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

TypeScript

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

Node 回调接收单个 ctx 对象，且必须是同步的。也可以提供接收 { sessionId, toolName } 的 checkBeforeTool(ctx)，timeoutMs（默认 5000）限制每次 check* 调用的时长。抛出异常、挂起或返回格式错误决策的 check* 回调会 fail-closed 为 deny；失败的 recordAfterLlm 会被忽略。

Python

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)

Python 守卫还可以定义 check_before_tool(session_id, tool_name)。 session.set_budget_guard(guard, timeout_ms=5000) 可在运行中的会话上安装守卫。异常、无法解析的决策和超时都会 fail-closed。

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

Go 回调遵循相同的 fail-closed 契约：回调报错、超时或返回无法解析的决策时，受保护的 LLM 或工具调用会被拒绝。

四种 SDK 的原生守卫具有等价的决策结构：

返回值	效果
`None` / `null` / `{ decision: 'allow' }`	继续执行 LLM 调用。
`{ decision: 'soft', resource, consumed, limit, message? }`	发出 `BudgetThresholdHit`（kind 为 `soft`）并继续执行。
`{ decision: 'deny', resource, reason }`	发出 `BudgetThresholdHit`（kind 为 `hard`），并以 `Budget exhausted on '<resource>': <reason>` 使调用失败。

被拒绝的调用在 Python 中抛出 RuntimeError，在 Node 中 reject，在 Go 中返回 error；会话保持打开。在所有 SDK 中，缺失的守卫方法都会被当作宽松默认值处理；回调出错、超时或返回格式错误的 check* 决策时，都会 fail-closed 为 deny。

集群事件词汇

AgentEvent 预留了三个集群级变体，并带有稳定的 JSON 标签，便于宿主和外部生产者针对同一 schema：

BudgetThresholdHit { resource, kind, consumed, limit, message? }（budget_threshold_hit）—— 预算守卫返回 soft（kind 为 soft）或 deny（kind 为 hard，原因放在 message 中）时，agent 循环会在会话事件流上发出它。
PassivationRequested { reason, deadline_ms? }（passivation_requested）—— 宿主请求会话进入一个安全、可持久化的状态，以便将其从当前节点驱逐。deadline_ms 若存在，是强制关闭前的 Unix 纪元截止时间（毫秒）。
PeerInvocation { from_session_id, from_tenant_id?, correlation_id? }（peer_invocation）—— 另一个会话调用了本会话。这些标签让接收方能够把调用归属回其源租户和关联链。

框架只产生 BudgetThresholdHit。它从不发出 PassivationRequested 或 PeerInvocation；使用它们的宿主通过自己拥有的传输自行发出。这些变体不是钩子事件类型，因此 registerHook / register_hook 的处理器收不到它们——请在事件流上观察它们（钩子事件列表参见钩子）。

确定性 ID 与时间（重放）

希望在另一节点上对某次运行进行逐位一致重放的集群，必须消除常规运行中两处不确定性的来源：随机 ID 和挂钟时间。Rust 核心将二者建模在一个 HostEnv { id_generator, clock } 之后。默认实现把 UUID 生成器与系统时钟配对；重放工具会换入 SequentialIdGenerator 和 FixedClock，使得对相同输入的重新执行在任意节点上都产生相同的 ID 和时间戳，从而产生相同的输出。

四种 SDK 都提供同一确定性配置。ID 前缀和固定时间戳相互独立——省略的字段保留系统默认实现——因此重放时请同时设置两者，并在每次重放开始时用相同的值重新创建配置。

Rust

Node.js

Python

Rust

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

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

循环检查点与运行恢复

配置了 sessionStore / session_store 后，智能体循环会在每一轮的工具调用全部落定后、于该轮末尾持久化一个以运行 id 为键的循环检查点。运行在进程内结束时，该检查点会被删除。任何共享同一存储的节点都可以恢复该运行。检查点提供的是事实而不是决策：恢复节点用它为一个全新的事实日志线程播种，再折叠该日志，由日志选择下一步。

Rust

Node.js

Python

Rust

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

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

恢复的工作会记录为一个新运行；存储中检查点所属的运行保持不变。如果不存在检查点，但会话的事实日志已有事实，resume_run 会改为折叠该日志。只有在事实日志也为空时，才会出现下面两种错误：

resume_run requires a session_store on this session：未配置存储；回退到一个全新会话。
no loop checkpoint found for run 'X'：该运行从未到达其第一个检查点，或检查点已在运行结束时删除；将该运行视为已落定或已丢失。

日志中缺少结果的工具调用会在恢复时执行一次。需要精确、可安全重放的恢复标识的宿主使用 spawnRecoveryWithRunId。完整规则参见持久化。

长时运行会话的保留上限

运行数小时或数天的会话会在四个内存存储中累积状态：运行记录、每次运行的事件缓冲区、追踪事件，以及终态子智能体任务快照。若不加限制，它们会随会话寿命增长——对短寿命会话无妨，对长寿命会话则是真实的泄漏。

SessionRetentionLimits 为这些存储设置上限。默认值如下：

字段	默认值
`max_runs_retained`	64
`max_events_per_run`	2,048
`max_event_bytes_per_run`	8 MiB
`max_trace_events`	8,192
`max_terminal_subagent_tasks`	512

省略字段时保留这些默认值；只有明确需要无限保留时才设置 unbounded: true（Rust： SessionRetentionLimits::unbounded()）。上限是软性的：存储达到上限时，插入新条目会丢弃最旧的条目，且不返回错误。正在运行的子智能体任务永不被丢弃——只有终态快照会被驱逐，按完成时间从最旧开始。

Node 使用 retentionLimits，Python 使用 opts.retention_limits，Go 使用 SessionOptions.RetentionLimits。Rust 宿主使用 SessionOptions::with_retention_limits(...)。字段名和示例见限制。

另见： 多机部署 · 持久化 · 限制 · 钩子

#集群扩展点

#身份标签

#预算 / 成本守卫

#集群事件词汇

#确定性 ID 与时间（重放）

#循环检查点与运行恢复

#长时运行会话的保留上限

集群扩展点

身份标签

预算 / 成本守卫

集群事件词汇

确定性 ID 与时间（重放）

循环检查点与运行恢复

长时运行会话的保留上限