notebook-ai-extension/opencode_bridge/tests/test_routes.py

import json

import pytest
import tornado.httpclient


class FakeOpenCodeClient:
    """Drop-in replacement for OpenCodeClient with recording + canned responses."""

    def __init__(self):
        self.calls = []
        self.session_id = "fake-session-123"
        self.health_response = {"healthy": True, "version": "0.1.0"}
        self.providers_response = {"providers": [{"id": "anthropic", "models": [{"id": "claude"}]}]}
        # Canned session-messages list response (list_session_messages).
        # Default: one user + one assistant message, mixed text parts.
        self.messages_response = [
            {
                "info": {"role": "user", "id": "m1"},
                "parts": [{"type": "text", "text": "fix the bug"}],
            },
            {
                "info": {"role": "assistant", "id": "m2"},
                "parts": [{"type": "text", "text": "```python\nx = 1\n```"}],
            },
        ]

    @property
    def endpoint(self):
        return "http://fake-opencode"

    async def health(self):
        self.calls.append(("health",))
        return self.health_response

    async def list_providers(self):
        self.calls.append(("list_providers",))
        return self.providers_response

    async def create_session(self, title):
        self.calls.append(("create_session", title))
        return {"id": self.session_id, "title": title}

    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))
        return True

    async def list_session_messages(self, session_id):
        self.calls.append(("list_session_messages", session_id))
        return self.messages_response

    async def list_all_sessions(self):
        self.calls.append(("list_all_sessions",))
        return [
            {
                "id": "fake-session-123",
                "title": "jupyter:foo.ipynb",
                "createdAt": 1700000000000,
                "updatedAt": 1700000001000,
            },
            {
                "id": "other-session",
                "title": "manual session",
                "createdAt": 1700000010000,
                "updatedAt": 1700000011000,
            },
        ]

    async def create_session(self, title):
        self.calls.append(("create_session", title))
        # Return a fresh id (different from the default "fake-session-123").
        new_id = "newly-created-" + title
        return {"id": new_id, "title": title}

    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:
    """Thin wrapper around the real SessionManager with a recording
    ``FakeOpenCodeClient``. Lets route tests exercise the real
    multi-session logic without touching OpenCode.
    """

    def __init__(self, session_id: str = "fake-session-123", client=None) -> None:
        from opencode_bridge.session_manager import SessionManager
        # `_client` attribute used by some legacy tests. Allow injection
        # so a test can share its FakeOpenCodeClient with the SessionManager
        # (the real SessionManager calls into its own client_factory, NOT
        # the route's make_client).
        self._client = client if client is not None else FakeOpenCodeClient()
        self._sm = SessionManager(lambda: self._client)
        self.calls: list = self._client.calls
        # Legacy attribute used by list_sessions() etc.
        self._sessions: dict = {}

    # ---- delegation to the real SessionManager ----
    def _record(self, name, *args):
        self.calls.append((name, *args))
        return None

    def get_or_create(self, notebook_path: str):
        self._record("get_or_create", notebook_path)
        return self._sm.get_or_create(notebook_path)

    def peek(self, notebook_path: str):
        return self._sm.peek(notebook_path)

    def get_active(self, notebook_path: str):
        return self._sm.get_active(notebook_path)

    def list_sessions(self) -> list:
        return self._sm.list_sessions()

    def list_sessions_for_notebook(self, notebook_path: str):
        return self._sm.list_sessions_for_notebook(notebook_path)

    def bind_existing(self, notebook_path: str, session_id: str) -> None:
        self._record("bind_existing", notebook_path, session_id)
        self._sm.bind_existing(notebook_path, session_id)

    def set_active(self, notebook_path: str, session_id: str) -> bool:
        self._record("set_active", notebook_path, session_id)
        return self._sm.set_active(notebook_path, session_id)

    def unbind(self, notebook_path: str, session_id: str) -> bool:
        self._record("unbind", notebook_path, session_id)
        return self._sm.unbind(notebook_path, session_id)

    def unbind_active(self, notebook_path: str) -> bool:
        self._record("unbind_active", notebook_path)
        return self._sm.unbind_active(notebook_path)

    def invalidate(self, notebook_path: str) -> bool:
        self._record("invalidate", notebook_path)
        return self._sm.invalidate(notebook_path)

    async def release(self, notebook_path: str) -> bool:
        self._record("release", notebook_path)
        return await self._sm.release(notebook_path)

    async def delete_session(self, notebook_path: str, session_id: str) -> bool:
        self._record("delete_session", notebook_path, session_id)
        return await self._sm.delete_session(notebook_path, session_id)

    async def create_new(self, notebook_path: str, title: str | None = None) -> str:
        self._record("create_new", notebook_path, title)
        return await self._sm.create_new(notebook_path, title)

    # ---- internal-state proxies (used by tests that poke internals) ----
    @property
    def _active(self) -> dict:
        return self._sm._active

    @property
    def _bound(self) -> dict:
        return self._sm._bound


