U-024: c_ws_subscribe + execution_event 实时推送（最小连通版）

Python stdlib 没有 WebSocket 支持，src/ws_subscribe.py 手写 RFC 6455 握手（hashlib+base64 算 Sec-WebSocket-Accept）+ 帧编解码（服务端帧不加掩码，客户端帧强制解掩码），不引入 websockets 之类第三方库（05 §5.7）。推送用轮询而非写路径 hook：execution_event 的写入分散在十几个调用点（org_node_trigger/work_graph_execute/work_node_run/task_create/ bug_record/...），没有单一写入路径可以插桩，backend_server.py 的 _handle_ws_subscribe 劫持 ThreadingHTTPServer 连接后每 0.5 秒按 rowid 增量查一次 execution_event（project_id/可选 task_id 过滤），轻松满足 REQ-12 的 ≤2 秒目标。断线重连按 03-C §3.14 走 REST 快照，不在 WS 层做消息重放，新连接只看连接那一刻起的新事件。 daemon_threads = True 避免长连接 WS 线程阻塞进程退出。验证：test_backend_server.py 用原始 socket 走真实握手+推送延迟断言；另外用 Node 原生 WebSocket 客户端做跨实现互操作性验证，连真实运行的服务器，从建连到收到 c_task_create 触发的真实推送全程走通。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-024: c_ws_subscribe + execution_event 实时推送（最小连通版）
Python stdlib 没有 WebSocket 支持，src/ws_subscribe.py 手写 RFC 6455 握手（hashlib+base64 算 Sec-WebSocket-Accept）+ 帧编解码（服务端帧不加掩码，客户端帧强制解掩码），不引入 websockets 之类第三方库（05 §5.7）。推送用轮询而非写路径 hook：execution_event 的写入分散在十几个调用点（org_node_trigger/work_graph_execute/work_node_run/task_create/ bug_record/...），没有单一写入路径可以插桩，backend_server.py 的 _handle_ws_subscribe 劫持 ThreadingHTTPServer 连接后每 0.5 秒按 rowid 增量查一次 execution_event（project_id/可选 task_id 过滤），轻松满足 REQ-12 的 ≤2 秒目标。断线重连按 03-C §3.14 走 REST 快照，不在 WS 层做消息重放，新连接只看连接那一刻起的新事件。 daemon_threads = True 避免长连接 WS 线程阻塞进程退出。验证：test_backend_server.py 用原始 socket 走真实握手+推送延迟断言；另外用 Node 原生 WebSocket 客户端做跨实现互操作性验证，连真实运行的服务器，从建连到收到 c_task_create 触发的真实推送全程走通。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent cacd03bf
Showing 6 changed files with 508 additions and 7 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/backend_server.py
src/test_backend_server.py
src/test_ws_subscribe.py
src/ws_subscribe.py
@@ -231,6 +231,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-024 [Value Path]** `c_ws_subscribe` + `execution_event` 实时推送（最小连通版）
   - 依赖：U-012, U-013（work graph 执行、work node run 是最早、最稳定产出 `execution_event` 的两个来源，足够验证推送链路本身是否通）
   - Done：`c_ws_subscribe` 对真实 `execution_event` 做实时推送，≤2 秒延迟达成；REQ-12 happy 用例通过（boundary/failure——尤其是断线重连拉快照在多样事件场景下的表现——留给 U-024.5 补齐完整覆盖）
