图解 Pi AgentPi Agent, Illustratedv1.1.0
包Packages· client
演示Slides
Run modes已完成Donepackages/client · v1.1.0

client@earendil-works/pi-client

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

1.1.0
版本version
8
源文件src files
1,135
源码行src LOC
3
测试文件test files
27
测试用例test cases
2
导出路径export paths
依赖Depends onchordprotocol被依赖Used by—

这个包是什么What this package ispi-client 是远程 pi 会话的客户端一端,且与传输无关:你给它一个 ByteTransportFactory,它负责握手(并核对物理端点报告的逻辑 serverId)、按 id 关联请求、取消、带外附着和带缓冲的服务订阅。它不构造类型化的服务代理——createClientServiceTransport() 把一个按需解析的路由适配成 Chord 的 RemoteServiceTransport,类型由应用的 Chord 绑定提供。它从不自动重连。pi-client is the client end of remote pi sessions, and transport-neutral: give it a ByteTransportFactory and it handles the handshake (checking the logical serverId the physical endpoint reports), id-correlated requests, cancellation, out-of-band attachments and buffered service subscriptions. It builds no typed service proxies: createClientServiceTransport() adapts a lazily resolved route into a Chord RemoteServiceTransport, and types come from the application's Chord bindings. It never reconnects on its own.

1,135 行 TypeScript,两个导入路径:运行时无关的根入口 ".",以及只在 Node / Bun 上可用的 "./unix"(Unix socket 传输与发现)。依赖 chord 与 pi-protocol。唯一的使用者是 coding-agent 的 experimental/ 运行时。1,135 lines of TypeScript, two import paths: the runtime-neutral root "." and the Node / Bun-only "./unix" (Unix-socket transport and discovery). Depends on chord and pi-protocol. Its only user is coding-agent's experimental/ runtime.

01Run modes

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

一句话结论TakeawayClient(client.ts)面向调用方:请求、订阅、附着与监听器;它内部的 Connection(connection.ts)管状态机、握手和 ServerMessageDecoder;字节的进出全部经过 ByteTransport 契约(transport.ts)。unix.ts 是一个独立子路径,提供 Unix socket 传输与本机服务器发现。Client (client.ts) faces the caller: requests, subscriptions, attachments and listeners; its internal Connection (connection.ts) owns the state machine, the handshake and the ServerMessageDecoder; every byte goes through the ByteTransport contract (transport.ts). unix.ts is a separate subpath with a Unix-socket transport and local server discovery.

pi-client 的结构:Client、Connection 与可替换的字节传输pi-client: Client, Connection and a pluggable byte transport
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 根入口不碰 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).
  • Client 只认 @earendil-works/chord 的服务控制调用(目录、订阅、退订)和状态解码器,不解释任何应用契约。Client knows only @earendil-works/chord's service-control calls (catalogue, subscribe, unsubscribe) and state decoder; it interprets no application contract.
  • coding-agent 里有两种传输:createUnixTransportFactory() 和 radius-relay.ts 的 WebSocket 中继传输。coding-agent uses two transports: createUnixTransportFactory() and the WebSocket relay transport in radius-relay.ts.

工作原理How it works

  1. const client = await Client.connect({ serverId, transportFactory }):失败时它会自己 dispose()。const client = await Client.connect({ serverId, transportFactory }): on failure it disposes itself.
  2. 低层调用:client.request({ serverId: client.serverId }, { serviceId, member, args })。A low-level call: client.request({ serverId: client.serverId }, { serviceId, member, args }).
  3. 类型化调用:把 createClientServiceTransport(client, getTarget) 交给 Chord 绑定。Typed calls: hand createClientServiceTransport(client, getTarget) to a Chord binding.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/index.tsL1index.ts根入口the root entry
packages/client/src/client.tsL62Client面向调用方the caller-facing class
packages/client/src/connection.tsL41Connection状态机与握手state machine and handshake
packages/client/src/transport.tsL1ByteTransport传输契约the transport contract
packages/client/src/unix.tsL88createUnixTransportFactory()Unix 传输the Unix transport
02Run modes

公共 APIPublic API#

