esengine--deepseek-reasonix/desktop/bot_bridge.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

	"reasonix/internal/bot"
	"reasonix/internal/event"
)

// errDriveBusy signals that a takeover drive could not start because the target
// controller was already running a turn. The hub translates it into a
// user-facing "session busy" message rather than a generic drive failure.
var errDriveBusy = errors.New("desktop session busy")

// botBridgeHub 是 bot 网关对桌面端的"上帝视角"桥（bot.DesktopBridge 的实现）。
//
// 职责边界（刻意保持窄）：
//   - 观察：tabEventSink.Emit 把每个桌面会话的事件旁路到 observe；hub 记录
//     待审批/待回答项，并把审批请求、任务完成/出错推送给订阅聊天。
//   - 遥控审批：/desktop approve|deny|answer 经 App.ApproveTab /
//     AnswerQuestionForTab 按 tab 寻址回写。controller 侧幂等（先到者赢），
//     桌面 UI 与远端并发应答互不干扰。
//   - 不做：向桌面会话注入输入、抢占写 lease。将来的显式接管在
//     bot.DesktopBridge 上扩展，底层走 internal/control 的接管端口。
//
// observe 跑在 controller 的事件 goroutine 上，绝不能做网络调用——通知统一
// 进有界队列由 worker 异步发送，队列满时丢弃并告警。
type botBridgeHub struct {
	sessions   func() []bot.DesktopSessionInfo
	approveTab func(tabID, id string, allow, session, persist bool)
	answerTab  func(tabID, id string, answers []QuestionAnswer)
	notify     func(ctx context.Context, connectionID, domain string, msg bot.OutboundMessage) (bot.SendResult, error)
	// drive 把一条远程文本提交为 tab 的新 turn，并把该 turn 的输出转发回 route。
	drive func(tabID, text string, route bot.DesktopWatchRoute) error
	// announce 往会话 transcript 里发一条 Notice，让桌面用户看到接管状态变化。
	announce func(tabID, text string)
	// persistWatchers 把订阅全集回写用户配置（bot.desktop_watchers）。
	persistWatchers func(routes []bot.DesktopWatchRoute) error
	// takeoverChanged 通知桌面前端刷新（TabMeta.RemoteControlled 变化）。
	takeoverChanged func()
	logger          *slog.Logger

	mu       sync.Mutex
	watchers map[string]bot.DesktopWatchRoute
	pending  map[string]desktopPendingPrompt
	// takeovers: tabID -> 驾驶该会话的聊天路由；takeoverTabs: routeKey -> tabID。
	takeovers    map[string]bot.DesktopWatchRoute
	takeoverTabs map[string]string
	// watchSeq 单调递增，标记订阅快照的新旧；persist 时用它丢弃过期写入。
	watchSeq uint64
	// watchPersistDirty keeps a failed local mutation authoritative in memory;
	// a later runtime refresh must not silently restore the older disk snapshot.
	watchPersistDirty bool

	// persistMu 串行化订阅落盘，并保证只写最新快照（见 SetWatch）。
	persistMu      sync.Mutex
	lastPersistSeq uint64

	queue chan desktopBridgeNotification
}

type desktopPendingPrompt struct {
	tabID     string
	kind      string // "approval" | "ask"
	tool      string
	subject   string
	questions []event.AskQuestion
}

// desktopBridgeNotification 是一条待推送的桌面事件。text 与 card 都按订阅路由
// 现做，因此群聊(多用户)与私聊可给出不同详略——命令行/错误详情只发私聊，群里
// 只给摘要。route 非 nil 时定向发给该聊天(不看 watch 订阅)，用于接管收回等必达通知。
type desktopBridgeNotification struct {
	text  func(route bot.DesktopWatchRoute) string
	card  func(route bot.DesktopWatchRoute) *bot.InteractiveCard
	route *bot.DesktopWatchRoute
}

