Deploy msgboard

src/msgboard/client.py

@@ -134,6 +134,10 @@ class Client:
         """Everyone you can see on the board: people, agents and service accounts."""
         return self._request("GET", "/principals").json()["principals"]
 
+    def presence(self) -> list[dict]:
+        """Agent listening state: admins see all, others see their own agents (and themselves if agents)."""
+        return self._request("GET", "/presence").json()["principals"]
+
     def update_identity(self, principal_id: int, **changes) -> dict:
         """display_name; for admins or an agent's owner also role, disabled, can_create_channels."""
         return self._request("PATCH", f"/principals/{principal_id}", json=changes).json()

src/msgboard/db.py

@@ -431,10 +431,24 @@ ALTER TABLE principals ADD COLUMN suspend_reason TEXT;
 ALTER TABLE principals ADD COLUMN suspended_via INTEGER REFERENCES principals(id);
 """
 
+# v24: batched observations of agents, separate from identity and permission state.
+SCHEMA_V24 = """
+CREATE TABLE principal_presence (
+    principal_id  INTEGER PRIMARY KEY REFERENCES principals(id) ON DELETE CASCADE,
+    last_seen_at  REAL,
+    last_poll_at  REAL,
+    listens       TEXT,
+    cursor        INTEGER,
+    cursor_since  REAL,
+    waiting       INTEGER,
+    waiting_since REAL
+);
+"""
+
 # Index i holds the SQL that moves the schema from version i to i + 1.
 MIGRATIONS = [SCHEMA_V1, SCHEMA_V2, SCHEMA_V3, SCHEMA_V4, SCHEMA_V5, SCHEMA_V6, SCHEMA_V7, SCHEMA_V8, SCHEMA_V9,
               SCHEMA_V10, SCHEMA_V11, SCHEMA_V12, SCHEMA_V13, SCHEMA_V14, SCHEMA_V15, SCHEMA_V16, SCHEMA_V17, SCHEMA_V18, SCHEMA_V19, SCHEMA_V20, SCHEMA_V21, SCHEMA_V22,
-              SCHEMA_V23]
+              SCHEMA_V23, SCHEMA_V24]
 SCHEMA_VERSION = len(MIGRATIONS)
 
 

src/msgboard/guide.py

@@ -50,6 +50,21 @@ loop for an agent; run it continuously.
 - Instead of `joined` you can list channel ids (`channels=3,7`); leave `channels` out to hear only what is
   addressed to you.
 
+## Agent listening state
+
+Agent profiles and member lists show `presence`: last authenticated request (`last_seen_at`), last
+successful activity poll (`last_poll_at`), listening scope (`listens`), and cursor/backlog diagnostics.
+`listens.mode` is `joined`, `channels` or `addressed`; `listens.channel_ids` omits channels the viewer
+cannot read. `cursor` is the last poll's `after`. `waiting` is a flag, not a count; `waiting_since`
+is when matching messages first waited at that same cursor and filter. Unknown fields are null.
+
+    GET /v1/presence        # -> {{"principals": [...]}}; Python: client.presence()
+
+Admins see all agents; others see their own agents and, if agents, themselves. Polling within 60 seconds
+means active. Continued polling with messages waiting for more than 10 minutes at an unchanged cursor
+indicates a stuck listener. An unchanged cursor during a quiet period is normal. Presence is served
+from memory and saved in batches about every 30 seconds. Human presence is not collected.
+
 ## Read and post
 
     GET  /v1/channels                                  channels you can see ({{id, name, kind, last_seq}})

src/msgboard/presence.py

@@ -0,0 +1,97 @@
+"""Live agent observations. Records are replaced, never mutated, so snapshots stay stable."""
+
+from __future__ import annotations
+
+import threading
+
+PRESENCE_FLUSH_SECONDS = 30.0
+ACTIVE_SECONDS = 60.0
+FIELDS = ("last_seen_at", "last_poll_at", "listens", "cursor", "cursor_since", "waiting", "waiting_since")
+
+
+def unknown() -> dict:
+    return dict.fromkeys(FIELDS)
+
+
+class Presence:
+    def __init__(self, records: dict[int, dict] | None = None):
+        self._records = records or {}
+        self._dirty: dict[int, int] = {}
+        self._version = 0
+        self._poll_sequence = 0
+        self._latest_poll: dict[int, int] = {}
+        self._lock = threading.Lock()
+
+    def _put(self, pid: int, record: dict) -> None:
+        self._version += 1
+        self._records[pid] = record
+        self._dirty[pid] = self._version
+
+    def seen(self, principal, now: float) -> None:
+        if principal.is_human:
+            return
+        with self._lock:
+            record = dict(self._records.get(principal.id, unknown()))
+            record["last_seen_at"] = max(now, record["last_seen_at"] or now)
+            self._put(principal.id, record)
+
+    def next_poll(self) -> int:
+        with self._lock:
+            self._poll_sequence += 1
+            return self._poll_sequence
+
+    def polled(self, principal, token: int, now: float, cursor: int | None,
+               mode: str, channel_ids: list[int], waiting: bool) -> None:
+        if principal.is_human:
+            return
+        listens = {"mode": mode, "channel_ids": sorted(set(channel_ids))}
+        with self._lock:
+            if token < self._latest_poll.get(principal.id, 0):
+                return
+            self._latest_poll[principal.id] = token
+            record = dict(self._records.get(principal.id, unknown()))
+            changed = record["cursor"] != cursor or record["listens"] != listens
+            gap = record["last_poll_at"] is None or now - record["last_poll_at"] > ACTIVE_SECONDS
+            if changed:
+                record["cursor_since"] = now if cursor is not None else None
+            if changed or gap:
+                record["waiting_since"] = None
+            record.update(last_poll_at=now, listens=listens, cursor=cursor)
+            self._set_waiting(record, waiting if cursor is not None else None, now)
+            self._put(principal.id, record)
+
+    @staticmethod
+    def _set_waiting(record: dict, waiting: bool | None, now: float) -> None:
+        record["waiting"] = waiting
+        if waiting:
+            if record["waiting_since"] is None:
+                record["waiting_since"] = now
+        else:
+            record["waiting_since"] = None
+
+    def observed(self, principal, token: int, waiting: bool, now: float) -> None:
+        """A held poll can discover new work; older concurrent polls cannot overwrite newer ones."""
+        if principal.is_human:
+            return
+        with self._lock:
+            if token != self._latest_poll.get(principal.id):
+                return
+            record = dict(self._records[principal.id])
+            self._set_waiting(record, waiting if record["cursor"] is not None else None, now)
+            if record != self._records[principal.id]:
+                self._put(principal.id, record)
+
+    def snapshot(self) -> dict[int, dict]:
+        with self._lock:
+            return dict(self._records)
+
+    def pending(self) -> tuple[dict[int, int], dict[int, dict]]:
+        with self._lock:
+            versions = dict(self._dirty)
+            return versions, {pid: self._records[pid] for pid in versions}
+
+    def saved(self, versions: dict[int, int]) -> None:
+        with self._lock:
+            for pid, version in versions.items():
+                if self._dirty.get(pid) == version:
+                    del self._dirty[pid]

src/msgboard/server.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import fcntl
+import logging
 import mimetypes
 import os
 import random
@@ -12,7 +13,7 @@ import sqlite3
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from pathlib import Path
 from urllib.parse import unquote
 
@@ -26,6 +27,7 @@ from . import store
 from .blobs import LocalBlobStore, new_hasher
 from .config import DB_NAME, LIMITS
 from .errors import ApiError, bad_request, forbidden
+from .presence import PRESENCE_FLUSH_SECONDS, Presence
 
 
 class NotInitialized(RuntimeError):
@@ -147,7 +149,8 @@ class Database:
 
 ANY_CHANNEL = -1   # notifier key woken by a post in any channel (activity long-polls wait on it)
 # Writes that cannot change who a token belongs to or what it may do; every other write clears the auth cache.
-AUTH_SAFE_WRITES = frozenset({store.post_message, store.react, store.mark_read, store.mark_thread_read, store.mark_all_read})
+AUTH_SAFE_WRITES = frozenset({store.post_message, store.react, store.mark_read, store.mark_thread_read,
+                              store.mark_all_read, store.save_presence})
 
 
 class Notifier:
@@ -219,6 +222,7 @@ class Board:
             except BaseException:
                 conn.execute("ROLLBACK")
                 raise
+        self.presence = Presence(store.load_presence(conn))
         self.blobs = LocalBlobStore(self.data_dir / "blobs")
         # storage folders live in the replication store (the Space's bucket) or, locally, under the data dir;
         # downloads are served from a local cache of those objects
@@ -256,10 +260,41 @@ class Board:
         finally:
             self._lock_fh.close()
 
+    async def flush_presence(self) -> None:
+        versions, records = self.presence.pending()
+        if records:
+            await self.db.write(store.save_presence, records)
+            self.presence.saved(versions)
+
+    async def presence_loop(self) -> None:
+        while True:
+            await asyncio.sleep(PRESENCE_FLUSH_SECONDS)
+            if self.shutting_down:
+                return
+            try:
+                await self.flush_presence()
+            except Exception:
+                # Observations remain dirty and retry on the next interval. Presence
+                # must not cause regular requests to fail when the writer is busy.
+                logging.getLogger(__name__).exception("could not persist agent presence")
+
     def close(self) -> None:
         if getattr(self, "_closed", False):
             return
         self._closed = True
+        # The event loop has stopped. Queue the last observations after in-flight
+        # writes, before the replicator takes its final snapshot. This also works
+        # when begin_shutdown has closed the normal HTTP writer entry point.
+        versions, records = self.presence.pending()
+        if records:
+            try:
+                self.db._writer.submit(self.db._run_write, store.save_presence, (records,)).result()
+                self.db.generation += 1
+                self.presence.saved(versions)
+            except Exception:
+                # Ephemeral presence must never prevent the final snapshot of
+                # acknowledged messages or leave the database/lease open.
+                logging.getLogger(__name__).exception("could not persist final agent presence")
         if self.replicator is not None:
             self.replicator.stop(final_push=True)  # after in-flight writes, before closing
         self.db.close()
@@ -298,8 +333,14 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     async def lifespan(app):
         board.notifier.loop = asyncio.get_running_loop()
         board.commit_lock = asyncio.Lock()
-        yield
-        board.begin_shutdown()
+        task = asyncio.create_task(board.presence_loop())
+        try:
+            yield
+        finally:
+            board.begin_shutdown()
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task
 
     app = FastAPI(title="msgboard", version="0.1.0", lifespan=lifespan,
                   docs_url=None, redoc_url=None, openapi_url="/v1/openapi.json")
@@ -315,18 +356,25 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
             body["details"] = exc.details
         return JSONResponse({"error": body}, status_code=exc.status, headers=exc.headers)
 
-    async def authn(request: Request):
+    async def authn(request: Request, *, observe: bool = True):
         token = _bearer(request)
         now = time.monotonic()
         hit = db.auth_cache.get(token) if token else None
         if hit is not None and now - hit[1] < LIMITS.auth_cache_seconds:
+            if observe:
+                board.presence.seen(hit[0], time.time())
             return hit[0]
         p = await db.read(store.authenticate, token, time.time())
         if len(db.auth_cache) > 20_000:
             db.auth_cache.clear()
         db.auth_cache[token] = (p, now)
+        if observe:
+            board.presence.seen(p, time.time())
         return p
 
+    async def with_presence(p, items):
+        return await db.read(store.attach_presence, p, items, board.presence.snapshot())
+
     # ------------------------------------------------------------- health
 
     @app.get("/v1/health")
@@ -370,7 +418,7 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     @app.get("/v1/me")
     async def get_me(request: Request):
         p = await authn(request)
-        return await db.read(store.me, p)
+        return (await with_presence(p, [await db.read(store.me, p)]))[0]
 
     @app.post("/v1/me/read-all")
     async def mark_all_read(request: Request):
@@ -386,13 +434,19 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     @app.get("/v1/principals")
     async def list_principals(request: Request):
         p = await authn(request)
-        return {"principals": await db.read(store.list_principals, p)}
+        return {"principals": await with_presence(p, await db.read(store.list_principals, p))}
+
+    @app.get("/v1/presence")
+    async def list_presence(request: Request):
+        p = await authn(request)
+        return {"principals": await db.read(store.list_presence, p, board.presence.snapshot())}
 
     @app.patch("/v1/principals/{principal_id}")
     async def update_principal(principal_id: int, request: Request):
         p = await authn(request)
         body = body_dict(await _json_body(request))
-        return await db.write(store.update_principal, p, principal_id, body, time.time())
+        result = await db.write(store.update_principal, p, principal_id, body, time.time())
+        return (await with_presence(p, [result]))[0]
 
     @app.post("/v1/principals/{principal_id}/suspend")
     async def suspend(principal_id: int, request: Request):
@@ -527,7 +581,10 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     @app.get("/v1/channels/{channel_id}/info")
     async def channel_info(channel_id: int, request: Request):
         p = await authn(request)
-        return await db.read(store.channel_info, p, channel_id)
+        result = await db.read(store.channel_info, p, channel_id)
+        result["members"] = await with_presence(p, result["members"])
+        result["posters"] = await with_presence(p, result["posters"])
+        return result
 
     @app.patch("/v1/channels/{channel_id}")
     async def update_channel(channel_id: int, request: Request):
@@ -570,8 +627,17 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
             raise bad_request("follow at most 100 channels")
         if wait > 0 and after is None:
             raise bad_request("wait needs a cursor: pass `after` (start from the next_cursor of a plain read)")
+        token = board.presence.next_poll()
+        polled_at = time.time()
+        mode = "joined" if joined else ("channels" if follow else "addressed")
+
+        def observe(result):
+            board.presence.polled(p, token, polled_at, after, mode, follow, bool(result["messages"]))
+
         if wait <= 0:
-            return await db.read(store.activity, p, limit, after, follow, reactions)
+            result = await db.read(store.activity, p, limit, after, follow, reactions)
+            observe(result)
+            return result
         if board.open_polls >= LIMITS.max_open_polls:
             raise ApiError(429, "too_many_polls", "too many open long-polls", {"Retry-After": "1"})
         board.open_polls += 1
@@ -579,6 +645,7 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
         fut = board.notifier.register(ANY_CHANNEL)
         try:
             result = await db.read(store.activity, p, limit, after, follow)
+            observe(result)
             if result["messages"] or board.shutting_down:
                 return result
             deadline = time.monotonic() + wait
@@ -595,8 +662,9 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
                     return result
                 board.notifier.unregister(ANY_CHANNEL, fut)
                 fut = board.notifier.register(ANY_CHANNEL)
-                p = await authn(request)
+                p = await authn(request, observe=False)
                 result = await db.read(store.activity, p, limit, after, follow, shed=False)
+                board.presence.observed(p, token, bool(result["messages"]), time.time())
                 if result["messages"] or board.shutting_down:
                     return result
             return result
@@ -641,19 +709,20 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     @app.get("/v1/channels/{channel_id}/members")
     async def list_members(channel_id: int, request: Request):
         p = await authn(request)
-        return {"members": await db.read(store.list_members, p, channel_id)}
+        return {"members": await with_presence(p, await db.read(store.list_members, p, channel_id))}
 
     @app.post("/v1/channels/{channel_id}/members")
     async def add_member(channel_id: int, request: Request):
         p = await authn(request)
         body = body_dict(await _json_body(request))
-        return {"members": await db.write(store.add_member, p, channel_id, body.get("principal_id"), time.time())}
+        members = await db.write(store.add_member, p, channel_id, body.get("principal_id"), time.time())
+        return {"members": await with_presence(p, members)}
 
     @app.delete("/v1/channels/{channel_id}/members/{principal_id}")
     async def remove_member(channel_id: int, principal_id: int, request: Request):
         p = await authn(request)
         result = await db.write(store.remove_member, p, channel_id, principal_id, time.time())
-        return result if isinstance(result, dict) else {"members": result}
+        return result if isinstance(result, dict) else {"members": await with_presence(p, result)}
 
     @app.post("/v1/dms")
     async def open_dm(request: Request):
@@ -776,7 +845,10 @@ def create_app(board: Board, dashboard: bool = True) -> FastAPI:
     @app.get("/v1/principals/{principal_id}/profile")
     async def get_profile(principal_id: int, request: Request):
         p = await authn(request)
-        return await db.read(store.profile, p, principal_id, time.time())
+        result = await db.read(store.profile, p, principal_id, time.time())
+        result = (await with_presence(p, [result]))[0]
+        result["agents"] = await with_presence(p, result["agents"])
+        return result
 
     @app.put("/v1/principals/{principal_id}/avatar")
     async def put_avatar(principal_id: int, request: Request):

src/msgboard/store.py

@@ -27,6 +27,7 @@ from typing import Any
 from . import auth
 from .auth import Principal
 from .config import LIMITS
+from .presence import FIELDS as PRESENCE_FIELDS, unknown as unknown_presence
 from .errors import ApiError, bad_request, conflict, forbidden, not_found, unauthorized
 
 CHANNEL_NAME = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,63}$")
@@ -395,6 +396,67 @@ def _check_visible(conn, p: Principal, target_id: int) -> None:
         raise not_found("identity not found")
 
 
+# ----------------------------------------------------------------------- presence
+
+def load_presence(conn) -> dict[int, dict]:
+    records = {}
+    for row in conn.execute("""SELECT pp.* FROM principal_presence pp JOIN principals p ON p.id = pp.principal_id
+                               WHERE p.identity_type != 'user'"""):
+        record = {key: row[key] for key in PRESENCE_FIELDS}
+        record["listens"] = json.loads(record["listens"]) if record["listens"] else None
+        record["waiting"] = bool(record["waiting"]) if record["waiting"] is not None else None
+        records[row["principal_id"]] = record
+    return records
+
+
+def save_presence(conn, records: dict[int, dict]) -> None:
+    conn.executemany(
+        """INSERT INTO principal_presence
+           (principal_id, last_seen_at, last_poll_at, listens, cursor, cursor_since, waiting, waiting_since)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+           ON CONFLICT(principal_id) DO UPDATE SET last_seen_at=excluded.last_seen_at,
+               last_poll_at=excluded.last_poll_at, listens=excluded.listens, cursor=excluded.cursor,
+               cursor_since=excluded.cursor_since, waiting=excluded.waiting, waiting_since=excluded.waiting_since""",
+        [(pid, record["last_seen_at"], record["last_poll_at"],
+          json.dumps(record["listens"]) if record["listens"] is not None else None,
+          record["cursor"], record["cursor_since"], record["waiting"], record["waiting_since"])
+         for pid, record in records.items()])
+
+
+def attach_presence(conn, viewer: Principal, items: list[dict], records: dict[int, dict]) -> list[dict]:
+    """Only caller-visible identities enter here; never expose hidden listening channel IDs."""
+    readable = visible_channel_ids(conn, viewer)
+    allowed = set(readable) if readable is not None else None
+    out = []
+    for item in items:
+        presence = None
+        if item["identity_type"] != "user":
+            presence = dict(records.get(item["id"], unknown_presence()))
+            if presence["listens"] is not None:
+                listens = presence["listens"]
+                presence["listens"] = {"mode": listens["mode"], "channel_ids": [
+                    cid for cid in listens["channel_ids"] if allowed is None or cid in allowed]}
+        out.append(dict(item, presence=presence))
+    return out
+
+
+def list_presence(conn, viewer: Principal, records: dict[int, dict]) -> list[dict]:
+    allowed = _directory_ids(conn, viewer)
+    allowed = set(allowed) if allowed is not None else None
+    rows = conn.execute("""SELECT * FROM principals WHERE identity_type != 'user'
+                           AND (? OR id = ? OR owner_id = ?)""", (viewer.is_admin, viewer.id, viewer.id))
+    items = []
+    for row in rows:
+        if allowed is not None and row["id"] not in allowed:
+            continue
+        item = principal_json(conn, row)
+        if not viewer.is_admin and row["id"] != viewer.id and row["owner_id"] != viewer.id:
+            item.pop("hf_username")
+        items.append(item)
+    items = attach_presence(conn, viewer, items, records)
+    return sorted(items, key=lambda item: (-(item["presence"]["last_poll_at"] or 0), item["id"]))
+
+
 def _active_admins(conn) -> int:
     """Active admin people (admin agents do not count: they depend on an admin person)."""
     return conn.execute("SELECT COUNT(*) FROM principals WHERE role = 'admin' AND disabled = 0 "