v1.1.0 Pi v1.1.0 源码分析 Analysis of Pi v1.1.0 · 2026-10-07
server server 实验性的 pi 服务端 experimental server package for pi
Run modes Infra Run modes
← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言 ← → navigate · F fullscreen · O overview · T theme · L language
01 · Server、SessionRouter 与可替换的监听器 01 · Server, SessionRouter and pluggable listeners Run modes
Server、SessionRouter 与可替换的监听器 Server, SessionRouter and pluggable listeners pi-server: 1,958 lines — handshake, route-based dispatch, attachments to the app's Sessions packages/server/src transports/unix → "./unix" testing/ → "./testing" 依赖与使用者 Dependencies and users 4 index.ts 根入口 "."：export * 自 errors、listener、server、types root ".": export * from errors, listener, server, types 576 server.ts Server：连接、握手、请求分派 Server: connections, handshake, dispatch toProtocolError · failProtocol 312 session-router.ts SessionRouter：Session 与附着（内部） SessionRouter: Sessions, attachments (internal) runForClient · releaseAttachment 68 types.ts ServerHost 契约 ServerHost contract 57 errors.ts 5 个错误子类 5 error subclasses 8 listener.ts ServerListener 37 connection.ts ByteConnection（内部） ByteConnection (internal) 476 createUnixServer getUnixSocketPath · createUnixListener 绑定、发布、身份校验 bind, publish, identity checks 420 createTestServer TestServerHost · TestHarness ProtocolTestClient · Deferred @earendil-works/chord ^1.1.0 服务控制解码、状态编码器、Context 与取消 service-control decoding, state encoder, Context and abort @earendil-works/pi-protocol ^1.1.0 ClientMessageDecoder · encodeServerMessage · isServerId server.ts createUnixServer：~/.pi/server/<id>.sock createUnixServer: ~/.pi/server/<id>.sock services/server.ts RoutedServerServiceHost：服务器级服务 RoutedServerServiceHost: server-wide services session-worker-manager.ts RoutedSessionHandle：每个 Session 一个 worker RoutedSessionHandle: one worker per Session radius-relay.ts 中继来的连接交给 server.accept() relayed connections go to server.accept() 依赖 Depends on 使用者：coding-agent 的 experimental/（4 个文件导入它） Used by: coding-agent experimental/ (4 importing files) 目录可用 PI_SERVER_DIR 覆盖。 PI_SERVER_DIR overrides the directory. Server 本身不懂任何服务契约，也不保存 Session：它只核对路由，把调用交给应用通过 ServerHost 提供的能力。 The Server knows no service contract and stores no Session: it checks routes and hands calls to capabilities the app supplies through ServerHost. 标签 = 行数 tag = lines 核心 core Session 路由 Session routing Unix / 使用方 Unix / users 测试与依赖 testing & deps 监听器负责传输层的认证；实验性的 Unix 监听器不做认证，只依靠文件权限。 Listeners own transport-level authentication; the experimental Unix listener does none and relies on file permissions.
02 · ServerHost：应用提供的能力 02 · ServerHost: capabilities the app provides Run modes
ServerHost：应用提供的能力 ServerHost: capabilities the app provides ServerHost: every capability the app hands to the Server «app» ServerHost<TMetadata> + serverServices: RoutedServerServiceHost + resolveSession(sessionId, ctx): TMetadata + openSession(metadata, ctx): RoutedSessionHandle «app» RoutedServerServiceHost + attachClient(presentation, ctx) «per conn» RoutedServerServiceAttachment + invokeService(call, publish, ctx) + release(ctx) «per Session» RoutedSessionHandle + attachClient(ctx): RoutedSessionAttachment + terminated?: Promise<Error | undefined> + close(ctx): Promise<void> «per attach» RoutedSessionAttachment + invokeService(call, publish, ctx) + release(ctx) «from Server» RoutedServerPresentation + attachSession(sessionId, ctx) + detachSession(ctx) + prepareSessionRemoval(sessionId, ctx) serverServices 握手时 at handshake openSession() 每次 attach per attach 作为参数 passed in 握手时： At handshake: serverServices.attachClient(presentation) 为每个连接建一个服务器级端点；连接断开时 release()。 serverServices.attachClient(presentation) builds one server-wide endpoint per connection; release() when it drops. attach 时： On attach: resolveSession → openSession（同一 Session 只开一次）→ handle.attachClient() 为这次附着取得一个租约。 resolveSession → openSession (once per Session) → handle.attachClient() takes a lease for this attachment. terminated： terminated: 意外退出时以错误 resolve，正常关闭时以 undefined resolve；Server 随即让该 Session 的所有附着失效。 resolves with an error on unexpected exit, undefined after a normal close; the Server then invalidates every attachment. publish： publish: invokeService 的第二个参数；订阅的更新经它回到 Server，再编码成 service_update。 invokeService's second argument; subscription updates come back through it and become service_update. coding-agent：serverServices 来自 services/server.ts；openSession 交给 SessionWorkerManager，每个 Session 跑在自己的 worker 进程里。 coding-agent: serverServices comes from services/server.ts; openSession goes to SessionWorkerManager, each Session running in its own worker process. 服务器级 server-wide Session 级 Session level Server 提供给应用 Server → app 返回 returns 参数 argument 服务器级端点拿到一个 RoutedServerPresentation：attachSession、detachSession、prepareSessionRemoval——应用的管理服务就是用它们 attach 的。 The server-wide endpoint receives a RoutedServerPresentation: attachSession, detachSession, prepareSessionRemoval; the app's management service attaches through these.
03 · 连接的五个阶段 03 · Five connection stages Run modes
连接的五个阶段 Five connection stages Five stages of a server connection: the handshake must finish within 5 s finishHandshake(state, hello) server.ts failProtocol(state, error) server.ts disconnect(state)：只做一次 disconnect(state): runs once awaitingHello handshaking ready closing closed 超时 5 s · 首条不是 hello · 坏帧 timeout 5 s · first is not hello · bad frame 传输 onClose / onError transport onClose / onError 收到 hello got hello hello 已发出 hello sent 版本检查 Version check 不支持 → hello_error { code: "version" } unsupported → hello_error { code: "version" } serverServices.attachClient(presentation) presentation = { attachSession, detachSession, prepareSessionRemoval } 再检查一次 Check again 期间关闭或断开 → release() 刚得到的端点，就此结束 closed or dropped meanwhile → release() the new endpoint and stop send hello { version, serverId } 发送成功 → stage = ready，清掉握手定时器 sent → stage = ready, clear the handshake timer 握手期间到达的请求挂在 state.handshake 之后，ready 时才处理，不会丢。 Requests arriving during the handshake wait on state.handshake and run once ready: none are lost. stage = closing；把 hello_error { error } 编码成最后一帧， stage = closing; encode hello_error { error } as the final frame, connection.close(finalFrame) 发完它再关闭，然后 disconnect。 connection.close(finalFrame) sends it, then closes; then disconnect. 握手之后的协议错误（例如第二个 hello）也走这条路。 Protocol errors after the handshake (a second hello) take this path too. ProtocolValidationError → invalid_request。 ProtocolValidationError → invalid_request. stage = closed，清掉握手定时器 stage = closed, clear the handshake timer abort 所有进行中的请求："Client disconnected" abort every active request: "Client disconnected" 清空订阅编码器；连接数变化 → onConnectionCountChanged clear subscription encoders; onConnectionCountChanged sessions.disconnect() 与 serverServices.release() 并行 sessions.disconnect() and serverServices.release() in parallel Server.close()：先关所有监听器，再把每个连接置为 closing 并关闭，最后 SessionRouter.close() 关闭所有托管的 Session。 Server.close(): close every listener, set each connection to closing and close it, then SessionRouter.close() closes every hosted Session. 1 2 3 4 1 2 3 4 前进 advance 协议失败 protocol failure 可以处理请求 serving requests 清理 cleanup 握手期间到达的请求挂在 state.handshake 之后，ready 时才处理，不会丢。 Requests that arrive during the handshake wait on state.handshake and run once ready; none are lost.
04 · 一个请求的处理 04 · Handling one request Run modes
一个请求的处理 Handling one request handleRequest: check, route to a Session or server-wide services, and send subscription updates after the response handleRequest(state, envelope) server.ts 失败时 On failure toProtocolError(error) server.ts handleCancel(state, envelope) server.ts invalid_request invalid_request wrong_server invalid_request activeRequests.has(id) parseServiceCall(envelope.call) Chord 校验调用的形状 Chord checks the call's shape activeRequests.set(id, { controller, target }) context = withAbortSignal(controller.signal) target.serverId !== this.serverId 订阅 id 已有编码器 subscription id already has an encoder 路由 Route "sessionId" in target ? sessions.executeServiceCall : serverServices.invokeService 订阅：必须返回快照 Subscribe: a snapshot is required encodeSnapshot，安装 ServiceStateEncoder encodeSnapshot, install a ServiceStateEncoder 发送回应，再冲刷 pendingUpdates Send the response, then flush pendingUpdates 之后的更新直接编码成 service_update later updates go straight out as service_update finally：从 activeRequests 删掉自己（只删自己那一项）。 finally: remove its own entry from activeRequests. 还没回应： Not yet responded: response { ok: false, error }；已 abort → cancelled，否则 toProtocolError response { ok: false, error }; aborted → cancelled, else toProtocolError 已经回应（冲刷更新时出错）： Already responded (failed while flushing): onError，然后关闭这条连接——订阅流不能半途断掉 onError, then close the connection: a stream must not break halfway 安装了编码器却没回应成功 → 移除编码器。 Encoder installed but no response → the encoder is removed. 退订成功 → 删掉对应的编码器，之后的更新被丢弃。 A successful unsubscribe deletes the encoder; later updates are dropped. ServerError · RemoteServiceError 原样：code + message as is: code + message ProtocolValidationError invalid_request + message 其它任何错误 anything else internal_error "Internal server error" + onError serverId 不是本服务器 → 忽略；id 与 target 都相同 → serverId of another server → ignored; same id and target → controller.abort(AbortError)；处理器看到 abort 后回应 cancelled。 controller.abort(AbortError); the handler then answers cancelled. 取消从不单独回应：它只影响原请求的回应。 A cancel never gets its own response; it only shapes the original one. 同一连接上的请求并发处理；只有 attach / detach 与 Session 调用的“准入”按连接串行（见 SessionRouter）。 Requests on one connection run concurrently; only attach / detach and the admission of Session calls are serialised per connection (see SessionRouter). 1 2 3 4 5 6 7 8 请求路径 request path 错误回应 error responses 错误映射 error mapping 取消 cancellation 错误回应用 toProtocolError() 映射：ServerError / RemoteServiceError 原样，ProtocolValidationError → invalid_request，其它 → internal_error。 Error responses are mapped by toProtocolError(): ServerError / RemoteServiceError as is, ProtocolValidationError → invalid_request, anything else → internal_error.
05 · SessionRouter：attach 与准入 05 · SessionRouter: attach and admission Run modes
SessionRouter：attach 与准入 SessionRouter: attach and admission SessionRouter: attach serialised per connection, every Session call checked against its attachment acquire → open(sessionId) session-router.ts executeServiceCall(call, target, client, …) session-router.ts attachClient(client, sessionId) 正在关闭 → ServerDrainingError closing → ServerDrainingError runForClient(client, op) 按连接排队：前一个操作结束后才开始 queued per connection: starts after the previous op 已附着同一个 Session？ Already attached to this Session? 是 → 直接返回（幂等） yes → return at once (idempotent) acquire(sessionId) 已托管 → 正在打开 → open() hosted → opening → open() 释放当前附着 Release the current attachment publish = false：新路由马上会推送 publish = false: the new route follows handle.attachClient(ctx) 新附着 id = randomUUID()，拿到租约 new attachment id = randomUUID(), take the lease publishAttachment(client, target) attachment { serverId, sessionId, attachmentId } session_not_attached server_draining hostedSessions openingSessions attachmentsByClient clientOperations 拿到租约后再查一次：Session 已失效、连接已断或服务器在关闭 → 释放租约并 ServerDrainingError。 After the lease, check again: Session invalidated, client gone or server closing → release it and ServerDrainingError. host.resolveSession(sessionId) 得到元数据；找不到 / 有歧义由应用抛 SessionNotFound / Ambiguous metadata; the app throws SessionNotFound / Ambiguous host.openSession(metadata) 得到 RoutedSessionHandle；期间开始关闭 → handle.close() 并拒绝 a RoutedSessionHandle; closing meanwhile → handle.close() and reject hostedSessions.set(metadata.id, …) 并监听 handle.terminated → invalidate() and watch handle.terminated → invalidate() openingSessions 让并发的 attach 共享同一次打开；打开失败时所有人一起失败，之后可以重试。 openingSessions lets concurrent attaches share one open; a failure fails them all, and a later retry may succeed. runForClient → requireAttachment sessionId 与 attachmentId 都要等于本连接的当前附着 sessionId and attachmentId must equal this connection's attachment lease.invokeService(call, publish, ctx) 只在队列里“准入”；结果在队列外等待，所以调用可以并发 only admission is queued; the result is awaited outside, so calls run concurrently trackOperation(attachment, result) 释放附着前会等这些调用结算 release waits for these calls to settle 不匹配、过期或服务器级 target → session_not_attached；关闭中或连接已断 → server_draining。 mismatched, stale or server-level target → session_not_attached; closing or client gone → server_draining. 内部状态 Internal state 1 2 3 1 2 3 排队 queueing Session 与租约 Session and lease 路由 route 拒绝 rejections 只有准入排队；调用结果在队列外等待，所以同一 Session 上的调用可以并发。 Only admission is queued; results are awaited outside the queue, so calls on one Session run concurrently.
06 · 附着的结束 06 · How attachments end Run modes
附着的结束 How attachments end Six ways an attachment ends, all through one release path 触发 Triggers releaseAttachment(attachment) 之后：Session 本身 Afterwards: the Session itself 关闭服务器时 On shutdown SessionRouter.close() detachSession() 推送 attachment: null pushes attachment: null attach 另一个 Session attach another Session 不推送 null，随后推送新路由 no null; the new route follows 连接断开 connection lost 不推送（已经没有连接） no push (no connection left) handle.terminated invalidate：所有附着，推送 null invalidate: every attachment, push null prepareSessionRemoval() 所有附着 + handle.close() every attachment + handle.close() Server.close() 全部附着 + 全部 handle.close() all attachments + every handle.close() releasing ??= … 只释放一次；并发调用共享这个 Promise released once; concurrent callers share the promise allSettled(attachment.operations) 先等已准入的调用结算 wait for admitted calls to settle lease.release(ctx) 租约仍在获取中 → 先等 acquiring still acquiring → await it first clearAttachment(…, publish) 从 Session 与连接上摘掉；publish 时推送 null detach from Session and client; null if publish release 失败也一定 clearAttachment（finally）。 clearAttachment runs even if release fails (finally). 断开 / detach / 切换 drop / detach / switch Session 仍被托管；是否退役由应用决定 stays hosted; the app decides on retirement terminated 从 hostedSessions 删掉；下次 attach 重新打开 removed from hostedSessions; reopened on next attach 删除 / 关闭服务器 removal / shutdown handle.close()：释放 worker 与写入权 handle.close(): frees the worker and writer 等进行中的连接操作和打开 wait for client ops and opens 释放每个 Session 的每个附着 release every attachment 关闭每个 handle close every handle 有失败 → AggregateError failures → AggregateError Server.closed 因此 reject。 so Server.closed rejects. README：连接断开时，本地的回应立即拒绝，但附着要等已准入的服务调用结算后才释放。 README: losing a connection rejects its local responses, but the attachment is released only after admitted service calls settle. 1 2 3 4 连接触发 connection triggers Session 失效 Session failure 释放与关闭 release and close 连接断开只释放附着，Session 仍被托管；什么时候让 worker 退役由应用决定。 A dropped connection only releases the attachment; the Session stays hosted, and the app decides when its worker retires.
07 · Unix 监听器 07 · The Unix listener Run modes
Unix 监听器 The Unix listener The Unix listener: bind privately, publish atomically with a hard link, remove only its own socket UnixListener.start(accept) transports/unix/listener.ts 例子：README 的目录 Example: the README's directory removeStaleSocket / isSocketLive close()：只删自己的 socket close(): remove only its own socket UnixByteConnection mkdir(dir, { recursive, mode: 0o700 }) 私有目录 a private directory removeStaleSocket(path) · (bindPath) 不是 socket → 拒绝；还活着 → "already running" not a socket → refuse; alive → "already running" listen(bindPath) bind-<sha256(path) 前 8 位>，与公共路径同目录 bind-<first 8 hex of sha256(path)>, same directory lstat(bindPath) → { dev, ino } 记住自己 socket 的身份 remember the identity of its own socket link(bindPath, path) 硬链接发布：原子，且目标已存在时失败 publish by hard link: atomic, fails if the target exists chmod(path, 0o600) 只有属主能连接（Windows 上跳过） owner-only connections (skipped on Windows) unlink(bindPath) 只留下公共路径 only the public path remains 任何一步失败：关闭 net.Server，删掉未发布的 bind 路径，再按身份清理。 Any step fails: close the net.Server, remove the unpublished bind path, clean up by identity. /run/user/1000/pi/ bind-0842f039 0f8fad5b-d9cb-469f-a165-70867728950e.sock // ⑤ link 之后两者是同一个 inode，⑦ 删掉 bind // ⑤ after link both are one inode; ⑦ drops bind // getUnixSocketPath(serverId, dir) bind-0842f039 = sha256("/run/user/1000/pi/<id>.sock") 的前 8 位（构建时计算） bind-0842f039 = first 8 hex of sha256 of the public path (computed at build time) 连得上 connects 活着 → 拒绝启动 alive → refuse to start 1000 ms 没反应 no answer in 1000 ms 也当作活着（宁可不删） treated as alive too (never delete) ECONNREFUSED · ENOENT · … 陈旧：改名 stale-xxxxxx，核对 dev/ino 后才 unlink stale: rename to stale-xxxxxx, unlink after a dev/ino check 改名期间路径被换掉 → 换回去并报错。 If the path changed meanwhile, put it back and fail. lstat(path) 的 dev/ino 等于记下的身份 → 改名 cleanup-xxxxxx → 再核对 → unlink； lstat(path) dev/ino equals the remembered identity → rename to cleanup-xxxxxx → recheck → unlink; 否则保留（可能是接替它的新服务器的 socket）。 otherwise leave it (it may belong to the server that replaced this one). maxPendingBytes 默认 4 × 帧上限，且必须 ≥ 帧上限 + 4；超过 → 断开慢的对端。 maxPendingBytes: 4 × the frame limit by default, ≥ limit + 4; beyond → drop the slow peer. close(final)：写完队列，end(final)；5000 ms 后强制 destroy。 close(final): drain the queue, end(final); force destroy after 5000 ms. README：对端认证是应用的策略，实验性的 Unix 传输不做认证——依靠 0700 目录与 0600 socket 的文件权限。 README: peer authentication is application policy and not implemented by the experimental Unix transport; it relies on 0700 / 0600 file permissions. 1 2 3 4 5 6 7 活性探测：连得上或 1000 ms 没反应都算“还活着”——宁可启动失败，也不删别人的 socket。 Liveness probe: connecting, or no answer within 1000 ms, both count as alive; better to fail start-up than delete another server's socket.
08 · 错误码与测试工具 08 · Error codes and the testing kit Run modes
错误码与测试工具 Error codes and the testing kit Errors that may cross the protocol boundary, and the ./testing kit ServerError 的子类（errors.ts） ServerError subclasses (errors.ts) ./testing：测试工具 ./testing: the testing kit src/testing wrong_server session_not_found session_ambiguous session_not_attached server_draining conformance · 20 protocol · 6 server · 6 unix · 6 listener · 2 unix-connection · 1 类 Class code 在哪里产生 Where it arises WrongServerError "Request was addressed to another server" handleRequest：在访问任何仓库之前 handleRequest: before any repository access SessionNotFoundError "Session was not found" 应用的 resolveSession 抛出 thrown by the app's resolveSession SessionAmbiguousError "Session ID matches more than one session" 应用的 resolveSession 抛出（一个 id 对应多个） thrown by the app's resolveSession (one id, several) SessionNotAttachedError "Session is not attached to this client" requireAttachment：路由不匹配或已过期 requireAttachment: mismatched or stale route ServerDrainingError "Server is draining" 关闭中 attach / 调用 / 删除，或连接已断 attach / call / removal while closing, or client gone 应用也可以直接 throw new ServerError(code, message)：code 还可以是 Chord 的 8 个 RemoteServiceErrorCode 之一。 Apps may also throw new ServerError(code, message); code may also be one of Chord's 8 RemoteServiceErrorCodes. 其它任何异常 → internal_error "Internal server error"，细节只交给 onError。 Anything else → internal_error "Internal server error"; details go only to onError. createTestServer(options) 未启动的 Server，默认 serverId …0001 an unstarted Server, default serverId …0001 TestServerHost 内存 ServerHost an in-memory ServerHost TestHarness · createTestServerServices 假的 Session 与服务器级服务 fake Sessions and server-wide services ProtocolTestClient 按帧收发、断言的协议客户端 a frame-level client for assertions connectUnixTestClient(path) 经真实 Unix socket 连接 connect over a real Unix socket Deferred<T> 手动控制时序 control timing by hand 传输实现可以拿它跑同一套一致性测试。 Transport implementations can run the same conformance tests. 测试（共 41 个 test） Tests (41 test cases) conformance.test.ts 覆盖多附着、过期路由、跨服务器请求、未知 / 有歧义的 Session、Harness 终止与重开、关闭时的清理失败。 conformance.test.ts covers multiple attachments, stale routes, cross-server requests, unknown / ambiguous Sessions, Harness termination and reopening, and cleanup failures on shutdown. 错误 errors 测试 testing wrong_server 在访问任何仓库之前检查：发错服务器的请求不会触发 resolveSession。wrong_server is checked before any repository access: a request for another server never triggers resolveSession.
扩展点 Extension points pi-coding-agent
server 的扩展点 Where to plug into server 1
ServerHost ServerHost
应用接入点：serverServices、resolveSession、openSession。 The app's entry point: serverServices, resolveSession, openSession.
2
RoutedSessionHandle · RoutedSessionAttachment RoutedSessionHandle · RoutedSessionAttachment
把 Session 放在任何地方：同进程、worker 进程或远端。 Host a Session anywhere: in process, in a worker process, or remotely.
3
ServerListener ServerListener
接入新的传输：start(accept) 交出已认证的字节连接。 Add a transport: start(accept) hands over authenticated byte connections.
4
ServerError(code, message) ServerError(code, message)
给客户端一个可识别的错误码和消息。 Give clients a recognisable error code and message.
5
./testing ./testing
用 createTestServer 与 ProtocolTestClient 测试你的宿主或传输。 Test your host or transport with createTestServer and ProtocolTestClient.