// isSharedChat 判断一个聊天是否为多用户场景(群/话题群/服务器频道)。私聊/单聊
// 只有操作者本人,可安全展示命令行等敏感详情。
func isSharedChat(ct bot.ChatType) bool {
	return ct != bot.ChatDM && ct != bot.ChatDirect
}

func constText(s string) func(bot.DesktopWatchRoute) string {
	return func(bot.DesktopWatchRoute) string { return s }
}

const (
	botBridgeQueueSize     = 64
	botBridgeSendTimeout   = 15 * time.Second
	botBridgeSubjectLimit  = 200
	botBridgePendingLimit  = 200
	botBridgeErrTextLimit  = 300
	botBridgePromptPreview = 500
)

// botBridgeDeps 打包 hub 对宿主(App)的全部依赖，便于测试注入。
type botBridgeDeps struct {
	sessions        func() []bot.DesktopSessionInfo
	approveTab      func(tabID, id string, allow, session, persist bool)
	answerTab       func(tabID, id string, answers []QuestionAnswer)
	notify          func(ctx context.Context, connectionID, domain string, msg bot.OutboundMessage) (bot.SendResult, error)
	drive           func(tabID, text string, route bot.DesktopWatchRoute) error
	announce        func(tabID, text string)
	persistWatchers func(routes []bot.DesktopWatchRoute) error
	takeoverChanged func()
	logger          *slog.Logger
}

func newBotBridgeHub(deps botBridgeDeps) *botBridgeHub {
	logger := deps.logger
	if logger == nil {
		logger = slog.Default()
	}
	h := &botBridgeHub{
		sessions:        deps.sessions,
		approveTab:      deps.approveTab,
		answerTab:       deps.answerTab,
		notify:          deps.notify,
		drive:           deps.drive,
		announce:        deps.announce,
		persistWatchers: deps.persistWatchers,
		takeoverChanged: deps.takeoverChanged,
		logger:          logger.With("component", "bot_bridge"),
		watchers:        make(map[string]bot.DesktopWatchRoute),
		pending:         make(map[string]desktopPendingPrompt),
		takeovers:       make(map[string]bot.DesktopWatchRoute),
		takeoverTabs:    make(map[string]string),
		queue:           make(chan desktopBridgeNotification, botBridgeQueueSize),
	}
	go h.run()
	return h
}

// observe 接收某个桌面会话的一条事件。在 controller 事件 goroutine 上运行，
// 只做内存记账和入队，不做任何阻塞调用。
func (h *botBridgeHub) observe(tabID string, e event.Event) {
	switch e.Kind {
	case event.ApprovalRequest:
		h.mu.Lock()
		h.rememberPendingLocked(e.Approval.ID, desktopPendingPrompt{
			tabID:   tabID,
			kind:    "approval",
			tool:    e.Approval.Tool,
			subject: truncateForBridge(e.Approval.Subject, botBridgeSubjectLimit),
		})
		watching := len(h.watchers) > 0
		h.mu.Unlock()
		if watching {
			h.enqueue(h.approvalNotification(tabID, e.Approval))
		}
	case event.AskRequest:
		h.mu.Lock()
		h.rememberPendingLocked(e.Ask.ID, desktopPendingPrompt{
			tabID:     tabID,
			kind:      "ask",
			questions: e.Ask.Questions,
		})
		watching := len(h.watchers) > 0
		h.mu.Unlock()
		if watching {
			h.enqueue(h.askNotification(tabID, e.Ask))
		}
	case event.TurnDone:
		h.mu.Lock()
		for id, p := range h.pending {
			if p.tabID == tabID {
				delete(h.pending, id)
			}
		}
		watching := len(h.watchers) > 0
		h.mu.Unlock()
		if !watching {
			return
		}
		if e.Err != nil && strings.Contains(e.Err.Error(), "context canceled") {
			// 桌面端主动停止的任务不推送，避免正常操作变成噪音。
			return
		}
		h.enqueue(h.turnDoneNotification(tabID, e.Err))
	}
}