一句话结论Takeaway2 个导入路径(另有 ./package.json)。根入口从 4 个模块导出 5 个值和 10 个类型:Client、createClientServiceTransport 与三种错误,其余是类型;Unix socket 传输与发现在 ./unix。2 import paths (plus ./package.json). The root exports 5 values and 10 types from 4 modules: Client, createClientServiceTransport and three errors, the rest are types; the Unix-socket transport and discovery live in ./unix.

导入路径Import path目标Target说明Notes
@earendil-works/pi-client./dist/index.jsClient、Chord 传输适配器、错误类与传输契约Client, the Chord transport adapter, error classes and the transport contract
@earendil-works/pi-client/unix./dist/unix.jsNode / Bun 专用:Unix socket 传输与本机发现Node / Bun only: the Unix-socket transport and local discovery
@earendil-works/pi-client/package.json./package.json包元数据package metadata

package.json · exportsL8

常用导出Key exports

模块Module常用导出Key exports用途Use
packages/client/src/client.tsClient连接、请求、订阅、附着connect, request, subscribe, attach
packages/client/src/client.tscreateClientServiceTransport把路由适配成 Chord 传输adapt a route to a Chord transport
packages/client/src/errors.tsServerError DisconnectedError ClientDisposedError三种错误the three errors
packages/client/src/transport.tsByteTransport ByteTransportFactory ByteTransportHandlers自定义传输custom transports
packages/client/src/types.tsClientOptions ServiceSubscription ConnectionStateChange选项与句柄options and handles

子路径 ./unixSubpath ./unix

模块Module常用导出Key exports用途Use
packages/client/src/unix.tscreateUnixTransportFactory一个 Unix socket 传输工厂a Unix-socket transport factory
packages/client/src/unix.tsdiscoverUnixServers列出可以握手的本机服务器list local servers that complete a handshake
packages/client/src/unix.tsUnixTransportOptions UnixServerRoute DiscoverUnixServersOptions选项与结果类型option and result types

根入口 index.ts 的全部导出(按来源模块)Every root index.ts export, by source module

展开全部 4 个来源模块Show all 4 source modules5 values10 types
来源模块Source module值导出Value exports类型导出Type exports
packages/client/src/client.tsClient createClientServiceTransport—
packages/client/src/errors.tsClientDisposedError DisconnectedError ServerError—
packages/client/src/transport.ts—ByteTransport ByteTransportFactory ByteTransportHandlers
packages/client/src/types.ts—AttachmentChangeListener ClientOptions ConnectionState ConnectionStateChange ListenerErrorHandler ServiceSubscription Unsubscribe

由构建脚本解析 export { … } from 语句生成。Generated by parsing the export { … } from statements at build time.

03Run modes

连接的三个状态Three connection states#

一句话结论TakeawayConnection 在 disconnected → connecting → connected 之间切换。每次 connect() 都是一次带新序号的尝试:调用工厂得到一个全新的传输,先发客户端 hello,再等待服务端 hello,并要求其中的 serverId 与期望值完全一致。任何失败、关闭或 disconnect() 都回到 disconnected。Connection moves between disconnected → connecting → connected. Every connect() is an attempt with a new sequence number: the factory builds a fresh transport, the client hello goes first, then the server hello must arrive with exactly the expected serverId. Any failure, close or disconnect() returns to disconnected.

Connection 的三个状态与握手失败Connection in three states, and handshake failures
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 旧尝试的回调带着旧 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.
  • 握手前的数据、第一条不是 hello、握手后再来 hello / hello_error,都是协议错误。Data before the client hello, a first message other than hello, or another hello / hello_error after connecting are all protocol errors.
  • 进入 disconnected 时 Client 清空 hello 与附着、拒绝全部挂起请求、丢弃全部订阅,再通知 onConnectionStateChange。Entering disconnected, the Client clears hello and the attachment, rejects every pending request, drops every subscription, then notifies onConnectionStateChange.

