feat(session): add unbounded connect and linked send

astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json

@@ -15,7 +15,13 @@
     "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.send.invalid": "Usage: /send <umo> <message or attachments>. Keep the command header in the message text.",
+    "session.connect.usage": "Usage: /session connect <UMO>. Omit the UMO to show the current unbounded link.",
+    "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.disconnect.ok": "Session disconnected.",
+    "session.disconnect.missing": "No unbounded link in this session.",
+    "session.send.invalid": "Usage: /send <umo> <message or attachments>, or /send after /session connect. Keep the command header in the message text.",
     "session.send.accepted": "The target platform accepted the message.",
     "session.send.partial": "Only part of the message was accepted. Resending may create duplicates.",
     "session.send.failed": "The target platform rejected the message.",

astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json

@@ -15,7 +15,13 @@
     "session.watches.empty": "你在 {source} 没有有效的监听。",
     "session.watches.body": "你的有效监听：\n{watches}",
     "session.watches.usage": "用法：/session watches [监听会话|this]。监听会话可省略或写 this 表示当前会话。",
-    "session.send.invalid": "用法：/send <umo> <文字或附件>。请保留消息文字中的指令头。",
+    "session.connect.usage": "用法：/session connect <UMO>。省略 UMO 可查看当前无期限连接。",
+    "session.connect.failed": "无法连接，请检查 UMO 和当前连接数量。",
+    "session.connect.ok": "已连接到 {umo}。将无限期转发其新消息。不带 UMO 的 /send 会发往该会话。",
+    "session.connect.status": "当前已连接到 {umo}。将无限期转发其新消息。不带 UMO 的 /send 会发往该会话。",
+    "session.disconnect.ok": "已断开会话连接。",
+    "session.disconnect.missing": "当前会话没有无期限连接。",
+    "session.send.invalid": "用法：/send <umo> <文字或附件>，或先 /session connect 后再 /send。请保留消息文字中的指令头。",
     "session.send.accepted": "目标平台已接受消息。",
     "session.send.partial": "目标平台仅接受了部分消息，重新发送可能造成重复。",
     "session.send.failed": "目标平台拒绝了消息。",

astrbot/builtin_stars/builtin_commands/commands/session.py

@@ -1,6 +1,10 @@
 from astrbot.api import Subject, star
 from astrbot.api.event import AstrMessageEvent
-from astrbot.api.platform import MAX_WATCH_TTL_SECONDS, MIN_WATCH_TTL_SECONDS
+from astrbot.api.platform import (
+    MAX_WATCH_TTL_SECONDS,
+    MIN_WATCH_TTL_SECONDS,
+    MessageSession,
+)
 
 from .reply import reply_i18n
 from .target import resolve_target_umo
@@ -28,6 +32,14 @@ def parse_watch_spec(spec: str, current_umo: str) -> tuple[str, str, int | None]
     return _resolve_listener(parts[0], current_umo), parts[1], int(parts[2])
 
 
+def _is_umo(token: str) -> bool:
+    try:
+        MessageSession.from_str(token)
+    except ValueError, KeyError:
+        return False
+    return bool(token)
+
+
 def parse_unwatch_spec(spec: str, current_umo: str) -> tuple[str, str]:
     """Parse `/session unwatch [listener|this] <target>`."""
     parts = spec.split()
@@ -283,12 +295,74 @@ async def watches(self, event: AstrMessageEvent, spec: str = "") -> None:
             ),
         )
 
-    async def send(
-        self, event: AstrMessageEvent, target_umo: str, content: str
-    ) -> None:
+    async def connect(self, event: AstrMessageEvent, target: str) -> None:
+        """Connect the current session to a target, or show the current link."""
+        target = target.strip()
+        if not target:
+            try:
+                item = await self.context.bridges.connection(event)
+            except PermissionError:
+                await reply_i18n(self.context, event, "session.bridge.denied")
+                return
+            if item is None:
+                await reply_i18n(self.context, event, "session.connect.usage")
+                return
+            await reply_i18n(
+                self.context,
+                event,
+                "session.connect.status",
+                umo=item.target_umo,
+            )
+            return
+        if len(target.split()) != 1:
+            await reply_i18n(self.context, event, "session.connect.usage")
+            return
+        try:
+            item = await self.context.bridges.connect(event, target)
+        except PermissionError:
+            await reply_i18n(self.context, event, "session.bridge.denied")
+            return
+        except ValueError, LookupError:
+            await reply_i18n(self.context, event, "session.connect.failed")
+            return
+        await reply_i18n(
+            self.context,
+            event,
+            "session.connect.ok",
+            umo=item.target_umo,
+        )
+
+    async def disconnect(self, event: AstrMessageEvent) -> None:
+        """Drop the unbounded link owned by the current actor."""
+        try:
+            removed = await self.context.bridges.disconnect(event)
+        except PermissionError:
+            await reply_i18n(self.context, event, "session.bridge.denied")
+            return
+        await reply_i18n(
+            self.context,
+            event,
+            "session.disconnect.ok" if removed else "session.disconnect.missing",
+        )
+
+    async def send(self, event: AstrMessageEvent, spec: str) -> None:
         """Send the event's rich body; parsed text alone loses attachment order."""
