NetDevAutomate
diff --git a/‎docs/plans/2026-05-09-refactor-agent-session-transport-plan.md‎
Lines changed: 88 additions & 0 deletions b/‎docs/plans/2026-05-09-refactor-agent-session-transport-plan.md‎
Lines changed: 88 additions & 0 deletions
diff --git a/‎packages/studyloop/src/studyloop/web/routes/session.py‎
Lines changed: 154 additions & 63 deletions b/‎packages/studyloop/src/studyloop/web/routes/session.py‎
Lines changed: 154 additions & 63 deletions
@@ -120,6 +120,94 @@ earlier. Ruff clean, format clean.
   now because `test_session_runtime.py` + the production
   `/session/ws` route still depend on it.
 
+## Amendment #5 (2026-05-10, session 3) — §1.5 FastAPI wiring landed
+
+`/api/session/ws` now runs the new transport stack. The route no longer
+reads `app.state.agent_session_manager`; it binds the already-acquired
+`active.ActiveSession` to the WebSocket and pumps transport events both
+ways. Legacy `session_runtime/` stays on disk (its unit tests in
+`test_session_runtime.py` still run) but is no longer in the request
+path for the web UI.
+
+**What landed**
+
+| PR | Commit | What landed |
+|---|---|---|
+| PR-B-5 | (this session) | `web/routes/session.py` — `live_session_socket` rewritten. Pre-accept Origin guard (plan Blocker B1) + `?study_session_id=` match + `TaskGroup` + `except*` pumps + binary `OutputBytes` out, text JSON for lifecycle events. 8 new tests in `test_web_session_ws.py`. Two stale legacy-protocol tests removed from `test_web_live_session.py`. |
+
+**Route contract (authoritative for frontend §1.7)**
+
+- **Preconditions**: caller must have already obtained an active
+  session (via `POST /api/session/start` — still driving the legacy
+  path today; PTY-acquire from REST is §1.5b follow-up) and know its
+  `study_session_id`. Without an active session, the WS closes 1008
+  before `accept()`.
+- **URL**: `/api/session/ws?study_session_id=<id>`. If the query param
+  does not match `active.current().study_session_id`, close 1008.
+- **Origin allow-list**: localhost/127.0.0.1 any port by default.
+  Operators behind a reverse proxy can extend via the
+  `STUDYLOOP_ALLOWED_ORIGINS` env var (comma-separated).
+- **Inbound frames** (JSON text):
+  - `{"type": "input", "data": "<string>"}` → `transport.send_input(data.encode("utf-8"))`
+  - `{"type": "resize", "cols": N, "rows": N}` → `transport.resize(cols, rows)`
+  - `{"type": "stop"}` → `transport.cancel()` then return
+  - Unknown types are silently dropped (no error channel).
+- **Outbound frames**:
+  - `OutputBytes.data` → **binary** frame, verbatim bytes
+  - `Started` → `{"type": "started", "agent": <name>}`
+  - `Stopped` → `{"type": "stopped", "returncode": N|null, "reason": <str>}` (terminal — pump stops)
+  - `TransportError` → `{"type": "transport_error", "message": <str>}`
+  - `AgentMessage` → `{"type": "agent_message", "kind": <str>, "payload": <obj>}`
+- **Shutdown**: `finally: await active.release()`. Always runs — closes the
+  transport and clears `session_state.json` regardless of which pump
+  raised. Guarantees a subsequent `acquire()` will not hit
+  `SessionAlreadyActiveError` on a clean disconnect.
+
+**Deliberate scope cuts for this PR**
+
+- **`POST /api/session/start` not yet re-pointed at `active.acquire()`**
+  — it still runs the legacy tmux+ttyd flow. Switching it requires
+  deciding how/when to call `active.acquire(config, lambda: PTYTransport(...))`
+  with a resolved `persona_file` + `build_launch_cmd` wiring. User
+  deferred this to "needs a lot of Playwright testing and iteration"
+  and marked it as the §1.5b follow-up. Today the route is usable via
+  tests only; the production flow still needs the REST start hook.
+- **Feature flag `STUDYLOOP_TRANSPORT=ttyd` / `?transport=ttyd`** —
+  the WS route no longer serves ttyd at all; the iframe path is
+  reached via `/terminal/` in `terminal_proxy.py`. The *WS-level*
+  feature flag described in plan §1.9 becomes a frontend concern
+  ("load iframe vs open WS") rather than a server concern. Rewording
+  §1.9 for this reality is part of §1.6–1.9 when the frontend lands.
+- **ACP branch removed** from the WS protocol — the legacy route had
+  a `{type: "start", transport: "acp"}` stub that returned an error.
+  The new route has no `start` frame; there's nothing to return an
+  error from. ACP support will route through its own `ACPTransport`
+  when Phase 2 lands.
+
+**Test-suite state after Amendment #5**
+
+1980 passed, 145 deselected, 0 failures. +8 new tests
+(`test_web_session_ws.py`): origin guard, session-id mismatch, no-
+active-session rejection, Started/Output/Stopped pump, input / resize /
+stop inbound frames, release-on-disconnect. −2 stale tests
+(`test_live_session_websocket_streams_events_and_accepts_input`,
+`test_live_session_websocket_rejects_acp_until_handshake_is_implemented`).
+Net +6 vs. the 1974 at Amendment #4. Ruff clean, format clean.
+
+### Open follow-ups after Amendment #5
+
+- **§1.5b REST→PTY bridge**: `POST /api/session/start` with
+  `transport=pty` should call `active.acquire(config, PTYTransport)`
+  (skip tmux/ttyd, do persona setup inline, return the `study_session_id`
+  for the WS query param). Separate PR so frontend iteration can drive
+  the shape via Playwright. Biggest remaining plan-§1.5 item.
+- **§1.6–1.9 xterm.js frontend** — now unblocked at the server side;
+  browser can `new WebSocket("/api/session/ws?study_session_id=…")`
+  and speak the contract above.
+- **Legacy `session_runtime/` removal** — viable once §1.5b also
+  lands (the WS route no longer references it; only
+  `test_session_runtime.py` does). Follow-up chore.
+
 ## Enhancement Summary (Deepened 2026-05-09)
 
 Plan deepened with nine parallel research agents (framework docs, best
 
@@ -576,79 +576,170 @@ def get_session_options() -> dict[str, list[dict]]:
     }
 
 
