协议与接口

BlockX 的组件间协议只有三类：gRPC（控制面与结果面）、UDS 上的长度前缀 JSON 帧（Worker 与 Python Executor）、etcd 上的 JSON 心跳（Worker 注册与发现）。所有跨模块共享的协议定义都收敛在 api/ 目录；transport handler、存储模型和状态机不放在这里。整体位置见系统架构总览。 api/ 的约束来自 docs/specs/code-style.md §3.1：只放协议结构、消息包络和字段约定；结构体保持”数据对象”角色，允许少量无副作用的校验或编码辅助，但不依赖外部 IO。

调用方	被调方	传输	定义文件
Client	Coordinator	gRPC unary `ReserveWorkerSlot`	`api/grpc/coordinator/coordinator.proto`
Client	Bundle Coordinator	gRPC unary `ReserveWorkerSlot`（独立 service 身份）	`api/grpc/bundlecoordinator/bundle_coordinator.proto`
Coordinator	Worker	gRPC unary `RequestTaskSlot`	`api/grpc/worker/worker.proto`
Client	Worker	gRPC unary `SubmitTask` / `GetTaskResult`，server-streaming `WatchTasks`	`api/grpc/worker/worker.proto`
Client	Sync Invoker	gRPC unary `Invoke` / `DebugInvoke` / `Precheck`	`api/grpc/syncinvoker/sync_invoker.proto`
Worker、Sync Invoker	Python Executor	UDS，`uint32` 长度前缀 + JSON 包络（可选二进制 sidecar）	`api/uds/types.go`、`api/uds/codec.go`
Worker	etcd	lease 绑定的 JSON 心跳 key	`api/etcd/registry.go`、`api/etcd/types.go`
Worker、Sync Invoker	BlockDB / Meta / NodeRPC	gRPC / HTTP，客户端在 `internal/sdk/`	见后端适配器

gRPC 服务

四个 proto 文件都放在 api/grpc/<service>/，生成的 Go 代码放在同级的 *pb/ 目录（workerpb/、coordinatorpb/、bundlecoordinatorpb/、syncinvokerpb/）。生成命令：

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
make proto

