cua-sandbox Fleet Builder API 迁移实战：用不可变流式 Builder 重构沙箱舰队请求构建

cua-sandbox Fleet Builder API 迁移实战：用不可变流式 Builder 重构沙箱舰队请求构建

【免费下载链接】cuaScale computer-use 2.0 with open-source drivers, cross-OS fleets, and benchmarks for training, evaluation, and data generation.项目地址: https://gitcode.com/GitHub_Trending/cua/cua

本文基于 docs/superpowers/plans/2026-08-07-sandbox-fleet-builder-api.md 实现计划展开，结合当前仓库中已落地的源码与测试，完整还原一次「从直接构造 Fleet 记录到生成式 Builder API」的 SDK 内部迁移：如何用 AST 契约测试锁定重构边界、如何在cua_sandbox顶层再导出 7 个 Builder、如何迁移传输层与测试夹具，并最终通过打包、发行与代码质量验证。读者读完可掌握一套可复制的「先立契约、再动代码、TDD 验证、打包兜底」的 SDK 演进方法论，以及 cua-fleet Builder API 的完整用法。

背景与目标：为什么需要 Builder API

cua-sandbox 是 CUA 项目的 Python 沙箱 SDK（位于 libs/python/cua-sandbox），负责向调用方提供"临时/持久化沙箱计算环境"的能力。它通过底层生成式 UniFFI Python 绑定fleet_sdk（由cua-fleet发行包提供）与 Fleet 云端交互，构建两类核心请求：

• Template 请求（CreateTemplateRequest）：描述虚拟机模板——容器磁盘镜像、CPU/内存、探针、服务端口、固件等；
• Pool 请求（CreatePoolRequest）：描述热池（warm pool）规格——副本数、模板引用、自动扩缩容、TTL 等。

在迁移之前，沙箱 SDK 与测试通过「直接调用 Fleet 记录类的构造函数」来构造这些请求图。迁移的目标非常明确：

将 Python 沙箱 SDK 及其测试，从直接构造 builder-enabled Fleet 记录，迁移到生成的 Fleet builder API，同时保留所有既有公开沙箱 API。

计划给出了清晰的架构决策（见原文档 "Architecture" 段）：

• 从cua_sandbox再导出生成的 builder 伴生类；
• 在既有边界处用不可变流式 Builder（immutable fluent builders）构造 Fleet template 与 pool 请求图；
• 未提供生成 builder 的 Fleet 记录继续使用构造函数；
• 用一个聚焦的 AST 契约测试，防止 builder-enabled 记录重新回归直接构造。

全局约束：迁移的红线

原计划列出六条全局约束，是任何实现都不可逾越的边界，也是理解后续每个任务设计意图的钥匙：

1. 保留cua_sandbox全部既有公开名称、签名与返回的 Fleet 记录类型——迁移是内部的，对外 API 完全不变；
2. 增加 builder 导出，但不删除 legacy 记录导出或构造函数兼容性——两类用法并存；
3. 将沙箱 SDK 锁定到 builder-enabled 的cua-fleet发行版——计划写作时指向==0.1.7，当前仓库 pyproject.toml 已演进为cua-fleet==0.1.17；
4. 迁移cua_sandbox包与tests下所有对 builder-enabled Fleet 记录的直接调用；
5. CreateClaimRequest、ClaimSpec、HttpHeader、HttpRequest、HttpResponse、CyclopsConfiguration、CyclopsCredentials以及 FleetSandbox的构造函数调用保持不变——因为生成的 SDK 没有为它们提供 builder；
6. 不编辑 libs/fleet/sdk-bindings 下的生成文件，也不在用户明确要求前创建提交。

其中第 5 条尤其关键：它划定了「哪些必须迁移、哪些保持原样」的精确范围，避免了为追求形式统一而过度重构。

Task 1：用 AST 契约测试锁定 Builder 使用契约

迁移最容易发生的回归，是后续开发者图省事又写回CreatePoolRequest(...)之类的直接构造。因此计划的第一步不是改代码，而是先写一个会失败（RED）的源码契约测试，把"不允许直接构造 builder-enabled 记录"固化成一条不可违背的规则。

七个 builder-enabled 记录

契约测试将以下七个 Fleet 记录类列为「必须走 Builder」名单（见 test_fleet_builder_usage.py 的BUILDER_ENABLED_RECORDS）：

计划中的首个契约测试版本很直接：遍历cua_sandbox/与tests/下所有*.py，解析 AST，凡是ast.Call且函数名落在名单内即记为违规：

