v1.1.0Pi v1.1.0 源码分析Analysis of Pi v1.1.0 · 2026-10-07

clientclient

基于分帧 CBOR 的远程 pi 会话客户端Transport-neutral client for remote pi sessions over framed CBOR bytes

Run modesInfraRun modes

← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言← → navigate · F fullscreen · O overview · T theme · L language

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.001
01 · 一个 Client、一个 Connection、一个可替换的传输01 · One Client, one Connection, a pluggable transportRun modes

一个 Client、一个 Connection、一个可替换的传输One Client, one Connection, a pluggable transport

pi-client: 1,135 lines — one Client, one Connection, a pluggable byte transportpackages/client/srcunix.ts → 子路径 "./unix"（仅 Node / Bun）unix.ts → subpath "./unix" (Node / Bun only)依赖与使用者Dependencies and users12index.ts根入口 "."：Client、createClientServiceTransport、3 个错误类、传输与选项类型root entry ".": Client, createClientServiceTransport, 3 error classes, transport and option types479client.tsClient：请求关联、订阅、附着Client: correlation, subscriptions, attachmentcreateClientServiceTransport245connection.tsConnection：状态机、握手、解码Connection: state machine, handshakeServerMessageDecoder34errors.ts3 个错误类3 error classes32types.ts选项与订阅类型options, subscriptions18transport.tsByteTransport 契约ByteTransport contract16promise.tsPromise 解析器promise resolverscreateUnixTransportFactoryUnixByteTransport · maxPendingBytes299discoverUnixServersreaddir → lstat → 握手探测readdir → lstat → handshake probe@earendil-works/chord ^1.1.0服务调用解析、订阅状态解码、RemoteServiceTransportservice-call parsers, state decoder, RemoteServiceTransport@earendil-works/pi-protocol ^1.1.0encodeClientMessage · ServerMessageDecoder · isServerIdclient-runtime.ts发现 Unix 服务器，Client.connectdiscover Unix servers, Client.connectservices/connection.tscreateClientServiceTransport → 类型化服务createClientServiceTransport → typed servicesserver.ts确认刚启动的服务器能握手check a freshly started server handshakesradius-relay.ts自定义 ByteTransport：WebSocket 中继a custom ByteTransport over a WebSocket relay依赖Depends on使用者：coding-agent 的 experimental/（5 个文件导入它）Used by: coding-agent experimental/ (5 importing files)README：断开或 dispose 后，挂起的请求在本地 reject；Client 不会自动重连，也不会重放请求。README: on disconnect or disposal, pending requests reject locally; the Client never reconnects or replays requests automatically.标签 = 行数tag = lines核心coreUnix 传输 / 使用方Unix transport / users依赖dependencies入口entry使用uses

根入口不碰 Node API：浏览器或 worker 也可以用，只要提供自己的传输（例如 WebSocket）。The root entry touches no Node APIs: a browser or worker can use it with its own transport (a WebSocket, say).

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.002
02 · 连接的三个状态02 · Three connection statesRun modes

连接的三个状态Three connection states

Connection in three states: every connect() is a fresh attemptConnectionStateconnection.ts握手阶段的失败Handshake failures进入 disconnected 时，Client 会Entering disconnected, the Clientdisconnectedconnectingconnectedhello_errorfirst ≠ helloserverId ≠data before hellofactory throwsconnect()：新的 id、新解码器connect(): new id, new decoder收到匹配的 hellomatching hello关闭 · 出错 · disconnect()close · error · disconnect()握手失败（见右侧）handshake failure (right)connecting 期间While connectingtransportFactory(handlers)打开一个全新的 ByteTransportopen a fresh ByteTransportsend hello { version }第一帧总是客户端的 hellothe first frame is always the client hellodecoder.push(chunk)ServerMessageDecoder 增量解码ServerMessageDecoder decodes incrementallyhello.serverId === options.serverId物理端点必须是预期的逻辑服务器the endpoint must be the expected logical server旧尝试的回调带着旧 id：#isCurrent(id) 为假就被忽略；迟到的 transport 直接 close()。Callbacks from an old attempt carry its id: #isCurrent(id) is false, so they are ignored; a late transport is closed.非 disconnected 时再 connect() → DisconnectedError("Client is already …")。connect() while not disconnected → DisconnectedError("Client is already …").→ ServerError(code)，例如 version→ ServerError(code), e.g. versionExpected server hello as first messageConnected server … does not match …客户端 hello 发出前就收到数据data before the client hello was sent→ DisconnectedError(cause)清空 hello 与 attachmentclear hello and attachment以该错误 reject 所有挂起请求reject every pending request with that error丢弃全部服务订阅drop every service subscription通知 onConnectionStateChangenotify onConnectionStateChangeconnected 之后再收到 hello / hello_error 也是协议错误；解码器在关闭时 end()，半帧会变成截断错误。A hello / hello_error after connecting is a protocol error too; on close the decoder end()s, so a partial frame becomes a truncation error.12341234前进advance失败 / 关闭failure / close握手错误handshake errors清理cleanup