工作原理How it works

  1. client.onConnectionStateChange(({ state, error }) => …) 观察状态;client.connectionState 读取当前值。Observe with client.onConnectionStateChange(({ state, error }) => …); read client.connectionState for the current value.
  2. maxFrameLength 必须是 1 到 0xffffffff 的安全整数,否则构造时抛 TypeError。maxFrameLength must be a safe integer from 1 to 0xffffffff, or the constructor throws a TypeError.
  3. 关闭时解码器 end():残留半帧会变成截断错误,作为断开原因上报。On close the decoder end()s: a leftover partial frame becomes a truncation error, reported as the disconnect reason.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/connection.tsL67connect()开始一次尝试start an attempt
packages/client/src/connection.tsL120#openTransport()打开传输并发 helloopen the transport, send hello
packages/client/src/connection.tsL163#handleMessage()握手校验handshake checks
packages/client/src/connection.tsL234#fail()回到 disconnectedback to disconnected
packages/client/src/client.tsL345#handleConnectionStateChange()断开时的清理cleanup on disconnect
04Run modes

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

一句话结论Takeaway#request() 先过三道守卫(已 dispose、未连接、已 abort),通过后才分配 request-N 并登记到 #pendingRequests,再编码、发送。回应按 id 取回条目:ok: true 解析结果,ok: false 变成带 code 的 ServerError。abort 立即在本地拒绝,并在已发送时补发一条 cancel。#request() passes three guards first (disposed, not connected, already aborted), and only then allocates request-N, registers it in #pendingRequests, encodes and sends. A response takes the entry back by id: ok: true resolves the result, ok: false becomes a ServerError with a code. An abort rejects locally at once and, if the request was sent, follows up with a cancel.

一个请求:守卫、登记、编码、发送与按 id 结算One request: guards, registration, encoding, sending and settlement by id
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 回应可以乱序到达;没有对应条目的回应是协议错误,会断开连接。Responses may arrive out of order; a response with no matching entry is a protocol error and drops the connection.
  • abort 之后条目留在表里,直到服务端的 cancelled 回应到来,所以那条回应不算“无主”。After an abort the entry stays until the server's cancelled response arrives, so that response is not an orphan.
  • 调用在发送前用 Chord 的 parseServiceCall() 校验;编码失败会取回条目并拒绝,什么也不发送。The call is validated with Chord's parseServiceCall() before sending; an encoding failure takes back the entry and rejects without sending anything.

工作原理How it works

  1. await client.request(target, call, signal):signal 是普通的 AbortSignal。await client.request(target, call, signal): signal is an ordinary AbortSignal.
  2. client.serviceCatalogue(target) 是同一路径上的一个预置调用,结果用 parseServiceCatalogue() 校验。client.serviceCatalogue(target) is a preset call on the same path, checked with parseServiceCatalogue().
  3. 错误处理:catch (e) { if (e instanceof ServerError && e.code === "cancelled") … }。Error handling: catch (e) { if (e instanceof ServerError && e.code === "cancelled") … }.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/client.tsL238#request()守卫、登记、发送guard, register, send
packages/client/src/client.tsL252sendCancel()补发 cancelthe follow-up cancel
packages/client/src/client.tsL304#handleMessage()按 id 结算settle by id
packages/client/src/client.tsL370#rejectPendingRequests()断开时全部拒绝reject all on disconnect
packages/client/src/client.tsL159serviceCatalogue()预置调用a preset call
05Run modes

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

一句话结论TakeawaysubscribeService() 先登记一个 service-N 监听器(hydrated = false, ready = false),再发订阅请求。快照之前到达的更新原样缓存;快照解码后 hydrated = true 并解码积压;调用方装好快照、调用 start() 之后,缓冲的更新才按序交付。交付经 deliveryTail 串行,异步监听器也不会乱序。subscribeService() registers a service-N listener first (hydrated = false, ready = false), then sends the subscribe request. Updates arriving before the snapshot are cached raw; once the snapshot decodes, hydrated = true and the backlog is decoded; only after the caller installs the snapshot and calls start() do buffered updates flow, in order. Delivery is chained on deliveryTail, so async listeners stay ordered too.

一个服务订阅:快照之前缓冲,start() 之后按序交付One service subscription: buffered before the snapshot, ordered after start()
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 每个订阅有自己的 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.
  • 断开时 #serviceListeners 整体清空;订阅不会自动恢复。On disconnect #serviceListeners is cleared wholesale; subscriptions are not restored.

