4.1 服务端与会话管理

@pi/server 把"一个 Agent 会话"变成"可被远程连接的服务"。这一节深入讲服务端如何组织连接、命令路由、会话生命周期、快照广播，以及如何处理握手与错误。

整体结构

PiServer（packages/server/src/server.ts:34）是核心对象，由几个协作者组成：

PiServer
 ├── listeners[]             监听器：TCP / Unix socket，负责"接客"
 ├── connections[]           当前所有活动连接（每个是一个 ConnectionState）
 ├── LiveSessionManager      会话管理器：路由命令、管理会话生命周期
 └── ServerSnapshotPublisher 快照发布器：广播"会话列表"的一致性

关键设计：server 不实现 Agent。它通过一个 PiSessionBackend 抽象来驱动真正的 Agent 会话 —— backend 才是真正跑 Agent 循环、执行工具的地方（比如 coding-agent 的会话运行时）。server 只是"协议翻译 + 状态广播"。

客户端 ──字节──► PiServer ──命令──► PiSessionBackend（真正跑 Agent）
                    │                   │
                    └──广播快照◄─────────┘ 事件

连接生命周期：一个状态机

每个连接（ConnectionState）都有一个阶段（stage），从接受到关闭走完一个状态机：

accept() → awaitingHello → handshaking → ready →（closing / closed）

1. 接受连接（accept）

accept()（server.ts:107）为每个新连接创建 ConnectionState，并设置握手超时（默认 5s）：

accept(connection): ByteConnectionHandler {
	const handshakeTimeout = setTimeout(() => {
		void this.failProtocol(state, { code: "invalid_request", message: "Handshake timeout" });
	}, this.handshakeTimeoutMs);

	state = {
		id: randomUUID(),
		decoder: new ClientMessageDecoder({ maxFrameLength }),
		sessionIds: new Set(),
		stage: "awaitingHello",      // ← 初始阶段：等第一帧 hello
		handshakeComplete: false,
		handshakeTimeout,
	};
	this.connections.add(state);
	return { onData: (c) => this.receive(state, c), onClose: ..., onError: ... };
}

2. 增量解码（receive）

receive()（server.ts:165）把收到的字节喂给 ClientMessageDecoder（增量切帧 + CBOR + 校验，见 3.2），解出一条条消息再分发：

receive(state, chunk) {
	let messages = state.decoder.push(chunk);   // 增量解码，可能解出多条
	for (const message of messages) {
		this.dispatchMessage(state, message);
	}
}

3. 分发与握手（dispatchMessage）

dispatchMessage()（server.ts:180）按阶段处理：

dispatchMessage(state, message) {
	if (state.stage === "awaitingHello") {
		if (message.type !== "hello") → failProtocol("must be hello");  // 第一帧必须是 hello
		state.stage = "handshaking";
		state.handshake = finishHandshake(state, message);
		return;
	}
	if (message.type === "hello") → failProtocol("hello only as first");  // 只能发一次
	if (state.stage === "ready") → handleRequest(state, message);          // 正常请求
	// handshaking 中 → 等握手完成再处理
}

finishHandshake()（server.ts:216）校验版本、取初始快照、发 ServerHello：

async finishHandshake(state, hello) {
	if (!isSupportedProtocolVersion(hello.version)) → failProtocol("version");  // 版本不匹配
	const snapshot = await this.snapshots.get(undefined, state);   // 取初始全貌
	await this.sendMessage(state, { type: "hello", version, connectionId, snapshot });
	state.stage = "ready";   // 握手成功，进入可处理请求阶段
}

强制第一帧：客户端必须先发 hello，且只能发一次。这保证了"连接 → 版本确认 → 才允许干活"。

请求-响应循环

握手进入 ready 后，客户端发 RequestEnvelope（带 id），handleRequest()（server.ts:247）路由到 LiveSessionManager.executeCommand，再把结果（带相同 id）回包成 ResponseEnvelope：

async handleRequest(state, envelope) {
	try {
		const result = await this.sessions.executeCommand(state, envelope.request);
		await this.sendMessage(state, { type: "response", id: envelope.id, ok: true, result });
	} catch (error) {
		await this.sendMessage(state, { type: "response", id: envelope.id, ok: false, error: toProtocolError(error) });
	}
}

id 配对：响应带请求的 id，客户端据此匹配（即使乱序也正确，见 4.2）。

命令路由：LiveSessionManager

packages/server/src/sessions.ts 的 LiveSessionManager 把命令路由到具体的会话运行时。它维护：

liveSessions: Map<id, LiveSession> —— 当前活跃的会话。
openingSessions: Map<id, Promise<LiveSession>> —— 正在打开的会话（防重入）。