旧尝试的回调带着旧 id，被 #isCurrent(id) 过滤掉；工厂迟到返回的传输会被直接 close()。Callbacks from an old attempt carry its id and are filtered by #isCurrent(id); a transport the factory returns late is simply closed.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.003
03 · 一个请求的一生03 · The life of one requestRun modes

一个请求的一生The life of one request

One request: guards, then register, encode, send; settled by id#request(target, call, signal)client.ts#pendingRequests：按 id 关联#pendingRequests: correlated by idMap<string, PendingRequest>一个回应或事件到达之后When a response or event arrivesClientDisposedErrorDisconnectedErrorAbortErrorencode errorok: trueok: falseabort()no matchtransform throwsdisconnectedif (this.#disposed)if (!this.connected)if (signal?.aborted)直接拒绝，什么也不发送reject at once, send nothingid = `request-${++seq}`#pendingRequests.set(id, { resolve, reject, cleanup })encodeClientMessage(request)parseServiceCall + maxFrameLength；失败则取回条目并拒绝parseServiceCall + maxFrameLength; on failure take back and rejectconnection.send(frame); sent = true发送前就已 abort → 发送后立刻补发 cancelaborted before the send → the cancel follows at once前三个守卫失败时没有 id、没有条目，也没有任何字节离开客户端。When one of the first three guards fails there is no id, no entry, and no byte leaves the client."request-1" → { resolve, reject, cleanup }"request-2" → { resolve, reject, cleanup }// 回应可以乱序到达；取出时 cleanup 移除 abort 监听// responses may arrive out of order; cleanup removes the abort listenerresolve(result)；订阅时先经 transform 解码快照resolve(result); a subscription decodes its snapshot in transformreject(new ServerError(error))：带 code 与 messagereject(new ServerError(error)): carries code and message本地立刻 reject；已发送则补发 cancel { id, target }reject locally now; if sent, send cancel { id, target }Response has no matching request → 断开连接Response has no matching request → disconnectProtocolValidationError → 断开连接ProtocolValidationError → disconnect所有条目一次性以断开原因 rejectevery entry rejects with the disconnect reasonabort 之后条目仍留在表里，直到服务端的 cancelled 回应到来——所以那条回应不会被当成“没有匹配的请求”。After abort the entry stays in the map until the server's cancelled response arrives, so that response is never a "no matching request".123456请求路径request path成功success失败failure关联表correlation map

回应可以乱序到达；没有对应条目的回应是协议错误，会断开连接。Responses may arrive out of order; a response with no matching entry is a protocol error and drops the connection.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.004
04 · 订阅：先快照，后更新04 · Subscriptions: snapshot first, then updatesRun modes

订阅：先快照，后更新Subscriptions: snapshot first, then updates

  • 每个订阅有自己的 Chord 状态解码器（createServiceStateDecoder()）：快照和更新都经过它。Every subscription has its own Chord state decoder (createServiceStateDecoder()); snapshots and updates both go through it.
  • dispose() 只在仍连接且路由仍是当前路由时发退订；随后等待 deliveryTail 并清空缓冲。dispose() sends the unsubscribe only while connected and while the route is current; then it awaits deliveryTail and clears the buffers.
One service subscription: updates buffer before the snapshot and flow in order only after start()Serverpi-serverClient#serviceListeners应用 / 绑定App / bindingChord 服务绑定Chord service bindingsubscribeService(target, serviceId, mode, listener)service-N · hydrated = false · ready = falserequest { call: subscribe(service-N, serviceId, mode) }service_update（早于快照）service_update (before the snapshot)原始 JSON → queuedWireUpdatesraw JSON → queuedWireUpdatesresponse { ok: true, result: snapshot }decodeSnapshot → hydrated = true → 解码积压decodeSnapshot → hydrated = true → decode backlog{ id, target, snapshot, start, dispose }应用先装好快照the app installs the snapshot firstservice_updateready = false → queuedstart()ready = true，按序冲刷 queuedready = true, flush queued in orderservice_update已就绪：解码后直接交付ready: decode and deliverlistener(update)经 deliveryTail 串行，异步监听器也不乱序chained on deliveryTail: async listeners stay ordereddispose()已连接且 target 仍有效时才退订unsubscribe only while connected and the target is currentrequest { call: unsubscribe(service-N) }然后 await deliveryTail，清空缓冲then await deliveryTail and clear the buffers监听器抛出的错误只交给 onListenerError，不会影响连接；断开时 #serviceListeners 被整体清空，订阅不会自动恢复。Listener errors only reach onListenerError and never affect the connection; on disconnect #serviceListeners is cleared and subscriptions are not restored.调用 / 请求call / request回应 / 更新response / update
图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.005
05 · 附着与 Chord 传输适配器05 · Attachments and the Chord transport adapterRun modes

附着与 Chord 传输适配器Attachments and the Chord transport adapter

The server pushes attachments out of band; createClientServiceTransport resolves the route on every call带外附着：Client.attachmentOut-of-band attachment: Client.attachmentcreateClientServiceTransport(client, getTarget)client.tsservices/connection.tscoding-agentattachment 信封attachment envelope{ attachment: SessionTarget | null }serverId 检查serverId check属于别的服务器 → 断开连接another server → disconnect#setAttachment三个字段都相同 → 忽略all three fields equal → ignoredonAttachmentChange通知监听器；undefined = 未附着notify; undefined = not attached服务器级服务Server-wide services() => ({ serverId: client.serverId })Session 服务Session services() => client.attachmentnull、断开或 dispose → attachment 回到 undefined；附着信封在 attach 的回应之前送达，所以回应到来时路由已可用。null, a disconnect or dispose → attachment returns to undefined; the envelope arrives before the attach response, so the route is ready when it resolves.// 返回 Chord 的 RemoteServiceTransport// returns a Chord RemoteServiceTransportinvoke(call, context)→ client.request(target(), call, context.abortSignal)subscribe(serviceId, mode, listener, context)→ client.subscribeService(target(), serviceId, mode, …)→ { snapshot, activate: start(), close: dispose() }// 更新以 BACKGROUND_CONTEXT 交给 Chord 监听器// updates reach the Chord listener with BACKGROUND_CONTEXT// target() 在每次调用时求值；undefined →// target() is evaluated per call; undefined →Error("Remote service target is unavailable")只做适配：类型化服务代理与应用契约由应用自己的 Chord 绑定提供。An adapter only: typed service proxies and application contracts come from the app's own Chord bindings.client.attachmentdispose() 只在 #targetIsCurrent 时退订：dispose() unsubscribes only while #targetIsCurrent:{ serverId } → 与 hello.serverId 相同{ serverId } → equals hello.serverIdSession 目标 → 三个字段都等于当前附着Session target → all three equal the attachment过期的附着不再发退订：服务器早已丢弃它。A stale attachment sends no unsubscribe: the server dropped it.附着状态attachment state适配器adaptercoding-agent读取路由reads the route

附着信封先于 attach 的回应送达：回应 resolve 时 client.attachment 已经可用。The attachment envelope arrives before the attach response: when it resolves, client.attachment is already set.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.006
06 · 字节传输契约与 Unix 实现06 · The byte-transport contract and the Unix implementationRun modes

字节传输契约与 Unix 实现The byte-transport contract and the Unix implementation

The byte transport is pluggable: the contract is send / close and three callbacks契约The contracttransport.tsUnixByteTransport.send(chunk)unix.ts另一个实现Another implementationcoding-agent radius-relay.ts检查ChecksUint8Array？未关闭？pending + 长度 ≤ maxPendingBytes？a Uint8Array? not closed? pending + length ≤ maxPendingBytes?登记并复制Count and copy#pendingBytes += 长度 · chunk.slice()#pendingBytes += length · chunk.slice()排进写队列Chain on the write queue#writeTail.then(() => #write(bytes))socket.write(bytes, callback)返回 false → 还要等 "drain"returned false → also wait for "drain"完成Settlefinally：#pendingBytes −= 长度finally: #pendingBytes −= lengthRadiusClientByteTransport同一契约，跑在 WebSocket 中继上；OrderedWebSocketWriter 保序same contract over a WebSocket relay; OrderedWebSocketWriter keeps orderinterface ByteTransport {send(chunk): Promise<void>close(): void}interface ByteTransportHandlers {onData(chunk) · onClose() · onError(error)}type ByteTransportFactory =(handlers) => ByteTransport | Promise<…>send 必须按调用顺序送达，并遵守背压send must deliver in call order and honor backpressureclose() 可重复调用close() must be safe to repeat恰好一个终止回调：onClose 或 onErrorexactly one terminal callback: onClose or onError每次连接尝试都由工厂新建并完成认证the factory builds and authenticates one per attempt超限 → rejectover limit → reject写入中 socket 关闭socket closed mid-writewrite 回调带错误write callback errormaxPendingBytes 默认 = 4 × 16 MiB 帧上限 = 64 MiBmaxPendingBytes defaults to 4 × the 16 MiB frame limit = 64 MiB失败的写不会卡住队列：#writeTail 吞掉错误，只有调用方的 Promise 被拒绝。A failed write never blocks the queue: #writeTail swallows it; only the caller's promise rejects.Windows 上不支持；连接前关闭 → "closed before connecting"。Not supported on Windows; closing before connect → "closed before connecting".RadiusClientReconnect：断开后 1 s 起指数退避（上限 30 s），RadiusClientReconnect: after a disconnect, back off from 1 s (max 30 s),reconnect() 后重新附着原 Session——这是应用的策略，不是 Client 的。reconnect(), then reattach the same Session: app policy, not the Client's.Connection 只依赖这个契约：Unix socket、WebSocket 中继或测试替身都能接上。Connection depends only on this contract: a Unix socket, a WebSocket relay or a test double all plug in.1234契约contractUnix 实现Unix implementation中继实现relay implementation失败failure

maxPendingBytes 默认是帧上限的 4 倍（64 MiB）；超过就拒绝这次 send，Connection 随之断开。maxPendingBytes defaults to 4 × the frame limit (64 MiB); beyond it the send rejects and the Connection drops.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.007
07 · 发现本机服务器07 · Discovering local serversRun modes

发现本机服务器Discovering local servers

discoverUnixServers: read a directory, take serverIds from file names, probe up to 16 at a time每个候选：probeUnixServerPer candidate: probeUnixServerunix.ts安静地忽略（不算这台服务器）Quietly ignored (not this server)其它错误：整个发现失败Anything else: discovery failsreaddir(directory)目录不存在 → 返回 []missing directory → []文件名过滤Filter names<serverId>.sock 且 isServerId<serverId>.sock with isServerId工作者池Worker poolmin(16, 候选数) 个，共享 nextIndexmin(16, candidates), shared nextIndex结果Result按 serverId 排序的 { serverId, path }{ serverId, path } sorted by serverIdtimeoutProtocolValidationErrorDisconnectedErrorServerError(version)ENOENTECONNREFUSEDECONNRESETEPIPEETIMEDOUTlstat(path).isSocket()不是 socket → 跳过；已消失（ENOENT）→ 跳过not a socket → skip; vanished (ENOENT) → skipnew Client({ serverId, transportFactory })期望的 serverId 来自文件名the expected serverId comes from the file namerace(client.connect(), 1000 ms)完整握手：版本与 serverId 都要对a full handshake: version and serverId must matchfinally: client.dispose()销毁 socket 并等它真正 closedestroy the socket and wait for its close只读：从不删除任何文件——陈旧 socket 的清理归服务器端的监听器。Read-only: it never deletes anything; stale-socket cleanup belongs to the server's listener.timeoutMs 必须是 1 … 2,147,483,647 的整数；Windows 上直接报错。timeoutMs must be an integer from 1 to 2,147,483,647; Windows throws.DisconnectedError 只在没有 cause 时忽略DisconnectedError is ignored only without a causefailure ??= { error }：记下第一个错误，failure ??= { error }: remember the first error,所有工作者停止领取新候选，every worker stops taking candidates,Promise.all 之后抛出它。and it is thrown after Promise.all.例如目录不可读（EACCES）。For example an unreadable directory (EACCES).coding-agent 的 client-runtime.ts 用它列出本机的服务器，再对选中的路由 Client.connect。coding-agent's client-runtime.ts uses it to list local servers, then Client.connects to the chosen route.1234文件系统filesystem探测probing忽略ignored失败failure

