diff --git a/clients/terminal/api.py b/clients/terminal/api.py index 311bd5c..741b130 100644 --- a/clients/terminal/api.py +++ b/clients/terminal/api.py @@ -98,4 +98,25 @@ client = _client() resp = await client.post(f"/sessions/{session_id}/stop") resp.raise_for_status() + return resp.json() + + +async def list_terminals(session_id: str) -> list[dict]: + """List the session's active persistent terminals (background dev servers, + watchers). Seeds the TUI terminals modal + status-bar count on attach.""" + client = _client() + resp = await client.get(f"/sessions/{session_id}/terminals") + resp.raise_for_status() + return resp.json().get("terminals", []) + + +async def close_terminal(session_id: str, terminal_name: str) -> dict: + """Force-close one persistent terminal by name (from the /terminals modal).""" + from urllib.parse import quote + + client = _client() + resp = await client.post( + f"/sessions/{session_id}/terminals/{quote(terminal_name, safe='')}/close" + ) + resp.raise_for_status() return resp.json() \ No newline at end of file diff --git a/navi/api/routes/sessions.py b/navi/api/routes/sessions.py index 2c65e32..40348df 100644 --- a/navi/api/routes/sessions.py +++ b/navi/api/routes/sessions.py @@ -246,6 +246,50 @@ return {"session_id": session_id, "tasks": tasks} +@router.get("/{session_id}/terminals") +async def list_session_terminals( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """List the session's active persistent terminals (background dev servers, + watchers, …). Seeds the TUI terminals modal + status-bar count on attach.""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + from navi.api.deps import get_orchestrator + + orchestrator = get_orchestrator() + terminals: list[dict] = [] + if orchestrator and orchestrator._container and orchestrator._container.terminal_manager: + terminals = orchestrator._container.terminal_manager.list(session_id) + return {"session_id": session_id, "terminals": terminals} + + +@router.post("/{session_id}/terminals/{terminal_name}/close") +async def close_session_terminal( + session_id: str, + terminal_name: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Force-close one persistent terminal by name (from the TUI /terminals modal).""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + from navi.api.deps import get_orchestrator + + orchestrator = get_orchestrator() + if not (orchestrator and orchestrator._container and orchestrator._container.terminal_manager): + return {"session_id": session_id, "terminal_name": terminal_name, "closed": False} + closed = await orchestrator._container.terminal_manager.close(session_id, terminal_name) + return {"session_id": session_id, "terminal_name": terminal_name, "closed": closed} + + @router.patch("/{session_id}/pin") async def pin_session( session_id: str, diff --git a/tests/integration/test_api_routes.py b/tests/integration/test_api_routes.py index b882377..30dbabd 100644 --- a/tests/integration/test_api_routes.py +++ b/tests/integration/test_api_routes.py @@ -159,6 +159,72 @@ response = client.get("/sessions/nonexistent/todos") assert response.status_code == 404 + # ── terminals ──────────────────────────────────────────────────────────── + + class _FakeTm: + def __init__(self, items=None, closed=True): + self._items = items or [] + self.closed_calls: list = [] + self._closed_result = closed + + def list(self, session_id): + return self._items + + async def close(self, session_id, name): + self.closed_calls.append((session_id, name)) + return self._closed_result + + @pytest.mark.anyio + async def test_list_terminals_empty(self, client, make_session): + session = await make_session("secretary") + response = client.get(f"/sessions/{session.id}/terminals") + assert response.status_code == 200 + data = response.json() + assert data["session_id"] == session.id + assert data["terminals"] == [] + + @pytest.mark.anyio + async def test_list_terminals_returns_active(self, client, make_session): + session = await make_session("secretary") + from navi.api.deps import _resolve_container + + container = _resolve_container() + fake = self._FakeTm([ + {"name": "dev", "description": "dev server", "status": "busy", + "pid": 123, "uptime_seconds": 5}, + ]) + container.terminal_manager = fake + try: + response = client.get(f"/sessions/{session.id}/terminals") + finally: + container.terminal_manager = None + assert response.status_code == 200 + terminals = response.json()["terminals"] + assert len(terminals) == 1 + assert terminals[0]["name"] == "dev" + + @pytest.mark.anyio + async def test_close_terminal(self, client, make_session): + session = await make_session("secretary") + from navi.api.deps import _resolve_container + + container = _resolve_container() + fake = self._FakeTm(closed=True) + container.terminal_manager = fake + try: + response = client.post(f"/sessions/{session.id}/terminals/dev/close") + finally: + container.terminal_manager = None + assert response.status_code == 200 + data = response.json() + assert data["terminal_name"] == "dev" + assert data["closed"] is True + assert fake.closed_calls == [(session.id, "dev")] + + def test_list_terminals_not_found(self, client): + response = client.get("/sessions/nonexistent/terminals") + assert response.status_code == 404 + @pytest.mark.anyio async def test_pin_session(self, client, make_session): session = await make_session("secretary")