+  - **实现补充（2026-07-31）**：Python stdlib 没有 WebSocket 支持，`src/ws_subscribe.py` 手写了 RFC 6455 握手（`Sec-WebSocket-Accept` 用 hashlib+base64 算）+ 帧编解码（`socket` 原始读写，服务端帧不加掩码，客户端帧强制解掩码），没有引入 `websockets` 之类的第三方库（05 §5.7）。推送机制是**轮询**，不是写路径 hook——`execution_event` 的写入分散在 `org_node_trigger`/`work_graph_execute`/`work_node_run`/`task_create`/`bug_record` 等十几个调用点，没有单一写入路径可以插桩，Unit 名字里的"最小连通版"本身也在暗示这个方向；`_handle_ws_subscribe` 用 `ThreadingHTTPServer` 劫持连接后每 0.5 秒查一次 `execution_event`（按 `rowid` 增量，`project_id`/可选 `task_id` 过滤），轻松满足 REQ-12 的 ≤2 秒目标（该数字本身写的是"建议"而非硬 SLA）。03-C §3.14 自己写明断线重连走 REST 快照补齐、不在 WebSocket 层做消息重放缓冲，所以新连接只看"连接那一刻起"的新事件，不做历史重放。给 `ThreadingHTTPServer` 加了 `daemon_threads = True`，避免长连接的 WS 线程阻塞进程退出。端到端验证：`src/test_backend_server.py` 用原始 socket 走真实握手+推送延迟断言；另外用 Node 原生 `WebSocket` 客户端（跨实现的真实互操作性验证，不是同一套代码自己测自己）连真实运行的服务器，从建连到收到 `c_task_create` 触发的真实 `task_status_changed` 推送全程走通。
   - 覆盖：01-M `execution_event`；03-C `c_ws_subscribe`；REQ-12（部分）
 - **U-024.5（2026-07-30 新增）[Value Path]** REQ-12 完整验收（事件多样性联调）
@@ -38,7 +38,7 @@
     {"id": "U-022", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_project_setup + c_pick_folder 全平台（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-007", "U-008"], "coverage": {"model": [], "view": ["v_project_setup"], "controller": ["c_pick_folder", "c_project_create"], "process": [], "requirement": ["REQ-11"]}, "done": "REQ-11 里 v_project_setup/c_pick_folder/c_project_create 部分通过；c_pick_folder 补齐 Windows/Linux；c_project_create 首次接上 HTTP 路由；REQ-11 完整 happy（含建 task）随 U-023 累计满足"},
     {"id": "U-023", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_goal_composer + v_bug_feedback（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-022"], "coverage": {"model": [], "view": ["v_goal_composer", "v_bug_feedback"], "controller": ["c_task_create", "c_bug_report_create"], "process": [], "requirement": []}, "done": "能通过 UI 创建 task 和人工 bug_record；补上 c_bug_report_create 的 project_id 必填缺口（03-C input_contract 遗漏）；related_task_id 简化为文本输入，等 U-025 的 c_task_list 落地后换真下拉"},
-    {"id": "U-024", "phase": "PH3", "sub_phase": "PH3-S2", "name": "c_ws_subscribe + execution_event 实时推送（最小连通版）", "classification": "Value Path", "depends_on": ["U-012", "U-013"], "coverage": {"model": ["execution_event"], "view": [], "controller": ["c_ws_subscribe"], "process": [], "requirement": ["REQ-12"]}, "done": "推送链路本身连通，≤2 秒延迟达成；REQ-12 happy 用例通过（boundary/failure 留给 U-024.5）"},
+    {"id": "U-024", "phase": "PH3", "sub_phase": "PH3-S2", "name": "c_ws_subscribe + execution_event 实时推送（最小连通版，见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-012", "U-013"], "coverage": {"model": ["execution_event"], "view": [], "controller": ["c_ws_subscribe"], "process": [], "requirement": ["REQ-12"]}, "done": "推送链路本身连通，≤2 秒延迟达成；REQ-12 happy 用例通过（boundary/failure 留给 U-024.5）；手写 RFC 6455 握手+帧编解码（无第三方 websockets 依赖），轮询式推送（无单一写路径可插桩），Node 原生 WebSocket 客户端验证过真实互操作性"},
     {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（2026-07-30 新增）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准"},
     {"id": "U-025", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_dashboard + v_task_board + v_task_detail", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_dashboard", "v_task_board", "v_task_detail"], "controller": [], "process": [], "requirement": []}, "done": "三个页面可正常展示实时数据"},
     {"id": "U-026", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_graph_visualizer", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_graph_visualizer"], "controller": [], "process": [], "requirement": []}, "done": "Org/Work Graph 双层可视化含 waiting_for_quota 信息态"},
@@ -10,20 +10,24 @@ via a `powershell` FolderBrowserDialog script and Linux via `zenity`/
 the U-002 `project_scaffold.create_project()` function that had no HTTP
 route until now) for `v_project_setup`'s submit flow, `c_task_create`/
 `c_bug_report_create` (03-C §3.1/§3.1.1, U-023) for `v_goal_composer`/
