新增专业团队全局话题接口

hermes-plugins/yaoyao/dashboard/group_plugin_api.py

@@ -1137,7 +1137,8 @@ async def capabilities() -> dict[str, object]:
         "limits": limits_payload(),
         "eventTypes": sorted(EVENT_TYPES),
         "features": [
-            "hostFlow", "roomInstructions", "roomAvatar", "nodeWorker", "nodeRegistry"
+            "hostFlow", "roomInstructions", "roomAvatar", "globalTopics",
+            "nodeWorker", "nodeRegistry"
         ],
     }
 
@@ -1166,6 +1167,16 @@ async def list_topics(
     return {"items": page.items, "nextCursor": page.next_cursor}
 
 
+async def list_global_topics(
+    limit: int = Query(default=50, ge=1, le=100),
+    cursor: str | None = Query(default=None, min_length=1, max_length=4096),
+) -> dict[str, object]:
+    page = await _store_call(
+        "list_global_topics", limit=limit, cursor=cursor
+    )
+    return {"items": page.items, "nextCursor": page.next_cursor}
+
+
 async def list_pinned_topics(
     limit: int = Query(default=100, ge=1, le=100),
 ) -> dict[str, object]:
@@ -1535,6 +1546,7 @@ def _build_router() -> APIRouter:
     )
     built.add_api_route("/capabilities", capabilities, methods=["GET"])
     built.add_api_route("/rooms", list_rooms, methods=["GET"])
+    built.add_api_route("/topics", list_global_topics, methods=["GET"])
     built.add_api_route("/topics/pinned", list_pinned_topics, methods=["GET"])
     built.add_api_route("/rooms", create_room, methods=["POST"])
     built.add_api_route("/rooms/{room_id}", get_room, methods=["GET"])

hermes-plugins/yaoyao/dashboard/group_store.py

@@ -1967,6 +1967,70 @@ class GroupStore:
             )
         return CursorPage(items=items, next_cursor=next_cursor)
 
+    def list_global_topics(
+        self, *, limit: int, cursor: str | None
+    ) -> CursorPage:
+        """Return active topics across active rooms in stable professional order."""
+        if (
+            not isinstance(limit, int)
+            or isinstance(limit, bool)
+            or not 1 <= limit <= 100
+        ):
+            raise ValueError("limit must be an integer from 1 to 100")
+        cursor_values = (
+            self._decode_global_topic_cursor(cursor)
+            if cursor is not None
+            else None
+        )
+        query = """SELECT topic.*,
+            (SELECT COUNT(*) FROM group_messages AS counted
+             WHERE counted.topic_id = topic.id AND counted.visible = 1)
+                AS message_count,
+            COALESCE((SELECT CASE WHEN TRIM(latest.content) != ''
+                                  THEN latest.content ELSE latest.error END
+             FROM group_messages AS latest
+             WHERE latest.topic_id = topic.id AND latest.visible = 1
+               AND (TRIM(latest.content) != '' OR TRIM(latest.error) != '')
+             ORDER BY latest.seq DESC LIMIT 1), '') AS preview,
+            (SELECT COUNT(*) FROM group_messages AS unread
+             WHERE unread.topic_id = topic.id AND unread.visible = 1
+               AND unread.sender_kind = 'agent'
+               AND (TRIM(unread.content) != '' OR TRIM(unread.error) != '')
+               AND unread.seq > topic.last_read_message_seq) AS unread_count,
+            COALESCE((SELECT MAX(visible_message.seq)
+             FROM group_messages AS visible_message
+             WHERE visible_message.topic_id = topic.id
+               AND visible_message.visible = 1), 0) AS latest_message_seq
+            FROM group_topics AS topic
+            INNER JOIN group_rooms AS room ON room.id = topic.room_id
+            WHERE topic.archived = 0 AND room.archived = 0"""
+        params: list[object] = []
+        if cursor_values is not None:
+            pinned, updated_at, topic_id = cursor_values
+            query += """ AND (
+                topic.pinned < ? OR (
+                    topic.pinned = ? AND (
+                        topic.updated_at < ? OR (
+                            topic.updated_at = ? AND topic.id < ?
+                        )
+                    )
+                )
+            )"""
+            params.extend((pinned, pinned, updated_at, updated_at, topic_id))
+        query += " ORDER BY topic.pinned DESC, topic.updated_at DESC, topic.id DESC LIMIT ?"
+        params.append(limit + 1)
+        with self.read_transaction() as connection:
+            rows = connection.execute(query, params).fetchall()
+        has_more = len(rows) > limit
+        items = [self._topic_summary(row) for row in rows[:limit]]
+        next_cursor = None
+        if has_more and items:
+            last = rows[limit - 1]
+            next_cursor = self._encode_global_topic_cursor(
+                bool(last["pinned"]), last["updated_at"], last["id"]
+            )
+        return CursorPage(items=items, next_cursor=next_cursor)
+
     def list_pinned_topics(self, *, limit: int) -> CursorPage:
         if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100:
             raise ValueError("limit must be an integer from 1 to 100")