工作原理How it works

  1. const sub = await client.subscribeService(target, serviceId, mode, listener);装好 sub.snapshot 后调用 sub.start()。const sub = await client.subscribeService(target, serviceId, mode, listener); install sub.snapshot, then call sub.start().
  2. 监听器抛出的错误只交给 ClientOptions.onListenerError。Listener errors only reach ClientOptions.onListenerError.
  3. 通过 Chord 绑定使用时,activate 对应 start()、close 对应 dispose()。Through a Chord binding, activate maps to start() and close to dispose().

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/client.tsL172subscribeService()登记并订阅register and subscribe
packages/client/src/client.tsL316hydrated快照前缓存cache before the snapshot
packages/client/src/client.tsL216start()开始交付begin delivery
packages/client/src/client.tsL417#deliverServiceUpdate()串行交付ordered delivery
packages/client/src/types.tsL16ServiceSubscription订阅句柄the subscription handle
06Run modes

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

一句话结论Takeaway附着由服务器带外推送:attachment 信封里的 serverId 必须是本服务器,否则断开;三个字段都相同的更新被忽略,其余通知 onAttachmentChange。createClientServiceTransport(client, getTarget) 在每次调用时才求值 getTarget(),所以 Session 服务可以直接传 () => client.attachment。The server pushes attachments out of band: the serverId in an attachment envelope must be this server's or the connection drops; an update with all three fields equal is ignored, anything else notifies onAttachmentChange. createClientServiceTransport(client, getTarget) evaluates getTarget() on every call, so Session services can pass () => client.attachment directly.

带外附着与 createClientServiceTransportOut-of-band attachments and createClientServiceTransport
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 附着信封先于 attach 的回应送达:回应 resolve 时 client.attachment 已经可用。The attachment envelope arrives before the attach response: when it resolves, client.attachment is already set.
  • 路由不存在时,适配器抛 "Remote service target is unavailable",不会发出请求。Without a route the adapter throws "Remote service target is unavailable" and sends nothing.
  • #targetIsCurrent():服务器级目标与 hello 的 serverId 比较,Session 目标与当前附着的三个字段比较。#targetIsCurrent(): a server target is compared with the hello's serverId, a Session target with all three fields of the current attachment.

工作原理How it works

  1. 服务器级服务:createClientServiceTransport(client, () => ({ serverId: client.serverId }))。Server-wide services: createClientServiceTransport(client, () => ({ serverId: client.serverId })).
  2. Session 服务:createClientServiceTransport(client, () => client.attachment)。Session services: createClientServiceTransport(client, () => client.attachment).
  3. client.onAttachmentChange((a) => …):undefined 表示没有附着。client.onAttachmentChange((a) => …): undefined means no attachment.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/client.tsL448createClientServiceTransport()Chord 适配器the Chord adapter
packages/client/src/client.tsL398#setAttachment()去重并通知dedupe and notify
packages/client/src/client.tsL423#targetIsCurrent()路由是否仍有效is the route current
packages/coding-agent/src/experimental/services/connection.tsL204connection.tscoding-agent 的 Session 服务coding-agent's Session services
07Run modes

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

一句话结论TakeawayByteTransport 只有 send(chunk): Promise<void> 和 close();工厂收到 onData / onClose / onError 三个回调,每次连接尝试新建一个已认证的传输。UnixByteTransport 用一条 Promise 链保序,按 maxPendingBytes 限制排队字节,写入返回 false 时等待 drain。ByteTransport is just send(chunk): Promise<void> and close(); the factory receives three callbacks, onData / onClose / onError, and builds one authenticated transport per attempt. UnixByteTransport keeps order with a promise chain, caps queued bytes with maxPendingBytes, and waits for drain when a write returns false.

ByteTransport 契约与 Unix 写队列The ByteTransport contract and the Unix write queue
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • maxPendingBytes 默认是帧上限的 4 倍(64 MiB);超过就拒绝这次 send,Connection 随之断开。maxPendingBytes defaults to 4 × the frame limit (64 MiB); beyond it the send rejects and the Connection drops.
  • 每个 chunk 都先 slice() 复制,调用方之后修改缓冲也不影响写出的字节。Every chunk is copied with slice() first, so later changes to the caller's buffer cannot affect what is written.
  • 失败的写不会堵住队列:#writeTail 吞掉错误,只有该次调用的 Promise 被拒绝。A failed write never blocks the queue: #writeTail swallows it, and only that call's promise rejects.