// rememberPendingLocked 记录待处理项；容量兜底防泄漏（正常路径 TurnDone 会清理）。
func (h *botBridgeHub) rememberPendingLocked(id string, p desktopPendingPrompt) {
	if strings.TrimSpace(id) == "" {
		return
	}
	if len(h.pending) >= botBridgePendingLimit {
		h.pending = make(map[string]desktopPendingPrompt)
	}
	h.pending[id] = p
}

func (h *botBridgeHub) enqueue(n desktopBridgeNotification) {
	select {
	case h.queue <- n:
	default:
		h.logger.Warn("desktop bridge notification queue full; dropping")
	}
}

func (h *botBridgeHub) run() {
	for n := range h.queue {
		h.deliver(n)
	}
}

func (h *botBridgeHub) deliver(n desktopBridgeNotification) {
	h.mu.Lock()
	var routes []bot.DesktopWatchRoute
	if n.route != nil {
		routes = []bot.DesktopWatchRoute{*n.route}
	} else {
		routes = h.watcherRoutesLocked()
	}
	notify := h.notify
	h.mu.Unlock()
	if notify == nil || len(routes) == 0 {
		return
	}
	// Fan out per route: a single slow/hung connection must not hold the queue
	// worker for its full timeout and back-pressure everyone else's approvals.
	var wg sync.WaitGroup
	for _, route := range routes {
		wg.Add(1)
		go func(route bot.DesktopWatchRoute) {
			defer wg.Done()
			msg := bot.OutboundMessage{
				ChatID:   route.ChatID,
				ChatType: route.ChatType,
			}
			if n.text != nil {
				msg.Text = n.text(route)
			}
			if n.card != nil {
				msg.Card = n.card(route)
			}
			ctx, cancel := context.WithTimeout(context.Background(), botBridgeSendTimeout)
			defer cancel()
			if _, err := notify(ctx, route.ConnectionID, route.Domain, msg); err != nil {
				h.logger.Warn("desktop bridge notification send failed", "platform", route.Platform, "err", err)
			}
		}(route)
	}
	wg.Wait()
}

// tabLabel 把 tabID 解析成人类可读的会话名。
func (h *botBridgeHub) tabLabel(tabID string) string {
	if s, ok := h.sessionByTabID(tabID); ok {
		if label := strings.TrimSpace(s.Label); label != "" {
			return label
		}
		if title := strings.TrimSpace(s.Topic); title != "" {
			return title
		}
	}
	return "(未命名会话)"
}

func (h *botBridgeHub) sessionByTabID(tabID string) (bot.DesktopSessionInfo, bool) {
	if h.sessions == nil {
		return bot.DesktopSessionInfo{}, false
	}
	for _, s := range h.sessions() {
		if s.TabID == tabID {
			return s, true
		}
	}
	return bot.DesktopSessionInfo{}, false
}

func (h *botBridgeHub) approvalNotification(tabID string, approval event.Approval) desktopBridgeNotification {
	label := h.tabLabel(tabID)
	// The approval subject is the pending command line; only reveal it in a
	// private chat. In a shared chat show the tool name and point the operator
	// to the desktop / a DM instead of leaking the command to the whole group.
	subjectFor := func(route bot.DesktopWatchRoute) string {
		if isSharedChat(route.ChatType) {
			return "(命令详情仅在桌面端或私聊显示)"
		}
		return truncateForBridge(approval.Subject, botBridgeSubjectLimit)
	}
	return desktopBridgeNotification{
		text: func(route bot.DesktopWatchRoute) string {
			return fmt.Sprintf("⚠️ 桌面会话「%s」需要批准操作\n工具: %s\n操作: %s\n\nID: `%s`\n用 /desktop approve %s 批准，/desktop deny %s 拒绝。桌面端先处理则以先到者为准。",
				label, approval.Tool, subjectFor(route), approval.ID, approval.ID, approval.ID)
		},
		card: func(route bot.DesktopWatchRoute) *bot.InteractiveCard {
			return &bot.InteractiveCard{
				Header: "桌面会话需要批准",
				Elements: []bot.InteractiveCardElement{
					{Tag: "markdown", Content: fmt.Sprintf("**会话**: %s\n\n**工具**: %s\n\n**操作**: %s\n\nID: `%s`", label, approval.Tool, subjectFor(route), approval.ID)},
					{Tag: "action", Extra: map[string]any{
						"actions": []map[string]any{
							desktopCardButton("允许一次", "primary", "/desktop approve "+approval.ID, route),
							desktopCardButton("拒绝", "danger", "/desktop deny "+approval.ID, route),
						},
					}},
				},
			}
		},
	}
}