make proto 定义在仓库根 Makefile，一次生成四个 service，并顺带触发 proto-blockdb、proto-meta、proto-localtestservice（这三个属于后端适配器，proto 在 internal/sdk/*/proto/）。改完 .proto 必须重新生成并把 *pb/ 一起提交。

`WorkerService`（`api/grpc/worker/worker.proto`）

service WorkerService {
  rpc RequestTaskSlot(RequestTaskSlotRequest) returns (RequestTaskSlotResponse);
  rpc SubmitTask(SubmitTaskRequest) returns (SubmitTaskResponse);
  rpc GetTaskResult(GetTaskResultRequest) returns (GetTaskResultResponse);
  rpc WatchTasks(WatchTasksRequest) returns (stream TaskUpdate);
}

RequestTaskSlot(task_id, ttl_ms) -> (slot_id, state)：Coordinator 调用；ttl_ms 只控制 ALLOCATED 的自动回收，不是执行超时。
SubmitTask(task: TaskInput, slot_id) -> state：slot_id 可空，为空时 Worker 内联做一次 RequestTaskSlot。TaskInput 字段：task_id、function_call_config{type, config bytes}、result_handler{type, config bytes}（可选）、task_timeout_ms（可选，0 表示 Worker 默认，负数或超过 Worker maxTaskTimeoutMs 都被拒绝）。两个 config 都是不透明 JSON blob，Worker 在入口校验它是合法 JSON（internal/worker/adapters/wire.go submitTaskFromPB）。
GetTaskResult(task_id) -> (task_id, state, result?)：result 只在终态出现。
WatchTasks(task_ids) -> stream TaskUpdate：先对每个 task_id 推一条当前快照，之后每个 task 终态推一条 is_terminal=true 的更新。Worker 不认识的 task_id 用 TaskUpdate.error 非空表达，不让整条 stream 失败。它同时是运行凭证：task 建立过 Watch 后，最后一条 stream 断开会启动 watchDisconnectGraceMs，超时则 task 以 WATCH_DISCONNECTED 失败。
state 字段是字符串，取值来自 internal/common/types/types.go 的 TaskState：ALLOCATED、RUNNING、SUCCEEDED、FAILED。
Client 走 Coordinator 拿到 worker_addr 后，SubmitTask / GetTaskResult / WatchTasks 都直连该 Worker，Coordinator 不中转结果。

metadata 约定：请求归属走 client-id（兼容 x-instance-id），Worker 在 SubmitTask 边界用 internal/common/clientid 归一化；shadow 转发用 x-blockx-shadow-submit: true 标记，接收端会剥掉 ResultHandler（internal/worker/adapters/shadow_forwarder.go）。

`CoordinatorService` 与 `BundleCoordinatorService`

service CoordinatorService {
  rpc ReserveWorkerSlot(ReserveWorkerSlotRequest) returns (ReserveWorkerSlotResponse);
}

ReserveWorkerSlot(task_id) -> (task_id, worker_addr, slot_id)。slot TTL 由 Coordinator 配置决定，请求里没有 chainId 或 reserveReqId。 bundle_coordinator.proto 定义了同名方法和字段完全相同的 bundlecoordinator.v1.BundleCoordinatorService，两份 message 互不引用。拆开的唯一目的是让上游 proxy 能按 method path（/coordinator.v1.CoordinatorService/ReserveWorkerSlot vs /bundlecoordinator.v1.BundleCoordinatorService/ReserveWorkerSlot）分流。internal/coordinator/adapters/bundle_grpc_server.go 的 BundleCoordinatorServer 只做 message 转换，然后委托给共享的 CoordinatorServer。详见 Bundle 集群。

`SyncInvokerService`（`api/grpc/syncinvoker/sync_invoker.proto`）

service SyncInvokerService {
  rpc Invoke(InvokeRequest) returns (InvokeResponse);
  rpc DebugInvoke(DebugInvokeRequest) returns (DebugInvokeResponse);
  rpc Precheck(PrecheckRequest) returns (PrecheckResponse);
}

一次 unary 请求执行一个 top-level call。请求用 oneof function_source { function_id | inline_source }（必须恰好一个），args_json 是 JSON 数组原始 bytes，deadline_unix_ms 是绝对 deadline。
响应携带 call_id、success、result_json 或 Failure{code, message, retryable, origin}、execution_duration_ms、queue_wait_ms。DebugInvoke 多一个 DebugOutput（stdout/stderr 尾部、error_stack、local_vars_json）。
FailureCode 与 FailureOrigin 是 proto 枚举，直接从 proto 读；执行阶段的失败作为响应数据返回（gRPC status 仍是 OK），只有准入层拒绝才是非 OK status，且 call_id 通过 trailer x-blockx-sync-call-id 回传（internal/syncinvoker/adapters/grpc_server.go）。
Precheck 只跑静态审计不执行；审计不通过是正常响应里的 pass=false + findings，非 OK status 保留给”给不出结论”（FAILED_PRECONDITION 审计关闭、UNAVAILABLE 审计器故障、INVALID_ARGUMENT 空源码）。

详见 Sync Invoker。

错误模型

BlockX 区分两层错误，规则来自 docs/specs/code-style.md §7：

准入 / 协议错误：参数不合法、slot 无效、无容量、服务未就绪。这些以 gRPC status 返回，不进入 TaskResult。
执行终态：task 已激活后的 Builder / Call / Plugin / 超时失败。这些收敛到 TaskResult，gRPC status 是 OK。

稳定业务码定义在 internal/common/types/types.go：

type ErrorCode string

const (
	ErrInvalidArgument  ErrorCode = "InvalidArgument"
	ErrInternal         ErrorCode = "Internal"
	ErrNoSlot           ErrorCode = "NoSlot"
	ErrUnavailable      ErrorCode = "Unavailable"
	ErrInvalidSlot      ErrorCode = "InvalidSlot"
	ErrNotFound         ErrorCode = "NotFound"
	ErrNoExecutableSlot ErrorCode = "NoExecutableSlot"
	ErrSlotUncertain    ErrorCode = "SlotUncertain"
)

在 gRPC 上，业务码只通过 gRPC status code 表达，没有 status details 也没有 trailer。Worker 侧映射在 internal/worker/adapters/wire.go workerCodeToGRPC，Coordinator 侧映射在 internal/coordinator/adapters/grpc_server.go coordCodeToGRPC，Coordinator 调 Worker 时的反向映射在 internal/coordinator/adapters/worker_rpc.go grpcErrToSlotError：

业务码	gRPC code	说明
`InvalidArgument`	`InvalidArgument`	参数错误，修正前不要重试
`NoSlot`、`NoExecutableSlot`	`ResourceExhausted`	无可用 slot；Coordinator 反向解析为 `NoSlot` 并按背压处理
`Unavailable`	`Unavailable`	未就绪或正在退出，未分配 slot
`InvalidSlot`	`FailedPrecondition`	`slotId` 不存在、过期或与 `taskId` 不匹配
`NotFound`	`NotFound`	task 不存在或结果已被清理
`SlotUncertain`	`Aborted`	已发起预占但无法确认，等一个 TTL 再重试
其他	`Internal`

worker.proto 顶部声明了 enum ErrorCode，但 Go 代码没有任何地方使用 workerpb.ErrorCode，也不附加 status details；调用方只能靠 status code 分支。若要恢复细粒度业务码，应通过 google.rpc.ErrorInfo details 挂上去。

UDS 协议（Worker ↔ Executor）

Worker 在 <SocketDir>/blockx-worker-<pid>.sock 上监听（internal/worker/adapters/executor/adapter.go NewAdapter），Executor 进程启动后主动连上来，靠首个 Heartbeat 完成附着。Sync Invoker 复用同一个 executor adapter 和同一套协议。字段语义见 api/uds/types.go 的注释和 docs/specs/worker-executor-connection-and-python-sdk-hook.md §4。

帧格式

api/uds/codec.go 定义了两种帧，最大 16 MiB（maxFrameSize）：

legacy 帧：   uint32(BE) 总长度 | JSON 对象
sidecar 帧：  uint32(BE) 总长度 | "BXB1" | uint32(BE) JSON 长度 | JSON 对象 | 原始字节

BXB1 是自描述 magic：没有 sidecar 的消息与旧格式逐字节相同；带 sidecar 的帧让 protobuf bytes（例如 BlockDB 请求 / 响应）跨 Worker/Executor 边界时不必 base64。JSON 对象是 MessageEnvelope：

type MessageEnvelope struct {
	ProtocolVersion string            `json:"protocolVersion"`
	MessageType     string            `json:"messageType"`
	ExecutorID      string            `json:"executorId"`
	CallID          string            `json:"callId,omitempty"`
	RequestID       string            `json:"requestId,omitempty"`
	Payload         json.RawMessage   `json:"payload"`
	TraceContext    map[string]string `json:"traceContext,omitempty"`
	BinaryPayload   []byte            `json:"-"`
}

ProtocolVersion 当前恒为 "v1"（internal/worker/adapters/executor/send.go）。TraceContext 装 W3C traceparent / tracestate，用于让 span 链跨过 UDS。

编解码入口

函数 / 类型	用途
`WriteFrame(w, env)`	编码并写一帧；`Payload` 原样拼接，不再校验是否合法 JSON，调用方保证
`ReadFrame(r)` / `ReadFrameInto(r, env)`	读一帧到 owned envelope
`BorrowedFrameReader.Read / Release`	零拷贝读帧，`Payload` 只在 `Release` 前有效；需要逃逸的数据用 `DecodePayloadOwned`
`EncodePayload / EncodePayloadInto`	payload 结构体转 `json.RawMessage`
`DecodePayload / DecodePayloadOwned`	`UseNumber` 语义解码，避免 uint256 精度丢失
`IsJSONNull(raw)`	判断 `RawMessage` 是否为空或 `null`

Go 侧用 bytedance/sonic 加速，wire 字节与 encoding/json 一致（api/uds/codec_test.go 的 TestWriteFrameStdlibCanReadSonicOutput 等用例钉住这一点）。

消息类型与 payload

api/uds/types.go 的常量：

const (
	// Worker -> Executor
	MsgExecuteCall  = "ExecuteCall"
	MsgResumeCall   = "ResumeCall"
	MsgCancelCall   = "CancelCall"
	MsgHeartbeatAck = "HeartbeatAck"

	// Executor -> Worker
	MsgCallWaiting   = "CallWaiting"
	MsgCallCompleted = "CallCompleted"
	MsgCallFailed    = "CallFailed"
	MsgHeartbeat     = "Heartbeat"
)

方向	消息	payload 类型	要点
W→E	`ExecuteCall`	`ExecuteCallPayload`	`taskId`、`functionId`、`entrySelector`、`functionCodeDigest` 或 `functionCode`、`args`、`callBudgetMs`、`attemptSeq`、`debug`、可选 `audit`（`AuditGatePayload`）
W→E	`ResumeCall`	`ResumeCallPayload`	`resumeKind`、`result` 或 `error`、`budgetUsedMs`；`NewResumeCallPayload` 会把实现了 `StructuredError()` 的错误展开成 map
W→E	`CancelCall`	`CancelCallPayload`	`reason`、`attemptSeq`
W→E	`HeartbeatAck`	无
E→W	`CallWaiting`	`CallWaitingPayload`	`waitKind` 为 `io` / `subcall` / `function_code`；`taskId` 必填；IO 用 `mode`、`operation`、`backend`、`cacheKey`、`request`（`RawMessage`）；subcall 用 `functionId`、`args`、`ancestry`、`grantedBudgetMs`
E→W	`CallCompleted`	`CallCompletedPayload`	`output`（`RawMessage`，全程不解码）、执行时长、debug 尾部
E→W	`CallFailed`	`CallFailedPayload`	`errorKind`、`detailCode`、`childOrigin`、`retryable`、`errorMessage`、`errorStack`、`localVars`
E→W	`Heartbeat`	`HeartbeatPayload`	队列计数、内存、CPU、`runningCallId`、`callStateCounts`、`lastSchedulerActiveAtMs`

Python 侧对应实现：python/blockx_executor/wire_keys.py（字段名与消息类型常量 MT_* / F_* / FP_*）、python/blockx_executor/messages.py（MessageEnvelope dataclass）、python/blockx_executor/uds_codec.py（encode_frame、FrameDecoder、dumps_wire / loads_wire，同样识别 BXB1）、python/blockx_executor/uds_session.py（UDSSession 单连接 reader/writer）。跨语言契约测试在 python/tests/test_executor_protocol_contract.py 和 python/tests/test_uds_codec.py。改 payload 字段时 Go 与 Python 两边都要动，见 Python Executor。

etcd 注册与心跳

api/etcd/registry.go 定义 registry prefix 常量与解析器：

const (
	LegacyBlockWorkerRegistryPrefix  = "/blockx/workers/"
	LegacyBundleWorkerRegistryPrefix = "/blockx/bundle-workers/"

	ProdDefaultBlockWorkerRegistryPrefix  = "/blockx/prod/lanes/default/workers/"
	ProdDefaultBundleWorkerRegistryPrefix = "/blockx/prod/lanes/default/bundle-workers/"
	TestDefaultBundleWorkerRegistryPrefix = "/blockx/test/lanes/default/bundle-workers/"
)

ParseRegistryPrefix(prefix) 只接受两类格式：legacy 两条固定路径（映射为隐式 prod/default），或 v2 的 /blockx/<env>/lanes/<lane>/{workers|bundle-workers}/。必须以 / 结尾，段只允许小写字母、数字和中间连字符。
Worker key 是 prefix + workerAddr（internal/worker/adapters/etcd_publisher.go key()），值是 WorkerHeartbeat 的 JSON，绑定到一个 lease（默认 TTL 10s），由 KeepAlive 续期；Deregister 撤销 lease 即删除 key。心跳默认每 200ms 发一次（HeartbeatIntervalMs，生产建议 2000ms）。
入口默认前缀：cmd/worker 用 /blockx/workers/，cmd/bundle_worker 用 /blockx/bundle-workers/；cmd/coordinator watch legacy block 前缀，cmd/bundle_coordinator 同时 watch legacy 与 prod v2 bundle 前缀，可用 COORDINATOR_REGISTRY_PREFIXES 覆盖。Coordinator 只 watch 精确前缀列表，不允许 watch /blockx/ 根。

心跳字段（api/etcd/types.go）：

type WorkerHeartbeat struct {
	WorkerAddr           string  `json:"workerAddr"`
	TaskSlots            int     `json:"taskSlots"`
	ReservedSlots        int     `json:"reservedSlots"`
	RunningTasks         int     `json:"runningTasks"`
	CPUPercent           float64 `json:"cpuPercent"`
	MemoryPercent        float64 `json:"memoryPercent"`
	MemoryAvailableBytes uint64  `json:"memoryAvailableBytes,omitempty"`
	MemoryTotalBytes     uint64  `json:"memoryTotalBytes,omitempty"`
	ProcessRSSBytes      uint64  `json:"processRssBytes,omitempty"`
	ErrorRate            float64 `json:"errorRate"`
	LastHeartbeatMs      int64   `json:"lastHeartbeatMs"`
}

Coordinator 侧 internal/coordinator/core/worker_view.go 用 type WorkerHeartbeat = etcd.WorkerHeartbeat 直接复用，internal/coordinator/adapters/etcd_watcher.go decodeWorkerHeartbeat 在 workerAddr 为空时回退到 key 去掉前缀。环境和 lane 不进入心跳字段，也不进入 protobuf。

任务结果模型

TaskResult 只表达 task 级结果，不含每个 call 的返回明细。proto 定义（worker.proto）：

message PluginResult {
  string plugin_name  = 1;
  bool   success      = 2;
  string failure_code = 3;
  bytes  result       = 4; // JSON-encoded plugin result payload
}

message ExecuteResult {
  string               failure_code    = 1;
  bool                 retryable       = 2;
  repeated PluginResult plugin_results = 3;
}

message TaskResult {
  bool          success        = 1;
  ExecuteResult execute_result = 2; // present only on terminal result
}

终态由外层 state 表达：SUCCEEDED 对应 success=true，FAILED 对应 success=false。TIMED_OUT 之类不是独立终态，而是 execute_result.failure_code。
代码里出现的 task 级 failure_code：BUILDER_FAILED、CALL_FAILED、PLUGIN_FAILED、TIMED_OUT、WATCH_DISCONNECTED、SLOT_LOST（internal/worker/core/worker.go、internal/worker/core/dispatcher.go）；adapter 侧还会用 ACTIVATION_FAILED、IO_SCOPE_FAILED、BUILDER_NOT_FOUND、PLUGIN_NOT_FOUND、CANCELLED、UNKNOWN_FAILURE（internal/worker/adapters/orchestrator_phases.go）。这些是字符串常量，没有集中枚举。
PluginResult.result 是 JSON bytes；ReturnValueResultHandler 把 call 输出数组放在这里回给 Client。
Go 内存形态在 internal/common/types/types.go（TaskResult / ExecuteResult / PluginResult / TaskUpdate），internal/worker/adapters/wire.go taskResultToPB 负责转 proto。

Go 类型 PluginResult 带 FailureMsg 和 Retryable 字段，但 worker.proto 的 PluginResult 没有对应 field，taskResultToPB 也不会传输它们。Client 拿不到 plugin 级失败消息，只有 failure_code。

修改协议的约定

来自 docs/specs/code-style.md §3.1 与 AGENTS.md：

优先新增可选字段或新增枚举值。不要静默改既有字段的含义、类型或错误码语义；proto 里不要复用已删除的 field number。
任何 api/ 变更都要在同一个 change 里同步对应 spec（worker.md、task-resource-coordinator.md、sync-invoker.md、worker-executor-connection-and-python-sdk-hook.md 等）。不要让代码里的结构体偷偷成为新的协议真相。
共享协议不散落在 adapter 内，统一收敛到 api/；api/ 里的类型不承载业务流程，也不依赖外部 IO。
改 UDS payload 时 Go（api/uds/types.go）和 Python（python/blockx_executor/wire_keys.py 及对应 payload 构造）必须一起改，并补 python/tests/test_executor_protocol_contract.py 用例。旧 peer 靠 JSON 宽松对象语义容忍新字段；BXB1 sidecar 帧旧 peer 无法消费，Worker 与 Executor 必须来自同一镜像。
改 gRPC 错误映射时同时改 wire.go 的正向映射与 worker_rpc.go 的反向映射，并更新 internal/worker/adapters/wire_test.go、internal/coordinator/adapters/worker_rpc_test.go。
改 registry prefix 规则时更新 api/etcd/registry_test.go 与 docs/specs/2026-07-28-registry-prefix-migration.md。
拆新的 gRPC service 身份（如 bundle coordinator）时，独立 proto、独立 message，不互相引用，让 method path 自己承担路由身份。
跨组件行为用 e2e/contract/（coordinator-worker、coordinator-etcd）和 e2e/system/ 的契约 / 系统测试兜底，见测试组织与命令。

开始使用

架构

组件

开发指南

gRPC 服务

`WorkerService`（`api/grpc/worker/worker.proto`）

`CoordinatorService` 与 `BundleCoordinatorService`

`SyncInvokerService`（`api/grpc/syncinvoker/sync_invoker.proto`）

错误模型

UDS 协议（Worker ↔ Executor）

帧格式

编解码入口

消息类型与 payload

etcd 注册与心跳

任务结果模型

修改协议的约定

相关文档

​gRPC 服务

​WorkerService（api/grpc/worker/worker.proto）

​CoordinatorService 与 BundleCoordinatorService

​SyncInvokerService（api/grpc/syncinvoker/sync_invoker.proto）

​错误模型

​UDS 协议（Worker ↔ Executor）

​帧格式

​编解码入口

​消息类型与 payload

​etcd 注册与心跳

​任务结果模型

​修改协议的约定

​相关文档

gRPC 服务

`WorkerService`（`api/grpc/worker/worker.proto`）

`CoordinatorService` 与 `BundleCoordinatorService`

`SyncInvokerService`（`api/grpc/syncinvoker/sync_invoker.proto`）

错误模型

UDS 协议（Worker ↔ Executor）

帧格式

编解码入口

消息类型与 payload

etcd 注册与心跳

任务结果模型

修改协议的约定

相关文档