数据截至 (上游 commit eb980a5c9eea)
加密同步流与不可读的 RPC 中继
30 秒导读： 手机、服务器、终端里的 CLI 三方要实时对上话，但服务器没有密钥、读不懂任何一个字节。 这一章讲 Happy 怎么在「中间人是瞎子」的前提下，把消息排好序、把状态改对、还能让手机远程喊 CLI 执行一条命令。
本章只讲传输层。密钥怎么来、配对怎么握手见 01 端到端加密与配对握手； 消息解密后的内容怎么解释见 04 远程回合 与 06 多种 agent 后端与 App 端的消息归一。
1. 这是什么（零基础也能懂）
一句话定义： Happy 的传输层是一条「服务器看不懂内容的三端同步管道」。
它要解决的场景。 你在笔记本终端里跑着 happy claude，人走了；在地铁上掏出手机想接着聊。
手机和笔记本不在一 个网络里，只能靠中间那台服务器转发。可 Happy 的卖点是端到端加密——
服务器必须在读不懂内容的情况下，把事情办对。
「办对」具体是三件事：
| 要办的事 | 白话 | 难在哪 |
|---|---|---|
| 消息不乱序、不丢 | 手机上看到的对话顺序，和终端里发生的顺序一致 | 网络会断，socket 会掉包，重连后得知道「我落下了哪几条」 |
| 状态改不打架 | 手机改标题的同时 CLI 也在改，不能互相覆盖 | 两边都在改同一个值，而服务器不能「合并」密文 |
| 手机能指挥电脑 | 在手机上点一下，笔记本上真的跑了 git status | 请求要精确送到「那一台机器上的那一个会话」，且参数不能让服务器看见 |
一句话直觉： 把服务器当成一个不识字的邮局。 信封上写着房间号（路由信息，明文），信纸上是密文。邮局只认房间号，永远不拆信。
2. 顶层全景：一条 Socket.IO 通道，三种业务流
2.1 三端怎么接进来
三种客户端连的是同一个 Socket.IO 端点（path: '/v1/updates'），靠握手里的 clientType 区分身份：
happy-app(手机/网页) clientType: 'user-scoped' ──┐
│ ┌──────────────────────────┐
happy-cli(某个会话) clientType: 'session-scoped' ──┼─────►│ happy-server │
+ sessionId │ │ Socket.IO /v1/updates │
│ │ + REST /v3/sessions/... │
happy daemon(某台机器) clientType: 'machine-scoped' ──┘ │ │
+ machineId │ 只见 base64 密文，不解密│
└──────────────────────────┘
鉴权发生在 Socket.IO 的 middleware 里而不是 connection 回调里，这是个有意的选择：
packages/happy-server/sources/app/api/socket.ts:78-121 的注释写明了理由——
如果把异步的 auth.verifyToken 放在 connection 回调里，客户端的 connect 事件会先于 handler 挂载触发，
那段窗口里到达的 rpc-register、rpc-call 会被静默丢弃。
2.2 三条业务流
同一条 socket 上跑着语义完全不同的三条流，外加一条「丢了也无所谓」的易失流：
| 通道 | 传什么 | 载体 | 语义 |
|---|---|---|---|
| ① 有序消息日志 | 对话正文（密文） | 写读走 REST /v3/sessions/:id/messages；推送走 socket update 事件、body.t = 'new-message' | 只追加，每条一个会话内自增的 seq |
| ② 状态版本号 | 会话元数据 / agentState（密文） | socket update-metadata / update-state（带 ack）；推送走 update、body.t = 'update-session' | 单值覆盖，乐观并发 |
| ③ 实时 RPC | 调用参数与返回值（密文） | socket rpc-register / rpc-call / rpc-request | 一问一答，不落库 |
| ④ 易失事件 | 在线、思考中、token 用量 | socket ephemeral 事件 | 尽力而为，丢了就丢了 |
为什么「消息」要用 REST 写而不是 socket 发？ 因为 ① 需要可重放的持久顺序：
写入要落库、要分配 seq、要能按 seq 回头补拉。socket 只负责「提速」——
告诉你有新东西了，顺便捎上内容；真正的正确性由 REST 兜底。这是本章最重要的一条主线，下一节展开。
3. 通道一：有序消息日志
它要解决的小问题： 断线重连之后，我怎么知道自己落下了哪几条消息？
3.1 思路：给每条消息发一个号码牌
服务器给每个会话维护一个自增计数器，每写入一条消息就发一个号。
allocateSessionSeqBatch（packages/happy-server/sources/storage/seq.ts:30-45）用一次
session.seq += count 的原子自增拿到一整段连续号码，再切成 [start … end] 分给这一批消息——
一次数据库往返搞定整批，而不是逐条 allocateSessionSeq。
客户端只需记住一个数：lastSeq。重连后拿着它问服务器「比这个大的都给我」。
3.2 收发双循环
CLI 端把「收」和「发」做成两个互不阻塞的独立循环，各自由一个 InvalidateSync 驱动
（packages/happy-cli/src/api/apiSession.ts:242-243）：
┌──────────────── receiveSync ────────────────┐
socket 'connect' ─┤ │
seq 断号 ─┤ invalidate() ──► fetchMessages() │
│ GET /v3/.../messages │
└─────────────────────────────────────────────┘
┌──────────────── sendSync ───────────────────┐
enqueueMessage() ─┤ invalidate() ──► flushOutbox() │
│ POST /v3/.../messages │
└─────────────────────────────────────────────┘
InvalidateSync（packages/happy-cli/src/utils/sync.ts:14-27，_doSync 在 55-73）是个只有 70 行的小类，
但它是整个同步层的地基。它的语义是：「我脏了，请重跑」，而不是「请跑一次」。
- 正在跑的时候再
invalidate()，不会并发第二次，只置一个_invalidatedDouble标记； - 当前这轮跑完，看到标记就再跑一轮，然后清标记（
sync.ts:66-72）； - 整个
_command外面包着backoff（packages/happy-cli/src/utils/time.ts:43），失败自动重试。
所以调用方永远不用关心「现在能不能跑」「跑失败怎么办」，只管喊脏。
这段演示它的核心想法：
// 示意，非源码：两条循环，socket 只是提速器
const receiveSync = new InvalidateSync(() => fetchMessages()); // 把服务器的新消息拉下来
const sendSync = new InvalidateSync(() => flushOutbox()); // 把本地待发队列推上去
socket.on('connect', () => receiveSync.invalidate()); // 一连上就补齐
socket.on('update', (u) => {
if (u.body.message.seq === lastSeq + 1) applyLocally(u); // 连号：直接用，零 HTTP
else receiveSync.invalidate(); // 断号：别猜，回去拉
});
重点看最后两行：socket 推送只在「号码正好接上」时被信任，否则一律退回 REST 全量补拉。
3.3 真实实现：快路径与回退
CLI 的 update 处理器（apiSession.ts:304-360）里，new-message 分支只有一个门槛：
if (typeof messageSeq !== 'number' || messageSeq !== this.lastSeq + 1 || data.body.message.content.t !== 'encrypted') {
this.receiveSync.invalidate();
return;
}
—— apiSession.ts:313-318。序号不连、或内容不是 encrypted 密文封套，就不猜，直接把 receiveSync 喊脏。
只有严格连号时才走下面的 decrypt + routeIncomingMessage + this.lastSeq = messageSeq（apiSession.ts:319-329）。
App 端是同一套判断：packages/happy-app/sources/sync/sync.ts:2278-2291，
incomingSeq === currentLastSeq + 1 才 enqueueMessages，否则 this.getMessagesSync(sid).invalidate()。
两端独立实现、结论一致，说明这是设计约定而非巧合。
3.4 回退路径：fetchMessages 的分页与防死循环
fetchMessages（apiSession.ts:581-643）是一个 after_seq 向前翻页的 while 循环：
- 每轮
GET /v3/sessions/:id/messages?after_seq=<lastSeq>&limit=100； - 服务端
take: limit + 1多取一条来判断hasMore（packages/happy-server/sources/app/api/routes/v3SessionRoutes.ts:114-120）； - 停滞保护：若
hasMore为真但这一页的maxSeq没有前进，直接 break，避免无限循环 （apiSession.ts:631-637）；App 端有一份对应的fetchForwardSince，同样的保护在sync.ts:2036-2040。
单条消息解密失败只跳过这一条并记日志（apiSession.ts:620-626），不会让整轮同步崩掉。
重连时还有个开关： skipExistingMessages()（apiSession.ts:919-921）置位后，下一次 fetchMessages 只推进
lastSeq、不投递消息（apiSession.ts:582-587, 611）——用于「接管一个已存在的会话」时不把历史重放一遍，
细节见 03 本地与远程双模态。
3.5 发送侧：localId 去重 + 批量
enqueueMessage（apiSession.ts:675-684）做三件事：加密、给一个 randomUUID() 当 localId、喊 sendSync 脏。
const encrypted = encodeBase64(encrypt(this.encryptionKey, this.encryptionVariant, content));
this.pendingOutbox.push({ content: encrypted, localId: randomUUID() });
—— apiSession.ts:676-680。localId 是客户端生成的幂等键，这是「重试安全」的全部秘密。
flushOutbox（apiSession.ts:647-673）按 MAX_OUTBOX_BATCH_SIZE = 50（apiSession.ts:645）切批，
而且从队尾切：
const batchStart = this.pendingOutbox.length - batchSize;
const batch = this.pendingOutbox.slice(batchStart);
—— apiSession.ts:652-653。注释说明了意图（apiSession.ts:648-649）：先发最新的，让用户立刻看到近期活动，
积压的旧消息在后续批次里回填。
服务端的去重在一个事务里完成（v3SessionRoutes.ts:158-211）：
- 请求体内先按
localId去重，同一批里重复的只留第一条（v3SessionRoutes.ts:148-155）； - 事务内查
sessionMessage里已存在的localId（v3SessionRoutes.ts:160-172）； - 只给新的那些分配 seq 并插入，已存在的原样回显（
v3SessionRoutes.ts:181-205）。
于是客户端可以无脑重试整批，不会产生重复消息。
广播在事务之外（v3SessionRoutes.ts:215-234）：逐条 allocateUserSeq 拿一个用户级 update 序号，
buildNewMessageUpdate 打包，再 emitUpdate 到 all-interested-in-session 房间。
3.6 一个可预期的浪费
这条 REST 写入路径没有 skipSenderConnection（v3SessionRoutes.ts:229-233）——
HTTP 请求本来就没有对应的 socket 可跳过。所以 CLI 自己 POST 上去的消息，还会经 socket 原路广播回它自己的会话房间。
CLI 收到后，UserMessageSchema 和 FileEventMessageSchema 都不匹配，落到 this.emit('message', message)
（apiSession.ts:551-579）；生产代码里没有任何地方订阅 ApiSessionClient 的 'message' 事件，等于丢弃 (inferred)。
3.7 CLI 与 App 的取历史策略不同
CLI（apiSession.ts:fetchMessages） | App（sync.ts:fetchMessages） | |
|---|---|---|
| 首次拉取方向 | 从 after_seq=0 向前 | 从 before_seq=2147483647 向后取最新一页 |
| 用意 | CLI 只关心「新指令」，历史在本地转录文件里 | 打开长会话不能卡在拉全量历史上 |
| 关键符号 | fetchMessages | fetchInitialLatestPage（sync.ts:1982-2011）、loadOlderMessages（sync.ts:2073） |
| 触发时机 | 连接建立 / seq 断号 | onSessionVisible（sync.ts:292），按会话懒加载 |
服务端为此提供了互斥的两个游标参数，契约写在 v3SessionRoutes.ts:8-25：
after_seq 正向升序、before_seq 反向降序，两者不能同时给。
4. 通道二：状态版本号（乐观并发）
它要解决的小问题： 手机和 CLI 同时改会话标题，谁赢？怎么让输的那个知道自己输了？
4.1 思路：比对版本号，输了就重来
会话上有两个可变值，各自带一个版本号：
| 字段 | 版本号 | 装什么 | 谁常改 |
|---|---|---|---|
metadata | metadataVersion | 路径、标题、summary、lifecycleState | 两端都改 |
agentState | agentStateVersion | 权限请求、controlledByUser 等运行时状态 | CLI 为主 |
写入协议是 compare-and-set：客户端把「我以为的版本」一起发上去，服务器只在版本相等时才写。
4.2 服 务端：两道防线
sessionUpdateHandler（packages/happy-server/sources/app/api/socket/sessionUpdateHandler.ts）先读一次比对
（:33-38），再用一句带版本条件的 updateMany 落库：
const { count } = await db.session.updateMany({
where: { id: sid, metadataVersion: expectedVersion },
data: { metadata, metadataVersion: expectedVersion + 1 }
});
if (count === 0) { callback({ result: 'version-mismatch', ... }); return null; }
—— sessionUpdateHandler.ts:40-50。第一次读比对只是快速失败；真正的防线是 where 里的版本条件，
count === 0 说明在读和写之间被别人抢先了。update-state 是完全对称的一份（sessionUpdateHandler.ts:106-116）。
写成功后广播 buildUpdateSessionUpdate（sessionUpdateHandler.ts:58-64），只带新值和新版本号。
4.3 客户端：AsyncLock + backoff 组成的重试
CLI 侧的 updateMetadata（apiSession.ts:923-942）长这样（结构演示）：
// 示意，非源码：带版本号的读-改-写，失败就重跑
async function updateMetadata(mutate) {
const next = mutate(localMetadata); // 在本地副本上改
const ans = await socket.emitWithAck('update-metadata', {
expectedVersion: localVersion, // 我以为的版本
metadata: encrypt(next)
});
if (ans.result === 'version-mismatch') {
localVersion = ans.version; // 服务器更新，先接受
localMetadata = decrypt(ans.metadata);
throw new Error('retry'); // 抛出去，交给 backoff 重跑整个函数
}
}
三个真实细节：
throw是故意的。apiSession.ts:936真的throw new Error('Metadata version mismatch')， 外层backoff（apiSession.ts:925）捕获后重跑——而重跑会用刚刚同步下来的新版本重新执行handler， 所以这是「rebase 后重放」，不是「覆盖」。metadataLock/agentStateLock两把独立的AsyncLock（apiSession.ts:209-210）保证同一进程内的更新串行， 否则两个并发的updateMetadata会用同一个expectedVersion互相踩。AsyncLock实现见packages/happy-cli/src/utils/lock.ts:inLock。- ack 的三态：
result | 含义 | 客户端动作 |
|---|---|---|
success | 写入成功 | 用回包里的值和 version 更新本地（apiSession.ts:928-930） |
version-mismatch | 版本过期 | 若回包版本更高就先吸收，然后抛错触发重试（apiSession.ts:931-936） |
error | 硬错误 | 静默忽略（apiSession.ts:937-939） |
4.4 接收侧：只接受更大的版本
update-session 推送到达时，CLI 用 > 而不是 !== 过滤（apiSession.ts:331、346）：
if (data.body.metadata && data.body.metadata.version > this.metadataVersion) { ... }
乱序到达的旧版本推送会被直接扔掉。App 端不做这个比较，而是无条件采用推送里带的版本
（sync.ts:2320-2331）——因为 App 是 user-scoped 单点消费者，并发写压力主要在 CLI 侧 (inferred)。
4.5 archived：一个走状态通道的「退出信号」
Happy 没有专门的「关闭会话」消息类型；归档是 metadata 里的一个字段。
CLI 在解密新 metadata 后检查 lifecycleState：
if (meta?.lifecycleState === 'archiveRequested' || meta?.lifecycleState === 'archived') { ... this.emit('archived'); }
—— apiSession.ts:335-344。runClaude.ts:617-620 订阅这个事件并执行 cleanup()，进程退出。
例外由 suppressNextArchiveSignal()（apiSession.ts:915-917）提供：重连场景下要吃掉一次归档信号，
避免刚接管就自杀。这条路径属于 03 本地与远程双模态 的范畴。
4.6 顺带一提：易失通道
session-alive 用 socket.volatile.emit 发（apiSession.ts:859-864）——volatile 表示「缓冲区满就丢弃」。
服务端收到后写进 activityCache 并广播 ephemeral（sessionUpdateHandler.ts:140-190）。
「在线」「正在思考」「token 用量」全走这条通道，因为它们过期即无价值，不值得占用有序日志的 seq。
5. 通道三：不可读的 RPC 中继
它要解决的小问题： 手机上点「看一下 git 状态」，怎么让那一台笔记本上的那一个会话去执行， 而且服务器不知道执行了什么？
5.1 思路：把 Socket.IO 房间当成路由表
服务器不维护任何「谁能提供什么方法」的注册表。它复用 Socket.IO 已有的房间机制：
- CLI 声明「我能处理
M」→socket.join('rpc:<userId>:<M>')； - 手机要调用
M→ 服务器io.in('rpc:<userId>:<M>').fetchSockets()找人； - CLI 掉线 → Socket.IO 自动把它移出所有房间。
rpcHandler.ts:1-14 的顶部注释把这个设计的收益说得很直白：
没有 Redis 键、没有 TTL、没有 Lua 脚本、没有保活刷新路径。断线清理是免费的。
文件末尾 rpcHandler.ts:258-259 再次强调：没有 disconnect handler，因为不需要。
5.2 方法名里的作用域前缀
关键在于方法名不是 bash，而是 <sessionId>:bash。
RpcHandlerManager 的构造参数 scopePrefix 决定了这个前缀
（packages/happy-cli/src/api/rpc/RpcHandlerManager.ts:137-139）：
private getPrefixedMethod(method: string): string {
return `${this.scopePrefix}:${method}`;
}
会话客户端传的是 sessionId（apiSession.ts:246-251），守护进程传的是 machineId（见
05 守护进程与机器）。调用方按同样规则拼：
App 的 sessionRPC 拼 ${sessionId}:${method}（packages/happy-app/sources/sync/apiSocket.ts:152-167），
machineRPC 拼 ${machineId}:${method}（apiSocket.ts:172-187）。
于是「路由到哪台机器的哪个会话」这件事，完全由字符串拼接和房间名解决，服务器一行业务逻辑都不用写。
它只在打指标时用 baseMethodName（rpcHandler.ts:75-78）把前缀切掉，好让 bash 的调用量能聚合统计。
CLI 侧注册的方法在 packages/happy-cli/src/modules/common/registerCommonHandlers.ts：
bash（:176）、readFile（:262）、writeFile（:282）、listDirectory（:348）、
getDirectoryTree（:407）、ripgrep（:493）、difftastic（:523）。
5.3 一次调用的完整时序
手机(user-scoped) 服务器 CLI(session-scoped)
│ │ │
│ │◄──── 'rpc-register' ─────────┤ join rpc:<uid>:<sid>:bash
│ │ │
├─ 'rpc-call' ──────────►│ fetchSockets(房间) │
│ {method:"<sid>:bash", │ ├ 空 → waitForRoomMember │
│ params: 密文} │ └ 命中 → 取 targets[0] │
│ ├──── 'rpc-request' 原样转发 ──►│ 解密 params
│ │ │ 跑 handler
│ │◄──── ack(密文 result) ────────┤ 加密 result
├◄─ ack {ok:true, │ │
│ result: 密文} │ │
服务器只做了一次字段搬运： rpcHandler.ts:219-220 把 { method, params } 原封不动地
emitWithAck('rpc-request', ...) 给目标 socket，拿回来的东西塞进 { ok: true, result: response }
（rpcHandler.ts:243）。params 和 result 从头到尾是 base64 密文。
CLI 侧的解密 / 加密在 RpcHandlerManager.handleRequest（RpcHandlerManager.ts:64-96）：
进来 decrypt(...decodeBase64(request.params))，出去 encodeBase64(encrypt(..., result))。
连「方法不存在」和异常信息也是加密返回的（RpcHandlerManager.ts:70-75、89-95）——
服务器连「调用失败了没有」都看不出来。
5.4 找人：跨副本 + 重连宽限窗
服务器可能有多个进程副本，目标 socket 未必连在本副本上。
REDIS_URL 存在时挂上 @socket.io/redis-streams-adapter（socket.ts:50-52），
fetchSockets() 和跨副本 ack 都由它提供。
查找分两步（rpcHandler.ts:178-192）：
- 先用 2 秒超时试一次
fetchRoomSockets； - 空了就进
waitForRoomMember（rpcHandler.ts:109-126），在RPC_RECONNECT_GRACE_MS = 15_000的窗口内轮询。
轮询用指数退避的超时而不是固定超时：RPC_LOOKUP_FETCH_TIMEOUTS_MS = [2_000, 4_000, 8_000]
（rpcHandler.ts:23）。注释解释得很清楚（rpcHandler.ts:19-22）：Redis 慢的时候，
固定短超时会让「超时 → 重试 → 又超时」的请求洪水放大流上的压力；逐次拉长反而更容易成功。
这个宽限窗的现实意义：daemon 正在重连的那两秒里发来的 RPC，不会直接报 "not available"。
5.5 判死：ack 和存活轮询赛跑
这是本章最巧的一处。rpcHandler.ts:205-241：
const ackPromise = target.timeout(RPC_CALL_TIMEOUT_MS).emitWithAck('rpc-request', { method, params });
// ... 同时起一个 presencePoll，发现 target 离开房间就 throw
const response = await Promise.race([ackPromise, presencePoll]);
为什么需要这个赛跑？ 注释（rpcHandler.ts:206-218）写明：
emitWithAck 根本不知道目标 socket 已经死了。daemon 所在的 pod 被杀时，
cluster adapter 的 BROADCAST 请求会一直等一个永远不会来的 BROADCAST_ACK，
只能等到用户设的 30 秒（RPC_CALL_TIMEOUT_MS）才超时；
adapter 自己的心跳探活要 ~10 秒，而且不会主动取消已排队的广播。
轮询 fetchSockets 是唯一能「发现目标不见了」并快速中止的办法（~2-4 秒）。
两个防误判的细节：
- 轮询的 fetch 超时压到
RPC_PRESENCE_FETCH_TIMEOUT_MS = 500（rpcHandler.ts:27）， 免得每次轮询自己被拖满 adapter 的 10 秒心跳超时； - 要连续 2 次没看到目标才判死（
rpcHandler.ts:229-233），单次 Redis 抖动不算数。
结果被分成 success / target_disconnected / timeout / not_available 等标签打进 Prometheus
（rpcHandler.ts:36-64，finish() 在 :164-169）。
5.6 重连后谁来补注册？
RPC 房间是 socket 级的，断线即清空。所以 CLI 在每次 connect 时重放全部注册：
onSocketConnect(socket: Socket): void {
this.socket = socket;
for (const [prefixedMethod] of this.handlers) socket.emit('rpc-register', { method: prefixedMethod });
}
—— RpcHandlerManager.ts:98-103，由 apiSession.ts:282 的 connect 回调调用。
registerHandler 在未连接时只入本地 Map、不发包（RpcHandlerManager.ts:45-47），等下次连上统一补。
6. 服务器为什么读不懂：路由只认房间名，负载只是密文
前面反复说「服务器只搬密文」。证据集中在 packages/happy-server/sources/app/events/eventRouter.ts。
6.1 房间命名与路由表
addConnection（eventRouter.ts:234-249）按连接类型把 socket 塞进房间：
所有连接 → user:<userId>
user-scoped → user:<userId>:user-scoped
session-scoped → user:<userId>:session:<sessionId>
machine-scoped → user:<userId>:machine:<machineId>
RecipientFilter（eventRouter.ts:43-47）是业务侧表达「发给谁」的语言，
getRoomsForFilter（eventRouter.ts:320）把它翻译成房间名：
| RecipientFilter | 展开成的房间 | 典型用途 |
|---|---|---|
all-user-authenticated-connections | user:<uid> | 账号级更新 |
user-scoped-only | user:<uid>:user-scoped | 只给 App：在线状态、机器上下线 |
all-interested-in-session | user:<uid>:session:<sid> + user:<uid>:user-scoped | 新消息、会话状态变更 |
machine-scoped-only | user:<uid>:machine:<mid> + user:<uid>:user-scoped | 机器状态变更 |
后两者是并集，靠 Socket.IO 自身对重复目标去重（eventRouter.ts:326-331 注释）。
skipSenderConnection 存在时改用 socket.broadcast.to(rooms)（eventRouter.ts:344-348）。
6.2 builder 函数里没有一处解密
buildNewMessageUpdate（eventRouter.ts:390-415）接的 content 类型是 SessionMessageContent，
落库时的形状是 { t: 'encrypted', c: <base64> }（v3SessionRoutes.ts:190-193），
builder 只是把它连同 id / seq / localId / 时间戳原样塞进 body。
buildUpdateSessionUpdate（eventRouter.ts:417-429）更干脆：metadata 和 agentState
都是 { value: string; version: number }，value 就是 base64 密文。
服务器唯一读得懂的东西是：谁（userId）、哪个会话（sessionId）、第几条（seq）、什么时候（timestamp）。 这就是 Happy 的元数据泄露面——内容保密，通信图谱不保密。
7. App 侧的对照结构
App（packages/happy-app/sources/sync/）是同一套思想的另一份实现，差异主要来自「一个 App 要同时看住 N 个会话」。
| 关注点 | CLI（apiSession.ts） | App（sync.ts / apiSocket.ts） |
|---|---|---|
| 连接类型 | session-scoped + sessionId | user-scoped（apiSocket.ts:88-107） |
| 重连策略 | reconnection: false，自己写 startSmartReconnect，还要问 shouldReconnect()（网络 + 笔记本盖子）（apiSession.ts:1004-1025） | Socket.IO 内建重连，reconnectionAttempts: Infinity |
| 同步器数量 | 全局 2 个：sendSync / receiveSync | 每会话一份：messagesSync、sendSync 两个 Map（sync.ts:112-114） |
| 重连后动作 | receiveSync.invalidate()（apiSession.ts:283） | onReconnected 里刷 sessions / machines / artifacts / friends / feed，消息则留给懒加载（sync.ts:2148-2168） |
| 事件分发 | 逐个 socket.on(...) | socket.onAny + 一张 messageHandlers Map（apiSocket.ts:308-316） |
一个值得注意的取舍：服务端把 connectionStateRecovery 注释掉了（socket.ts:44-46）。
注释说明这是为了和多进程改造前的行为保持一致——目前所有客户端在每次重连时都走完整的 REST 重取路径。
App 侧因此在 connect 时判断 !this.socket?.recovered 才触发 onReconnected（apiSocket.ts:280-282），
为将来打开这个开关留好了接口。
8. 巧妙之处（可以抄走的）
-
把「喊脏」和「执行」分开。
InvalidateSync让所有调用点只需表达「数据可能变了」， 去重、串行、重试、backoff 全部下沉（packages/happy-cli/src/utils/sync.ts:14-27）。 这套抽象在 CLI 和 App 里都被完整复制，是两端代码结构相似的根本原因。 -
实时推送只做加速，不做正确性。
seq === lastSeq + 1是唯一的信任条件， 任何不确定都退回 REST 全量补拉（apiSession.ts:313-318）。 于是「socket 丢消息」这个最难测的故障类别，被降级成「多一次 HTTP」。 -
客户端生成幂等键。
localId由客户端randomUUID()（apiSession.ts:679）， 服务端在事务里按它去重（v3SessionRoutes.ts:160-181）。整批重发天然安全。 -
用现成的房间当分布式注册表。 RPC 不引入任何新的存储：注册 = join，注销 = leave， 掉线清理由 Socket.IO 免费提供（
rpcHandler.ts:1-14、258-259）。 -
给
emitWithAck配一个存活轮询做赛跑。 承认底层原语探测不了对端死亡， 用一个便宜的旁路把 30 秒的等待压到 2-4 秒，并要求连续两次未命中才判死（rpcHandler.ts:219-241）。 -
重试即 rebase。
version-mismatch时先吸收服务器的新值，再抛错让backoff重跑 handler， 于是「重试」自动变成「在新状态上重放我的意图」而不是「覆盖」（apiSession.ts:931-940）。 -
发送从队尾切批。 用户最关心最新的那几条，积压的旧消息慢慢回填（
apiSession.ts:648-653）。
9. 边界与局限
- 元数据不保密。 服务器知道 userId、sessionId、machineId、消息条数、时间、RPC 方法名的后缀
（
baseMethodName会把它打进指标，rpcHandler.ts:75-78）。加密保护的只有内容。 seq是 Postgres int4。 App 里用2_147_483_647当反向分页哨兵值就是基于这个假设 （sync.ts:76-81），超出即溢出。- RPC 同一房间多 socket 时只取第一个，并且只打一条 warn（
rpcHandler.ts:193-198）。 同一会话被两个 CLI 进程注册时，调用落到谁身上是不确定的。 result: 'error'被静默吞掉。 状态更新的硬错误在 CLI 侧只有注释、没有日志也没有上抛 （apiSession.ts:937-939、964-967）。connectionStateRecovery关着，所以每次重连都是完整重取（socket.ts:36-46）。- CLI 自己 POST 的消息会被广播回自己，产生一次无人消费的回声（见 §3.6）。
- 附件走的是另一条路（request-upload / request-download + 独立的 blob 密钥），
不在有序日志里，只有一个
ref进消息（apiSession.ts:393-517）。
10. 代码地图（导航索引）
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| CLI 会话客户端总入口 | packages/happy-cli/src/api/apiSession.ts | ApiSessionClient |
| socket 连接参数（clientType / path） | packages/happy-cli/src/api/apiSession.ts | 构造函数里的 io(configuration.serverUrl, {...}) |
| 新消息快路径与回退 | packages/happy-cli/src/api/apiSession.ts | socket.on('update', ...) 的 new-message 分支 |
| 拉取循环 / 推送循环 | packages/happy-cli/src/api/apiSession.ts | fetchMessages、flushOutbox、MAX_OUTBOX_BATCH_SIZE |
| 出站队列与幂等键 | packages/happy-cli/src/api/apiSession.ts | enqueueMessage、pendingOutbox |
| 乐观并发写 | packages/happy-cli/src/api/apiSession.ts | updateMetadata、updateAgentState、metadataLock |
| 归档退出信号 | packages/happy-cli/src/api/apiSession.ts | emit('archived')、suppressNextArchiveSignal |
| 智能重连 | packages/happy-cli/src/api/apiSession.ts | startSmartReconnect、shouldReconnect |
| 同步原语 | packages/happy-cli/src/utils/sync.ts | InvalidateSync、_doSync |
| 串行锁 | packages/happy-cli/src/utils/lock.ts | AsyncLock.inLock |
| RPC 作用域前缀与加解密 | packages/happy-cli/src/api/rpc/RpcHandlerManager.ts | getPrefixedMethod、handleRequest、onSocketConnect |
| CLI 注册的通用 RPC 方法 | packages/happy-cli/src/modules/common/registerCommonHandlers.ts | registerCommonHandlers |
| 服务端 socket 装配与鉴权 | packages/happy-server/sources/app/api/socket.ts | startSocket、io.use(...) middleware |
| 房间路由表 | packages/happy-server/sources/app/events/eventRouter.ts | addConnection、getRoomsForFilter、RecipientFilter |
| 更新包构造（只搬密文） | packages/happy-server/sources/app/events/eventRouter.ts | buildNewMessageUpdate、buildUpdateSessionUpdate |
| 消息读写端点 | packages/happy-server/sources/app/api/routes/v3SessionRoutes.ts | v3SessionRoutes、getMessagesQuerySchema |
| seq 分配 | packages/happy-server/sources/storage/seq.ts | allocateSessionSeqBatch、allocateUserSeq |
| 版本比较落库 | packages/happy-server/sources/app/api/socket/sessionUpdateHandler.ts | sessionUpdateHandler |
| RPC 中继 | packages/happy-server/sources/app/api/socket/rpcHandler.ts | rpcHandler、rpcRoom、waitForRoomMember、fetchRoomSockets |
| App socket 与 RPC 调用 | packages/happy-app/sources/sync/apiSocket.ts | ApiSocket、sessionRPC、machineRPC |
| App 同步引擎 | packages/happy-app/sources/sync/sync.ts | Sync、handleUpdate、fetchInitialLatestPage、flushOutbox |
接下去读： 03 本地与远程双模态 讲这条同步流怎么支撑「把终端里的会话抢过来」； 04 远程回合 讲消息内容进入 SDK 之后发生了什么； 05 守护进程与机器 讲 machine-scoped 连接和它的 RPC 怎么凭空开一个新会话。 回到 Happy — 架构与原理。