func (h *botBridgeHub) askNotification(tabID string, ask event.Ask) desktopBridgeNotification {
	label := h.tabLabel(tabID)
	var b strings.Builder
	fmt.Fprintf(&b, "❓ 桌面会话「%s」在等待回答:\n", label)
	for i, q := range ask.Questions {
		fmt.Fprintf(&b, "\n**%d. %s**\n", i+1, truncateForBridge(q.Prompt, botBridgePromptPreview))
		for j, opt := range q.Options {
			fmt.Fprintf(&b, "  %d. %s\n", j+1, opt.Label)
		}
	}
	fmt.Fprintf(&b, "\nID: `%s`\n用 /desktop answer %s <选项编号或文本> 回答；桌面端先处理则以先到者为准。", ask.ID, ask.ID)
	privateText := b.String()
	sharedText := fmt.Sprintf("❓ 桌面会话「%s」正在等待回答（问题详情仅在桌面端或私聊显示）。\n\nID: `%s`", label, ask.ID)
	textFor := func(route bot.DesktopWatchRoute) string {
		if isSharedChat(route.ChatType) {
			return sharedText
		}
		return privateText
	}

	var card func(route bot.DesktopWatchRoute) *bot.InteractiveCard
	if len(ask.Questions) == 1 && len(ask.Questions[0].Options) > 0 {
		options := ask.Questions[0].Options
		card = func(route bot.DesktopWatchRoute) *bot.InteractiveCard {
			if isSharedChat(route.ChatType) {
				return nil
			}
			actions := make([]map[string]any, 0, len(options))
			for i, opt := range options {
				optLabel := strings.TrimSpace(opt.Label)
				if optLabel == "" {
					optLabel = fmt.Sprintf("选项 %d", i+1)
				}
				actions = append(actions, desktopCardButton(optLabel, "primary", fmt.Sprintf("/desktop answer %s %d", ask.ID, i+1), route))
			}
			return &bot.InteractiveCard{
				Header: "桌面会话在等待回答",
				Elements: []bot.InteractiveCardElement{
					{Tag: "markdown", Content: privateText},
					{Tag: "action", Extra: map[string]any{"actions": actions}},
				},
			}
		}
	}
	return desktopBridgeNotification{text: textFor, card: card}
}

func (h *botBridgeHub) turnDoneNotification(tabID string, err error) desktopBridgeNotification {
	label := h.tabLabel(tabID)
	if err != nil {
		// Error text can contain paths/tokens; only detail it in a private chat.
		return desktopBridgeNotification{text: func(route bot.DesktopWatchRoute) string {
			if isSharedChat(route.ChatType) {
				return fmt.Sprintf("❌ 桌面会话「%s」任务出错（详情见桌面端或私聊）。", label)
			}
			return fmt.Sprintf("❌ 桌面会话「%s」任务出错: %s", label, truncateForBridge(err.Error(), botBridgeErrTextLimit))
		}}
	}
	return desktopBridgeNotification{text: constText(fmt.Sprintf("✅ 桌面会话「%s」任务完成。", label))}
}

func desktopCardButton(label, style, command string, route bot.DesktopWatchRoute) map[string]any {
	return map[string]any{
		"tag":  "button",
		"text": map[string]string{"tag": "plain_text", "content": label},
		"type": style,
		"value": map[string]string{
			"command":   command,
			"chat_type": string(route.ChatType),
		},
	}
}