-`v_bug_feedback`'s submit flows, and serving the U-008 frontend shell as
-static files (no build step, 05 §5.7: whitelisted by exact path, not a
-generic directory walker, so there's no path-traversal surface to reason
-about).
+`v_bug_feedback`'s submit flows, `c_ws_subscribe` (03-C §3.14, U-024,
+hand-rolled RFC 6455 WebSocket handshake + framing in src/ws_subscribe.py --
+no `websockets` dependency) for real-time `execution_event` push, and
+serving the U-008 frontend shell as static files (no build step, 05 §5.7:
+whitelisted by exact path, not a generic directory walker, so there's no
+path-traversal surface to reason about).
 """
 import json
 import os
 import platform
+import socket
 import subprocess
 import threading
+import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from src import bug_record, paths, project_scaffold, schema, task_create
+from src import bug_record, paths, project_scaffold, schema, task_create, ws_subscribe
 _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
@@ -103,8 +107,16 @@ def _pick_folder_linux():
     raise DialogUnavailableError("neither zenity nor kdialog is installed")
+_WS_PROJECTS_PREFIX = "/api/ws/projects/"
+
+
 class _Handler(BaseHTTPRequestHandler):
     def do_GET(self):
+        parsed = urllib.parse.urlsplit(self.path)
+        if parsed.path.startswith(_WS_PROJECTS_PREFIX):
+            self._handle_ws_subscribe(parsed.path, parsed.query)
+            return
+
         entry = _STATIC_FILES.get(self.path)
         if entry is None:
             self._send_json(404, {"error": "NOT_FOUND"})
@@ -209,6 +221,65 @@ class _Handler(BaseHTTPRequestHandler):
         else:
             self._send_json(200, result)
+    def _handle_ws_subscribe(self, path, query_string):
+        """c_ws_subscribe (03-C §3.14): WS /api/ws/projects/:project_id.
+        Rejects with a normal HTTP response (no 101) before ever touching
+        the socket-hijack path if the request isn't a real WS upgrade or
+        project_id doesn't exist -- 03-C: "直接拒绝连接", not a connection
+        that opens and then immediately closes."""
+        project_id = path[len(_WS_PROJECTS_PREFIX):]
+        if not project_id or "/" in project_id:
+            self._send_json(404, {"error": "NOT_FOUND"})
+            return
+        task_id = urllib.parse.parse_qs(query_string).get("task_id", [None])[0]
+
+        ws_key = self.headers.get("Sec-WebSocket-Key")
+        if self.headers.get("Upgrade", "").lower() != "websocket" or not ws_key:
+            self._send_json(400, {"error": "INVALID_INPUT", "detail": "expected a WebSocket upgrade request"})
+            return
+
+        with self.server.conn_lock:
+            project_exists = self.server.conn.execute(
+                "SELECT 1 FROM project WHERE project_id = ?", (project_id,)
+            ).fetchone() is not None
+        if not project_exists:
+            self._send_json(404, {"error": "PROJECT_NOT_FOUND"})
+            return
+
+        accept_key = ws_subscribe.compute_accept_key(ws_key)
+        self.connection.sendall(
+            (
+                "HTTP/1.1 101 Switching Protocols\r\n"
+                "Upgrade: websocket\r\n"
+                "Connection: Upgrade\r\n"
+                "Sec-WebSocket-Accept: " + accept_key + "\r\n\r\n"
+            ).encode("ascii")
+        )
+        self.close_connection = True
+
+        with self.server.conn_lock:
+            after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
+        self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
+        try:
+            while True:
+                try:
+                    opcode, _payload = ws_subscribe.decode_client_frame(self.connection)
+                    if opcode == ws_subscribe.OPCODE_CLOSE:
+                        break
+                    if opcode == ws_subscribe.OPCODE_PING:
+                        self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_PONG))
+                except socket.timeout:
+                    pass  # normal: nothing from the client this poll tick
+
+                with self.server.conn_lock:
+                    events, after_rowid = ws_subscribe.poll_new_events(
+                        self.server.conn, project_id, after_rowid, task_id=task_id
+                    )
+                for event in events:
+                    self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
+        except (ConnectionError, OSError):
+            pass  # client disconnected -- nothing more to send
+
     def _send_json(self, status, obj):
         body = json.dumps(obj).encode("utf-8")
         self.send_response(status)
@@ -230,6 +301,10 @@ def _ensure_schema(conn):
 class _Server(ThreadingHTTPServer):
+    # long-lived WS connections (c_ws_subscribe) shouldn't be able to block
+    # process shutdown -- ThreadingMixIn.daemon_threads defaults to False.
+    daemon_threads = True
+
     def server_close(self):
         super().server_close()
         self.conn.close()
@@ -6,18 +6,21 @@ Run: python3 -m unittest src.test_backend_server -v
 """
 import ast