@@ -7311,3 +7375,57 @@ class GroupStore:
         if room_id != expected_room_id:
             raise ValueError("cursor does not belong to room")
         return updated_at, topic_id
+
+    @staticmethod
+    def _encode_global_topic_cursor(
+        pinned: bool, updated_at: float, topic_id: str
+    ) -> str:
+        raw = json.dumps(
+            ["global-topic", int(pinned), updated_at, topic_id],
+            separators=(",", ":"),
+        ).encode()
+        return base64.urlsafe_b64encode(raw).decode().rstrip("=")
+
+    @classmethod
+    def _decode_global_topic_cursor(
+        cls, cursor: str
+    ) -> tuple[int, float, str]:
+        if (
+            not isinstance(cursor, str)
+            or not cursor
+            or not re.fullmatch(r"[A-Za-z0-9_-]+", cursor)
+        ):
+            raise ValueError("cursor is malformed")
+        try:
+            raw = base64.b64decode(
+                cursor + "=" * (-len(cursor) % 4), altchars=b"-_", validate=True
+            )
+            decoded = json.loads(raw.decode("utf-8"))
+            if (
+                not isinstance(decoded, list)
+                or len(decoded) != 4
+                or decoded[0] != "global-topic"
+                or isinstance(decoded[1], bool)
+                or decoded[1] not in (0, 1)
+                or isinstance(decoded[2], bool)
+                or not isinstance(decoded[2], (int, float))
+                or not math.isfinite(decoded[2])
+            ):
+                raise ValueError
+            pinned = int(decoded[1])
+            updated_at = float(decoded[2])
+            topic_id = cls._canonical_uuid(decoded[3], "cursor topic id")
+            if cursor != cls._encode_global_topic_cursor(
+                bool(pinned), updated_at, topic_id
+            ):
+                raise ValueError
+            return pinned, updated_at, topic_id
+        except (
+            ValueError,
+            TypeError,
+            UnicodeDecodeError,
+            json.JSONDecodeError,
+            binascii.Error,
+            OverflowError,
+        ) as error:
+            raise ValueError("cursor is malformed") from error

hermes-plugins/yaoyao/dashboard/tests/test_group_topics_contract.py

@@ -11,7 +11,7 @@ import unittest
 import uuid
 from io import BytesIO
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 from fastapi import UploadFile
 from starlette.datastructures import Headers
@@ -54,6 +54,7 @@ class GroupTopicsContractTests(unittest.TestCase):
             if hasattr(route, "methods")
         }
         self.assertIn(("/v1/rooms/{room_id}/topics", ("GET",)), route_methods)
+        self.assertIn(("/v1/topics", ("GET",)), route_methods)
         self.assertIn(
             ("/v1/rooms/{room_id}/topics/{topic_id}", ("PATCH",)),
             route_methods,
@@ -71,6 +72,93 @@ class GroupTopicsContractTests(unittest.TestCase):
             ("/v1/rooms/{room_id}/uploads", ("POST",)), route_methods
         )
 
