From 82b088118693d288e839e9dd127670af159cb89f Mon Sep 17 00:00:00 2001 From: bboonnee Date: Fri, 31 Jul 2026 17:18:05 +0800 Subject: [PATCH] U-024: c_ws_subscribe + execution_event 实时推送（最小连通版） --- docs/development-plan.md | 1 + docs/sdd/unit_matrix.json | 2 +- src/backend_server.py | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- src/test_backend_server.py | 159 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/test_ws_subscribe.py | 143 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/ws_subscribe.py | 125 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 508 insertions(+), 7 deletions(-) create mode 100644 src/test_ws_subscribe.py create mode 100644 src/ws_subscribe.py diff --git a/docs/development-plan.md b/docs/development-plan.md index 38baffe..87885ed 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -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 完整验收（事件多样性联调） diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 5a76edc..ab7192d 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -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 信息态"}, diff --git a/src/backend_server.py b/src/backend_server.py index 497ff38..e1d0856 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -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() diff --git a/src/test_backend_server.py b/src/test_backend_server.py index a30fa97..de653a8 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -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.""" diff --git a/src/test_ws_subscribe.py b/src/test_ws_subscribe.py new file mode 100644 index 0000000..acdd4da --- /dev/null +++ b/src/test_ws_subscribe.py @@ -0,0 +1,143 @@ +"""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() diff --git a/src/ws_subscribe.py b/src/ws_subscribe.py new file mode 100644 index 0000000..06f5888 --- /dev/null +++ b/src/ws_subscribe.py @@ -0,0 +1,125 @@ +"""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 -- libgit2 0.21.4