executeCommand()（sessions.ts:52）按命令分发：

switch (command.command) {
	case "list":   return { sessions: await this.listSummaries(connection) };
	case "create": { const live = await this.acquire(id, () => backend.createSession(options)); ... }
	case "attach": { const live = await this.acquire(sessionId, () => backend.openSession(sessionId)); ... }
	case "prompt": { const live = this.requireAttached(connection, sessionId); ... }
	case "steer":  ...
	case "abort":  ...
	// ...
}

关键点：

create / attach 通过 backend 真正创建/打开会话（backend.createSession / backend.openSession），然后 attach 到当前连接。
prompt / steer / abort 等操作要求连接先 attach 到该会话（requireAttached），否则报错。
每个命令最后会 broadcastSnapshot 发出最新快照，并 broadcastServerSnapshot 通知会话列表变化。

快照发布：ServerSnapshotPublisher

packages/server/src/snapshots.ts 负责一致性。它维护一个自增的 revision，并用 broadcastQueue 把多次广播串行化（避免并发推乱序）：

broadcast() {
	const broadcast = this.broadcastQueue.then(() => this.performBroadcast());
	this.broadcastQueue = broadcast.catch(reportError);
	return broadcast;
}

async performBroadcast() {
	const ready = [...connections].filter(c => c.stage === "ready" && !c.disconnected);
	const revision = ++this.revision;
	const models = await backend.listModels();
	for (const conn of ready) {
		const snapshot = { ...(await this.get(models, conn)), revision };
		await sendMessage(conn, { type: "event", event: { type: "server_snapshot", snapshot } });
	}
}

为什么用 revision + 队列：

revision 让客户端能判断"我看到的快照是不是最新"。
broadcastQueue 保证广播按序执行，不会因并发调用把旧快照覆盖新快照。

会话的事件推送（主动）

服务端不只是被动响应。当会话有变化（后端 Agent 产生事件），它会主动推 EventEnvelope：

事件	携带内容	何时
`session_snapshot`	某会话的权威快照	会话状态大改
`session_progress`	增量（`item_started` / `assistant_delta` / `item_updated` / `item_finished`）	Agent 边界跑边推
`session_removed`	会话 id	会话被删除
`server_snapshot`	会话列表全貌	列表有增减

服务端把 backend（Agent 运行时）的细粒度事件，归一化成协议层的三类产物：ServerSnapshot（全局）、SessionSnapshot（单会话权威态）、TranscriptProgress（增量）。

传输无关：listeners 与 ByteConnection

服务端与具体传输方式解耦：

interface PiServerListener {
	start(onConnection: (conn: ByteConnection) => ByteConnectionHandler): Promise<void>;
	close(): Promise<void>;
	address?: string;
}

监听器可以是 TCP、Unix socket。ByteConnection 提供 onData / onClose / onError 回调。这样同一个 PiServer 可以跑在多种传输上（见 transports/）。

错误模型与关闭

PiServerError 与协议错误码对应（server.ts:346 的 toProtocolError）：

type ProtocolErrorCode = "version" | "busy" | "session_locked" | "not_found" | "invalid_request";

版本不匹配 → version
服务端忙 → busy
会话被其他客户端锁定 → session_locked
找不到会话 → not_found
非法请求 → invalid_request

关闭与清理：连接断开时 transportClosed → disconnect（从 connections 移除、通知会话 disconnect、并 broadcast 让其他客户端看到列表变化）。服务端整体 close 时，先关闭所有连接、再关闭会话与 backend。

小结

PiServer = 监听器 + 会话管理器 + 快照广播器。
连接是一个状态机：awaitingHello → handshaking → ready → closed，强制第一帧 hello + 版本校验。
命令由 LiveSessionManager 路由到 backend 会话运行时；id 配对响应。
快照（权威）+ 增量（优化），revision + 广播队列保证一致性。
server 不实现 Agent，只做协议翻译与广播。

真实源码位置

PiServer：packages/server/src/server.ts:34
连接状态机 / 握手：packages/server/src/server.ts:107,180,216
命令路由：packages/server/src/sessions.ts:52
快照广播：packages/server/src/snapshots.ts:44
错误映射：packages/server/src/server.ts:346

下一步：4.2 客户端与连接。

4.1 服务端与会话管理 ​

整体结构 ​

连接生命周期：一个状态机 ​

1. 接受连接（accept） ​

2. 增量解码（receive） ​

3. 分发与握手（dispatchMessage） ​

请求-响应循环 ​

命令路由：LiveSessionManager ​

快照发布：ServerSnapshotPublisher ​

会话的事件推送（主动） ​

传输无关：listeners 与 ByteConnection ​

错误模型与关闭 ​

小结 ​