+    def test_capabilities_advertise_global_topics(self) -> None:
+        with patch.object(
+            group_plugin_api,
+            "_store_call",
+            new_callable=AsyncMock,
+            side_effect=[new_id(), 0],
+        ):
+            payload = asyncio.run(group_plugin_api.capabilities())
+        self.assertIn("globalTopics", payload["features"])
+
+    def test_global_topics_are_stably_paginated_and_exclude_archived_scope(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = GroupStore(Path(directory) / "group.db")
+            store.initialize()
+
+            def make_room(name: str) -> dict[str, object]:
+                return store.create_room({
+                    "requestId": new_id(),
+                    "name": name,
+                    "cwd": "",
+                    "agents": [{"profile": "default"}],
+                })
+
+            def make_topic(room_id: str, title: str) -> str:
+                created = store.create_human_message(
+                    room_id,
+                    request_id=new_id(),
+                    client_message_id=new_id(),
+                    topic_id=new_id(),
+                    content=title,
+                    mention_agent_ids=[],
+                )
+                return created["message"]["topicId"]
+
+            room_a = make_room("设计团队")
+            room_b = make_room("发布团队")
+            archived_room = make_room("归档团队")
+            pinned_id = make_topic(room_a["id"], "置顶但较早")
+            recent_a_id = make_topic(room_a["id"], "最近二")
+            archived_topic_id = make_topic(room_a["id"], "归档话题")
+            recent_b_id = make_topic(room_b["id"], "最近一")
+            hidden_room_topic_id = make_topic(archived_room["id"], "隐藏团队话题")
+
+            with store.write_transaction() as connection:
+                values = {
+                    pinned_id: (1, 100.0, 0),
+                    recent_a_id: (0, 300.0, 0),
+                    archived_topic_id: (1, 500.0, 1),
+                    recent_b_id: (0, 300.0, 0),
+                    hidden_room_topic_id: (1, 600.0, 0),
+                }
+                for topic_id, (pinned, updated_at, archived) in values.items():
+                    connection.execute(
+                        "UPDATE group_topics SET pinned = ?, updated_at = ?, archived = ? WHERE id = ?",
+                        (pinned, updated_at, archived, topic_id),
+                    )
+                connection.execute(
+                    "UPDATE group_rooms SET archived = 1 WHERE id = ?",
+                    (archived_room["id"],),
+                )
+
+            first = store.list_global_topics(limit=2, cursor=None)
+            second = store.list_global_topics(limit=2, cursor=first.next_cursor)
+            ordered_recent = sorted(
+                [recent_a_id, recent_b_id], reverse=True
+            )
+            self.assertEqual(
+                [item["id"] for item in first.items],
+                [pinned_id, ordered_recent[0]],
+            )
+            self.assertEqual(
+                [item["id"] for item in second.items],
+                [ordered_recent[1]],
+            )
+            self.assertIsNotNone(first.next_cursor)
+            self.assertIsNone(second.next_cursor)
+            self.assertNotIn(archived_topic_id, {
+                item["id"] for item in [*first.items, *second.items]
+            })
+            self.assertNotIn(hidden_room_topic_id, {
+                item["id"] for item in [*first.items, *second.items]
+            })
+            with self.assertRaisesRegex(ValueError, "cursor is malformed"):
+                store.list_global_topics(limit=2, cursor="not-a-valid-cursor")
+
     def test_group_upload_persists_server_readable_attachment(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             store = GroupStore(Path(directory) / "group.db")

src/client/api/groups.ts

@@ -72,6 +72,17 @@ export async function getPinnedGroupTopics(limit = 100): Promise<GroupTopicPage>
   return { items: values(source.items ?? source.topics ?? payload).map(normalizeGroupTopic).filter(topic => topic.id && topic.roomId), nextCursor: null }
 }
 
+export async function getGlobalGroupTopics(cursor?: string, limit = 50): Promise<GroupTopicPage> {
+  const payload = unwrapData(await apiRequest<unknown>(apiUrl(`${BASE}/topics`, {
+    cursor, limit: Math.max(1, Math.min(100, limit)),
+  })))
+  const source = record(payload)
+  return {
+    items: values(source.items ?? source.topics ?? payload).map(normalizeGroupTopic).filter(topic => topic.id && topic.roomId),
+    nextCursor: typeof source.nextCursor === 'string' ? source.nextCursor : typeof source.next_cursor === 'string' ? source.next_cursor : null,
+  }
+}
+
 export async function updateGroupTopic(roomId: string, topicId: string, input: { title?: string; pinned?: boolean }): Promise<GroupTopicSummary> {
   return normalizeGroupTopic(unwrapData(await apiRequest<unknown>(`${BASE}/rooms/${encodeURIComponent(roomId)}/topics/${encodeURIComponent(topicId)}`, {
     method: 'PATCH', body: { requestId: requestId(), ...input } as JsonValue,

src/server/routes.ts

@@ -976,6 +976,11 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
       search: searchFrom(ctx, ['limit']),
     })
   })
+  router.get('/api/app/groups/topics', async (ctx) => {
+    await proxy(ctx, dependencies.upstream, '/api/plugins/yaoyao/v1/topics', {
+      search: searchFrom(ctx, ['limit', 'cursor']),
+    })
+  })
   router.post('/api/app/groups/rooms', async (ctx) => {
     await proxy(ctx, dependencies.upstream, '/api/plugins/yaoyao/v1/rooms', {
       method: 'POST', requestBody: body(ctx),

tests/server/app.test.ts

@@ -335,6 +335,17 @@ describe('8800 BFF', () => {
     const roomID = '11111111-1111-4111-8111-111111111111'
     const topicID = '22222222-2222-4222-8222-222222222222'
 
+    await request(runtime.app.callback())
+      .get('/api/app/groups/topics?limit=40&cursor=next-global-topic&archived=true&ignored=drop-me')
+      .set('Host', '127.0.0.1:8800')
+      .expect(200)
+    const globalTopics = records.at(-1)!
+    expect(globalTopics.path).toBe('/api/plugins/yaoyao/v1/topics')
+    expect(globalTopics.search.get('limit')).toBe('40')
+    expect(globalTopics.search.get('cursor')).toBe('next-global-topic')
+    expect(globalTopics.search.has('archived')).toBe(false)
+    expect(globalTopics.search.has('ignored')).toBe(false)
+
     await request(runtime.app.callback())
       .get(`/api/app/groups/rooms/${roomID}/topics?limit=25&cursor=next-topic&archived=true&ignored=drop-me`)
       .set('Host', '127.0.0.1:8800')