+        spec = spec.strip()
+        dest = ""
+        target_in_header = False
+        if spec and _is_umo(spec.split()[0]):
+            dest = spec.split()[0]
+            target_in_header = True
         try:
-            result = await self.context.bridges.send(event, target_umo.strip())
+            if not dest:
+                item = await self.context.bridges.connection(event)
+                if item is None:
+                    await reply_i18n(self.context, event, "session.send.invalid")
+                    return
+                dest = item.target_umo
+            result = await self.context.bridges.send(
+                event, dest, target_in_header=target_in_header
+            )
         except PermissionError:
             await reply_i18n(self.context, event, "session.bridge.denied")
             return

astrbot/builtin_stars/builtin_commands/main.py

@@ -134,6 +134,20 @@ async def session_watches(
         """List active cross-session watches."""
         await self.session_c.watches(event, spec)
 
+    @filter.permission("session.watch")
+    @session.command("connect")
+    async def session_connect(
+        self, event: AstrMessageEvent, target: GreedyStr = GreedyStr("")
+    ) -> None:
+        """Link the current session to another session without expiry."""
+        await self.session_c.connect(event, target)
+
+    @filter.permission("session.read")
+    @session.command("disconnect")
+    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.block")
     @session.command("block")
     async def session_block(
@@ -155,11 +169,10 @@ async def session_unblock(
     async def send_to_session(
         self,
         event: AstrMessageEvent,
-        target_umo: str,
-        content: GreedyStr = GreedyStr(""),
+        spec: GreedyStr = GreedyStr(""),
     ) -> None:
-        """Send text and attachments through the target session's bot account."""
-        await self.session_c.send(event, target_umo, content)
+        """Send text and attachments through a target or the connected session."""
+        await self.session_c.send(event, spec)
 
     @filter.command_group("conversation")
     def conversation(self) -> None:

astrbot/core/platform/message_projection.py

@@ -233,22 +233,31 @@ def envelope_from_event(event: AstrMessageEvent) -> MessageEnvelope:
 
 
 def envelope_from_send_event(
-    event: AstrMessageEvent, target_umo: str
+    event: AstrMessageEvent,
+    target_umo: str,
+    *,
+    target_in_header: bool = True,
 ) -> MessageEnvelope:
     """Remove only the command header, preserving the body and attachment order.
 
-    Command text may span multiple Plain components. The bound target must match
-    the header; a mismatch fails closed rather than forwarding command text.
+    Command text may span multiple Plain components. When the bound target is in
+    the header it must match; a mismatch fails closed rather than forwarding
+    command text. Linked `/send` without a UMO strips only the command token.
     """
     envelope = envelope_from_event(event)
     text = "".join(
         str(part.value)
         for part in envelope.content
         if isinstance(part, PortablePart) and part.kind == ContentKind.TEXT
     )
-    match = re.match(r"""^\s*\S+\s+("[^"]+"|'[^']+'|\S+)[ \t]*""", text)
-    if match is None or match.group(1).strip("\"'") != target_umo:
-        raise ValueError("Cannot locate the send command header")
+    if target_in_header:
+        match = re.match(r"""^\s*\S+\s+("[^"]+"|'[^']+'|\S+)[ \t]*""", text)
+        if match is None or match.group(1).strip("\"'") != target_umo:
+            raise ValueError("Cannot locate the send command header")
+    else:
+        match = re.match(r"""^\s*\S+[ \t]*""", text)
+        if match is None:
+            raise ValueError("Cannot locate the send command header")
     remaining = match.end()
     content: list[PortablePart | NativeContent] = []
     for part in envelope.content:

astrbot/core/platform/session_bridge.py

@@ -40,15 +40,17 @@
 
 @dataclass(frozen=True, slots=True)
 class SessionWatch:
-    """One expiring watch owned by a trusted authorization subject."""
+    """One watch or unbounded link owned by a trusted authorization subject."""
 
     source_umo: str
     target_umo: str
     subject_id: str
-    expires_at: float
+    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()))
 
 
@@ -91,6 +93,7 @@ def __init__(
         )
         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._forwarded: OrderedDict[tuple[str, str, str], None] = OrderedDict()
         self._message_ids: OrderedDict[tuple[str, str, str], str] = OrderedDict()
@@ -208,7 +211,56 @@ async def list_watches(
         await self._notify_expired_watches(expired)
         return items
 
-    async def send(self, event: AstrMessageEvent, target_umo: str) -> DeliveryReceipt:
+    async def connect(
+        self,
+        event: AstrMessageEvent,
+        target_umo: str,
+    ) -> SessionWatch:
+        """Create an unbounded 1:1 link from the current session to the target."""
+        listener = event.unified_msg_origin.strip()
+        target = target_umo.strip()
+        if listener == target:
+            raise ValueError("Source and target sessions must differ")
+        subject, context = self._actor(event)
+        await self._authorize(subject, context, listener, "session.watch")
+        await self._authorize(subject, context, target, "session.watch")
+        if not self._get_capabilities(target).available:
+            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
+
+    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())
+        async with self._lock:
+            return self._links.pop(key, None) is not None
+
+    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())
+        async with self._lock:
+            grant = self._links.get(key)
+        return grant.watch if grant is not None else None
+
+    async def send(
+        self,
+        event: AstrMessageEvent,
+        target_umo: str,
+        *,
+        target_in_header: bool = True,
+    ) -> DeliveryReceipt:
         """Send the command's ordered rich content under target authorization."""
         subject, context = self._actor(event)
         await self._authorize(
@@ -217,7 +269,9 @@ async def send(self, event: AstrMessageEvent, target_umo: str) -> DeliveryReceip
         await self._authorize(subject, context, target_umo, "session.send")
         if not self._get_capabilities(target_umo).proactive:
             raise ValueError("Target adapter does not support proactive delivery")
-        envelope = envelope_from_send_event(event, target_umo)
+        envelope = envelope_from_send_event(
+            event, target_umo, target_in_header=target_in_header
+        )
         if not envelope.content:
             raise ValueError("A message or attachment is required")
 
@@ -234,10 +288,24 @@ 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) -> 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: 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:
         watch = grant.watch
