Mini05：沿着一次请求理解 IPC 与网络通信

在 Java 里用 RPC 框架时，连接、收包和请求配对往往藏在框架后面。读 MiniClaude 的 asyncio + TCP + JSON 实现，这些工作突然都露了出来：谁调用连接回调？发完请求之后谁收响应？Agent 已经开始跑了，CLI 退出会怎样？

这篇沿着一次 agent.run，从客户端发出请求，走到服务端返回 run_id、持续推送事件，再看断线和关闭。原来分散的 23 个问题，都能放回这条路径。

1 请求先从哪个进程出发

S1 的 mini run 把 AgentRunner、AgentLoop、provider、tool 和终端打印放在同一个 Python 进程里，任务跑完，进程退出。

S2 把 Agent 执行移到常驻的 mini-core daemon，mini run 成了客户端：发命令、接事件、显示结果。mini tui 也是客户端。所谓“双进程架构”是服务端与客户端两类角色；同时开 daemon、CLI 和 TUI，实际就是三个进程。

这样拆分，可以复用 daemon 中的 Agent 核心和进程内 EventBus，不必每次运行都重建服务，还能让多个界面看同一次执行。常驻也不等于永远不退出：异常、信号和主动关闭仍会结束 daemon。

CLI 和 TUI 复用 SocketClient 的发送、接收和请求配对逻辑，事件回调分别接终端打印器和 Textual/RichLog 界面更新函数。复用通信层，不等于行为完全相同：S2 笔记里的 CLI 发 agent.run，TUI 只发 event.subscribe 观察；后续 S3+ 笔记记录了 TUI 经 session.send_message 发起运行的能力。不能把“纯观察者”写成 TUI 永久的限制。S2 的分工有命令与观察分离的意思，但仅凭两个客户端，还不能说实现了完整的 CQRS。

另一个容易混淆的层次是 BashTool。CLI/TUI 不是 daemon 启动和管理的子进程；daemon 通过 asyncio.create_subprocess_shell(...) 启动的 shell 才是它的子进程，shell 又可能启动命令进程。工具进程有独立 PID，通常继承调用者的环境和权限上下文，也可以显式改变环境。几个终端里的进程是应用角色上的平级，不保证有同一个操作系统父进程。Python 子进程文档、环境继承说明

2 连接进来后 字节怎样变成请求

回调由 asyncio 调用

传给 start_server 的是函数对象，不是在这一行调用它。每当一个连接建立，asyncio 把该连接的 reader, writer 交给回调；如果回调是协程函数，会自动安排为 Task。一个连接的回调可以循环处理很多条消息，并非每来一条消息就新建连接。asyncio Streams

# handle_connection 已定义为 async def，监听端口由操作系统分配。
server = await asyncio.start_server(handle_connection, "127.0.0.1", 0)
# 正确：传 handle_connection；不是 handle_connection()。

读连接代码时常看到 except SomeError: pass。语法要求代码块内有语句，pass 是“什么都不做”的一种写法；可以换成日志、raise、return 等合适语句，不是必须保留 pass 这个单词。注释不能替代语句。若写成 except Exception: pass，还要问清楚为什么可以吞掉异常，而不是把它当成网络代码的固定模板。Python pass 语句

TCP 管字节 分帧约定管消息边界

本机两个进程用 loopback TCP 通信，已经是 IPC，不必先引入 Kafka、Redis 或 multiprocessing.Pipe。这条路径有三层：

TCP 不保留应用每次 write() 的边界：两次写可能被一起读到，一次写也可能分几次到达。按行读取是在字节流之上恢复消息边界。JSON-RPC 本身不规定使用 TCP，更不规定用换行分帧。TCP 规范 §2.2、JSON-RPC 规范

“一行一个 JSON”常称 JSON Lines / NDJSON。这里不能用带缩进、多行输出的 JSON；字符串内容里的换行要由 JSON 序列化器编码成 \n，它不等于报文末尾的实际换行字节。NDJSON 分帧约定

import json

class IpcError(Exception):
    pass

def encode_frame(message):
    text = json.dumps(message, ensure_ascii=False, allow_nan=False,
                      separators=(",", ":"))
    return text.encode("utf-8") + b"\n"

async def read_frame(reader):
    line = await reader.readline()
    if not line:
        raise EOFError("连接结束")
    if not line.endswith(b"\n"):
        raise IpcError("EOF 前收到未结束的帧")
    return json.loads(line.decode("utf-8"))

