Add group topic rename, activity, and unread tracking

hermes-plugins/yaoyao/dashboard/group_orchestrator.py

@@ -137,6 +137,14 @@ class _Store(Protocol):
         self, room_id: str, command: Mapping[str, object]
     ) -> dict[str, object]: ...
 
+    def update_topic(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]: ...
+
+    def mark_topic_read(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]: ...
+
     def add_agent(
         self, room_id: str, command: Mapping[str, object]
     ) -> dict[str, object]: ...
@@ -749,6 +757,56 @@ class GroupOrchestrator:
             self.wake()
             return dict(result)
 
+    async def update_topic(
+        self, *, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        _required_string(room_id, "roomId")
+        _required_string(topic_id, "topicId")
+        safe_command = self._public_command(command)
+        request_id = str(safe_command["requestId"])
+        return await self._tracked_control(
+            lambda: self._update_topic(room_id, topic_id, safe_command),
+            task_name=f"yaoyao-group-update-topic-{request_id}",
+            failure_message="YaoYao group topic update failed",
+        )
+
+    async def _update_topic(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        async with self._room_serial(room_id):
+            result = await asyncio.to_thread(
+                self.store.update_topic, room_id, topic_id, command
+            )
+            if not isinstance(result, Mapping):
+                raise GroupOrchestratorError("Updated topic is invalid")
+            self.wake()
+            return dict(result)
+
+    async def mark_topic_read(
+        self, *, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        _required_string(room_id, "roomId")
+        _required_string(topic_id, "topicId")
+        safe_command = self._public_command(command)
+        request_id = str(safe_command["requestId"])
+        return await self._tracked_control(
+            lambda: self._mark_topic_read(room_id, topic_id, safe_command),
+            task_name=f"yaoyao-group-mark-topic-read-{request_id}",
+            failure_message="YaoYao group topic read update failed",
+        )
+
+    async def _mark_topic_read(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        async with self._room_serial(room_id):
+            result = await asyncio.to_thread(
+                self.store.mark_topic_read, room_id, topic_id, command
+            )
+            if not isinstance(result, Mapping):
+                raise GroupOrchestratorError("Topic read response is invalid")
+            self.wake()
+            return dict(result)
+
     async def add_agent(
         self, *, room_id: str, command: Mapping[str, object]
     ) -> dict[str, object]:

hermes-plugins/yaoyao/dashboard/group_plugin_api.py

@@ -117,6 +117,8 @@ RequestIDRequest = _protocol.RequestIDRequest
 SendMessageRequest = _protocol.SendMessageRequest
 UpdateAgentRequest = _protocol.UpdateAgentRequest
 UpdateRoomRequest = _protocol.UpdateRoomRequest
+UpdateTopicRequest = _protocol.UpdateTopicRequest
+MarkTopicReadRequest = _protocol.MarkTopicReadRequest
 
 GroupStore = _store_module.GroupStore
 GroupStoreError = _store_module.GroupStoreError
@@ -1059,6 +1061,28 @@ async def list_topics(
     return {"items": page.items, "nextCursor": page.next_cursor}
 
 
+async def update_topic(
+    room_id: UUID, topic_id: UUID, request: UpdateTopicRequest
+) -> dict[str, object]:
+    return await _runtime_call(
+        "update_topic",
+        room_id=str(room_id),
+        topic_id=str(topic_id),
+        command=_command(request, exclude_unset=True),
+    )
+
+
+async def mark_topic_read(
+    room_id: UUID, topic_id: UUID, request: MarkTopicReadRequest
+) -> dict[str, object]:
+    return await _runtime_call(
+        "mark_topic_read",
+        room_id=str(room_id),
+        topic_id=str(topic_id),
+        command=_command(request, exclude_unset=True),
+    )
+
+
 async def create_room(request: CreateRoomRequest) -> dict[str, object]:
     return await _runtime_call("create_room", command=_create_room_command(request))
 
@@ -1253,6 +1277,14 @@ def _build_router() -> APIRouter:
         methods=["POST"],
     )
     built.add_api_route("/rooms/{room_id}/topics", list_topics, methods=["GET"])
+    built.add_api_route(
+        "/rooms/{room_id}/topics/{topic_id}", update_topic, methods=["PATCH"]
+    )
+    built.add_api_route(
+        "/rooms/{room_id}/topics/{topic_id}/read",
+        mark_topic_read,
+        methods=["PATCH"],
+    )
     built.add_api_route(
         "/rooms/{room_id}/uploads", upload_group_files, methods=["POST"]
     )

hermes-plugins/yaoyao/dashboard/group_protocol.py

@@ -18,7 +18,7 @@ from pydantic import (
 )
 
 
-PROTOCOL_VERSION = 7
+PROTOCOL_VERSION = 8
 MAX_AGENTS_PER_ROOM = 8
 MAX_MESSAGE_BYTES = 64 * 1024
 MAX_TOOL_STATE_BYTES = 256 * 1024
@@ -57,6 +57,7 @@ EVENT_TYPES = frozenset({
     "agent.status",
     "message.upsert",
     "topic.updated",
+    "room.activity",
     "interaction.requested",
     "interaction.resolved",
     "run.updated",
@@ -172,6 +173,24 @@ class UpdateRoomRequest(GroupModel):
         return None if value is None else normalize_max_reply_rounds(value)
 
 
+class UpdateTopicRequest(GroupModel):
+    request_id: UUID = Field(alias="requestId")
+    title: str = Field(min_length=1, max_length=120)
+
+    @field_validator("title")
+    @classmethod
+    def normalize_title(cls, value: str) -> str:
+        normalized = " ".join(value.split()).strip()
+        if not normalized:
+            raise ValueError("title must not be blank")
+        return normalized
+
+
+class MarkTopicReadRequest(GroupModel):
+    request_id: UUID = Field(alias="requestId")
+    through_seq: StrictInt = Field(alias="throughSeq", ge=0)
+
+
 class AddAgentRequest(GroupModel):
     request_id: UUID = Field(alias="requestId")
     profile: str = Field(min_length=1, max_length=100)

hermes-plugins/yaoyao/dashboard/group_store.py

@@ -41,7 +41,7 @@ from .group_protocol import (
 )
 
 
-SCHEMA_VERSION = 7
+SCHEMA_VERSION = 8
 _WAL_RETRY_DELAYS = (0.01, 0.02, 0.04, 0.08)
 EVENT_RETENTION_SECONDS = 30 * 24 * 60 * 60
 MIN_RETAINED_EVENTS = 100_000
@@ -120,6 +120,7 @@ _SCHEMA_STATEMENTS = (
         id TEXT PRIMARY KEY,
         room_id TEXT NOT NULL REFERENCES group_rooms(id) ON DELETE CASCADE,
         title TEXT NOT NULL,
+        last_read_message_seq INTEGER NOT NULL DEFAULT 0,
         created_at REAL NOT NULL,
         updated_at REAL NOT NULL,
         UNIQUE(room_id, id)
@@ -475,13 +476,16 @@ class GroupStore:
             raw_version = "6"
         if raw_version == "6":
             self._migrate_v6_to_v7(connection)
+            raw_version = "7"
+        if raw_version == "7":
+            self._migrate_v7_to_v8(connection)
         self._validated_schema_version(
             self._metadata_value_from_connection(connection, "schema_version")
         )
         self._validated_epoch(
             self._metadata_value_from_connection(connection, "journal_epoch")
         )
-        self._validate_v7_columns(connection)
+        self._validate_v8_columns(connection)
         self._validate_v7_values(connection)
 
     @staticmethod
@@ -1128,11 +1132,27 @@ class GroupStore:
         )
 
     @staticmethod
-    def _validate_v7_columns(connection: sqlite3.Connection) -> None:
+    def _migrate_v7_to_v8(connection: sqlite3.Connection) -> None:
+        """Add one monotonic human-read watermark to every topic."""
+        columns = {
+            str(row["name"])
+            for row in connection.execute("PRAGMA table_info(group_topics)")
+        }
+        if "last_read_message_seq" not in columns:
+            connection.execute(
+                """ALTER TABLE group_topics ADD COLUMN
+                last_read_message_seq INTEGER NOT NULL DEFAULT 0"""
+            )
+        connection.execute(
+            "UPDATE group_meta SET value = '8' WHERE key = 'schema_version'"
+        )
+
+    @staticmethod
+    def _validate_v8_columns(connection: sqlite3.Connection) -> None:
         expected_columns = {
             "group_meta": {"key", "value"},
             "group_rooms": {"id", "name", "cwd", "max_reply_rounds", "created_at", "updated_at", "archived"},
-            "group_topics": {"id", "room_id", "title", "created_at", "updated_at"},
+            "group_topics": {"id", "room_id", "title", "last_read_message_seq", "created_at", "updated_at"},
             "group_agents": {"id", "room_id", "profile", "display_name", "display_name_key", "description", "stored_session_id", "last_context_message_seq", "enabled", "reply_without_mention", "is_host", "model_override", "provider_override", "reasoning_effort_override", "fast_mode_override", "session_config_json", "created_at", "updated_at"},
             "group_agent_topic_state": {"agent_id", "topic_id", "last_context_message_seq", "created_at", "updated_at"},
             "group_messages": {"seq", "id", "room_id", "topic_id", "sender_kind", "sender_id", "sender_name", "root_message_id", "reply_to_message_id", "client_message_id", "content", "reasoning", "tool_state_json", "status", "error", "visible", "created_at", "updated_at"},
@@ -1152,6 +1172,7 @@ class GroupStore:
             ("group_agent_runs", "topic_id"): ("TEXT", 1, None),
             ("group_interactions", "topic_id"): ("TEXT", 1, None),
             ("group_agent_topic_state", "last_context_message_seq"): ("INTEGER", 1, "0"),
+            ("group_topics", "last_read_message_seq"): ("INTEGER", 1, "0"),
         }
         for table, required in expected_columns.items():
             rows = list(connection.execute(f"PRAGMA table_info({table})"))
@@ -1252,6 +1273,15 @@ class GroupStore:
                 WHERE typeof(last_context_message_seq) != 'integer'
                    OR last_context_message_seq < 0
                 LIMIT 1""",
+            """SELECT 1 FROM group_topics AS topic
+                WHERE typeof(topic.last_read_message_seq) != 'integer'
+                   OR topic.last_read_message_seq < 0
+                   OR topic.last_read_message_seq > COALESCE((
+                       SELECT MAX(message.seq) FROM group_messages AS message
+                       WHERE message.topic_id = topic.id
+                         AND message.visible = 1
+                   ), 0)
+                LIMIT 1""",
             """SELECT 1 FROM group_messages AS message
                 LEFT JOIN group_topics AS topic ON topic.id = message.topic_id
                 WHERE topic.id IS NULL OR topic.room_id != message.room_id
@@ -1663,8 +1693,19 @@ class GroupStore:
         params.append(limit + 1)
         with self.connection() as connection:
             rows = connection.execute(query, params).fetchall()
+            items = []
+            for row in rows[:limit]:
+                summary = self._room_summary(row)
+                activity = self._room_activity_from_connection(
+                    connection, str(row["id"])
+                )
+                summary.update({
+                    "activeRunCount": activity["activeRunCount"],
+                    "unreadCount": activity["unreadCount"],
+                    "lastMessage": activity["lastMessage"],
+                })
+                items.append(summary)
         has_more = len(rows) > limit
-        items = [self._room_summary(row) for row in rows[:limit]]
         next_cursor = None
         if has_more and items:
             last = rows[limit - 1]
@@ -1696,7 +1737,16 @@ class GroupStore:
              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
+             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 WHERE topic.room_id = ?"""
         params: list[object] = [canonical_room_id]
         if cursor_values is not None:
@@ -1717,6 +1767,48 @@ class GroupStore:
             )
         return CursorPage(items=items, next_cursor=next_cursor)
 
+    def room_activity(self, room_id: str) -> dict[str, object]:
+        canonical_room_id = self._canonical_uuid(room_id, "roomId")
+        with self.read_transaction() as connection:
+            self._room_detail(connection, canonical_room_id, include_archived=True)
+            return self._room_activity_from_connection(connection, canonical_room_id)
+
+    def _room_activity_from_connection(
+        self, connection: sqlite3.Connection, room_id: str
+    ) -> dict[str, object]:
+        counts = connection.execute(
+            """SELECT
+              (SELECT COUNT(*) FROM group_agent_runs AS run
+               WHERE run.room_id = ?
+                 AND run.status IN ('queued', 'running', 'awaiting_input'))
+                AS active_run_count,
+              (SELECT COUNT(*) FROM group_messages AS message
+               JOIN group_topics AS topic ON topic.id = message.topic_id
+               WHERE message.room_id = ? AND message.visible = 1
+                 AND message.sender_kind = 'agent'
+                 AND (TRIM(message.content) != '' OR TRIM(message.error) != '')
+                 AND message.seq > topic.last_read_message_seq)
+                AS unread_count""",
+            (room_id, room_id),
+        ).fetchone()
+        latest = connection.execute(
+            """SELECT id FROM group_messages
+            WHERE room_id = ? AND visible = 1
+            ORDER BY seq DESC LIMIT 1""",
+            (room_id,),
+        ).fetchone()
+        last_message = None
+        if latest is not None:
+            last_message = self._message_wire(
+                self._message_with_execution_row(connection, latest["id"])
+            )
+        return {
+            "roomId": room_id,
+            "activeRunCount": int(counts["active_run_count"]),
+            "unreadCount": int(counts["unread_count"]),
+            "lastMessage": last_message,
+        }
+
     def update_room(
         self, room_id: str, command: Mapping[str, object]
     ) -> dict[str, object]:
@@ -1762,6 +1854,93 @@ class GroupStore:
             action,
         )
 
+    def update_topic(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        canonical_room_id = self._canonical_uuid(room_id, "roomId")
+        canonical_topic_id = self._canonical_uuid(topic_id, "topicId")
+        command = self._command(command, {"requestId", "title"})
+        request_id = self._command_request_id(command)
+        title = self._topic_title(self._string(command, "title"))
+
+        def action(connection: sqlite3.Connection) -> dict[str, object]:
+            self._active_room(connection, canonical_room_id)
+            self._topic_row(connection, canonical_room_id, canonical_topic_id)
+            now = self._now()
+            connection.execute(
+                """UPDATE group_topics SET title = ?, updated_at = ?
+                WHERE id = ? AND room_id = ?""",
+                (title, now, canonical_topic_id, canonical_room_id),
+            )
+            topic = self._topic_summary_by_id(
+                connection, canonical_room_id, canonical_topic_id
+            )
+            self._append_event(
+                connection, canonical_room_id, "topic.updated", topic,
+                created_at=now,
+            )
+            return topic
+
+        return self.execute_idempotent(
+            "topic.updated",
+            request_id,
+            self._scoped_payload(command, canonical_room_id, canonical_topic_id),
+            action,
+        )
+
+    def mark_topic_read(
+        self, room_id: str, topic_id: str, command: Mapping[str, object]
+    ) -> dict[str, object]:
+        canonical_room_id = self._canonical_uuid(room_id, "roomId")
+        canonical_topic_id = self._canonical_uuid(topic_id, "topicId")
+        command = self._command(command, {"requestId", "throughSeq"})
+        request_id = self._command_request_id(command)
+        through_seq = self._nonnegative_int(command.get("throughSeq"), "throughSeq")
+
+        def action(connection: sqlite3.Connection) -> dict[str, object]:
+            self._active_room(connection, canonical_room_id)
+            topic_row = self._topic_row(
+                connection, canonical_room_id, canonical_topic_id
+            )
+            maximum = int(connection.execute(
+                """SELECT COALESCE(MAX(seq), 0) FROM group_messages
+                WHERE room_id = ? AND topic_id = ? AND visible = 1""",
+                (canonical_room_id, canonical_topic_id),
+            ).fetchone()[0])
+            if through_seq > maximum:
+                raise ValueError("throughSeq exceeds the latest visible message")
+            changed = through_seq > int(topic_row["last_read_message_seq"])
+            if changed:
+                connection.execute(
+                    """UPDATE group_topics SET last_read_message_seq = ?
+                    WHERE id = ? AND room_id = ?""",
+                    (through_seq, canonical_topic_id, canonical_room_id),
+                )
+            topic = self._topic_summary_by_id(
+                connection, canonical_room_id, canonical_topic_id
+            )
+            room = self._room_activity_from_connection(
+                connection, canonical_room_id
+            )
+            if changed:
+                now = self._now()
+                self._append_event(
+                    connection, canonical_room_id, "topic.updated", topic,
+                    created_at=now,
+                )
+                self._append_event(
+                    connection, canonical_room_id, "room.activity", room,
+                    created_at=now,
+                )
+            return {"topic": topic, "room": room}
+
+        return self.execute_idempotent(
+            "topic.read",
+            request_id,
+            self._scoped_payload(command, canonical_room_id, canonical_topic_id),
+            action,
+        )
+
     def archive_room(
         self, room_id: str, command: Mapping[str, object]
     ) -> dict[str, object]:
@@ -5102,6 +5281,14 @@ class GroupStore:
                 VALUES (?, ?, 'topic.updated', ?, ?)""",
                 (epoch, room_id, self._canonical_json(topic), event_time),
             )
+        if event_type in {"message.upsert", "run.updated"}:
+            activity = self._room_activity_from_connection(connection, room_id)
+            connection.execute(
+                """INSERT INTO group_events
+                (epoch, room_id, event_type, payload_json, created_at)
+                VALUES (?, ?, 'room.activity', ?, ?)""",
+                (epoch, room_id, self._canonical_json(activity), event_time),
+            )
         return int(cursor)
 
     def _append_room_updated_summary(
@@ -5236,7 +5423,16 @@ class GroupStore:
              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
+             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
             WHERE topic.id = ? AND topic.room_id = ?""",
             (topic_id, room_id),
@@ -6122,6 +6318,8 @@ class GroupStore:
             "title": row["title"],
             "preview": preview[:_TOPIC_PREVIEW_LENGTH],
             "messageCount": int(row["message_count"]),
+            "unreadCount": int(row["unread_count"]),
+            "latestMessageSeq": int(row["latest_message_seq"]),
             "createdAt": row["created_at"],
             "updatedAt": row["updated_at"],
         }

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

@@ -37,8 +37,9 @@ def new_id() -> str:
 
 class GroupTopicsContractTests(unittest.TestCase):
     def test_protocol_and_router_advertise_topics(self) -> None:
-        self.assertEqual(PROTOCOL.PROTOCOL_VERSION, 7)
+        self.assertEqual(PROTOCOL.PROTOCOL_VERSION, 8)
         self.assertIn("topic.updated", PROTOCOL.EVENT_TYPES)
+        self.assertIn("room.activity", PROTOCOL.EVENT_TYPES)
         request = PROTOCOL.SendMessageRequest.model_validate({
             "requestId": new_id(),
             "clientMessageId": new_id(),
@@ -53,6 +54,19 @@ class GroupTopicsContractTests(unittest.TestCase):
             if hasattr(route, "methods")
         }
         self.assertIn(("/v1/rooms/{room_id}/topics", ("GET",)), route_methods)
+        self.assertIn(
+            ("/v1/rooms/{room_id}/topics/{topic_id}", ("PATCH",)),
+            route_methods,
+        )
+        self.assertIn(
+            ("/v1/rooms/{room_id}/topics/{topic_id}/read", ("PATCH",)),
+            route_methods,
+        )
+        read_request = PROTOCOL.MarkTopicReadRequest.model_validate({
+            "requestId": new_id(),
+            "throughSeq": 12,
+        })
+        self.assertEqual(read_request.through_seq, 12)
         self.assertIn(
             ("/v1/rooms/{room_id}/uploads", ("POST",)), route_methods
         )
@@ -169,6 +183,219 @@ class GroupTopicsContractTests(unittest.TestCase):
             self.assertEqual(updated["maxReplyRounds"], 5)
             self.assertEqual(updated["agents"], [before_agent])
 
+    def test_topic_title_update_is_idempotent_and_emits_topic_event(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = GroupStore(Path(directory) / "group.db")
+            store.initialize()
+            room = store.create_room({
+                "requestId": new_id(),
+                "name": "话题群",
+                "cwd": "",
+                "agents": [{"profile": "default"}],
+            })
+            created = store.create_human_message(
+                room["id"],
+                request_id=new_id(),
+                client_message_id=new_id(),
+                topic_id=new_id(),
+                content="原始标题",
+                mention_agent_ids=[],
+            )
+            topic_id = created["message"]["topicId"]
+            request_id = new_id()
+            before = store.latest_cursor()
+
+            renamed = store.update_topic(
+                room["id"],
+                topic_id,
+                {"requestId": request_id, "title": "  新的\n话题标题  "},
+            )
+            repeated = store.update_topic(
+                room["id"],
+                topic_id,
+                {"requestId": request_id, "title": "  新的\n话题标题  "},
+            )
+
+            self.assertEqual(renamed["title"], "新的 话题标题")
+            self.assertEqual(repeated, renamed)
+            events = store.events_after(before, 10)
+            self.assertEqual(len(events), 1)
+            self.assertEqual(events[0]["eventType"], "topic.updated")
+            self.assertEqual(events[0]["payload"]["title"], "新的 话题标题")
+
+    def test_schema_v7_migrates_topic_read_position_without_data_loss(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "group.db"
+            store = GroupStore(path)
+            store.initialize()
+            room = store.create_room({
+                "requestId": new_id(),
+                "name": "迁移群",
+                "cwd": "",
+                "agents": [{"profile": "default"}],
+            })
+            created = store.create_human_message(
+                room["id"],
+                request_id=new_id(),
+                client_message_id=new_id(),
+                topic_id=new_id(),
+                content="保留消息",
+                mention_agent_ids=[],
+            )
+            topic_id = created["message"]["topicId"]
+            with store.connection() as connection:
+                original_message_count = connection.execute(
+                    "SELECT COUNT(*) FROM group_messages"
+                ).fetchone()[0]
+            with store.write_transaction() as connection:
+                columns = {
+                    row["name"]
+                    for row in connection.execute("PRAGMA table_info(group_topics)")
+                }
+                if "last_read_message_seq" in columns:
+                    connection.execute(
+                        "ALTER TABLE group_topics DROP COLUMN last_read_message_seq"
+                    )
+                connection.execute(
+                    "UPDATE group_meta SET value = '7' WHERE key = 'schema_version'"
+                )
+
+            migrated = GroupStore(path)
+            migrated.initialize()
+
+            self.assertEqual(migrated.schema_version(), 8)
+            with migrated.connection() as connection:
+                topic = connection.execute(
+                    "SELECT * FROM group_topics WHERE id = ?", (topic_id,)
+                ).fetchone()
+                message_count = connection.execute(
+                    "SELECT COUNT(*) FROM group_messages"
+                ).fetchone()[0]
+            self.assertEqual(topic["last_read_message_seq"], 0)
+            self.assertEqual(message_count, original_message_count)
+
+    def test_room_activity_counts_active_runs_and_unread_agent_messages(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = GroupStore(Path(directory) / "group.db")
+            store.initialize()
+            room = store.create_room({
+                "requestId": new_id(),
+                "name": "活动群",
+                "cwd": "",
+                "agents": [{"profile": "default"}],
+            })
+            agent = room["agents"][0]
+            topic_id = new_id()
+            created = store.create_human_message(
+                room["id"],
+                request_id=new_id(),
+                client_message_id=new_id(),
+                topic_id=topic_id,
+                content="用户问题",
+                mention_agent_ids=[agent["id"]],
+            )
+            trigger = created["message"]
+            run = created["runs"][0]
+            with store.write_transaction() as connection:
+                connection.execute(
+                    "UPDATE group_agent_runs SET status = 'running' WHERE id = ?",
+                    (run["id"],),
+                )
+                visible_ids = [new_id(), new_id()]
+                for index, message_id in enumerate(visible_ids, start=2):
+                    connection.execute(
+                        """INSERT INTO group_messages
+                        (id, room_id, topic_id, sender_kind, sender_id, sender_name,
+                         root_message_id, content, status, visible, created_at, updated_at)
+                        VALUES (?, ?, ?, 'agent', ?, '夭夭', ?, ?, 'completed', 1, ?, ?)""",
+                        (
+                            message_id,
+                            room["id"],
+                            topic_id,
+                            agent["id"],
+                            trigger["rootMessageId"],
+                            f"回复 {index}",
+                            index,
+                            index,
+                        ),
+                    )
+                connection.execute(
+                    """INSERT INTO group_messages
+                    (id, room_id, topic_id, sender_kind, sender_id, sender_name,
+                     root_message_id, content, status, visible, created_at, updated_at)
+                    VALUES (?, ?, ?, 'agent', ?, '夭夭', ?, '隐藏', 'completed', 0, 4, 4)""",
+                    (
+                        new_id(), room["id"], topic_id, agent["id"],
+                        trigger["rootMessageId"],
+                    ),
+                )
+
+            activity = store.room_activity(room["id"])
+
+            self.assertEqual(activity["activeRunCount"], 1)
+            self.assertEqual(activity["unreadCount"], 2)
+            self.assertEqual(activity["lastMessage"]["id"], visible_ids[-1])
+
+    def test_mark_topic_read_is_monotonic_idempotent_and_emits_activity(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = GroupStore(Path(directory) / "group.db")
+            store.initialize()
+            room = store.create_room({
+                "requestId": new_id(),
+                "name": "已读群",
+                "cwd": "",
+                "agents": [{"profile": "default"}],
+            })
+            agent = room["agents"][0]
+            topic_id = new_id()
+            trigger = store.create_human_message(
+                room["id"], request_id=new_id(), client_message_id=new_id(),
+                topic_id=topic_id, content="问题", mention_agent_ids=[],
+            )["message"]
+            with store.write_transaction() as connection:
+                connection.execute(
+                    """INSERT INTO group_messages
+                    (id, room_id, topic_id, sender_kind, sender_id, sender_name,
+                     root_message_id, content, status, visible, created_at, updated_at)
+                    VALUES (?, ?, ?, 'agent', ?, '夭夭', ?, '回复', 'completed', 1, 2, 2)""",
+                    (
+                        new_id(), room["id"], topic_id, agent["id"],
+                        trigger["rootMessageId"],
+                    ),
+                )
+                through_seq = connection.execute(
+                    "SELECT MAX(seq) FROM group_messages WHERE topic_id = ?",
+                    (topic_id,),
+                ).fetchone()[0]
+            request_id = new_id()
+            before = store.latest_cursor()
+
+            first = store.mark_topic_read(
+                room["id"], topic_id,
+                {"requestId": request_id, "throughSeq": through_seq},
+            )
+            repeated = store.mark_topic_read(
+                room["id"], topic_id,
+                {"requestId": request_id, "throughSeq": through_seq},
+            )
+            stale = store.mark_topic_read(
+                room["id"], topic_id,
+                {"requestId": new_id(), "throughSeq": through_seq - 1},
+            )
+
+            self.assertEqual(first, repeated)
+            self.assertEqual(stale["room"]["unreadCount"], 0)
+            events = store.events_after(before, 10)
+            self.assertEqual(
+                [event["eventType"] for event in events],
+                ["topic.updated", "room.activity"],
+            )
+            with self.assertRaises(ValueError):
+                store.mark_topic_read(
+                    room["id"], topic_id,
+                    {"requestId": new_id(), "throughSeq": through_seq + 100},
+                )
+
     def test_topics_page_filter_compatibility_and_event_order(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             store = GroupStore(Path(directory) / "group.db")
@@ -197,9 +424,12 @@ class GroupTopicsContractTests(unittest.TestCase):
                 [
                     "message.upsert",
                     "topic.updated",
+                    "room.activity",
                     "message.upsert",
                     "topic.updated",
+                    "room.activity",
                     "run.updated",
+                    "room.activity",
                 ],
             )
             self.assertEqual(events[0]["payload"]["topicId"], topic_a)
@@ -211,11 +441,16 @@ class GroupTopicsContractTests(unittest.TestCase):
                     "title",
                     "preview",
                     "messageCount",
+                    "unreadCount",
+                    "latestMessageSeq",
                     "createdAt",
                     "updatedAt",
                 },
             )
             self.assertEqual(events[1]["payload"]["id"], topic_a)
+            self.assertEqual(events[2]["payload"]["roomId"], room["id"])
+            self.assertEqual(events[2]["payload"]["activeRunCount"], 0)
+            self.assertEqual(events[-1]["payload"]["activeRunCount"], 1)
 
             second = store.create_human_message(
                 room["id"],
@@ -1257,7 +1492,7 @@ class GroupTopicsContractTests(unittest.TestCase):
 
             store = GroupStore(path)
             store.initialize()
-            self.assertEqual(store.schema_version(), 7)
+            self.assertEqual(store.schema_version(), 8)
             with store.connection() as migrated:
                 room = migrated.execute("SELECT * FROM group_rooms").fetchone()
                 agent = migrated.execute("SELECT * FROM group_agents").fetchone()

src/client/api/groups.ts

@@ -1,5 +1,5 @@
 import type {
-  GroupAgent, GroupCapabilities, GroupInteraction, GroupMessagePage, GroupRoomDetail, GroupRoomPage, GroupRun, GroupTopicPage, JsonValue, UploadReference,
+  GroupAgent, GroupCapabilities, GroupInteraction, GroupMessagePage, GroupRoomDetail, GroupRoomPage, GroupRun, GroupTopicPage, GroupTopicSummary, JsonValue, UploadReference,
 } from '@shared/types'
 import { apiRequest, apiUrl, unwrapData } from './client'
 import {
@@ -40,6 +40,12 @@ export async function getGroupTopics(roomId: string, cursor?: string, limit = 50
   }
 }
 
+export async function updateGroupTopic(roomId: string, topicId: string, input: { title: string }): Promise<GroupTopicSummary> {
+  return normalizeGroupTopic(unwrapData(await apiRequest<unknown>(`${BASE}/rooms/${encodeURIComponent(roomId)}/topics/${encodeURIComponent(topicId)}`, {
+    method: 'PATCH', body: { requestId: requestId(), ...input } as JsonValue,
+  })))
+}
+
 export interface AgentSeed {
   profile: string
   displayName: string

src/server/routes.ts

@@ -674,6 +674,18 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
       search: searchFrom(ctx, ['limit', 'cursor']),
     })
   })
+  router.patch('/api/app/groups/rooms/:roomID/topics/:topicID', async (ctx) => {
+    const topicID = canonicalUUID(ctx.params.topicID, 'topic ID')
+    await proxy(ctx, dependencies.upstream, groupPath(ctx, `/topics/${topicID}`), {
+      method: 'PATCH', requestBody: body(ctx),
+    })
+  })
+  router.patch('/api/app/groups/rooms/:roomID/topics/:topicID/read', async (ctx) => {
+    const topicID = canonicalUUID(ctx.params.topicID, 'topic ID')
+    await proxy(ctx, dependencies.upstream, groupPath(ctx, `/topics/${topicID}/read`), {
+      method: 'PATCH', requestBody: body(ctx),
+    })
+  })
   router.get('/api/app/groups/rooms/:roomID/messages', async (ctx) => {
     await proxy(ctx, dependencies.upstream, groupPath(ctx, '/messages'), {
       search: searchFrom(ctx, ['topicId', 'beforeSeq', 'afterSeq', 'limit']),

tests/server/app.test.ts

@@ -328,6 +328,36 @@ describe('8800 BFF', () => {
     expect(topics.search.get('cursor')).toBe('next-topic')
     expect(topics.search.has('ignored')).toBe(false)
 
+    const bootstrap = await request(runtime.app.callback())
+      .get('/api/app/bootstrap')
+      .set('Host', '127.0.0.1:8800')
+      .expect(200)
+    await request(runtime.app.callback())
+      .patch(`/api/app/groups/rooms/${roomID}/topics/${topicID}`)
+      .set('Host', '127.0.0.1:8800')
+      .set('Origin', 'http://127.0.0.1:8800')
+      .set('Cookie', cookieHeader(bootstrap))
+      .set('X-CSRF-Token', bootstrap.body.csrfToken)
+      .send({ requestId: '33333333-3333-4333-8333-333333333333', title: '新标题' })
+      .expect(200)
+    const renamed = records.at(-1)!
+    expect(renamed.method).toBe('PATCH')
+    expect(renamed.path).toBe(`/api/plugins/yaoyao/v1/rooms/${roomID}/topics/${topicID}`)
+    expect(renamed.body).toEqual({ requestId: '33333333-3333-4333-8333-333333333333', title: '新标题' })
+
+    await request(runtime.app.callback())
+      .patch(`/api/app/groups/rooms/${roomID}/topics/${topicID}/read`)
+      .set('Host', '127.0.0.1:8800')
+      .set('Origin', 'http://127.0.0.1:8800')
+      .set('Cookie', cookieHeader(bootstrap))
+      .set('X-CSRF-Token', bootstrap.body.csrfToken)
+      .send({ requestId: '44444444-4444-4444-8444-444444444444', throughSeq: 42 })
+      .expect(200)
+    const markedRead = records.at(-1)!
+    expect(markedRead.method).toBe('PATCH')
+    expect(markedRead.path).toBe(`/api/plugins/yaoyao/v1/rooms/${roomID}/topics/${topicID}/read`)
+    expect(markedRead.body).toEqual({ requestId: '44444444-4444-4444-8444-444444444444', throughSeq: 42 })
+
     await request(runtime.app.callback())
       .get(`/api/app/groups/rooms/${roomID}/messages?topicId=${topicID}&beforeSeq=40&limit=20`)
       .set('Host', '127.0.0.1:8800')