Fix Hub FastAPI lifespan startup and shutdown

frontends/hub.py

@@ -171,6 +171,8 @@ def phone_session():
 
 # ------------------------------- server (python hub.py) -------------------------------
 if __name__ == '__main__':
+    import logging
+    from contextlib import asynccontextmanager
     from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, Response
     from fastapi.responses import FileResponse, JSONResponse; import uvicorn
     HERE, OK_ORIGIN = os.path.dirname(os.path.abspath(__file__)), ('', f'http://127.0.0.1:{PORT}', f'http://localhost:{PORT}')
@@ -183,7 +185,16 @@ if __name__ == '__main__':
         with open(TOKEN_FILE, 'w', encoding='utf-8') as f: f.write(TOKEN)
     except OSError: pass
     STATUS = {'offline': 404, 'gone': 404, 'timeout': 504, 'nosupport': 501, 'badop': 400, 'busy': 409, 'idle': 409}
-    app, peers, meta, waits, seq, pcache, pdead = FastAPI(), {}, {}, {}, [0], {}, {}  # ws / hello / futures / row / probe
+    p2p = None
+    @asynccontextmanager
+    async def lifespan(app):
+        if p2p is None:
+            yield
+        else:
+            async with p2p["lifespan"](app):
+                yield
+
+    app, peers, meta, waits, seq, pcache, pdead = FastAPI(lifespan=lifespan), {}, {}, {}, [0], {}, {}  # ws / hello / futures / row / probe
     @app.middleware('http')
     async def guard(req: Request, call_next):
         if req.scope['server'][1] == PORT: return Response(status_code=404)   # bus port speaks WebSocket only
@@ -278,9 +289,10 @@ if __name__ == '__main__':
     # ---- optional P2P pairing; delete this block to remove it completely ----
     try:
         from hub_p2p import install as _install_p2p
-        _install_p2p(app, web_port=WEB_PORT, token=TOKEN, here=HERE)
-    except Exception:
-        pass  # missing plug-in/client/dependency must not affect the hub
+    except ImportError:
+        logging.getLogger(__name__).warning("P2P unavailable; local Hub only", exc_info=True)
+    else:
+        p2p = _install_p2p(app, web_port=WEB_PORT, token=TOKEN, here=HERE)
 
     @app.get('/')
     async def index(): return FileResponse(os.path.join(HERE, 'hub.html'))
@@ -288,9 +300,16 @@ if __name__ == '__main__':
     async def vendor(f: str):
         p = os.path.join(HERE, 'desktop', 'static', 'vendor', os.path.basename(f))
         return FileResponse(p) if os.path.exists(p) else Response(status_code=404)
-    def serve(p):
-        uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=p, log_level='warning')).run()
+    # Only the web loop owns P2P startup/shutdown; the bus shares routes, not lifespan.
+    web = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=WEB_PORT,
+                                      log_level='warning', lifespan='on'))
+    bus = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=PORT,
+                                      log_level='warning', lifespan='off'))
     print(f'[hub] bus ws://127.0.0.1:{PORT}/ws (no HTTP)  |  panel http://127.0.0.1:{WEB_PORT}/?t={TOKEN}')
-    threading.Thread(target=serve, args=(WEB_PORT,), daemon=True).start()
-    try: serve(PORT)
+    thread = threading.Thread(target=web.run, daemon=True)
+    thread.start()
+    try: bus.run()
     except OSError: sys.exit(f'hub already running on {PORT}')
+    finally:
+        web.should_exit = True
+        thread.join(timeout=10)

frontends/hub_p2p.py