func truncateForBridge(s string, limit int) string {
	s = strings.TrimSpace(s)
	runes := []rune(s)
	if len(runes) <= limit {
		return s
	}
	return string(runes[:limit]) + "…"
}

// ---- bot.DesktopBridge 实现 ----

func (h *botBridgeHub) Sessions() []bot.DesktopSessionInfo {
	if h.sessions == nil {
		return nil
	}
	sessions := h.sessions()
	h.mu.Lock()
	byTab := make(map[string][]bot.DesktopPendingInfo, len(h.pending))
	for id, p := range h.pending {
		byTab[p.tabID] = append(byTab[p.tabID], bot.DesktopPendingInfo{ID: id, Kind: p.kind, Tool: p.tool})
	}
	h.mu.Unlock()
	for i := range sessions {
		if pend := byTab[sessions[i].TabID]; len(pend) > 0 {
			sort.Slice(pend, func(a, b int) bool { return pend[a].ID < pend[b].ID })
			sessions[i].Pending = pend
		}
	}
	return sessions
}

func (h *botBridgeHub) SetWatch(route bot.DesktopWatchRoute, enable bool) error {
	h.mu.Lock()
	if enable {
		h.watchers[route.Key()] = route
	} else {
		delete(h.watchers, route.Key())
	}
	h.watchSeq++
	h.watchPersistDirty = true
	seq := h.watchSeq
	routes := h.watcherRoutesLocked()
	persist := h.persistWatchers
	h.mu.Unlock()
	if persist == nil {
		return nil
	}
	// Serialize persists and drop stale ones: two concurrent SetWatch calls
	// (different connections) compute snapshots under h.mu but write config
	// outside it, so their writes could otherwise reorder and let an older
	// snapshot clobber a newer one, silently losing a subscription.
	h.persistMu.Lock()
	defer h.persistMu.Unlock()
	if seq <= h.lastPersistSeq {
		return nil
	}
	if err := persist(routes); err != nil {
		return err
	}
	h.lastPersistSeq = seq
	h.mu.Lock()
	if h.watchSeq == seq {
		h.watchPersistDirty = false
	}
	h.mu.Unlock()
	return nil
}

func (h *botBridgeHub) watcherVersion() uint64 {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.watchSeq
}

// seedWatchers applies a config snapshot only if no watch command changed the
// runtime after the config read began. Fresh external config edits still apply;
// stale refreshes and failed local persists do not erase newer runtime state.
func (h *botBridgeHub) seedWatchers(routes []bot.DesktopWatchRoute, expectedSeq uint64) {
	h.persistMu.Lock()
	defer h.persistMu.Unlock()
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.watchSeq != expectedSeq || h.watchPersistDirty {
		return
	}
	h.watchers = make(map[string]bot.DesktopWatchRoute, len(routes))
	for _, r := range routes {
		if strings.TrimSpace(r.ChatID) == "" {
			continue
		}
		h.watchers[r.Key()] = r
	}
}

func (h *botBridgeHub) watcherRoutesLocked() []bot.DesktopWatchRoute {
	routes := make([]bot.DesktopWatchRoute, 0, len(h.watchers))
	for _, r := range h.watchers {
		routes = append(routes, r)
	}
	sort.Slice(routes, func(i, j int) bool { return routes[i].Key() < routes[j].Key() })
	return routes
}

func (h *botBridgeHub) Watching(route bot.DesktopWatchRoute) bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	_, ok := h.watchers[route.Key()]
	return ok
}