这是本文选择的严格行分帧策略：EOF 前有字节但没有行尾，也不当成完整帧。readline() 本身会返回这段残留数据，并不会替应用判断协议是否完整。实际连接还需设置并处理帧大小限制、无效 UTF-8 和 JSON；这些读取或解析失败都应进入连接错误处理，不能让等待者一直挂着。读取与 EOF 行为

3 发出请求后 谁来收响应

先看线上几种不同的消息：

{"jsonrpc":"2.0","method":"agent.run","params":{"goal":"读文件"},"id":"q1"}
{"jsonrpc":"2.0","id":"q1","result":{"run_id":"r1"}}
{"kind":"event","event":{"type":"run.started","run_id":"r1"}}

q1 配对这一次 RPC，r1 标识长时间运行的 Agent Run，二者不能混用。返回 run_id 表示服务端接受并创建了运行，不表示 Agent 已成功完成；后续进度和终态走事件。

JSON-RPC 的 notification 是没有 id 成员的请求对象，例如 {"jsonrpc":"2.0","method":"log","params":{}}，接收方不得返回响应。id: null 不等于没有 id，规范不推荐客户端用它作为请求 ID。上面的 kind: "event" 则是项目自定义封套，不是标准 JSON-RPC notification。JSON-RPC 请求、通知与响应

一个读循环负责所有入站消息

在原笔记的设计中，send_command() 写请求后等待自己的 Future；唯一的 run_event_loop() 持续读同一个 socket，交给 _dispatch() 分流：

• 响应按请求 id 找到 _pending[id]，填入结果或异常。

• 事件交给 on_event 回调，不进入 _pending。

原笔记把 JSON 解析也放在 _dispatch(line)；本文先由 read_frame() 解析，再传给 _dispatch(msg)。分工没有变：读循环拿数据，dispatcher 不再读 socket。

_pending 是待响应表，不是事件队列。Future 代表尚未得到的结果；等待它只挂起当前任务，其他就绪任务仍可继续。可以类比 Java 的 CompletableFuture，但这里的 asyncio Future 不是线程安全对象。Python Future

下面两段是同一个教学客户端的发送与分流方法。前提是连接已建立、读循环已启动，_pending 是字典，_write_lock 是 asyncio.Lock，_on_event 是异步回调。

async def send_command(self, method, params, timeout=3):
    req_id = uuid.uuid4().hex
    fut = asyncio.get_running_loop().create_future()
    self._pending[req_id] = fut                 # 先登记，再发送
    try:
        async with asyncio.timeout(timeout):  # 包括写入等待与响应等待
            async with self._write_lock:
                self._writer.write(encode_frame({
                    "jsonrpc": "2.0", "id": req_id,
                    "method": method, "params": params,
                }))
                await self._writer.drain()
            return await fut
    finally:
        self._pending.pop(req_id, None)        # 超时/取消也清理
        if not fut.done():
            fut.cancel()
        elif not fut.cancelled():
            fut.exception()  # 写入阶段提前退出时，也取走已到达的异常

async def _dispatch(self, msg):
    # 教学子集：只接收单条 RPC 响应和自定义事件封套。
    if not isinstance(msg, dict):
        raise IpcError("需要对象报文")
    if "jsonrpc" in msg:
        if (msg["jsonrpc"] != "2.0" or "id" not in msg
                or "method" in msg
                or (("result" in msg) == ("error" in msg))):
            raise IpcError("响应结构错误")
        fut = self._pending.pop(msg["id"], None)
        if fut is None or fut.done():
            return                            # 迟到/未知响应，实际可记日志
        if "error" in msg:
            fut.set_exception(IpcError(str(msg["error"])))
        else:
            fut.set_result(msg["result"])
    elif msg.get("kind") == "event":
        await self._on_event(msg["event"])
    else:
        raise IpcError("未知报文类型")

这里保留两个很小但容易写错的细节：按成员存在与否判断 id，不要用 if id 漏掉合法的 0 或空字符串；成功结果直接取 msg["result"]，不能写 msg.get("result") or {}，否则 0、False、[]、null 都会被改掉。失败响应必须走异常路径；标准错误对象包含整数 code、字符串 message 和可选 data。上面为突出配对机制，省略了 ID 类型、错误对象等完整校验，也不实现 batch 或服务端反向请求，不能当作完整 JSON-RPC 库。JSON-RPC 响应与错误对象

kind 保留了事件类别的扩展位置，以后可以另行约定 system 等类别，但不能据此声称这些类别已经实现。新增类别仍需要约定语义、接收端处理与旧客户端的兼容策略。

为什么要先启动读循环

read_task = asyncio.create_task(client.run_event_loop())
try:
    await client.send_command("event.subscribe", {
        "topics": ["run.*", "tool.*"], "scope": "global",
    })
    result = await client.send_command("agent.run", {"goal": "读文件"})
