> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Call 执行子系统
> Worker 内把 call list 调度到 Python Executor 池执行的子系统：DispatcherCore、执行 adapter、Executor Pool、子函数调用与 UDS 协议
Call 执行子系统位于 Worker 进程内部，负责把 Call Builder 产出的 `call list` 变成聚合后的 `outputs`。它是 [Worker](/components/worker) 的 Executor 阶段实现，在 [系统架构总览](/architecture/overview) 中对应 `4.2.4 Call 执行子系统`。Python Executor 进程本身见 [Python Executor](/components/python-executor)，本页只把它当作 UDS 对端。
## 职责与边界
负责：
* 用一个 Worker 全局的 `DispatcherCore` 调度多个 task 的 call，按 task / instance 公平放量。
* 为顶层 call 做 `Filter -> Score` 选 Executor；子函数调用固定绑回父 call 所在 Executor。
* 在 task 内做 call result cache 与 singleflight 去重。
* 处理 call 的挂起（IO / 子函数 / 源码拉取）与恢复，把 IO 请求转交 [IO 访问子系统](/components/io-subsystem)。
* 管理 Python Executor 进程池（spawn / 重启 / 沙箱），并维护与每个 Executor 的 UDS 长连接。
* 汇聚成功根 call 的输出为 `outputs []any`，交回 `WorkerCore` 进入 Plugin 阶段。
不负责：
* 生成 task、分配 slot、Worker 选址（归 [Coordinator](/components/coordinator) 与 Worker）。
* 函数源码快照与审计本身（归 [Function Code View](/components/function-code)），本页只消费 `TaskFunctionView.ResolveForDispatch`。
* 真实 BlockDB / RPC 访问（归 IO 子系统），Executor 不直连外部依赖。
核心不变量：
* `DispatcherCore` 是 Sans-IO 纯内存状态机，单写者：只有 dispatch actor goroutine（`Orchestrator.RunDispatchLoop`）能改它；其他 goroutine 只能投递 `dispatchEvent`。锁不变量由 `internal/worker/adapters/actor_invariants_test.go` 强制。
* 一个 call 整个生命周期只对应一个 `DispatchCallContext`；retry 复用同一 `callID`，只递增 `AttemptCount`。
* `DONE` 是唯一终态；`setCallTerminal` 是唯一记账点，重复调用幂等。
* 子函数调用有独立的 `DispatchCallContext`，`MaxAttempts = 1`，不进入 `TotalCount` / `outputs`，只把结果通过 `ResumeCall` 回送父 call。
* 子调用 ID 由进程级单调 `subcallSeq` 生成（`{parent}/sub-{n}`），worker 进程内全局唯一。
* Worker 侧没有 per-call 定时器：执行预算 `CallBudgetMs` 由 Executor 本地强制，墙钟外层保护是 task deadline 加 wedge backstop。
* Executor 终态帧到达即释放 adapter 预留（transport truth），与 dispatcher 的 `UnbindCall` 幂等。
## 代码位置
| 路径 | 用途 |
| ----------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- |
| `cmd/worker/main.go`、`internal/worker/app/app.go` | 装配：创建 `executor.Adapter`、`PoolManager`、启动 dispatch loop 与 500ms tick（快照同步、wedge 检查、心跳超时） |
| `internal/worker/core/dispatcher.go` | `DispatcherCore` 事件处理、`TryDispatchNext`、instance 公平调度、子调用注册/绑定 |
| `internal/worker/core/dispatcher_internal.go` | cache key、`selectExecutor`、retry/backoff、子树取消、`FlushDirtyTasks` 收敛 epilogue |
| `internal/worker/core/dispatcher_types.go` | `DispatchCallContext`、`TaskDispatchState`、`ExecutorSnapshot`、全部 `DispatchCommand` |
| `internal/worker/core/config.go` | `DispatcherConfig` 与默认值 |
| `internal/worker/adapters/orchestrator_dispatch.go` | `executeDispatchCommands`：把 core 命令翻译成 Bind / 解析代码 / UDS 下发；`HandleIO` |
| `internal/worker/adapters/orchestrator_subcall.go` | `HandleSubcall`、`pendingSubcall`、`sendSubcallResume` |
| `internal/worker/adapters/orchestrator_actor.go` | dispatch actor：`dispatchEvent` 类型与 `applyXxx` 处理器 |
| `internal/worker/adapters/orchestrator_wedge.go` | `CheckWedgedExecutors` 硬杀 backstop |
| `internal/worker/adapters/exec_adapter.go` | `ExecutorAdapter` 接口（orchestrator 对 UDS adapter 的依赖面） |
| `internal/worker/adapters/executor/adapter.go` | UDS listener、`executorConn`、预留（reservations / draining）、`Snapshots`、writer loop |
| `internal/worker/adapters/executor/handlers.go` | Executor→Worker 消息处理：Heartbeat / CallCompleted / CallFailed / CallWaiting |
| `internal/worker/adapters/executor/send.go`、`payload_buffer.go` | Worker→Executor 下发、outbox、payload 缓冲池 |
| `internal/worker/adapters/executor/function_code_source.go` | `waitKind=function_code` 的源码回送 |
| `internal/worker/adapters/executor/pool.go`、`spawner.go`、`sandbox_config.go` | Executor Pool Manager：slot 状态机、bare process / containerd sandbox 两种 spawn |
| `api/uds/types.go`、`api/uds/codec.go` | UDS 消息类型、payload 结构、帧编解码 |
| `internal/worker/core/dispatcher*_test.go`、`retry_backoff_test.go` | core 单测 |
| `internal/worker/adapters/orchestrator_dispatch_test.go`、`orchestrator_subcall_test.go`、`adapters/executor/*_test.go`、`api/uds/*_test.go` | adapter / 协议单测 |
| `cmd/worker/worker_process_*_test.go` | process E2E（真实 Python Executor） |
| `docs/specs/call-execution-subsystem.md` | 本子系统 spec（重点） |
| `docs/specs/worker-executor-connection-and-python-sdk-hook.md` | 连接模型与 UDS contract |
## 核心类型与接口
* `core.DispatcherCore`（`internal/worker/core/dispatcher.go`）：全局调度器。输入是 `HandleXxx` 事件方法，输出是 `[]DispatchCommand`。
* `core.DispatchCallContext`（`dispatcher_types.go`）：单个 call 的调度载体，含 `State`、`ExecutorID`、`AttemptCount`、`BudgetMs`、`IsSubcall`、`BatchIdx`、`ArgsJSON`。
* `core.TaskDispatchState`（`dispatcher_types.go`）：per-task 状态，含 `ReadyQueue`、`InflightCalls`、`CallCache`、`Singleflight`、`PendingBindSubcalls`、`ChildSubcalls`，以及流式生产字段（`ProductionOpen`、`BatchRemaining`、`PendingSettle`、`CollectedOutputs`）。
* `core.ExecutorSnapshot`（`dispatcher_types.go`）：adapter 提供的 Executor 容量视图，含 `Healthy`、`RunnableCount`、`UsableContexts`、`HeartbeatSeq`、`RunningCallID`、`LastSchedulerActiveAtMs`。
* `core.DispatchCommand`（`dispatcher_types.go`）：sealed 接口，实现有 `BindCall`、`UnbindCall`、`CancelCallCmd`、`DispatchCallCmd`、`ArmCallRetry`、`ConvergeTaskCalls`、`CompleteCallFromCache`、`BatchSettled`、`StopCallProduction`。
* `core.DispatcherCore.TryDispatchNext / tryDispatchNextFair`：一次放量决策；`selectExecutor` 做 `Filter -> Score`。
* `core.DispatcherCore.PrepareCallPhaseWithDigests + CommitCallPhase`、`CommitCallStream + PrepareCallBatchWithDigests + CommitCallBatch`：一次性与流式两种 call 注册入口；`Prepare*` 无副作用可 off-actor 运行。
* `core.DispatcherCore.FlushDirtyTasks`（`dispatcher_internal.go`）：收敛 epilogue，按 task 分组产出 `StopCallProduction -> 排空 -> BatchSettled -> ConvergeTaskCalls`。
* `adapters.ExecutorAdapter`（`exec_adapter.go`）：orchestrator 依赖的 UDS adapter 接口。
```go theme={null}
type ExecutorAdapter interface {
Bind(ctx context.Context, callID, executorID string) error
BindSubcall(ctx context.Context, callID, parentCallID, executorID string) error
Unbind(ctx context.Context, callID, executorID string)
MarkDraining(ctx context.Context, callID, executorID string)
IsDraining(callID, executorID string) bool
SendExecuteCall(ctx context.Context, executorID, callID string, payload uds.ExecuteCallPayload) error
SendResumeCall(ctx context.Context, executorID, callID, requestID string, payload uds.ResumeCallPayload) error
SendCancelCall(ctx context.Context, executorID, callID, reason string, attemptSeq int) error
Snapshots(ctx context.Context) ([]core.ExecutorSnapshot, error)
// SetCallbacks / SetIOHandler / SetSubcallHandler / SetFunctionCodeResolver / SetExecutorLostCallback
}
```
* `adapters.Orchestrator.executeDispatchCommands`（`orchestrator_dispatch.go`）：命令执行器。`Bind`/`Unbind` 在 actor 上同步执行；`SendExecuteCall`/`SendResumeCall`/`SendCancelCall` 一律 off-actor goroutine。
* `executor.IOHandler`、`executor.SubcallHandler`、`executor.FunctionCodeResolver`（`adapters/executor/adapter.go`）：adapter 收到 `CallWaiting` 后回调 orchestrator 的三个入口。
* `executor.Adapter`（`adapter.go`）：每个 Executor 一个 `executorConn`，持有 `reservations`、`draining`、`dispatchedAttempts`、`usableContexts`；`Snapshots()` 由此派生。
* `executor.PoolManager`、`PoolConfig`、`SlotState`（`pool.go`）：进程 slot 状态机 `RUNNING / WAITING / BACKOFF / STOPPED`；`executorSpawner` 抽象 bare process 与 sandbox。
* `uds.MessageEnvelope` 与 payload 结构（`api/uds/types.go`）：`ExecuteCallPayload`、`ResumeCallPayload`、`CancelCallPayload`、`CallWaitingPayload`、`CallCompletedPayload`、`CallFailedPayload`、`HeartbeatPayload`。
## 数据流 / 执行流程
交互模式：Executor reader goroutine / 定时器 / gRPC 把事实投递为 `dispatchEvent`（`evCallCompleted`、`evCallWaiting`、`evSubcallRequest`、`evExecutorSnapshot` 等）；actor 调 `DispatcherCore.HandleXxx` 得到命令；`executeDispatchCommands` 执行命令；每个事件结束都走 `flushDispatchEpilogue`。
```mermaid theme={null}
sequenceDiagram
participant WC as WorkerCore
participant O as Orchestrator actor
participant D as DispatcherCore
participant EA as executor.Adapter
participant EX as Python Executor
participant IO as TaskIOScope
WC->>O: StartCallPhase(taskId, callList)
O->>O: TaskFunctionView 解析各 call 的 digest
O->>D: PrepareCallPhaseWithDigests + CommitCallPhase
loop TryDispatchNext
D-->>O: BindCall(callId, executorId)
O->>EA: Bind (同步, 预留 context)
O->>D: HandleExecutorBound
D-->>O: DispatchCallCmd(functionId, digest, args, budgetMs)
O->>O: ResolveForDispatch(functionId, digest) 校验
O->>EA: SendExecuteCall (off-actor)
EA->>EX: ExecuteCall 帧
end
EX->>EA: CallWaiting(waitKind=io | subcall | function_code)
EA->>O: onCallWaiting -> HandleCallWaiting (RUNNING->WAITING)
alt waitKind=io
EA->>IO: HandleIO(taskId, callId, req)
IO-->>EA: result
else waitKind=subcall
EA->>O: HandleSubcall -> AddSubcallJSON + TryBindSubcallToParent
O->>EX: ExecuteCall(子 call, ParentCallID)
EX->>O: CallCompleted(子 call)
end
EA->>EX: ResumeCall(requestId, result, budgetUsedMs)
EA->>O: onCallResumed -> HandleCallResumed (WAITING->RUNNING)
EX->>EA: CallCompleted(output)
EA->>EA: releaseOnTerminal
EA->>O: onCallCompleted -> HandleCallCompleted
O->>D: FlushDirtyTasks
D-->>O: ConvergeTaskCalls(success, outputs)
O->>WC: HandleTaskPhaseFinished(PhaseCalls)
```
关键步骤说明：
`runCallPhase`（`orchestrator_phases.go`）在 task 激活的 goroutine 上按 `taskCodeEpoch` 解析每个 call 的 `functionCodeDigest`，调用 `PrepareCallPhaseWithDigests`，再投递 `evCallPhaseStarted` 让 actor `CommitCallPhase`。流式 builder 走 `CommitCallStream` + 逐批 `CommitCallBatch`，生产者受 `StreamBuildConfig.MaxOutstandingBatches` 信用窗口背压，`BatchSettled` 归还一个信用。
`TryDispatchNext` 从 `ReadyQueue` 队首取 call：先查 `CallCache`（命中直接 `CompleteCallFromCache`），再查 `Singleflight`（同 key 连续 call 一次性登记为 waiter），否则 `selectExecutor` 选 Executor、登记 singleflight leader、置 `RUNNING`、`AttemptCount++`、`BudgetMs = CallDeadlineMs`，输出 `BindCall`。
`selectExecutor` 过滤：`!Healthy`；调度器停摆超过 `SchedulerStallTimeoutMs`；`headroom = UsableContexts - localInflight <= 0`；补偿后 `runnable >= ExecutorMaxRunnable`；`localInflight >= ExecutorMaxInflight`；`MemoryBytes >= ExecutorMemoryHighWatermark`。打分：`runnable` 低优先，再 `localInflight` 低，再 `headroom` 高，最后 `ExecutorID` 字典序。`ExecutorSelectionSampleSize > 0` 时只扫前 N 个 eligible 候选（power-of-d 近似）。
`BindCall` 在 actor 上同步调 `Adapter.Bind`（子调用走 `BindSubcall`，父有活预留即放行），成功后 `HandleExecutorBound` 输出 `DispatchCallCmd`。adapter 用 `resolveFunctionCodeForDispatch` 按 `(functionID, digest)` 从 task pinned view 校验并取 `EntrySelector`，组装 `ExecuteCallPayload`（只带 digest，不带源码），在新 goroutine 里过 audit gate、做 alive / draining 检查后 `SendExecuteCall`。Bind 失败对根 call 输出 `ArmCallRetry(now)` 且不消耗 attempt 预算，子调用则重新 park 到 `PendingBindSubcalls`。
Executor 发 `CallWaiting`。`waitKind=function_code` 不改变 dispatcher 状态，adapter 直接用 `FunctionCodeResolver` 回源码。`io` / `subcall` 触发 `HandleCallWaiting`，adapter 在 `processWaitRequest` goroutine 里调 `IOHandler.HandleIO`（按 payload 回显的 `TaskID` 找 runtime，不查 dispatcher）或 `SubcallHandler.HandleSubcall`。结果经 `SendResumeCall` 阻塞到帧真正 flush 才触发 `HandleCallResumed`；发送失败则合成父 call 的 `resume_send_failed` 失败事件（retryable，带 attempt 序号）。
`CallCompleted` / `CallFailed` 到达 adapter 后先 `releaseOnTerminal` 释放预留，再回调 `applyCallCompleted` / `applyCallFailed`。core 写 `CallCache`、解决 singleflight waiter、取消该 call 的子调用子树，`markTaskDirty`。`flushDispatchEpilogue` 循环 `FlushDirtyTasks` 到静止：`TerminalRootCount == TotalCount`（流式还要 `ProductionDone`）时输出 `ConvergeTaskCalls`，adapter 翻译成 `WorkerCore.HandleTaskPhaseFinished(PhaseCalls)`。
### 子函数调用路径
1. Executor 内 SDK 发 `CallWaiting(waitKind=subcall, functionId, args, ancestry, grantedBudgetMs)`。
2. `HandleSubcall` 投递 `evSubcallRequest`，`applySubcallRequest` 在 actor 上校验：深度 `len(ancestry) < MaxSubcallDepth`、环检测（`CacheKeyForJSON` 与祖先 key 比对）、请求来源 executor 必须等于父 call 当前绑定的 executor（拒绝旧 attempt 的迟到请求）。
3. `resolveFunctionCodeForDispatch` 取子函数 digest，`AddSubcallJSON` 创建子 `DispatchCallContext`（继承父 `BatchIdx`，`BudgetMs = grantedBudgetMs`），记录 `pendingSubcalls[childID]`，`TryBindSubcallToParent` 绑回父 Executor；父未绑定或父 Executor 不健康时 park 到 `PendingBindSubcalls`，等 `HandleExecutorBound(parent)` 或 `HandleExecutorSnapshotUpdated` 重试。
4. 子 call 终态后 `sendSubcallResume` 把结果或 `subcallError`（含 `retryable`）经 `ResumeCall` 回送父 Executor，`BudgetUsedMs` 让父扣减预算。cache / singleflight 命中的子调用走 `CompleteCallFromCache` 直接 resume 父。
5. 父 call 终态（含 attempt 失败重试）时 `finalizeSubcallSubtree` 递归取消整棵子树，输出 `CancelCallCmd`；adapter 把预留标为 draining 并 best-effort `SendCancelCall`。
### Executor Pool 与连接
* `PoolManager.Start` 为每个 slot spawn 一个 `python -m blockx_executor --executor-id exec-N --socket-path ... --executor-max-inflight ... --executor-max-runnable ...`。进程退出后 `waitLoop` 按 `RestartDelayMs` 重启，`RestartWindowMs` 内超过 `MaxRestarts` 次进入 `BACKOFF` 等 `BackoffMs`。`KillExecutor` 供 wedge backstop 硬杀。
* `ExecutorSpawnMode=sandbox` 时 `sandboxSpawner` 通过 `github.com/Chaintable/emulator/sandbox` 起 containerd 容器：无网络、只挂载 UDS socket 目录、只透传 `childEnvKeys` 白名单环境变量、可选 CPU / 内存限额。设计见 blockx 仓库 `docs/specs/executor-sandbox-isolation.md`。
* Executor 主动连 `Adapter.SocketPath()`（`/blockx-worker-.sock`），首个有效 `Heartbeat` 完成 attach；没有显式注册消息。每条连接一个 reader goroutine 和一个 writer goroutine；writer 从 `outbox`（深度 256）取帧，一次最多 coalesce 64 帧后共享一次 `Flush`。
* `ExecuteCall` 非阻塞入队，outbox 满返回可重试错误；`ResumeCall` / `CancelCall` 阻塞到 flush 完成，因为 dispatcher 状态推进依赖它真正到达。
* 500ms tick：`SyncExecutorSnapshots` 把 `Adapter.Snapshots()` 喂给 `HandleExecutorSnapshotUpdated`；`CheckWedgedExecutors` 硬杀 `RunningCallID != ""` 且 `LastSchedulerActiveAtMs` 停摆超过 `SchedulerStallTimeoutMs` 的 Executor；`CheckHeartbeatTimeouts` 把超过 `HeartbeatTimeoutMs` 无心跳的连接判为 transport lost，触发 `onExecutorLost` → `applyExecutorLost` 先刷快照再对该 Executor 上全部 inflight call 发 retryable 失败。
### UDS 协议
帧格式：4 字节大端长度前缀 + JSON `MessageEnvelope`；带二进制 sidecar 时 body 以 `BXB1` 魔数 + 4 字节 JSON 长度开头，JSON 之后紧跟原始字节（BlockDB protobuf 走此路径，避免 base64）。单帧上限 16 MiB。编解码用 sonic，`json.RawMessage` 字段原样拼接。
| 方向 | `messageType` | payload | 说明 |
| --------------- | --------------- | ---------------------- | ----------------------------------------------------------------------------------------------------------------------- |
| Worker→Executor | `ExecuteCall` | `ExecuteCallPayload` | `TaskID`、`FunctionID`、`FunctionCodeDigest`、`EntrySelector`、`Args`、`ParentCallID`、`CallBudgetMs`、`AttemptSeq`、可选 `Audit` |
| Worker→Executor | `ResumeCall` | `ResumeCallPayload` | `ResumeKind`、`Result` 或 `Error`、`BudgetUsedMs`；`RequestID` 关联挂起 |
| Worker→Executor | `CancelCall` | `CancelCallPayload` | `Reason`、`AttemptSeq`，best-effort |
| Worker→Executor | `HeartbeatAck` | 无 | 非阻塞，丢失无害 |
| Executor→Worker | `CallWaiting` | `CallWaitingPayload` | `WaitKind = io / subcall / function_code`，必须回显 `TaskID`；IO 字段或子调用字段 |
| Executor→Worker | `CallCompleted` | `CallCompletedPayload` | `Output`（raw JSON）、`BudgetUsedMs`、耗时字段 |
| Executor→Worker | `CallFailed` | `CallFailedPayload` | `ErrorKind`、`Retryable`、`ErrorMessage`、`ErrorStack`、`LocalVars` 等 |
| Executor→Worker | `Heartbeat` | `HeartbeatPayload` | `RunnableCount`、`InflightCount`、`AvailableContexts`、`MemoryBytes`、`LastSchedulerActiveAtMs`、`RunningCallID`、CPU 指标 |
## 状态与生命周期
`DispatchCallContext.State` 只有四态（`CallReady / CallRunning / CallWaiting / CallDone`）。Executor 内部的 runnable / greenlet 细状态不上升到 Worker。
```mermaid theme={null}
stateDiagram-v2
[*] --> READY: addCall
READY --> RUNNING: BindCall 成功 (AttemptCount++)
READY --> DONE: cache hit / singleflight waiter 被 leader 解决 / 排空取消
RUNNING --> READY: bind 失败 (回滚 attempt) 或 retryable 失败且未耗尽 MaxAttempts
RUNNING --> WAITING: HandleCallWaiting (io / subcall)
WAITING --> RUNNING: HandleCallResumed
WAITING --> READY: 等待中的 attempt retryable 失败 (resume_send_failed / executor lost)
RUNNING --> DONE: CallCompleted 或最终失败
WAITING --> DONE: 最终失败或子树被取消
DONE --> [*]: removeCall (batch settle / RemoveTask)
```
* 重试与退避：`handleAttemptFailure` 在 `retryable && AttemptCount < MaxAttempts && !StopDispatching` 时置 `READY`，`computeBackoff` 按 `base * 2^(attempt-1)` 加最多 5% 抖动输出 `ArmCallRetry`；adapter 用 `time.AfterFunc` 到时投递 `evRetryReady`。base 优先取 task 的 `CallRetryPolicy.BaseBackoffMs`，否则 `DispatcherConfig.BaseBackoffMs`。
* 陈旧事件：`CallFailureStale` 丢弃 `READY && AttemptCount > 0`（退避中）或 `attemptSeq` 不匹配的失败事件；已 `DONE` 的 call 忽略一切迟到终态。
* 失败策略：`CallFailureFastFail` 下任一根 call 最终失败即 `markStopDispatching`，之后不再补发，`drainCancelledCalls` 在 inflight 归零后把余下 call 全部取消收敛。
* 预算：根 call 每次 attempt `BudgetMs = CallDeadlineMs`（默认 5000ms，0 无限），只计 CPU + IO 后端 + 子调用 + sleep，排队不计；子调用拿父切出的 `grantedBudgetMs`。
* 预留生命周期（adapter）：`Bind` 加入 `reservations`；正常终态 `Unbind`；取消路径 `MarkDraining`，等 Executor 终态帧 `releaseOnTerminal` 才释放；重连 era 清空全部预留。
* 输出保留：`MaxCollectedOutputBytes > 0` 且累计输出超限时 task 以 `OUTPUT_BYTES_EXCEEDED` 停产并丢弃输出。
## 配置
来自 `internal/worker/app/config.go`（`WorkerFullConfig`）与 `internal/worker/core/config.go`（`DispatcherConfig`）。环境变量覆盖见 `internal/worker/app/app.go`。
| 字段 | 默认值 | 说明 |
| ------------------------------------------------------------------------------------------ | ---------------------------------- | ------------------------------------------------------------------ |
| `ExecutorCount`（`EXECUTOR_COUNT`） | 8 | Executor 进程数（`PoolConfig.ExecutorCount`） |
| `ExecutorSpawnMode`（`EXECUTOR_SPAWN_MODE`） | `process` | `process` 或 `sandbox`；沙箱另需 `EXECUTOR_SANDBOX_IMAGE_REF` 等 |
| `Dispatcher.ExecutorMaxInflight` | 128 | 单 Executor 本地 inflight 上限，同时作为 `--executor-max-inflight` 传给 Python |
| `Dispatcher.ExecutorMaxRunnable` | 128 | 单 Executor runnable 阈值（心跳采样，含完成数补偿） |
| `Dispatcher.ExecutorSelectionSampleSize`（`EXECUTOR_SELECTION_SAMPLE_SIZE`） | 0 | 0 全扫；N>0 只对前 N 个 eligible 候选打分 |
| `Dispatcher.ExecutorMemoryHighWatermark` | 0 | Executor RSS 高水位过滤，0 关闭 |
| `Dispatcher.TaskMaxInflightCalls` | 1024 | task 内放量窗口 |
| `Dispatcher.CallDeadlineMs`（`CALL_DEADLINE_MS`） | 5000 | 根 call 每次 attempt 的执行预算（名字沿用，语义已是 budget） |
| `Dispatcher.SchedulerStallTimeoutMs`（`SCHEDULER_STALL_TIMEOUT_MS`） | 派生：`CallDeadlineMs + 5000` | wedge backstop 阈值；显式值若 `<= CallDeadlineMs + 1000` 会被抬到派生值 |
| `Dispatcher.MaxSubcallDepth` | 8 | 子调用最大深度 |
| `Dispatcher.BaseBackoffMs` | 10 | 重试退避基数（task 未指定时） |
| `Dispatcher.InstanceFairness`（`DISPATCH_INSTANCE_FAIRNESS`） | false | 开启 instance 级 max-min 公平放量 |
| `Dispatcher.InstanceFairnessQuantum`（`DISPATCH_INSTANCE_FAIRNESS_QUANTUM`） | 0 | 连续服务同一 instance 的 bind 数；0/1 等价逐 call 重选，上限 1024 |
| `Dispatcher.MaxCollectedOutputBytes`（`MAX_COLLECTED_OUTPUT_BYTES`） | 0 | 单 task 保留输出字节上限 |
| `Worker.DefaultMaxAttempts`（`CALL_RETRY_MAX_ATTEMPTS`） | 3 | call 内部重试次数上限 |
| `ExecutorRestartDelayMs / ExecutorMaxRestarts / ExecutorRestartWindow / ExecutorBackoffMs` | 1000 / 10 / 60000 / 30000 | Pool 重启节流 |
| `Admission.ExecutorTaskSlots`（`EXECUTOR_TASK_SLOTS`） | 32 | 同时处于 Executor 阶段的 task 数 |
| `StreamBuild.MaxOutstandingBatches / BatchMaxRows / BatchMaxArgsBytes` | 2 / 8192 / 32 MiB | 流式生产窗口与批大小（`StreamBuild.Enabled` 默认 false） |
| `executor.Config.HeartbeatTimeoutMs` | 10000（与 stall 耦合后为 `stall + 5000`） | 心跳超时判 transport lost |
## 扩展点
* 改调度策略：`selectExecutor` / `betterExecutorCandidate`（`dispatcher_internal.go`）改 Filter / Score；`tryDispatchNextFair`（`dispatcher.go`）改公平模型。跑 `dispatcher_select_bench_test.go`、`dispatcher_fairness_bench_test.go` 看回归。
* 新增 core 事件或命令：在 `dispatcher_types.go` 加 `DispatchCommand` 实现，在 `orchestrator_dispatch.go` 的 `executeDispatchCommands` 加分支；新事件在 `orchestrator_actor.go` 加 `evXxx` 类型和 `applyXxx`，并保证只在 actor 上改 `dispCore`（`actor_invariants_test.go` 会拦）。任何可能影响收敛的入口必须 `markTaskDirty`，收敛只在 `FlushDirtyTasks` 里发生。
* 新增 UDS 消息类型或 payload 字段：`api/uds/types.go` 加常量与结构，`adapters/executor/adapter.go` 的 `handleMessage` 加路由，`handlers.go` 加处理器；同步改 Python 侧协议与 `api/uds/codec_test.go`。
* 新增 `waitKind`：`handlers.go` 的 `handleCallWaiting` / `processWaitRequest` 加分支；决定它是否进入 dispatcher `WAITING` 状态（`function_code` 就不进入）。
* 新增 Executor spawn 方式：实现 `executorSpawner` 与 `executorProcess`（`spawner.go`），在 `newExecutorSpawner` 注册，`sandbox_config.go` 补校验。
* 调整 Executor 心跳字段：`uds.HeartbeatPayload` → `handlers.go` 的 `handleHeartbeat` / `heartbeatUpdateExisting` → `core.ExecutorSnapshot`。
## 测试
```bash theme={null}
# core 单测（纯内存，无 Python）
go test ./internal/worker/core/ -run 'Dispatcher|Backoff|Subcall' -count=1
# adapter / 协议单测
go test ./internal/worker/adapters/ -run 'Dispatch|Subcall|Actor' -count=1
go test ./internal/worker/adapters/executor/ ./api/uds/ -count=1
# 竞态检查（改 actor / adapter 边界必跑）
go test -race ./internal/worker/...
# process E2E：启动真实 worker + Python executor，需要 python/.venv
go test ./cmd/worker/ -run 'WorkerProcess_(Subcall|ExecutorLost|ExecutorPoolRestart|Wedge)' -count=1
```
* core 测试按主题分文件：`dispatcher_test.go`（基础调度）、`dispatcher_subcall_test.go`、`dispatcher_stream_test.go`（流式生产 / batch settle / 输出上限）、`dispatcher_instance_fairness_test.go` 与 `dispatcher_fairness_quantum_test.go`、`retry_backoff_test.go`、`dispatcher_stale_failure_test.go`、`dispatcher_wedge_test.go`。
* adapter 测试通过 `ExecutorAdapter` stub 注入发送失败等路径：`orchestrator_dispatch_test.go`、`orchestrator_subcall_test.go`、`orchestrator_send_failure_test.go`、`orchestrator_capacity_failure_test.go`。
* `adapters/executor/` 测试用真实 UDS socket 起 fake executor：`adapter_test.go`、`pool_test.go`、`resume_pool_test.go`、`write_coalesce_test.go`、`payload_buffer_test.go`。
* `api/uds/codec_test.go`、`bench_test.go` 覆盖帧编解码与二进制 sidecar。
* E2E 在 `cmd/worker/worker_process_*_test.go`，测试组织约定见 [测试组织与命令](/development/testing) 与 blockx 仓库 `docs/specs/worker-test-organization.md`。
## 相关文档
blockx 仓库 spec：
* `docs/specs/call-execution-subsystem.md`：本子系统设计（重点），含 Filter/Score、重试、子调用、流式生产。
* `docs/specs/worker-executor-connection-and-python-sdk-hook.md`：连接附着、失效判定、UDS 消息 contract、SDK hook。
* `docs/specs/architecture.md` §4.2.4：子系统在整体架构中的位置。
* `docs/specs/2026-07-23-execute-call-function-code-reference.md`：`ExecuteCall` 只带 digest、`waitKind=function_code` 按需取源码。
* `docs/specs/2026-07-30-python-executor-cpu-optimization.md`：结论性影响是 BlockDB 请求 / 响应走 `BXB1` 二进制 sidecar，Worker 用原始字节做 IO cache key。
* `docs/specs/executor-sandbox-isolation.md`：containerd 沙箱 spawn 模式。
站内相关页面：
* [Worker](/components/worker)、[Python Executor](/components/python-executor)、[Function Code View](/components/function-code)、[IO 访问子系统](/components/io-subsystem)、[Plugin 系统](/components/plugins)
* [Task 生命周期](/architecture/task-lifecycle)、[协议与接口](/architecture/protocols)