+import base64
 import http.client
 import json
 import os
 import shutil
+import socket
 import subprocess
 import sys
 import tempfile
 import threading
+import time
 import unittest
 from unittest.mock import patch
-from src import backend_server
+from src import backend_server, ws_subscribe
 class ServerStartStopTest(unittest.TestCase):
@@ -389,6 +392,160 @@ class TaskAndBugReportRouteTest(unittest.TestCase):
         self.assertEqual(body["error"], "INVALID_PARAM")
+def _ws_handshake_request(path, key):
+    return (
+        "GET {} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Upgrade: websocket\r\n"
+        "Connection: Upgrade\r\n"
+        "Sec-WebSocket-Key: {}\r\n"
+        "Sec-WebSocket-Version: 13\r\n"
+        "\r\n"
+    ).format(path, key).encode("ascii")
+
+
+def _read_http_response_head(sock):
+    """Reads until the blank line terminating an HTTP response's headers --
+    good enough for both a 101 upgrade and a plain error response, since
+    neither carries a body in this handler."""
+    buf = b""
+    while b"\r\n\r\n" not in buf:
+        chunk = sock.recv(4096)
+        if not chunk:
+            break
+        buf += chunk
+    return buf.decode("ascii", errors="replace")
+
+
+def _read_server_frame(sock):
+    """Server->client frames are never masked (RFC 6455 §5.1) -- simpler
+    decode than ws_subscribe.decode_client_frame, which expects masking."""
+    first_two = _recv_exact(sock, 2)
+    length = first_two[1] & 0x7F
+    if length == 126:
+        length = int.from_bytes(_recv_exact(sock, 2), "big")
+    elif length == 127:
+        length = int.from_bytes(_recv_exact(sock, 8), "big")
+    payload = _recv_exact(sock, length)
+    return json.loads(payload.decode("utf-8"))
+
+
+def _recv_exact(sock, n):
+    chunks = []
+    remaining = n
+    while remaining > 0:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            raise ConnectionError("socket closed")
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
+class WsSubscribeRouteTest(unittest.TestCase):
+    """U-024: c_ws_subscribe end-to-end over a real raw socket -- the actual
+    RFC 6455 handshake + framing, not mocked, since that's exactly the
+    hand-rolled logic (no `websockets` dependency, 05 §5.7) most likely to
+    have a real bug."""
+
+    def setUp(self):
+        self.server = backend_server.create_server(db_path=":memory:")
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.server.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.server.conn.commit()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+
+    def _connect(self):
+        host, port = self.server.server_address
+        return socket.create_connection((host, port), timeout=5)
+
+    def test_handshake_succeeds_with_correct_accept_key(self):
+        sock = self._connect()
+        self.addCleanup(sock.close)
+        key = base64.b64encode(b"0123456789012345").decode("ascii")
+        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
+        head = _read_http_response_head(sock)
+        self.assertIn("101 Switching Protocols", head.splitlines()[0])
+        expected_accept = ws_subscribe.compute_accept_key(key)
+        self.assertIn("Sec-WebSocket-Accept: " + expected_accept, head)
+
+    def test_nonexistent_project_rejected_before_upgrade(self):
+        sock = self._connect()
+        self.addCleanup(sock.close)
+        key = base64.b64encode(b"0123456789012345").decode("ascii")
+        sock.sendall(_ws_handshake_request("/api/ws/projects/no-such-project", key))
+        head = _read_http_response_head(sock)
+        self.assertIn("404", head.splitlines()[0])
+        self.assertNotIn("101", head.splitlines()[0])
+
+    def test_non_upgrade_request_rejected(self):
+        sock = self._connect()
+        self.addCleanup(sock.close)
+        sock.sendall(b"GET /api/ws/projects/proj-1 HTTP/1.1\r\nHost: x\r\n\r\n")
+        head = _read_http_response_head(sock)
+        self.assertIn("400", head.splitlines()[0])
+
+    def test_new_execution_event_pushed_within_two_seconds(self):
+        sock = self._connect()
+        self.addCleanup(sock.close)
+        key = base64.b64encode(b"0123456789012345").decode("ascii")
+        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
+        _read_http_response_head(sock)
+
+        start = time.monotonic()
+        with self.server.conn_lock:
+            self.server.conn.execute(
+                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+                "VALUES ('evt-1', 'proj-1', NULL, 'budget_updated', '{\"x\": 1}')"
+            )
+            self.server.conn.commit()
+
+        sock.settimeout(5)
+        event = _read_server_frame(sock)
+        elapsed = time.monotonic() - start
+        self.assertLess(elapsed, 2.0)
+        self.assertEqual(event["event_id"], "evt-1")
+        self.assertEqual(event["event_type"], "budget_updated")
+        self.assertEqual(event["payload"], {"x": 1})
+
+    def test_task_id_filter_excludes_unrelated_events(self):
+        with self.server.conn_lock:
+            self.server.conn.execute(
+                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+                "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
+            )
+            self.server.conn.commit()
+
+        sock = self._connect()
+        self.addCleanup(sock.close)
+        key = base64.b64encode(b"0123456789012345").decode("ascii")
+        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1?task_id=task-1", key))
+        _read_http_response_head(sock)
+
+        with self.server.conn_lock:
+            self.server.conn.execute(
+                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+                "VALUES ('evt-unrelated', 'proj-1', NULL, 'budget_updated', '{}')"
+            )
+            self.server.conn.execute(
+                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+                "VALUES ('evt-related', 'proj-1', 'task-1', 'task_status_changed', '{}')"
+            )
+            self.server.conn.commit()
+
+        sock.settimeout(5)
+        event = _read_server_frame(sock)
+        self.assertEqual(event["event_id"], "evt-related")
+
+
 class ZeroThirdPartyDependencyTest(unittest.TestCase):
     """05 §5.7: no Flask/FastAPI/Express/Node.js -- checked statically so this
     doesn't silently rot if someone adds an import later."""