from __future__ import annotations import ast from pathlib import Path PACKAGE_ROOT = Path(__file__).parents[1] BUILDER_ENABLED_RECORDS = { "CreatePoolRequest", "CreateTemplateRequest", "OsGymSandboxTemplateSpec", "OsGymSandboxWarmPoolSpec", "SandboxService", "SandboxTemplateRef", "VmTemplate", } def test_builder_enabled_fleet_records_use_generated_builders() -> None: violations: list[str] = [] source_roots = (PACKAGE_ROOT / "cua_sandbox", PACKAGE_ROOT / "tests") for source_root in source_roots: for path in source_root.rglob("*.py"): tree = ast.parse(path.read_text(), filename=str(path)) for node in ast.walk(tree): if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name): continue if node.func.id in BUILDER_ENABLED_RECORDS: relative_path = path.relative_to(PACKAGE_ROOT) violations.append(f"{relative_path}:{node.lineno}: {node.func.id}") assert violations == [], "Direct Fleet record constructors remain:\n" + "\n".join(violations)

按计划执行uv run --project libs/python/cua-sandbox pytest libs/python/cua-sandbox/tests/test_fleet_builder_usage.py -q，预期结果是FAIL，并列出cua_sandbox/transport/fleet_cloud.py与tests/test_pool.py中的直接构造调用——这正为后续 Task 3、Task 4 的迁移提供了明确的"待办清单"。

从计划到落地的演进：绑定作用域解析

当前仓库中的 test_fleet_builder_usage.py 把这份契约测试实现得远比计划中的初版严谨。计划的 Interfaces 要求检测器必须能解析三类导入形态：

• from ... import ... as ...别名导入（如from fleet_sdk import VmTemplate as FleetVm）；
• import fleet_sdk as ...模块别名下的属性调用（如fleet.VmTemplate()）；
• import cua_sandbox as ...属性调用（如cua.SandboxService()）。

落地版实现了一个完整的_BindingScope作用域链 +_BuilderRecordCallVisitor，通过维护模块/函数/类的词法作用域，能够正确处理：

• 直接导入、别名导入、模块属性调用的识别（对应测试test_finds_direct_builder_record_imports、test_finds_aliased_builder_record_imports、test_finds_builder_record_module_attributes）；
• 本地同名类/函数/赋值对 Fleet 名称的遮蔽（shadowing）——被遮蔽后不应误报（test_ignores_unrelated_local_builder_record_name、test_ignores_direct_import_after_local_shadowing）；
• 函数内导入不泄漏到兄弟或外层作用域（test_function_imports_do_not_leak_to_sibling_or_outer_scopes）；
• 模块别名被重绑定后不再当作 Fleet 模块（test_ignores_module_attribute_after_alias_rebinding）；
• 嵌套作用域可解析外层未遮蔽的 Fleet 绑定（test_nested_scope_resolves_unshadowed_outer_fleet_binding）。

这些聚焦的 snippet 测试（每个导入形态一个）保证了契约检测器本身的行为可被单独验证，是"测试的测试"，也防止契约测试因误报/漏报而在未来被绕过。

Task 2：通过 cua_sandbox 导出 Fleet Builders

契约测试锁定了"必须用 Builder"，接下来要让 Builder 真正可用。Task 2 的核心是在cua_sandbox的公开命名空间再导出七个 Builder，让外部使用者只需要from cua_sandbox import CreatePoolRequestBuilder即可，无需感知底层fleet_sdk的存在。

先写一个失败的公开导出测试

计划先在 tests/test_pool.py 的from cua_sandbox import (...)导入列表中加入全部七个 Builder 名，并新增test_public_pool_schema_exports_generated_builders()——这个测试同时也演示了 Builder API 的完整拼装链路：

def test_public_pool_schema_exports_generated_builders() -> None: service = SandboxServiceBuilder().name("server").target_port(8000).build() vm_template = ( VmTemplateBuilder() .container_disk_image("registry.example/workspace:latest") .services([service]) .build() ) template_request = ( CreateTemplateRequestBuilder() .namespace("default") .name("workspace") .spec(OsGymSandboxTemplateSpecBuilder().vm_template(vm_template).build()) .build() ) pool_request = ( CreatePoolRequestBuilder() .namespace("default") .spec( OsGymSandboxWarmPoolSpecBuilder() .replicas(1) .sandbox_template_ref(SandboxTemplateRefBuilder().name("workspace").build()) .build() ) .build() ) assert template_request.spec.vm_template.services == [service] assert pool_request.spec.sandbox_template_ref.name == "workspace"

此时运行该测试会得到collection ERROR——因为cua_sandbox尚未导出这些名字，验证了 RED 阶段。

锁定 builder-enabled 的 Fleet 发行版

随后在pyproject.toml中把cua-fleet==0.0.10升级为 builder-enabled 版本（计划为==0.1.7），并运行：

uv lock --project libs/python/cua-sandbox --upgrade-package cua-fleet

预期uv.lock解析出目标版本。计划还专门要求处理两个配套测试：

• test_fleet_sdk_packaging.py：移除其中与 Fleet 版本无关的cua-sandbox项目版本硬编码断言，避免每次发版都要附带改动；随后把版本断言更新为目标版本、把期望 registry 更新为https://wheels.cua.ai/simple，先观察 RED 再改到 GREEN；
• test_fleet_sdk_distribution.py：先运行观察 RED，再把期望的发行版本从0.0.7改为目标版本（当前仓库 test_fleet_sdk_distribution.py 断言cua_fleet_distribution.version == "0.1.16"，并校验fleet_sdk/__init__.py确实来自该发行包），重跑至 GREEN。

这条路径保证「锁版本」不止是改一行依赖声明，还通过打包与发行测试验证了依赖解析的真实状态。

再导出全部七个 Builder

在 cua_sandbox/init.py 的既有from fleet_sdk import (...)块中加入七个 Builder，并在__all__中与对应记录名相邻添加字符串：

CreatePoolRequestBuilder CreateTemplateRequestBuilder OsGymSandboxTemplateSpecBuilder OsGymSandboxWarmPoolSpecBuilder SandboxServiceBuilder SandboxTemplateRefBuilder VmTemplateBuilder

当前仓库正是这样实现的——__all__中每个*Record后面紧跟着对应的*RecordBuilder，公开命名空间同时保留记录类型与 Builder。WarmPoolAutoscalingBuilder也随之上线，供热池自动扩缩容场景使用。修改后重跑导出测试，预期GREEN。

Task 3：用 Builder 构建 Fleet Cloud 请求

契约与导出就绪后，进入核心迁移：改造 cua_sandbox/transport/fleet_cloud.py 中_FleetCloudTransport的_template_request()与_pool_request()两个内部方法，要求请求记录值完全不变。

替换 builder-enabled 导入

保留用于注解的请求记录类型（CreateTemplateRequest、CreatePoolRequest），其余 builder-enabled 记录导入替换为七个 Builder，以及Firmware、PreservedJson、ServiceProtocol、WarmPoolAutoscaling等辅助值。

流式构建服务记录

_template_request()先把端口表展开为SandboxService列表（当前实现见 fleet_cloud.py）：

services = [ SandboxServiceBuilder() .name(name) .target_port(port) .protocol(ServiceProtocol.TCP) .build() for name, port in service_ports.items() ]

注意默认端口表的组装逻辑：显式传入services时以"server"端口为基准合并其余命名服务；未传时则从镜像暴露端口self._image._ports生成port-<port>命名服务（见 fleet_cloud.py）。测试 test_fleet_cloud_transport.py 验证了expose(3000)后请求中服务列表为[("server", 8000), ("port-3000", 3000)]。

构建 VM、模板规格与模板请求

vm_template_builder = ( VmTemplateBuilder() .container_disk_image(self._image._registry) .image_pull_secret("ecr-credentials") .probes( PreservedJson.from_json( json.dumps({"readinessProbe": {"tcpSocket": {"port": 8000}}}) ) ) .services(services) ) if self._cpu is not None: vm_template_builder = vm_template_builder.cpu_cores(self._cpu) if self._memory_mb is not None: vm_template_builder = vm_template_builder.memory(f"{self._memory_mb}Mi") template_spec = OsGymSandboxTemplateSpecBuilder().vm_template(vm_template_builder.build()).build() return ( CreateTemplateRequestBuilder() .namespace(self._name) .name(self._name) .spec(template_spec) .build() )

计划明确了一条重要约定：当既有值为None时省略可选 setter——生成式记录对这些字段仍会保持None，因此省略与显式置空结果等价，但代码更干净。

当前仓库实现在此基础上又补充了两处真实业务逻辑（见 fleet_cloud.py）：

• ECR 拉取凭证按需附加：仅当镜像来自账号私有 ECR 时（_needs_ecr_pull_secret判断.dkr.ecr.与.amazonaws.com后缀）才.image_pull_secret("ecr-credentials")。注释解释了原因：网关的准入策略会把「附带了拉取凭证」解读为"需要执行 ECR 白名单"，对公有镜像反而会导致拒绝拉取；
• Windows 镜像强制 UEFI 固件：Windows 客户机磁盘只按 UEFI 构建（见registry/qemu_builder.py），而 Fleet 模式默认固件是 BIOS，因此os_type == "windows"时追加.firmware(Firmware.EFI)，否则 Windows 镜像在 SeaBIOS 下永远无法通过就绪探针。测试test_windows_image_boots_uefi与test_linux_image_leaves_firmware_at_the_schema_default（test_fleet_cloud_transport.py）精确锁定了这一分支。

