feat: async + SSE message flow with interactive permission/question UI

Replace the synchronous /edit (wait for full markdown response) with an
async + SSE flow per demo.html so the user sees text stream in real-time
and can interact with permission/question events the agent raises.

Backend
-------
- EditHandler now calls OpenCode /session/:id/prompt_async and returns
  immediately with {ok, sessionId, notebookPath}; the LLM reply is no
  longer embedded in this response.
- New GlobalEventHandler proxies OpenCode /global/event as
  text/event-stream. Server forwards ALL events; the client filters.
  A too-eager server-side ?session= filter was silently dropping events
  the client would have accepted, so it was removed.
- New PermissionReplyHandler + QuestionReplyHandler forward user
  replies (once/always/reject and freeform answer) back to OpenCode
  Serve at /session/:sid/permissions/:permId and
  /session/:sid/question/:qId/reply.
- OpenCodeClient gains send_message_async(), stream_global_events()
  (async generator over the SSE feed via tornado streaming_callback),
  reply_permission() and reply_question(). Legacy send_message_sync
  removed.

Frontend
--------
- subscribeOpenCodeEvents() opens a fetch+reader SSE client (XSRF
  token injected from serverSettings); AbortController-backed close()
  is idempotent.
- OpenCodeInlinePrompt.applyEvent() routes events into the streaming
  UI: text delta -> assistant message (re-rendered as markdown on
  every delta so the user sees formatted <pre><code> blocks in
  real-time, not raw fence source); reasoning/tool/permission/question
  get collapsible details blocks. session.idle resets stream pointers
  and fires onStreamEnd.
- Permission/question blocks render real interactive UI: three
  buttons (once/always/reject) for permission, a text input + submit
  for question. Click handlers post through the new API routes and
  show success/failure status in place.
- Centralized idle detection (4 shapes: top-level + payload-nested
  x {session.idle, session.status/idle}) inside the prompt; cell
  action reacts via onStreamEnd callback rather than re-parsing
  event types.
- System / workspace / pty / lsp / mcp / installation events AND
  session-level control events (agent.switched, model.switched,
  file.edited) are not rendered in the frontend.

Tests
-----
- 52 backend pytest (was 43)
- 58 frontend jest (was 46)

design.md section 3 updated for the new async /edit response shape
and the new GET /events SSE endpoint.

Co-Authored-By: Claude <noreply@anthropic.com>

opencode_bridge/opencode_client.py

@@ -1,8 +1,22 @@
-"""Async HTTP client for the local OpenCode server."""
+"""Async HTTP client for the local OpenCode server.
 
+Two interaction modes:
+
+- **Request/response** (one-shot) - ``health``, ``list_providers``,
+  ``create_session``, ``send_message_async``, ``abort``, ``delete_session``,
+  ``list_session_messages``. ``send_message_async`` fires the prompt and
+  returns immediately; the actual LLM response is consumed via the SSE
+  stream (next mode).
+- **Streaming** (long-lived) - :meth:`stream_global_events` is an async
+  generator over OpenCode Serve's ``GET /global/event`` SSE feed. It
+  yields each event as a parsed JSON dict until the upstream closes.
+"""
+
+import asyncio
 import atexit
 import json
-from typing import Any, Optional
+import logging
+from typing import Any, AsyncIterator, Optional
 
 from tornado.httpclient import AsyncHTTPClient, HTTPRequest
 from tornado.httputil import HTTPHeaders
@@ -10,6 +24,9 @@ from tornado.httputil import HTTPHeaders
 from .config import OpenCodeConfig
 
 
+log = logging.getLogger("opencode_bridge.opencode_client")
+
+
 class OpenCodeError(Exception):
     """Raised when an OpenCode API request fails."""
 
@@ -64,22 +81,28 @@ class OpenCodeClient:
     async def create_session(self, title: str) -> dict[str, Any]:
         return await self._request("POST", "/session", {"title": title})
 
-    async def send_message_sync(
+    async def send_message_async(
         self,
         session_id: str,
         parts: list[dict[str, Any]],
         provider_id: Optional[str] = None,
         model_id: Optional[str] = None,
         system: Optional[str] = None,
-    ) -> dict[str, Any]:
+    ) -> None:
+        """Fire-and-forget POST /session/:id/prompt_async.
+
+        Returns once OpenCode has accepted the prompt and queued it for
+        processing - NOT when the LLM is done. Subscribe to
+        :meth:`stream_global_events` to follow the agent's progress.
+        """
         body: dict[str, Any] = {"parts": parts}
         if provider_id is not None and model_id is not None:
             body["model"] = {"providerID": provider_id, "modelID": model_id}
         if system is not None:
             body["system"] = system