+_WS_CLOSE_POLICY = 1008  # RFC 6455 Policy Violation
+
+
+def _allowed_ws_origins() -> set[str]:
+    """Return the allow-list for Origin on the live session WS.
+
+    Localhost only by default — this is a single-user tool. Expand via
+    ``STUDYLOOP_ALLOWED_ORIGINS`` (comma-separated) if the operator fronts
+    the app behind a reverse proxy. See plan Blocker B1 (CSRF via cross-
+    origin WS upgrade; RFC 6455 has no same-origin enforcement and CORS
+    does not apply to WS).
+    """
+    import os
+
+    defaults = {
+        "http://127.0.0.1:8788",
+        "http://localhost:8788",
+        "http://127.0.0.1",
+        "http://localhost",
+    }
+    extra = os.environ.get("STUDYLOOP_ALLOWED_ORIGINS", "").strip()
+    if extra:
+        return defaults | {o.strip() for o in extra.split(",") if o.strip()}
+    return defaults
+
+
+def _origin_allowed(origin: str) -> bool:
+    if not origin:
+        return False
+    allowed = _allowed_ws_origins()
+    if origin in allowed:
+        return True
+    # Allow any port on localhost / 127.0.0.1 (dev server port may vary).
+    for prefix in ("http://127.0.0.1", "http://localhost"):
+        if origin.startswith(prefix + ":") or origin == prefix:
+            return True
+    return False
+
+
 @router.websocket("/session/ws")
 async def live_session_socket(websocket: WebSocket) -> None:
