From 24adb568da0f10ece15efa7f331b3b42029a675d Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 12:55:12 +0000 Subject: [PATCH] =?UTF-8?q?fix(host):=20=5Fread=5Floop=20=E8=B7=B3?= =?UTF-8?q?=E8=BF=87=E9=9D=9E=20dict=20params=20=E7=AD=89=E8=84=8F?= =?UTF-8?q?=E6=B6=88=E6=81=AF=EF=BC=8C=E4=B8=8D=E6=9D=80=E6=AD=BB=E8=AF=BB?= =?UTF-8?q?=E5=BE=AA=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 顶层为 dict 但 params 为真值非 dict 时，旧代码在 params.get 处抛 AttributeError，被外层 except 吞掉、读线程退出，后续正常消息 再也送达不了。改后校验跳过并记 stderr 日志、脏消息不进 events、 读循环继续；None/缺失/假值沿用旧 or {} 语义。审计同函数其它 get 路径：一并守住不可哈希的响应 id（pending.get）与 threadId / thread.id（turn_queues.get）的 TypeError 致死路径；params.thread 已有 isinstance 守卫、_append_trace 已有 try 包裹，均不动。 --- host.py | 43 ++++++++- tests/test_host_read_loop_params.py | 144 ++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 tests/test_host_read_loop_params.py diff --git a/host.py b/host.py index da2bc2f..1ea91d5 100644 --- a/host.py +++ b/host.py @@ -504,16 +504,57 @@ class Server: self._answer_server_request(msg) continue if "id" in msg: + mid = msg["id"] + if isinstance(mid, (list, dict)): + # id 在此被当 pending 字典的 key：不可哈希时 + # 旧代码在 pending.get 处抛 TypeError，被外层 + # except 吞掉、读线程退出。这类 id 本就不可能 + # 命中 pending（key 全是 request 发出的 int）， + # 跳过本条、读循环继续。 + sys.stderr.write( + "host _read_loop: skip response with " + "unhashable id type=%s value=%.200s\n" + % (type(mid).__name__, repr(mid))) + sys.stderr.flush() + continue with self.pending_lock: - q = self.pending.get(msg["id"]) + q = self.pending.get(mid) if q is not None: q.put(msg) continue # 通知 params = msg.get("params") or {} + if not isinstance(params, dict): + # 残留项：params 为真值非 dict（如 "foo"/123/ + # [1,2]/True）时旧代码在下一行 params.get 抛 + # AttributeError，被外层 except 吞掉、读线程退出， + # 后续正常消息再也送达不了。脏消息跳过、不进 + # events、不当正常请求处理，读循环继续。 + # None/缺失/假值沿用旧的 `or {}` 语义（已是 + # dict，不走此分支），正常消息语义不动。 + sys.stderr.write( + "host _read_loop: skip notification with " + "non-dict params type=%s value=%.200s\n" + % (type(params).__name__, repr(params))) + sys.stderr.flush() + continue thread_id = params.get("threadId") if not thread_id and isinstance(params.get("thread"), dict): thread_id = params["thread"].get("id") + if not isinstance(thread_id, str) and thread_id is not None: + # thread_id 只用于 turn_queues 路由：不可哈希的 + # 值（如 list/dict）旧代码在 turn_queues.get 处 + # 抛 TypeError 杀死读线程；非 str 的 key 本就不 + # 可能命中 turn_queues（key 全是 str），一律视为 + # 不可路由，事件本身仍保留、只跳过路由。 + if isinstance(thread_id, (list, dict)): + sys.stderr.write( + "host _read_loop: skip turn routing with " + "unhashable thread_id type=%s value=%.200s\n" + % (type(thread_id).__name__, + repr(thread_id))) + sys.stderr.flush() + thread_id = None with self.events_lock: self.events.append(msg) if self.trace_path is not None: diff --git a/tests/test_host_read_loop_params.py b/tests/test_host_read_loop_params.py new file mode 100644 index 0000000..1d1cb6b --- /dev/null +++ b/tests/test_host_read_loop_params.py @@ -0,0 +1,144 @@ +"""_read_loop 脏消息回归测试（残留项 1：params 非 dict 等）。 + +不启动 app-server：Server 实例用 object.__new__ 绕开 __init__ +（__init__ 会拉起 codex 子进程），直接同步跑 _read_loop 消费 +伪造的 stdout 行。改前行为：真值非 dict 的 params 在 params.get +处抛 AttributeError，被外层 except 吞掉、读循环退出，脏消息后 +的正常消息再也送达不了；不可哈希的 id/threadId 同理在 +pending.get / turn_queues.get 处抛 TypeError 杀死读循环。 + +范围说明：顶层非 dict 消息（"method" in msg 等路径）是上一轨 +#266 的范围，不在本文件覆盖；本文件只覆盖顶层为 dict 的情形。 +""" + +import contextlib +import io +import json +import queue +import sys +import threading +import unittest +from pathlib import Path + +try: + import host + from host import Server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + import host + from host import Server + + +class FakeProc: + def __init__(self, lines): + self.stdout = lines + self.stdin = io.BytesIO() + + +def _normal_notification(): + return {"method": "item/completed", + "params": {"threadId": "th-1", + "item": {"type": "agentMessage", "id": "m1", + "text": "done"}}} + + +class ReadLoopParamsTestCase(unittest.TestCase): + def _run(self, lines): + srv = object.__new__(Server) + srv.proc = FakeProc([json.dumps(m) for m in lines]) + srv.pending = {} + srv.pending_lock = threading.Lock() + srv.turn_queues = {} + srv.events = [] + srv.events_lock = threading.Lock() + srv.trace_path = None + srv._trace_lines = 0 + err = io.StringIO() + with contextlib.redirect_stderr(err): + srv._read_loop() + return srv, err.getvalue() + + def test_non_dict_params_skipped_and_loop_continues(self): + # 真值非 dict 的 params 各型：脏消息不进 events，读循环 + # 继续、脏消息后的正常通知仍送达。 + for dirty_params in ("foo", 123, [1, 2], True, 1.5): + with self.subTest(params=dirty_params): + dirty = {"method": "item/completed", + "params": dirty_params} + srv, err = self._run([dirty, _normal_notification()]) + self.assertEqual(srv.events, [_normal_notification()]) + self.assertIn("non-dict params", err) + + def test_null_missing_empty_params_preserved(self): + # None/缺失/空 dict 沿用旧的 `or {}` 语义：不属致死脏消息， + # 事件仍照常追加（改前改后行为一致，不得以“跳过”丢通知）。 + for dirty in ({"method": "item/completed", "params": None}, + {"method": "item/completed"}, + {"method": "item/completed", "params": {}}): + with self.subTest(dirty=dirty): + srv, _err = self._run([dirty, _normal_notification()]) + self.assertEqual(srv.events, [dirty, _normal_notification()]) + + def test_unhashable_thread_id_keeps_event_and_loop_continues(self): + # threadId / thread.id 不可哈希时：事件本身仍保留（params + # 是合法 dict），只跳过 turn 路由，读循环继续。 + cases = [ + {"method": "turn/completed", + "params": {"threadId": ["not", "a", "str"]}}, + {"method": "turn/completed", + "params": {"thread": {"id": {"x": 1}}}}, + ] + for dirty in cases: + with self.subTest(dirty=dirty): + srv, err = self._run([dirty, _normal_notification()]) + self.assertEqual( + srv.events, [dirty, _normal_notification()]) + self.assertIn("unhashable thread_id", err) + + def test_valid_turn_completed_still_routed(self): + # 正常语义不动：str threadId 的 turn/completed 仍进 turn 队列。 + srv = object.__new__(Server) + turn_q = queue.Queue() + done = {"method": "turn/completed", + "params": {"threadId": "th-1"}} + srv.proc = FakeProc([json.dumps(done)]) + srv.pending = {} + srv.pending_lock = threading.Lock() + srv.turn_queues = {"th-1": turn_q} + srv.events = [] + srv.events_lock = threading.Lock() + srv.trace_path = None + srv._trace_lines = 0 + srv._read_loop() + self.assertEqual(srv.events, [done]) + self.assertEqual(turn_q.get_nowait(), done) + + def test_unhashable_response_id_skipped_and_pending_delivered(self): + # 响应 id 不可哈希时：该条跳过，之后 id 正常的响应仍送达 + # pending 队列（pending key 全是 int，list/dict id 本就 + # 不可能命中）。 + for bad_id in (["unhashable"], {"x": 1}): + with self.subTest(bad_id=bad_id): + srv = object.__new__(Server) + pend_q = queue.Queue() + resp = {"id": 7, "result": {"ok": True}} + srv.proc = FakeProc([json.dumps({"id": bad_id, + "result": {}}), + json.dumps(resp)]) + srv.pending = {7: pend_q} + srv.pending_lock = threading.Lock() + srv.turn_queues = {} + srv.events = [] + srv.events_lock = threading.Lock() + srv.trace_path = None + srv._trace_lines = 0 + err = io.StringIO() + with contextlib.redirect_stderr(err): + srv._read_loop() + self.assertEqual(pend_q.get_nowait(), resp) + self.assertEqual(srv.events, []) + self.assertIn("unhashable id", err.getvalue()) + + +if __name__ == "__main__": + unittest.main()