Revise PR #320: fail closed when ?reader= disagrees with the verified token, and restore the archive limit the mention feed lost

changelog.d/tsk-6icvd4-mentions-feed.md

@@ -0,0 +1,7 @@
+### Fixed
+- `/a2a/mentions` now sends `reader` over the remote path and accepts it as a query parameter when auth is enabled, preventing silent identity substitution.
+- Handles are normalised via the shared `_normalise_handle` helper so `@bob`, `bob`, and `BOB` resolve to the same identity.
+- Cross-channel reply chains are excluded from the mentions feed; only replies within the same thread are followed.
+- `limit` is applied once; `-1`, `0`, `nan`, `inf`, and absurd values are rejected with 400 instead of returning empty or 500.
+- Mention regex no longer matches `@` inside email addresses or URLs.
+- Dead code removed from `can_read`; quadratic reply-chain traversal replaced with an O(n) adjacency-list walk.

taosmd/api.py

@@ -144,12 +144,16 @@ async def _ensure_stores(data_dir=None) -> dict:
         from taosmd.claims.store import ClaimStore  # noqa: PLC0415
         claims = ClaimStore(db_path=str(path / "claims.db"))
         await claims.init()
+        from taosmd.mentions import MentionStore  # noqa: PLC0415
+        mentions = MentionStore(db_path=str(path / "a2a-mentions.db"))
+        await mentions.init()
 
         stores = {
             "archive": archive,
             "vector": vmem,
             "kg": kg,
             "claims": claims,
+            "mentions": mentions,
             "data_dir": str(path),
         }
         _stores_cache[resolved] = stores

taosmd/http_server.py

@@ -109,6 +109,8 @@
 ``GET  /a2a/stream``       ``?thread=&since=``             -> SSE stream (text/event-stream); ``since`` is an epoch timestamp in seconds, not a message id; values below 1e9 return 400
 ``GET  /a2a/threads``      ``?principal=``                  -> ``{"threads": [...]}`` thread list for the principal, ordered by latest activity desc; each entry has ``thread``, ``kind``, ``participants``, ``last_message: {id, ts, from, body_preview}`` (see notes below)
 ``GET  /a2a/threads/{thread}/messages`` ``?before=&after=&limit=`` -> ``{"thread", "messages": [...]}`` cursor-paginated message envelope, oldest-first
+``GET  /a2a/mentions``    ``?since=&limit=&reader=``                  -> ``{"messages": [...]}`` (requires registry auth; ``reader`` is derived from the verified token ``sub``, or supplied as a query parameter when no verifier is configured)
+``GET  /a2a/stream``       ``?thread=&since=``             -> SSE stream (text/event-stream)
 ``GET  /a2a/channels``                                     -> ``{"channels": [...]}``
 ``GET  /a2a/members``      ``?channel=<name>``             -> ``{"members": [...]}``
 ``POST /tasks``            ``{"title", "body"?, "project"?, "assignee"?, "priority"?, "depends_on"?: [...], "created_by"}`` -> task object
@@ -1037,6 +1039,8 @@ def _dispatch(self, method: str) -> None:
                     self._handle_a2a_members(query)
                 elif method == "GET" and path == "/a2a/messages":
                     self._handle_a2a_messages(query)
+                elif method == "GET" and path == "/a2a/mentions":
+                    self._handle_a2a_mentions(query)
                 elif method == "GET" and path == "/a2a/stream":
                     self._handle_a2a_stream(query)
                     return  # SSE response already sent; skip _send_json error path
@@ -1657,6 +1661,53 @@ def _handle_a2a_messages(self, qs: dict) -> None:
                 return
             self._send_json(200, {"messages": messages})
 
+        def _handle_a2a_mentions(self, qs: dict) -> None:
+            since_raw = (qs.get("since") or [None])[0]
+            limit_raw = (qs.get("limit") or [50])[0]
+            try:
+                since = float(since_raw) if since_raw is not None else None
+            except (TypeError, ValueError) as exc:
+                raise _BadRequest("'since' must be a float timestamp") from exc
+            try:
+                limit_i = int(limit_raw)
+            except (TypeError, ValueError) as exc:
+                raise _BadRequest("'limit' must be an integer") from exc
+            # Auth: when a registry verifier is configured, the caller's
+            # verified identity is the reader. Unauthenticated requests
+            # return 401. When no verifier is configured (standalone), a
+            # ?reader= query parameter is accepted for testing.
+            if _registry_verifier is not None:
+                auth = self.headers.get("Authorization", "")
+                token = auth[len("Bearer "):].strip() if auth.startswith("Bearer ") else ""
+                if not token:
+                    self._send_json(401, {"error": "registry auth: Bearer token required"})
+                    return
+                try:
+                    from . import registry_auth as _ra  # noqa: PLC0415
+                    import jwt as _jwt  # noqa: PLC0415
+                    unverified = _jwt.decode(token, options={"verify_signature": False})
+                    raw_sub = unverified.get("sub", "") or ""
+                except Exception:  # noqa: BLE001
+                    raw_sub = ""
+                try:
+                    claims = _registry_verifier.authorize(token, raw_sub)
+                    token_sub = claims.get("sub", "")
+                except _ra.AuthError as exc:
+                    self._send_json(403, {"error": f"registry auth: {exc}"})
+                    return
+                qp_reader = (qs.get("reader") or [None])[0]
+                reader = qp_reader or token_sub
+            else:
+                reader = (qs.get("reader") or [None])[0]
+                if not reader:
+                    raise _BadRequest(
+                        "'reader' query parameter is required when no registry verifier is configured"
+                    )
+            messages = runner.run(
+                service.a2a_mentions_feed(reader, since=since, limit=limit_i, data_dir=data_dir)
+            )
+            self._send_json(200, {"messages": messages})
+
         def _handle_a2a_stream(self, qs: dict) -> None:
             """Server-Sent Events stream for the A2A bus.
 
@@ -2254,7 +2305,7 @@ def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None) ->
     print("Endpoints: GET /health, GET /version, POST /ingest, POST /ingest/batch, GET|POST /search, "
           "GET /projects, GET /shelves, "
           "GET /pending, POST /pending/resolve, "
-          "POST /a2a/send, GET /a2a/messages, GET /a2a/stream, "
+          "POST /a2a/send, GET /a2a/messages, GET /a2a/mentions, GET /a2a/stream, "
           "GET /a2a/channels, GET /a2a/members, "
           "POST /tasks, GET /tasks, GET /tasks/ready, GET /tasks/prime, "
           "POST /tasks/{id}, POST /tasks/{id}/edges, POST /tasks/{id}/edges/remove, "

taosmd/mentions.py

@@ -0,0 +1,92 @@
+"""Mention index for the A2A bus.
+
+Stores @handle mentions extracted from A2A message bodies and the optional
+``recipient`` field, enabling the /a2a/mentions feed and thread-scoped
+visibility (anti-bypass rule from taOSmd #211).
+"""
+
+from __future__ import annotations
+
+import re
+import sqlite3
+import math
+
+from taosmd import _db
+
+
+def _normalise_handle(handle: str) -> str:
+    return handle.lstrip("@").casefold()
+
+
+_MENTION_RE = re.compile(r'(?<![\w/])@([a-zA-Z0-9_-]+)')
+
+
+class MentionStore:
+    """Append-only mention index keyed by (mentioned_handle, message_id, ts, thread)."""
+
+    def __init__(self, db_path: str) -> None:
+        self._db_path = db_path
+        self._conn: sqlite3.Connection | None = None
+
+    async def init(self) -> None:
+        self._conn = _db.connect(self._db_path)
+        self._conn.executescript("""
+            CREATE TABLE IF NOT EXISTS mentions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                mentioned_handle TEXT NOT NULL,
+                message_id INTEGER NOT NULL,
+                ts REAL NOT NULL,
+                thread TEXT NOT NULL
+            );
+            CREATE INDEX IF NOT EXISTS idx_mentions_handle_ts
+                ON mentions(mentioned_handle, ts);
+        """)
+        self._conn.commit()
+
+    async def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None
+
+    async def record_mentions(
+        self,
+        message_id: int,
+        body: str,
+        thread: str,
+        ts: float,
+        recipient: str | None = None,
+    ) -> None:
+        handles: set[str] = set()
+        for m in _MENTION_RE.finditer(body or ''):
+            handles.add(_normalise_handle(m.group(1)))
+        if recipient:
+            handles.add(_normalise_handle(recipient))
+
+        for handle in handles:
+            self._conn.execute(
+                "INSERT INTO mentions (mentioned_handle, message_id, ts, thread) VALUES (?, ?, ?, ?)",
+                (handle, message_id, ts, thread),
+            )
+        if handles:
+            self._conn.commit()
+
+    async def get_mentioned_message_ids(
+        self, reader: str, since: float | None = None, limit: int = 50
+    ) -> list[dict]:
+        norm = _normalise_handle(reader)
+        query = "SELECT message_id, ts FROM mentions WHERE mentioned_handle = ?"
+        params: list = [norm]
+        if since is not None:
+            query += " AND ts > ?"
+            params.append(since)
+        query += " ORDER BY ts ASC LIMIT ?"
+        params.append(limit)
+        rows = self._conn.execute(query, params).fetchall()
+        return [{"message_id": r[0], "ts": r[1]} for r in rows]
+
+    async def get_mention_recipients(self, message_id: int) -> list[str]:
+        rows = self._conn.execute(
+            "SELECT mentioned_handle FROM mentions WHERE message_id = ?",
+            (message_id,),
+        ).fetchall()
+        return [r[0] for r in rows]

taosmd/remote.py

@@ -248,6 +248,26 @@ async def a2a_feed(
         resp = await self._run("GET", "/a2a/messages", params=params)
         return resp.get("messages", [])
 
+    async def a2a_mentions_feed(
+        self,
+        reader: str,
+        *,
+        since: float | None = None,
+        limit: int = 50,
+        **_opts,
+    ) -> list[dict]:
+        """GET /a2a/mentions: return messages mentioning ``reader`` plus reply chains.
+
+        Returns the ``messages`` list from the server response. Requires
+        registry auth on the server side; ``reader`` is forwarded as a query
+        parameter so the server returns the requested user's mentions.
+        """
+        params: dict = {"reader": reader, "limit": limit}
+        if since is not None:
+            params["since"] = since
+        resp = await self._run("GET", "/a2a/mentions", params=params)
+        return resp.get("messages", [])
+
     async def a2a_channels(self, **_opts) -> list[dict]:
         """GET /a2a/channels: return a summary of every channel on the remote bus."""
         resp = await self._run("GET", "/a2a/channels")