+"""U-024 tests: c_ws_subscribe's pure pieces (handshake key, frame
+encode/decode, event polling) tested directly, no live socket needed for
+these -- the live-socket end-to-end path is covered in
+src/test_backend_server.py's WsSubscribeRouteTest.
+
+Run: python3 -m unittest src.test_ws_subscribe -v
+"""
+
+import io
+import json
+import unittest
+import uuid
+
+from src import schema, ws_subscribe
+
+
+class ComputeAcceptKeyTest(unittest.TestCase):
+    def test_matches_rfc6455_example(self):
+        # RFC 6455 §1.3's own worked example
+        self.assertEqual(
+            ws_subscribe.compute_accept_key("dGhlIHNhbXBsZSBub25jZQ=="),
+            "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",
+        )
+
+
+class _FakeSocket:
+    """Minimal recv()-only stand-in so decode_client_frame can be tested
+    without a real socket -- feeds back a bytes buffer in arbitrary chunks."""
+
+    def __init__(self, data):
+        self._buf = io.BytesIO(data)
+
+    def recv(self, n):
+        return self._buf.read(n)
+
+
+class FrameEncodeDecodeTest(unittest.TestCase):
+    def test_text_frame_roundtrip_short_payload(self):
+        frame = ws_subscribe.encode_text_frame("hello")
+        # server frames aren't masked -- decode via a masked wrapper to
+        # reuse decode_client_frame's parsing logic for the length/opcode
+        # bits (masking bit is separately exercised in the masked test).
+        self.assertEqual(frame[0], 0x81)  # FIN=1, opcode=text
+        self.assertEqual(frame[1], 5)  # unmasked, length=5
+        self.assertEqual(frame[2:], b"hello")
+
+    def test_text_frame_medium_payload_uses_16bit_length(self):
+        text = "x" * 200
+        frame = ws_subscribe.encode_text_frame(text)
+        self.assertEqual(frame[1], 126)
+        self.assertEqual(int.from_bytes(frame[2:4], "big"), 200)
+        self.assertEqual(frame[4:].decode(), text)
+
+    def test_decode_masked_client_frame(self):
+        payload = b"ping-ish"
+        mask_key = bytes([0x12, 0x34, 0x56, 0x78])
+        masked_payload = bytes(b ^ mask_key[i % 4] for i, b in enumerate(payload))
+        raw = bytes([0x81, 0x80 | len(payload)]) + mask_key + masked_payload
+        opcode, decoded = ws_subscribe.decode_client_frame(_FakeSocket(raw))
+        self.assertEqual(opcode, ws_subscribe.OPCODE_TEXT)
+        self.assertEqual(decoded, payload)
+
+    def test_decode_close_frame_empty_payload(self):
+        raw = bytes([0x88, 0x80]) + bytes([0, 0, 0, 0])  # masked, zero-length
+        opcode, decoded = ws_subscribe.decode_client_frame(_FakeSocket(raw))
+        self.assertEqual(opcode, ws_subscribe.OPCODE_CLOSE)
+        self.assertEqual(decoded, b"")
+
+    def test_decode_raises_connection_error_on_truncated_socket(self):
+        with self.assertRaises(ConnectionError):
+            ws_subscribe.decode_client_frame(_FakeSocket(b"\x81"))
+
+
+class PollNewEventsTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-2', 'P2', '/tmp/proj-2', 1, 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'queued', 'planner')"
+        )
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def _insert_event(self, project_id, event_type="task_status_changed", task_id=None, payload=None):
+        self.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (str(uuid.uuid4()), project_id, task_id, event_type, json.dumps(payload or {})),
+        )
+        self.conn.commit()
+
+    def test_no_new_events_returns_empty_list_and_same_rowid(self):
+        after = ws_subscribe.latest_rowid(self.conn, "proj-1")
+        events, new_after = ws_subscribe.poll_new_events(self.conn, "proj-1", after)
+        self.assertEqual(events, [])
+        self.assertEqual(new_after, after)
+
+    def test_new_event_after_baseline_is_returned(self):
+        after = ws_subscribe.latest_rowid(self.conn, "proj-1")
+        self._insert_event("proj-1", event_type="budget_updated", payload={"x": 1})
+        events, new_after = ws_subscribe.poll_new_events(self.conn, "proj-1", after)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["event_type"], "budget_updated")
+        self.assertEqual(events[0]["payload"], {"x": 1})
+        self.assertGreater(new_after, after)
+
+    def test_events_from_other_project_are_not_returned(self):
+        after = ws_subscribe.latest_rowid(self.conn, "proj-1")
+        self._insert_event("proj-2")
+        events, _ = ws_subscribe.poll_new_events(self.conn, "proj-1", after)
+        self.assertEqual(events, [])
+
+    def test_task_id_filter_narrows_results(self):
+        after = ws_subscribe.latest_rowid(self.conn, "proj-1")
+        self._insert_event("proj-1", task_id="task-1")
+        self._insert_event("proj-1", task_id=None)
+        events, _ = ws_subscribe.poll_new_events(self.conn, "proj-1", after, task_id="task-1")
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["task_id"], "task-1")
+
+    def test_already_seen_events_not_returned_again(self):
+        after = ws_subscribe.latest_rowid(self.conn, "proj-1")
+        self._insert_event("proj-1")
+        events, after2 = ws_subscribe.poll_new_events(self.conn, "proj-1", after)
+        self.assertEqual(len(events), 1)
+        events_again, after3 = ws_subscribe.poll_new_events(self.conn, "proj-1", after2)
+        self.assertEqual(events_again, [])
+        self.assertEqual(after2, after3)
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""c_ws_subscribe (03-C §3.14): real-time execution_event push over a raw
+WebSocket connection (RFC 6455), hand-rolled with hashlib/base64/socket --
+pure stdlib, no `websockets`/`tornado` (05 §5.7).
+
+U-024 scope ("最小连通版"): a poll loop is what actually pushes events, not
+a write-side hook -- execution_event rows are inserted from many different
+call sites across this codebase (org_node_trigger, work_graph_execute,
+work_node_run, task_create, bug_record, ...) with no single write path to
+instrument; polling the table every POLL_INTERVAL_SECONDS decouples this
+from all of them and easily clears REQ-12's own <=2s target (framed there
+as a suggested non-functional goal, not a hard SLA). Reconnect catch-up is
+explicitly routed through REST snapshot endpoints per 03-C §3.14's own
+"断线重连约束" note, not WS-layer replay -- a fresh connection only sees
+events from its own connect time onward, no history replay to build here.
+REQ-12's boundary/failure cases are U-024.5's job once more event-generating
+Units exist to test against.
+"""
+
+import base64
+import hashlib
+import json
+
+_WS_MAGIC_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+POLL_INTERVAL_SECONDS = 0.5
+
+OPCODE_TEXT = 0x1
+OPCODE_CLOSE = 0x8
+OPCODE_PING = 0x9
+OPCODE_PONG = 0xA
+
+
+def compute_accept_key(client_key):
+    """RFC 6455 §1.3 handshake: SHA1(client_key + magic GUID), base64-encoded."""
+    digest = hashlib.sha1((client_key + _WS_MAGIC_GUID).encode("ascii")).digest()
+    return base64.b64encode(digest).decode("ascii")
+
+
+def encode_frame(payload_bytes, opcode=OPCODE_TEXT):
+    """Server->client frames are never masked (RFC 6455 §5.1)."""
+    length = len(payload_bytes)
+    if length < 126:
+        header = bytes([0x80 | opcode, length])
+    elif length < 65536:
+        header = bytes([0x80 | opcode, 126]) + length.to_bytes(2, "big")
+    else:
+        header = bytes([0x80 | opcode, 127]) + length.to_bytes(8, "big")
+    return header + payload_bytes
+
+
+def encode_text_frame(text):
+    return encode_frame(text.encode("utf-8"), opcode=OPCODE_TEXT)
+
+
+def decode_client_frame(sock):
+    """Reads one client->server frame (always masked per RFC 6455 §5.1).
+    Returns (opcode, payload_bytes). Raises ConnectionError if the socket
+    closes mid-frame; a plain read timeout propagates as socket.timeout,
+    left for the caller to catch (that's the normal "nothing arrived this
+    poll tick" case, not an error)."""
+    first_two = _recv_exact(sock, 2)
+    opcode = first_two[0] & 0x0F
+    masked = first_two[1] & 0x80
+    length = first_two[1] & 0x7F
+    if length == 126:
+        length = int.from_bytes(_recv_exact(sock, 2), "big")
+    elif length == 127:
+        length = int.from_bytes(_recv_exact(sock, 8), "big")
+    mask_key = _recv_exact(sock, 4) if masked else b""
+    payload = _recv_exact(sock, length)
+    if masked:
+        payload = bytes(b ^ mask_key[i % 4] for i, b in enumerate(payload))
+    return opcode, payload
+
+
+def _recv_exact(sock, n):
+    if n == 0:
+        return b""
+    chunks = []
+    remaining = n
+    while remaining > 0:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            raise ConnectionError("socket closed while reading a WebSocket frame")
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
+def latest_rowid(conn, project_id):
+    """Starting point for a fresh subscription -- 'now', not full history."""
+    row = conn.execute(
+        "SELECT MAX(rowid) FROM execution_event WHERE project_id = ?", (project_id,)
+    ).fetchone()
+    return row[0] or 0
+
+
+def poll_new_events(conn, project_id, after_rowid, task_id=None):
+    """Returns (events, new_after_rowid) for execution_event rows with
+    rowid > after_rowid, matching project_id (and task_id if given, per
+    03-C §3.14's optional server-side filter), oldest first. Each event
+    dict matches the output_contract field-for-field."""
+    query = (
+        "SELECT rowid, event_id, event_type, task_id, run_id, node_exec_id, payload, created_at "
+        "FROM execution_event WHERE project_id = ? AND rowid > ?"
+    )
+    params = [project_id, after_rowid]
+    if task_id:
+        query += " AND task_id = ?"
+        params.append(task_id)
+    query += " ORDER BY rowid"
+    rows = conn.execute(query, params).fetchall()
+    events = [
+        {
+            "event_id": r[1],
+            "event_type": r[2],
+            "task_id": r[3],
+            "run_id": r[4],
+            "node_exec_id": r[5],
+            "payload": json.loads(r[6]),
+            "created_at": r[7],
+        }
+        for r in rows
+    ]
+    new_after_rowid = rows[-1][0] if rows else after_rowid
+    return events, new_after_rowid