finally:
    await client.close()  # 教学客户端约定：关闭连接并等待读任务结束

这段只展示启动次序；真实 CLI 还要等对应 Run 的终态再关闭。先订阅、再发起运行，也是在避免运行太快而漏掉最初的事件；订阅确认后的事件边界、按 run_id 区分并行运行，仍需由协议和客户端共同约定。

如果先 await send_command()，却把启动 reader 写在它后面，就会出现“发的人等响应，唯一能收的人还没开始”的互相等待。原因是当前代码把收包职责交给了独立读循环，不是 await 阻塞了整个事件循环，也不是 TCP 规定必须恰好两个协程。多个请求可以各自等待，仍共用一个读循环；不要让它们分别去抢同一 StreamReader。asyncio 任务与等待

两条 TCP 可以把 RPC 和事件分开，但不会自动变成“只需一个协程”：两路仍需并发读取或统一的就绪事件调度，RPC 并发时仍需配对响应。两条连接可以连到同一个服务端监听端口，代价主要是两套连接生命周期及其关联关系，不是必然多占一个监听端口。一条连接减少管理对象，却仍需分流，也可能让大事件和响应在同一字节流里相互等待。这是协议取舍，不是协程数量公式。

4 daemon 收到请求 为什么很快就能回复

服务端解析 method 后，从 handler 表找到处理函数；处理函数创建后台 Run 任务，便可以返回 run_id。create_task() 安排协程执行，不会等它跑完。命令的“接受成功”和 Agent 的“执行成功”因此是两件事。

原笔记的两个阶段要放在一起看：

# 历史 S2：同一 daemon 只允许一个尚未完成的 Run。
if self._current_run_task is not None and not self._current_run_task.done():
    raise RuntimeError("a run is already in progress")
self._current_run_task = asyncio.create_task(runner.run())

# 历史 S3：改为保存多个 Run 的 Task，以下是另一阶段，不与上段连用。
run_task = asyncio.create_task(
    self._sessions.send_message(session.id, cmd.goal, run_id=run_id)
)
self._running_runs.add(run_task)
run_task.add_done_callback(self._running_runs.discard)

S2 的单变量是应用层准入检查，不是 TCP、asyncio 或“一个 Agent”天然只能跑一项任务。S3 笔记改成 _running_runs: set[asyncio.Task[Any]]，支持多个 Run，runner 的创建放在 SessionManager 内部，handler 调 send_message。集合持有任务引用，完成后 discard 移除；元素已经不在集合里也不会报错。后台任务引用与清理

这个集合本身既不是并发上限，也不是线程锁。它没有处理任务异常、共享资源互斥或同一 Session 的冲突。原笔记强调每个 Run 有自己的 AgentLoop、ExecutionContext、TaskManager，应理解为运行状态的隔离目标，而不是“所有外部资源都互不干扰”的保证。多 Run 也不等于把一个 Run 的步骤全部并行化；不同 Run 的 LLM 请求仍可能在时间上重叠，是否限流要看调用层。

add_done_callback(discard) 只清引用，不会读取异常。daemon 还需要观察任务失败，关闭时对存活任务发出取消并等待清理完成。只遍历 cancel() 后立刻退出，不能证明它们已经停干净。

5 Run 的事件怎样走到另一个进程

EventBus 只走进程内函数调用

这一版笔记里的 EventBus，本质是 handler 列表和调用循环：

async def publish(self, event):
    for handler in list(self._subs):
        await handler(event)

# 组装时注册；不是 IpcEventBroadcaster 构造器偷偷订阅。
bus.subscribe(event_writer.handle)
bus.subscribe(broadcaster.handle)

一个事件交给多个订阅者叫 fan-out，扇出，不是英文缩写；多个来源汇入一个处理点叫 fan-in。上面的 fan-out 是依次 await，不等于并行执行。正常返回只说明这些回调完成了各自承诺的工作：若某个回调只是入队，不能据此说文件已经持久化或远端已显示；某个回调抛错而未被隔离，后续回调也可能没有执行。

EventBus 没有因为叫“总线”就跨进程。真正跨出去的是 IpcEventBroadcaster.handle：接到进程内事件，把模型转换为字典，包装成 EventPushEnvelope，写入订阅者的 socket。原笔记中特别注明，订阅发生在 CoreApp.run() 组装处，方法名是 handle；构造 broadcaster 和把它接上 EventBus 是两步。

