From 32a2f60e941d7a7e7ace6dbd59bd3d1f5a3eade9 Mon Sep 17 00:00:00 2001
From: BegoniaHe <tianyihebegonia1005@gmail.com>
Date: Sun, 13 Sep 2026 22:17:13 +0200
Subject: [PATCH] feat(session): add unbounded connect and linked send

Add /session connect|disconnect as a 1:1 in-memory link that forwards
the target without expiry, and let /send omit the UMO after connect.

AI-Generated: true
Generated-At: 2026-09-13T20:17:07Z
---
 .../.astrbot-plugin/i18n/en-US.json           |   8 +-
 .../.astrbot-plugin/i18n/zh-CN.json           |   8 +-
 .../builtin_commands/commands/session.py      |  84 +++++++++++-
 .../builtin_stars/builtin_commands/main.py    |  21 ++-
 astrbot/core/platform/message_projection.py   |  21 ++-
 astrbot/core/platform/session_bridge.py       | 102 +++++++++++++--
 astrbot/core/star/plugin_context.py           |  24 +++-
 docs/en/dev/star/guides/send-message.md       |  15 ++-
 docs/en/use/authorization.md                  |   2 +-
 docs/en/use/command.md                        |   5 +-
 docs/en/use/webui.md                          |   2 +-
 docs/zh/dev/star/guides/send-message.md       |   7 +-
 docs/zh/use/authorization.md                  |   2 +-
 docs/zh/use/command.md                        |   5 +-
 docs/zh/use/webui.md                          |   2 +-
 tests/unit/test_builtin_command_extensions.py |  18 +++
 tests/unit/test_message_protocol.py           | 120 +++++++++++++++++-
 17 files changed, 398 insertions(+), 48 deletions(-)

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 26bc5cb122..3dde6fa3c1 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
@@ -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.",
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 9ce03d1ed1..9ac6bbe909 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
@@ -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": "目标平台拒绝了消息。",
diff --git a/astrbot/builtin_stars/builtin_commands/commands/session.py b/astrbot/builtin_stars/builtin_commands/commands/session.py
index ac15890abf..3b9aac15cb 100644
--- a/astrbot/builtin_stars/builtin_commands/commands/session.py
+++ b/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
diff --git a/astrbot/builtin_stars/builtin_commands/main.py b/astrbot/builtin_stars/builtin_commands/main.py
index 76bc842bc1..4f4d7da24a 100644
--- a/astrbot/builtin_stars/builtin_commands/main.py
+++ b/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:
diff --git a/astrbot/core/platform/message_projection.py b/astrbot/core/platform/message_projection.py
index 9ed847a851..751fc8ccdb 100644
--- a/astrbot/core/platform/message_projection.py
+++ b/astrbot/core/platform/message_projection.py
@@ -233,12 +233,16 @@ 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(
@@ -246,9 +250,14 @@ def envelope_from_send_event(
         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:
diff --git a/astrbot/core/platform/session_bridge.py b/astrbot/core/platform/session_bridge.py
index 99d631feee..6450271a2d 100644
--- a/astrbot/core/platform/session_bridge.py
+++ b/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:
diff --git a/astrbot/core/star/plugin_context.py b/astrbot/core/star/plugin_context.py
index 35f586aa54..ecfcde1dcf 100644
--- a/astrbot/core/star/plugin_context.py
+++ b/astrbot/core/star/plugin_context.py
@@ -202,9 +202,29 @@ async def list(
         """List active watches owned by the event's trusted actor."""
         return await self._manager.list_watches(event, source_umo=source_umo)
 
-    async def send(self, event: AstrMessageEvent, target_umo: str) -> DeliveryReceipt:
+    async def connect(self, event: AstrMessageEvent, target_umo: str) -> SessionWatch:
+        """Authorize and create an unbounded link from the current session."""
+        return await self._manager.connect(event, target_umo)
+
+    async def disconnect(self, event: AstrMessageEvent) -> bool:
+        """Remove the unbounded link owned by the event's trusted actor."""
+        return await self._manager.disconnect(event)
+
+    async def connection(self, event: AstrMessageEvent) -> SessionWatch | None:
+        """Return the unbounded link for the current actor in this session."""
+        return await self._manager.connection(event)
+
+    async def send(
+        self,
+        event: AstrMessageEvent,
+        target_umo: str,
+        *,
+        target_in_header: bool = True,
+    ) -> DeliveryReceipt:
         """Authorize and send the command body, attachments, and quote."""
-        return await self._manager.send(event, target_umo)
+        return await self._manager.send(
+            event, target_umo, target_in_header=target_in_header
+        )
 
 
 class ModelCapability:
diff --git a/docs/en/dev/star/guides/send-message.md b/docs/en/dev/star/guides/send-message.md
index ee370296a2..2aa95427fe 100644
--- a/docs/en/dev/star/guides/send-message.md
+++ b/docs/en/dev/star/guides/send-message.md
@@ -78,14 +78,17 @@ async def watch_room(self, event: AstrMessageEvent, target_umo: str):
 - `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.
 - `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.
-- `send(event, target_umo)`: deliver the current message body and attachments after stripping the command header; returns `DeliveryReceipt`.
+- `connect(event, target_umo)`: create an unbounded 1:1 link from the current session; returns `SessionWatch` with no expiry.
+- `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 stay in
-memory and disappear when they expire or the process restarts. 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.
+`session.send`, and both sessions must share a configuration. Watches and
+links stay in memory and disappear when they expire or the process restarts.
+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.
 
 ## Rich-Media Chains
 
diff --git a/docs/en/use/authorization.md b/docs/en/use/authorization.md
index 2f71142499..03e68c9065 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. Authorization is checked again before submission. Stopping a watch 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. 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 16cd66f626..a34f3c9707 100644
--- a/docs/en/use/command.md
+++ b/docs/en/use/command.md
@@ -71,9 +71,12 @@ The user ID from `/session info` can be granted current-session `session_admin`
 - `/session watch [listener|this] <target> [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 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] <target>`: Stop a matching watch that you own. Omit the listener or write `this` for the current session.
+- `/session connect <UMO>`: 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 disconnect`: Drop the unbounded link in the current session; requires `session.read`.
 - `/send <UMO> [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 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 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 per actor. Authorization is checked again for each forwarded message, so revocation stops the watch. 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. 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.
 
 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/en/use/webui.md b/docs/en/use/webui.md
index 1ff80136ee..91fc500400 100644
--- a/docs/en/use/webui.md
+++ b/docs/en/use/webui.md
@@ -144,7 +144,7 @@ WebUI supports multiple Dashboard accounts. First startup creates a bootstrap `r
 
 **More → Authorization** opens `/authorization`. Dashboard accounts, IM session owners, and `/admin grant` are not the same identity. Group grants, step-up, and the role table are in [Authorization](./authorization).
 
-Cross-session watches and `/send` have no Dashboard management surface in this stage. After binding `instance_operator`, create, list, and stop watches only with the [built-in IM commands](./command#cross-session-watches-and-sending). The command-management page can enable or disable those commands, but it does not list active watches.
+Cross-session watches, connects, and `/send` have no Dashboard management surface in this stage. After binding `instance_operator`, create, list, and stop watches or links only with the [built-in IM commands](./command#cross-session-watches-and-sending). The command-management page can enable or disable those commands, but it does not list active watches or links.
 
 The developer model is in [Architecture](/en/dev/architecture#unified-authorization).
 
diff --git a/docs/zh/dev/star/guides/send-message.md b/docs/zh/dev/star/guides/send-message.md
index 36e26f3c5a..c44d7a9f43 100644
--- a/docs/zh/dev/star/guides/send-message.md
+++ b/docs/zh/dev/star/guides/send-message.md
@@ -70,9 +70,12 @@ async def watch_room(self, event: AstrMessageEvent, target_umo: str):
 - `watch(event, target_umo, *, source_umo=None, ttl_seconds=None)`：用事件上的可信主体创建有期限监听，返回 `SessionWatch`。`source_umo` 是接收转发的监听会话，缺省为当前会话；`ttl_seconds` 范围 60–864000，缺省 43200。到期后会向监听会话发送结束通知。
 - `unwatch(event, target_umo, *, source_umo=None)`：停止当前主体创建的对应监听。
 - `list(event, *, source_umo=None)`：列出当前主体的有效监听。`source_umo` 是监听会话，缺省为当前会话。
-- `send(event, target_umo)`：把当前消息去掉指令头后的正文和附件投递到目标会话，返回 `DeliveryReceipt`。
+- `connect(event, target_umo)`：从当前会话创建无期限的一对一连接，返回没有过期时间的 `SessionWatch`。
+- `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`，且两个会话属于同一配置。监听和连接保存在内存中，到期或重启后清空。不要自己构造 `SessionBridgeManager`。`SessionWatch` 和时长常量可从 `astrbot.api.platform` 导入。当前没有 Dashboard 管理面，插件也不应假设存在对应 HTTP API。
 
 ## 富媒体消息链
 
diff --git a/docs/zh/use/authorization.md b/docs/zh/use/authorization.md
index fb2411ab9a..c5799a4874 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` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。每次投递前重新检查权限，停止监听或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。
+`/session watch` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。`/session connect` 用同一套授权规则创建无期限的一对一连接。每次投递前重新检查权限，停止监听或连接、或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。
 
 ## 二次验证（step-up）
 
diff --git a/docs/zh/use/command.md b/docs/zh/use/command.md
index 27d4bca242..61a978f440 100644
--- a/docs/zh/use/command.md
+++ b/docs/zh/use/command.md
@@ -71,9 +71,12 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、
 - `/session watch [监听会话|this] <被监听会话> [秒数]`：把被监听会话后续收到的消息转发到监听会话，需要 `session.watch`。监听会话可省略或写 `this`，表示当前会话。时长 60–864000 秒（最多 10 天），默认 43200 秒（12 小时）。到期后会在监听会话发送结束通知。
 - `/session watches [监听会话|this]`：查看你创建的监听和剩余时间。省略参数或写 `this` 表示当前会话作为监听端。
 - `/session unwatch [监听会话|this] <被监听会话>`：停止你创建的指定监听。监听会话可省略或写 `this`。
+- `/session connect <UMO>`：把当前会话无期限连接到目标会话，需要 `session.watch`。目标会话的新消息会一直转发，直到 `/session disconnect` 或进程重启。每个主体在每个监听会话只能有一条连接，再次连接会替换目标。省略 UMO 可查看当前连接。
+- `/session disconnect`：断开当前会话的无期限连接，需要 `session.read`。
 - `/send <UMO> [内容]`：借助目标平台的 Bot 账号发送文字和同一条消息中的附件，需要 `session.send`。可以只附图片而不填写正文；不会占用 `reply` 指令。
+- `/send [内容]`：在 `/session connect` 之后，不写 UMO 也会发往已连接的目标会话。可以只附图片。
 
-监听和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。监听保存在内存中，到期或重启后清空，每人最多 16 项。每次转发都会重新检查权限，撤权后停止。对同一对会话再次 `/session watch` 会重置时长。
+监听、连接和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。监听和连接保存在内存中，到期或重启后清空，每人最多 16 条监听和 16 条连接。每次转发都会重新检查权限，撤权后停止。对同一对会话再次 `/session watch` 会重置时长。
 
 正文保留消息链中的图文先后顺序；不能混排的目标拆成多条消息。跨平台提及转成文字，引用优先通过已接受消息的 ID 映射还原；映射不存在时附引用摘要。无法解析的附件和不支持的原生内容会保留文字占位。平台自己的卡片、私有语法和小程序不能保证在别的平台重现。
 
diff --git a/docs/zh/use/webui.md b/docs/zh/use/webui.md
index ebba67db06..ca1778282b 100644
--- a/docs/zh/use/webui.md
+++ b/docs/zh/use/webui.md
@@ -138,7 +138,7 @@ WebUI 支持多个 Dashboard 账户。首次启动会创建 bootstrap `root` 账
 
 侧栏 **更多功能 → 授权管理** 打开 `/authorization`。Dashboard 账户、IM 会话 owner 和 `/admin grant` 不是同一套身份；群聊授权、二次验证（step-up）和角色表见 [授权管理](./authorization)。
 
-跨会话监听和 `/send` 目前没有 Dashboard 管理面。绑定 `instance_operator` 后，只能在 IM 或 WebChat 里用 [内置指令](./command#跨会话监听与发送) 创建、查看和停止监听。指令管理页可以启用或禁用这些指令，但不会列出活跃监听。
+跨会话监听、连接和 `/send` 目前没有 Dashboard 管理面。绑定 `instance_operator` 后，只能在 IM 或 WebChat 里用 [内置指令](./command#跨会话监听与发送) 创建、查看和停止监听或连接。指令管理页可以启用或禁用这些指令，但不会列出活跃监听或连接。
 
 开发模型见[项目架构](/dev/architecture#统一授权系统)。
 
diff --git a/tests/unit/test_builtin_command_extensions.py b/tests/unit/test_builtin_command_extensions.py
index 502bb530ad..9ce1a35cc2 100644
--- a/tests/unit/test_builtin_command_extensions.py
+++ b/tests/unit/test_builtin_command_extensions.py
@@ -261,6 +261,8 @@ def test_all_builtin_extension_commands_use_native_command_schemas():
         "session_watch",
         "session_unwatch",
         "session_watches",
+        "session_connect",
+        "session_disconnect",
         "session_block",
         "session_unblock",
         "send_to_session",
@@ -1097,6 +1099,8 @@ def command_names(group: CommandGroupFilter) -> set[str]:
             "watch",
             "unwatch",
             "watches",
+            "connect",
+            "disconnect",
         },
         "conversation": {
             "create",
@@ -1161,6 +1165,8 @@ def test_non_public_builtin_commands_declare_the_planned_actions():
         "session_watch": "session.watch",
         "session_unwatch": "session.read",
         "session_watches": "session.read",
+        "session_connect": "session.watch",
+        "session_disconnect": "session.read",
         "session_block": "session.block",
         "session_unblock": "session.block",
         "send_to_session": "session.send",
@@ -1225,6 +1231,18 @@ def test_normalized_builtin_paths_resolve_and_legacy_subcommands_do_not():
         "clear": False,
     }
 
+    connected = engine.resolve("session connect target:GroupMessage:room")
+    assert connected.resolution.command_path == ("session", "connect")
+    connected_entry = connected.resolution.entries[0]
+    assert dict(engine.bind(connected_entry, connected).values) == {
+        "target": "target:GroupMessage:room"
+    }
+
+    linked_send = engine.resolve("send hello")
+    assert linked_send.resolution.command_path == ("send",)
+    send_entry = linked_send.resolution.entries[0]
+    assert dict(engine.bind(send_entry, linked_send).values) == {"spec": "hello"}
+
     bot_leave = engine.resolve("bot leave --confirm")
     assert bot_leave.resolution.command_path == ("bot", "leave")
     bot_entry = bot_leave.resolution.entries[0]
diff --git a/tests/unit/test_message_protocol.py b/tests/unit/test_message_protocol.py
index 578a0f4046..c9de9bbd34 100644
--- a/tests/unit/test_message_protocol.py
+++ b/tests/unit/test_message_protocol.py
@@ -1043,10 +1043,118 @@ async def test_terminate_clears_watches_and_message_maps():
     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
+
+    assert _is_umo("target:GroupMessage:room")
+    assert not _is_umo("hello")
+    assert not _is_umo("")
+
+
+def test_send_projection_strips_command_only_when_target_not_in_header():
+    from astrbot.core.message.components import Plain
+    from astrbot.core.platform.message_projection import envelope_from_send_event
+
+    event = _event(components=[Plain("/send hello there")])
+    envelope = envelope_from_send_event(
+        event, "target:GroupMessage:room", target_in_header=False
+    )
+    assert [
+        part.value for part in envelope.content if part.kind == ContentKind.TEXT
+    ] == ["hello there"]
+    with pytest.raises(ValueError):
+        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
+
+    replies: list[str] = []
+    link = SimpleNamespace(target_umo="target:GroupMessage:room", expires_at=None)
+
+    async def translate(_event, key, **_kwargs):
+        replies.append(key)
+        return key
+
+    context = SimpleNamespace(
+        bridges=SimpleNamespace(
+            connect=AsyncMock(return_value=link),
+            connection=AsyncMock(return_value=link),
+            disconnect=AsyncMock(return_value=True),
+            send=AsyncMock(return_value=SimpleNamespace(status="accepted")),
+        ),
+        i18n=SimpleNamespace(t=translate),
+    )
+    event = _event()
+    event.set_result = lambda _result: None
+    commands = SessionCommands(context)
+    await commands.connect(event, "")
+    await commands.connect(event, "target:GroupMessage:room")
+    await commands.disconnect(event)
+    await commands.send(event, "")
+    await commands.send(event, "target:GroupMessage:room hello")
+    assert replies == [
+        "session.connect.status",
+        "session.connect.ok",
+        "session.disconnect.ok",
+        "session.send.accepted",
+        "session.send.accepted",
+    ]
+    context.bridges.send.assert_any_await(
+        event, "target:GroupMessage:room", target_in_header=False
+    )
+    context.bridges.send.assert_any_await(
+        event, "target:GroupMessage:room", target_in_header=True
+    )
+
+
 @pytest.mark.asyncio
 async def test_session_commands_report_denied_without_actor():
     from astrbot.builtin_stars.builtin_commands.commands.session import SessionCommands
@@ -1061,6 +1169,9 @@ async def translate(_event, key, **_kwargs):
         bridges=SimpleNamespace(
             unwatch=AsyncMock(side_effect=PermissionError("denied")),
             list=AsyncMock(side_effect=PermissionError("denied")),
+            connect=AsyncMock(side_effect=PermissionError("denied")),
+            disconnect=AsyncMock(side_effect=PermissionError("denied")),
+            connection=AsyncMock(side_effect=PermissionError("denied")),
         ),
         i18n=SimpleNamespace(t=translate),
     )
@@ -1069,4 +1180,11 @@ async def translate(_event, key, **_kwargs):
     commands = SessionCommands(context)
     await commands.unwatch(event, "target:GroupMessage:room")
     await commands.watches(event)
-    assert replies == ["session.bridge.denied", "session.bridge.denied"]
+    await commands.connect(event, "target:GroupMessage:room")
+    await commands.disconnect(event)
+    assert replies == [
+        "session.bridge.denied",
+        "session.bridge.denied",
+        "session.bridge.denied",
+        "session.bridge.denied",
+    ]