func (h *botBridgeHub) Approve(approvalID string, allow bool) (string, error) {
	approvalID = strings.TrimSpace(approvalID)
	h.mu.Lock()
	p, ok := h.pending[approvalID]
	if ok && p.kind == "approval" {
		delete(h.pending, approvalID)
	}
	h.mu.Unlock()
	if !ok || p.kind != "approval" {
		return "", fmt.Errorf("未找到待处理的审批 %s（可能已在桌面端处理或已超时）。用 /desktop status 查看当前会话。", approvalID)
	}
	if h.approveTab == nil {
		return "", fmt.Errorf("桌面端审批通道不可用。")
	}
	h.approveTab(p.tabID, approvalID, allow, false, false)
	action := "批准"
	if !allow {
		action = "拒绝"
	}
	return fmt.Sprintf("已提交%s「%s」的操作（%s）。桌面端若已先处理，以先到者为准。", action, h.tabLabel(p.tabID), p.tool), nil
}

func (h *botBridgeHub) AskQuestions(askID string) ([]event.AskQuestion, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	p, ok := h.pending[strings.TrimSpace(askID)]
	if !ok || p.kind != "ask" {
		return nil, false
	}
	return p.questions, true
}

func (h *botBridgeHub) Answer(askID string, answers []event.AskAnswer) (string, error) {
	askID = strings.TrimSpace(askID)
	h.mu.Lock()
	p, ok := h.pending[askID]
	if ok && p.kind == "ask" {
		delete(h.pending, askID)
	}
	h.mu.Unlock()
	if !ok || p.kind != "ask" {
		return "", fmt.Errorf("未找到待回答的提问 %s（可能已在桌面端回答或已超时）。", askID)
	}
	if h.answerTab == nil {
		return "", fmt.Errorf("桌面端问答通道不可用。")
	}
	out := make([]QuestionAnswer, 0, len(answers))
	for _, an := range answers {
		out = append(out, QuestionAnswer{QuestionID: an.QuestionID, Selected: an.Selected})
	}
	h.answerTab(p.tabID, askID, out)
	return fmt.Sprintf("已提交「%s」的回答。桌面端若已先处理，以先到者为准。", h.tabLabel(p.tabID)), nil
}

// ---- 显式接管 ----

func (h *botBridgeHub) Takeover(route bot.DesktopWatchRoute, tabID string) (string, error) {
	tabID = strings.TrimSpace(tabID)
	// DM only. In a group the binding is keyed on the group chat, so after an
	// admin takes over, ANY allowlisted member's plain message would be diverted
	// to drive the session — a privilege escalation past the admin gate that
	// establishes the takeover. Restricting to DM keeps the driver identical to
	// the operator who established it.
	if route.ChatType != bot.ChatDM {
		return "", fmt.Errorf("接管仅支持私聊：在群里接管会让其他成员也能驱动你的桌面会话。请在与 bot 的私聊中接管。")
	}
	session, ok := h.sessionByTabID(tabID)
	if !ok {
		return "", fmt.Errorf("未找到会话 %s。用 /desktop status 查看可接管的会话。", tabID)
	}
	if session.Detached {
		return "", fmt.Errorf("会话「%s」在后台运行，暂不支持接管；请先在桌面端打开它。", h.tabLabel(tabID))
	}
	h.mu.Lock()
	if holder, held := h.takeovers[tabID]; held && holder.Key() != route.Key() {
		h.mu.Unlock()
		return "", fmt.Errorf("会话「%s」已被另一个聊天接管。", h.tabLabel(tabID))
	}
	// 同一聊天换目标：先解除旧绑定，并记下旧 tab 以便公告解除。
	released := ""
	if prev, ok := h.takeoverTabs[route.Key()]; ok && prev != tabID {
		delete(h.takeovers, prev)
		released = prev
	}
	h.takeovers[tabID] = route
	h.takeoverTabs[route.Key()] = tabID
	announce := h.announce
	changed := h.takeoverChanged
	h.mu.Unlock()
	if announce != nil {
		if released != "" {
			announce(released, "IM 远程接管已解除（接管方切换到了另一个会话）。")
		}
		announce(tabID, "此会话已被 IM 远程接管（bot 管理员）。在此本地发送任意消息即可收回控制。")
	}
	if changed != nil {
		changed()
	}
	label := h.tabLabel(tabID)
	return fmt.Sprintf("已接管「%s」。现在直接发消息即可驱动它，输出会流回本聊天；/desktop release 解除接管。桌面端本地发言会自动收回控制。", label), nil
}