-        key = (watch.subject_id, watch.source_umo, watch.target_umo)
-        if self._watches.get(key) is not grant or watch.expires_at <= monotonic():
+        key = self._store_key(watch)
+        if not self._grant_active(key, grant, monotonic()):
             raise PermissionError("Watch is no longer active")
         await self._authorize(
             grant.subject, grant.context, watch.source_umo, "session.watch"
@@ -257,6 +325,10 @@ async def observe(self, envelope: MessageEnvelope) -> None:
                 (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
             )
         await self._notify_expired_watches(expired)
         for key, grant in watches:
@@ -269,10 +341,7 @@ async def observe(self, envelope: MessageEnvelope) -> None:
                     grant.subject, grant.context, watch.target_umo, "session.watch"
                 )
                 async with self._lock:
-                    if (
-                        self._watches.get(key) is not grant
-                        or watch.expires_at <= monotonic()
-                    ):
+                    if not self._grant_active(key, grant, monotonic()):
                         continue
                     dedup = (watch.source_umo, origin, envelope.source_message_id or "")
                     if envelope.source_message_id:
@@ -308,8 +377,9 @@ async def observe(self, envelope: MessageEnvelope) -> None:
                     )
             except PermissionError:
                 async with self._lock:
-                    if self._watches.get(key) is grant:
-                        self._watches.pop(key)
+                    store = self._store_for(grant.watch)
+                    if store.get(key) is grant:
+                        store.pop(key)
             except asyncio.CancelledError:
                 raise
             except Exception:
@@ -486,7 +556,7 @@ async def _submit(
     def _purge(self, now: float) -> tuple[_WatchGrant, ...]:
         expired: list[_WatchGrant] = []
         for key, grant in tuple(self._watches.items()):
-            if grant.watch.expires_at <= now:
+            if grant.watch.expires_at is not None and grant.watch.expires_at <= now:
                 self._watches.pop(key)
                 expired.append(grant)
         return tuple(expired)
@@ -518,7 +588,10 @@ def _done(done: asyncio.Task) -> None:
     async def _expire_watch(
         self, key: tuple[str, str, str], grant: _WatchGrant
     ) -> None:
-        delay = max(0.0, grant.watch.expires_at - monotonic())
+        expires_at = grant.watch.expires_at
+        if expires_at is None:
+            return
+        delay = max(0.0, expires_at - monotonic())
         await asyncio.sleep(delay)
         async with self._lock:
             if self._watches.get(key) is not grant:
@@ -549,6 +622,7 @@ async def terminate(self) -> None:
             tasks = list(self._expiry_tasks.values())
             self._expiry_tasks.clear()
             self._watches.clear()
+            self._links.clear()
             self._forwarded.clear()
             self._message_ids.clear()
         for task in tasks: