feat: add terminal session management with HTTP/WebSocket routes and session history

tests/test_sessions.py

@@ -7,12 +7,13 @@ import tempfile
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import patch
 
 from nest.server import create_app
 from nest.session_store import SessionStore
 from nest.sessions import SessionManager
 from tests.test_server import FASTAPI_AVAILABLE, fake_application
-from uvm.errors import ConflictError, GuestAgentUnavailableError, NotFoundError
+from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError
 
 
 if FASTAPI_AVAILABLE:
@@ -29,7 +30,10 @@ class FakeTerminal:
         self.closed = False
 
     async def read(self):
-        return await self.events.get()
+        event = await self.events.get()
+        if isinstance(event, Exception):
+            raise event
+        return event
 
     async def write(self, data):
         self.input.append(data)
@@ -176,6 +180,45 @@ class SessionManagerTests(unittest.IsolatedAsyncioTestCase):
         await self.manager.shutdown()
         self.assertTrue(self.application.terminals[0].closed)
 
+    async def test_guest_loss_ends_the_session_and_retains_output(self):
+        record = await self.manager.create("vm-test", 24, 80)
+        terminal = self.application.terminals[0]
+        await terminal.events.put(GuestExecutionError("VM connection lost", outcome_unknown=True))
+        async with asyncio.timeout(2):
+            while (await self.manager.get("vm-test", record["id"]))["status"] == "active":
+                await asyncio.sleep(0.01)
+        saved, _start, output = await self.manager.read("vm-test", record["id"], 0)
+        self.assertEqual(saved["status"], "interrupted")
+        self.assertEqual(output, b"guest$ ")
+        self.assertTrue(terminal.closed)
+
+    async def test_shutdown_finishes_inflight_output_before_releasing_history_ownership(self):
+        import threading
+        entered, release = threading.Event(), threading.Event()
+        history = self.manager._history()
+        append = history.append
+
+        def slow_append(*args):
+            entered.set()
+            release.wait(2)
+            append(*args)
+
+        with patch.object(history, "append", side_effect=slow_append):
+            record = await self.manager.create("vm-test", 24, 80)
+            self.assertTrue(await asyncio.to_thread(entered.wait, 2))
+            closing = asyncio.create_task(self.manager.shutdown())
+            try:
+                await asyncio.sleep(0.03)
+                self.assertFalse(closing.done())
+                with self.assertRaises(ConflictError):
+                    SessionStore(history.path)
+            finally:
+                release.set()
+                await closing
+        reopened = SessionStore(history.path)
+        self.addCleanup(reopened.close)
+        self.assertEqual(reopened.read("vm-test", record["id"], 0)[2], b"guest$ ")
+
 
 @unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required")
 class SessionApiTests(unittest.TestCase):
@@ -265,3 +308,15 @@ class SessionApiTests(unittest.TestCase):
         self.assertEqual(transcript, b"guest$ pwd\r\r\nguest$ ")
         self.assertEqual(self.client.get(self.base, headers=self.headers).json()[0]["status"], "closed")
         self.assertTrue(self.application.terminals[0].closed)
+
+    def test_malformed_input_ends_only_the_browser_attachment(self):
+        record = self.create()
+        ws = self.connect(record)
+        self.output(ws)
+        ws.send_json({"type": "input", "data": "not base64!"})
+        while (message := ws.receive_json())["type"] == "session":
+            pass
+        self.assertEqual(message["type"], "error")
+        self.assertEqual(self.application.terminals[0].input, [])
+        self.assertFalse(self.application.terminals[0].closed)
+        self.assertEqual(self.client.get(self.base + "/" + record["id"], headers=self.headers).json()["status"], "active")