@@ -2,6 +2,7 @@
 import asyncio
 import os
 import socket
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import Request, Response
@@ -146,15 +147,33 @@ poll();</script>''', media_type="text/html")
             "error": state["error"],
         })
 
-    async def startup():
+    @asynccontextmanager
+    async def lifespan(app):
         if state["started"]:
-            return
+            raise RuntimeError("P2P lifespan is already running")
         state["started"] = True
         try:
-            load_room(name, path=rooms)
-        except KeyError:
-            return
-        start_group()
+            try:
+                load_room(name, path=rooms)
+            except KeyError:
+                pass  # Unpaired hubs remain available for /pair.
+            else:
+                start_group()
+            yield
+        finally:
+            group = state["group"]
+            links = [*group._out.values(), *group._in.values()] if group else []
+            tasks = [t for t in [state["task"], state["legacy"],
+                                 *state.get("group_tasks", [])] if t is not None]
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            for task in links:
+                if not task.done() and not task.cancelling():
+                    task.cancel()
+            await asyncio.gather(*links, return_exceptions=True)
+            state.update(task=None, legacy=None, group=None, group_tasks=[],
+                         invite=None, started=False, status="idle", error=None)
 
-    app.add_event_handler("startup", startup)
+    state["lifespan"] = lifespan
     return state

frontends/tests/test_hub_p2p_lifespan.py

@@ -0,0 +1,132 @@
+"""FastAPI lifespan integration; no real pairing records or network access."""
+import ast
+import asyncio
+from contextlib import asynccontextmanager
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+from fastapi import FastAPI
+
+FRONTENDS = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(FRONTENDS))
+import hub_p2p
+
+
+class FakeGroup:
+    def __init__(self, *args, **kwargs):
+        self._out, self._in = {}, {}
+        self.members, self._ws = set(), None
+        self.ran = self.closed = False
+
+    async def run(self):
+        self.ran = True
+        try:
+            await asyncio.Event().wait()
+        finally:
+            for task in [*self._out.values(), *self._in.values()]:
+                task.cancel()
+            self.closed = True
+
+
+class LifespanTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.loader = self.enterContext(patch.object(hub_p2p, 'load_room', return_value={
+            'signal_url': 'ws://unused.invalid/ws', 'room': 'test-only',
+        }))
+        self.enterContext(patch.object(hub_p2p, 'Group', FakeGroup))
+        # Exactly the same explicit FastAPI lifespan delegation as hub.py.
+        @asynccontextmanager
+        async def lifespan(app):
+            async with self.state['lifespan'](app):
+                yield
+        self.app = FastAPI(lifespan=lifespan)
+        self.state = hub_p2p.install(self.app, web_port=1, token='test-only', here=FRONTENDS)
+
+    @asynccontextmanager
+    async def running(self):
+        incoming, outgoing = asyncio.Queue(), asyncio.Queue()
+        scope = {'type': 'lifespan', 'asgi': {'version': '3.0', 'spec_version': '2.0'}, 'state': {}}
+        task = asyncio.create_task(self.app(scope, incoming.get, outgoing.put))
+        try:
+            await incoming.put({'type': 'lifespan.startup'})
+            result = await asyncio.wait_for(outgoing.get(), 2)
+            if result['type'] != 'lifespan.startup.complete':
+                await task  # Propagate the original startup exception.
+                self.fail(result)
+            yield
+            await incoming.put({'type': 'lifespan.shutdown'})
+            result = await asyncio.wait_for(outgoing.get(), 2)
+            self.assertEqual(result['type'], 'lifespan.shutdown.complete')
+            await task
+        finally:
+            if not task.done():
+                task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+    async def test_saved_pairing_auto_starts_and_can_restart(self):
+        for _ in range(2):
+            async with self.running():
+                group = self.state['group']
+                self.assertIsInstance(group, FakeGroup)
+                await asyncio.sleep(0)
+                self.assertTrue(group.ran)
+                self.assertTrue(self.state['started'])
+            self.assertTrue(group.closed)
+            self.assertFalse(self.state['started'])
+            self.assertIsNone(self.state['group'])
+
+    async def test_unpaired_starts_without_creating_pairing(self):
+        self.loader.side_effect = KeyError('unpaired')
+        async with self.running():
+            self.assertIsNone(self.state['group'])
+            self.assertIsNone(self.state['task'])
+            self.assertEqual(self.state['status'], 'idle')
+
+    async def test_startup_error_is_not_silently_ignored(self):
+        self.loader.side_effect = ValueError('invalid pairing configuration')
+        with self.assertRaisesRegex(ValueError, 'invalid pairing'):
+            async with self.running():
+                self.fail('Startup must fail')
+        self.assertFalse(self.state['started'])
+
+    async def test_shutdown_awaits_links_pairing_and_legacy(self):
+        closed = set()
+        async def worker(name):
+            try:
+                await asyncio.Event().wait()
+            finally:
+                await asyncio.sleep(0)  # Cleanup must finish, not just receive cancel.
+                closed.add(name)
+        async with self.running():
+            group = self.state['group']
+            group._out['a'] = asyncio.create_task(worker('out'))
+            group._in['b'] = asyncio.create_task(worker('in'))
+            self.state['task'] = asyncio.create_task(worker('pair'))
+            self.state['legacy'] = asyncio.create_task(worker('legacy'))
+            await asyncio.sleep(0)
+        self.assertEqual(closed, {'out', 'in', 'pair', 'legacy'})
+        self.assertFalse([t for t in asyncio.all_tasks() if t is not asyncio.current_task()])
+
+    async def test_duplicate_lifespan_rejected_without_stopping_owner(self):
+        async with self.running():
+            group = self.state['group']
+            with self.assertRaisesRegex(RuntimeError, 'already running'):
+                async with self.state['lifespan'](self.app):
+                    self.fail('Duplicate startup')
+            self.assertIs(self.state['group'], group)
+            self.assertTrue(self.state['started'])
+
+    def test_only_web_server_owns_lifespan(self):
+        tree = ast.parse((FRONTENDS / 'hub.py').read_text())
+        configs = {}
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == 'Config':
+                kw = {k.arg: k.value for k in node.keywords}
+                configs[kw['port'].id] = ast.literal_eval(kw['lifespan'])
+        self.assertEqual(configs, {'WEB_PORT': 'on', 'PORT': 'off'})
+
+
+if __name__ == '__main__':
+    unittest.main()