-    """Bidirectional live agent session socket.
-
-    Client messages:
-    - {"type": "start", "topic": "...", "energy": 5, "agent": "codex", "transport": "pty"}
-    - {"type": "input", "text": "..."}
-    - {"type": "stop"}
+    """Bidirectional live agent session socket (plan §1.5).
+
+    The route binds an already-acquired ``active.ActiveSession`` to the
+    WS client. Sessions are acquired via ``POST /api/session/start``
+    (which calls ``active.acquire(config, PTYTransport)``); the WS then
+    streams transport events out and pumps control frames in.
+
+    Inbound JSON control frames:
+    - ``{"type": "input", "data": "..."}``   → ``transport.send_input``
+    - ``{"type": "resize", "cols": N, "rows": N}`` → ``transport.resize``
+    - ``{"type": "stop"}``                    → ``transport.cancel``
+
+    Outbound framing:
+    - ``OutputBytes.data`` → **binary** frame (verbatim PTY bytes)
+    - ``Started``/``Stopped``/``TransportError``/``AgentMessage`` → text
+      JSON frames (``{"type": ..., ...}``)
+
+    Close codes:
+    - 1008 (Policy Violation) if Origin is disallowed, if no session is
+      active, or if ``?study_session_id`` does not match the active one.
+    - 1000 normal close on ``stop`` frame or transport-emitted ``Stopped``.
     """
-    await websocket.accept()
-    manager = websocket.app.state.agent_session_manager
-    session_id: str | None = None
-    event_task: asyncio.Task[None] | None = None
+    from studyloop.session import active as session_active
+    from studyloop.session.transport import (
+        AgentMessage,
+        OutputBytes,
+        Started,
+        Stopped,
+        TransportError,
+    )
 
-    async def forward_events(events) -> None:  # type: ignore[no-untyped-def]
-        async for event in events:
-            await websocket.send_json(event.as_dict())
+    # --- Pre-accept guards -----------------------------------------------
 
-    try:
-        while True:
-            message = await websocket.receive_json()
-            message_type = message.get("type")
-            if message_type == "start":
-                requested_transport = str(message.get("transport") or "pty")
-                if requested_transport == "acp":
-                    await websocket.send_json(
-                        {
-                            "type": "error",
-                            "data": {
-                                "message": (
-                                    "ACP transport is scaffolded but not enabled yet. "
-                                    "Use Browser terminal for Gemini or Kiro."
-                                )
-                            },
-                        }
-                    )
-                    continue
-                if session_id is not None:
-                    await websocket.send_json(
-                        {"type": "error", "data": {"message": "Session already started"}}
-                    )
-                    continue
-                session_id, events = await manager.start_session(
-                    topic=str(message.get("topic") or "Study Session"),
-                    energy=int(message.get("energy") or 5),
-                    agent=message.get("agent"),
-                    transport=requested_transport,
+    # Origin check (plan Blocker B1). Must happen before accept() — the
+    # handshake has not yet completed, so close-without-accept sends an
+    # HTTP 403 response per Starlette semantics, which the client sees as
+    # a failed upgrade. After accept() we can only send a WS close frame.
+    origin = websocket.headers.get("origin", "")
+    if not _origin_allowed(origin):
+        logger.warning("WS /session/ws rejected: disallowed origin=%r", origin)
+        await websocket.close(code=_WS_CLOSE_POLICY)
+        return
+
+    requested = websocket.query_params.get("study_session_id")
+    current = await session_active.current()
+    if current is None:
+        await websocket.close(code=_WS_CLOSE_POLICY)
+        return
+    if requested and requested != current.study_session_id:
+        logger.warning(
+            "WS /session/ws rejected: requested=%r active=%r",
+            requested,
+            current.study_session_id,
+        )
+        await websocket.close(code=_WS_CLOSE_POLICY)
+        return
+
+    await websocket.accept()
+    transport = current.transport
+
+    async def pty_to_ws() -> None:
+        """Pump transport events → WS frames until the session ends."""
+        async for event in transport.events():
+            if isinstance(event, OutputBytes):
+                await websocket.send_bytes(event.data)
+            elif isinstance(event, Started):
+                await websocket.send_json({"type": "started", "agent": event.agent})
+            elif isinstance(event, Stopped):
+                await websocket.send_json(
+                    {
+                        "type": "stopped",
+                        "returncode": event.returncode,
+                        "reason": event.reason,
+                    }
                 )
-                event_task = asyncio.create_task(forward_events(events))
-            elif message_type == "input":
-                if session_id is None:
-                    await websocket.send_json(
-                        {"type": "error", "data": {"message": "No active session"}}
-                    )
-                    continue
-                await manager.send(session_id, str(message.get("text") or ""))
-            elif message_type == "stop":
-                if session_id:
-                    await manager.stop(session_id)
-                    session_id = None
-                await websocket.close()
-                return
-            else:
+                return  # Stopped is terminal — stop pumping.
+            elif isinstance(event, TransportError):
+                await websocket.send_json({"type": "transport_error", "message": event.message})
+            elif isinstance(event, AgentMessage):
                 await websocket.send_json(
-                    {"type": "error", "data": {"message": "Unknown message type"}}
+                    {"type": "agent_message", "kind": event.kind, "payload": event.payload}
                 )
-    except WebSocketDisconnect:
+
+    async def ws_to_pty() -> None:
+        """Read WS control frames and forward to transport."""
+        while True:
+            frame = await websocket.receive_json()
+            ftype = frame.get("type")
+            if ftype == "input":
+                data = frame.get("data", "")
+                if isinstance(data, str):
+                    await transport.send_input(data.encode("utf-8"))
+            elif ftype == "resize":
+                try:
+                    cols = int(frame.get("cols", 80))
+                    rows = int(frame.get("rows", 24))
+                except (TypeError, ValueError):
+                    continue
+                await transport.resize(cols, rows)
+            elif ftype == "stop":
+                await transport.cancel()
+                return
+            # Silently drop unknown frame types — no error channel needed.
+
+    # --- Pump with TaskGroup (plan Blocker B5) ---------------------------
+    #
+    # TaskGroup raises ExceptionGroup on any child exception; the WS
+    # disconnect paths come up as ExceptionGroup[WebSocketDisconnect,
+    # ConnectionClosedOK, ...]. ``except*`` unpacks cleanly without
+    # reaching for .exceptions.
+
+    try:
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(pty_to_ws(), name="ws-pty-to-ws")
+            tg.create_task(ws_to_pty(), name="ws-ws-to-pty")
+    except* WebSocketDisconnect:
         pass
+    except* OSError as eg:
+        logger.error("PTY I/O error on /session/ws: %s", eg.exceptions)
+    except* Exception as eg:  # pragma: no cover — defensive
+        logger.exception("unexpected error on /session/ws: %s", eg.exceptions)
     finally:
-        if event_task:
-            event_task.cancel()
-        if session_id:
-            await manager.stop(session_id)
+        await session_active.release()
 
 
 @router.post("/session/end")