构建热池规格与请求

_pool_request()的迁移同样直观（当前实现见 fleet_cloud.py）：

def _pool_request(self) -> CreatePoolRequest: template_ref = SandboxTemplateRefBuilder().name(self._name).build() pool_spec = ( OsGymSandboxWarmPoolSpecBuilder() .replicas(self._replicas) .sandbox_template_ref(template_ref) .build() ) return CreatePoolRequestBuilder().namespace(self._name).spec(pool_spec).build()

当前版本进一步支持了自动扩缩容与创建 TTL 的条件追加（autoscaling、ttl_seconds_after_created），并有对应的传输层测试覆盖：test_pool_request_carries_the_requested_autoscaling、test_pool_request_carries_the_requested_creation_ttl、test_pool_request_leaves_creation_ttl_unset_by_default，以及test_transport_rejects_invalid_creation_ttl（校验 TTL 必须是0 ~ 2^32-1的整数，拒绝-1、True、"3600"、1.5、2**32）。

验证：请求字段不变

运行聚焦测试确认迁移无行为变化：

uv run --project libs/python/cua-sandbox pytest \ libs/python/cua-sandbox/tests/test_fleet_cloud_client.py \ libs/python/cua-sandbox/tests/test_fleet_cloud_transport.py -q

预期 PASS。例如test_registry_image_becomes_typed_template_request断言迁移后cpu_cores == 4、memory == "8192Mi"、镜像与服务列表完全符合预期——这正是"请求记录值不变"的机器证明。

Task 4：将沙箱测试夹具迁移到 Builder

传输层迁移完成后，轮到测试自身：把 tests/test_pool.py 中的pool_request()、template_request()夹具以及自定义 VM 夹具全部改为 Builder 构造，同时保持每个断言所依赖的记录身份与取值不变。

转换pool_request()夹具

def pool_request( *, name: str = "foo", template_name: str | None = None, replicas: int = 1, ) -> CreatePoolRequest: template_ref = SandboxTemplateRefBuilder().name(template_name or name).build() spec = ( OsGymSandboxWarmPoolSpecBuilder() .replicas(replicas) .sandbox_template_ref(template_ref) .build() ) return CreatePoolRequestBuilder().namespace(name).spec(spec).build()

转换template_request()夹具

def template_request( *, name: str = "foo", image: str = "example:latest", services: dict[str, int] | None = None, vm_template: VmTemplate | None = None, ) -> CreateTemplateRequest: if vm_template is None: built_services = [ SandboxServiceBuilder() .name(service_name) .target_port(port) .protocol(ServiceProtocol.TCP) .build() for service_name, port in (services or {"server": 8000}).items() ] vm_template = ( VmTemplateBuilder() .container_disk_image(image) .image_pull_secret("ecr-credentials") .services(built_services) .build() ) spec = OsGymSandboxTemplateSpecBuilder().vm_template(vm_template).build() return CreateTemplateRequestBuilder().namespace(name).name(name).spec(spec).build()

转换自定义 VM 夹具

对于需要完整自定义 VM 的场景（当前实现见 test_pool.py，对应计划中的自定义 VM fixture）：

service = ( SandboxServiceBuilder() .name("server") .target_port(8000) .protocol(ServiceProtocol.TCP) .build() ) vm_template = ( VmTemplateBuilder() .container_disk_image("registry.example/workspace:latest") .runtime(RuntimeKind.KUBEVIRT) .image_pull_secret("workspace-pull") .cpu_cores(10) .memory("20Gi") .firmware(Firmware.EFI) .services([service]) .build() ) request = template_request(vm_template=vm_template)

这里展示了 Builder 相比构造函数的显著优势：RuntimeKind.KUBEVIRT、Firmware.EFI等枚举值以具名 setter 传入，可读性远高于位置参数堆叠，且每个 setter 返回值仍是 Builder 本身，支持链式调用。

验证：契约 + 池测试全绿

uv run --project libs/python/cua-sandbox pytest \ libs/python/cua-sandbox/tests/test_fleet_builder_usage.py \ libs/python/cua-sandbox/tests/test_pool.py -q

预期 PASS，且契约测试不再报告任何直接构造。测试如test_template_reconcile_preserves_named_services（服务列表[("server", 8000), ("mcp", 3000)]）、test_reconcile_preserves_replicas_and_template_reference（replicas == 2、模板引用为workspace）等，从夹具迁移前后行为一致的角度完成了最终确认。

