fix(host): _read_loop 跳过非 dict params 等脏消息，不杀死读循环

顶层为 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

@@ -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:

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()