-        return await self._request(
+        await self._request(
             "POST",
-            "/session/%s/message" % session_id,
+            "/session/%s/prompt_async" % session_id,
             body,
         )
 
@@ -87,6 +110,41 @@ class OpenCodeClient:
         result = await self._request("POST", "/session/%s/abort" % session_id)
         return result is not None
 
+    async def reply_permission(
+        self,
+        session_id: str,
+        permission_id: str,
+        response: str,
+    ) -> bool:
+        """Respond to a `permission.asked` event.
+
+        ``response`` must be one of ``"once"`` / ``"always"`` / ``"reject"``,
+        matching the OpenCode Serve API. Returns True on success.
+        """
+        result = await self._request(
+            "POST",
+            "/session/%s/permissions/%s" % (session_id, permission_id),
+            {"response": response},
+        )
+        return result is not None
+
+    async def reply_question(
+        self,
+        session_id: str,
+        question_id: str,
+        answer: str,
+    ) -> bool:
+        """Reply to a `question.asked` event with the user's freeform answer.
+
+        Returns True on success.
+        """
+        result = await self._request(
+            "POST",
+            "/session/%s/question/%s/reply" % (session_id, question_id),
+            {"answer": answer},
+        )
+        return result is not None
+
     async def delete_session(self, session_id: str) -> bool:
         result = await self._request("DELETE", "/session/%s" % session_id)
         return result is not None
@@ -95,9 +153,9 @@ class OpenCodeClient:
         """List all messages in the given OpenCode session.
 
         Returns the raw OpenCode response: a list of
-        `{ info: { role: "user"|"assistant", ... }, parts: [...] }`.
+        ``{ info: { role: "user"|"assistant", ... }, parts: [...] }``.
         The server extension is responsible for projecting this into a
-        frontend-friendly `{role, content}[]` shape.
+        frontend-friendly ``{role, content}[]`` shape.
         """
         return await self._request("GET", "/session/%s/message" % session_id)
 
@@ -105,6 +163,104 @@ class OpenCodeClient:
     def endpoint(self) -> str:
         return self._config.url
 
+    async def stream_global_events(self) -> AsyncIterator[dict[str, Any]]:
+        """Stream OpenCode Serve's ``GET /global/event`` SSE feed.
+
+        Yields one parsed JSON dict per ``data: {...}`` event block. The
+        stream runs until:
+
+        - the upstream disconnects (end of stream),
+        - the consumer stops iterating (caller cancels this generator),
+        - or a network error occurs (logged and the generator returns).
+
+        The consumer is responsible for filtering events by session ID
+        (the upstream pushes events for ALL sessions, plus system events).
+        """
+        queue: asyncio.Queue = asyncio.Queue()
+        end_sentinel = object()
+
+        def on_chunk(chunk: bytes) -> None:
+            if chunk:
+                queue.put_nowait(chunk)
+
+        headers = HTTPHeaders({
+            "Accept": "text/event-stream",
+            "Cache-Control": "no-cache",
+        })
+        request_kwargs: dict[str, Any] = {
+            "method": "GET",
+            "headers": headers,
+            "request_timeout": 0,  # 0 = no timeout for long-lived streams
+            # streaming_callback must live on the HTTPRequest itself; tornado
+            # rejects any fetch() kwargs when an HTTPRequest is passed.
+            "streaming_callback": on_chunk,
+        }
+        if self._config.auth is not None:
+            request_kwargs["auth_username"] = self._config.auth[0]
+            request_kwargs["auth_password"] = self._config.auth[1]
+
+        request = HTTPRequest(self._config.url + "/global/event", **request_kwargs)
+
+        async def _run_fetch() -> None:
+            try:
+                await self._http_client.fetch(request, raise_error=False)
+            except Exception as e:
+                log.warning("OpenCode /global/event fetch error: %s", e)
+            finally:
+                queue.put_nowait(end_sentinel)
+
+        runner_task = asyncio.create_task(_run_fetch())
+
+        buffer = ""
+        try:
+            while True:
+                chunk = await queue.get()
+                if chunk is end_sentinel:
+                    break
+                if not isinstance(chunk, bytes):
+                    continue
+                buffer += chunk.decode("utf-8", errors="replace")
+                while "\n\n" in buffer:
+                    block, buffer = buffer.split("\n\n", 1)
+                    parsed = self._parse_sse_block(block)
+                    if parsed is not None:
+                        yield parsed
+            # Drain any trailing data after the last \n\n
+            if buffer.strip():
+                parsed = self._parse_sse_block(buffer)
+                if parsed is not None:
+                    yield parsed
+        finally:
+            if not runner_task.done():
+                runner_task.cancel()
+                try:
+                    await runner_task
+                except (asyncio.CancelledError, Exception):
+                    pass
+
+    @staticmethod
+    def _parse_sse_block(block: str) -> Optional[dict[str, Any]]:
+        """Parse a single SSE event block into a JSON dict (or None).
+
+        SSE format reminder: a block is one or more ``field: value`` lines
+        separated by newlines; multiple ``data:`` lines are concatenated
+        with ``\\n``. We only care about ``data:``.
+        """
+        data_parts: list[str] = []
+        for line in block.split("\n"):
+            if line.startswith("data:"):
+                data_parts.append(line[5:].lstrip())
+        if not data_parts:
+            return None
+        data = "\n".join(data_parts).strip()
+        if not data:
+            return None
+        try:
+            result = json.loads(data)
+        except json.JSONDecodeError:
+            return None
+        return result if isinstance(result, dict) else None
+
     async def _request(
         self,
         method: str,

opencode_bridge/routes.py

@@ -49,7 +49,7 @@ def get_session_manager(handler: APIHandler) -> SessionManager:
 
 
 def _build_request_body(prompt: str, context: dict) -> dict:
-    """Build full request body for OpenCode POST /session/:id/message.
+    """Build full request body for OpenCode POST /session/:id/prompt_async.
 
     Returns dict with 'parts' (list) and 'system' (str) keys. No mode concept:
     the LLM interprets the user's natural-language instruction, with the code
@@ -105,6 +105,26 @@ def _strip_code_fence(s: str) -> str:
     return s
 
 
+def _extract_session_id(event: dict) -> str | None:
+    """Best-effort session ID extraction from an OpenCode event.
+
+    OpenCode's event shape is not strictly documented; this mirrors the
+    heuristics in demo.html ``handleGlobalEvent``:
+      properties.sessionID / properties.sessionId /
+      payload.properties.sessionID / syncEvent.aggregateID
+    """
+    props = event.get("properties") or {}
+    sid = props.get("sessionID") or props.get("sessionId")
+    if sid:
+        return sid
+    payload_props = (event.get("payload") or {}).get("properties") or {}
+    sid = payload_props.get("sessionID") or payload_props.get("sessionId")
+    if sid:
+        return sid
+    sync = event.get("syncEvent") or (event.get("payload") or {}).get("syncEvent") or {}
+    return sync.get("aggregateID")
+
+
 class HelloRouteHandler(APIHandler):
     # The following decorator should be present on all verb methods (head, get, post,
     # patch, put, delete, options) to ensure only authorized user can request the
@@ -154,6 +174,13 @@ class ProvidersHandler(APIHandler):
 
 
 class EditHandler(APIHandler):
+    """Async edit: fire prompt_async, return sessionId immediately.
+
+    The LLM's response is consumed via the separate ``/opencode-bridge/events``
+    SSE endpoint. Frontend subscribes to SSE after this call returns and
+    processes events incrementally.
+    """
+
     @tornado.web.authenticated
     async def post(self):
         try:
@@ -178,44 +205,174 @@ class EditHandler(APIHandler):
         try:
             sid = await sm.get_or_create(notebook_path)
             request_body = _build_request_body(prompt, context)
-            result = await client.send_message_sync(
+            await client.send_message_async(
                 sid,
                 request_body["parts"],
                 provider_id=provider_id,
                 model_id=model_id,
                 system=request_body["system"],
             )
-
-            text_parts = [
-                p.get("text", "")
-                for p in result.get("parts", [])
-                if p.get("type") == "text"
-            ]
-            # The AI's reply is markdown (code in ```fences```, optional
-            # explanation). Pass it through unchanged so the frontend
-            # `marked.parse` can render the code blocks and structure.
-            markdown = "\n".join(text_parts).strip()
-
             self.finish(json.dumps({
                 "ok": True,
-                "markdown": markdown,
                 "sessionId": sid,
                 "notebookPath": notebook_path,
             }))
         except OpenCodeError as e:
             # If session is invalid on OpenCode side, invalidate cache.
-            # 404 / 410 / "session not found" -> next get_or_create will recreate.
             if "404" in str(e) or "not found" in str(e).lower():
                 sm.invalidate(notebook_path)
                 log.warning("invalidated dead session for %s", notebook_path)
-            log.exception("edit failed")
+            log.exception("edit (async) failed")
             self.set_status(502)
             self.finish(json.dumps({"ok": False, "error": str(e)}))
         except Exception as e:
-            log.exception("edit failed")
+            log.exception("edit (async) failed")
+            self.set_status(502)
+            self.finish(json.dumps({"ok": False, "error": str(e)}))
+
+
+class GlobalEventHandler(APIHandler):
+    """SSE proxy for OpenCode Serve's ``GET /global/event``.
+
+    Streams every event from the upstream SSE feed to the client as
+    ``text/event-stream`` so EventSource / fetch+reader on the frontend
+    can consume them directly. The frontend filters by session ID —
+    we deliberately do NOT filter on the server because OpenCode's
+    event-shape has varied across versions and a too-eager server
+    filter can silently drop events the client would have accepted.
+
+    The connection is held open until the client disconnects, the upstream
+    closes, or an unrecoverable error occurs.
+    """
+
+    @tornado.web.authenticated
+    async def get(self):
+        # The ?session= query param is accepted for API compatibility
+        # but the server does NOT use it to filter — see docstring.
+        _ = self.get_query_argument("session", None)
+
+        self.set_header("Content-Type", "text/event-stream")
+        self.set_header("Cache-Control", "no-cache")
+        # Disable proxy buffering (e.g. nginx) so chunks reach the client
+        # immediately.
+        self.set_header("X-Accel-Buffering", "no")
+        # Prevent tornado from closing the connection on its own keep-alive
+        # timer — we hold it open as long as the upstream is alive.
+        self.set_header("Connection", "keep-alive")
+
+        client = make_client(self)
+        try:
+            async for event in client.stream_global_events():
+                payload = json.dumps(event, ensure_ascii=False)
+                self.write("data: %s\n\n" % payload)
+                # flush() pushes the chunk to the client; without it tornado
+                # would buffer until the response is "finished".
+                await self.flush()
+        except tornado.iostream.StreamClosedError:
+            # Client went away — normal disconnect, not an error.
+            log.info("SSE client disconnected")
+        except Exception as e:
+            log.exception("SSE proxy error")
+            try:
+                self.write("data: %s\n\n" % json.dumps({
+                    "type": "error",
+                    "message": str(e),
+                }))
+                await self.flush()
+            except Exception:
+                # Connection already gone; nothing to send.
+                pass
+
+
+class PermissionReplyHandler(APIHandler):
+    """Respond to a `permission.asked` event for a given session.
+
+    POST ``/opencode-bridge/permissions/:permId?session=:sid`` with body
+    ``{"response": "once"|"always"|"reject"}``. Forwards to OpenCode
+    Serve's ``POST /session/:sid/permissions/:permId``.
+    """
+
+    VALID_RESPONSES = ("once", "always", "reject")
+
+    @tornado.web.authenticated
+    async def post(self, perm_id: str):
+        session_id = self.get_query_argument("session", "")
+        if not session_id:
+            self.set_status(400)
+            self.finish(json.dumps({"error": "missing 'session' query parameter"}))
+            return
+        try:
+            body = json.loads(self.request.body) if self.request.body else {}
+        except json.JSONDecodeError as e:
+            self.set_status(400)
+            self.finish(json.dumps({"error": "bad request: %s" % e}))
+            return
+        response = body.get("response")
+        if response not in self.VALID_RESPONSES:
+            self.set_status(400)
+            self.finish(json.dumps({
+                "error": "response must be one of %s" % list(self.VALID_RESPONSES)
+            }))
+            return
+        client = make_client(self)
+        try:
+            ok = await client.reply_permission(session_id, perm_id, response)
+            self.finish(json.dumps({
+                "ok": ok,
+                "permissionId": perm_id,
+                "response": response,
+            }))
+        except OpenCodeError as e:
+            log.exception("permission reply failed")
+            self.set_status(502)
+            self.finish(json.dumps({"ok": False, "error": str(e)}))
+        except Exception as e:
+            log.exception("permission reply failed")
+            self.set_status(502)
+            self.finish(json.dumps({"ok": False, "error": str(e)}))
+
+
+class QuestionReplyHandler(APIHandler):
+    """Reply to a `question.asked` event with a freeform answer.
+
+    POST ``/opencode-bridge/questions/:qId/reply?session=:sid`` with body
+    ``{"answer": "..."}``. Forwards to OpenCode Serve's
+    ``POST /session/:sid/question/:qId/reply``.
+    """
+
+    @tornado.web.authenticated
+    async def post(self, question_id: str):
+        session_id = self.get_query_argument("session", "")
+        if not session_id:
+            self.set_status(400)
+            self.finish(json.dumps({"error": "missing 'session' query parameter"}))
+            return
+        try:
+            body = json.loads(self.request.body) if self.request.body else {}
+        except json.JSONDecodeError as e:
+            self.set_status(400)
+            self.finish(json.dumps({"error": "bad request: %s" % e}))
+            return
+        answer = body.get("answer")
+        if not isinstance(answer, str) or not answer.strip():
+            self.set_status(400)
+            self.finish(json.dumps({"error": "missing 'answer'"}))
+            return
+        client = make_client(self)
+        try:
+            ok = await client.reply_question(session_id, question_id, answer)
+            self.finish(json.dumps({
+                "ok": ok,
+                "questionId": question_id,
+            }))
+        except OpenCodeError as e:
+            log.exception("question reply failed")
+            self.set_status(502)
+            self.finish(json.dumps({"ok": False, "error": str(e)}))
+        except Exception as e:
+            log.exception("question reply failed")
             self.set_status(502)
             self.finish(json.dumps({"ok": False, "error": str(e)}))
-        # NO finally delete — session is reused per notebook.
 
 
 class SessionListHandler(APIHandler):
@@ -310,9 +467,21 @@ def setup_route_handlers(web_app):
         (url_path_join(base_url, "opencode-bridge", "health"), HealthHandler),
         (url_path_join(base_url, "opencode-bridge", "providers"), ProvidersHandler),
         (url_path_join(base_url, "opencode-bridge", "edit"), EditHandler),
+        (url_path_join(base_url, "opencode-bridge", "events"), GlobalEventHandler),
         (url_path_join(base_url, "opencode-bridge", "sessions"), SessionListHandler),
         (url_path_join(base_url, "opencode-bridge", "session"), SessionReleaseHandler),
         (url_path_join(base_url, "opencode-bridge", "session-messages"), SessionMessagesHandler),
+        # /permissions/<perm_id>  and  /questions/<q_id>/reply
+        # use a URLSpec with a regex so the path param is forwarded to
+        # the handler as a positional argument.
+        tornado.web.URLSpec(
+            url_path_join(base_url, "opencode-bridge", "permissions", r"([A-Za-z0-9_\-]+)"),
+            PermissionReplyHandler,
+        ),
+        tornado.web.URLSpec(
+            url_path_join(base_url, "opencode-bridge", "questions", r"([A-Za-z0-9_\-]+)", "reply"),
+            QuestionReplyHandler,
+        ),
     ]
 
     web_app.add_handlers(host_pattern, handlers)

opencode_bridge/tests/test_client.py

@@ -98,28 +98,30 @@ async def test_auth_absent_when_no_password(base_config: OpenCodeConfig) -> None
 
 
 @pytest.mark.asyncio
-async def test_send_message_sync_includes_model_when_provider_and_model_given(
+async def test_send_message_async_includes_model_when_provider_and_model_given(
     base_config: OpenCodeConfig,
 ) -> None:
+    """send_message_async posts to /prompt_async with provider/model when both set."""
     client, mock = _make_client(base_config, [(200, {"done": True})])
     parts = [{"type": "text", "text": "hello"}]
-    result = await client.send_message_sync("s1", parts, provider_id="p1", model_id="m1")
-    assert result == {"done": True}
+    result = await client.send_message_async("s1", parts, provider_id="p1", model_id="m1")
+    assert result is None
     call = mock.calls[0]
-    assert call["request"].url == "http://127.0.0.1:4096/session/s1/message"
+    assert call["request"].url == "http://127.0.0.1:4096/session/s1/prompt_async"
+    assert call["request"].method == "POST"
     body = json.loads(call["request"].body.decode("utf-8"))
     assert body["parts"] == parts
     assert body["model"] == {"providerID": "p1", "modelID": "m1"}
 
 
 @pytest.mark.asyncio
-async def test_send_message_sync_omits_model_when_provider_or_model_missing(
+async def test_send_message_async_omits_model_when_provider_or_model_missing(
     base_config: OpenCodeConfig,
 ) -> None:
     client, mock = _make_client(base_config, [(200, {"done": True})])
     parts = [{"type": "text", "text": "hello"}]
-    result = await client.send_message_sync("s1", parts)
-    assert result == {"done": True}
+    result = await client.send_message_async("s1", parts)
+    assert result is None
     call = mock.calls[0]
     body = json.loads(call["request"].body.decode("utf-8"))
     assert body["parts"] == parts
@@ -127,7 +129,7 @@ async def test_send_message_sync_omits_model_when_provider_or_model_missing(
 
 
 @pytest.mark.asyncio
-async def test_send_message_sync_includes_system_when_provided() -> None:
+async def test_send_message_async_includes_system_when_provided() -> None:
     """system param is forwarded into the request body when not None."""
     config = OpenCodeConfig(
         url="http://x:1", user="u", password="", request_timeout_seconds=120
@@ -136,7 +138,7 @@ async def test_send_message_sync_includes_system_when_provided() -> None:
     mock.responses.append((200, {"info": {}, "parts": []}))
     client = OpenCodeClient(config, http_client=mock)
 
-    await client.send_message_sync("sid", [{"type": "text", "text": "hi"}], system="be brief")
+    await client.send_message_async("sid", [{"type": "text", "text": "hi"}], system="be brief")
 
     assert len(mock.calls) == 1
     body = json.loads(mock.calls[0]["request"].body.decode("utf-8"))
@@ -144,6 +146,83 @@ async def test_send_message_sync_includes_system_when_provided() -> None:
     assert body["parts"] == [{"type": "text", "text": "hi"}]
 
 
+@pytest.mark.asyncio
+async def test_stream_global_events_yields_parsed_events(
+    base_config: OpenCodeConfig,
+) -> None:
+    """stream_global_events yields one dict per `data: {...}` SSE block."""
+
+    class _StreamingMock(MockHTTPClient):
+        async def fetch(self, request, **kwargs):
+            self.calls.append({"request": request, "kwargs": kwargs})
+            # Simulate OpenCode's /global/event: three events back-to-back.
+            chunks = [
+                b"data: {\"type\": \"session.next.text.delta\", \"properties\": "
+                b"{\"sessionID\": \"s1\", \"delta\": \"hello \"}}\n\n",
+                b"data: {\"type\": \"session.next.text.delta\", \"properties\": "
+                b"{\"sessionID\": \"s1\", \"delta\": \"world\"}}\n\n",
+                b"data: {\"type\": \"session.idle\", \"properties\": "
+                b"{\"sessionID\": \"s1\"}}\n\n",
+            ]
+            # streaming_callback now lives on the HTTPRequest (per tornado
+            # API — fetch() rejects kwargs when an HTTPRequest is passed).
+            cb = getattr(request, "streaming_callback", None)
+            if cb is not None:
+                for c in chunks:
+                    cb(c)
+            # Empty body, code 200 to terminate normally.
+            return tornado.httpclient.HTTPResponse(
+                request=request,
+                code=200,
+                headers=tornado.httputil.HTTPHeaders(),
+                buffer=BytesIO(b""),
+            )
+
+    mock = _StreamingMock()
+    client = OpenCodeClient(base_config, http_client=mock)
+
+    events = []
+    async for ev in client.stream_global_events():
+        events.append(ev)
+
+    assert len(events) == 3
+    assert events[0]["type"] == "session.next.text.delta"
+    assert events[0]["properties"]["delta"] == "hello "
+    assert events[1]["properties"]["delta"] == "world"
+    assert events[2]["type"] == "session.idle"
+    # Stream used the SSE accept header and no timeout.
+    call = mock.calls[0]
+    assert call["kwargs"] == {"raise_error": False}
+    assert call["request"].method == "GET"
+    assert call["request"].url == "http://127.0.0.1:4096/global/event"
+    assert call["request"].request_timeout == 0
+    assert call["request"].headers.get("Accept") == "text/event-stream"
+    assert call["request"].streaming_callback is not None
+
+
+def test_parse_sse_block_handles_concatenated_data_lines() -> None:
+    """Multi-line `data:` blocks are joined with \n (per SSE spec)."""
+    parsed = OpenCodeClient._parse_sse_block(
+        "data: {\"a\": 1,\ndata:  \"b\": 2}"
+    )
+    assert parsed == {"a": 1, "b": 2}
+
+
+def test_parse_sse_block_ignores_non_data_fields() -> None:
+    parsed = OpenCodeClient._parse_sse_block("event: ping\ndata: {\"k\": \"v\"}\n\n")
+    assert parsed == {"k": "v"}
+
+
+def test_parse_sse_block_returns_none_on_empty() -> None:
+    assert OpenCodeClient._parse_sse_block("") is None
+    assert OpenCodeClient._parse_sse_block("event: ping\n\n") is None
+    assert OpenCodeClient._parse_sse_block(": heartbeat\n\n") is None
+
+
+def test_parse_sse_block_returns_none_on_bad_json() -> None:
+    assert OpenCodeClient._parse_sse_block("data: not-json\n\n") is None
+
+
 @pytest.mark.asyncio
 async def test_delete_session_returns_bool_by_status(base_config: OpenCodeConfig) -> None:
     client, mock = _make_client(base_config, [(200, True), (404, False)])
@@ -154,6 +233,39 @@ async def test_delete_session_returns_bool_by_status(base_config: OpenCodeConfig
     assert mock.calls[0]["request"].url == "http://127.0.0.1:4096/session/s1"
 
 
+@pytest.mark.asyncio
+async def test_reply_permission_posts_response(base_config: OpenCodeConfig) -> None:
+    """reply_permission forwards the response string to /session/:id/permissions/:permId."""
+    client, mock = _make_client(base_config, [(200, True)])
+    ok = await client.reply_permission("s1", "perm-1", "once")
+    assert ok is True
+    call = mock.calls[0]
+    assert call["request"].url == "http://127.0.0.1:4096/session/s1/permissions/perm-1"
+    assert call["request"].method == "POST"
+    body = json.loads(call["request"].body.decode("utf-8"))
+    assert body == {"response": "once"}
+
+
+@pytest.mark.asyncio
+async def test_reply_permission_returns_false_on_404(
+    base_config: OpenCodeConfig,
+) -> None:
+    client, mock = _make_client(base_config, [(404, False)])
+    assert await client.reply_permission("s1", "perm-1", "reject") is False
+
+
+@pytest.mark.asyncio
+async def test_reply_question_posts_answer(base_config: OpenCodeConfig) -> None:
+    """reply_question forwards the answer to /session/:id/question/:qid/reply."""
+    client, mock = _make_client(base_config, [(200, True)])
+    ok = await client.reply_question("s1", "q-7", "use 3.12")
+    assert ok is True
+    call = mock.calls[0]
+    assert call["request"].url == "http://127.0.0.1:4096/session/s1/question/q-7/reply"
+    body = json.loads(call["request"].body.decode("utf-8"))
+    assert body == {"answer": "use 3.12"}
+
+
 @pytest.mark.asyncio
 async def test_non_2xx_response_raises_with_body(base_config: OpenCodeConfig) -> None:
     client, mock = _make_client(base_config, [(500, {"error": "boom"})])

opencode_bridge/tests/test_routes.py

@@ -12,10 +12,6 @@ class FakeOpenCodeClient:
         self.session_id = "fake-session-123"
         self.health_response = {"healthy": True, "version": "0.1.0"}
         self.providers_response = {"providers": [{"id": "anthropic", "models": [{"id": "claude"}]}]}
-        self.message_response = {
-            "info": {"id": "msg-1"},
-            "parts": [{"type": "text", "text": "def foo():\n    return 42\n"}],
-        }
         # Canned session-messages list response (list_session_messages).
         # Default: one user + one assistant message, mixed text parts.
         self.messages_response = [
@@ -45,9 +41,13 @@ class FakeOpenCodeClient:
         self.calls.append(("create_session", title))
         return {"id": self.session_id, "title": title}
 
-    async def send_message_sync(self, session_id, parts, provider_id=None, model_id=None, system=None):
-        self.calls.append(("send_message_sync", session_id, parts, system))
-        return self.message_response
+    async def send_message_async(
+        self, session_id, parts, provider_id=None, model_id=None, system=None
+    ):
+        self.calls.append(
+            ("send_message_async", session_id, parts, system, provider_id, model_id)
+        )
+        return None
 
     async def delete_session(self, session_id):
         self.calls.append(("delete_session", session_id))
@@ -57,6 +57,19 @@ class FakeOpenCodeClient:
         self.calls.append(("list_session_messages", session_id))
         return self.messages_response
 
+    async def reply_permission(self, session_id, permission_id, response):
+        self.calls.append(("reply_permission", session_id, permission_id, response))
+        return True
+
+    async def reply_question(self, session_id, question_id, answer):
+        self.calls.append(("reply_question", session_id, question_id, answer))
+        return True
+
+    async def stream_global_events(self):
+        """Async generator — tests can monkey-patch this to control the stream."""
+        if False:  # pragma: no cover - never executed, just makes this an asyncgen
+            yield {}
+
 
 class FakeSessionManager:
     """Drop-in replacement for SessionManager with recording."""
@@ -131,6 +144,12 @@ async def test_providers_handler(monkeypatch, jp_fetch):
 
 
 async def test_edit_handler(monkeypatch, jp_fetch):
+    """POST /opencode-bridge/edit fires prompt_async and returns sessionId only.
+
+    The async path returns immediately with `{ok, sessionId, notebookPath}` —
+    the actual LLM response is consumed via the /events SSE endpoint, not
+    embedded in this response.
+    """
     fake = FakeOpenCodeClient()
     monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
 
@@ -159,31 +178,68 @@ async def test_edit_handler(monkeypatch, jp_fetch):
     )
     assert response.code == 200
     payload = json.loads(response.body)
-    assert payload["ok"] is True
-    # Server now returns the AI reply as raw markdown (```fences``` kept
-    # so the frontend marked.parse renders code blocks).
-    assert payload["markdown"] == "def foo():\n    return 42"
-    assert payload["sessionId"] == "fake-session-123"
-    assert payload["notebookPath"] == "test.ipynb"
+    assert payload == {
+        "ok": True,
+        "sessionId": "fake-session-123",
+        "notebookPath": "test.ipynb",
+    }
+    # Async path: no `markdown` field. The LLM reply is consumed via SSE.
+    assert "markdown" not in payload
 
-    # Session manager was used, not direct create/delete on client
+    # Session manager was used (not direct create/delete on client).
     call_names = [c[0] for c in fake.calls]
     assert "create_session" not in call_names
     assert "delete_session" not in call_names
-    assert "send_message_sync" in call_names
+    assert "send_message_async" in call_names
+    assert "send_message_sync" not in call_names  # legacy sync path is gone
 
-    # send_message_sync received the session ID from the manager
-    send_call = [c for c in fake.calls if c[0] == "send_message_sync"][0]
+    # send_message_async received the session ID from the manager.
+    send_call = [c for c in fake.calls if c[0] == "send_message_async"][0]
     assert send_call[1] == "fake-session-123"
-    # system prompt was passed (v3+ allows markdown/fences in the reply)
+    # system prompt was passed.
     assert "你是一个代码助手" in send_call[3]
 
-    # SessionManager.get_or_create was called with the notebook path
+    # SessionManager.get_or_create was called with the notebook path.
     sm_call_names = [c[0] for c in fake_sm.calls]
     assert sm_call_names == ["get_or_create"]
     assert fake_sm.calls[0][1] == "test.ipynb"
 
 
+async def test_edit_handler_includes_model_when_provider_and_model_given(
+    monkeypatch, jp_fetch
+):
+    """providerId/modelId are forwarded into the async send body when both set."""
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+    monkeypatch.setattr(
+        "opencode_bridge.routes.get_session_manager",
+        lambda h: FakeSessionManager(),
+    )
+
+    body = json.dumps({
+        "prompt": "x",
+        "context": {
+            "notebookPath": "n.ipynb",
+            "cellId": "c",
+            "language": "python",
+            "cellIndex": 0,
+            "totalCells": 1,
+            "source": "",
+            "previousCode": None,
+            "error": None,
+        },
+        "providerId": "anthropic",
+        "modelId": "claude-sonnet-4-20250514",
+    })
+    response = await jp_fetch("opencode-bridge", "edit", method="POST", body=body)
+    assert response.code == 200
+
+    send_call = [c for c in fake.calls if c[0] == "send_message_async"][0]
+    # (send_message_async, sid, parts, system, provider_id, model_id)
+    assert send_call[4] == "anthropic"
+    assert send_call[5] == "claude-sonnet-4-20250514"
+
+
 async def test_session_list_handler(monkeypatch, jp_fetch):
     fake_sm = FakeSessionManager()
     fake_sm._sessions = {
@@ -287,3 +343,119 @@ async def test_session_release_handler(monkeypatch, jp_fetch):
     assert payload["notebookPath"] == "foo.ipynb"
     assert payload["deleted"] is True
     assert ("release", "foo.ipynb") in fake_sm.calls
+
+
+async def test_permission_reply_forwards_response(monkeypatch, jp_fetch):
+    """POST /opencode-bridge/permissions/:permId?session=:sid forwards to client."""
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+
+    response = await jp_fetch(
+        "opencode-bridge", "permissions", "perm-42",
+        method="POST",
+        params={"session": "ses-1"},
+        body=json.dumps({"response": "once"}),
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload == {
+        "ok": True,
+        "permissionId": "perm-42",
+        "response": "once",
+    }
+    assert ("reply_permission", "ses-1", "perm-42", "once") in fake.calls
+
+
+async def test_permission_reply_rejects_invalid_response(monkeypatch, jp_fetch):
+    """response must be one of once/always/reject — 400 otherwise."""
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+
+    with pytest.raises(tornado.httpclient.HTTPClientError) as exc_info:
+        await jp_fetch(
+            "opencode-bridge", "permissions", "perm-1",
+            method="POST",
+            params={"session": "ses-1"},
+            body=json.dumps({"response": "yes"}),
+        )
+    assert exc_info.value.code == 400
+    assert "yes" not in fake.calls and fake.calls == []
+
+
+async def test_permission_reply_requires_session(monkeypatch, jp_fetch):
+    """Missing ?session=... is a 400."""
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+    with pytest.raises(tornado.httpclient.HTTPClientError) as exc_info:
+        await jp_fetch(
+            "opencode-bridge", "permissions", "perm-1",
+            method="POST",
+            body=json.dumps({"response": "once"}),
+        )
+    assert exc_info.value.code == 400
+
+
+async def test_question_reply_forwards_answer(monkeypatch, jp_fetch):
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+
+    response = await jp_fetch(
+        "opencode-bridge", "questions", "q-7", "reply",
+        method="POST",
+        params={"session": "ses-2"},
+        body=json.dumps({"answer": "use Python 3.12"}),
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload == {"ok": True, "questionId": "q-7"}
+    assert ("reply_question", "ses-2", "q-7", "use Python 3.12") in fake.calls
+
+
+async def test_question_reply_rejects_empty_answer(monkeypatch, jp_fetch):
+    fake = FakeOpenCodeClient()
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+    with pytest.raises(tornado.httpclient.HTTPClientError) as exc_info:
+        await jp_fetch(
+            "opencode-bridge", "questions", "q-1", "reply",
+            method="POST",
+            params={"session": "ses-1"},
+            body=json.dumps({"answer": "   "}),
+        )
+    assert exc_info.value.code == 400
+    assert fake.calls == []
+
+
+async def test_global_event_handler_forwards_all_events_unfiltered(
+    monkeypatch, jp_fetch
+):
+    """Server-side SSE proxy must NOT filter by session — it forwards
+    everything so the client can decide. (Previously a too-eager server
+    filter silently dropped events the client would have accepted.)
+
+    We mock stream_global_events to push three events with different
+    sessionID shapes (one matching, one not, one missing). The proxy
+    should pass all three through as data: lines.
+    """
+    fake = FakeOpenCodeClient()
+
+    async def fake_stream():
+        yield {"type": "session.next.text.delta", "properties": {"sessionID": "ses-1", "delta": "A"}}
+        yield {"type": "session.next.text.delta", "properties": {"sessionID": "ses-OTHER", "delta": "B"}}
+        yield {"type": "server.connected", "properties": {}}
+
+    fake.stream_global_events = fake_stream
+    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
+
+    # Use ?session=ses-1 to confirm the param is accepted but doesn't filter.
+    response = await jp_fetch(
+        "opencode-bridge", "events",
+        params={"session": "ses-1"},
+    )
+    assert response.code == 200
+    body = response.body.decode("utf-8")
+    # All three events should appear in the body, regardless of sessionID.
+    assert '"delta": "A"' in body
+    assert '"delta": "B"' in body
+    assert '"server.connected"' in body
+    # The body should use the SSE format: each event as a `data:` line.
+    assert body.count("data: ") == 3