diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json index 9b355da35e..b4c455f6eb 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json @@ -6,8 +6,8 @@ "commands": { "session.bridge.denied": "This operation requires instance operator permission for both sessions in the same configuration.", "session.watch.failed": "Cannot start the watch. Check the UMO and active watch limit.", - "session.watch.usage": "Usage: /session watch [listener|this] [seconds]. Omit the listener or write this for the current session. Duration is 60–864000 seconds, default 43200.", - "session.watch.ttl_invalid": "Watch duration must be between 60 and 864000 seconds.", + "session.watch.usage": "Usage: /session watch [listener|this] [seconds]. Omit the listener or write this for the current session. Duration is 1–864000 seconds, default 43200.", + "session.watch.ttl_invalid": "Watch duration must be between 1 and 864000 seconds.", "session.watch.ok": "Watching {umo} into {source} for {seconds} seconds. New incoming messages will be shared with everyone in the listening session.", "session.unwatch.ok": "Watch stopped.", "session.unwatch.missing": "No matching watch owned by you.", @@ -15,10 +15,17 @@ "session.watches.empty": "You have no active watches in {source}.", "session.watches.body": "Your active watches:\n{watches}", "session.watches.usage": "Usage: /session watches [listener|this]. Omit the listener or write this for the current session.", - "session.connect.usage": "Usage: /session connect . Omit the UMO to show the current unbounded link.", + "session.connect.usage": "Usage: /session connect . Omit the UMO to show the current unbounded link. Extra tokens such as a duration are not accepted.", "session.connect.failed": "Cannot connect. Check the UMO and active link limit.", "session.connect.ok": "Connected to {umo}. Incoming messages will be forwarded without expiry. /send without a UMO uses this target.", "session.connect.status": "Connected to {umo}. Incoming messages are forwarded without expiry. /send without a UMO uses this target.", + "session.links.empty": "No visible session-bridge rules.", + "session.links.body": "Session-bridge rules:\n{links}", + "session.links.ttl_unbounded": "unbounded", + "session.links.ttl_seconds": "{seconds}s", + "session.unlink.usage": "Usage: /session unlink . rule_id is 12 lowercase hex characters.", + "session.unlink.ok": "Removed that session-bridge rule.", + "session.unlink.missing": "No matching rule to remove.", "session.disconnect.ok": "Session disconnected.", "session.disconnect.missing": "No unbounded link in this session.", "session.send.invalid": "Usage: /send , or /send after /session connect. Keep the command header in the message text.", diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json index 5f846b1f79..6ec412bbbc 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json @@ -6,8 +6,8 @@ "commands": { "session.bridge.denied": "此操作需要同一配置下两个会话的实例管理员权限。", "session.watch.failed": "无法开始监听，请检查 UMO 和当前监听数量。", - "session.watch.usage": "用法：/session watch [监听会话|this] <被监听会话> [秒数]。监听会话可省略或写 this 表示当前会话。时长 60–864000 秒，默认 43200。", - "session.watch.ttl_invalid": "监听时长须在 60 到 864000 秒之间。", + "session.watch.usage": "用法：/session watch [监听会话|this] <被监听会话> [秒数]。监听会话可省略或写 this 表示当前会话。时长 1–864000 秒，默认 43200。", + "session.watch.ttl_invalid": "监听时长须在 1 到 864000 秒之间。", "session.watch.ok": "已在 {source} 监听 {umo}，有效期 {seconds} 秒。后续收到的消息将对监听会话所有成员可见。", "session.unwatch.ok": "已停止监听。", "session.unwatch.missing": "没有找到属于你的该项监听。", @@ -15,10 +15,17 @@ "session.watches.empty": "你在 {source} 没有有效的监听。", "session.watches.body": "你的有效监听：\n{watches}", "session.watches.usage": "用法：/session watches [监听会话|this]。监听会话可省略或写 this 表示当前会话。", - "session.connect.usage": "用法：/session connect 。省略 UMO 可查看当前无期限连接。", + "session.connect.usage": "用法：/session connect 。省略 UMO 可查看当前无期限连接。不接受秒数或其他额外参数。", "session.connect.failed": "无法连接，请检查 UMO 和当前连接数量。", "session.connect.ok": "已连接到 {umo}。将无限期转发其新消息。不带 UMO 的 /send 会发往该会话。", "session.connect.status": "当前已连接到 {umo}。将无限期转发其新消息。不带 UMO 的 /send 会发往该会话。", + "session.links.empty": "没有可见的会话桥接规则。", + "session.links.body": "会话桥接规则：\n{links}", + "session.links.ttl_unbounded": "无期限", + "session.links.ttl_seconds": "{seconds}秒", + "session.unlink.usage": "用法：/session unlink 。rule_id 为 12 位小写十六进制。", + "session.unlink.ok": "已撤销该会话桥接规则。", + "session.unlink.missing": "没有找到可撤销的规则。", "session.disconnect.ok": "已断开会话连接。", "session.disconnect.missing": "当前会话没有无期限连接。", "session.send.invalid": "用法：/send <文字或附件>，或先 /session connect 后再 /send。请保留消息文字中的指令头。", diff --git a/astrbot/builtin_stars/builtin_commands/commands/session.py b/astrbot/builtin_stars/builtin_commands/commands/session.py index 3b9aac15cb..52ceceb18a 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/session.py +++ b/astrbot/builtin_stars/builtin_commands/commands/session.py @@ -50,6 +50,17 @@ def parse_unwatch_spec(spec: str, current_umo: str) -> tuple[str, str]: return _resolve_listener(parts[0], current_umo), parts[1] +def parse_unlink_spec(spec: str) -> str: + """Parse `/session unlink `.""" + parts = spec.split() + if len(parts) != 1: + raise ValueError("Invalid unlink arguments") + rule_id = parts[0] + if len(rule_id) != 12 or any(char not in "0123456789abcdef" for char in rule_id): + raise ValueError("Invalid unlink arguments") + return rule_id + + def parse_watches_spec(spec: str, current_umo: str) -> str: """Parse `/session watches [listener|this]`.""" parts = spec.split() @@ -332,6 +343,53 @@ async def connect(self, event: AstrMessageEvent, target: str) -> None: umo=item.target_umo, ) + async def links(self, event: AstrMessageEvent) -> None: + """List watch/connect edges visible to the current actor.""" + try: + items = await self.context.bridges._manager.list_links(event) + except PermissionError: + await reply_i18n(self.context, event, "session.bridge.denied") + return + if not items: + await reply_i18n(self.context, event, "session.links.empty") + return + lines = [] + for watch, kind in items: + if watch.expires_at is None: + ttl = await self.context.i18n.t(event, "session.links.ttl_unbounded") + else: + ttl = await self.context.i18n.t( + event, + "session.links.ttl_seconds", + seconds=watch.remaining_seconds, + ) + lines.append( + f"{watch.rule_id} {kind} {watch.source_umo} -> {watch.target_umo} ({ttl})" + ) + await reply_i18n( + self.context, + event, + "session.links.body", + links="\n".join(lines), + ) + + async def unlink(self, event: AstrMessageEvent, spec: str) -> None: + """Remove a watch or connect by public id.""" + try: + rule_id = parse_unlink_spec(spec) + removed = await self.context.bridges._manager.unlink(event, rule_id) + except ValueError: + await reply_i18n(self.context, event, "session.unlink.usage") + return + except PermissionError: + await reply_i18n(self.context, event, "session.bridge.denied") + return + await reply_i18n( + self.context, + event, + "session.unlink.ok" if removed else "session.unlink.missing", + ) + async def disconnect(self, event: AstrMessageEvent) -> None: """Drop the unbounded link owned by the current actor.""" try: diff --git a/astrbot/builtin_stars/builtin_commands/main.py b/astrbot/builtin_stars/builtin_commands/main.py index 4f4d7da24a..31de1e160e 100644 --- a/astrbot/builtin_stars/builtin_commands/main.py +++ b/astrbot/builtin_stars/builtin_commands/main.py @@ -148,6 +148,20 @@ async def session_disconnect(self, event: AstrMessageEvent) -> None: """Drop the unbounded session link for the current session.""" await self.session_c.disconnect(event) + @filter.permission("session.read") + @session.command("links") + async def session_links(self, event: AstrMessageEvent) -> None: + """List visible session-bridge watch and connect rules.""" + await self.session_c.links(event) + + @filter.permission("session.read") + @session.command("unlink") + async def session_unlink( + self, event: AstrMessageEvent, spec: GreedyStr = GreedyStr("") + ) -> None: + """Remove a session-bridge rule by id.""" + await self.session_c.unlink(event, spec) + @filter.permission("session.block") @session.command("block") async def session_block( diff --git a/astrbot/core/core_lifecycle.py b/astrbot/core/core_lifecycle.py index 4c132ea963..a25715f2a7 100644 --- a/astrbot/core/core_lifecycle.py +++ b/astrbot/core/core_lifecycle.py @@ -571,6 +571,7 @@ async def _initialize(self) -> None: self.execution_context.session_bridge_manager.observe ) await self.platform_manager.initialize() + await self.execution_context.session_bridge_manager.restore() # 初始化关闭控制面板的事件 self.dashboard_shutdown_event = asyncio.Event() diff --git a/astrbot/core/db/po/__init__.py b/astrbot/core/db/po/__init__.py index 32b46f7f3d..65d67c80d7 100644 --- a/astrbot/core/db/po/__init__.py +++ b/astrbot/core/db/po/__init__.py @@ -39,6 +39,7 @@ ) from astrbot.core.db.po.preferences import Preference from astrbot.core.db.po.projects import ChatUIProject, SessionProjectRelation +from astrbot.core.db.po.session_bridge import SessionBridgeRule from astrbot.core.db.po.sessions import PlatformSession, UmoAlias from astrbot.core.db.po.statistics import PlatformStat, ProviderStat from astrbot.core.db.po.webchat import WebChatThread @@ -77,6 +78,7 @@ "PlatformStat", "Preference", "ProviderStat", + "SessionBridgeRule", "SessionProjectRelation", "UmoAlias", "WebChatThread", diff --git a/astrbot/core/db/po/registry.py b/astrbot/core/db/po/registry.py index e43cb10f63..74346def44 100644 --- a/astrbot/core/db/po/registry.py +++ b/astrbot/core/db/po/registry.py @@ -35,6 +35,7 @@ ) from astrbot.core.db.po.preferences import Preference from astrbot.core.db.po.projects import ChatUIProject, SessionProjectRelation +from astrbot.core.db.po.session_bridge import SessionBridgeRule from astrbot.core.db.po.sessions import PlatformSession, UmoAlias from astrbot.core.db.po.statistics import PlatformStat, ProviderStat from astrbot.core.db.po.webchat import WebChatThread @@ -73,6 +74,7 @@ PlatformStat, Preference, ProviderStat, + SessionBridgeRule, SessionProjectRelation, UmoAlias, WebChatThread, diff --git a/astrbot/core/db/po/session_bridge.py b/astrbot/core/db/po/session_bridge.py new file mode 100644 index 0000000000..72496f6eb7 --- /dev/null +++ b/astrbot/core/db/po/session_bridge.py @@ -0,0 +1,34 @@ +from sqlmodel import JSON, Column, Field, SQLModel, UniqueConstraint + +from astrbot.core.db.po.mixins import TimestampMixin + + +class SessionBridgeRule(TimestampMixin, SQLModel, table=True): + """Persisted directed edge for a session-bridge watch, connect, or later pair.""" + + __tablename__ = "session_bridge_rules" # type: ignore + + rule_id: str = Field(primary_key=True, max_length=12, nullable=False) + subject_id: str = Field(nullable=False, max_length=512, index=True) + source_umo: str = Field(nullable=False, max_length=512) + target_umo: str = Field(nullable=False, max_length=512) + source_config_id: str = Field(nullable=False, max_length=128, index=True) + target_config_id: str = Field(nullable=False, max_length=128, index=True) + kind: str = Field(nullable=False, max_length=32, index=True) + expires_at: int | None = Field(default=None) + header: bool = Field(default=True, nullable=False) + pair_id: str | None = Field(default=None, max_length=12) + match: dict = Field(default_factory=dict, sa_type=JSON, nullable=False) + except_: dict = Field( + default_factory=dict, + sa_column=Column("except", JSON, nullable=False), + ) + + __table_args__ = ( + UniqueConstraint( + "subject_id", + "source_umo", + "target_umo", + name="uix_session_bridge_direction", + ), + ) diff --git a/astrbot/core/db/protocols.py b/astrbot/core/db/protocols.py index 3313dc8bf9..e7c1fc1ace 100644 --- a/astrbot/core/db/protocols.py +++ b/astrbot/core/db/protocols.py @@ -37,6 +37,7 @@ PlatformStat, Preference, ProviderStat, + SessionBridgeRule, SessionProjectRelation, UmoAlias, WebChatThread, @@ -911,6 +912,82 @@ async def get_umo_aliases( ) -> list[UmoAlias]: ... +@runtime_checkable +class SessionBridgeStore(Protocol): + """Operations for persisted session-bridge directed edges.""" + + async def insert_session_bridge_rule( + self, + *, + subject_id: str, + source_umo: str, + target_umo: str, + source_config_id: str, + target_config_id: str, + kind: str, + expires_at: int | None = None, + header: bool = True, + pair_id: str | None = None, + match: dict | None = None, + except_: dict | None = None, + rule_id: str | None = None, + ) -> SessionBridgeRule: + return T.cast("SessionBridgeRule", None) + + async def get_session_bridge_rule(self, rule_id: str) -> SessionBridgeRule | None: + return None + + async def get_session_bridge_rule_by_direction( + self, + subject_id: str, + source_umo: str, + target_umo: str, + ) -> SessionBridgeRule | None: + return None + + async def list_session_bridge_rules(self) -> list[SessionBridgeRule]: + return [] + + async def list_session_bridge_rules_by_subject( + self, subject_id: str + ) -> list[SessionBridgeRule]: + return [] + + async def list_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ) -> list[SessionBridgeRule]: + return [] + + async def list_session_bridge_rules_touching_config( + self, config_id: str + ) -> list[SessionBridgeRule]: + return [] + + async def update_session_bridge_rule( + self, + rule_id: str, + *, + expires_at: int | None = None, + kind: str | None = None, + target_umo: str | None = None, + source_config_id: str | None = None, + target_config_id: str | None = None, + header: bool | None = None, + pair_id: str | None = None, + match: dict | None = None, + except_: dict | None = None, + ) -> SessionBridgeRule | None: + return None + + async def delete_session_bridge_rule(self, rule_id: str) -> None: + return None + + async def delete_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ) -> None: + return None + + @runtime_checkable class ChatProjectStore(Protocol): """Operations for ChatUI projects and their session relations.""" @@ -1056,6 +1133,7 @@ class DashboardStore( "PlatformSessionStore", "PluginRuntimeStore", "PreferenceStore", + "SessionBridgeStore", "SessionManagementStore", "StatisticsStore", "StatisticsSessionStore", diff --git a/astrbot/core/db/sqlite.py b/astrbot/core/db/sqlite.py index c9adb614fc..fe0e89b7fe 100644 --- a/astrbot/core/db/sqlite.py +++ b/astrbot/core/db/sqlite.py @@ -13,6 +13,7 @@ from astrbot.core.db.stores.personas import PersonaStoreMixin from astrbot.core.db.stores.preferences import PreferenceStoreMixin from astrbot.core.db.stores.projects import ChatProjectStoreMixin +from astrbot.core.db.stores.session_bridge import SessionBridgeStoreMixin from astrbot.core.db.stores.sessions import PlatformSessionStoreMixin from astrbot.core.db.stores.statistics import StatisticsStoreMixin from astrbot.core.db.stores.webchat import WebChatThreadStoreMixin @@ -33,6 +34,7 @@ class SQLiteDatabase( CommandStoreMixin, CronStoreMixin, PlatformSessionStoreMixin, + SessionBridgeStoreMixin, UmoAliasStoreMixin, ChatProjectStoreMixin, BaseDatabase, diff --git a/astrbot/core/db/stores/__init__.py b/astrbot/core/db/stores/__init__.py index 990844bbf3..5a86e6993e 100644 --- a/astrbot/core/db/stores/__init__.py +++ b/astrbot/core/db/stores/__init__.py @@ -11,6 +11,7 @@ from astrbot.core.db.stores.personas import PersonaStoreMixin from astrbot.core.db.stores.preferences import PreferenceStoreMixin from astrbot.core.db.stores.projects import ChatProjectStoreMixin +from astrbot.core.db.stores.session_bridge import SessionBridgeStoreMixin from astrbot.core.db.stores.sessions import PlatformSessionStoreMixin from astrbot.core.db.stores.statistics import StatisticsStoreMixin from astrbot.core.db.stores.webchat import WebChatThreadStoreMixin @@ -29,6 +30,7 @@ "PersonaStoreMixin", "PlatformSessionStoreMixin", "PreferenceStoreMixin", + "SessionBridgeStoreMixin", "StatisticsStoreMixin", "UmoAliasStoreMixin", "WebChatThreadStoreMixin", diff --git a/astrbot/core/db/stores/session_bridge.py b/astrbot/core/db/stores/session_bridge.py new file mode 100644 index 0000000000..fdc1c152be --- /dev/null +++ b/astrbot/core/db/stores/session_bridge.py @@ -0,0 +1,277 @@ +import secrets + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession +from sqlmodel import col, delete, select, update + +from astrbot.core.db.po import SessionBridgeRule +from astrbot.core.db.stores.mixin import DatabaseStoreMixin, store_session + +_RULE_ID_ATTEMPTS = 8 +_RULE_FIELD_NOT_SET = object() + + +class SessionBridgeStoreMixin(DatabaseStoreMixin): + async def insert_session_bridge_rule( + self, + *, + subject_id: str, + source_umo: str, + target_umo: str, + source_config_id: str, + target_config_id: str, + kind: str, + expires_at: int | None = None, + header: bool = True, + pair_id: str | None = None, + match: dict | None = None, + except_: dict | None = None, + rule_id: str | None = None, + ) -> SessionBridgeRule: + """Insert one directed session-bridge edge, allocating a rule id if needed. + + Args: + subject_id: Authorization subject that owns the edge. + source_umo: Listening session that receives forwarded messages. + target_umo: Observed session whose inbound traffic is forwarded. + source_config_id: Config id of the listening session at write time. + target_config_id: Config id of the observed session at write time. + kind: Edge kind. S1 writes ``watch`` or ``connect``. + expires_at: Wall-clock UTC unix seconds, or None for unbounded edges. + header: Whether deliveries include a source header. S1 writes True. + pair_id: Paired-edge id. S1 writes None. + match: Inclusive filter document. S1 writes ``{}``. + except_: Exclusive filter document. S1 writes ``{}``. + rule_id: Optional 12-hex id. Allocated on collision retry when omitted. + + Returns: + The inserted row. + + Raises: + IntegrityError: Direction uniqueness failed, or a caller-supplied + ``rule_id`` collided. + RuntimeError: An id could not be allocated after retries. + """ + match_payload = {} if match is None else match + except_payload = {} if except_ is None else except_ + last_error: IntegrityError | None = None + for _ in range(_RULE_ID_ATTEMPTS): + allocated = rule_id or secrets.token_hex(6) + try: + async with store_session(self) as session: + session: AsyncSession + async with session.begin(): + row = SessionBridgeRule( + rule_id=allocated, + subject_id=subject_id, + source_umo=source_umo, + target_umo=target_umo, + source_config_id=source_config_id, + target_config_id=target_config_id, + kind=kind, + expires_at=expires_at, + header=header, + pair_id=pair_id, + match=match_payload, + except_=except_payload, + ) + session.add(row) + await session.flush() + await session.refresh(row) + return row + except IntegrityError as exc: + last_error = exc + if rule_id is not None: + raise + orig = str(getattr(exc, "orig", exc)) + if "rule_id" not in orig: + raise + raise RuntimeError( + "Unable to allocate a session bridge rule id" + ) from last_error + + async def get_session_bridge_rule(self, rule_id: str) -> SessionBridgeRule | None: + """Return one rule by public id.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule).where( + col(SessionBridgeRule.rule_id) == rule_id + ) + ) + return result.scalar_one_or_none() + + async def get_session_bridge_rule_by_direction( + self, + subject_id: str, + source_umo: str, + target_umo: str, + ) -> SessionBridgeRule | None: + """Return the edge for one owner and direction, if any.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule).where( + col(SessionBridgeRule.subject_id) == subject_id, + col(SessionBridgeRule.source_umo) == source_umo, + col(SessionBridgeRule.target_umo) == target_umo, + ) + ) + return result.scalar_one_or_none() + + async def list_session_bridge_rules(self) -> list[SessionBridgeRule]: + """Return every persisted session-bridge edge.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule).order_by( + col(SessionBridgeRule.created_at), + col(SessionBridgeRule.rule_id), + ) + ) + return list(result.scalars().all()) + + async def list_session_bridge_rules_by_subject( + self, subject_id: str + ) -> list[SessionBridgeRule]: + """Return edges owned by one authorization subject.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule) + .where(col(SessionBridgeRule.subject_id) == subject_id) + .order_by( + col(SessionBridgeRule.created_at), + col(SessionBridgeRule.rule_id), + ) + ) + return list(result.scalars().all()) + + async def list_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ) -> list[SessionBridgeRule]: + """Return connect edges for one owner and listening session.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule) + .where( + col(SessionBridgeRule.subject_id) == subject_id, + col(SessionBridgeRule.source_umo) == source_umo, + col(SessionBridgeRule.kind) == "connect", + ) + .order_by( + col(SessionBridgeRule.created_at), + col(SessionBridgeRule.rule_id), + ) + ) + return list(result.scalars().all()) + + async def list_session_bridge_rules_touching_config( + self, config_id: str + ) -> list[SessionBridgeRule]: + """Return edges whose source or target config id equals ``config_id``.""" + async with store_session(self) as session: + session: AsyncSession + result = await session.execute( + select(SessionBridgeRule) + .where( + (col(SessionBridgeRule.source_config_id) == config_id) + | (col(SessionBridgeRule.target_config_id) == config_id) + ) + .order_by( + col(SessionBridgeRule.created_at), + col(SessionBridgeRule.rule_id), + ) + ) + return list(result.scalars().all()) + + async def update_session_bridge_rule( + self, + rule_id: str, + *, + expires_at: int | None | object = _RULE_FIELD_NOT_SET, + kind: str | None | object = _RULE_FIELD_NOT_SET, + target_umo: str | None | object = _RULE_FIELD_NOT_SET, + source_config_id: str | None | object = _RULE_FIELD_NOT_SET, + target_config_id: str | None | object = _RULE_FIELD_NOT_SET, + header: bool | None | object = _RULE_FIELD_NOT_SET, + pair_id: str | None | object = _RULE_FIELD_NOT_SET, + match: dict | None | object = _RULE_FIELD_NOT_SET, + except_: dict | None | object = _RULE_FIELD_NOT_SET, + ) -> SessionBridgeRule | None: + """Patch selected columns on one rule. + + Args: + rule_id: Public edge id. + expires_at: Wall-clock expiry, or None to clear. + kind: Replacement kind. + target_umo: Replacement observed session. + source_config_id: Replacement listener config id. + target_config_id: Replacement observed-session config id. + header: Replacement header flag. + pair_id: Replacement pair id. + match: Replacement inclusive filter document. + except_: Replacement exclusive filter document. + + Returns: + The updated row, or None when ``rule_id`` is missing. + """ + updates: dict = {} + for key, val in { + "expires_at": expires_at, + "kind": kind, + "target_umo": target_umo, + "source_config_id": source_config_id, + "target_config_id": target_config_id, + "header": header, + "pair_id": pair_id, + "match": match, + "except_": except_, + }.items(): + if val is _RULE_FIELD_NOT_SET: + continue + updates[key] = val + if not updates: + return await self.get_session_bridge_rule(rule_id) + async with store_session(self) as session: + session: AsyncSession + async with session.begin(): + await session.execute( + update(SessionBridgeRule) + .where(col(SessionBridgeRule.rule_id) == rule_id) + .values(**updates) + .execution_options(synchronize_session="fetch") + ) + result = await session.execute( + select(SessionBridgeRule).where( + col(SessionBridgeRule.rule_id) == rule_id + ) + ) + return result.scalar_one_or_none() + + async def delete_session_bridge_rule(self, rule_id: str) -> None: + """Delete one rule by public id.""" + async with store_session(self) as session: + session: AsyncSession + async with session.begin(): + await session.execute( + delete(SessionBridgeRule).where( + col(SessionBridgeRule.rule_id) == rule_id + ) + ) + + async def delete_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ) -> None: + """Delete every connect edge for one owner and listening session.""" + async with store_session(self) as session: + session: AsyncSession + async with session.begin(): + await session.execute( + delete(SessionBridgeRule).where( + col(SessionBridgeRule.subject_id) == subject_id, + col(SessionBridgeRule.source_umo) == source_umo, + col(SessionBridgeRule.kind) == "connect", + ) + ) diff --git a/astrbot/core/execution_context.py b/astrbot/core/execution_context.py index eb6970ab59..d3807d1521 100644 --- a/astrbot/core/execution_context.py +++ b/astrbot/core/execution_context.py @@ -2,7 +2,7 @@ import logging from asyncio import Queue, QueueFull from collections.abc import Awaitable, Callable -from typing import TYPE_CHECKING, Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol, cast from astrbot.core.agent.follow_up import FollowUpCoordinator from astrbot.core.agent.hooks import BaseAgentRunHooks @@ -17,7 +17,7 @@ from astrbot.core.computer.computer_client import ComputerRuntime from astrbot.core.config.astrbot_config import AstrBotConfig from astrbot.core.conversation_mgr import ConversationManager -from astrbot.core.db.protocols import PluginRuntimeStore +from astrbot.core.db.protocols import PluginRuntimeStore, SessionBridgeStore from astrbot.core.exceptions import ProviderNotFoundError from astrbot.core.group_sender_concurrency import GroupOutboundGate from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager @@ -249,6 +249,7 @@ def __init__( ), get_locale=self._session_locale, get_platform_family=self._adapter_family, + store=cast(SessionBridgeStore, db), ) """Runtime-owned expiring cross-session watch state.""" self.conversation_manager = conversation_manager diff --git a/astrbot/core/platform/session_bridge.py b/astrbot/core/platform/session_bridge.py index 6450271a2d..21414b6700 100644 --- a/astrbot/core/platform/session_bridge.py +++ b/astrbot/core/platform/session_bridge.py @@ -3,14 +3,15 @@ from __future__ import annotations import asyncio +import math from collections import OrderedDict from collections.abc import Awaitable, Callable -from dataclasses import dataclass, replace -from time import monotonic +from dataclasses import replace +from time import time from typing import TYPE_CHECKING from astrbot import logger -from astrbot.core.auth.models import AuthContext, Resource, Subject +from astrbot.core.auth.models import AuthContext, Resource, Role, Subject from astrbot.core.message.message_event_result import MessageChain from .message_delivery import plan_message_delivery @@ -26,41 +27,21 @@ from .message_renderers import render_source_header from .message_session import MessageSession from .send_result import DeliveryAttempt, DeliveryReceipt, PlatformSendResult +from .session_bridge_state import GrantKey, SessionBridgeState, SessionWatch, WatchGrant -MIN_WATCH_TTL_SECONDS = 60 +MIN_WATCH_TTL_SECONDS = 1 DEFAULT_WATCH_TTL_SECONDS = 12 * 60 * 60 MAX_WATCH_TTL_SECONDS = 10 * 24 * 60 * 60 if TYPE_CHECKING: from astrbot.core.auth.service import AuthorizationService + from astrbot.core.db.po.session_bridge import SessionBridgeRule + from astrbot.core.db.protocols import SessionBridgeStore from astrbot.core.file_token_service import FileTokenService from .astr_message_event import AstrMessageEvent -@dataclass(frozen=True, slots=True) -class SessionWatch: - """One watch or unbounded link owned by a trusted authorization subject.""" - - source_umo: str - target_umo: str - subject_id: str - expires_at: float | None - - @property - def remaining_seconds(self) -> int: - if self.expires_at is None: - return 0 - return max(0, int(self.expires_at - monotonic())) - - -@dataclass(frozen=True, slots=True) -class _WatchGrant: - watch: SessionWatch - subject: Subject - context: AuthContext - - class SessionBridgeManager: """Manage bounded watches, rechecking authority before every delivery.""" @@ -77,6 +58,7 @@ def __init__( get_callback_base: Callable[[str], str] | None = None, get_locale: Callable[[str], Awaitable[str]] | None = None, get_platform_family: Callable[[str], str] | None = None, + store: SessionBridgeStore, default_ttl_seconds: int = DEFAULT_WATCH_TTL_SECONDS, max_watches_per_subject: int = 16, ) -> None: @@ -92,9 +74,7 @@ def __init__( MAX_WATCH_TTL_SECONDS, max(MIN_WATCH_TTL_SECONDS, default_ttl_seconds) ) self._max_watches_per_subject = max(1, max_watches_per_subject) - self._watches: dict[tuple[str, str, str], _WatchGrant] = {} - self._links: dict[tuple[str, str], _WatchGrant] = {} - self._expiry_tasks: dict[tuple[str, str, str], asyncio.Task] = {} + self._state = SessionBridgeState(store) self._forwarded: OrderedDict[tuple[str, str, str], None] = OrderedDict() self._message_ids: OrderedDict[tuple[str, str, str], str] = OrderedDict() self._lock = asyncio.Lock() @@ -124,6 +104,11 @@ async def _authorize( if not decision.allowed: raise PermissionError("Session operation is not authorized") + def _config_id(self, umo: str) -> str: + if self._get_config_id is None: + raise PermissionError("Authorization context is unavailable") + return self._get_config_id(umo) + def _ttl_seconds(self, ttl_seconds: int | None) -> int: if ttl_seconds is None: return self._default_ttl_seconds @@ -153,27 +138,23 @@ async def watch( raise ValueError("Target adapter is unavailable") if not self._get_capabilities(listener).proactive: raise ValueError("Source adapter cannot receive forwarded messages") - now = monotonic() + now = time() key = (subject.id, listener, target) async with self._lock: - expired = self._purge(now) - owned_count = sum(item[0] == subject.id for item in self._watches) - if ( - key not in self._watches - and owned_count >= self._max_watches_per_subject - ): - raise ValueError("Watch limit exceeded") - if key not in self._watches and len(self._watches) >= 1024: - raise ValueError("Runtime watch limit exceeded") - previous = self._expiry_tasks.pop(key, None) - if previous is not None: - previous.cancel() - watch = SessionWatch(listener, target, subject.id, now + ttl) - grant = _WatchGrant(watch, subject, context) - self._watches[key] = grant - self._arm_expiry(key, grant) + expired = await self._state.purge(now) + grant = await self._state.save_watch( + subject=subject, + context=context, + source_umo=listener, + target_umo=target, + source_config_id=self._config_id(listener), + target_config_id=self._config_id(target), + expires_at=math.ceil(now + ttl), + max_per_subject=self._max_watches_per_subject, + ) + self._state.arm_expiry(key, grant, self._expire_watch) await self._notify_expired_watches(expired) - return watch + return grant.watch async def unwatch( self, @@ -187,11 +168,7 @@ async def unwatch( listener = (source_umo or event.unified_msg_origin).strip() key = (subject.id, listener, target_umo.strip()) async with self._lock: - removed = self._watches.pop(key, None) is not None - task = self._expiry_tasks.pop(key, None) - if task is not None: - task.cancel() - return removed + return await self._state.drop_watch(subject.id, key[1], key[2]) async def list_watches( self, @@ -202,12 +179,8 @@ async def list_watches( subject, _ = self._actor(event) listener = (source_umo or event.unified_msg_origin).strip() async with self._lock: - expired = self._purge(monotonic()) - items = tuple( - grant.watch - for key, grant in self._watches.items() - if key[:2] == (subject.id, listener) - ) + expired = await self._state.purge(time()) + items = self._state.watches_for(subject.id, listener) await self._notify_expired_watches(expired) return items @@ -228,32 +201,113 @@ async def connect( raise ValueError("Target adapter is unavailable") if not self._get_capabilities(listener).proactive: raise ValueError("Source adapter cannot receive forwarded messages") - key = (subject.id, listener) async with self._lock: - owned_count = sum(item[0] == subject.id for item in self._links) - if key not in self._links and owned_count >= self._max_watches_per_subject: - raise ValueError("Watch limit exceeded") - if key not in self._links and len(self._links) >= 1024: - raise ValueError("Runtime watch limit exceeded") - watch = SessionWatch(listener, target, subject.id, None) - self._links[key] = _WatchGrant(watch, subject, context) - return watch + grant = await self._state.save_connect( + subject=subject, + context=context, + source_umo=listener, + target_umo=target, + source_config_id=self._config_id(listener), + target_config_id=self._config_id(target), + max_per_subject=self._max_watches_per_subject, + ) + return grant.watch async def disconnect(self, event: AstrMessageEvent) -> bool: """Remove the unbounded link owned by the current actor in this session.""" subject, _ = self._actor(event) - key = (subject.id, event.unified_msg_origin.strip()) + listener = event.unified_msg_origin.strip() async with self._lock: - return self._links.pop(key, None) is not None + return await self._state.drop_connect(subject.id, listener) async def connection(self, event: AstrMessageEvent) -> SessionWatch | None: """Return the unbounded link for the current actor in this session.""" subject, _ = self._actor(event) - key = (subject.id, event.unified_msg_origin.strip()) + listener = event.unified_msg_origin.strip() async with self._lock: - grant = self._links.get(key) + grant = self._state.connect_for(subject.id, listener) return grant.watch if grant is not None else None + async def list_links( + self, event: AstrMessageEvent + ) -> tuple[tuple[SessionWatch, str], ...]: + """List watch/connect edges visible to the current actor.""" + subject, _ = self._actor(event) + role = await self._operator_role(event) + current_config = self._config_id(event.unified_msg_origin) + now = time() + async with self._lock: + expired = await self._state.purge(now) + await self._notify_expired_watches(expired) + own = await self._state.stored_rules_by_subject(subject.id) + if role in {Role.ROOT, Role.OPERATOR}: + rows = await self._state.list_stored_rules() + elif role is Role.INSTANCE_OPERATOR: + extra = await self._state.stored_rules_touching_config(current_config) + merged = {row.rule_id: row for row in own} + merged.update({row.rule_id: row for row in extra}) + rows = list(merged.values()) + else: + rows = own + items: list[tuple[SessionWatch, str]] = [] + for row in rows: + if row.kind not in {"watch", "connect"}: + continue + if ( + row.kind == "watch" + and row.expires_at is not None + and row.expires_at <= now + ): + continue + items.append((self._watch_from_row(row), row.kind)) + return tuple(items) + + async def unlink(self, event: AstrMessageEvent, rule_id: str) -> bool: + """Remove a watch or connect by public id under creator or operator rules.""" + subject, _ = self._actor(event) + stored = await self._state.stored_rule(rule_id) + if stored is None or stored.kind == "pair": + if stored is not None: + raise ValueError("Pair edges cannot be unlinked") + return False + if stored.subject_id != subject.id: + role = await self._operator_role(event) + current_config = self._config_id(event.unified_msg_origin) + allowed = role in {Role.ROOT, Role.OPERATOR} or ( + role is Role.INSTANCE_OPERATOR + and current_config in {stored.source_config_id, stored.target_config_id} + ) + if not allowed: + raise PermissionError("Session operation is not authorized") + async with self._lock: + await self._state.delete_rule(rule_id) + return True + + def _watch_from_row(self, row: SessionBridgeRule) -> SessionWatch: + expires_at = None if row.expires_at is None else float(row.expires_at) + return SessionWatch( + row.source_umo, + row.target_umo, + row.subject_id, + expires_at, + row.rule_id, + ) + + async def _operator_role(self, event: AstrMessageEvent) -> Role | None: + if self._authorization is None: + return None + subject, context = self._actor(event) + umo = event.unified_msg_origin.strip() + decision = await self._authorization.authorize( + subject, + "session.watch", + Resource.session(self._config_id(umo), umo), + context, + ) + if not decision.allowed: + return None + return decision.effective_role + async def send( self, event: AstrMessageEvent, @@ -288,24 +342,16 @@ async def check_authority() -> None: locale=await self._locale_for(target_umo), ) - def _store_for(self, watch: SessionWatch) -> dict: - return self._links if watch.expires_at is None else self._watches + def _store_key(self, watch: SessionWatch) -> GrantKey: + return self._state.store_key(watch) - def _store_key(self, watch: SessionWatch) -> tuple[str, ...]: - if watch.expires_at is None: - return (watch.subject_id, watch.source_umo) - return (watch.subject_id, watch.source_umo, watch.target_umo) + def _grant_active(self, key: GrantKey, grant: WatchGrant, now: float) -> bool: + return self._state.grant_active(key, grant, now) - def _grant_active(self, key: tuple, grant: _WatchGrant, now: float) -> bool: - watch = grant.watch - if self._store_for(watch).get(key) is not grant: - return False - return watch.expires_at is None or watch.expires_at > now - - async def _check_watch(self, grant: _WatchGrant) -> None: + async def _check_watch(self, grant: WatchGrant) -> None: watch = grant.watch key = self._store_key(watch) - if not self._grant_active(key, grant, monotonic()): + if not self._grant_active(key, grant, time()): raise PermissionError("Watch is no longer active") await self._authorize( grant.subject, grant.context, watch.source_umo, "session.watch" @@ -320,20 +366,14 @@ async def observe(self, envelope: MessageEnvelope) -> None: return origin = envelope.source_umo async with self._lock: - expired = self._purge(monotonic()) - watches = tuple( - (key, grant) - for key, grant in self._watches.items() - if grant.watch.target_umo == origin - ) + tuple( - (key, grant) - for key, grant in self._links.items() - if grant.watch.target_umo == origin - ) + expired = await self._state.purge(time()) + watches = self._state.grants_observing(origin) await self._notify_expired_watches(expired) for key, grant in watches: watch = grant.watch try: + if not self._get_capabilities(watch.source_umo).available: + continue await self._authorize( grant.subject, grant.context, watch.source_umo, "session.watch" ) @@ -341,7 +381,7 @@ async def observe(self, envelope: MessageEnvelope) -> None: grant.subject, grant.context, watch.target_umo, "session.watch" ) async with self._lock: - if not self._grant_active(key, grant, monotonic()): + if not self._grant_active(key, grant, time()): continue dedup = (watch.source_umo, origin, envelope.source_message_id or "") if envelope.source_message_id: @@ -377,9 +417,7 @@ async def observe(self, envelope: MessageEnvelope) -> None: ) except PermissionError: async with self._lock: - store = self._store_for(grant.watch) - if store.get(key) is grant: - store.pop(key) + await self._state.delete_rule(grant.watch.rule_id) except asyncio.CancelledError: raise except Exception: @@ -553,57 +591,23 @@ async def _submit( self._message_ids.popitem(last=False) return receipt - def _purge(self, now: float) -> tuple[_WatchGrant, ...]: - expired: list[_WatchGrant] = [] - for key, grant in tuple(self._watches.items()): - if grant.watch.expires_at is not None and grant.watch.expires_at <= now: - self._watches.pop(key) - expired.append(grant) - return tuple(expired) - - def _arm_expiry(self, key: tuple[str, str, str], grant: _WatchGrant) -> None: - previous = self._expiry_tasks.pop(key, None) - if previous is not None: - previous.cancel() - task = asyncio.create_task( - self._expire_watch(key, grant), - name=f"session-watch-expire:{key[1]}:{key[2]}", - ) - self._expiry_tasks[key] = task - - def _done(done: asyncio.Task) -> None: - if self._expiry_tasks.get(key) is done: - self._expiry_tasks.pop(key, None) - if done.cancelled(): - return - try: - exc = done.exception() - except Exception: - return - if exc is not None: - logger.warning("Session watch expiry task failed") - - task.add_done_callback(_done) - - async def _expire_watch( - self, key: tuple[str, str, str], grant: _WatchGrant - ) -> None: + async def _expire_watch(self, key: GrantKey, grant: WatchGrant) -> None: expires_at = grant.watch.expires_at if expires_at is None: return - delay = max(0.0, expires_at - monotonic()) + delay = max(0.0, expires_at - time()) await asyncio.sleep(delay) async with self._lock: - if self._watches.get(key) is not grant: + if self._state.get(key) is not grant: return - self._watches.pop(key, None) + await self._state.delete_rule(grant.watch.rule_id) await self._notify_expired(grant) - async def _notify_expired_watches(self, expired: tuple[_WatchGrant, ...]) -> None: + async def _notify_expired_watches(self, expired: tuple[WatchGrant, ...]) -> None: for grant in expired: await self._notify_expired(grant) - async def _notify_expired(self, grant: _WatchGrant) -> None: + async def _notify_expired(self, grant: WatchGrant) -> None: locale = await self._locale_for(grant.watch.source_umo) text = localize(locale, "astrbot.msg.watch.expired", umo=grant.watch.target_umo) try: @@ -616,13 +620,100 @@ async def _notify_expired(self, grant: _WatchGrant) -> None: except Exception: logger.warning("Session watch expiry notice failed") + async def restore(self) -> None: + """Reload persisted edges, dropping expired or unauthorized rows.""" + rows = await self._state.list_stored_rules() + now = time() + expired: list[WatchGrant] = [] + async with self._lock: + for row in rows: + if row.kind not in {"watch", "connect"}: + continue + if ( + row.kind == "watch" + and row.expires_at is not None + and row.expires_at <= now + ): + await self._state.discard_stored_rule(row.rule_id) + try: + expired.append(self._grant_for_notice(row)) + except ValueError: + logger.warning("Session watch expiry notice skipped") + continue + grant = await self._restore_row(row) + if grant is None: + continue + if grant.kind == "connect": + existing = self._state.connect_for( + grant.watch.subject_id, grant.watch.source_umo + ) + if ( + existing is not None + and existing.watch.rule_id != grant.watch.rule_id + ): + await self._state.delete_rule(existing.watch.rule_id) + self._state.index_grant(grant) + self._state.arm_expiry( + self._state.store_key(grant.watch), grant, self._expire_watch + ) + await self._notify_expired_watches(tuple(expired)) + + def _grant_for_notice(self, row: SessionBridgeRule) -> WatchGrant: + subject = Subject.from_id(row.subject_id) + context = AuthContext(subject=subject, source="im", authenticated=True) + return self._state.grant_from_row(row, subject, context) + + async def _restore_row(self, row: SessionBridgeRule) -> WatchGrant | None: + try: + subject = Subject.from_id(row.subject_id) + source = "webchat" if subject.kind == "dashboard-account" else "im" + source_config_id = self._config_id(row.source_umo) + target_config_id = self._config_id(row.target_umo) + context = AuthContext( + subject=subject, + source=source, + config_id=source_config_id, + authenticated=True, + origin_session_resource_id=Resource.session( + source_config_id, row.source_umo + ).id, + ) + await self._authorize(subject, context, row.source_umo, "session.watch") + context = AuthContext( + subject=subject, + source=source, + config_id=target_config_id, + authenticated=True, + origin_session_resource_id=Resource.session( + source_config_id, row.source_umo + ).id, + ) + await self._authorize(subject, context, row.target_umo, "session.watch") + row = await self._state.persist_config_ids( + row, source_config_id, target_config_id + ) + except PermissionError, ValueError: + await self._state.discard_stored_rule(row.rule_id) + return None + except Exception: + logger.warning("Session bridge restore skipped a rule") + return None + grant_context = AuthContext( + subject=subject, + source=source, + config_id=source_config_id, + authenticated=True, + origin_session_resource_id=Resource.session( + source_config_id, row.source_umo + ).id, + ) + return self._state.grant_from_row(row, subject, grant_context) + async def terminate(self) -> None: """Cancel expiry tasks and drop in-memory watches.""" async with self._lock: - tasks = list(self._expiry_tasks.values()) - self._expiry_tasks.clear() - self._watches.clear() - self._links.clear() + tasks = self._state.take_expiry_tasks() + self._state.clear_grants() self._forwarded.clear() self._message_ids.clear() for task in tasks: diff --git a/astrbot/core/platform/session_bridge_state.py b/astrbot/core/platform/session_bridge_state.py new file mode 100644 index 0000000000..4dbf43dc31 --- /dev/null +++ b/astrbot/core/platform/session_bridge_state.py @@ -0,0 +1,406 @@ +"""Live grant index and SQLite IO for session-bridge directed edges.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable, Coroutine +from dataclasses import dataclass +from time import time +from typing import TYPE_CHECKING + +from astrbot import logger +from astrbot.core.auth.models import AuthContext, Subject + +if TYPE_CHECKING: + from astrbot.core.db.po.session_bridge import SessionBridgeRule + from astrbot.core.db.protocols import SessionBridgeStore + +PAIR_OCCUPIED = "Direction is occupied by a pair" +GrantKey = tuple[str, str, str] + + +@dataclass(frozen=True, slots=True) +class SessionWatch: + """One watch or unbounded link owned by a trusted authorization subject.""" + + source_umo: str + target_umo: str + subject_id: str + expires_at: float | None + rule_id: str + + @property + def remaining_seconds(self) -> int: + if self.expires_at is None: + return 0 + return max(0, int(self.expires_at - time())) + + +@dataclass(frozen=True, slots=True) +class WatchGrant: + """Trusted actor snapshot paired with one live watch or connect.""" + + watch: SessionWatch + subject: Subject + context: AuthContext + kind: str + source_config_id: str + target_config_id: str + + +ExpireWatch = Callable[[GrantKey, WatchGrant], Coroutine[object, object, None]] + + +class SessionBridgeState: + """Unified direction index, connect 1:1 map, expiry tasks, and store IO.""" + + def __init__(self, store: SessionBridgeStore) -> None: + self._store = store + self._grants: dict[GrantKey, WatchGrant] = {} + self._connect_by_listener: dict[tuple[str, str], GrantKey] = {} + self._expiry_tasks: dict[GrantKey, asyncio.Task[None]] = {} + + def count_kind(self, subject_id: str, kind: str) -> int: + return sum( + 1 + for grant in self._grants.values() + if grant.watch.subject_id == subject_id and grant.kind == kind + ) + + def total_kind(self, kind: str) -> int: + return sum(1 for grant in self._grants.values() if grant.kind == kind) + + def get(self, key: GrantKey) -> WatchGrant | None: + return self._grants.get(key) + + def connect_for(self, subject_id: str, source_umo: str) -> WatchGrant | None: + key = self._connect_by_listener.get((subject_id, source_umo)) + return None if key is None else self._grants.get(key) + + def watches_for(self, subject_id: str, source_umo: str) -> tuple[SessionWatch, ...]: + return tuple( + grant.watch + for key, grant in self._grants.items() + if key[:2] == (subject_id, source_umo) and grant.kind == "watch" + ) + + def grants_observing(self, origin: str) -> tuple[tuple[GrantKey, WatchGrant], ...]: + return tuple( + (key, grant) + for key, grant in self._grants.items() + if grant.watch.target_umo == origin + ) + + def store_key(self, watch: SessionWatch) -> GrantKey: + return (watch.subject_id, watch.source_umo, watch.target_umo) + + def grant_active(self, key: GrantKey, grant: WatchGrant, now: float) -> bool: + if self._grants.get(key) is not grant: + return False + return grant.watch.expires_at is None or grant.watch.expires_at > now + + async def purge(self, now: float) -> tuple[WatchGrant, ...]: + expired: list[WatchGrant] = [] + for key, grant in tuple(self._grants.items()): + expires_at = grant.watch.expires_at + if expires_at is not None and expires_at <= now: + await self._store.delete_session_bridge_rule(grant.watch.rule_id) + self._drop_memory(key) + task = self._expiry_tasks.pop(key, None) + if task is not None: + task.cancel() + expired.append(grant) + return tuple(expired) + + def arm_expiry( + self, + key: GrantKey, + grant: WatchGrant, + expire: ExpireWatch, + ) -> None: + previous = self._expiry_tasks.pop(key, None) + if previous is not None: + previous.cancel() + if grant.watch.expires_at is None: + return + task = asyncio.create_task( + expire(key, grant), + name=f"session-watch-expire:{key[1]}:{key[2]}", + ) + self._expiry_tasks[key] = task + + def _done(done: asyncio.Task[None]) -> None: + if self._expiry_tasks.get(key) is done: + self._expiry_tasks.pop(key, None) + if done.cancelled(): + return + try: + exc = done.exception() + except Exception: + return + if exc is not None: + logger.warning("Session watch expiry task failed") + + task.add_done_callback(_done) + + def take_expiry_tasks(self) -> list[asyncio.Task[None]]: + tasks = list(self._expiry_tasks.values()) + self._expiry_tasks.clear() + return tasks + + def clear_grants(self) -> None: + self._grants.clear() + self._connect_by_listener.clear() + + async def save_watch( + self, + *, + subject: Subject, + context: AuthContext, + source_umo: str, + target_umo: str, + source_config_id: str, + target_config_id: str, + expires_at: int, + max_per_subject: int, + ) -> WatchGrant: + key = (subject.id, source_umo, target_umo) + stored = await self._store.get_session_bridge_rule_by_direction( + subject.id, source_umo, target_umo + ) + if stored is not None and stored.kind == "pair": + raise ValueError(PAIR_OCCUPIED) + if stored is not None and stored.kind == "watch": + updated = await self._store.update_session_bridge_rule( + stored.rule_id, + expires_at=expires_at, + source_config_id=source_config_id, + target_config_id=target_config_id, + ) + row = updated or stored + grant = self._grant_from_row(row, subject, context) + self._index(grant) + return grant + if self.count_kind(subject.id, "watch") >= max_per_subject: + raise ValueError("Watch limit exceeded") + if self.total_kind("watch") >= 1024: + raise ValueError("Runtime watch limit exceeded") + if stored is not None: + await self._delete_row(stored.rule_id, key) + row = await self._store.insert_session_bridge_rule( + subject_id=subject.id, + source_umo=source_umo, + target_umo=target_umo, + source_config_id=source_config_id, + target_config_id=target_config_id, + kind="watch", + expires_at=expires_at, + ) + grant = self._grant_from_row(row, subject, context) + self._index(grant) + return grant + + async def save_connect( + self, + *, + subject: Subject, + context: AuthContext, + source_umo: str, + target_umo: str, + source_config_id: str, + target_config_id: str, + max_per_subject: int, + ) -> WatchGrant: + key = (subject.id, source_umo, target_umo) + stored = await self._store.get_session_bridge_rule_by_direction( + subject.id, source_umo, target_umo + ) + if stored is not None and stored.kind == "pair": + raise ValueError(PAIR_OCCUPIED) + if stored is not None and stored.kind == "connect": + row = await self.persist_config_ids( + stored, source_config_id, target_config_id + ) + grant = self._grant_from_row(row, subject, context) + self._index(grant) + return grant + listener = (subject.id, source_umo) + had_connect = listener in self._connect_by_listener + if not had_connect: + if self.count_kind(subject.id, "connect") >= max_per_subject: + raise ValueError("Watch limit exceeded") + if self.total_kind("connect") >= 1024: + raise ValueError("Runtime watch limit exceeded") + occupant = stored + await self._store.delete_session_bridge_connects_for_listener( + subject.id, source_umo + ) + self._drop_connects_for_listener(subject.id, source_umo) + if occupant is not None: + await self._delete_row(occupant.rule_id, key) + row = await self._store.insert_session_bridge_rule( + subject_id=subject.id, + source_umo=source_umo, + target_umo=target_umo, + source_config_id=source_config_id, + target_config_id=target_config_id, + kind="connect", + expires_at=None, + ) + grant = self._grant_from_row(row, subject, context) + self._index(grant) + return grant + + async def drop_watch( + self, subject_id: str, source_umo: str, target_umo: str + ) -> bool: + key = (subject_id, source_umo, target_umo) + grant = self._grants.get(key) + if grant is None or grant.kind != "watch": + stored = await self._store.get_session_bridge_rule_by_direction( + subject_id, source_umo, target_umo + ) + if stored is None or stored.kind != "watch": + return False + await self._delete_row(stored.rule_id, key) + return True + await self._delete_row(grant.watch.rule_id, key) + return True + + async def drop_connect(self, subject_id: str, source_umo: str) -> bool: + grant = self.connect_for(subject_id, source_umo) + if grant is None: + rows = await self._store.list_session_bridge_connects_for_listener( + subject_id, source_umo + ) + if not rows: + return False + for row in rows: + await self._delete_row( + row.rule_id, (row.subject_id, row.source_umo, row.target_umo) + ) + return True + await self._delete_row(grant.watch.rule_id, self.store_key(grant.watch)) + return True + + async def delete_rule(self, rule_id: str) -> WatchGrant | None: + for key, grant in tuple(self._grants.items()): + if grant.watch.rule_id == rule_id: + await self._delete_row(rule_id, key) + return grant + stored = await self._store.get_session_bridge_rule(rule_id) + if stored is None: + return None + await self._delete_row( + rule_id, (stored.subject_id, stored.source_umo, stored.target_umo) + ) + return None + + async def list_stored_rules(self) -> list[SessionBridgeRule]: + return await self._store.list_session_bridge_rules() + + async def stored_rule(self, rule_id: str) -> SessionBridgeRule | None: + return await self._store.get_session_bridge_rule(rule_id) + + async def stored_rules_by_subject(self, subject_id: str) -> list[SessionBridgeRule]: + return await self._store.list_session_bridge_rules_by_subject(subject_id) + + async def stored_rules_touching_config( + self, config_id: str + ) -> list[SessionBridgeRule]: + return await self._store.list_session_bridge_rules_touching_config(config_id) + + async def discard_stored_rule(self, rule_id: str) -> None: + await self._store.delete_session_bridge_rule(rule_id) + + async def persist_config_ids( + self, + row: SessionBridgeRule, + source_config_id: str, + target_config_id: str, + ) -> SessionBridgeRule: + if ( + row.source_config_id == source_config_id + and row.target_config_id == target_config_id + ): + return row + updated = await self._store.update_session_bridge_rule( + row.rule_id, + source_config_id=source_config_id, + target_config_id=target_config_id, + ) + return updated or row + + def index_grant(self, grant: WatchGrant) -> None: + self._index(grant) + + def grant_from_row( + self, + row: SessionBridgeRule, + subject: Subject, + context: AuthContext, + ) -> WatchGrant: + return self._grant_from_row(row, subject, context) + + def _grant_from_row( + self, + row: SessionBridgeRule, + subject: Subject, + context: AuthContext, + ) -> WatchGrant: + expires_at = None if row.expires_at is None else float(row.expires_at) + watch = SessionWatch( + row.source_umo, + row.target_umo, + row.subject_id, + expires_at, + row.rule_id, + ) + return WatchGrant( + watch, + subject, + context, + row.kind, + row.source_config_id, + row.target_config_id, + ) + + def _index(self, grant: WatchGrant) -> None: + key = self.store_key(grant.watch) + self._grants[key] = grant + listener = (grant.watch.subject_id, grant.watch.source_umo) + if grant.kind == "connect": + previous_key = self._connect_by_listener.get(listener) + if previous_key is not None and previous_key != key: + self._drop_memory(previous_key) + self._connect_by_listener[listener] = key + elif self._connect_by_listener.get(listener) == key: + self._connect_by_listener.pop(listener, None) + + def _drop_memory(self, key: GrantKey) -> WatchGrant | None: + grant = self._grants.pop(key, None) + listener = key[:2] + if self._connect_by_listener.get(listener) == key: + self._connect_by_listener.pop(listener, None) + return grant + + def _drop_connects_for_listener(self, subject_id: str, source_umo: str) -> None: + listener = (subject_id, source_umo) + key = self._connect_by_listener.pop(listener, None) + if key is not None: + self._grants.pop(key, None) + for grant_key, grant in tuple(self._grants.items()): + if ( + grant.kind == "connect" + and grant.watch.subject_id == subject_id + and grant.watch.source_umo == source_umo + ): + self._grants.pop(grant_key, None) + + async def _delete_row(self, rule_id: str, key: GrantKey) -> None: + await self._store.delete_session_bridge_rule(rule_id) + self._drop_memory(key) + task = self._expiry_tasks.pop(key, None) + current = asyncio.current_task() + if task is not None and task is not current: + task.cancel() diff --git a/docs/en/dev/star/guides/send-message.md b/docs/en/dev/star/guides/send-message.md index 2aa95427fe..8288945844 100644 --- a/docs/en/dev/star/guides/send-message.md +++ b/docs/en/dev/star/guides/send-message.md @@ -75,20 +75,23 @@ async def watch_room(self, event: AstrMessageEvent, target_umo: str): yield event.plain_result(f"watching {watch.target_umo}") ``` -- `watch(event, target_umo, *, source_umo=None, ttl_seconds=None)`: create an expiring watch owned by the event's trusted actor; returns `SessionWatch`. `source_umo` is the listening session that receives forwards and defaults to the current session. `ttl_seconds` is 60–864000, default 43200. When it expires, the listener session receives an end notice. +- `watch(event, target_umo, *, source_umo=None, ttl_seconds=None)`: create an expiring watch owned by the event's trusted actor; returns `SessionWatch`. `source_umo` is the listening session that receives forwards and defaults to the current session. `ttl_seconds` is 1–864000, default 43200. `expires_at` is wall-clock UTC unix seconds. `rule_id` is a stable 12-character lowercase hex id. When it expires, the listener session receives an end notice. - `unwatch(event, target_umo, *, source_umo=None)`: stop a matching watch this actor created. -- `list(event, *, source_umo=None)`: list this actor's active watches. `source_umo` is the listening session and defaults to the current session. -- `connect(event, target_umo)`: create an unbounded 1:1 link from the current session; returns `SessionWatch` with no expiry. +- `list(event, *, source_umo=None)`: list this actor's active watches (not connects). `source_umo` is the listening session and defaults to the current session. +- `connect(event, target_umo)`: create an unbounded 1:1 link from the current session; returns `SessionWatch` with `expires_at is None`. There is no `ttl_seconds` argument. - `disconnect(event)`: drop that link. - `connection(event)`: return the current unbounded link, or `None`. - `send(event, target_umo, *, target_in_header=True)`: deliver the current message body and attachments after stripping the command header; returns `DeliveryReceipt`. Linked `/send` without a UMO passes `target_in_header=False`. These methods call `authorize()` again. They require `session.watch` or `session.send`, and both sessions must share a configuration. Watches and -links stay in memory and disappear when they expire or the process restarts. +links persist in SQLite, so unexpired rules survive process restart. +`remaining_seconds` uses the wall clock; unbounded links return `0`. Do not construct `SessionBridgeManager` yourself. Import `SessionWatch` and -the duration constants from `astrbot.api.platform`. There is no Dashboard -management surface, and plugins must not assume a matching HTTP API. +the duration constants from `astrbot.api.platform`. `links` / `unlink` are +IM commands only in this stage; `SessionBridgeCapability` does not expose +them. There is no Dashboard management surface, and plugins must not assume +a matching HTTP API. ## Rich-Media Chains diff --git a/docs/en/use/authorization.md b/docs/en/use/authorization.md index 03e68c9065..c3f3a450c6 100644 --- a/docs/en/use/authorization.md +++ b/docs/en/use/authorization.md @@ -54,7 +54,7 @@ Profiles can bind separately to platforms, groups, or DMs. Editing `default` may `session.watch` and `session.send` are separate cross-session permissions requiring `instance_operator` or a higher role. Both sessions must belong to the same configuration. A group admin or private-session owner in the source session cannot use that status to watch another session. Existing cross-session restrictions on `session.manage` and `session.assign` remain in effect. -`/session watch` creates an expiring watch from the trusted event identity; command arguments cannot choose an actor. `/session connect` creates an unbounded 1:1 link with the same authorization rules. Authorization is checked again before submission. Stopping a watch or link, or revoking access, prevents subsequent queued submissions. Forwarded content is visible to everyone in the receiving session. This stage has no Dashboard management surface; see [Built-in commands](./command#cross-session-watches-and-sending). +`/session watch` creates an expiring watch from the trusted event identity; command arguments cannot choose an actor. `/session connect` creates an unbounded 1:1 link with the same authorization rules. Rules persist across restart. On restore, AstrBot rebuilds the subject with `Subject.from_id` and authorizes again; lost permission deletes the edge instead of silently continuing. Authorization is checked again before submission. Stopping a watch or link, or revoking access, prevents subsequent queued submissions. Forwarded content is visible to everyone in the receiving session. This stage has no Dashboard management surface; see [Built-in commands](./command#cross-session-watches-and-sending). ## Step-up diff --git a/docs/en/use/command.md b/docs/en/use/command.md index 8ebbd7084f..8519805a9f 100644 --- a/docs/en/use/command.md +++ b/docs/en/use/command.md @@ -68,15 +68,17 @@ The user ID from `/session info` can be granted current-session `session_admin` ### Cross-session watches and sending -- `/session watch [listener|this] [seconds]`: Forward subsequent incoming messages from the target into the listener session; requires `session.watch`. Omit the listener or write `this` for the current session. Duration is 60–864000 seconds (up to 10 days), default 43200 seconds (12 hours). When it expires, the listener session receives an end notice. +- `/session watch [listener|this] [seconds]`: Forward subsequent incoming messages from the target into the listener session; requires `session.watch`. Omit the listener or write `this` for the current session. Duration is 1–864000 seconds (up to 10 days), default 43200 seconds (12 hours), measured on the wall clock. When it expires, the listener session receives an end notice. - `/session watches [listener|this]`: List watches you created and their remaining time. Omit the argument or write `this` for the current session as the listener. - `/session unwatch [listener|this] `: Stop a matching watch that you own. Omit the listener or write `this` for the current session. -- `/session connect `: Link the current session to the target without expiry; requires `session.watch`. Incoming messages from the target are forwarded until `/session disconnect` or process restart. One link per actor and listener; connecting again replaces the previous target. `/session connect` with no argument shows the current link. +- `/session connect `: Link the current session to the target without expiry; requires `session.watch`. Incoming messages from the target are forwarded until `/session disconnect` or `/session unlink`. One link per actor and listener; connecting again replaces the previous target. `/session connect` with no argument shows the current link. Extra tokens such as a duration are rejected. - `/session disconnect`: Drop the unbounded link in the current session; requires `session.read`. +- `/session links`: List visible watches and connects with `rule_id`, kind, and remaining time or no expiry. Permission matches `watches` (`session.read`). Creators see only their own rules. An `instance_operator` on the current session configuration also sees rules whose source or target config id equals that configuration. `operator` / `root` see all rules. +- `/session unlink `: Remove a watch or connect by its 12-character lowercase hex id. Creators can remove their own rules after their role is revoked. An `instance_operator` on the current configuration can remove a rule whose either config id belongs to that configuration; `operator` / `root` can remove any rule. - `/send [content]`: Send text and attachments from the same message through the target Bot account; requires `session.send`. An attachment-only body is allowed. This does not register a `reply` command. - `/send [content]`: After `/session connect`, send to the linked target without repeating the UMO. Attachment-only bodies are allowed. -Watching, connecting, and sending require the current identity to hold `instance_operator` permission in the configuration shared by both sessions. Group admin or private-session ownership does not grant this access. Forwarded content is visible to everyone in the receiving session. Watches and links forward new incoming messages only, without reading history. They are held in memory, disappear when they expire or the process restarts, and are limited to 16 watches and 16 links per actor. Authorization is checked again for each forwarded message, so revocation stops the watch or link. Running `/session watch` again on the same pair resets the duration. +Watching, connecting, and sending require the current identity to hold `instance_operator` permission in the configuration shared by both sessions. Group admin or private-session ownership does not grant this access. Forwarded content is visible to everyone in the receiving session. Watches and links forward new incoming messages only, without reading history. Rules are stored in SQLite, so unexpired watches and all connects survive process restart; expired watches are cleared at startup or when they elapse, and the listener is notified. Limits remain 16 watches and 16 links per actor. Authorization is checked again for each forwarded message, so revocation stops delivery. Running `/session watch` again on the same pair keeps the `rule_id` and resets the duration. A watch and a connect on the same direction replace each other. Text and media retain their message-chain order; targets without mixed-content delivery receive separate messages. Cross-platform mentions become text. Quotes use accepted-message ID mappings when available and otherwise become a quote summary. Unavailable attachments and unsupported native content leave text placeholders. Platform cards, private syntax, and mini apps cannot be guaranteed to reproduce on another platform. diff --git a/docs/zh/dev/star/guides/send-message.md b/docs/zh/dev/star/guides/send-message.md index c44d7a9f43..c7e1ce9063 100644 --- a/docs/zh/dev/star/guides/send-message.md +++ b/docs/zh/dev/star/guides/send-message.md @@ -67,15 +67,15 @@ async def watch_room(self, event: AstrMessageEvent, target_umo: str): yield event.plain_result(f"watching {watch.target_umo}") ``` -- `watch(event, target_umo, *, source_umo=None, ttl_seconds=None)`：用事件上的可信主体创建有期限监听，返回 `SessionWatch`。`source_umo` 是接收转发的监听会话，缺省为当前会话；`ttl_seconds` 范围 60–864000，缺省 43200。到期后会向监听会话发送结束通知。 +- `watch(event, target_umo, *, source_umo=None, ttl_seconds=None)`：用事件上的可信主体创建有期限监听，返回 `SessionWatch`。`source_umo` 是接收转发的监听会话，缺省为当前会话；`ttl_seconds` 范围 1–864000，缺省 43200。`expires_at` 是墙钟 UTC unix 秒。`rule_id` 是 12 位小写十六进制稳定 id。到期后会向监听会话发送结束通知。 - `unwatch(event, target_umo, *, source_umo=None)`：停止当前主体创建的对应监听。 -- `list(event, *, source_umo=None)`：列出当前主体的有效监听。`source_umo` 是监听会话，缺省为当前会话。 -- `connect(event, target_umo)`：从当前会话创建无期限的一对一连接，返回没有过期时间的 `SessionWatch`。 +- `list(event, *, source_umo=None)`：列出当前主体的有效监听（不含 connect）。`source_umo` 是监听会话，缺省为当前会话。 +- `connect(event, target_umo)`：从当前会话创建无期限的一对一连接，返回 `expires_at is None` 的 `SessionWatch`。没有 `ttl_seconds` 参数。 - `disconnect(event)`：断开该连接。 - `connection(event)`：返回当前无期限连接，没有则是 `None`。 - `send(event, target_umo, *, target_in_header=True)`：把当前消息去掉指令头后的正文和附件投递到目标会话，返回 `DeliveryReceipt`。已连接且不带 UMO 的 `/send` 传 `target_in_header=False`。 -这些方法会再次调用 `authorize()`，要求 `session.watch` 或 `session.send`，且两个会话属于同一配置。监听和连接保存在内存中，到期或重启后清空。不要自己构造 `SessionBridgeManager`。`SessionWatch` 和时长常量可从 `astrbot.api.platform` 导入。当前没有 Dashboard 管理面，插件也不应假设存在对应 HTTP API。 +这些方法会再次调用 `authorize()`，要求 `session.watch` 或 `session.send`，且两个会话属于同一配置。监听和连接会持久化，重启后未过期的规则仍在。`remaining_seconds` 按墙钟计算，无期限返回 `0`。不要自己构造 `SessionBridgeManager`。`SessionWatch` 和时长常量可从 `astrbot.api.platform` 导入。`links` / `unlink` 本阶段只做 IM 指令，没有对应的 `SessionBridgeCapability` 方法。当前没有 Dashboard 管理面，插件也不应假设存在对应 HTTP API。 ## 富媒体消息链 diff --git a/docs/zh/use/authorization.md b/docs/zh/use/authorization.md index c5799a4874..87d2e50484 100644 --- a/docs/zh/use/authorization.md +++ b/docs/zh/use/authorization.md @@ -54,7 +54,7 @@ AstrBot 把 Dashboard 登录、IM 会话管理和高风险操作拆开。把群 `session.watch` 和 `session.send` 是独立的跨会话权限，要求 `instance_operator` 或更高角色，且两个会话必须属于同一配置。源会话的群管理员或私聊所有者身份不能用于监听其他会话。已有的 `session.manage`、`session.assign` 跨会话限制保持不变。 -`/session watch` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。`/session connect` 用同一套授权规则创建无期限的一对一连接。每次投递前重新检查权限，停止监听或连接、或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。 +`/session watch` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。`/session connect` 用同一套授权规则创建无期限的一对一连接。规则会持久化；启动时用 `Subject.from_id` 重建主体并再次鉴权，权限已无则删除该边，不会静默续转。每次投递前重新检查权限，停止监听或连接、或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。 ## 二次验证（step-up） diff --git a/docs/zh/use/command.md b/docs/zh/use/command.md index 0fa7d6b4f5..c0daa16ef7 100644 --- a/docs/zh/use/command.md +++ b/docs/zh/use/command.md @@ -68,15 +68,17 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、 ### 跨会话监听与发送 -- `/session watch [监听会话|this] <被监听会话> [秒数]`：把被监听会话后续收到的消息转发到监听会话，需要 `session.watch`。监听会话可省略或写 `this`，表示当前会话。时长 60–864000 秒（最多 10 天），默认 43200 秒（12 小时）。到期后会在监听会话发送结束通知。 +- `/session watch [监听会话|this] <被监听会话> [秒数]`：把被监听会话后续收到的消息转发到监听会话，需要 `session.watch`。监听会话可省略或写 `this`，表示当前会话。时长 1–864000 秒（最多 10 天），默认 43200 秒（12 小时），按墙钟计算。到期后会在监听会话发送结束通知。 - `/session watches [监听会话|this]`：查看你创建的监听和剩余时间。省略参数或写 `this` 表示当前会话作为监听端。 - `/session unwatch [监听会话|this] <被监听会话>`：停止你创建的指定监听。监听会话可省略或写 `this`。 -- `/session connect `：把当前会话无期限连接到目标会话，需要 `session.watch`。目标会话的新消息会一直转发，直到 `/session disconnect` 或进程重启。每个主体在每个监听会话只能有一条连接，再次连接会替换目标。省略 UMO 可查看当前连接。 +- `/session connect `：把当前会话无期限连接到目标会话，需要 `session.watch`。目标会话的新消息会一直转发，直到 `/session disconnect` 或 `/session unlink`。每个主体在每个监听会话只能有一条连接，再次连接会替换目标。省略 UMO 可查看当前连接。指令不接受秒数。 - `/session disconnect`：断开当前会话的无期限连接，需要 `session.read`。 +- `/session links`：列出可见的监听和连接，包含 `rule_id`、类型和剩余时间或无期限。权限与 `watches` 相同（`session.read`）。创建者只看到自己的规则；当前会话配置上的 `instance_operator` 额外看到两端任一配置 id 属于当前会话配置的规则；`operator` / `root` 看到全部。 +- `/session unlink `：按 12 位小写十六进制 id 撤销一条监听或连接。创建者可撤自己的规则（角色被撤后仍可）。本配置 `instance_operator` 可撤两端任一配置属于当前会话配置的规则；`operator` / `root` 可撤全部。 - `/send [内容]`：借助目标平台的 Bot 账号发送文字和同一条消息中的附件，需要 `session.send`。可以只附图片而不填写正文；不会占用 `reply` 指令。 - `/send [内容]`：在 `/session connect` 之后，不写 UMO 也会发往已连接的目标会话。可以只附图片。 -监听、连接和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。监听和连接保存在内存中，到期或重启后清空，每人最多 16 条监听和 16 条连接。每次转发都会重新检查权限，撤权后停止。对同一对会话再次 `/session watch` 会重置时长。 +监听、连接和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。规则写入 SQLite，进程重启后未过期的监听和全部连接仍在；过期监听会在启动或到期时清除并通知监听端。每人最多 16 条监听和 16 条连接。每次转发都会重新检查权限，撤权后停止投递。对同一对会话再次 `/session watch` 会保留 `rule_id` 并重置时长。同一方向的 watch 与 connect 会互相替换。 正文保留消息链中的图文先后顺序；不能混排的目标拆成多条消息。跨平台提及转成文字，引用优先通过已接受消息的 ID 映射还原；映射不存在时附引用摘要。无法解析的附件和不支持的原生内容会保留文字占位。平台自己的卡片、私有语法和小程序不能保证在别的平台重现。 diff --git a/tests/unit/db/test_protocols.py b/tests/unit/db/test_protocols.py index e5d8335aa1..361c448b7b 100644 --- a/tests/unit/db/test_protocols.py +++ b/tests/unit/db/test_protocols.py @@ -26,6 +26,7 @@ PersonaStore, PlatformSessionStore, PreferenceStore, + SessionBridgeStore, SessionManagementStore, StatisticsSessionStore, StatisticsStore, @@ -50,6 +51,7 @@ CommandStore, CronStore, PlatformSessionStore, + SessionBridgeStore, UmoAliasStore, ChatProjectStore, ) diff --git a/tests/unit/db/test_schema.py b/tests/unit/db/test_schema.py index 51599b20ce..4339251511 100644 --- a/tests/unit/db/test_schema.py +++ b/tests/unit/db/test_schema.py @@ -46,6 +46,7 @@ "platform_stats", "preferences", "provider_stats", + "session_bridge_rules", "session_project_relations", "umo_aliases", "webchat_threads", @@ -87,6 +88,7 @@ "PlatformStat", "Preference", "ProviderStat", + "SessionBridgeRule", "SessionProjectRelation", "UmoAlias", "WebChatThread", @@ -204,6 +206,41 @@ def inspect_commands(sync_conn): assert ("command_id",) in config_uniques +@pytest.mark.asyncio +async def test_session_bridge_rules_have_required_columns_and_direction_unique( + temp_db: SQLiteDatabase, +): + await temp_db.initialize() + + def inspect_rules(sync_conn): + return ( + _column_map(sync_conn, "session_bridge_rules"), + _unique_column_sets(sync_conn, "session_bridge_rules"), + ) + + async with temp_db.engine.connect() as conn: + columns, uniques = await conn.run_sync(inspect_rules) + + for name in ( + "rule_id", + "subject_id", + "source_umo", + "target_umo", + "source_config_id", + "target_config_id", + "kind", + "expires_at", + "header", + "pair_id", + "match", + "except", + ): + assert name in columns + assert columns["rule_id"]["nullable"] is False + assert columns["kind"]["nullable"] is False + assert ("subject_id", "source_umo", "target_umo") in uniques + + @pytest.mark.asyncio async def test_initialize_does_not_patch_leftover_columns( temp_db: SQLiteDatabase, diff --git a/tests/unit/db/test_session_bridge_store.py b/tests/unit/db/test_session_bridge_store.py new file mode 100644 index 0000000000..a6d4ff9edf --- /dev/null +++ b/tests/unit/db/test_session_bridge_store.py @@ -0,0 +1,112 @@ +import pytest +from sqlalchemy.exc import IntegrityError + +from astrbot.core.db.sqlite import SQLiteDatabase + + +async def _insert_rule( + temp_db: SQLiteDatabase, + *, + subject_id: str = "im:source:bot:actor", + source_umo: str = "source:FriendMessage:sender", + target_umo: str = "target:GroupMessage:room", + source_config_id: str = "default", + target_config_id: str = "default", + kind: str = "watch", + expires_at: int | None = 1_800_000_000, + **kwargs, +): + return await temp_db.insert_session_bridge_rule( + subject_id=subject_id, + source_umo=source_umo, + target_umo=target_umo, + source_config_id=source_config_id, + target_config_id=target_config_id, + kind=kind, + expires_at=expires_at, + **kwargs, + ) + + +@pytest.mark.asyncio +async def test_insert_session_bridge_rule_writes_s1_defaults( + temp_db: SQLiteDatabase, +): + await temp_db.initialize() + watch = await _insert_rule(temp_db) + connect = await _insert_rule( + temp_db, + target_umo="other:GroupMessage:room", + kind="connect", + expires_at=None, + ) + + assert len(watch.rule_id) == 12 + assert watch.rule_id.islower() + assert watch.rule_id.isalnum() + assert int(watch.rule_id, 16) >= 0 + assert watch.header is True + assert watch.pair_id is None + assert watch.match == {} + assert watch.except_ == {} + assert watch.kind == "watch" + assert watch.expires_at == 1_800_000_000 + assert connect.kind == "connect" + assert connect.expires_at is None + + +@pytest.mark.asyncio +async def test_direction_unique_constraint_rejects_duplicate_edge( + temp_db: SQLiteDatabase, +): + await temp_db.initialize() + await _insert_rule(temp_db) + with pytest.raises(IntegrityError): + await _insert_rule(temp_db, kind="connect", expires_at=None) + + +@pytest.mark.asyncio +async def test_session_bridge_rule_queries_update_and_delete( + temp_db: SQLiteDatabase, +): + await temp_db.initialize() + watch = await _insert_rule(temp_db) + other = await _insert_rule( + temp_db, + subject_id="im:source:bot:other", + source_umo="ops:FriendMessage:box", + target_umo="target:GroupMessage:room", + source_config_id="ops", + kind="connect", + expires_at=None, + ) + fetched = await temp_db.get_session_bridge_rule(watch.rule_id) + by_direction = await temp_db.get_session_bridge_rule_by_direction( + watch.subject_id, watch.source_umo, watch.target_umo + ) + owned = await temp_db.list_session_bridge_rules_by_subject(watch.subject_id) + touching = await temp_db.list_session_bridge_rules_touching_config("ops") + connects = await temp_db.list_session_bridge_connects_for_listener( + other.subject_id, other.source_umo + ) + updated = await temp_db.update_session_bridge_rule( + watch.rule_id, expires_at=1_800_000_100 + ) + + assert fetched is not None + assert fetched.rule_id == watch.rule_id + assert by_direction is not None + assert by_direction.rule_id == watch.rule_id + assert [item.rule_id for item in owned] == [watch.rule_id] + assert [item.rule_id for item in touching] == [other.rule_id] + assert [item.rule_id for item in connects] == [other.rule_id] + assert updated is not None + assert updated.expires_at == 1_800_000_100 + assert updated.kind == "watch" + + await temp_db.delete_session_bridge_connects_for_listener( + other.subject_id, other.source_umo + ) + await temp_db.delete_session_bridge_rule(watch.rule_id) + remaining = await temp_db.list_session_bridge_rules() + assert remaining == [] diff --git a/tests/unit/test_builtin_command_extensions.py b/tests/unit/test_builtin_command_extensions.py index 9ce1a35cc2..6ee85c5361 100644 --- a/tests/unit/test_builtin_command_extensions.py +++ b/tests/unit/test_builtin_command_extensions.py @@ -263,6 +263,8 @@ def test_all_builtin_extension_commands_use_native_command_schemas(): "session_watches", "session_connect", "session_disconnect", + "session_links", + "session_unlink", "session_block", "session_unblock", "send_to_session", @@ -1101,6 +1103,8 @@ def command_names(group: CommandGroupFilter) -> set[str]: "watches", "connect", "disconnect", + "links", + "unlink", }, "conversation": { "create", @@ -1167,6 +1171,8 @@ def test_non_public_builtin_commands_declare_the_planned_actions(): "session_watches": "session.read", "session_connect": "session.watch", "session_disconnect": "session.read", + "session_links": "session.read", + "session_unlink": "session.read", "session_block": "session.block", "session_unblock": "session.block", "send_to_session": "session.send", diff --git a/tests/unit/test_core_lifecycle.py b/tests/unit/test_core_lifecycle.py index c4037386f2..197391c5e8 100644 --- a/tests/unit/test_core_lifecycle.py +++ b/tests/unit/test_core_lifecycle.py @@ -434,6 +434,8 @@ async def test_initialize_sets_up_all_components( mock_execution_context.session_waiter_registry = SimpleNamespace( terminate=AsyncMock() ) + mock_execution_context.session_bridge_manager.restore = AsyncMock() + mock_execution_context.session_bridge_manager.terminate = AsyncMock() mock_plugin_manager = MagicMock() mock_plugin_manager.catalog.plugins.all.return_value = [] @@ -688,7 +690,9 @@ async def action(*_args) -> None: background_tasks=set(), session_waiter_registry=SimpleNamespace(terminate=AsyncMock()), session_bridge_manager=SimpleNamespace( - observe=AsyncMock(), terminate=AsyncMock() + observe=AsyncMock(), + terminate=AsyncMock(), + restore=AsyncMock(), ), ) plugin_manager = SimpleNamespace( diff --git a/tests/unit/test_message_protocol.py b/tests/unit/test_message_protocol.py index 95ce1262ef..480698a120 100644 --- a/tests/unit/test_message_protocol.py +++ b/tests/unit/test_message_protocol.py @@ -22,6 +22,7 @@ from astrbot.core.platform.route_identity import PlatformRouteIdentity from astrbot.core.platform.send_result import PlatformSendResult from astrbot.core.platform.session_bridge import SessionBridgeManager +from tests.unit.test_session_bridge import FakeSessionBridgeStore def test_message_i18n_catalogs_share_keys(): @@ -143,6 +144,7 @@ def _manager(send=None): lambda _: MessageDeliveryCapabilities(quote=True, media=frozenset({"image"})), authorization=authorization, get_config_id=lambda _: "default", + store=FakeSessionBridgeStore(), ) return manager, authorization, sender @@ -266,6 +268,7 @@ async def send(session, chain): authorize=AsyncMock(return_value=SimpleNamespace(allowed=True)) ), get_config_id=lambda _: "default", + store=FakeSessionBridgeStore(), ) source = "webchat:FriendMessage:source" target = "telegram:GroupMessage:target" @@ -309,6 +312,7 @@ async def send(session, chain): ), get_config_id=lambda _: "default", get_locale=AsyncMock(return_value="en-US"), + store=FakeSessionBridgeStore(), ) source = "webchat:FriendMessage:source" target = "telegram:GroupMessage:target" @@ -346,6 +350,7 @@ async def send(session, chain): get_platform_family=lambda umo: { "tg-main:FriendMessage:source": "telegram", }.get(umo, umo.split(":", 1)[0]), + store=FakeSessionBridgeStore(), ) source = "tg-main:FriendMessage:source" target = "napcat:GroupMessage:room" @@ -973,127 +978,6 @@ def test_parse_watch_spec_accepts_this_omission_and_duration(): ) -@pytest.mark.asyncio -async def test_watch_custom_source_and_ttl_and_rejects_out_of_range(): - from astrbot.core.platform.session_bridge import ( - DEFAULT_WATCH_TTL_SECONDS, - MAX_WATCH_TTL_SECONDS, - MIN_WATCH_TTL_SECONDS, - ) - - manager, authorization, _ = _manager() - event = _event() - listener = "other:FriendMessage:box" - target = "target:GroupMessage:room" - watch = await manager.watch( - event, target, source_umo=listener, ttl_seconds=MIN_WATCH_TTL_SECONDS - ) - assert watch.source_umo == listener - assert watch.target_umo == target - assert MIN_WATCH_TTL_SECONDS - 1 <= watch.remaining_seconds <= MIN_WATCH_TTL_SECONDS - assert [call.args[2].umo for call in authorization.authorize.await_args_list] == [ - listener, - target, - ] - default = await manager.watch(_event("source:FriendMessage:two"), target) - assert ( - DEFAULT_WATCH_TTL_SECONDS - 1 - <= default.remaining_seconds - <= DEFAULT_WATCH_TTL_SECONDS - ) - with pytest.raises(ValueError, match="Invalid watch duration"): - await manager.watch(event, target, ttl_seconds=MIN_WATCH_TTL_SECONDS - 1) - with pytest.raises(ValueError, match="Invalid watch duration"): - await manager.watch(event, target, ttl_seconds=MAX_WATCH_TTL_SECONDS + 1) - assert await manager.unwatch(event, target, source_umo=listener) - await manager.terminate() - - -@pytest.mark.asyncio -async def test_watch_expiry_notifies_listener(monkeypatch): - import asyncio - - sent = [] - - async def send(session, chain): - sent.append((str(session), chain.get_plain_text())) - return PlatformSendResult(session.platform_id, True, str(session)) - - async def instant_sleep(_delay): - return - - monkeypatch.setattr( - "astrbot.core.platform.session_bridge.asyncio.sleep", instant_sleep - ) - manager, _, _ = _manager(send) - source = "source:FriendMessage:sender" - target = "target:GroupMessage:room" - await manager.watch(_event(source), target, ttl_seconds=60) - await asyncio.sleep(0) - pending = [ - task - for task in asyncio.all_tasks() - if task.get_name().startswith("session-watch-expire") - ] - if pending: - await asyncio.gather(*pending) - assert sent == [(source, "对 target:GroupMessage:room 的监听已结束。")] - assert await manager.list_watches(_event(source)) == () - await manager.terminate() - - -@pytest.mark.asyncio -async def test_watch_expiry_notice_follows_locale(monkeypatch): - import asyncio - - sent = [] - - async def send(session, chain): - sent.append(chain.get_plain_text()) - return PlatformSendResult(session.platform_id, True, str(session)) - - async def instant_sleep(_delay): - return - - monkeypatch.setattr( - "astrbot.core.platform.session_bridge.asyncio.sleep", instant_sleep - ) - manager, _, _ = _manager(send) - manager._get_locale = AsyncMock(return_value="en-US") - await manager.watch(_event(), "target:GroupMessage:room", ttl_seconds=60) - await asyncio.sleep(0) - pending = [ - task - for task in asyncio.all_tasks() - if task.get_name().startswith("session-watch-expire") - ] - if pending: - await asyncio.gather(*pending) - assert sent == ["The watch on target:GroupMessage:room has ended."] - await manager.terminate() - - -@pytest.mark.asyncio -async def test_terminate_clears_watches_and_message_maps(): - manager, _, _ = _manager() - event = _event() - await manager.watch(event, "target:GroupMessage:room") - await manager.observe( - MessageEnvelope( - PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), - source_message_id="original", - content=(PortablePart(ContentKind.TEXT, "hello"),), - ) - ) - assert manager._watches - assert manager._message_ids - await manager.terminate() - assert manager._watches == {} - assert manager._links == {} - assert manager._forwarded == {} - assert manager._message_ids == {} - - def test_is_umo_accepts_session_strings(): from astrbot.builtin_stars.builtin_commands.commands.session import _is_umo @@ -1117,47 +1001,6 @@ def test_send_projection_strips_command_only_when_target_not_in_header(): envelope_from_send_event(event, "target:GroupMessage:room") -@pytest.mark.asyncio -async def test_connect_forwards_without_expiry_and_send_uses_link(): - from astrbot.core.message.components import Plain - from astrbot.core.star.plugin_context import SessionBridgeCapability - - sent = [] - - async def send(session, chain): - sent.append((str(session), chain.get_plain_text())) - return PlatformSendResult(session.platform_id, True, str(session)) - - manager, authorization, _ = _manager(send) - event = _event() - target = "target:GroupMessage:room" - capability = SessionBridgeCapability(manager) - link = await capability.connect(event, target) - assert link.expires_at is None - assert await capability.connection(event) == link - assert [call.args[2].umo for call in authorization.authorize.await_args_list] == [ - event.unified_msg_origin, - target, - ] - await manager.observe( - MessageEnvelope( - PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), - source_message_id="9", - sender=SenderSnapshot("1", "Alice", "napcat"), - content=(PortablePart(ContentKind.TEXT, "hello"),), - ) - ) - assert sent[-1][0] == event.unified_msg_origin - assert "hello" in sent[-1][1] - - send_event = _event(components=[Plain("/send ping")]) - receipt = await capability.send(send_event, target, target_in_header=False) - assert receipt.status == "accepted" - assert await capability.disconnect(event) - assert await capability.connection(event) is None - await manager.terminate() - - @pytest.mark.asyncio async def test_session_commands_connect_and_linked_send(): from astrbot.builtin_stars.builtin_commands.commands.session import SessionCommands diff --git a/tests/unit/test_session_bridge.py b/tests/unit/test_session_bridge.py new file mode 100644 index 0000000000..21705701d1 --- /dev/null +++ b/tests/unit/test_session_bridge.py @@ -0,0 +1,935 @@ +import secrets +from time import time +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from sqlalchemy.exc import IntegrityError + +from astrbot.core.auth.models import AuthContext, Resource, Role, Subject +from astrbot.core.platform.message_protocol import ( + ContentKind, + MessageDeliveryCapabilities, + MessageEnvelope, + PortablePart, + SenderSnapshot, +) +from astrbot.core.platform.message_type import MessageType +from astrbot.core.platform.route_identity import PlatformRouteIdentity +from astrbot.core.platform.send_result import PlatformSendResult +from astrbot.core.platform.session_bridge import ( + DEFAULT_WATCH_TTL_SECONDS, + MAX_WATCH_TTL_SECONDS, + MIN_WATCH_TTL_SECONDS, + SessionBridgeManager, +) + + +def _event(umo="source:FriendMessage:sender", *, sender="actor", components=()): + from astrbot.core.platform.message_session import MessageSession + + subject = Subject.im( + platform_instance="source", bot_account_id="bot", sender_id=sender + ) + session = MessageSession.from_str(umo) + return SimpleNamespace( + unified_msg_origin=umo, + subject=subject, + auth_context=AuthContext( + subject=subject, + source="im", + config_id="default", + authenticated=True, + origin_session_resource_id=Resource.session("default", umo).id, + ), + route_identity=PlatformRouteIdentity( + session.platform_id, session.message_type, session.session_id + ), + get_messages=lambda: list(components), + get_platform_name=lambda: "telegram", + get_sender_id=lambda: sender, + get_sender_name=lambda: sender, + get_self_id=lambda: "bot", + message_obj=SimpleNamespace(message_id="inbound"), + created_at=0.0, + ) + + +class FakeSessionBridgeStore: + def __init__(self) -> None: + self.rows: dict[str, SimpleNamespace] = {} + self.by_direction: dict[tuple[str, str, str], str] = {} + + def seed(self, **fields) -> SimpleNamespace: + payload = { + "header": True, + "pair_id": None, + "match": {}, + "except_": {}, + } + payload.update(fields) + row = SimpleNamespace(**payload) + self.rows[row.rule_id] = row + self.by_direction[(row.subject_id, row.source_umo, row.target_umo)] = ( + row.rule_id + ) + return row + + async def insert_session_bridge_rule(self, **kwargs) -> SimpleNamespace: + key = (kwargs["subject_id"], kwargs["source_umo"], kwargs["target_umo"]) + if key in self.by_direction: + raise IntegrityError("statement", {}, Exception("direction")) + rule_id = kwargs.get("rule_id") or secrets.token_hex(6) + if rule_id in self.rows: + raise IntegrityError("statement", {}, Exception("rule_id")) + return self.seed( + rule_id=rule_id, + expires_at=kwargs.get("expires_at"), + kind=kwargs["kind"], + subject_id=kwargs["subject_id"], + source_umo=kwargs["source_umo"], + target_umo=kwargs["target_umo"], + source_config_id=kwargs["source_config_id"], + target_config_id=kwargs["target_config_id"], + header=kwargs.get("header", True), + pair_id=kwargs.get("pair_id"), + match=kwargs.get("match") or {}, + except_=kwargs.get("except_") or {}, + ) + + async def get_session_bridge_rule(self, rule_id: str): + return self.rows.get(rule_id) + + async def get_session_bridge_rule_by_direction( + self, subject_id: str, source_umo: str, target_umo: str + ): + rule_id = self.by_direction.get((subject_id, source_umo, target_umo)) + return None if rule_id is None else self.rows.get(rule_id) + + async def list_session_bridge_rules(self): + return list(self.rows.values()) + + async def list_session_bridge_rules_by_subject(self, subject_id: str): + return [row for row in self.rows.values() if row.subject_id == subject_id] + + async def list_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ): + return [ + row + for row in self.rows.values() + if row.subject_id == subject_id + and row.source_umo == source_umo + and row.kind == "connect" + ] + + async def list_session_bridge_rules_touching_config(self, config_id: str): + return [ + row + for row in self.rows.values() + if row.source_config_id == config_id or row.target_config_id == config_id + ] + + async def update_session_bridge_rule(self, rule_id: str, **kwargs): + row = self.rows.get(rule_id) + if row is None: + return None + for key, value in kwargs.items(): + setattr(row, key, value) + return row + + async def delete_session_bridge_rule(self, rule_id: str) -> None: + row = self.rows.pop(rule_id, None) + if row is None: + return + self.by_direction.pop((row.subject_id, row.source_umo, row.target_umo), None) + + async def delete_session_bridge_connects_for_listener( + self, subject_id: str, source_umo: str + ) -> None: + for row in list(self.rows.values()): + if ( + row.subject_id == subject_id + and row.source_umo == source_umo + and row.kind == "connect" + ): + await self.delete_session_bridge_rule(row.rule_id) + + +def _manager(send=None, store=None, *, max_watches_per_subject=16, get_config_id=None): + authorization = SimpleNamespace( + authorize=AsyncMock( + return_value=SimpleNamespace(allowed=True, effective_role=None) + ) + ) + sender = send or AsyncMock( + return_value=PlatformSendResult("target", True, "target", message_ids=("sent",)) + ) + manager = SessionBridgeManager( + sender, + lambda _: MessageDeliveryCapabilities(quote=True, media=frozenset({"image"})), + authorization=authorization, + get_config_id=get_config_id or (lambda _: "default"), + store=store or FakeSessionBridgeStore(), + max_watches_per_subject=max_watches_per_subject, + ) + return manager, authorization, sender + + +def test_session_bridge_state_is_not_a_public_export(): + import astrbot.api.platform as api_platform + import astrbot.core.platform as core_platform + + assert "SessionBridgeState" not in api_platform.__all__ + assert "SessionBridgeState" not in core_platform.__all__ + assert not hasattr(api_platform, "SessionBridgeState") + + +@pytest.mark.asyncio +async def test_watch_records_rule_id_and_rejects_zero_ttl(): + manager, _, _ = _manager() + event = _event() + target = "target:GroupMessage:room" + watch = await manager.watch(event, target, ttl_seconds=1) + assert len(watch.rule_id) == 12 + assert watch.rule_id == watch.rule_id.lower() + int(watch.rule_id, 16) + assert watch.expires_at is not None + assert watch.expires_at > time() + with pytest.raises(ValueError, match="Invalid watch duration"): + await manager.watch(event, target, ttl_seconds=0) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_watch_custom_source_and_ttl_and_rejects_out_of_range(): + manager, authorization, _ = _manager() + event = _event() + listener = "other:FriendMessage:box" + target = "target:GroupMessage:room" + watch = await manager.watch( + event, target, source_umo=listener, ttl_seconds=MIN_WATCH_TTL_SECONDS + ) + assert watch.source_umo == listener + assert watch.target_umo == target + assert watch.expires_at is not None + assert watch.expires_at > time() + assert [call.args[2].umo for call in authorization.authorize.await_args_list] == [ + listener, + target, + ] + default = await manager.watch(_event("source:FriendMessage:two"), target) + assert ( + DEFAULT_WATCH_TTL_SECONDS - 1 + <= default.remaining_seconds + <= DEFAULT_WATCH_TTL_SECONDS + ) + with pytest.raises(ValueError, match="Invalid watch duration"): + await manager.watch(event, target, ttl_seconds=MIN_WATCH_TTL_SECONDS - 1) + with pytest.raises(ValueError, match="Invalid watch duration"): + await manager.watch(event, target, ttl_seconds=MAX_WATCH_TTL_SECONDS + 1) + assert await manager.unwatch(event, target, source_umo=listener) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_watch_expiry_notifies_listener(monkeypatch): + import asyncio + + sent = [] + + async def send(session, chain): + sent.append((str(session), chain.get_plain_text())) + return PlatformSendResult(session.platform_id, True, str(session)) + + async def instant_sleep(_delay): + return + + monkeypatch.setattr( + "astrbot.core.platform.session_bridge.asyncio.sleep", instant_sleep + ) + manager, _, _ = _manager(send) + source = "source:FriendMessage:sender" + target = "target:GroupMessage:room" + await manager.watch(_event(source), target, ttl_seconds=60) + await asyncio.sleep(0) + pending = [ + task + for task in asyncio.all_tasks() + if task.get_name().startswith("session-watch-expire") + ] + if pending: + await asyncio.gather(*pending) + assert sent == [(source, "对 target:GroupMessage:room 的监听已结束。")] + assert await manager.list_watches(_event(source)) == () + await manager.terminate() + + +@pytest.mark.asyncio +async def test_watch_expiry_notice_follows_locale(monkeypatch): + import asyncio + + sent = [] + + async def send(session, chain): + sent.append(chain.get_plain_text()) + return PlatformSendResult(session.platform_id, True, str(session)) + + async def instant_sleep(_delay): + return + + monkeypatch.setattr( + "astrbot.core.platform.session_bridge.asyncio.sleep", instant_sleep + ) + manager, _, _ = _manager(send) + manager._get_locale = AsyncMock(return_value="en-US") + await manager.watch(_event(), "target:GroupMessage:room", ttl_seconds=60) + await asyncio.sleep(0) + pending = [ + task + for task in asyncio.all_tasks() + if task.get_name().startswith("session-watch-expire") + ] + if pending: + await asyncio.gather(*pending) + assert sent == ["The watch on target:GroupMessage:room has ended."] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_terminate_clears_watches_and_message_maps(): + manager, _, _ = _manager() + event = _event() + await manager.watch(event, "target:GroupMessage:room") + await manager.observe( + MessageEnvelope( + PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), + source_message_id="original", + content=(PortablePart(ContentKind.TEXT, "hello"),), + ) + ) + assert manager._state.total_kind("watch") + assert manager._message_ids + await manager.terminate() + assert manager._state.total_kind("watch") == 0 + assert manager._state.total_kind("connect") == 0 + assert manager._forwarded == {} + assert manager._message_ids == {} + + +@pytest.mark.asyncio +async def test_connect_forwards_without_expiry_and_send_uses_link(): + from astrbot.core.message.components import Plain + from astrbot.core.star.plugin_context import SessionBridgeCapability + + sent = [] + + async def send(session, chain): + sent.append((str(session), chain.get_plain_text())) + return PlatformSendResult(session.platform_id, True, str(session)) + + manager, authorization, _ = _manager(send) + event = _event() + target = "target:GroupMessage:room" + capability = SessionBridgeCapability(manager) + link = await capability.connect(event, target) + assert link.expires_at is None + assert await capability.connection(event) == link + assert [call.args[2].umo for call in authorization.authorize.await_args_list] == [ + event.unified_msg_origin, + target, + ] + await manager.observe( + MessageEnvelope( + PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), + source_message_id="9", + sender=SenderSnapshot("1", "Alice", "napcat"), + content=(PortablePart(ContentKind.TEXT, "hello"),), + ) + ) + assert sent[-1][0] == event.unified_msg_origin + assert "hello" in sent[-1][1] + + send_event = _event(components=[Plain("/send ping")]) + receipt = await capability.send(send_event, target, target_in_header=False) + assert receipt.status == "accepted" + assert await capability.disconnect(event) + assert await capability.connection(event) is None + await manager.terminate() + + +@pytest.mark.asyncio +async def test_same_watch_direction_keeps_rule_id_and_resets_ttl(): + manager, _, _ = _manager() + event = _event() + target = "target:GroupMessage:room" + first = await manager.watch(event, target, ttl_seconds=10) + second = await manager.watch(event, target, ttl_seconds=100) + assert first.rule_id == second.rule_id + assert 90 <= second.remaining_seconds <= 100 + listed = await manager.list_watches(event) + assert len(listed) == 1 + await manager.terminate() + + +@pytest.mark.asyncio +async def test_connect_retarget_replaces_and_same_target_keeps_rule_id(): + manager, _, _ = _manager() + event = _event() + first = await manager.connect(event, "target:GroupMessage:one") + again = await manager.connect(event, "target:GroupMessage:one") + assert again.rule_id == first.rule_id + retarget = await manager.connect(event, "target:GroupMessage:two") + assert retarget.rule_id != first.rule_id + current = await manager.connection(event) + assert current is not None + assert current.target_umo == "target:GroupMessage:two" + assert manager._state.total_kind("connect") == 1 + await manager.terminate() + + +@pytest.mark.asyncio +async def test_cross_kind_replace_uses_new_rule_id(): + manager, _, _ = _manager() + event = _event() + target = "target:GroupMessage:room" + watch = await manager.watch(event, target, ttl_seconds=30) + link = await manager.connect(event, target) + assert link.rule_id != watch.rule_id + assert await manager.list_watches(event) == () + current = await manager.connection(event) + assert current is not None + assert current.rule_id == link.rule_id + watch_again = await manager.watch(event, target, ttl_seconds=30) + assert watch_again.rule_id != link.rule_id + assert await manager.connection(event) is None + await manager.terminate() + + +@pytest.mark.asyncio +async def test_pair_direction_rejects_watch_and_connect(): + store = FakeSessionBridgeStore() + store.seed( + rule_id="abcdef123456", + subject_id=_event().subject.id, + source_umo="source:FriendMessage:sender", + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="pair", + expires_at=None, + ) + manager, _, _ = _manager(store=store) + event = _event() + with pytest.raises(ValueError, match="occupied by a pair"): + await manager.watch(event, "target:GroupMessage:room", ttl_seconds=30) + with pytest.raises(ValueError, match="occupied by a pair"): + await manager.connect(event, "target:GroupMessage:room") + remaining = await store.get_session_bridge_rule("abcdef123456") + assert remaining is not None + assert remaining.kind == "pair" + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_reloads_unexpired_watch_and_connect(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="watchrule0001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + store.seed( + rule_id="connectrule01", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="other:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="connect", + expires_at=None, + ) + manager, _, _ = _manager(store=store) + await manager.restore() + watches = await manager.list_watches(event) + assert [item.rule_id for item in watches] == ["watchrule0001"] + link = await manager.connection(event) + assert link is not None + assert link.rule_id == "connectrule01" + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_deletes_expired_watch_and_notifies(): + sent = [] + + async def send(session, chain): + sent.append((str(session), chain.get_plain_text())) + return PlatformSendResult(session.platform_id, True, str(session)) + + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="expiredwatch1", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) - 10, + ) + manager, _, _ = _manager(send, store) + await manager.restore() + assert await manager.list_watches(event) == () + assert await store.get_session_bridge_rule("expiredwatch1") is None + assert sent == [ + (event.unified_msg_origin, "对 target:GroupMessage:room 的监听已结束。") + ] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_deletes_row_when_reauthorize_denied(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="revokedwatch1", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, authorization, _ = _manager(store=store) + authorization.authorize = AsyncMock(return_value=SimpleNamespace(allowed=False)) + await manager.restore() + assert await manager.list_watches(event) == () + assert await store.get_session_bridge_rule("revokedwatch1") is None + await manager.terminate() + + +@pytest.mark.asyncio +async def test_unavailable_adapter_keeps_restored_rule(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="hangingwatch1", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, _, send = _manager(store=store) + manager._get_capabilities = lambda _: MessageDeliveryCapabilities( + proactive=False, available=False + ) + await manager.restore() + await manager.observe( + MessageEnvelope( + PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), + source_message_id="keep-me", + content=(PortablePart(ContentKind.TEXT, "hello"),), + ) + ) + assert await store.get_session_bridge_rule("hangingwatch1") is not None + send.assert_not_called() + await manager.terminate() + + +@pytest.mark.asyncio +async def test_terminate_keeps_sqlite_rows(): + store = FakeSessionBridgeStore() + manager, _, _ = _manager(store=store) + watch = await manager.watch(_event(), "target:GroupMessage:room", ttl_seconds=30) + await manager.terminate() + assert await store.get_session_bridge_rule(watch.rule_id) is not None + assert manager._state.total_kind("watch") == 0 + + +@pytest.mark.asyncio +async def test_list_links_creator_and_instance_operator_visibility(): + owner = _event() + other = _event(sender="other") + store = FakeSessionBridgeStore() + store.seed( + rule_id="ownwatch00001", + subject_id=owner.subject.id, + source_umo=owner.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 60, + ) + store.seed( + rule_id="otherwatch001", + subject_id=other.subject.id, + source_umo="ops:FriendMessage:box", + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="ops", + kind="connect", + expires_at=None, + ) + store.seed( + rule_id="foreignwatch01", + subject_id=other.subject.id, + source_umo="alt:FriendMessage:x", + target_umo="alt:GroupMessage:y", + source_config_id="othercfg", + target_config_id="othercfg", + kind="watch", + expires_at=int(time()) + 60, + ) + manager, authorization, _ = _manager(store=store) + authorization.authorize = AsyncMock( + return_value=SimpleNamespace(allowed=True, effective_role=None) + ) + own_links = await manager.list_links(owner) + assert [item[0].rule_id for item in own_links] == ["ownwatch00001"] + authorization.authorize = AsyncMock( + return_value=SimpleNamespace( + allowed=True, effective_role=Role.INSTANCE_OPERATOR + ) + ) + operator_links = await manager.list_links(owner) + assert {item[0].rule_id for item in operator_links} == { + "ownwatch00001", + "otherwatch001", + } + authorization.authorize = AsyncMock( + return_value=SimpleNamespace(allowed=True, effective_role=Role.ROOT) + ) + root_links = await manager.list_links(owner) + assert {item[0].rule_id for item in root_links} == { + "ownwatch00001", + "otherwatch001", + "foreignwatch01", + } + await manager.terminate() + + +@pytest.mark.asyncio +async def test_unlink_allows_creator_after_revoke_and_scopes_instance_operator(): + owner = _event() + other = _event(sender="other") + store = FakeSessionBridgeStore() + store.seed( + rule_id="ownwatch00001", + subject_id=owner.subject.id, + source_umo=owner.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 60, + ) + store.seed( + rule_id="foreignwatch01", + subject_id=other.subject.id, + source_umo="alt:FriendMessage:x", + target_umo="alt:GroupMessage:y", + source_config_id="othercfg", + target_config_id="othercfg", + kind="watch", + expires_at=int(time()) + 60, + ) + manager, authorization, _ = _manager(store=store) + authorization.authorize = AsyncMock( + return_value=SimpleNamespace(allowed=False, effective_role=None) + ) + assert await manager.unlink(owner, "ownwatch00001") + assert await store.get_session_bridge_rule("ownwatch00001") is None + store.seed( + rule_id="samecfgwatch01", + subject_id=other.subject.id, + source_umo="ops:FriendMessage:box", + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="ops", + kind="connect", + expires_at=None, + ) + authorization.authorize = AsyncMock( + return_value=SimpleNamespace( + allowed=True, effective_role=Role.INSTANCE_OPERATOR + ) + ) + assert await manager.unlink(owner, "samecfgwatch01") + assert await store.get_session_bridge_rule("samecfgwatch01") is None + with pytest.raises(PermissionError): + await manager.unlink(owner, "foreignwatch01") + assert await store.get_session_bridge_rule("foreignwatch01") is not None + await manager.terminate() + + +@pytest.mark.asyncio +async def test_watch_limit_does_not_drop_existing_connect(): + manager, _, _ = _manager(max_watches_per_subject=1) + event = _event() + await manager.watch(event, "target:GroupMessage:one", ttl_seconds=30) + link = await manager.connect(event, "target:GroupMessage:two") + with pytest.raises(ValueError, match="Watch limit exceeded"): + await manager.watch(event, "target:GroupMessage:two", ttl_seconds=30) + current = await manager.connection(event) + assert current is not None + assert current.rule_id == link.rule_id + await manager.terminate() + + +@pytest.mark.asyncio +async def test_connect_limit_does_not_drop_existing_watch(): + manager, _, _ = _manager(max_watches_per_subject=1) + event = _event() + other = _event("source:FriendMessage:other") + await manager.connect(other, "target:GroupMessage:two") + watch = await manager.watch(event, "target:GroupMessage:one", ttl_seconds=30) + with pytest.raises(ValueError, match="Watch limit exceeded"): + await manager.connect(event, "target:GroupMessage:one") + listed = await manager.list_watches(event) + assert [item.rule_id for item in listed] == [watch.rule_id] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_skips_pair_and_discards_invalid_umo(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="pairrule00001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="pair", + expires_at=None, + ) + store.seed( + rule_id="badumorule001", + subject_id=event.subject.id, + source_umo="not-a-umo", + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, _, _ = _manager(store=store) + await manager.restore() + assert await store.get_session_bridge_rule("pairrule00001") is not None + assert await store.get_session_bridge_rule("badumorule001") is None + assert manager._state.total_kind("pair") == 0 + assert manager._state.total_kind("watch") == 0 + await manager.terminate() + + +@pytest.mark.asyncio +async def test_list_links_hides_expired_watches(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="expiredlink01", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) - 10, + ) + store.seed( + rule_id="liveconnect01", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="other:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="connect", + expires_at=None, + ) + manager, _, _ = _manager(store=store) + links = await manager.list_links(event) + assert [item[0].rule_id for item in links] == ["liveconnect01"] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_expired_invalid_subject_does_not_abort(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="expiredbadid1", + subject_id="not-a-subject", + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) - 10, + ) + store.seed( + rule_id="livewatch00001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="other:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, _, _ = _manager(store=store) + await manager.restore() + assert await store.get_session_bridge_rule("expiredbadid1") is None + watches = await manager.list_watches(event) + assert [item.rule_id for item in watches] == ["livewatch00001"] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_continues_after_unexpected_authorize_error(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="brokenwatch01", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + store.seed( + rule_id="goodwatch0001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="other:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, authorization, _ = _manager(store=store) + + async def authorize(_subject, _action, resource, _context): + if resource.umo == "target:GroupMessage:room": + raise RuntimeError("transient") + return SimpleNamespace(allowed=True, effective_role=None) + + authorization.authorize = authorize + await manager.restore() + assert await store.get_session_bridge_rule("brokenwatch01") is not None + watches = await manager.list_watches(event) + assert [item.rule_id for item in watches] == ["goodwatch0001"] + await manager.terminate() + + +@pytest.mark.asyncio +async def test_same_direction_refresh_updates_config_ids(): + store = FakeSessionBridgeStore() + manager, _, _ = _manager(store=store, get_config_id=lambda _: "ops") + event = _event() + target = "target:GroupMessage:room" + watch = await manager.watch(event, target, ttl_seconds=30) + manager._get_config_id = lambda _: "other" + watch_again = await manager.watch(event, target, ttl_seconds=30) + assert watch_again.rule_id == watch.rule_id + watch_row = await store.get_session_bridge_rule(watch.rule_id) + assert watch_row is not None + assert watch_row.source_config_id == "other" + assert watch_row.target_config_id == "other" + + link = await manager.connect(event, target) + manager._get_config_id = lambda _: "third" + link_again = await manager.connect(event, target) + assert link_again.rule_id == link.rule_id + link_row = await store.get_session_bridge_rule(link.rule_id) + assert link_row is not None + assert link_row.source_config_id == "third" + assert link_row.target_config_id == "third" + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_refreshes_stale_config_ids(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="staleconfig01", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="stale", + target_config_id="stale", + kind="watch", + expires_at=int(time()) + 3600, + ) + manager, _, _ = _manager(store=store) + await manager.restore() + row = await store.get_session_bridge_rule("staleconfig01") + assert row is not None + assert row.source_config_id == "default" + assert row.target_config_id == "default" + await manager.terminate() + + +@pytest.mark.asyncio +async def test_one_second_watch_stays_active_after_create(): + manager, _, _ = _manager() + event = _event() + created = time() + watch = await manager.watch(event, "target:GroupMessage:room", ttl_seconds=1) + key = manager._state.store_key(watch) + grant = manager._state.get(key) + assert grant is not None + assert watch.expires_at is not None + assert watch.expires_at >= created + 1 + assert manager._state.grant_active(key, grant, time()) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_restore_keeps_one_connect_per_listener(): + event = _event() + store = FakeSessionBridgeStore() + store.seed( + rule_id="oldconnect0001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:one", + source_config_id="default", + target_config_id="default", + kind="connect", + expires_at=None, + ) + store.seed( + rule_id="newconnect0001", + subject_id=event.subject.id, + source_umo=event.unified_msg_origin, + target_umo="target:GroupMessage:two", + source_config_id="default", + target_config_id="default", + kind="connect", + expires_at=None, + ) + manager, _, _ = _manager(store=store) + await manager.restore() + link = await manager.connection(event) + assert link is not None + assert link.rule_id == "newconnect0001" + assert await store.get_session_bridge_rule("oldconnect0001") is None + assert manager._state.total_kind("connect") == 1 + await manager.terminate() diff --git a/tests/unit/test_session_commands.py b/tests/unit/test_session_commands.py new file mode 100644 index 0000000000..8fc93ec678 --- /dev/null +++ b/tests/unit/test_session_commands.py @@ -0,0 +1,114 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from astrbot.builtin_stars.builtin_commands.commands.session import ( + SessionCommands, + parse_unlink_spec, +) +from astrbot.core.auth.models import AuthContext, Resource, Subject +from astrbot.core.platform.route_identity import PlatformRouteIdentity + + +def _event(umo="source:FriendMessage:sender", *, sender="actor"): + from astrbot.core.platform.message_session import MessageSession + + subject = Subject.im( + platform_instance="source", bot_account_id="bot", sender_id=sender + ) + session = MessageSession.from_str(umo) + return SimpleNamespace( + unified_msg_origin=umo, + subject=subject, + auth_context=AuthContext( + subject=subject, + source="im", + config_id="default", + authenticated=True, + origin_session_resource_id=Resource.session("default", umo).id, + ), + route_identity=PlatformRouteIdentity( + session.platform_id, session.message_type, session.session_id + ), + get_messages=lambda: [], + get_platform_name=lambda: "telegram", + get_sender_id=lambda: sender, + get_sender_name=lambda: sender, + get_self_id=lambda: "bot", + message_obj=SimpleNamespace(message_id="inbound"), + created_at=0.0, + set_result=lambda _result: None, + ) + + +def test_parse_unlink_spec_requires_twelve_hex(): + assert parse_unlink_spec("abcdef123456") == "abcdef123456" + with pytest.raises(ValueError): + parse_unlink_spec("ABCDEF123456") + with pytest.raises(ValueError): + parse_unlink_spec("short") + with pytest.raises(ValueError): + parse_unlink_spec("abcdef123456 extra") + + +@pytest.mark.asyncio +async def test_session_commands_connect_rejects_duration_token(): + replies: list[str] = [] + + async def translate(_event, key, **_kwargs): + replies.append(key) + return key + + context = SimpleNamespace( + bridges=SimpleNamespace(connect=AsyncMock()), + i18n=SimpleNamespace(t=translate), + ) + await SessionCommands(context).connect(_event(), "target:GroupMessage:room 60") + assert replies == ["session.connect.usage"] + context.bridges.connect.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_session_commands_links_and_unlink(): + replies: list[str] = [] + watch = SimpleNamespace( + rule_id="abcdef123456", + source_umo="source:FriendMessage:sender", + target_umo="target:GroupMessage:room", + remaining_seconds=12, + expires_at=1.0, + ) + link = SimpleNamespace( + rule_id="abc123def456", + source_umo="source:FriendMessage:sender", + target_umo="other:GroupMessage:room", + remaining_seconds=0, + expires_at=None, + ) + + async def translate(_event, key, **_kwargs): + replies.append(key) + return key + + manager = SimpleNamespace( + list_links=AsyncMock(return_value=((watch, "watch"), (link, "connect"))), + unlink=AsyncMock(return_value=True), + ) + context = SimpleNamespace( + bridges=SimpleNamespace(_manager=manager), + i18n=SimpleNamespace(t=translate), + ) + commands = SessionCommands(context) + event = _event() + await commands.links(event) + await commands.unlink(event, "abcdef123456") + await commands.unlink(event, "not-an-id") + assert replies == [ + "session.links.ttl_seconds", + "session.links.ttl_unbounded", + "session.links.body", + "session.unlink.ok", + "session.unlink.usage", + ] + manager.unlink.assert_awaited_once_with(event, "abcdef123456")