工作原理How it works

  1. createUnixTransportFactory({ path, maxPendingBytes }):path 不能为空,Windows 上不可用。createUnixTransportFactory({ path, maxPendingBytes }): the path must not be empty; unavailable on Windows.
  2. 自定义传输:实现 send / close,并且恰好调用一次 onClose 或 onError。A custom transport: implement send / close and call exactly one of onClose or onError.
  3. 两端配置相同的 maxFrameLength,并让 maxPendingBytes 至少容得下一帧。Configure the same maxFrameLength on both peers, with maxPendingBytes large enough for one frame.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/transport.tsL18ByteTransportFactory工厂契约the factory contract
packages/client/src/unix.tsL147UnixByteTransport写队列the write queue
packages/client/src/unix.tsL103connectUnixSocket()连接与终止回调connect and terminal callbacks
packages/client/src/unix.tsL93validateUnixTransportOptions()默认值与检查defaults and checks
packages/coding-agent/src/experimental/radius-relay.tsL437RadiusClientByteTransportWebSocket 中继传输the WebSocket relay transport
08Run modes

发现本机服务器Discovering local servers#

一句话结论TakeawaydiscoverUnixServers({ directory }) 读一个目录,只保留名为 <serverId>.sock 且 serverId 是规范 UUIDv4 的条目,然后用最多 16 个工作者并发探测:确认是 socket、用一个临时 Client 完成握手(默认 1000 ms 超时),最后按 serverId 排序返回 { serverId, path }。discoverUnixServers({ directory }) reads a directory, keeps only entries named <serverId>.sock whose serverId is a canonical UUIDv4, then probes them with up to 16 concurrent workers: check it is a socket, complete a handshake with a temporary Client (1000 ms timeout by default), and return { serverId, path } sorted by serverId.

discoverUnixServers:目录、过滤与并发握手探测discoverUnixServers: directory, filtering and concurrent handshake probes
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 只读:陈旧或无响应的 socket 只会被忽略,从不删除——清理归服务器端的监听器。Read-only: stale or unresponsive sockets are just ignored, never deleted; cleanup belongs to the server's listener.
  • 超时、协议错误、版本不符、无 cause 的断开以及 ENOENT / ECONNREFUSED / ECONNRESET / EPIPE / ETIMEDOUT 都算“不是这台服务器”。Timeouts, protocol errors, version mismatches, cause-less disconnects and ENOENT / ECONNREFUSED / ECONNRESET / EPIPE / ETIMEDOUT all count as “not this server”.
  • 其它错误让整个发现失败:第一个错误被记下,工作者停止领取新候选。Any other error fails discovery: the first one is remembered and workers stop taking candidates.

工作原理How it works

  1. const routes = await discoverUnixServers({ directory }),directory 用绝对路径,例如展开后的 ~/.pi/server。const routes = await discoverUnixServers({ directory }) with an absolute directory such as the expanded ~/.pi/server.
  2. timeoutMs 覆盖每次探测的超时(1 … 2,147,483,647)。timeoutMs overrides the per-probe timeout (1 … 2,147,483,647).
  3. 然后 Client.connect({ serverId: route.serverId, transportFactory: createUnixTransportFactory({ path: route.path }) })。Then Client.connect({ serverId: route.serverId, transportFactory: createUnixTransportFactory({ path: route.path }) }).

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/unix.tsL37discoverUnixServers()读目录与工作者池directory and worker pool
packages/client/src/unix.tsL237probeUnixServer()一次握手探测one handshake probe
packages/client/src/unix.tsL17MAX_CONCURRENT_DISCOVERY_PROBES并发上限 16concurrency 16
packages/coding-agent/src/experimental/client-runtime.tsL79client-runtime.tscoding-agent 的使用coding-agent's use
09Run modes

失败语义与恢复Failure semantics and recovery#