只读：陈旧或无响应的 socket 只会被忽略，从不删除——清理归服务器端的监听器。Read-only: stale or unresponsive sockets are just ignored, never deleted; cleanup belongs to the server's listener.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.008
08 · 失败语义与恢复08 · Failure semantics and recoveryRun modes

失败语义与恢复Failure semantics and recovery

Failure semantics: five errors and no automatic reconnectionPromise 会以什么拒绝What a promise can reject with断开之后：由应用决定After a disconnect: the app decides监听器的错误Listener errorserrors.tsServerError服务器说“不”：hello_error 或 ok: falsethe server said no: hello_error or ok: false.code：version · invalid_request · wrong_server · session_not_attached · cancelled ….code: version · invalid_request · wrong_server · session_not_attached · cancelled …errors.tsDisconnectedError没连上 / 连接没了not connected, or the connection is gone传输错误包成 cause；"Byte transport closed"；已在连接中transport errors as cause; "Byte transport closed"; already connectingerrors.tsClientDisposedErrordispose() 之后的任何调用any call after dispose()dispose 时所有挂起请求都以它拒绝every pending request rejects with it on disposepi-protocolProtocolValidationError对端违反协议：同时断开连接the peer broke the protocol: also disconnects坏帧 · serverId 不符 · 无主回应 · 他人的附着bad frame · serverId mismatch · orphan response · foreign attachmentsignal.reasonAbortError调用方 abortthe caller aborted只拒绝这一个请求；连接保持rejects just this request; the connection staysclient.reconnect()新的传输、新的握手（等同 connect）a fresh transport and handshake (same as connect)重新 attachattach again经应用的管理服务；得到新的 attachmentIdthrough the app's management service; a new attachmentId重新订阅subscribe again旧订阅在断开时已全部丢弃old subscriptions were all dropped on disconnect只重放安全的操作replay only safe operations已被接受的工作可能仍在远端完成accepted work may still complete remotelyonConnectionStateChange、onAttachmentChange 与订阅Exceptions from onConnectionStateChange, onAttachmentChange监听器抛出的异常只交给 ClientOptions.onListenerError；and subscription listeners only reach ClientOptions.onListenerError;它自己再抛也会被吞掉。if that throws too, it is swallowed.诊断不能影响协议或传输状态。Diagnostics cannot affect protocol or transport state.ServerError 的 code 与 message 都来自服务器；pi-server 只转发 ServerError 的消息，其它内部异常一律是 "Internal server error"。A ServerError's code and message come from the server; pi-server forwards only ServerError messages, other internal failures are always "Internal server error".1234错误类error classes恢复步骤recovery steps诊断diagnostics

断开时挂起的请求在本地拒绝，但已被服务器接受的工作仍可能在远端完成。On disconnect pending requests reject locally, but work the server accepted may still complete remotely.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.009
扩展点Extension pointspi-coding-agent

client 的扩展点Where to plug into client

1
ByteTransportFactoryByteTransportFactory

接入任何保序字节流：WebSocket、管道、测试替身。Plug in any order-preserving byte stream: a WebSocket, a pipe, a test double.

2
createClientServiceTransport(client, getTarget)createClientServiceTransport(client, getTarget)

把路由交给 Chord 绑定，得到类型化的远程服务。Hand a route to a Chord binding to get typed remote services.

3
ClientOptions.onListenerErrorClientOptions.onListenerError

收集监听器异常，而不影响连接。Collect listener exceptions without affecting the connection.

4
ClientOptions.maxFrameLength · maxPendingBytesClientOptions.maxFrameLength · maxPendingBytes

帧上限与 Unix 传输的排队上限；两端配置一致。The frame limit and the Unix transport's queue limit; configure both peers alike.

5
discoverUnixServers({ directory, timeoutMs })discoverUnixServers({ directory, timeoutMs })

列出某个目录里可以握手的本机服务器。List the local servers in a directory that complete a handshake.

← 返回完整图解← Back to the full walkthroughgithub.com/earendil-works/pi · v1.1.0
图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.010
←
1 / 10