# 机制摘录：已有 event、subscriptions；清理坏连接由连接管理层负责。
event_data = event.model_dump(mode="json")
for sub in list(subscriptions):
    if not matches(event_data, sub.topics, sub.scope):
        continue
    sub.writer.write(encode_frame({"kind": "event", "event": event_data}))
    await sub.writer.drain()

这就是进程内与进程间的桥接点。EventBus 可复用，CLI/TUI 只处理序列化后的数据，不直接持有 daemon 的 Python 对象。

Kafka、Redis Pub/Sub 也能跨进程，但不能因此混成同一种可靠性：Kafka 可按保留策略保存日志，消费者可以独立读取；同一 consumer group 内通常分摊分区，不是把每条记录广播给组内所有成员。Redis Pub/Sub 不提供断线历史补发，别把它和 Redis Streams 混淆。Kafka 消费模型、Redis Pub/Sub 交付语义

topic 和 scope 两层都匹配才推

from fnmatch import fnmatchcase

def matches(event, topics, scope):
    topic_ok = any(fnmatchcase(event["type"], p) for p in topics)
    scope_ok = (scope == "global" or
                (scope.startswith("run:") and
                 event.get("run_id") == scope[4:]))
    return topic_ok and scope_ok

run.* 匹配 run.started、run.finished，决定“看哪类事件”；global 或 run:r1 决定“看哪些 Run”。global 只是 scope 放行，仍然需要 topic 匹配。*、?、[seq] 是 shell 风格通配符，不是正则；原笔记用 fnmatch.fnmatch，此处用 fnmatchcase 明确保证不同操作系统上都区分大小写。Python fnmatch

这里也会遇到背压。writer.write() 可能先把数据放进本地缓冲，await drain() 是流量控制，不是“对端业务已处理”的确认。逐个等待慢客户端，可能拖住 broadcaster，进而拖住串行 EventBus；给每个客户端加队列能隔离一部分等待，却仍要决定队列满时等待、丢弃还是断开慢客户端。无界队列只是把拥堵换成内存增长。StreamWriter 写入与 drain

同样，客户端读循环若直接 await 慢 UI 回调，后面的 RPC 响应也会延迟。更危险的是在这个回调里调用并等待同一客户端的 send_command()：reader 正等回调结束，回调又等 reader 收响应。回调应短小，耗时工作交给受管理的任务或有界队列，而不是再制造一个失去管理的后台任务。

6 响应没回来 或者连接断了

请求表不仅要处理成功，还必须有退出路径。超时和调用者取消时，发送方法的 finally 移除待响应项；若写入阶段退出前 Future 已被置为失败，还要取出异常，避免留下无人观察的 Future。EOF、解码失败或连接异常时，读循环应让所有仍在等待的 Future 失败，并清空表。之后迟到的响应不能重新完成已结束的 Future。

# 读循环退出时的清理片段；每个仍在等待的请求都得到失败通知。
for fut in self._pending.values():
    if not fut.done():
        fut.set_exception(IpcError("连接已关闭"))
self._pending.clear()

CLI 还要同时关心“Run 到达终态”和“reader 提前结束”。如果只 await finished.wait()，reader 断了而终态事件再也收不到，CLI 仍可能永远等下去。一个简短的等待骨架是：

async def wait_cli_end(read_task, finished):
    terminal = asyncio.create_task(finished.wait())
    try:
        done, _ = await asyncio.wait(
            {read_task, terminal}, return_when=asyncio.FIRST_COMPLETED
        )
        if terminal in done:
            return 0   # 示例约定：finished 只代表已确认的成功终态
        await read_task  # 取出 reader 的异常；正常 EOF 也按通信失败处理
        return 1
    except (IpcError, OSError):
        return 1
    finally:
        terminal.cancel()
        await asyncio.gather(terminal, return_exceptions=True)

若协议区分成功、失败、取消，实际 CLI 应按终态状态选择退出码。外层负责关闭连接并等待 reader；同步入口再把异步返回值交给 sys.exit(asyncio.run(main()))。0 是本程序约定的成功，1 是失败；详细原因仍靠错误信息，不能只看数字。

客户端退出不会自动取消 daemon 自己持有的 Run。原设计把执行与观察分开，CLI 断线后，daemon 活着且没有“断线取消”策略，Run 就可继续，其他 TUI 也可继续看。反过来，主动取消 Run 应走明确的命令，而不是把关 socket 当取消操作。

原笔记里的 CLI 是一次性任务，断线后报错退出；TUI 外层有重连循环，连接失败或读循环退出后等待再试，并在新连接上重新订阅。示例里的固定 sleep(2) 是简单策略，不是所有系统的推荐参数；持续重连通常还要考虑退避、抖动和用户退出。

