Added a live chat stream for the open chat page

.claude/CLAUDE.md

@@ -6,7 +6,7 @@ The Schuly **Microsoft plugin** is the only caller.
 
 ## What it is / isn't
 
-- **Stateless** apart from the short-lived in-memory password sign-in jobs. No database, no stored tokens. The caller owns the FOCI refresh token and
+- **Stateless** apart from the short-lived in-memory password sign-in jobs and the live chat streams, which last only as long as their request. No database, no stored tokens. The caller owns the FOCI refresh token and
   sends it back in the `X-MS-Refresh-Token` header; reads return the rotated token in the
   same response header.
 - Sign-in is the **browser sign-in** (authorization code + PKCE in a WebView, used by the
@@ -44,6 +44,7 @@ The Schuly **Microsoft plugin** is the only caller.
 - `AccountRunner` strips `_native` from every result. No route calls OutNote's native reads today: chat
   messages stay on Graph's list because `ChatReader` and the callers need its attachments, mentions and
   reactions. `emails()` gets an explicit `$select` because the web's list shows `toRecipients`.
+- `GET /api/chats/live` (`ChatLive`) runs `note.listen(reconnect=False)` in a daemon thread for as long as the caller stays, so one thread and one Teams connection per open chat page, capped per process and per account. `listen()` stops only on `close()`, so the stream's `finally` must reach it. It forwards only `{kind, chatId, fromMe}` for messages, edits and deletions: no content, typing or presence. The route is undocumented and has no Graph fallback; the web app polls while it is down.
 - Never add `SQLiteCache` or any other on-disk cache. `stream()` cursors would have to live in the plugin.
 
 ## Cold start
@@ -109,7 +110,7 @@ src/
     logging_config.py         # logger names
     microsoft/                # AccountRunner, OutNoteClientFactory, BrokerCache, SharedBroker, AccountCache, MicrosoftHttpClients, DeviceLoginClient, BrowserSignInClient, ProfileReader
     graph/                    # TeamsFiles, GraphSearch, OneNoteClient, OneNoteRunner, PdfConverter, FileThumbnails, ChannelMessages, PeoplePhotos, MailFolderRoles
-    teams/                    # TeamsChatClient (skype token, AMS), ChatReader, ChatSender, ChatPresence, EmojiCatalogue
+    teams/                    # TeamsChatClient (skype token, AMS), ChatReader, ChatSender, ChatLive, ChatPresence, EmojiCatalogue
     giphy/                    # GifSearch (Teams' own GIF command with the account's sign-in; 503 when Teams refuses)
     files/                    # FileTextExtractor (lazy imports, zip-bomb guard, caps, paging)
     password_signin/          # JobStore, SignInJob (worker thread per job, MfaPrompt bridge), RateLimiter

requirements.txt

@@ -2,7 +2,7 @@ fastapi==0.122.0
 uvicorn==0.38.0
 httpx==0.28.1
 python-multipart==0.0.20
-outnote==2.2.0
+outnote==2.3.0
 ms-entrance==2.0.0
 wreq==0.13.0
 python-docx==1.2.0

src/api/controllers/chats_controller.py

@@ -19,6 +19,7 @@
 from src.application.queries.get_chat_presence_query import GetChatPresenceQuery
 from src.application.queries.list_chat_messages_query import ListChatMessagesQuery
 from src.application.queries.list_chats_query import ListChatsQuery
+from src.application.queries.open_chat_live_query import OpenChatLiveQuery
 
 router = APIRouter(prefix="/api")
 
@@ -73,6 +74,11 @@ async def chat_presence(self, chat_id: str, token: str = Header(..., alias=HttpH
         """The Teams status of the other person in a 1:1 chat: `{availability, activity}`, both null when there is none to show."""
         return await self.mediator.send(GetChatPresenceQuery(token, chat_id))
 
+    @router.get("/chats/live", tags=["Microsoft 365"], response_class=StreamingResponse, responses={200: {"content": {"application/x-ndjson": {}}}})
+    async def chats_live(self, token: str = Header(..., alias=HttpHeaders.REFRESH_TOKEN_HEADER)):
+        """Live chat events, one JSON object per line, for as long as the caller stays: `connected`, `message`, `message_updated` and `message_deleted` with `chatId` and `fromMe`, and `ping` to keep the connection up. It says what changed where and nothing else. When it ends, reconnect."""
+        return StreamingResponse(await self.mediator.send(OpenChatLiveQuery(token)), media_type="application/x-ndjson", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"})
+
     @router.get("/chats/{chat_id}/photo", tags=["Microsoft 365"], response_class=Response, responses={200: {"content": {"image/jpeg": {}}}})
     async def chat_photo(self, chat_id: str, token: str = Header(..., alias=HttpHeaders.REFRESH_TOKEN_HEADER)):
         return FileResponses.image(await self.mediator.send(GetChatPhotoQuery(token, chat_id)))

src/api/extensions/chats_extensions.py

@@ -1,5 +1,6 @@
 from src.api.di.service_collection import ServiceCollection
 from src.application.abstractions.i_chat_presence import IChatPresence
+from src.application.abstractions.i_chat_live import IChatLive
 from src.application.abstractions.i_chat_reader import IChatReader
 from src.application.abstractions.i_chat_sender import IChatSender
 from src.application.abstractions.i_teams_chat import ITeamsChat
@@ -14,8 +15,10 @@
 from src.application.queries.list_chat_messages_query import ListChatMessagesQuery, ListChatMessagesHandler
 from src.application.queries.get_chat_presence_query import GetChatPresenceQuery, GetChatPresenceHandler
 from src.application.queries.list_chats_query import ListChatsQuery, ListChatsHandler
+from src.application.queries.open_chat_live_query import OpenChatLiveQuery, OpenChatLiveHandler
 from src.application.services.chat_html_sanitizer import ChatHtmlSanitizer
 from src.infrastructure.teams.chat_presence import ChatPresence
+from src.infrastructure.teams.chat_live import ChatLive
 from src.infrastructure.teams.chat_reader import ChatReader
 from src.infrastructure.teams.chat_sender import ChatSender
 from src.infrastructure.teams.teams_chat_client import TeamsChatClient
@@ -27,6 +30,7 @@ def add_chats(services: ServiceCollection) -> ServiceCollection:
         services.add_singleton(ChatHtmlSanitizer)
         services.add_singleton(ITeamsChat, TeamsChatClient)
         services.add_singleton(IChatReader, ChatReader)
+        services.add_singleton(IChatLive, ChatLive)
         services.add_singleton(IChatPresence, ChatPresence)
         services.add_singleton(IChatSender, ChatSender)
 
@@ -40,5 +44,6 @@ def add_chats(services: ServiceCollection) -> ServiceCollection:
         services.add_handler(ScheduleChatMeetingCommand, ScheduleChatMeetingHandler)
         services.add_handler(MarkChatReadCommand, MarkChatReadHandler)
         services.add_handler(GetChatPhotoQuery, GetChatPhotoHandler)
+        services.add_handler(OpenChatLiveQuery, OpenChatLiveHandler)
         services.add_handler(GetChatPresenceQuery, GetChatPresenceHandler)
         return services

src/application/abstractions/i_chat_live.py

@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+from collections.abc import AsyncIterator
+
+from src.application.abstractions.i_account_broker import IAccountBroker
+from src.application.abstractions.i_graph_client import IGraphClient
+
+
+class IChatLive(ABC):
+    """Live chat events as NDJSON lines, for as long as the caller reads them."""
+
+    @abstractmethod
+    def open(self, note: IGraphClient, broker: IAccountBroker) -> AsyncIterator[bytes]: ...

src/application/abstractions/i_graph_client.py

@@ -13,6 +13,9 @@ def list(self, path, params=None, limit=None, client=None): ...
     @abstractmethod
     def post(self, path, body=None, client=None): ...
 
+    @abstractmethod
+    def listen(self, reconnect=True, presence=()): ...
+
     @abstractmethod
     def presences(self, user_ids): ...
 

src/application/queries/open_chat_live_query.py

@@ -0,0 +1,31 @@
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+
+from mediatorx import IQuery, IQueryHandler
+
+from src.application.abstractions.i_account_runner import IAccountRunner
+from src.application.abstractions.i_chat_live import IChatLive
+from src.application.options.microsoft_options import MicrosoftOptions
+from src.domain.validation.refresh_token_format import RefreshTokenFormat
+
+
+@dataclass
+class OpenChatLiveQuery(IQuery[AsyncIterator[bytes]]):
+    refresh_token: str = field(repr=False)
+
+
+class OpenChatLiveHandler(IQueryHandler[OpenChatLiveQuery, AsyncIterator[bytes]]):
+    def __init__(self, accounts: IAccountRunner, live: IChatLive, options: MicrosoftOptions):
+        self._accounts = accounts
+        self._live = live
+        self._options = options
+
+    async def handle(self, query: OpenChatLiveQuery) -> AsyncIterator[bytes]:
+        RefreshTokenFormat.check(query.refresh_token)
+
+        def run(note, broker):
+            # A dead sign-in is refused here, as a real status, before the stream starts.
+            broker(self._options.device_client_id)
+            return self._live.open(note, broker)
+
+        return await self._accounts.run(query.refresh_token, run)

src/infrastructure/teams/chat_live.py

@@ -0,0 +1,104 @@
+"""Live chat events for one open Schuly chat page, over OutNote's `listen()` (Teams' own push service, undocumented).
+
+The feed lives exactly as long as the request: no listener outlives its caller, and nothing is stored. It says only that something
+changed in a chat, never what: the caller reads the messages the normal way. `reconnect=False` on purpose, so a broken feed ends the
+stream and the caller sees it, falls back to polling and reconnects, rather than a stream that looks up and delivers nothing.
+"""
+
+import asyncio
+import json
+import logging
+import threading
+from collections import Counter
+from collections.abc import AsyncIterator
+
+from src.application.abstractions.i_chat_live import IChatLive
+from src.domain.errors.service_error import ServiceError
+from src.domain.validation.graph_ids import GraphIds
+from src.infrastructure.logging_config import LoggerNames
+
+
+class ChatLive(IChatLive):
+    """One thread and one Teams connection per open stream, capped because this service is one process."""
+
+    log = logging.getLogger(LoggerNames.SERVICE)
+    HEARTBEAT = 20.0  # seconds; keeps proxies from closing a quiet stream
+    MAX_STREAMS = 200
+    MAX_PER_ACCOUNT = 3  # a few tabs; each one is its own connection to Teams
+    PING = b'{"kind":"ping"}\n'
+    CHANGES = frozenset({"message", "message_updated", "message_deleted"})
+
+    def __init__(self):
+        self._guard = threading.Lock()
+        self._streams = Counter()
+
+    def _has_room(self, broker) -> bool:
+        return self._streams.total() < self.MAX_STREAMS and self._streams[broker] < self.MAX_PER_ACCOUNT
+
+    def _enter(self, broker) -> bool:
+        with self._guard:
+            if not self._has_room(broker):
+                return False
+            self._streams[broker] += 1
+            return True
+
+    def _leave(self, broker) -> None:
+        with self._guard:
+            self._streams[broker] -= 1
+            if self._streams[broker] <= 0:
+                del self._streams[broker]
+
+    def line(self, event) -> bytes | None:
+        """What the caller may see of an event, or None: everything else (typing, presence, read receipts) stays here."""
+        if event.kind == "connected":
+            return b'{"kind":"connected"}\n'
+        if event.kind not in self.CHANGES or not isinstance(event.chat_id, str):
+            return None
+        if not GraphIds.CHAT_ID_PATTERN.fullmatch(event.chat_id) or GraphIds.has_dot_segment(event.chat_id):
+            return None
+        return (json.dumps({"kind": event.kind, "chatId": event.chat_id, "fromMe": bool(event.from_me)}) + "\n").encode()
+
+    def open(self, note, broker) -> AsyncIterator[bytes]:
+        with self._guard:
+            if not self._has_room(broker):
+                raise ServiceError(503, "Too many live chat connections.", headers={"Retry-After": "60"})
+        return self._stream(note.listen(reconnect=False), broker)
+
+    def _pump(self, live, loop, queue) -> None:
+        def put(item) -> None:
+            try:
+                loop.call_soon_threadsafe(queue.put_nowait, item)
+            except RuntimeError:
+                live.close()  # the loop is gone, and so is the caller
+
+        try:
+            for event in live:
+                line = self.line(event)
+                if line:
+                    put(line)
+        except Exception as ex:  # network, auth or protocol: the caller reconnects
+            self.log.warning("Live chat feed ended: %s", type(ex).__name__)
+        finally:
+            put(None)
+
+    async def _stream(self, live, broker) -> AsyncIterator[bytes]:
+        # The slot is taken here and not in open(): a generator that never starts never reaches a finally.
+        if not self._enter(broker):
+            live.close()
+            return
+        try:
+            queue = asyncio.Queue()
+            threading.Thread(target=self._pump, args=(live, asyncio.get_running_loop(), queue), daemon=True, name="chat-live").start()
+            yield self.PING
+            while True:
+                try:
+                    line = await asyncio.wait_for(queue.get(), self.HEARTBEAT)
+                except TimeoutError:
+                    yield self.PING
+                    continue
+                if line is None:
+                    return
+                yield line
+        finally:
+            live.close()
+            self._leave(broker)