Add client-side Server-Sent Events parser

README.md

@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-21) — Server-Sent Events (SSE) Client Parser](#whats-new-2026-06-21--server-sent-events-sse-client-parser)
 - [What's new (2026-06-21) — Dotenv (.env) Parsing](#whats-new-2026-06-21--dotenv-env-parsing)
 - [What's new (2026-06-21) — RFC 9457 Problem Details Parsing](#whats-new-2026-06-21--rfc-9457-problem-details-parsing)
 - [What's new (2026-06-21) — Data Profiling & Schema Inference](#whats-new-2026-06-21--data-profiling--schema-inference)
@@ -132,6 +133,12 @@
 
 ---
 
+## What's new (2026-06-21) — Server-Sent Events (SSE) Client Parser
+
+Consume `text/event-stream` responses. Full reference: [`docs/source/Eng/doc/new_features/v80_features_doc.rst`](docs/source/Eng/doc/new_features/v80_features_doc.rst).
+
+- **`parse_event_stream` / `SSEParser` / `SSEEvent`** (`AC_parse_sse`): the MCP HTTP transport emits SSE, but nothing consumed it — a streaming LLM/agent/chatops endpoint left `http_request` with a raw blob. This implements the WHATWG event-stream parsing algorithm (`event`/`data`/`id`/`retry`, comments, the leading-space rule, blank-line dispatch) with an incremental `feed` for chunks and a one-shot `parse_event_stream`. Pure-stdlib, fully deterministic.
+
 ## What's new (2026-06-21) — Dotenv (.env) Parsing
 
 Read 12-factor `.env` files into config. Full reference: [`docs/source/Eng/doc/new_features/v79_features_doc.rst`](docs/source/Eng/doc/new_features/v79_features_doc.rst).

README/README_zh-CN.md

@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-21) — Server-Sent Events (SSE) 客户端解析器](#本次更新-2026-06-21--server-sent-events-sse-客户端解析器)
 - [本次更新 (2026-06-21) — Dotenv (.env) 解析](#本次更新-2026-06-21--dotenv-env-解析)
 - [本次更新 (2026-06-21) — RFC 9457 Problem Details 解析](#本次更新-2026-06-21--rfc-9457-problem-details-解析)
 - [本次更新 (2026-06-21) — 数据剖析与结构推断](#本次更新-2026-06-21--数据剖析与结构推断)
@@ -131,6 +132,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — Server-Sent Events (SSE) 客户端解析器
+
+消费 `text/event-stream` 响应。完整参考:[`docs/source/Zh/doc/new_features/v80_features_doc.rst`](../docs/source/Zh/doc/new_features/v80_features_doc.rst)。
+
+- **`parse_event_stream` / `SSEParser` / `SSEEvent`**(`AC_parse_sse`):MCP 的 HTTP 传输会发出 SSE,但没有任何东西消费它 —— 一个流式的 LLM/agent/chatops 端点会让 `http_request` 拿到原始 blob。本功能实现 WHATWG event-stream 解析算法(`event`/`data`/`id`/`retry`、注释、前导空格规则、空行派发),并提供逐块的增量 `feed` 与一次性的 `parse_event_stream`。纯标准库、完全确定。
+
 ## 本次更新 (2026-06-21) — Dotenv (.env) 解析
 
 把 12-factor `.env` 文件读进配置。完整参考:[`docs/source/Zh/doc/new_features/v79_features_doc.rst`](../docs/source/Zh/doc/new_features/v79_features_doc.rst)。

README/README_zh-TW.md

@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-21) — Server-Sent Events (SSE) 用戶端解析器](#本次更新-2026-06-21--server-sent-events-sse-用戶端解析器)
 - [本次更新 (2026-06-21) — Dotenv (.env) 解析](#本次更新-2026-06-21--dotenv-env-解析)
 - [本次更新 (2026-06-21) — RFC 9457 Problem Details 解析](#本次更新-2026-06-21--rfc-9457-problem-details-解析)
 - [本次更新 (2026-06-21) — 資料剖析與結構推斷](#本次更新-2026-06-21--資料剖析與結構推斷)
@@ -131,6 +132,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — Server-Sent Events (SSE) 用戶端解析器
+
+消費 `text/event-stream` 回應。完整參考:[`docs/source/Zh/doc/new_features/v80_features_doc.rst`](../docs/source/Zh/doc/new_features/v80_features_doc.rst)。
+
+- **`parse_event_stream` / `SSEParser` / `SSEEvent`**(`AC_parse_sse`):MCP 的 HTTP 傳輸會發出 SSE,但沒有任何東西消費它 —— 一個串流的 LLM/agent/chatops 端點會讓 `http_request` 拿到原始 blob。本功能實作 WHATWG event-stream 解析演算法(`event`/`data`/`id`/`retry`、註解、前導空白規則、空白行派發),並提供逐塊的增量 `feed` 與一次性的 `parse_event_stream`。純標準函式庫、完全具決定性。
+
 ## 本次更新 (2026-06-21) — Dotenv (.env) 解析
 
 把 12-factor `.env` 檔案讀進設定。完整參考:[`docs/source/Zh/doc/new_features/v79_features_doc.rst`](../docs/source/Zh/doc/new_features/v79_features_doc.rst)。

docs/source/Eng/doc/new_features/v80_features_doc.rst

@@ -0,0 +1,45 @@
+Server-Sent Events (SSE) Client Parser
+======================================
+
+The MCP HTTP transport *emits* Server-Sent Events, but nothing consumed them:
+an LLM, agent, or chatops endpoint that streams ``text/event-stream`` left
+``http_request`` holding a raw, unparsed blob. This implements the WHATWG
+event-stream parsing algorithm — ``event`` / ``data`` / ``id`` / ``retry``
+fields, comment lines, the leading-space rule, and blank-line dispatch — with
+an incremental ``feed`` for streamed chunks.
+
+Pure standard library (``re``); imports no ``PySide6``. The parser is pure and
+fully deterministic, so streaming logic is CI-testable without a live server.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import parse_event_stream, SSEParser
+
+    # Parse a complete response body:
+    for event in parse_event_stream(response_text):
+        handle(event.event, event.data, event.id)
+
+    # Or parse incrementally as chunks arrive:
+    parser = SSEParser()
+    for chunk in stream:
+        for event in parser.feed(chunk):
+            handle(event)
+    for event in parser.close():           # flush a trailing event
+        handle(event)
+
+``SSEEvent`` is the dispatched ``(event, data, id, retry)`` tuple (``event``
+defaulting to ``"message"``). ``SSEParser.feed`` buffers a partial trailing line
+across calls and returns each event completed by a blank line; ``close``
+flushes a final event when the stream ends without one. ``id`` and ``retry``
+persist across subsequent events per the spec. ``parse_event_stream`` is the
+one-shot helper for a complete blob and flushes the trailing event.
+
+Executor command
+----------------
+
+``AC_parse_sse`` parses a ``text`` blob into ``{events}`` (each
+``{event, data, id, retry}``). It is exposed as the MCP tool ``ac_parse_sse``
+and as a Script Builder command under **Data**.

docs/source/Eng/eng_index.rst

@@ -102,6 +102,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v77_features_doc
    doc/new_features/v78_features_doc
    doc/new_features/v79_features_doc
+   doc/new_features/v80_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

docs/source/Zh/doc/new_features/v80_features_doc.rst

@@ -0,0 +1,40 @@
+Server-Sent Events(SSE)用戶端解析器
+==================================
+
+MCP 的 HTTP 傳輸會*發出* Server-Sent Events,但沒有任何東西消費它:一個串流 ``text/event-stream``
+的 LLM、agent 或 chatops 端點,會讓 ``http_request`` 拿到未經解析的原始 blob。本功能實作 WHATWG
+event-stream 解析演算法 —— ``event`` / ``data`` / ``id`` / ``retry`` 欄位、註解行、前導空白規則,以及
+空白行派發 —— 並提供逐塊串流的增量 ``feed``。
+
+純標準函式庫(``re``);不匯入 ``PySide6``。解析器為純函式且完全具決定性,因此串流邏輯可在無線上伺服器
+的情況下於 CI 測試。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import parse_event_stream, SSEParser
+
+    # 解析完整回應內文:
+    for event in parse_event_stream(response_text):
+        handle(event.event, event.data, event.id)
+
+    # 或在資料塊抵達時逐塊解析:
+    parser = SSEParser()
+    for chunk in stream:
+        for event in parser.feed(chunk):
+            handle(event)
+    for event in parser.close():           # 沖出尾端事件
+        handle(event)
+
+``SSEEvent`` 是派發出的 ``(event, data, id, retry)`` 組合(``event`` 預設 ``"message"``)。
+``SSEParser.feed`` 在多次呼叫間緩衝尾端不完整的行,並回傳每個由空白行完成的事件;``close`` 會在串流
+未以空白行結尾時沖出最後一個事件。``id`` 與 ``retry`` 依規範在後續事件間延續。``parse_event_stream``
+是處理完整 blob 的一次性輔助函式,並會沖出尾端事件。
+
+執行器命令
+----------
+
+``AC_parse_sse`` 把 ``text`` blob 解析成 ``{events}``(每筆 ``{event, data, id, retry}``)。它以 MCP
+工具 ``ac_parse_sse`` 以及 Script Builder 中 **Data** 分類下的命令提供。

docs/source/Zh/zh_index.rst

@@ -102,6 +102,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v77_features_doc
    doc/new_features/v78_features_doc
    doc/new_features/v79_features_doc
+   doc/new_features/v80_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

je_auto_control/__init__.py

@@ -388,6 +388,10 @@
     HttpProblemError, ProblemDetails, is_problem, parse_problem,
     raise_for_problem,
 )
+# Server-Sent Events (text/event-stream) client parser
+from je_auto_control.utils.sse_client import (
+    SSEEvent, SSEParser, parse_event_stream,
+)
 # W3C Trace Context propagation (traceparent / tracestate)
 from je_auto_control.utils.trace_context import (
     SpanContext, TraceContextError, child_context, extract_context,
@@ -898,6 +902,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "Cassette", "CassetteMissError",
     "HttpProblemError", "ProblemDetails", "is_problem", "parse_problem",
     "raise_for_problem",
+    "SSEEvent", "SSEParser", "parse_event_stream",
     "SpanContext", "TraceContextError", "child_context", "extract_context",
     "format_traceparent", "format_tracestate", "inject_context",
     "new_root_context", "new_span_id", "new_trace_id", "parse_traceparent",

je_auto_control/gui/script_builder/command_schema.py

@@ -1670,6 +1670,14 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Load a .env file into a values dict.",
     ))
+    specs.append(CommandSpec(
+        "AC_parse_sse", "Data", "SSE: Parse Event Stream",
+        fields=(
+            FieldSpec("text", FieldType.STRING,
+                      placeholder='event: ping\ndata: {"x": 1}\n\n'),
+        ),
+        description="Parse a text/event-stream blob into events.",
+    ))
     specs.append(CommandSpec(
         "AC_rate_limit", "Flow", "Rate Limit (Token Bucket)",
         fields=(

je_auto_control/utils/executor/action_executor.py

@@ -3044,6 +3044,12 @@ def _load_dotenv(path: str, override: Any = False) -> Dict[str, Any]:
     return {"values": load_dotenv(path, {}, override=bool(override))}
 
 
+def _parse_sse(text: str) -> Dict[str, Any]:
+    """Adapter: parse a text/event-stream blob into {events}."""
+    from je_auto_control.utils.sse_client import parse_event_stream
+    return {"events": [event.to_dict() for event in parse_event_stream(text)]}
+
+
 def _percentiles(samples: Any, qs: Any = None) -> Dict[str, Any]:
     """Adapter: exact percentiles of a numeric sample list (or JSON string)."""
     import json
@@ -4127,6 +4133,7 @@ def __init__(self):
             "AC_parse_problem": _parse_problem,
             "AC_parse_dotenv": _parse_dotenv,
             "AC_load_dotenv": _load_dotenv,
+            "AC_parse_sse": _parse_sse,
             "AC_unified_diff": _unified_diff,
             "AC_apply_unified": _apply_unified,
             "AC_three_way_merge": _three_way_merge,

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -3343,6 +3343,20 @@ def rate_limit_tools() -> List[MCPTool]:
     ]
 
 
+def sse_client_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_parse_sse",
+            description=("Parse a Server-Sent Events ('text/event-stream') "
+                         "'text' blob into {events} (event/data/id/retry), "
+                         "flushing a trailing event without a final blank line."),
+            input_schema=schema({"text": {"type": "string"}}, ["text"]),
+            handler=h.parse_sse,
+            annotations=READ_ONLY,
+        ),
+    ]
+
+
 def dotenv_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -4971,6 +4985,7 @@ def media_assert_tools() -> List[MCPTool]:
     feature_flag_tools, provenance_tools, json_contract_tools, chaos_tools,
     slo_tools, percentiles_tools, bulkhead_tools, http_cassette_tools,
     trace_context_tools, data_profile_tools, http_problem_tools, dotenv_tools,
+    sse_client_tools,
     saga_tools, decision_table_tools, locator_repair_tools,
     pii_text_tools, sarif_tools,
     screen_record_tools,

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -1751,6 +1751,11 @@ def load_dotenv(path, override=False):
     return _load_dotenv(path, override)
 
 
+def parse_sse(text):
+    from je_auto_control.utils.executor.action_executor import _parse_sse
+    return _parse_sse(text)
+
+
 def build_provenance(paths, builder_id="je_auto_control"):
     from je_auto_control.utils.provenance import build_provenance, subject_for
     subjects = [subject_for(path) for path in paths]

je_auto_control/utils/sse_client/__init__.py

@@ -0,0 +1,6 @@
+"""Client-side Server-Sent Events parsing for AutoControl."""
+from je_auto_control.utils.sse_client.sse_client import (
+    SSEEvent, SSEParser, parse_event_stream,
+)
+
+__all__ = ["SSEEvent", "SSEParser", "parse_event_stream"]

je_auto_control/utils/sse_client/sse_client.py

@@ -0,0 +1,106 @@
+"""Client-side Server-Sent Events (``text/event-stream``) parser.
+
+The MCP HTTP transport *emits* SSE, but nothing consumed it: an LLM, agent, or
+chatops endpoint that streams ``text/event-stream`` left ``http_request`` with a
+raw, unparsed blob. This implements the WHATWG event-stream parsing algorithm —
+``event`` / ``data`` / ``id`` / ``retry`` fields, comment lines, the leading
+space rule, and blank-line dispatch — with incremental ``feed`` for chunks.
+
+Pure standard library (``re``); imports no ``PySide6``. The parser is pure and
+fully deterministic, so streaming logic is CI-testable without a live server.
+"""
+import re
+from dataclasses import dataclass
+from typing import List, Optional
+
+_LINE_SPLIT = re.compile(r"\r\n|\r|\n")
+
+
+@dataclass(frozen=True)
+class SSEEvent:
+    """One dispatched Server-Sent Event."""
+
+    event: str = "message"
+    data: str = ""
+    id: Optional[str] = None
+    retry: Optional[int] = None
+
+    def to_dict(self) -> dict:
+        """Return a JSON-friendly view of the event."""
+        return {"event": self.event, "data": self.data,
+                "id": self.id, "retry": self.retry}
+
+
+class SSEParser:
+    """Incremental WHATWG ``text/event-stream`` parser."""
+
+    def __init__(self) -> None:
+        self._buffer = ""
+        self._event = ""
+        self._data: List[str] = []
+        self._last_id: Optional[str] = None
+        self._retry: Optional[int] = None
+
+    def feed(self, chunk: str) -> List[SSEEvent]:
+        """Feed a chunk of stream text; return any complete events."""
+        lines = _LINE_SPLIT.split(self._buffer + chunk)
+        self._buffer = lines.pop()          # trailing partial line
+        events = [event for event in (self._process_line(line)
+                                      for line in lines) if event is not None]
+        return events
+
+    def close(self) -> List[SSEEvent]:
+        """Flush a trailing line and any pending event at end of stream."""
+        events: List[SSEEvent] = []
+        if self._buffer:
+            trailing = self._process_line(self._buffer)
+            self._buffer = ""
+            if trailing is not None:
+                events.append(trailing)
+        final = self._dispatch()
+        if final is not None:
+            events.append(final)
+        return events
+
+    def _process_line(self, line: str) -> Optional[SSEEvent]:
+        if line == "":
+            return self._dispatch()
+        if line.startswith(":"):
+            return None
+        field, _, value = line.partition(":")
+        if value.startswith(" "):
+            value = value[1:]
+        self._set_field(field, value)
+        return None
+
+    def _set_field(self, field: str, value: str) -> None:
+        if field == "event":
+            self._event = value
+        elif field == "data":
+            self._data.append(value)
+        elif field == "id" and "\x00" not in value:
+            self._last_id = value
+        elif field == "retry" and value.isdigit():
+            self._retry = int(value)
+
+    def _dispatch(self) -> Optional[SSEEvent]:
+        if not self._data:
+            self._event = ""
+            return None
+        event = SSEEvent(event=self._event or "message",
+                         data="\n".join(self._data),
+                         id=self._last_id, retry=self._retry)
+        self._event = ""
+        self._data = []
+        return event
+
+
+def parse_event_stream(text: str) -> List[SSEEvent]:
+    """Parse a complete ``text/event-stream`` blob into events.
+
+    Flushes a trailing event even when the stream lacks a final blank line.
+    """
+    parser = SSEParser()
+    events = parser.feed(text or "")
+    events.extend(parser.close())
+    return events