一句话结论Takeaway调用方会看到五种拒绝:ServerError(服务器拒绝,带 code)、DisconnectedError(没连上或连接断了,传输错误放在 cause)、ClientDisposedError、ProtocolValidationError(对端违反协议,并且断开)和 AbortError。Client 从不自动重连或重放:断开之后 reconnect()、重新 attach、重新订阅,只重放已知安全的操作——这是应用的策略。Callers see five kinds of rejection: ServerError (the server refused, with a code), DisconnectedError (not connected or the connection dropped; transport errors in cause), ClientDisposedError, ProtocolValidationError (the peer broke the protocol, and the connection drops) and AbortError. The Client never reconnects or replays on its own: after a disconnect, reconnect(), attach again, subscribe again, and replay only operations known to be safe; that is the application's policy.

五种错误与断开之后的恢复Five errors and recovery after a disconnect
← 左右滑动查看,点击放大 →← swipe to scroll, tap to enlarge →
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

要点Key points

  • 断开时挂起的请求在本地拒绝,但已被服务器接受的工作仍可能在远端完成。On disconnect pending requests reject locally, but work the server accepted may still complete remotely.
  • coding-agent 的 RadiusClientReconnect 就是这样一种应用策略:指数退避重连并重新附着原来的 Session。coding-agent's RadiusClientReconnect is one such application policy: reconnect with exponential backoff and reattach the same Session.
  • 监听器的异常只交给 onListenerError,诊断不能影响协议或传输状态。Listener exceptions only reach onListenerError; diagnostics cannot affect protocol or transport state.

工作原理How it works

  1. e instanceof ServerError → 看 e.code;e instanceof DisconnectedError → 看 e.cause。e instanceof ServerError → inspect e.code; e instanceof DisconnectedError → inspect e.cause.
  2. await using client = … 或 await client.dispose():两者等价(Symbol.asyncDispose)。await using client = … or await client.dispose(): equivalent (Symbol.asyncDispose).
  3. client.reconnect() 只是再走一次 connect():新传输、新握手。client.reconnect() is just another connect(): a new transport and a new handshake.

关键文件Key files · v1.1.0

文件File符号Symbol作用Role
packages/client/src/errors.tsL3ServerError带 code 的服务器错误server error with a code
packages/client/src/errors.tsL13DisconnectedError断开,带 causedisconnected, with cause
packages/client/src/errors.tsL20ClientDisposedError已 disposedisposed
packages/client/src/client.tsL134reconnect()显式重连explicit reconnection
packages/coding-agent/src/experimental/radius-relay.tsL344RadiusClientReconnect一种应用策略one application policy
10Run modes

扩展点Extension points#

ByteTransportFactory接入任何保序字节流:WebSocket、管道、测试替身。Plug in any order-preserving byte stream: a WebSocket, a pipe, a test double.
createClientServiceTransport(client, getTarget)把路由交给 Chord 绑定,得到类型化的远程服务。Hand a route to a Chord binding to get typed remote services.
ClientOptions.onListenerError收集监听器异常,而不影响连接。Collect listener exceptions without affecting the connection.
ClientOptions.maxFrameLength · maxPendingBytes帧上限与 Unix 传输的排队上限;两端配置一致。The frame limit and the Unix transport's queue limit; configure both peers alike.
discoverUnixServers({ directory, timeoutMs })列出某个目录里可以握手的本机服务器。List the local servers in a directory that complete a handshake.
11Run modes

测试Tests#

一句话结论Takeaway三个测试文件:client.test.ts 用内存传输覆盖握手、乱序回应、取消、订阅缓冲、附着与断开后的拒绝;unix-transport.test.ts 在真实 Unix socket 上跑完整握手与截断帧;unix.test.ts 覆盖发现的排序、过滤、陈旧 socket、超时与 16 路并发上限。Three test files: client.test.ts uses an in-memory transport for the handshake, out-of-order responses, cancellation, subscription buffering, attachments and rejection after disconnect; unix-transport.test.ts runs a full handshake and truncated frames over a real Unix socket; unix.test.ts covers discovery ordering, filtering, stale sockets, timeouts and the 16-probe limit.

用例数 = 源码中 it( / test( 的出现次数(静态统计,不运行测试)。Cases = occurrences of it( / test( in the source (static count; tests were not run).

12Run modes

交叉链接Cross-links#