> ## 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.
# IO 访问子系统
> Worker 内统一承接 BlockDB / NodeRPC / 能力后端访问的 IO 层：WorkerIOScope、TaskIOScope、缓存、singleflight、自适应准入与 backend adapter 装配
IO 访问子系统是 Worker 进程内的一层资源治理代码。Builder、Executor 中的用户函数（经 Python SDK 和 UDS 回送）、Writer plugin 的所有 BlockDB / RPC / 能力后端访问都经过它，再由各 backend adapter 发起真实网络调用。它在系统中的位置见 [架构总览](/architecture/overview)。
它不按 core / adapter 的事件-命令模式组织，而是采用 **Scoped Resource Context**：Worker 级共享资源（`WorkerIOScope`）里嵌套 task 级作用域（`TaskIOScope`），请求生命周期嵌套在 task 生命周期里。
## 职责与边界
负责：
* 提供 `TaskIOScope.Read / Write` 这一个同步入口，供 Builder、Executor IO 回送、Plugin 共用。
* 请求路径固定为：Worker 级读缓存 → task 内 singleflight → task IO 窗口 → backend admission → adapter。
* 持有 Worker 级共享读缓存（IO cache）和独立的 system cache。
* 对带 Builder / Result retry scope 的请求执行有界重试。
* 把请求生命周期绑定到 task：scope 关闭后取消在途请求、拒绝新请求、丢弃迟到结果。
* 统一错误模型（`model.IOError`）与 per-task IO 统计。
不负责：
* 决定 call 是否被调度（那是 Dispatcher 的事，见 [Call 执行子系统](/components/call-execution)）。
* 真实协议调用、连接池、错误分类的具体逻辑——这些在 `internal/sdk/*` 的 adapter 里，见 [后端适配器](/components/backend-adapter)。
* call result cache（归 Call 执行子系统）。
* 跨 task 的写幂等或事务恢复。
核心不变量：
* 每个 backend kind 只有一个 admission controller，由 `assembly.Build` 强制包装（`adaptive.WrapBackend`）；`TaskIOScope` 前面不再叠第二层共享 quota。
* 写请求不进缓存、不做 singleflight，但同样经过 task 窗口和 backend admission。
* `req.Timeout()` 只约束获得 permit 之后的真实 adapter 调用；task 窗口和 admission 排队没有本地超时，只受请求 context 与 scope 生命周期约束。
* 空 `CacheKey()` 表示"不可缓存"，跳过缓存和 singleflight，直接走窗口 + adapter。
* `TaskIOScope.Close` 幂等；关闭后不得再回填缓存或 singleflight。
* IO cache key 由 `(backend, operation)` 命名空间 + 请求归一化 key 组成，不同后端 / 接口不会碰撞。
## 代码位置
| 路径 | 用途 |
| --------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------- |
| `internal/io/core/worker_scope.go` | `WorkerIOScope`：adapter 注册表、IO cache / system cache、活跃 task scope 表、cache 命名空间 |
| `internal/io/core/task_scope.go` | `TaskIOScope`：`Read / Write / Close`、singleflight leader 逻辑、迟到结果检查 |
| `internal/io/core/retry.go` | `RetryScope`、`WithRetryScope`、`runIORetry` 有界重试与退避 |
| `internal/io/core/task_scope_stats.go` 等 | `Snapshot / IOStats / BackendHits`、等待错误分类、`classifyBackendError` |
| `internal/io/core/model/` | `BackendAdapter`、`ReadReq / WriteReq`、`IOError`、`WorkerIOConfig`、`BackendRegistration`、结果与统计类型 |
| `internal/io/core/stats/` | 命中计数、per-call IO 时长、per-(backend, operation) CKMS 分位数、IO 错误 drain 聚类 |
| `internal/io/adaptive/` | 自适应并发 `Limiter`（窗口化 AIMD + 软上限探测 + 公平 lane）、`Backend` wrapper、`AIMDConfig`、传输层错误判定 |
| `internal/io/adaptive/observed/` | `WrapBackendRegistration`：包一层 adaptive 并注册 `blockx_io_adaptive_*` 指标 |
| `internal/io/assembly/` | `Module` 声明 + `Build`：从 module 列表构造 `[]BackendRegistration`，强制准入包装、stub 回退 |
| `internal/io/cache/` | 泛型 TTL + LRU + byte 记账缓存 `cache.Cache[K]` |
| `internal/io/sflight/` | 泛型 singleflight `sflight.Table[K]` |
| `internal/io/quota/` | task IO 窗口用的 channel 信号量 `quota.Semaphore` |
| `internal/io/capabilitywire/` | `DecodeBinaryRequest`：拆 gRPC 样式能力请求的 UDS 二进制信封 |
| `internal/worker/app/app.go`、`internal/worker/app/config.go` | Worker 侧 module 装配表、`DefaultIOConfig`、env override |
| `cmd/syncinvoker/io.go` | Sync Invoker 侧的 module 装配（禁用共享读缓存） |
| `internal/worker/adapters/orchestrator_dispatch.go` | `Orchestrator.HandleIO`：Executor IO 请求进入 `TaskIOScope` 的入口 |
| `internal/worker/adapters/orchestrator_phases.go` | task 激活时 `NewTaskScope`，终态时 `Close("task_terminal")`，Builder / Writer 阶段打 retry scope |
| `internal/sdk/localtestservice/` | 能力后端的活模板（adapter + service + proto） |
| `internal/io/**/*_test.go` | 单元测试与基准 |
| `docs/specs/io-subsystem.md` | 设计 spec（blockx 仓库） |
| `docs/io-backend-module-design.md`、`docs/capability-backend-guide.md` | backend module 设计与新增能力后端操作手册（blockx 仓库） |
## 核心类型与接口
* `model.BackendAdapter`（`internal/io/core/model/adapter.go`）：adapter 必须实现的最小接口。`Write` 返回 `(*WriteResult, error)`。
```go theme={null}
type BackendAdapter interface {
Read(ctx context.Context, req ReadReq) (*ReadResult, error)
Write(ctx context.Context, req WriteReq) (*WriteResult, error)
Close() error
}
```
* `model.ReadReq` / `model.WriteReq`（同文件）：请求接口。`ReadReq` 有 `Backend() BackendKind`、`Operation() string`、`CacheKey() string`、`Timeout() time.Duration`；`WriteReq` 没有 `CacheKey`。具体请求类型由各 SDK 包定义（如 `internal/sdk/blockdb/types.go`），Executor 回送的请求由 `executor.IORequest` 实现。
* `model.ErrorClassifier`：adapter 可选实现 `ClassifyError(err) *IOError`；没实现时未知错误一律归 `retryable_io`。
* `model.CachedReadValidator`：可缓存读的可选能力 `AcceptCachedRead(data []byte) bool`，返回 false 时旁路缓存值并进入同 key 的 singleflight 刷新（当前使用者是 `internal/sdk/iceberg/io_adapter.go`）。
* `model.IOError` / `model.IOErrorKind`：统一错误模型。`Kind` 取值 `system_error`、`param_error`、`retryable_io`、`non_retryable`、`timeout`、`scope_closed`；`Code` 是细粒度可观测码（如 `io:no_adapter_registered`、`context:timeout`）；`StructuredError()` 生成回给 Python SDK 的 `errorKind / retryable / detailCode`。
* `model.BackendRegistration`：`Kind`、`Adapter`、`SystemCache bool`（读走 system cache）、`ReadOnly bool`（`Write` 直接以 `io:write_not_supported` 拒绝）。
* `core.WorkerIOScope`（`worker_scope.go`）：`NewWorkerIOScope(WorkerIOScopeParams)`、`NewTaskScope(*commontypes.TaskCtx)`、`Close()`、`Snapshot()`。`WorkerIOScopeParams.DisableSharedReadCache` 供 Sync Invoker 关闭跨 call 结果复用。
* `core.TaskIOScope`（`task_scope.go`）：`Read`、`Write`、`Close(reason)`、`Snapshot()`、`IOStats()`、`BackendHits()`、`AddCallIO / FinalizeCallIO`、`BackendIOStats()`、`DrainIOErrorClusters()`。
* `core.RetryScope` / `core.WithRetryScope`（`retry.go`）：只有 `RetryScopeBuilder`、`RetryScopeResult` 两个框架标签能开启 IO-local retry；未知标签 fail-closed。
* `adaptive.Backend` / `adaptive.WrapBackend(next, Config)`（`adaptive/backend.go`）：给任意 adapter 外包一个独占 `Limiter`；同时实现 `ErrorClassifier`，把 admission 错误映射为 canonical 的 `context:timeout` / `context:canceled` / `io:temporarily_unavailable`。
* `adaptive.Limiter` / `adaptive.Permit` / `adaptive.Outcome`（`adaptive/limiter.go`）：`Acquire(ctx) (*Permit, error)`、`Permit.Done(outcome)`（幂等）；`Outcome` 三值 `OutcomeSuccess / OutcomeOverloaded / OutcomeIgnore`。
* `adaptive.OutcomeClassifier`：adapter 可选实现 `ClassifyAdaptiveOutcome(err) Outcome`；没实现时成功记 `Success`，错误记 `Ignore`。实现者见 `internal/sdk/noderpc/adaptive.go`、`internal/sdk/blockdb/adaptive.go`、`internal/sdk/logicaltypes/adaptive.go`、`internal/sdk/meta/adaptive.go`。
* `adaptive.AIMDConfig`（`adaptive/config.go`）：运维面的 AIMD 配置；`ResolveAdmission(initialLimit)` 在 `Enabled=false` 时退化为 `FixedConfig`。
* `assembly.Module` / `assembly.Build`（`assembly/assembly.go`）：见"扩展点"。
* `capabilitywire.DecodeBinaryRequest(kind, req) (method, payload, error)`：见"扩展点"。
## 数据流 / 执行流程
一次来自 Executor 的读请求（Python SDK → UDS → Worker → backend）：
```mermaid theme={null}
flowchart TD
A["Python SDK read_req / BridgeChannel"] -->|"UDS CallWaiting, waitKind=io"| B["executor.Adapter.processWaitRequest 构造 executor.IORequest"]
B --> C["Orchestrator.HandleIO 查 taskRuntime, 注入 clientid"]
C --> D["TaskIOScope.Read"]
D --> E{"scope closing?"}
E -->|"是"| Z1["返回 scope_closed"]
E -->|"否"| F{"CacheKey 为空?"}
F -->|"是, 不可缓存旁路"| L
F -->|"否"| G{"Worker IO cache 命中?"}
G -->|"是, 且 validator 接受"| Z2["返回 FromCache=true"]
G -->|"否"| H{"sflight.Join 是 leader?"}
H -->|"否, 作为 waiter"| I["waitForLeader 等 entry.Done / scopeCtx / ctx"]
H -->|"是"| L["runIORetry 发起一次 attempt"]
L --> M["quota.Semaphore.Acquire 获取 task IO 窗口"]
M --> N["adaptive.Backend.Read: Limiter.Acquire 拿 permit"]
N --> O["backendCallContext 套 req.Timeout, 调 raw adapter.Read"]
O --> P["permit.Done(outcome), 释放窗口"]
P --> Q{"结果"}
Q -->|"retryable_io 且带 retry scope"| L
Q -->|"其他错误"| Z3["completeSingleflight 广播错误"]
Q -->|"成功"| R{"scope 已关闭?"}
R -->|"是"| Z4["丢弃结果, 返回 scope_closed"]
R -->|"否"| S["cache.Set, completeSingleflight 广播数据"]
S --> T["ReadResult 经 ResumeCall 回 Executor, BudgetUsedMs 取 BackendLatencyMs"]
```
要点：
* 入口是 `internal/worker/adapters/executor/handlers.go` 的 `processWaitRequest`：按 `CallWaitingPayload` 的 `Mode / Operation / Backend / CacheKey / TimeoutMs / Request` 构造 `executor.IORequest`（带二进制 sidecar 时用 `NewIORequestWithBinary`，`DeriveCacheKeyFromBinary=true` 时以 sidecar 字节直接做 cache key），再调 `IOHandler.HandleIO`。
* `Orchestrator.HandleIO`（`orchestrator_dispatch.go`）按 `taskID` 找到 `taskRuntime`，把 instance id 写进 ctx（供公平 lane 用），Executor 来源的 `blockdb` 写在这里被拒绝，然后调 `rt.io.Read / Write`；成功后把 `BackendLatencyMs` 记进 per-call IO 统计。
* 回程：`ResumeCall.BudgetUsedMs` 取 `ReadResult.BudgetChargedMs()`——缓存 / singleflight 命中为 0，真实读为 adapter 测得的后端时长；admission 和窗口排队时间不计入 budget。请求经二进制 sidecar 到达的，结果 `Data` 也走 sidecar 回去（binary-in → binary-out）。
* Builder 与 Writer 不经 UDS，直接持有 `TaskIOScope`（Builder 拿的是 `cbtypes.TaskIOReader`，plugin 拿 `types.TaskIO`），ctx 分别带 `RetryScopeBuilder` / `RetryScopeResult`。
* 写路径与读路径的差别只在：跳过缓存与 singleflight、先检查 `ReadOnly`；其余（窗口、admission、重试、错误分类）相同。
### 缓存与命名空间
`WorkerIOScope` 持有两个 `cache.Cache[ioCacheKey]` 实例：
* IO cache：默认 4096 entries / 16 MiB，TTL 固定 1 分钟（`workerIOCacheTTL`），跨 task 共享，Worker 关闭时 `Purge`。
* system cache：默认 64 entries / 512 MiB / 20 分钟，只服务注册时声明 `SystemCache: true` 的 backend（当前是 `iceberg` 的 `resolve_data_files`）。
`ioCacheKey{namespace, request}` 中的 `namespace` 是 `(backend, operation)` 的数字 ID：BlockDB 两个 kind、`logicalTypes`、`meta`、`router`、`localtestservice`、`iceberg` 的已知 operation 在 `knownIOCacheNamespace` 里有编译期 ID；NodeRPC 的 JSON-RPC 方法是开放集合，与其他未知组合一起走有界（4096）动态注册表，注册表耗尽时该请求退化为不可缓存读。第三类缓存 call result cache 不在本子系统内。
### 自适应准入
`adaptive.Backend` 是每个 backend 唯一的进程级准入门。`Limiter` 实现窗口化 AIMD：
* 每个采样窗口（默认 1s）内首次 `Overloaded` 立即乘法下降（`BackoffRatio` 默认 0.5，下限 `MinLimit`）；同窗口再次下降要求新代际样本达到 `RepeatBackoffMinSamples` 且过载比例达到 `RepeatBackoffOverloadRatio`，且不超过 `MaxDecreasesPerWindow`。
* `Ignore` 比例达到 `IgnoreRatioThreshold`（0.10）时窗口失去增长资格，达到 `IgnoreBackoffRatioThreshold`（0.30）时窗口结算时收缩一次。
* 增长只在窗口结束时发生，要求无过载、样本够、饱和（最大 in-flight 达到 limit）、连续健康窗口达到 `IncreaseAfterHealthyWindows`；步长 `IncreaseStep`（0 表示 initial 的 1%，至少 1）。
* 超过已学习的 `softLimit` 时增长变成探测：探测代际过载则精确退回 `probeBaseLimit`，连续失败按窗口数指数退避（上限 `ProbeBackoffMaxWindows`）。
* limit 降到 in-flight 以下时不取消存量请求，用 shrink debt 偿还。
* 不起 goroutine，窗口由 `Acquire / Done` 事件惰性推进。
`Config.LaneKeyFromCtx` 非空时进入公平 lane 模式（`adaptive/lanes.go`）：等待者按 key 分道，空出的容量给 in-flight 最少的 lane；AIMD 策略不变。Worker 通过 `IOFairQueueBackends` 配置对指定 backend 开启，key 取 `clientid.FromContext`。
`observed.WrapBackendRegistration` 在包装的同时按 `BackendKind` 注册 `blockx_io_adaptive_limit / _pressure / _overloads_total / _ignored_total / _limit_transitions_total / _admission_waits_total / _admission_wait_seconds_total`。IO 层其余指标是 `blockx_task_io_ops_total`、`blockx_task_io_backend_duration_milliseconds`、`blockx_task_io_cache_hits_total`、`blockx_task_io_singleflight_hits_total`，定义在 `internal/obs/metrics.go` 与 `internal/obs/metrics_io_adaptive.go`。
## 状态与生命周期
`TaskIOScope` 只有"活跃"和"closing"两个状态：
* 创建：task 激活时 `Orchestrator` 调 `ioScope.NewTaskScope(ioTaskCtx)`（`orchestrator_phases.go`），失败记为 `task_activation_failed`。scope 持有派生自 task ctx 的 `scopeCtx`。
* 请求：每次 `Read / Write` 派生 `ioCtx`，`context.AfterFunc(scopeCtx, ioCancel)` 使 scope 关闭能取消在途 adapter 调用；每个 attempt 单独 `Acquire / Release` task 窗口，退避期间不占窗口和 permit。
* 关闭：task 终态时 `rt.io.Close("task_terminal")`。`Close` 用 `CompareAndSwap` 保证幂等，然后依次：取消 `scopeCtx`、`sfTable.DrainAll(scope_closed)` 唤醒所有 waiter、从 `WorkerIOScope.activeScopes` 移除。
* 迟到结果：leader 在 `runIORetry` 成功返回后再检查一次 `closing`，已关闭则返回 `scope_closed`，不写缓存也不广播；`sflight.Table.Complete` 与 `DrainAll` 在同一把锁下做幂等 close，避免 double-close panic。
* Worker 关闭：`WorkerIOScope.Close` 先关所有活跃 task scope，`Purge` 两个缓存，再逐个 `adapter.Close()`（adaptive wrapper 先关 limiter 唤醒 waiter，再关 raw adapter）。
错误分类与重试：
* 等待期错误由 `classifyWaitContextError` 归一：scope 已 closing → `scope_closed` / `io:temporarily_unavailable`；`DeadlineExceeded` → `timeout` / `context:timeout`；`Canceled` → `scope_closed` / `context:canceled`；其他 → `system_error`。没有独立的"排队超时"错误。
* backend 错误由 `classifyBackendError` 归一：`*IOError` 原样透传 → adapter 的 `ClassifyError` → 否则 `retryable_io`。
* `shouldRetryIO` 只在 `ioErr.Retryable()`、未用完 `RetryMaxRetries`、ctx 带已知 retry scope 且 ctx 未结束时放行。退避用 `backoff/v5`：默认三次窗口 250–500ms、500ms–1s、1–2s，单次上限 3s。
* 写重试是 at-least-once：IO core 只重放当前 typed write request。BlockDB 全部 API（含 BundleWrite 的 `InitWriteJob / CommitWriteJob / CancelWriteJob`）保证幂等，因此 Result scope 内的 BlockDB 写保持可重试；此前的 `WithoutRetryScope` 退出机制已在 commit `aae6b77a` 删除。接入不满足幂等契约的写 API 时必须先调整 retry scope 或 backend 契约。
* singleflight leader 拥有整个 retry 循环，waiter 只等最终结果；缓存只回填最终成功值。
## 配置
Worker 侧 IO 配置在 `WorkerFullConfig.IO`（类型 `iocore.WorkerIOConfig`，定义在 `internal/io/core/model/adapter.go`，`internal/io/core/types.go` 只做别名；默认值来自 `internal/worker/app/config.go` 的 `DefaultIOConfig`）：
| 字段（JSON） | 默认 | 说明 |
| ------------------------------------ | ------- | ------------------------------------------------------------------------------------------------------------ |
| `io.taskMaxInflightIO` | 1024 | 单 task 在途 IO 上限（task 窗口）。没有 env override |
| `io.retryMaxRetries` | 3 | 带 retry scope 的逻辑 IO 在首次 attempt 后最多追加的重试数；0 关闭。env `WORKER_IO_RETRY_MAX_RETRIES` |
| `io.workerIoCacheMaxEntries` | 4096 | IO cache entry 上限。env `WORKER_IO_CACHE_MAX_ENTRIES`（兼容 `TASK_IO_CACHE_MAX_ENTRIES`） |
| `io.workerIoCacheMaxBytes` | 16 MiB | IO cache 估算驻留字节上限。env `WORKER_IO_CACHE_MAX_BYTES`（兼容 `TASK_IO_CACHE_MAX_BYTES`） |
| `io.systemIoCacheMaxEntries` | 64 | system cache entry 上限。env `SYSTEM_IO_CACHE_MAX_ENTRIES` |
| `io.systemIoCacheMaxBytes` | 512 MiB | system cache 字节上限。env `SYSTEM_IO_CACHE_MAX_BYTES` |
| `io.systemIoCacheTtlMs` | 1200000 | system cache TTL；负数关闭过期。env `SYSTEM_IO_CACHE_TTL_MS` |
| `io.systemIoCacheRefreshConcurrency` | 2 | Iceberg 全表 planning 并发上限，同时是 `iceberg` backend 的固定 admission limit。env `SYSTEM_IO_CACHE_REFRESH_CONCURRENCY` |
backend admission 配置在 `WorkerFullConfig` 顶层：
| 字段（JSON） | 默认 | 说明 |
| -------------------------------------------------------------------------- | ------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `nodeRpcInitialLimit` / `blockdbInitialLimit` / `logicalTypesInitialLimit` | 1024 / 16 / 16 | 各 backend 的启动并发（不是静态 quota）。env `NODE_RPC_INITIAL_LIMIT`、`BLOCKDB_INITIAL_LIMIT`、`LOGICAL_TYPES_INITIAL_LIMIT`；`blockdb` 与 `blockdb_executor` 共用同一份 BlockDB 策略，各自独立 limiter |
| `nodeRpcAimd` / `blockdbAimd` / `logicalTypesAimd` | `adaptive.DefaultAIMDConfig()` | `enabled=true`、`minLimit=1`、`maxLimit=0`（不设上限）、`increaseAfterHealthyWindows=2`、`backoffRatio=0.5`、`windowMs=1000`、`minSamples=20`、`repeatBackoffOverloadRatio=0.30`、`repeatBackoffMinSamples=20`、`maxDecreasesPerWindow=2`、`ignoreRatioThreshold=0.10`、`ignoreBackoffRatioThreshold=0.30`、`probeBackoffMaxWindows=64`。env 前缀 `NODE_RPC_AIMD_*`、`BLOCKDB_AIMD_*`、`LOGICAL_TYPES_AIMD_*`（见 `applyAIMDEnvOverrides`） |
| `ioFairQueueBackends` | 空 | 开启公平 lane 的 backend kind 列表。env `IO_FAIR_QUEUE_BACKENDS` |
| `localTestServiceDisabled` / `routerDisabled` | false | 不入装配表，调用得到 `io:no_adapter_registered`。env `LOCAL_TEST_SERVICE_DISABLED`、`ROUTER_DISABLED` |
| `phaseIoTimeoutMs` | 10000 | Builder / Writer 阶段 IO 的 `req.Timeout()`，独立于 call budget。env `PHASE_IO_TIMEOUT_MS` |
Worker 与 Sync Invoker 的装配表注册的 backend kind 是 `rpc`、`blockdb`、`blockdb_executor`、`logicalTypes`、`iceberg`、`localtestservice`、`router`；`internal/sdk/meta` 定义了 `BackendMeta`，但两处装配表都没有注册它。
Sync Invoker（`cmd/syncinvoker/io.go`）只设 `TaskMaxInflightIO: 1024`，并以 `DisableSharedReadCache: true` 构造 `WorkerIOScope`。
## 扩展点
### 新增一个 capability backend
以 `internal/sdk/localtestservice/` 为模板（`docs/capability-backend-guide.md` 是逐文件手册），平台代码零改动。Go 侧需要：
在 `internal/sdk//` 下声明 `const Backend iocore.BackendKind = ""`（不得复用存量 kind），写 `proto/.proto` 作为 gRPC 服务定义，生成代码进 `gen/`。
`service.go` 实现生成的 server 接口，业务逻辑全在这里。`adapter.go` 实现 `model.BackendAdapter`：`Read` 用 `capabilitywire.DecodeBinaryRequest(string(Backend), req)` 拆出 `(grpc_method, protobuf 字节)`，`proto.Unmarshal` 后调 service，响应 `proto.Marshal` 进 `ReadResult.Data`；未知方法返回 `IOErrParam` + `:unsupported_method`，信封 / proto 不合形返回 `:bad_request`。只读能力的 `Write` 不会被调到（module 声明 `ReadOnly: true`）。外部依赖型后端应再实现 `model.ErrorClassifier` 和 `adaptive.OutcomeClassifier`。
导出 `Module()`（可带依赖入参）返回 `assembly.Module{Kind, ReadOnly, Build, Stub, Admission}`。进程内轻后端用 `adaptive.FixedConfig(n)`；外部依赖型后端用 `AIMDConfig.ResolveAdmission`，`Enabled` 判断配置是否齐全，`Stub` 给 dev 回退（不给则未配置时拒绝启动）。需要借用其他 backend 时按 `localtestservice.BlockDBReader` 的方式注入 raw adapter：借用不持有（`Close` 不关它）、准入自负、本地 ctx 结束的错误先原样放行再委托依赖分类。
`internal/worker/app/config.go` 加开关字段，`internal/worker/app/app.go` 加 env override 并 `backendModules = append(backendModules, .Module(...))`；`cmd/syncinvoker/io.go` 同样 append。`deploy/env/worker.env.example` 补一行。
`adapter_test.go` 覆盖：正常路径、参数默认值、未知 operation、malformed 信封、malformed proto、`Write` 拒绝、`Module()` 声明；组合后端再加 typed 请求转发、依赖未注入干净失败、依赖错误分类透传。e2e 在 `cmd/worker/` 照 `worker_process_localtestservice_test.go` 写正例（`FUNCTION_CODE_AUDIT_MODE=enforce` 下 task SUCCEEDED）和 `_DISABLED=true` 负例（task FAILED）；devstub 函数注册在 `internal/worker/app/function_code.go`。
Python 侧（blockx-py 类、`connection.py` channel getter、executor `BridgeChannel` 接线、`python/blockx_audit/tables.py` 五处表增量）见 [Python Executor](/components/python-executor) 和 `docs/capability-backend-guide.md` §3–§4。
`assembly.Module` 的字段：
```go theme={null}
type Module struct {
Kind iocoremodel.BackendKind
SystemCache bool
ReadOnly bool
Enabled func() bool
Build func(adm adaptive.Config) (iocoremodel.BackendAdapter, error)
Stub func() iocoremodel.BackendAdapter
Admission func() (adaptive.Config, error)
OuterWrap func(inner iocoremodel.BackendAdapter) (iocoremodel.BackendAdapter, error)
}
```
`OuterWrap` 是唯一允许放在 admission 外侧的层（当前只有 NodeRPC 语义合批 `noderpc.NewBatchingBackend`，由 `NODE_RPC_SEMANTIC_BATCH_MODE` 门控，默认 `off`）。
### 其他常见改动
* 给热点 `(backend, operation)` 加编译期 cache 命名空间：改 `worker_scope.go` 的 `knownIOCacheNamespace` 和对应枚举，并更新 `TestKnownIOCacheNamespacesAreDistinctEnums`。
* 调整某个 backend 的 AIMD 反馈：改该 SDK 包的 `adaptive.go`（`ClassifyAdaptiveOutcome`），不要动 `internal/io/adaptive`；判断传输层错误用 `adaptive.IsTransportFailure / IsTransportFailureMessage`。
* 新增可缓存但需要请求级校验的读：让请求类型实现 `model.CachedReadValidator`。
## 测试
```bash theme={null}
go test ./internal/io/...
go test ./internal/io/core/ -run 'TestTaskScope|TestWorkerScope'
go test ./internal/io/adaptive/ -run 'TestLimiter|TestFairLane|TestBackend'
go test ./internal/sdk/localtestservice/ ./internal/io/...
go test -count=1 -run TestWorkerProcess_LocalTestService ./cmd/worker/
```
测试组织：
* `internal/io/core/core_test.go`：请求路径主干——缓存命中 / 跨 task 共享（`TestWorkerScope_CacheIsSharedAcrossTaskLifetimes`）、singleflight（`TestTaskScope_ReadSingleflight`）、task 窗口获取 / 释放（`TestTaskScope_TaskWindowWaitsUntilCapacityIsReleased`、`TestTaskScope_IOBackendAttemptReleasesTaskWindow`）、scope 关闭与迟到结果（`TestTaskScope_ActiveBackendContextLifecycle`、`TestTaskScope_ReadAfterClose`、`TestTaskScope_CloseIdempotent`、`TestWorkerScope_CloseCascades`）、只读后端拒写、空 cacheKey 旁路（`task_scope_bypass_test.go`）。
* `internal/io/core/retry_test.go`：retry scope 门控、单个 leader 拥有重试、attempt 间释放资源、退避边界。
* `internal/io/core/task_scope_drain_test.go`：`Complete` 与 `DrainAll` 并发不 double-close。
* `internal/io/adaptive/limiter_test.go`、`lanes_test.go`、`backend_test.go`、`backend_budget_test.go`：AIMD 各条规则、探测、公平 lane、admission 排队不计入 budget、超时在 admission 后才开始。
* `internal/io/assembly/assembly_test.go`：stub 回退、重复 kind、失败时关闭已建 adapter、`OuterWrap`。
* `internal/io/capabilitywire/wire_test.go`：信封三方绑定、shape 错误 fail-closed。
* `internal/io/sflight/sflight_test.go`、`internal/io/cache/cache_test.go`：singleflight 与缓存的独立单测。
* 基准：`internal/io/core/bench_test.go`、`adaptive/lanes_bench_test.go`。
* 与 Executor 联动的进程级 e2e 在 `cmd/worker/worker_process_*_test.go`（如 `worker_process_blockdb_bridge_test.go`、`worker_process_localtestservice_test.go`）。更多见 [测试组织与命令](/development/testing)。
## 相关文档
blockx 仓库中的 spec 与设计文档：
* `docs/specs/io-subsystem.md`：本子系统的行为规范（Scoped Resource Context、adapter 职责、AIMD 反馈分类、Close 语义、缓存策略与 fork 风险、配置项）。
* `docs/io-backend-module-design.md`：backend module 装配层的设计动机与不变量。
* `docs/capability-backend-guide.md`：新增能力后端的逐文件操作手册与检查单。
* `docs/specs/architecture.md` §4.2.5 / §4.2.6：IO 访问子系统与共享状态缓存在总体架构中的定位。
* `docs/specs/noderpc-semantic-batching.md`：NodeRPC 语义合批（`OuterWrap` 的使用者）。
* `docs/deploy.md`：IO / AIMD 相关环境变量的部署说明。
站内相关页面：
WorkerIOScope 的宿主，task scope 的创建与关闭时机。
BlockDB / NodeRPC / logical-types / Iceberg 等 adapter 的具体实现与错误分类。
IO 请求在 Python 侧如何被拦截并经 UDS 回送。
HandleIO 所在的 Orchestrator 与 call result cache。