async def test_hello(jp_fetch):
    # When
    response = await jp_fetch("opencode-bridge", "hello")

    # Then
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload == {
            "data": (
                "Hello, world!"
                " This is the '/opencode-bridge/hello' endpoint."
                " Try visiting me in your browser!"
            ),
        }


async def test_health_handler(monkeypatch, jp_fetch):
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)

    response = await jp_fetch("opencode-bridge", "health")
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["ok"] is True
    assert payload["version"] == "0.1.0"
    assert ("health",) in fake.calls


async def test_providers_handler(monkeypatch, jp_fetch):
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)

    response = await jp_fetch("opencode-bridge", "providers")
    assert response.code == 200
    payload = json.loads(response.body)
    assert "providers" in payload
    assert payload["providers"][0]["id"] == "anthropic"
    assert ("list_providers",) in fake.calls


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)

    fake_sm = FakeSessionManager(session_id="fake-session-123")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: fake_sm
    )

    body = json.dumps({
        "prompt": "Add type hints",
        "context": {
            "notebookPath": "test.ipynb",
            "cellId": "cell-1",
            "language": "python",
            "cellIndex": 0,
            "totalCells": 1,
            "source": "def foo(): return 42\n",
            "previousCode": None,
            "error": None,
        },
    })
    response = await jp_fetch(
        "opencode-bridge", "edit",
        method="POST",
        body=body,
    )
    assert response.code == 200
    payload = json.loads(response.body)
    # The exact sessionId comes from the FakeClient; just check the
    # shape — keys are present and ok is true.
    assert payload["ok"] is True
    assert isinstance(payload["sessionId"], str) and payload["sessionId"]
    assert payload["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).
    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_async" in call_names
    assert "send_message_sync" not in call_names  # legacy sync path is gone

    # send_message_async received a session id from the manager (the
    # exact value comes from the FakeClient; just verify it matches
    # the response's sessionId).
    send_call = [c for c in fake.calls if c[0] == "send_message_async"][0]
    assert send_call[1] == payload["sessionId"]
    # system prompt was passed.
    assert "你是一个代码助手" in send_call[3]

    # SessionManager.get_or_create was called with the notebook path.
    # (The real SessionManager internally calls create_session on the
    # FakeClient too, so the shared calls list has both entries.)
    sm_call_names = [c[0] for c in fake_sm.calls]
    assert "get_or_create" in sm_call_names
    get_or_create_call = [c for c in fake_sm.calls if c[0] == "get_or_create"][0]
    assert get_or_create_call[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()
    # Seed the underlying real SessionManager's state so list_sessions
    # returns the expected debug mapping.
    fake_sm._active["foo.ipynb"] = "sid-1"
    fake_sm._active["bar.ipynb"] = "sid-2"
    fake_sm._bound["foo.ipynb"] = ["sid-1"]
    fake_sm._bound["bar.ipynb"] = ["sid-2"]
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: fake_sm
    )

    response = await jp_fetch("opencode-bridge", "sessions")
    assert response.code == 200
    payload = json.loads(response.body)
    assert "sessions" in payload
    paths = {s["notebookPath"] for s in payload["sessions"]}
    assert paths == {"foo.ipynb", "bar.ipynb"}


async def test_session_messages_handler_returns_empty_when_no_session(
    monkeypatch, jp_fetch
) -> None:
    # No session registered for this notebook -> handler returns
    # {"messages": []} WITHOUT calling OpenCodeClient (peek short-circuits).
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)

    fake_sm = FakeSessionManager(session_id="")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: fake_sm
    )

    response = await jp_fetch(
        "opencode-bridge", "session-messages",
        method="GET",
        params={"notebook": "fresh.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload == {"messages": []}
    # No OpenCode call was made.
    assert all(c[0] != "list_session_messages" for c in fake.calls)


async def test_session_messages_handler_projects_opencode_messages(
    monkeypatch, jp_fetch
) -> None:
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)

    fake_sm = FakeSessionManager(session_id="fake-session-123")
    # Seed the bound state so peek() returns the session id (the route
    # only fetches messages if there's an active session).
    fake_sm._active["test.ipynb"] = "fake-session-123"
    fake_sm._bound["test.ipynb"] = ["fake-session-123"]
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: fake_sm
    )

    response = await jp_fetch(
        "opencode-bridge", "session-messages",
        method="GET",
        params={"notebook": "test.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload == {
        "messages": [
            {"role": "user", "content": "fix the bug"},
            {"role": "assistant", "content": "```python\nx = 1\n```"},
        ]
    }
    # The OpenCode client was called with the session id from the manager.
    assert ("list_session_messages", "fake-session-123") in fake.calls


async def test_session_messages_handler_requires_notebook(
    monkeypatch, jp_fetch
) -> None:
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    # jp_fetch raises HTTPClientError on 4xx; assert the handler 400s
    # (the body would contain "missing 'notebook' query parameter").
    with pytest.raises(tornado.httpclient.HTTPClientError) as exc_info:
        await jp_fetch("opencode-bridge", "session-messages", method="GET")
    assert exc_info.value.code == 400


async def test_session_release_handler(monkeypatch, jp_fetch):
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)

    fake_sm = FakeSessionManager()
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: fake_sm
    )

    # Set up state: get_or_create actually binds a session to the
    # notebook in the real SessionManager (via the FakeClient). release
    # then has something to delete.
    await fake_sm.get_or_create("foo.ipynb")

    response = await jp_fetch(
        "opencode-bridge", "session",
        method="DELETE",
        params={"notebook": "foo.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["ok"] is True
    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


# ---------------------------------------------------------------------------
# Multi-session management endpoints
# ---------------------------------------------------------------------------


async def test_all_sessions_handler_returns_global_list(monkeypatch, jp_fetch):
    """GET /sessions/all proxies OpenCode Serve's GET /session."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    response = await jp_fetch("opencode-bridge", "sessions", "all")
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["ok"] is True
    assert len(payload["sessions"]) == 2
    # The fake returns both the default session and a hand-rolled "other-session".
    ids = {s["id"] for s in payload["sessions"]}
    assert "fake-session-123" in ids
    assert "other-session" in ids
    assert ("list_all_sessions",) in fake.calls


async def test_notebook_sessions_get_returns_bound_with_metadata(
    monkeypatch, jp_fetch
):
    """GET /sessions/notebook?notebook=... returns the bound sessions
    enriched with title/createdAt from the global list."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    sm = FakeSessionManager()
    sm._active["foo.ipynb"] = "fake-session-123"
    sm._bound["foo.ipynb"] = ["fake-session-123"]
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )

    response = await jp_fetch(
        "opencode-bridge", "sessions", "notebook",
        params={"notebook": "foo.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["notebookPath"] == "foo.ipynb"
    assert payload["activeSessionId"] == "fake-session-123"
    assert len(payload["sessions"]) == 1
    s = payload["sessions"][0]
    assert s["sessionId"] == "fake-session-123"
    assert s["title"] == "jupyter:foo.ipynb"
    assert s["isActive"] is True


async def test_notebook_sessions_post_creates_and_binds(monkeypatch, jp_fetch):
    """POST /sessions/notebook creates a new session on OpenCode and
    binds it as active."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    # Share the client with the SessionManager so the test can assert
    # against `fake.calls` (the real SessionManager calls its own
    # client_factory, NOT the route's make_client).
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager",
        lambda h: FakeSessionManager(client=fake),
    )

    response = await jp_fetch(
        "opencode-bridge", "sessions", "notebook",
        method="POST",
        params={"notebook": "foo.ipynb", "title": "fresh"},
        body=json.dumps({}),
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["ok"] is True
    assert payload["sessionId"] == "newly-created-fresh"
    # The new session is the active one.
    assert payload["activeSessionId"] == "newly-created-fresh"
    # OpenCode was called.
    assert ("create_session", "fresh") in fake.calls


async def test_notebook_sessions_put_binds_existing(monkeypatch, jp_fetch):
    """PUT /sessions/notebook binds an existing sessionId without
    calling OpenCode. Sets it as active."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    sm = FakeSessionManager()
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )

    response = await jp_fetch(
        "opencode-bridge", "sessions", "notebook",
        method="PUT",
        params={"notebook": "foo.ipynb", "sessionId": "some-existing-sid"},
        body=json.dumps({}),
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["sessionId"] == "some-existing-sid"
    assert payload["activeSessionId"] == "some-existing-sid"
    # No OpenCode round-trip for the bind.
    assert ("create_session", ...) not in [
        c for c in fake.calls if c[0] == "create_session"
    ]


async def test_active_session_set_requires_bound(monkeypatch, jp_fetch):
    """PUT /sessions/active?sessionId=X fails with 400 if X is not bound."""
    monkeypatch.setattr(
        "opencode_bridge.routes.make_client", lambda h: FakeOpenCodeClient()
    )
    sm = FakeSessionManager()
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )
    with pytest.raises(tornado.httpclient.HTTPClientError) as exc_info:
        await jp_fetch(
            "opencode-bridge", "sessions", "active",
            method="PUT",
            params={"notebook": "foo.ipynb", "sessionId": "never-bound"},
            body=json.dumps({}),
        )
    assert exc_info.value.code == 400


async def test_active_session_set_switches(monkeypatch, jp_fetch):
    """PUT /sessions/active successfully switches when the sessionId is bound."""
    monkeypatch.setattr(
        "opencode_bridge.routes.make_client", lambda h: FakeOpenCodeClient()
    )
    sm = FakeSessionManager()
    sm.bind_existing("foo.ipynb", "sid-A")
    sm.bind_existing("foo.ipynb", "sid-B")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )
    response = await jp_fetch(
        "opencode-bridge", "sessions", "active",
        method="PUT",
        params={"notebook": "foo.ipynb", "sessionId": "sid-A"},
        body=json.dumps({}),
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["activeSessionId"] == "sid-A"


async def test_active_session_delete_unbinds_only(monkeypatch, jp_fetch):
    """DELETE /sessions/active unbinds the active session; other
    bound sessions are kept; OpenCode is NOT called."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    sm = FakeSessionManager()
    sm.bind_existing("foo.ipynb", "sid-A")
    sm.bind_existing("foo.ipynb", "sid-B")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )
    response = await jp_fetch(
        "opencode-bridge", "sessions", "active",
        method="DELETE",
        params={"notebook": "foo.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["activeSessionId"] is None
    # sid-A is still bound; sid-B was the active and is gone.
    assert sm.list_sessions_for_notebook("foo.ipynb") == ["sid-A"]
    # No OpenCode call.
    assert not any(c[0] == "delete_session" for c in fake.calls)


async def test_active_session_get_returns_id_or_null(monkeypatch, jp_fetch):
    monkeypatch.setattr(
        "opencode_bridge.routes.make_client", lambda h: FakeOpenCodeClient()
    )
    sm = FakeSessionManager()
    sm.bind_existing("foo.ipynb", "sid-A")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )
    response = await jp_fetch(
        "opencode-bridge", "sessions", "active",
        params={"notebook": "foo.ipynb"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["activeSessionId"] == "sid-A"


async def test_notebook_sessions_delete_removes_session(monkeypatch, jp_fetch):
    """DELETE /sessions/notebook?notebook=...&sessionId=... deletes
    the session on OpenCode AND removes its binding."""
    fake = FakeOpenCodeClient()
    monkeypatch.setattr("opencode_bridge.routes.make_client", lambda h: fake)
    # Share the client with the SessionManager so the test can assert
    # against `fake.calls` (the real SessionManager calls its own
    # client_factory, NOT the route's make_client).
    sm = FakeSessionManager(client=fake)
    sm.bind_existing("foo.ipynb", "sid-A")
    sm.bind_existing("foo.ipynb", "sid-B")
    monkeypatch.setattr(
        "opencode_bridge.routes.get_session_manager", lambda h: sm
    )
    response = await jp_fetch(
        "opencode-bridge", "sessions", "notebook",
        method="DELETE",
        params={"notebook": "foo.ipynb", "sessionId": "sid-A"},
    )
    assert response.code == 200
    payload = json.loads(response.body)
    assert payload["deleted"] is True
    # OpenCode delete was called for sid-A.
    assert ("delete_session", "sid-A") in fake.calls
    # sid-A is gone; sid-B remains active.
    assert sm.list_sessions_for_notebook("foo.ipynb") == ["sid-B"]
    assert sm.get_active("foo.ipynb") == "sid-B"