func (h *botBridgeHub) Release(route bot.DesktopWatchRoute) (string, error) {
	h.mu.Lock()
	tabID, ok := h.takeoverTabs[route.Key()]
	if ok {
		delete(h.takeoverTabs, route.Key())
		delete(h.takeovers, tabID)
	}
	announce := h.announce
	changed := h.takeoverChanged
	h.mu.Unlock()
	if !ok {
		return "", fmt.Errorf("本聊天当前没有接管任何桌面会话。")
	}
	if announce != nil {
		announce(tabID, "IM 远程接管已解除。")
	}
	if changed != nil {
		changed()
	}
	return fmt.Sprintf("已解除对「%s」的接管。", h.tabLabel(tabID)), nil
}

func (h *botBridgeHub) TakeoverTab(route bot.DesktopWatchRoute) string {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.takeoverTabs[route.Key()]
}

func (h *botBridgeHub) DriveInput(route bot.DesktopWatchRoute, text string) (string, error) {
	h.mu.Lock()
	tabID := h.takeoverTabs[route.Key()]
	h.mu.Unlock()
	if tabID == "" {
		return "", fmt.Errorf("本聊天没有接管任何桌面会话。")
	}
	session, ok := h.sessionByTabID(tabID)
	if !ok || session.Detached {
		// 会话被关闭或转入后台：自动解除绑定，避免消息黑洞。
		h.mu.Lock()
		delete(h.takeoverTabs, route.Key())
		delete(h.takeovers, tabID)
		h.mu.Unlock()
		if changed := h.takeoverChanged; changed != nil {
			changed()
		}
		return "", fmt.Errorf("被接管的会话已关闭或转入后台，接管已自动解除。")
	}
	if session.Running {
		return "", h.busyError(tabID)
	}
	if h.drive == nil {
		return "", fmt.Errorf("桌面端驱动通道不可用。")
	}
	if err := h.drive(tabID, text, route); err != nil {
		if errors.Is(err, errDriveBusy) {
			return "", h.busyError(tabID)
		}
		return "", fmt.Errorf("驱动失败: %v", err)
	}
	return "", nil
}

func (h *botBridgeHub) busyError(tabID string) error {
	return fmt.Errorf("会话「%s」正在执行中，等它完成后再发；或用 /desktop watch on 订阅完成通知。", h.tabLabel(tabID))
}

// reclaimFromDesktop 在桌面用户本地提交输入时收回控制权：解除绑定并通知
// 远端聊天。由 App.SubmitToTab 调用（bridge 自己的驱动不走这条路）。
func (h *botBridgeHub) reclaimFromDesktop(tabID string) {
	h.mu.Lock()
	route, ok := h.takeovers[tabID]
	if ok {
		delete(h.takeovers, tabID)
		delete(h.takeoverTabs, route.Key())
	}
	notify := h.notify
	changed := h.takeoverChanged
	h.mu.Unlock()
	if !ok {
		return
	}
	if changed != nil {
		changed()
	}
	if notify == nil {
		return
	}
	label := h.tabLabel(tabID)
	// 直接入通知队列（不依赖 watch 订阅）：接管者必须知道控制权没了。
	h.enqueue(desktopBridgeNotification{
		text:  constText(fmt.Sprintf("🔓 桌面端已收回会话「%s」的控制权，接管已解除。", label)),
		route: &route,
	})
}

// remoteControlledTabs 返回当前被接管的 tabID 集合（TabMeta 标记用）。
func (h *botBridgeHub) remoteControlledTabs() map[string]bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	if len(h.takeovers) == 0 {
		return nil
	}
	out := make(map[string]bool, len(h.takeovers))
	for tabID := range h.takeovers {
		out[tabID] = true
	}
	return out
}