重连只恢复通道，不自动补齐历史。 原笔记中的 --replay RUN_ID 是历史回放入口。能否无缝接上实时流，还取决于日志是否保留、从哪里续读、如何处理重复和缺口；不能仅凭 while True 就承诺不漏事件。重连后也应建立新的待响应表，不把旧连接上的未知结果偷偷当成成功。

7 事件写入队列之后 是否就不阻塞了

TraceWriter 的“队列 + drain task”是生产者消费者模式：emit() 先把记录入队，后台任务再取出写文件，减少产生日志的调用路径直接等待文件 I/O 的时间。这里的 drain task 是“消费队列的后台任务”，不要和 socket 的 writer.drain() 混为一谈。

后台协程仍在同一个事件循环线程上。 如果 _write_file() 内部执行同步文件写，即使函数叫 async def，仍可能阻塞整个循环。对于阻塞 I/O，可以使用真正的异步接口，或把同步调用交给线程，例如下面的教学片段。asyncio.to_thread

async def drain_records(queue, append_record):
    while True:
        record = await queue.get()
        try:
            await asyncio.to_thread(append_record, record)
        finally:
            queue.task_done()

这里 append_record 是同步写函数。片段把异常向外传播，owner 必须观察 drain task 的失败；不能把 task_done() 理解成写入成功凭证。如果消费者失败后队列还有记录，单独等 queue.join() 可能一直等不到，应同时监督消费者。

Queue() 默认无界，put_nowait() 不等容量，不代表系统能无限吞吐；有界队列满了会抛 QueueFull，await put() 则等待空位。选哪种取决于这类记录能否丢失。join() 等的是未完成计数归零，不是“队列看起来为空”，更不是文件已经可靠落盘。asyncio Queue

因此队列只改变等待发生的位置。它能解耦生产和消费，却不替你解决容量、写失败、线程中的写入何时真正结束，以及关闭时剩余记录怎么办。

8 最后由谁把连接 任务和进程关掉

关闭流程要顺着资源所有者走。对这个教学架构，可以按以下顺序检查：

1. daemon 停止接收新工作，决定让已有 Run 完成还是取消并等待；停止相关事件生产者。
2. 不再往日志队列放新记录，等待消费者完成已接受的记录，同时处理写失败和关闭超时。
3. 关闭文件、客户端连接和监听服务；socket 调 close() 后还应 await wait_closed()。
4. 回收由本进程持有的后台任务和工具子进程，记录未能完成的清理。

Unix 上，SIGTERM 可以被处理，程序可通过 loop.add_signal_handler 发起这套流程；只有已经接好处理器且清理成功完成，才叫优雅关闭。默认终止、磁盘写失败、关闭期限耗尽或随后遭到强杀，都可能丢数据。SIGKILL 不能捕获、忽略或处理，进程没有机会把内存队列清空。不能再把它们简化成“SIGTERM 不丢，SIGKILL 才丢”。Python 信号、asyncio Unix 信号处理

即使 queue.join() 返回，持久性仍取决于消费者承诺：Python 文件缓冲是否 flush、是否需要 fsync，以及存储层本身的保证。这些层次不能用一个“已写完”代替。Python fsync

最后回到进程关系：daemon 能管理自己创建的工具进程，超时时需要终止并等待回收；通过 shell 启动时，还要区分 shell PID 与它的后代，杀掉 shell 不保证整棵进程树都结束。CLI/TUI 则通常各自退出。不能因为 TUI 不是子进程，就断言 daemon 在操作系统层面“绝对不能 kill 它”；能否发信号取决于权限，是否应该这么做取决于应用的所有权约定。asyncio 子进程终止与 PID

Task 也有类似边界：普通 create_task() 不自动建立“创建者取消，所有孩子跟着取消”的结构；集合持有引用不等于完整的生命周期管理。需要作用域内共同退出时，可以考虑 TaskGroup；已经放到 daemon 长期管理的 Run，则由 daemon 明确承担等待、异常处理与关闭责任。

沿着一条请求复习

重读源码时，可以只追这几个问题：

• 谁拥有连接，谁是唯一 reader，报文在哪里分帧？

• 请求 ID 存在哪里，成功、错误、超时、EOF 分别怎样结束等待？

• 返回 run_id 后，Run 由谁持有，事件在哪里跨过进程边界？

• 最慢的客户端或日志消费者会把哪一层拖住？

• 观察者断线与执行取消有什么区别，重连后缺的事件从哪里来？

• 关闭时，谁停止生产、谁排空队列、谁等待任务和子进程真正退出？