3.1 字节定帧（Framing）

到目前为止，Agent 的所有消息都是内存里的 JS 对象。要让 Agent"跨进程"（比如编码 Agent 作为服务端、TUI 作为客户端），就必须把它们序列化成字节流。pi 采用 长度前缀（length-prefix）定帧 的方式，把"一条消息"和"一条消息"在字节流里切分开。

为什么需要定帧

网络/管道是字节流，没有"消息"的边界。你发两条消息，对方可能一次收到、也可能拆成几段收到。定帧解决的问题是：如何从字节流里可靠地切出一条条完整的消息。

pi 的方案：每条消息前面加 4 字节大端无符号长度。

┌──────────────┬──────────────────────────────┐
│  4 字节长度    │        payload（消息内容）      │
│  (big-endian) │                              │
└──────────────┴──────────────────────────────┘

编码：encodeFrame

packages/protocol/src/framing.ts:28：

export function encodeFrame(payload: Uint8Array): Uint8Array {
	const frame = new Uint8Array(4 + payload.byteLength);
	const length = payload.byteLength;
	frame[0] = length >>> 24;   // 最高位字节
	frame[1] = length >>> 16;
	frame[2] = length >>> 8;
	frame[3] = length;          // 最低位字节
	frame.set(payload, 4);      // 内容跟在长度后面
	return frame;
}

长度用 32 位无符号，所以单条消息最大约 4 GB（MAX_UINT32）。实际还有更小的上限 DEFAULT_MAX_FRAME_LENGTH = 16MB（framing.ts:6）做防御。

位运算在做什么？—— 把一个长度整数拆成 4 个字节

上面的 >>> 24 / >>> 16 / >>> 8 可能有点难懂。它们的全部目的：把 length 这个整数，拆成 4 个 0~255 的字节（因为 Uint8Array 每个元素只能存 0~255）。

先理解 >>>（无符号右移）：把所有位整体向右移 n 位，左边补 0，丢掉右边溢出的位。右移 8 位 = "丢掉低 8 位，只看高位的部分"。

用 length = 1234（二进制 00000100 11010010）演示，把它拆进 frame[0..3]。1234 的 32 位完整表示是 4 个 8 位字节：

1234 = 00000000 00000000 00000100 11010010   (32 位，4 个字节)
       └──────┘ └──────┘ └──────┘ └──────┘
        frame[0] frame[1] frame[2] frame[3]

写法	做了什么	结果	存到
length >>> 24	右移 24 位，只剩最高 8 位	0（00000000）	frame[0]
length >>> 16	右移 16 位，取次高 8 位	0（00000000）	frame[1]
length >>> 8	右移 8 位，取次低 8 位	4（00000100）	frame[2]
length	不移动，取最低 8 位	210（11010010）	frame[3]

验证：0×2²⁴ + 0×2¹⁶ + 4×2⁸ + 210 = 1024 + 210 = 1234 ✅

为什么是 24 / 16 / 8？ 因为 4 个字节的权重分别是 2²⁴、2¹⁶、2⁸、2⁰。从高到低：最高 8 位右移 24 位、次高 8 位右移 16 位、次低 8 位右移 8 位、最低 8 位不移动。

为什么右移后能存进 frame？ >>> 24 后只剩最高 8 位，值域 0~255；同理 >>> 16 / >>> 8 后 Uint8Array 自动截断保留低 8 位。所以 4 个结果都落在 0~255，安全。

用 length = 5（"hello" 有 5 字节）快速验证：5 = ...00000101，四个右移结果分别是 0, 0, 0, 5，帧头就是 [0,0,0,5]。

解码端 FrameDecoder 用相反的位运算拼回来：

length = (frame[0] << 24) | (frame[1] << 16) | (frame[2] << 8) | frame[3];

一句话：encodeFrame 的位运算 = 把长度整数拆成 4 个字节（>>> 24/16/8 分别取最高/次高/次低 8 位，最后一个不移动取最低 8 位），解码端再按反方向拼回。

解码：FrameDecoder

解码要处理黏包（一个 chunk 含多条消息）和半包（一条消息被拆到多个 chunk）。FrameDecoder（framing.ts:58）是一个增量状态机。

完整实现（可直接运行）

class FrameDecoder {
	// 缓存"还没凑够"的 header（4 字节）和 payload
	private header = new Uint8Array(4);
	private headerLen = 0;                // 已攒了几个 header 字节
	private payloadBuf: number[] = [];    // 已攒的 payload 字节
	private expected: number | undefined; // 本帧 payload 一共多长（undefined = 还在读 header）
	private payloadLen = 0;               // 已攒了多少 payload 字节