Task 5：验证打包与代码质量

迁移完成后，需要从「包能不能装、装完能不能用、代码干不干净」三个维度做最终验收。

发行与打包测试

uv run --project libs/python/cua-sandbox pytest \ libs/python/cua-sandbox/tests/test_fleet_sdk_distribution.py \ libs/python/cua-sandbox/tests/test_fleet_sdk_packaging.py -q

test_fleet_sdk_packaging.py 的断言非常严格，值得展开：

• pyproject.toml声明了精确版本的cua-fleet，且不直接依赖cua-train（训练栈与运行时沙箱解耦）；
• cua-fleet不在tool.uv.sources中（即不走本地路径源码，而是从https://wheels.cua.ai/simple私有索引解析发布包）；
• uv.lock中的cua-fleet版本与 registry source 与声明一致；
• 包不会把 checkout 里的绑定复制进 wheel（无hatch_build.py、无 hatch build hooks）；
• fleet_cloud.py、fleet.py、cyclops_http_client.py三个运行时文件都是直接from fleet_sdk import ...，没有sys.path注入或ctypes.CDLL加载本地库的旁路。

聚焦沙箱 Fleet 套件

uv run --project libs/python/cua-sandbox pytest \ libs/python/cua-sandbox/tests/test_fleet_builder_usage.py \ libs/python/cua-sandbox/tests/test_pool.py \ libs/python/cua-sandbox/tests/test_fleet_cloud_client.py \ libs/python/cua-sandbox/tests/test_fleet_cloud_transport.py \ libs/python/cua-sandbox/tests/test_fleet_transport.py -q

Ruff 代码质量

uv run --project libs/python/cua-sandbox --extra dev ruff check \ libs/python/cua-sandbox/cua_sandbox/__init__.py \ libs/python/cua-sandbox/cua_sandbox/transport/fleet_cloud.py \ libs/python/cua-sandbox/tests/test_fleet_builder_usage.py \ libs/python/cua-sandbox/tests/test_pool.py

发行版冒烟测试与最终 diff 审查

计划要求在libs/python/cua-sandbox的临时副本中执行：

uvx --from pdm==2.20.1 pdm lock uvx --from pdm==2.20.1 pdm build

随后在干净虚拟环境中用默认 PyPI 索引安装构建出的 wheel，验证fleet_sdk.CreatePoolRequestBuilder可以成功导入——这证明了最终用户不依赖私有索引也能解析到公开的cua-fleet发行版并拿到 Builder API。

最后审查 diff：

git diff --check && git diff -- \ libs/python/cua-sandbox/cua_sandbox/__init__.py \ libs/python/cua-sandbox/cua_sandbox/transport/fleet_cloud.py \ libs/python/cua-sandbox/tests/test_fleet_builder_usage.py \ libs/python/cua-sandbox/tests/test_pool.py

预期git diff --check退出码为 0，且 diff 只包含四类改动：builder 导出、builder 构造、源码契约、夹具迁移——没有夹带任何无关变更。

迁移全景回顾

这份实现计划展示了一条可复用的 SDK 演进路径，其核心方法论可归纳为四点：

1. 契约先行：任何重构开始前，先用 AST 契约测试把"不允许的写法"变成可执行的红线，让迁移范围有据可查、未来回归有门可挡。当前仓库的 test_fleet_builder_usage.py 甚至实现了完整的作用域链解析来对抗别名、遮蔽等真实代码形态。
2. 公开面不变：cua_sandbox.__init__同时保留记录类型与 Builder，对外 API 零破坏，CreateClaimRequest等无 Builder 的记录继续走构造函数。
3. 行为由测试锁定：传输层测试（test_fleet_cloud_transport.py）在迁移前后逐一断言请求字段取值（镜像、CPU、内存、服务列表、固件、副本数、TTL），确保"用 Builder 重构但不改变语义"不是口号而是机器验证。
4. 打包兜底：发行测试、打包测试与 PDM 冒烟测试共同保证「版本锁得住、wheel 装得上、Builder 导得出」。

对于任何希望在生成式 SDK 绑定之上构建业务库的团队，这套「先锁契约 → 再导出 → 迁移调用点 → 迁移夹具 → 打包验收」的流程，都是一份可以直接套用的工程模板。

【免费下载链接】cuaScale computer-use 2.0 with open-source drivers, cross-OS fleets, and benchmarks for training, evaluation, and data generation.项目地址: https://gitcode.com/GitHub_Trending/cua/cua