集群扩展点

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

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

身份标签

每个会话都可以携带四个不透明的身份标签。框架从不解释它们——它会把它们传播到钩子、追踪和 SessionData，并在恢复时还原它们。宿主正是借此把一个会话归属到租户、主体、智能体模板以及更广的关联链。

请将身份标签与 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',
});

// 读取器返回 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)

# 读取器是属性，返回 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))
}

预算 / 成本守卫

预算守卫让宿主针对成本或令牌预算对每一次 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("总结这个仓库。", 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);
  },
});

// 清除预算守卫
session.setBudgetGuard(null);

Node 回调接收单个 ctx 对象，且绝不能抛出异常。请用 try/catch 包裹逻辑并返回显式决策。卡住或无法解析的 check* 回调会 fail-closed 为 deny。

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)

# 如需清除：把 opts.budget_guard 设为 None，再重新创建会话。

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, "总结这个仓库。")
	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 }`	中止 LLM 调用。Python 抛出 `RuntimeError("Budget exhausted...")`；Node 以 `"Budget exhausted..."` 拒绝（reject）。

这种健壮性是刻意为之，但各 SDK 的失败语义略有不同：缺失的守卫方法会被当作宽松默认值处理。Python 回调出错会回退为 Allow。Node 回调不能 throw。Go 回调报错、超时或返回无法解析的 check* 决策时，会 fail-closed 为 deny。

集群事件词汇

宿主通过其钩子执行器，将集群级别的决策作为结构化的 AgentEvent 变体发出。会话内的钩子以统一方式订阅它们——与它们观察其他任何事件的方式相同——因此在宿主处编写的策略会原样呈现给智能体自身的钩子，无需特殊处理。

集群词汇如下：

BudgetThresholdHit { resource, kind, consumed, limit, message? } —— 预算守卫返回了 soft 决策（或宿主越过了它自己跟踪的某个阈值）。kind 用于区分软性警告与更硬性的限制。
PassivationRequested { reason, deadline_ms? } —— 宿主请求会话进入一个安全、可持久化的状态，以便将其从当前节点驱逐。deadline_ms 若存在，则表示强制驱逐前的宽限窗口。
PeerInvocation { from_session_id, from_tenant_id?, correlation_id? } —— 另一个会话调用了本会话。这些标签让接收方能够把调用归属回其源租户和关联链。

这些事件通过你的会话内钩子已经在使用的、经过验证的同一套钩子 API 来观察——Node 中为 session.registerHook，Python 中为 session.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/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())
}

循环检查点与运行恢复

配置了 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/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)
}

系统会为恢复的工作分配一个新的运行 id——存储中的原始运行保持不变。有两条错误路径值得处理：

resume_run requires a session_store —— 未配置存储；回退到一个全新会话。
no loop checkpoint found for run 'X' —— 该运行从未到达其第一个检查点，或已被清理；稍后重试，或将该运行视为丢失。

由于检查点只在工具轮次之间、绝不在工具执行中途生成，恢复的运行永远不会重放一个执行到一半的工具。存储细节参见持久化。

长时运行会话的保留上限

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

SessionRetentionLimits 为这些存储设置上限。省略字段时保留框架的有限默认值；只有明确需要无限保留时才设置 unbounded: true。驱逐采用严格的 FIFO，并且 正在运行的子智能体任务永不被丢弃——只有终态快照会被驱逐。

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

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

#集群扩展点

#身份标签

#预算 / 成本守卫

#集群事件词汇

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

#循环检查点与运行恢复

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

集群扩展点

身份标签

预算 / 成本守卫

集群事件词汇

确定性 ID 与时间（重放）

循环检查点与运行恢复

长时运行会话的保留上限