	// 输入任意字节块，返回解出的完整帧数组
	push(chunk: Uint8Array): Uint8Array[] {
		const frames: Uint8Array[] = [];
		let i = 0;

		while (i < chunk.byteLength) {
			// 阶段一：先攒够 4 字节 header，解析出"本帧 payload 多长"
			if (this.expected === undefined) {
				const take = Math.min(4 - this.headerLen, chunk.byteLength - i);
				this.header.set(chunk.subarray(i, i + take), this.headerLen);
				this.headerLen += take;
				i += take;
				if (this.headerLen < 4) break;   // header 还没攒够，等更多数据

				this.expected =
					(this.header[0]! << 24) | (this.header[1]! << 16) | (this.header[2]! << 8) | this.header[3]!;
				this.headerLen = 0;
				this.payloadBuf = [];
				this.payloadLen = 0;
				if (this.expected === 0) {        // 空帧
					frames.push(new Uint8Array());
					this.expected = undefined;
				}
				continue;
			}

			// 阶段二：按"本帧长度"攒 payload，攒够就弹出一帧
			const take = Math.min(this.expected - this.payloadLen, chunk.byteLength - i);
			for (let k = 0; k < take; k++) this.payloadBuf.push(chunk[i + k]!);
			this.payloadLen += take;
			i += take;

			if (this.payloadLen === this.expected) {   // 这一帧攒够了
				frames.push(new Uint8Array(this.payloadBuf));
				this.expected = undefined;
				this.payloadBuf = [];
				this.payloadLen = 0;
			}
		}
		return frames;
	}

	// 流结束时调用：若还有没读完的 header/payload，说明传输被截断
	end(): void {
		if (this.headerLen !== 0 || this.expected !== undefined) throw new Error("截断帧");
	}
}

核心思路：expected 是状态机的"旗帜" —— 为 undefined 时在读 header，读满 4 字节就得到本帧长度并进入"读 payload"；攒够 expected 个字节就弹出一整帧，回到读 header。不管 chunk 怎么切，每次 push 吐出的都是完整帧。

缩略版（理解骨架）

class FrameDecoder {
	push(chunk: Uint8Array): Uint8Array[] {
		// 边读边切：
		// 1. 先攒够 4 字节 header，解析出本帧长度
		// 2. 再按长度攒 payload
		// 3. 攒够整帧就弹出一条，继续读下一个
	}
	end(): void { /* 结束时若还有未读完的 header/payload → 抛 "截断帧" */ }
}

它内部用 header 缓冲区 + payloadBuf 数组，把任意大小的 chunk 增量切成完整帧。不管 chunk 怎么切，push 出来的都是完整帧。完整可运行版见下节示例，以及工程化项目 pi-principles/src/protocol/framing.ts。

完整示例

import { encodeFrame, FrameDecoder } from "@earendil-works/pi-protocol";

// 编码两条消息
const a = encodeFrame(new TextEncoder().encode("hello"));
const b = encodeFrame(new TextEncoder().encode("world"));

// 故意把字节流切得乱七八糟（模拟网络）
const mixed = new Uint8Array([...a.slice(0, 3), ...a.slice(3), ...b]);

const decoder = new FrameDecoder();
const frames = decoder.push(mixed);   // 一条 chunk 里解出两条完整帧
for (const f of frames) {
	console.log(new TextDecoder().decode(f)); // "hello" "world"
}

定帧与 CBOR 的关系

定帧只负责"切出完整的一段字节"，这段字节内部是什么格式由 CBOR 决定（见 3.2）。两者叠加就是 pi 的传输层：

字节流
  → FrameDecoder 切帧      （拿到一段段完整 payload）
  → decodeCbor 解 CBOR     （payload → JS 对象）
  → parseSchema 校验       （确认是合法的协议消息）

小结

• 定帧 = 每条消息前置 4 字节大端长度。
• FrameDecoder 是一个增量状态机，能处理黏包与半包。
• 定帧负责"切消息"，CBOR 负责"解内容"，Schema 负责"验合法性"。

真实源码位置

• 常量与上限：packages/protocol/src/framing.ts:1-6
• encodeFrame：packages/protocol/src/framing.ts:28
• FrameDecoder：packages/protocol/src/framing.ts:58

面试角度：为什么这样定帧

Q1：为什么用"长度前缀定帧"而不是用分隔符（如换行）？ 因为消息内容（CBOR）里可能包含任意字节，包括分隔符本身。用分隔符会出现"消息内的分隔符被误判为边界"的问题。长度前缀则完全基于"我告诉你这条多长"，不依赖内容，天然免疫内容里的任何字节。

Q2：为什么 FrameDecoder 要设计成"增量状态机"？ 因为网络/管道是字节流，一条消息可能被拆成多段到达（半包），一个包也可能含多条消息（黏包）。增量状态机记住"当前在哪一帧、还差多少字节"，无论字节怎么切，都能正确切出完整帧。这是可靠传输的基础。

Q3：为什么长度用 4 字节大端无符号？ 4 字节可表达约 4GB，足够容纳协议消息；大端（big-endian）是网络字节序的通用约定，方便跨平台解析。同时 pi 又设了更小的 DEFAULT_MAX_FRAME_LENGTH（16MB）做防御，防止恶意超大帧耗尽内存。

Q4：为什么 end() 时若还有未读完的帧要抛"截断帧"？ 因为正常结束的流不应出现"读到一半"的状态。抛错能及早暴露"对端没发完整就断了"的异常情况，而不是静默吞掉导致数据不完整。

下一步：3.2 CBOR 编解码。