gateway.go - DeepSeek-Reasonix

gateway.go

1	package bot
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"log/slog"
8	"os"
9	"sort"
10	"strconv"
11	"strings"
12	"sync"
13	"time"
14
15	"reasonix/internal/agent"
16	"reasonix/internal/boot"
17	"reasonix/internal/config"
18	"reasonix/internal/control"
19	"reasonix/internal/event"
20	"reasonix/internal/secrets"
21	)
22
23	// GatewayConfig 是 BotGateway 的配置。
24	type GatewayConfig struct {
25	Model string
26	ToolApprovalMode string
27	MaxSteps int
28	QueueMode string
29	QueueCap int
30	QueueDrop string
31	PairingEnabled bool
32	PairingTTL time.Duration
33	PairingMaxPending int
34	// IgnoreSelfMessages drops messages that are clearly sent by this bot. It
35	// uses configured SelfUserIDs plus recently returned outbound message IDs.
36	IgnoreSelfMessages bool
37	SelfUserIDs map[Platform][]string
38	ControlEnabled bool
39	ControlAddr string
40	ControlToken string
41	// ApprovalTimeout bounds how long a tool-approval/ask prompt blocks a bot
42	// session waiting for a remote user's reply. Zero falls back to
43	// defaultBotApprovalTimeout so an abandoned prompt can't wedge the bot forever
44	// (#4626, #4402). A negative value disables the timeout (wait indefinitely).
45	ApprovalTimeout time.Duration
46	WorkspaceRoot string
47	Channels map[Platform]ChannelConfig
48	ConnectionChannels map[string]ChannelConfig
49	Routes []RouteConfig
50	ConnectionAccess map[string]AccessConfig
51	Allowlist AllowlistConfig
52	Enabled map[Platform]bool
53	Debounce time.Duration
54	// OnInbound observes every allowlisted inbound message before dispatch.
55	//
56	// Reentrancy contract for all GatewayConfig callbacks (OnInbound,
57	// OnSessionReady, OnToolApprovalModeChange): they run synchronously on
58	// gateway-owned dispatch/turn goroutines; OnSessionReady can also run on a
59	// controller recovery/autosave goroutine. Stop drains all of those paths
60	// before returning. A callback must therefore never call Stop, nor block
61	// until a goroutine that does so completes — Stop would wait on the very
62	// goroutine running the callback, a guaranteed deadlock. Hosts that want to
63	// shut the gateway down in reaction to a callback must trigger the shutdown
64	// asynchronously.
65	OnInbound func(InboundMessage)
66	// OnSessionReady notifies the host after the bot has created, reused, or
67	// recovered the controller for an inbound remote. Hosts may persist the
68	// concrete session ID or keep the remote as a read-only channel.
69	OnSessionReady func(InboundMessage, string) error
70	// OnToolApprovalModeChange persists a remote IM request such as /yolo on.
71	// The gateway updates the live session and in-memory defaults first; this
72	// callback lets desktop save the chosen connection mode to user config.
73	OnToolApprovalModeChange func(InboundMessage, string) error
74	// Desktop, when the gateway is embedded in the desktop app, gives bot
75	// chats a god view over desktop sessions (/desktop commands): global
76	// status, event subscriptions, and remote approvals for any live desktop
77	// session. Nil when the gateway runs standalone (reasonix bot start).
78	Desktop DesktopBridge
79	}
80
81	// ChannelConfig overrides gateway defaults for one IM channel.
82	type ChannelConfig struct {
83	Model string
84	ToolApprovalMode string
85	WorkspaceRoot string
86	SessionMappings []SessionMapping
87	}
88
89	// SessionMapping is the runtime subset of a saved bot connection mapping used
90	// to route a remote chat/user/thread back to its intended workspace.
91	type SessionMapping struct {
92	RemoteID string
93	SessionID string
94	SessionSource string
95	ChatType string
96	UserID string
97	ThreadID string
98	Scope string
99	WorkspaceRoot string
100	UpdatedAt string
101	}
102
103	// RouteConfig applies per-remote overrides. Empty match fields are wildcards;
104	// the first matching route wins.
105	type RouteConfig struct {
106	ConnectionID string
107	Platform Platform
108	ChatType ChatType
109	ChatID string
110	UserID string
111	ThreadID string
112	Channel ChannelConfig
113	}
114
115	// AdapterBinding attaches an adapter instance to one saved bot connection.
116	// Feishu and Lark share PlatformFeishu, so ID/Domain keep their sessions,
117	// replies, and per-connection settings separated at runtime.
118	type AdapterBinding struct {
119	ID string
120	Domain string
121	Platform Platform
122	Adapter Adapter
123	}
124
125	// AllowlistConfig 控制哪些用户/群可以使用 bot。
126	type AllowlistConfig struct {
127	Enabled bool
128	AllowAll bool
129	Users map[Platform][]string
130	Approvers map[Platform][]string
131	Admins map[Platform][]string
132	Groups map[Platform][]string
133	}
134
135	// AccessConfig controls who may use one concrete bot connection.
136	type AccessConfig struct {
137	Enabled bool
138	AllowAll bool
139	PairingEnabled bool
140	Users []string
141	Groups []string
142	Approvers []string
143	Admins []string
144	}
145
146	// AdapterHealthSnapshot describes the gateway's current view of one adapter.
147	type AdapterHealthSnapshot struct {
148	ID string `json:"id"`
149	Platform Platform `json:"platform"`
150	Domain string `json:"domain,omitempty"`
151	Name string `json:"name,omitempty"`
152	Status string `json:"status"`
153	StartedAt time.Time `json:"started_at,omitempty"`
154	LastMessageAt time.Time `json:"last_message_at,omitempty"`
155	LastSendAt time.Time `json:"last_send_at,omitempty"`
156	LastErrorAt time.Time `json:"last_error_at,omitempty"`
157	LastError string `json:"last_error,omitempty"`
158	Messages int64 `json:"messages"`
159	Sends int64 `json:"sends"`
160	SendErrors int64 `json:"send_errors"`
161	Closed bool `json:"closed"`
162	}
163
164	// BotGateway 是 reasonix bot 消息网关，管理 Controller 生命周期、session 并发、
165	// 事件渲染和平台适配器。
166	type BotGateway struct {
167	cfg GatewayConfig
168	adapters []AdapterBinding
169	sessions *SessionManager
170	startErr []error
171
172	lifecycleMu sync.Mutex
173	started bool
174	stopped bool
175	runCancel context.CancelFunc
176	startDone chan struct{}
177	stopDone chan struct{}
178	gatewayWG sync.WaitGroup
179	turnWG sync.WaitGroup
180
181	mu sync.Mutex
182	controllers map[string]*sessionState // session key -> active state
183	pendingReactionCleanups map[string][]func()
184	allowlist map[Platform]map[string]bool
185	groupAllowlist map[Platform]map[string]bool
186	selfUserIDs map[Platform]map[string]bool
187	outboundMessageIDs map[string]time.Time
188	adapterHealth map[string]*AdapterHealthSnapshot
189	controlServer *controlHTTPServer
190	sessionOverrides map[string]sessionRuntimeOverride
191
192	logger *slog.Logger
193	}
194
195	// botController is the slice of the controller's driving port the gateway needs:
196	// session lifecycle, turn execution, and approval/ask handling. The bot never
197	// touches goals, checkpoints, or memory, so it depends on those sub-ports only —
198	// not the concrete *control.Controller and its ~99 methods.
199	type botController interface {
200	control.Lifecycle
201	control.TurnControl
202	control.Approvals
203	}
204
205	type sessionState struct {
206	lifecycleMu sync.Mutex
207	retired bool
208	ctrl botController
209	sink *sessionEventSink
210	leases *control.SessionLeaseKeeper
211	platform Platform
212	connectionID string
213	model string
214	workspaceRoot string
215	toolApprovalMode string
216	sessionPath string
217	// mappingDegraded records that this state intentionally runs on a fresh
218	// session because its session_mappings target could not be used at build
219	// time. It keeps later messages (whose profile re-resolves the mapping)
220	// from tearing the state down every turn; convergence back onto the
221	// mapped file happens on the next gateway restart.
222	mappingDegraded bool
223	cancel context.CancelFunc
224	pendingAsks map[string][]event.AskQuestion
225	pendingApprovals map[string]event.Approval
226	lastApprovalID string
227	lastAskID string
228	createdAt time.Time
229	lastActive time.Time
230	}
231
232	var errBotSessionRetired = errors.New("bot session retired during recovery")
233
234	type sessionRuntimeProfile struct {
235	model string
236	workspaceRoot string
237	toolApprovalMode string
238	sessionPath string
239	// sessionPathOptional marks sessionPath as a persisted session_mappings
240	// binding rather than an explicit /attach: when the mapped file cannot be
241	// loaded or leased, the session degrades to a fresh path instead of
242	// dropping the message (#6917).
243	sessionPathOptional bool
244	}
245
246	type sessionRuntimeOverride struct {
247	channel ChannelConfig
248	sessionPath string
249	label string
250	}
251
252	type sessionEventSink struct {
253	mu sync.RWMutex
254	target event.Sink
255	}
256
257	type pendingReactionAdapter interface {
258	AddPendingReaction(ctx context.Context, messageID string) (func(), error)
259	}
260
261	const outboundEchoTTL = 10 * time.Minute
262
263	func (s *sessionEventSink) setTarget(target event.Sink) {
264	s.mu.Lock()
265	defer s.mu.Unlock()
266	s.target = target
267	}
268
269	func (s *sessionEventSink) Emit(e event.Event) {
270	s.mu.RLock()
271	target := s.target
272	s.mu.RUnlock()
273	if target != nil {
274	target.Emit(e)
275	}
276	}
277
278	// NewGateway 创建一个新的 BotGateway。
279	func NewGateway(cfg GatewayConfig, adapters map[Platform]Adapter, logger slog.Logger) BotGateway {
280	bindings := make([]AdapterBinding, 0, len(adapters))
281	for plat, adapter := range adapters {
282	bindings = append(bindings, AdapterBinding{ID: string(plat), Platform: plat, Adapter: adapter})
283	}
284	return NewGatewayWithAdapterBindings(cfg, bindings, logger)
285	}
286
287	// NewGatewayWithAdapterBindings creates a gateway with one or more adapter
288	// instances per platform.
289	func NewGatewayWithAdapterBindings(cfg GatewayConfig, adapters []AdapterBinding, logger slog.Logger) BotGateway {
290	if logger == nil {
291	logger = slog.Default()
292	}
293	if cfg.Debounce <= 0 {
294	cfg.Debounce = 1500 * time.Millisecond
295	}
296	cfg.QueueMode = NormalizeQueueMode(cfg.QueueMode)
297	if cfg.QueueCap <= 0 {
298	cfg.QueueCap = DefaultQueueCap
299	}
300	cfg.QueueDrop = NormalizeQueueDrop(cfg.QueueDrop)
301	if cfg.PairingTTL <= 0 {
302	cfg.PairingTTL = defaultPairingTTL
303	}
304	if cfg.PairingMaxPending <= 0 {
305	cfg.PairingMaxPending = defaultPairingMaxPending
306	}
307	gw := &BotGateway{
308	cfg: cfg,
309	adapters: normalizeAdapterBindings(adapters),
310	sessions: NewSessionManager(cfg.Debounce),
311	controllers: make(map[string]*sessionState),
312	pendingReactionCleanups: make(map[string][]func()),
313	allowlist: make(map[Platform]map[string]bool),
314	groupAllowlist: make(map[Platform]map[string]bool),
315	selfUserIDs: make(map[Platform]map[string]bool),
316	outboundMessageIDs: make(map[string]time.Time),
317	adapterHealth: make(map[string]*AdapterHealthSnapshot),
318	sessionOverrides: make(map[string]sessionRuntimeOverride),
319	logger: logger.With("component", "bot_gateway"),
320	}
321	gw.buildAllowlist()
322	gw.buildSelfUserIDs()
323	for _, binding := range gw.adapters {
324	gw.setAdapterConfigured(binding)
325	}
326	return gw
327	}
328
329	func normalizeAdapterBindings(adapters []AdapterBinding) []AdapterBinding {
330	out := make([]AdapterBinding, 0, len(adapters))
331	for _, binding := range adapters {
332	if binding.Adapter == nil {
333	continue
334	}
335	if binding.Platform == "" {
336	binding.Platform = binding.Adapter.Platform()
337	}
338	if strings.TrimSpace(binding.ID) == "" {
339	binding.ID = string(binding.Platform)
340	}
341	binding.ID = strings.TrimSpace(binding.ID)
342	binding.Domain = strings.TrimSpace(binding.Domain)
343	out = append(out, binding)
344	}
345	return out
346	}
347
348	func (gw *BotGateway) buildAllowlist() {
349	for _, plat := range []Platform{PlatformQQ, PlatformFeishu, PlatformWeixin} {
350	gw.allowlist[plat] = make(map[string]bool)
351	if !gw.cfg.Allowlist.Enabled {
352	continue
353	}
354	addAllowlistUsers(gw.allowlist[plat], gw.cfg.Allowlist.Users[plat])
355	addAllowlistUsers(gw.allowlist[plat], gw.cfg.Allowlist.Admins[plat])
356	addAllowlistUsers(gw.allowlist[plat], gw.cfg.Allowlist.Approvers[plat])
357	gw.groupAllowlist[plat] = make(map[string]bool)
358	for _, gid := range gw.cfg.Allowlist.Groups[plat] {
359	gw.groupAllowlist[plat][gid] = true
360	}
361	}
362	}
363
364	func addAllowlistUsers(dst map[string]bool, users []string) {
365	for _, uid := range users {
366	uid = strings.TrimSpace(uid)
367	if uid != "" {
368	dst[uid] = true
369	}
370	}
371	}
372
373	func (gw *BotGateway) buildSelfUserIDs() {
374	for _, plat := range []Platform{PlatformQQ, PlatformFeishu, PlatformWeixin} {
375	gw.selfUserIDs[plat] = stringSet(gw.cfg.SelfUserIDs[plat])
376	}
377	}
378
379	// Start 启动所有已启用的平台适配器并开始处理消息。
380	func (gw *BotGateway) Start(ctx context.Context) (err error) {
381	gw.lifecycleMu.Lock()
382	if gw.stopped {
383	gw.lifecycleMu.Unlock()
384	return errors.New("bot gateway already stopped")
385	}
386	if gw.started {
387	gw.lifecycleMu.Unlock()
388	return errors.New("bot gateway already started")
389	}
390	gw.started = true
391	runCtx, cancel := context.WithCancel(ctx)
392	gw.runCancel = cancel
393	startDone := make(chan struct{})
394	gw.startDone = startDone
395	gw.lifecycleMu.Unlock()
396	defer func() {
397	if err != nil {
398	cancel()
399	}
400	gw.lifecycleMu.Lock()
401	if err != nil {
402	gw.runCancel = nil
403	}
404	close(startDone)
405	gw.lifecycleMu.Unlock()
406	}()
407
408	started := make([]AdapterBinding, 0, len(gw.adapters))
409	var startErr []error
410	for _, binding := range gw.adapters {
411	if !gw.cfg.Enabled[binding.Platform] {
412	gw.logger.Info("platform disabled, skipping", "platform", binding.Platform, "connection", binding.ID)
413	gw.markAdapterDisabled(binding)
414	continue
415	}
416	gw.logger.Info("starting adapter", "platform", binding.Platform, "connection", binding.ID, "domain", binding.Domain)
417	if err := binding.Adapter.Start(runCtx); err != nil {
418	wrapped := fmt.Errorf("start adapter %s: %w", binding.ID, err)
419	startErr = append(startErr, wrapped)
420	gw.markAdapterStartFailed(binding, err)
421	gw.logger.Warn("adapter start failed", "platform", binding.Platform, "connection", binding.ID, "domain", binding.Domain, "err", err)
422	continue
423	}
424	gw.markAdapterStarted(binding)
425	started = append(started, binding)
426	}
427	// SendToAdapter reads gw.adapters under gw.mu; publish the started set under
428	// the same lock.
429	gw.mu.Lock()
430	gw.adapters = started
431	gw.startErr = startErr
432	gw.mu.Unlock()
433	if len(started) == 0 && len(startErr) > 0 {
434	return errors.Join(startErr...)
435	}
436	if err := gw.startControlServer(runCtx); err != nil {
437	for _, binding := range started {
438	_ = binding.Adapter.Stop()
439	}
440	return err
441	}
442
443	// 合并所有适配器的消息通道
444	for _, binding := range gw.adapters {
445	gw.gatewayWG.Add(1)
446	go func() {
447	defer gw.gatewayWG.Done()
448	gw.dispatchLoop(runCtx, binding)
449	}()
450	}
451
452	return nil
453	}
454
455	func (gw *BotGateway) AdapterCount() int {
456	gw.mu.Lock()
457	defer gw.mu.Unlock()
458	return len(gw.adapters)
459	}
460
461	func (gw *BotGateway) StartErrors() []error {
462	gw.mu.Lock()
463	defer gw.mu.Unlock()
464	out := make([]error, len(gw.startErr))
465	copy(out, gw.startErr)
466	return out
467	}
468
469	// AdapterHealth returns a stable snapshot of all configured adapter instances.
470	func (gw *BotGateway) AdapterHealth() []AdapterHealthSnapshot {
471	gw.mu.Lock()
472	defer gw.mu.Unlock()
473	out := make([]AdapterHealthSnapshot, 0, len(gw.adapterHealth))
474	for _, health := range gw.adapterHealth {
475	if health == nil {
476	continue
477	}
478	out = append(out, *health)
479	}
480	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
481	return out
482	}
483
484	func (gw *BotGateway) setAdapterConfigured(binding AdapterBinding) {
485	gw.mu.Lock()
486	defer gw.mu.Unlock()
487	gw.ensureAdapterHealthLocked(binding).Status = "configured"
488	}
489
490	func (gw *BotGateway) markAdapterDisabled(binding AdapterBinding) {
491	gw.mu.Lock()
492	defer gw.mu.Unlock()
493	health := gw.ensureAdapterHealthLocked(binding)
494	health.Status = "disabled"
495	health.Closed = true
496	}
497
498	func (gw *BotGateway) markAdapterStarted(binding AdapterBinding) {
499	now := time.Now()
500	gw.mu.Lock()
501	defer gw.mu.Unlock()
502	health := gw.ensureAdapterHealthLocked(binding)
503	health.Status = "running"
504	health.StartedAt = now
505	health.LastError = ""
506	health.Closed = false
507	}
508
509	func (gw *BotGateway) markAdapterStartFailed(binding AdapterBinding, err error) {
510	gw.mu.Lock()
511	defer gw.mu.Unlock()
512	health := gw.ensureAdapterHealthLocked(binding)
513	health.Status = "error"
514	health.Closed = true
515	health.LastErrorAt = time.Now()
516	if err != nil {
517	health.LastError = err.Error()
518	}
519	}
520
521	func (gw *BotGateway) markAdapterMessage(binding AdapterBinding) {
522	now := time.Now()
523	gw.mu.Lock()
524	defer gw.mu.Unlock()
525	health := gw.ensureAdapterHealthLocked(binding)
526	health.Status = "running"
527	health.LastMessageAt = now
528	health.Messages++
529	health.Closed = false
530	}
531
532	func (gw *BotGateway) markAdapterClosed(binding AdapterBinding) {
533	gw.mu.Lock()
534	defer gw.mu.Unlock()
535	health := gw.ensureAdapterHealthLocked(binding)
536	if health.Status == "running" {
537	health.Status = "closed"
538	}
539	health.Closed = true
540	}
541
542	func (gw *BotGateway) markAdapterSend(binding AdapterBinding, err error) {
543	now := time.Now()
544	gw.mu.Lock()
545	defer gw.mu.Unlock()
546	health := gw.ensureAdapterHealthLocked(binding)
547	if err != nil {
548	health.SendErrors++
549	health.LastErrorAt = now
550	health.LastError = err.Error()
551	if health.Status == "running" {
552	health.Status = "degraded"
553	}
554	return
555	}
556	health.Sends++
557	health.LastSendAt = now
558	if health.Status == "degraded" {
559	health.Status = "running"
560	}
561	}
562
563	func (gw BotGateway) ensureAdapterHealthLocked(binding AdapterBinding) AdapterHealthSnapshot {
564	id := strings.TrimSpace(binding.ID)
565	if id == "" && binding.Adapter != nil {
566	id = binding.Adapter.Name()
567	}
568	if id == "" {
569	id = string(binding.Platform)
570	}
571	health := gw.adapterHealth[id]
572	if health == nil {
573	health = &AdapterHealthSnapshot{ID: id}
574	gw.adapterHealth[id] = health
575	}
576	health.Platform = binding.Platform
577	health.Domain = strings.TrimSpace(binding.Domain)
578	if binding.Adapter != nil {
579	health.Name = binding.Adapter.Name()
580	}
581	if strings.TrimSpace(health.Status) == "" {
582	health.Status = "configured"
583	}
584	return health
585	}
586
587	// Stop 停止所有适配器并关闭所有 session。它会等待 dispatch 与 turn goroutine
588	// 全部退出，所以绝不能在 GatewayConfig 回调里同步调用（见 OnInbound 的
589	// reentrancy contract），否则 Stop 会等待正在运行该回调的 goroutine 自己。
590	func (gw *BotGateway) Stop() {
591	gw.lifecycleMu.Lock()
592	if gw.stopped {
593	stopDone := gw.stopDone
594	gw.lifecycleMu.Unlock()
595	if stopDone != nil {
596	<-stopDone
597	}
598	return
599	}
600	gw.stopped = true
601	stopDone := make(chan struct{})
602	gw.stopDone = stopDone
603	cancel := gw.runCancel
604	gw.runCancel = nil
605	startDone := gw.startDone
606	gw.lifecycleMu.Unlock()
607	defer close(stopDone)
608
609	if cancel != nil {
610	cancel()
611	}
612	if startDone != nil {
613	<-startDone
614	}
615
616	// Cancel sessions that already exist before waiting for dispatch to drain.
617	// A dispatch already inside handleMessage may still publish a late session,
618	// so closeSessions is repeated after gatewayWG and turnWG reach zero.
619	gw.closeSessions()
620	for _, binding := range gw.adapters {
621	if err := binding.Adapter.Stop(); err != nil {
622	gw.logger.Warn("error stopping adapter", "platform", binding.Platform, "connection", binding.ID, "err", err)
623	}
624	gw.markAdapterClosed(binding)
625	}
626	gw.stopControlServer()
627	gw.gatewayWG.Wait()
628	gw.closeSessions()
629	gw.turnWG.Wait()
630	gw.closeSessions()
631	}
632
633	func (gw *BotGateway) closeSessions() {
634	var states []*sessionState
635	gw.mu.Lock()
636	for key, state := range gw.controllers {
637	states = append(states, state)
638	delete(gw.controllers, key)
639	}
640	gw.mu.Unlock()
641	for _, state := range states {
642	gw.closeSessionState(state)
643	}
644	}
645
646	// closeSessionState tears down a session state that has been unlinked from
647	// gw.controllers. runTurn publishes state.cancel under gw.mu on every turn —
648	// possibly after the state was already unlinked — so snapshot and clear the
649	// field inside the lock and invoke it outside (the same discipline as
650	// cancelActiveSession).
651	func (gw BotGateway) closeSessionState(state sessionState) {
652	if state == nil {
653	return
654	}
655	// Serialize retirement with recovery ownership handoffs. Stop unlinks
656	// sessions before turn goroutines drain, so a recovery callback captured by
657	// the controller can still arrive here. Marking the state retired under the
658	// same lock prevents that callback from reacquiring a lease after teardown;
659	// an already-running handoff completes before the lease is released below.
660	state.lifecycleMu.Lock()
661	if state.retired {
662	state.lifecycleMu.Unlock()
663	return
664	}
665	state.retired = true
666	state.lifecycleMu.Unlock()
667
668	gw.mu.Lock()
669	cancel := state.cancel
670	state.cancel = nil
671	gw.mu.Unlock()
672	if cancel != nil {
673	cancel()
674	}
675	if state.ctrl != nil {
676	state.ctrl.Close()
677	}
678	if state.leases != nil {
679	state.leases.Release()
680	}
681	}
682
683	// unlinkAndCloseSessionState removes state from the live gateway before closing
684	// it. It is used when a controller has already rotated its transcript but the
685	// replacement lease could not be acquired: retaining that state would let the
686	// next message reuse a controller that no longer owns its active session path.
687	func (gw BotGateway) unlinkAndCloseSessionState(key string, state sessionState) {
688	if state == nil {
689	return
690	}
691	gw.mu.Lock()
692	if gw.controllers[key] == state {
693	delete(gw.controllers, key)
694	}
695	gw.mu.Unlock()
696	gw.closeSessionState(state)
697	}
698
699	func (gw *BotGateway) dispatchLoop(ctx context.Context, binding AdapterBinding) {
700	for {
701	select {
702	case <-ctx.Done():
703	gw.markAdapterClosed(binding)
704	return
705	case msg, ok := <-binding.Adapter.Messages():
706	if !ok {
707	gw.markAdapterClosed(binding)
708	return
709	}
710	gw.markAdapterMessage(binding)
711	gw.handleMessage(ctx, binding, msg)
712	}
713	}
714	}
715
716	func (gw *BotGateway) handleMessage(ctx context.Context, binding AdapterBinding, msg InboundMessage) {
717	msg.Platform = binding.Platform
718	if msg.ConnectionID == "" {
719	msg.ConnectionID = binding.ID
720	}
721	if msg.Domain == "" {
722	msg.Domain = binding.Domain
723	}
724	if gw.isSelfMessage(msg) {
725	gw.logger.Debug("bot ignored self message", "platform", binding.Platform, "connection", msg.ConnectionID, "chat", hashID(msg.ChatID), "message", hashID(msg.MessageID), "user", hashID(msg.UserID))
726	return
727	}
728	src := msg.Session()
729	key := BuildSessionKey(src)
730	logFields := []any{
731	"platform", binding.Platform,
732	"connection", msg.ConnectionID,
733	"domain", msg.Domain,
734	"chat_type", msg.ChatType,
735	"chat", hashID(msg.ChatID),
736	"user", hashID(msg.UserID),
737	"operator", hashID(msg.OperatorID),
738	"thread", hashID(msg.ThreadID),
739	"message", hashID(msg.MessageID),
740	"text_chars", len([]rune(msg.Text)),
741	"session", key[:8],
742	}
743	gw.logger.Info("bot inbound message", logFields...)
744
745	// allowlist 检查
746	if !gw.checkAllowlist(binding.Platform, msg) {
747	gw.logger.Info("user not in allowlist", "platform", binding.Platform, "connection", msg.ConnectionID, "user", hashID(msg.UserID))
748	if gw.offerPairing(ctx, binding.Adapter, msg) {
749	return
750	}
751	_ = gw.sendText(ctx, binding.Adapter, msg, "抱歉，您没有使用此 bot 的权限。")
752	return
753	}
754	if gw.cfg.OnInbound != nil {
755	gw.cfg.OnInbound(msg)
756	}
757
758	if normalized, ok := gw.normalizeApprovalShortcut(key, msg.Text); ok {
759	msg.Text = normalized
760	} else if normalized, ok := gw.normalizeAskShortcut(key, msg.Text); ok {
761	msg.Text = normalized
762	} else if _, ok := decisionShortcutCommand(msg.Text); ok && gw.sessions.IsActive(key) {
763	_ = gw.sendText(ctx, binding.Adapter, msg, "没有找到可匹配的待处理操作。请重新触发一次操作后回复编号，或按消息中的 ID 使用 /approve、/deny 或 /answer。")
764	return
765	}
766
767	// 斜杠命令处理
768	if IsSlashBypass(msg.Text) {
769	gw.logger.Info("bot slash command", logFields...)
770	gw.handleSlashCommand(ctx, binding.Adapter, key, msg)
771	return
772	}
773
774	// 已接管桌面会话的聊天：普通消息直接驱动那个桌面会话，不进 bot 自己的
775	// 会话机器（斜杠命令仍走上面的分支，/desktop release 永远可达）。
776	if gw.divertToDesktopTakeover(ctx, binding.Adapter, msg) {
777	gw.logger.Info("bot message diverted to desktop takeover", logFields...)
778	return
779	}
780
781	cleanup := gw.addPendingReaction(ctx, binding.Platform, binding.Adapter, msg)
782
783	queueMode := gw.queueMode(key, msg)
784	if gw.sessions.IsActive(key) {
785	switch queueMode {
786	case QueueModeSteer:
787	if gw.steerActiveSession(ctx, binding.Adapter, key, msg) {
788	gw.logger.Info("bot message steered into active turn", "session", key[:8])
789	if cleanup != nil {
790	cleanup()
791	}
792	_ = gw.sendText(ctx, binding.Adapter, msg, "已收到，会并入当前任务。")
793	return
794	}
795	case QueueModeInterrupt:
796	gw.cancelActiveSession(key)
797	runReactionCleanups(gw.takeReactionCleanups(key))
798	result := gw.sessions.ReplacePending(key, msg)
799	gw.storeReactionCleanup(key, cleanup)
800	gw.logger.Info("bot active turn interrupted; newest message queued", "session", key[:8], "pending", result.Pending)
801	_ = gw.sendText(ctx, binding.Adapter, msg, "已停止当前任务，稍后处理这条新消息。")
802	return
803	}
804	}
805
806	// session 并发控制
807	result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{
808	Mode: queueMode,
809	Cap: gw.cfg.QueueCap,
810	Drop: gw.cfg.QueueDrop,
811	})
812	if result.Rejected {
813	gw.logger.Warn("bot queue rejected message", "session", key[:8], "pending", result.Pending, "mode", result.Mode)
814	if cleanup != nil {
815	cleanup()
816	}
817	_ = gw.sendText(ctx, binding.Adapter, msg, "当前会话排队已满，请稍后再发，或使用 /queue interrupt 中断当前任务。")
818	return
819	}
820	if result.Queued {
821	gw.logger.Debug("message queued", "session", key[:8], "mode", result.Mode, "pending", result.Pending, "dropped", result.Dropped)
822	gw.storeReactionCleanup(key, cleanup)
823	return
824	}
825	if !result.Acquired {
826	gw.logger.Debug("session busy without queue action", "session", key[:8])
827	gw.storeReactionCleanup(key, cleanup)
828	return
829	}
830
831	// Run the turn on its own goroutine so the dispatch loop stays free to read
832	// the next inbound message. A turn that hits interactive approval/ask blocks
833	// inside RunTurn waiting for ctrl.Approve/AnswerQuestion — and the ONLY path
834	// that calls those is handleSlashCommand on this same dispatch goroutine. Run
835	// it inline and the loop can never deliver the /approve (or card) reply that
836	// would unblock it: the session wedges until restart (#4701, #4863, #4402).
837	// Per-session serialization is still held by the session lock (active[key]),
838	// which the deferred Release inside runTurn clears.
839	gw.turnWG.Add(1)
840	go func() {
841	defer gw.turnWG.Done()
842	gw.runTurn(ctx, binding.Adapter, key, msg, cleanup)
843	}()
844	}
845
846	func (gw *BotGateway) queueMode(key string, msg InboundMessage) string {
847	return gw.sessions.QueueMode(key, gw.cfg.QueueMode)
848	}
849
850	func (gw *BotGateway) steerActiveSession(ctx context.Context, adapter Adapter, key string, msg InboundMessage) bool {
851	text := strings.TrimSpace(msg.Text)
852	if text == "" && len(msg.MediaURLs) == 0 && len(msg.Media) == 0 {
853	return false
854	}
855	gw.mu.Lock()
856	state, ok := gw.controllers[key]
857	gw.mu.Unlock()
858	if !ok \|\| state.ctrl == nil {
859	return false
860	}
861	text = gw.inputTextWithMedia(ctx, adapter, msg, state)
862	if strings.TrimSpace(text) == "" {
863	return false
864	}
865	controller, ok := state.ctrl.(interface{ TrySteer(string) bool })
866	return ok && controller.TrySteer(text)
867	}
868
869	func (gw *BotGateway) cancelActiveSession(key string) {
870	// state.cancel is rewritten under gw.mu on every turn (runTurn), so copy it
871	// inside the lock and invoke it outside.
872	var cancel context.CancelFunc
873	gw.mu.Lock()
874	state, ok := gw.controllers[key]
875	if ok && state != nil {
876	cancel = state.cancel
877	}
878	gw.mu.Unlock()
879	if !ok \|\| state == nil {
880	return
881	}
882	if cancel != nil {
883	cancel()
884	return
885	}
886	if state.ctrl != nil {
887	state.ctrl.Cancel()
888	}
889	}
890
891	func (gw *BotGateway) storeReactionCleanup(key string, cleanup func()) {
892	if cleanup == nil {
893	return
894	}
895	gw.mu.Lock()
896	defer gw.mu.Unlock()
897	gw.pendingReactionCleanups[key] = append(gw.pendingReactionCleanups[key], cleanup)
898	}
899
900	func (gw *BotGateway) flushReactionCleanups(key string, cleanup func()) {
901	stored := gw.takeReactionCleanups(key)
902	runReactionCleanups(stored)
903	if cleanup != nil {
904	cleanup()
905	}
906	}
907
908	func (gw *BotGateway) takeReactionCleanups(key string) []func() {
909	gw.mu.Lock()
910	defer gw.mu.Unlock()
911	stored := gw.pendingReactionCleanups[key]
912	delete(gw.pendingReactionCleanups, key)
913	return stored
914	}
915
916	func runReactionCleanups(cleanups []func()) {
917	for _, cleanup := range cleanups {
918	if cleanup != nil {
919	cleanup()
920	}
921	}
922	}
923
924	func makeReactionCleanup(cleanups []func()) func() {
925	if len(cleanups) == 0 {
926	return nil
927	}
928	return func() {
929	runReactionCleanups(cleanups)
930	}
931	}
932
933	func (gw *BotGateway) addPendingReaction(ctx context.Context, plat Platform, adapter Adapter, msg InboundMessage) func() {
934	if strings.TrimSpace(msg.MessageID) == "" {
935	return nil
936	}
937	reactor, ok := adapter.(pendingReactionAdapter)
938	if !ok {
939	return nil
940	}
941	cleanup, err := reactor.AddPendingReaction(ctx, msg.MessageID)
942	if err != nil {
943	gw.logger.Warn("pending reaction failed", "platform", plat, "err", err)
944	return nil
945	}
946	return cleanup
947	}
948
949	func (gw *BotGateway) isSelfMessage(msg InboundMessage) bool {
950	if !gw.cfg.IgnoreSelfMessages {
951	return false
952	}
953	actor := strings.TrimSpace(msg.UserID)
954	if strings.TrimSpace(msg.OperatorID) != "" {
955	actor = strings.TrimSpace(msg.OperatorID)
956	}
957	if actor != "" && gw.selfUserIDs[msg.Platform][actor] {
958	return true
959	}
960	messageID := strings.TrimSpace(msg.MessageID)
961	if messageID == "" {
962	return false
963	}
964	key := outboundMessageKey(msg.Platform, msg.ConnectionID, msg.Domain, msg.ChatID, messageID)
965	now := time.Now()
966	gw.mu.Lock()
967	defer gw.mu.Unlock()
968	gw.pruneOutboundMessagesLocked(now)
969	_, ok := gw.outboundMessageIDs[key]
970	return ok
971	}
972
973	func (gw *BotGateway) rememberOutboundMessage(platform Platform, connID, domain, chatID, messageID string) {
974	messageID = strings.TrimSpace(messageID)
975	if !gw.cfg.IgnoreSelfMessages \|\| messageID == "" {
976	return
977	}
978	now := time.Now()
979	key := outboundMessageKey(platform, connID, domain, chatID, messageID)
980	gw.mu.Lock()
981	defer gw.mu.Unlock()
982	gw.pruneOutboundMessagesLocked(now)
983	gw.outboundMessageIDs[key] = now.Add(outboundEchoTTL)
984	}
985
986	func (gw *BotGateway) pruneOutboundMessagesLocked(now time.Time) {
987	for key, expiresAt := range gw.outboundMessageIDs {
988	if !expiresAt.After(now) {
989	delete(gw.outboundMessageIDs, key)
990	}
991	}
992	}
993
994	func outboundMessageKey(platform Platform, connID, domain, chatID, messageID string) string {
995	return strings.Join([]string{
996	string(platform),
997	strings.TrimSpace(connID),
998	strings.TrimSpace(domain),
999	strings.TrimSpace(chatID),
1000	strings.TrimSpace(messageID),
1001	}, "\x00")
1002	}
1003
1004	func (gw *BotGateway) connectionAccess(msg InboundMessage) (AccessConfig, bool) {
1005	if gw.cfg.ConnectionAccess == nil {
1006	return AccessConfig{}, false
1007	}
1008	id := strings.TrimSpace(msg.ConnectionID)
1009	if id == "" {
1010	return AccessConfig{}, false
1011	}
1012	access, ok := gw.cfg.ConnectionAccess[id]
1013	if !ok {
1014	return AccessConfig{}, false
1015	}
1016	if !accessConfigActive(access) {
1017	return AccessConfig{}, false
1018	}
1019	return access, true
1020	}
1021
1022	func accessConfigActive(access AccessConfig) bool {
1023	return access.Enabled \|\|
1024	access.AllowAll \|\|
1025	access.PairingEnabled \|\|
1026	len(access.Users) > 0 \|\|
1027	len(access.Groups) > 0 \|\|
1028	len(access.Approvers) > 0 \|\|
1029	len(access.Admins) > 0
1030	}
1031
1032	func (gw *BotGateway) checkAllowlist(plat Platform, msg InboundMessage) bool {
1033	if access, ok := gw.connectionAccess(msg); ok {
1034	return checkConnectionAllowlist(access, msg)
1035	}
1036	if gw.cfg.Allowlist.AllowAll {
1037	return true
1038	}
1039	if !gw.cfg.Allowlist.Enabled {
1040	return false
1041	}
1042	actor := msg.UserID
1043	if msg.OperatorID != "" {
1044	actor = msg.OperatorID
1045	}
1046	if !gw.allowlist[plat][actor] {
1047	return false
1048	}
1049	groups := gw.groupAllowlist[plat]
1050	if chatUsesGroupAllowlist(msg.ChatType) && len(groups) > 0 && !groups[msg.ChatID] {
1051	return false
1052	}
1053	return true
1054	}
1055
1056	func checkConnectionAllowlist(access AccessConfig, msg InboundMessage) bool {
1057	if access.AllowAll {
1058	return true
1059	}
1060	if !access.Enabled {
1061	return false
1062	}
1063	actor := msg.UserID
1064	if msg.OperatorID != "" {
1065	actor = msg.OperatorID
1066	}
1067	users := stringSet(append(append(append([]string{}, access.Users...), access.Admins...), access.Approvers...))
1068	groups := stringSet(access.Groups)
1069	actorAllowed := users[actor]
1070	groupAllowed := chatUsesGroupAllowlist(msg.ChatType) && groups[msg.ChatID]
1071	if len(users) == 0 && len(groups) == 0 {
1072	return false
1073	}
1074	return actorAllowed \|\| groupAllowed
1075	}
1076
1077	func (gw *BotGateway) requireCommandRole(ctx context.Context, adapter Adapter, msg InboundMessage, role string) bool {
1078	if gw.checkCommandRole(msg.Platform, msg, role) {
1079	return true
1080	}
1081	_ = gw.sendText(ctx, adapter, msg, "抱歉，你没有执行此 bot 命令的权限。")
1082	return false
1083	}
1084
1085	func (gw *BotGateway) checkCommandRole(plat Platform, msg InboundMessage, role string) bool {
1086	actor := msg.UserID
1087	if msg.OperatorID != "" {
1088	actor = msg.OperatorID
1089	}
1090	if strings.TrimSpace(actor) == "" {
1091	return false
1092	}
1093	if access, ok := gw.connectionAccess(msg); ok {
1094	admins := stringSet(access.Admins)
1095	approvers := stringSet(access.Approvers)
1096	if len(admins) == 0 && len(approvers) == 0 {
1097	return true
1098	}
1099	if admins[actor] {
1100	return true
1101	}
1102	return role == "approver" && approvers[actor]
1103	}
1104	admins := stringSet(gw.cfg.Allowlist.Admins[plat])
1105	approvers := stringSet(gw.cfg.Allowlist.Approvers[plat])
1106	if len(admins) == 0 && len(approvers) == 0 {
1107	return true
1108	}
1109	if admins[actor] {
1110	return true
1111	}
1112	if role == "approver" && approvers[actor] {
1113	return true
1114	}
1115	return false
1116	}
1117
1118	func stringSet(values []string) map[string]bool {
1119	out := make(map[string]bool, len(values))
1120	for _, value := range values {
1121	value = strings.TrimSpace(value)
1122	if value != "" {
1123	out[value] = true
1124	}
1125	}
1126	return out
1127	}
1128
1129	func (gw *BotGateway) offerPairing(ctx context.Context, adapter Adapter, msg InboundMessage) bool {
1130	if access, ok := gw.connectionAccess(msg); ok {
1131	if !access.PairingEnabled {
1132	return false
1133	}
1134	} else if !gw.cfg.PairingEnabled {
1135	return false
1136	}
1137	req, created, err := CreateOrRefreshPairingRequest(msg, PairingConfig{
1138	Enabled: true,
1139	RequestTTL: gw.cfg.PairingTTL,
1140	MaxPendingPerPlatform: gw.cfg.PairingMaxPending,
1141	})
1142	if err != nil {
1143	gw.logger.Warn("bot pairing request failed", "platform", msg.Platform, "chat_type", msg.ChatType, "err", err)
1144	return false
1145	}
1146	prefix := "需要先完成配对。"
1147	if !created {
1148	prefix = "你已有待批准的配对请求。"
1149	}
1150	text := fmt.Sprintf("%s\n配对码: %s\n请在本机运行: reasonix bot pairing approve %s\n此码将在 %s 过期。",
1151	prefix, req.Code, req.Code, req.ExpiresAt.Local().Format("2006-01-02 15:04"))
1152	_ = gw.sendText(ctx, adapter, msg, text)
1153	return true
1154	}
1155
1156	func chatUsesGroupAllowlist(chatType ChatType) bool {
1157	switch chatType {
1158	case ChatGroup, ChatGuild, ChatThread:
1159	return true
1160	default:
1161	return false
1162	}
1163	}
1164
1165	func (gw *BotGateway) normalizeApprovalShortcut(key, text string) (string, bool) {
1166	approvalID := gw.currentPendingApprovalID(key)
1167	if approvalID == "" {
1168	return "", false
1169	}
1170	if gw.pendingApprovalIsRecovery(key, approvalID) {
1171	if command, ok := recoveryShortcutCommand(text, gw.pendingRecoveryCanGrantTask(key, approvalID)); ok {
1172	return command + " " + approvalID, true
1173	}
1174	return "", false
1175	}
1176	command, ok := approvalShortcutCommand(text)
1177	if !ok {
1178	return "", false
1179	}
1180	return command + " " + approvalID, true
1181	}
1182
1183	func approvalShortcutCommand(text string) (string, bool) {
1184	switch strings.ToLower(strings.TrimSpace(text)) {
1185	case "1", "y", "yes", "ok", "同意", "批准", "允许", "允许一次":
1186	return "/approve", true
1187	case "2", "0", "n", "no", "deny", "拒绝":
1188	return "/deny", true
1189	default:
1190	return "", false
1191	}
1192	}
1193
1194	func recoveryShortcutCommand(text string, canGrantTask bool) (string, bool) {
1195	switch strings.ToLower(strings.TrimSpace(text)) {
1196	case "1", "y", "yes", "ok", "继续", "继续此变更", "continue":
1197	return "/recovery-continue", true
1198	case "2", "a", "同类", "本任务允许", "allow similar":
1199	if canGrantTask {
1200	return "/recovery-continue-task", true
1201	}
1202	return "/recovery-revise", true
1203	case "3":
1204	if canGrantTask {
1205	return "/recovery-revise", true
1206	}
1207	return "", false
1208	case "修改", "修改方案", "换个办法", "revise":
1209	return "/recovery-revise", true
1210	default:
1211	return "", false
1212	}
1213	}
1214
1215	func (gw *BotGateway) pendingRecoveryCanGrantTask(key, id string) bool {
1216	gw.mu.Lock()
1217	defer gw.mu.Unlock()
1218	state, ok := gw.controllers[key]
1219	if !ok \|\| state.pendingApprovals == nil {
1220	return false
1221	}
1222	a, ok := state.pendingApprovals[id]
1223	return ok && a.Recovery != nil && a.Recovery.CanGrantTask
1224	}
1225
1226	func (gw *BotGateway) pendingApprovalIsRecovery(key, id string) bool {
1227	gw.mu.Lock()
1228	defer gw.mu.Unlock()
1229	state, ok := gw.controllers[key]
1230	if !ok \|\| state.pendingApprovals == nil {
1231	return false
1232	}
1233	a, ok := state.pendingApprovals[id]
1234	if !ok {
1235	return false
1236	}
1237	return strings.EqualFold(strings.TrimSpace(a.Kind), "recovery") \|\| a.Recovery != nil
1238	}
1239
1240	func decisionShortcutCommand(text string) (string, bool) {
1241	if command, ok := approvalShortcutCommand(text); ok {
1242	return command, true
1243	}
1244	if _, ok := askShortcutAnswer(text); ok {
1245	return "/answer", true
1246	}
1247	return "", false
1248	}
1249
1250	func (gw *BotGateway) currentPendingApprovalID(key string) string {
1251	gw.mu.Lock()
1252	defer gw.mu.Unlock()
1253	state, ok := gw.controllers[key]
1254	if !ok \|\| len(state.pendingApprovals) == 0 {
1255	return ""
1256	}
1257	if state.lastApprovalID != "" {
1258	if _, ok := state.pendingApprovals[state.lastApprovalID]; ok {
1259	return state.lastApprovalID
1260	}
1261	}
1262	for id := range state.pendingApprovals {
1263	return id
1264	}
1265	return ""
1266	}
1267
1268	func (gw *BotGateway) forgetPendingApproval(key, id string) {
1269	gw.mu.Lock()
1270	defer gw.mu.Unlock()
1271	state, ok := gw.controllers[key]
1272	if !ok \|\| state.pendingApprovals == nil {
1273	return
1274	}
1275	delete(state.pendingApprovals, id)
1276	if state.lastApprovalID == id {
1277	state.lastApprovalID = ""
1278	for nextID := range state.pendingApprovals {
1279	state.lastApprovalID = nextID
1280	break
1281	}
1282	}
1283	}
1284
1285	func (gw *BotGateway) normalizeAskShortcut(key, text string) (string, bool) {
1286	raw := strings.TrimSpace(text)
1287	if raw == "" \|\| strings.HasPrefix(raw, "/") {
1288	return "", false
1289	}
1290	askID := gw.currentPendingAskIDForReply(key)
1291	if askID == "" {
1292	return "", false
1293	}
1294	return "/answer " + askID + " " + raw, true
1295	}
1296
1297	func askShortcutAnswer(text string) (string, bool) {
1298	raw := strings.TrimSpace(text)
1299	if raw == "" {
1300	return "", false
1301	}
1302	if strings.ContainsAny(raw, " \t\n;=") {
1303	return "", false
1304	}
1305	if _, err := strconv.Atoi(raw); err == nil {
1306	return raw, true
1307	}
1308	return "", false
1309	}
1310
1311	func (gw *BotGateway) currentPendingAskIDForReply(key string) string {
1312	gw.mu.Lock()
1313	defer gw.mu.Unlock()
1314	state, ok := gw.controllers[key]
1315	if !ok \|\| len(state.pendingAsks) == 0 {
1316	return ""
1317	}
1318	if state.lastAskID != "" {
1319	if _, ok := state.pendingAsks[state.lastAskID]; ok {
1320	return state.lastAskID
1321	}
1322	}
1323	if len(state.pendingAsks) != 1 {
1324	return ""
1325	}
1326	for id := range state.pendingAsks {
1327	return id
1328	}
1329	return ""
1330	}
1331
1332	func (gw *BotGateway) handleSlashCommand(ctx context.Context, adapter Adapter, key string, msg InboundMessage) {
1333	switch {
1334	case strings.HasPrefix(msg.Text, "/stop"):
1335	var cancel context.CancelFunc
1336	gw.mu.Lock()
1337	if state, ok := gw.controllers[key]; ok {
1338	cancel = state.cancel
1339	}
1340	gw.mu.Unlock()
1341	if cancel != nil {
1342	cancel()
1343	}
1344	gw.sessions.ForceRelease(key)
1345	_ = gw.sendText(ctx, adapter, msg, "已停止当前任务。")
1346
1347	case strings.HasPrefix(msg.Text, "/new") \|\| strings.HasPrefix(msg.Text, "/reset"):
1348	var cancel context.CancelFunc
1349	gw.mu.Lock()
1350	state, ok := gw.controllers[key]
1351	if ok {
1352	cancel = state.cancel
1353	}
1354	gw.mu.Unlock()
1355	if ok {
1356	if cancel != nil {
1357	cancel()
1358	}
1359	// NewSession refuses to rotate while a turn is running; the cancel
1360	// above is asynchronous, so give the turn a bounded window to
1361	// unwind before rotating.
1362	deadline := time.Now().Add(5 * time.Second)
1363	for state.ctrl.Running() && time.Now().Before(deadline) {
1364	time.Sleep(10 * time.Millisecond)
1365	}
1366	if err := state.ctrl.NewSession(); err != nil {
1367	gw.logger.Warn("new session failed", "err", err)
1368	gw.sessions.ForceRelease(key)
1369	_ = gw.sendText(ctx, adapter, msg, "新会话创建失败，请稍后重试。")
1370	return
1371	}
1372	if state.leases != nil {
1373	if err := state.leases.Rebind(state.ctrl.SessionPath()); err != nil {
1374	gw.logger.Warn("new session lease failed", "err", control.SessionInUseMessage(err))
1375	gw.unlinkAndCloseSessionState(key, state)
1376	gw.sessions.ForceRelease(key)
1377	_ = gw.sendText(ctx, adapter, msg, "新会话创建失败：无法取得写入权限。请关闭其他 Reasonix 窗口或进程后重试。")
1378	return
1379	}
1380	}
1381	// /new leaves an attached transcript and continues in the freshly
1382	// rotated path. Clear only the path pin while preserving any project
1383	// override, otherwise the next message would rebuild the old attached
1384	// transcript and silently undo the rotation.
1385	gw.mu.Lock()
1386	if gw.controllers[key] == state {
1387	state.sessionPath = ""
1388	if override, exists := gw.sessionOverrides[key]; exists && override.sessionPath != "" {
1389	override.sessionPath = ""
1390	gw.sessionOverrides[key] = override
1391	}
1392	}
1393	gw.mu.Unlock()
1394	gw.rememberSessionReady(msg, state.ctrl)
1395	}
1396	gw.sessions.ForceRelease(key)
1397	_ = gw.sendText(ctx, adapter, msg, "已开始新会话。")
1398
1399	case strings.HasPrefix(msg.Text, "/approve"):
1400	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1401	return
1402	}
1403	// 从消息中解析 approval ID
1404	parts := strings.Fields(msg.Text)
1405	if len(parts) < 2 {
1406	_ = gw.sendText(ctx, adapter, msg, "用法: /approve <id>")
1407	return
1408	}
1409	gw.mu.Lock()
1410	state, ok := gw.controllers[key]
1411	gw.mu.Unlock()
1412	if ok && state.ctrl != nil {
1413	// Recovery cards map allow → continue for older clients that only know Approve.
1414	if gw.pendingApprovalIsRecovery(key, parts[1]) {
1415	_ = state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionContinue, "")
1416	} else {
1417	state.ctrl.Approve(parts[1], true, false, false)
1418	}
1419	gw.forgetPendingApproval(key, parts[1])
1420	_ = gw.sendText(ctx, adapter, msg, "已批准。")
1421	} else {
1422	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的待审批操作，请重新触发一次操作。")
1423	}
1424
1425	case strings.HasPrefix(msg.Text, "/deny"):
1426	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1427	return
1428	}
1429	parts := strings.Fields(msg.Text)
1430	if len(parts) < 2 {
1431	_ = gw.sendText(ctx, adapter, msg, "用法: /deny <id>")
1432	return
1433	}
1434	gw.mu.Lock()
1435	state, ok := gw.controllers[key]
1436	gw.mu.Unlock()
1437	if ok && state.ctrl != nil {
1438	if gw.pendingApprovalIsRecovery(key, parts[1]) {
1439	_ = state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionRevise, "")
1440	} else {
1441	state.ctrl.Approve(parts[1], false, false, false)
1442	}
1443	gw.forgetPendingApproval(key, parts[1])
1444	_ = gw.sendText(ctx, adapter, msg, "已拒绝。")
1445	} else {
1446	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的待审批操作，请重新触发一次操作。")
1447	}
1448
1449	case strings.HasPrefix(msg.Text, "/recovery-continue-task"):
1450	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1451	return
1452	}
1453	parts := strings.Fields(msg.Text)
1454	if len(parts) < 2 {
1455	_ = gw.sendText(ctx, adapter, msg, "用法: /recovery-continue-task <id>")
1456	return
1457	}
1458	gw.mu.Lock()
1459	state, ok := gw.controllers[key]
1460	gw.mu.Unlock()
1461	if ok && state.ctrl != nil {
1462	if err := state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionContinueTask, ""); err != nil {
1463	_ = gw.sendText(ctx, adapter, msg, "确认失败: "+err.Error())
1464	return
1465	}
1466	gw.forgetPendingApproval(key, parts[1])
1467	_ = gw.sendText(ctx, adapter, msg, "已继续；本任务内同类操作将自动执行，范围扩大或风险升级仍会确认。")
1468	} else {
1469	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的待确认操作。")
1470	}
1471
1472	case strings.HasPrefix(msg.Text, "/recovery-continue"):
1473	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1474	return
1475	}
1476	parts := strings.Fields(msg.Text)
1477	if len(parts) < 2 {
1478	_ = gw.sendText(ctx, adapter, msg, "用法: /recovery-continue <id>")
1479	return
1480	}
1481	gw.mu.Lock()
1482	state, ok := gw.controllers[key]
1483	gw.mu.Unlock()
1484	if ok && state.ctrl != nil {
1485	if err := state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionContinue, ""); err != nil {
1486	_ = gw.sendText(ctx, adapter, msg, "确认失败: "+err.Error())
1487	return
1488	}
1489	gw.forgetPendingApproval(key, parts[1])
1490	_ = gw.sendText(ctx, adapter, msg, "已继续。")
1491	} else {
1492	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的待确认操作。")
1493	}
1494
1495	case strings.HasPrefix(msg.Text, "/recovery-revise"):
1496	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1497	return
1498	}
1499	parts := strings.Fields(msg.Text)
1500	if len(parts) < 2 {
1501	_ = gw.sendText(ctx, adapter, msg, "用法: /recovery-revise <id> [补充要求]")
1502	return
1503	}
1504	feedback := strings.TrimSpace(strings.Join(parts[2:], " "))
1505	gw.mu.Lock()
1506	state, ok := gw.controllers[key]
1507	gw.mu.Unlock()
1508	if ok && state.ctrl != nil {
1509	if err := state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionRevise, feedback); err != nil {
1510	_ = gw.sendText(ctx, adapter, msg, "修改方案失败: "+err.Error())
1511	return
1512	}
1513	gw.forgetPendingApproval(key, parts[1])
1514	_ = gw.sendText(ctx, adapter, msg, "已拒绝当前变更并注入修改要求。")
1515	} else {
1516	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的恢复检查点。")
1517	}
1518
1519	case strings.HasPrefix(msg.Text, "/recovery-stop"):
1520	// Backward compatibility for cards rendered by an older client: reject
1521	// the proposed mutation but leave task cancellation to ordinary /stop.
1522	if !gw.requireCommandRole(ctx, adapter, msg, "approver") {
1523	return
1524	}
1525	parts := strings.Fields(msg.Text)
1526	if len(parts) < 2 {
1527	_ = gw.sendText(ctx, adapter, msg, "用法: /recovery-stop <id>")
1528	return
1529	}
1530	gw.mu.Lock()
1531	state, ok := gw.controllers[key]
1532	gw.mu.Unlock()
1533	if ok && state.ctrl != nil {
1534	if err := state.ctrl.ResolveRecovery(parts[1], agent.RecoveryActionRevise, "cancel this proposed action"); err != nil {
1535	_ = gw.sendText(ctx, adapter, msg, "取消变更失败: "+err.Error())
1536	return
1537	}
1538	gw.forgetPendingApproval(key, parts[1])
1539	_ = gw.sendText(ctx, adapter, msg, "已取消当前变更；如需停止整个任务，请使用 /stop。")
1540	} else {
1541	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话中的恢复检查点。")
1542	}
1543
1544	case strings.HasPrefix(msg.Text, "/answer"):
1545	parts := strings.Fields(msg.Text)
1546	if len(parts) < 3 {
1547	_ = gw.sendText(ctx, adapter, msg, "用法: /answer <id> <选项或 q1=选项;q2=选项>")
1548	return
1549	}
1550	askID := parts[1]
1551	rawAnswer := strings.TrimSpace(strings.Join(parts[2:], " "))
1552	gw.mu.Lock()
1553	state, ok := gw.controllers[key]
1554	var questions []event.AskQuestion
1555	if ok {
1556	questions = state.pendingAsks[askID]
1557	delete(state.pendingAsks, askID)
1558	if state.lastAskID == askID {
1559	state.lastAskID = ""
1560	for nextID := range state.pendingAsks {
1561	state.lastAskID = nextID
1562	break
1563	}
1564	}
1565	}
1566	gw.mu.Unlock()
1567	if !ok \|\| state.ctrl == nil {
1568	_ = gw.sendText(ctx, adapter, msg, "没有找到当前会话。")
1569	return
1570	}
1571	answers := parseAskAnswers(questions, rawAnswer)
1572	state.ctrl.AnswerQuestion(askID, answers)
1573	_ = gw.sendText(ctx, adapter, msg, "已提交回答。")
1574
1575	case strings.HasPrefix(msg.Text, "/yolo") \|\| strings.HasPrefix(msg.Text, "/mode"):
1576	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1577	return
1578	}
1579	mode, statusOnly, ok := parseToolApprovalModeCommand(msg.Text)
1580	if !ok {
1581	_ = gw.sendText(ctx, adapter, msg, "用法: /yolo on\|off\|auto\|status，或 /mode yolo\|ask\|auto")
1582	return
1583	}
1584	if statusOnly {
1585	_ = gw.sendText(ctx, adapter, msg, gw.toolApprovalModeStatusText(key, msg))
1586	return
1587	}
1588	persistErr := gw.setToolApprovalModeForMessage(key, msg, mode)
1589	text := toolApprovalModeChangedText(mode)
1590	if persistErr != nil {
1591	text += "\n当前会话已生效，但保存到设置失败：" + persistErr.Error()
1592	}
1593	_ = gw.sendText(ctx, adapter, msg, text)
1594
1595	case strings.HasPrefix(msg.Text, "/queue"):
1596	mode, clear, statusOnly, ok := parseQueueCommand(msg.Text)
1597	if !ok {
1598	_ = gw.sendText(ctx, adapter, msg, "用法: /queue steer\|followup\|collect\|interrupt\|status\|default")
1599	return
1600	}
1601	if statusOnly {
1602	_ = gw.sendText(ctx, adapter, msg, gw.queueStatusText(key, msg))
1603	return
1604	}
1605	if clear {
1606	gw.sessions.ClearQueueMode(key)
1607	_ = gw.sendText(ctx, adapter, msg, "已恢复默认队列模式："+queueModeLabel(gw.queueMode(key, msg))+"。")
1608	return
1609	}
1610	gw.sessions.SetQueueMode(key, mode)
1611	_ = gw.sendText(ctx, adapter, msg, "已切换队列模式："+queueModeLabel(mode)+"。")
1612
1613	case slashCommandVerb(msg.Text) == "/projects":
1614	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1615	return
1616	}
1617	query := strings.TrimSpace(strings.TrimPrefix(msg.Text, "/projects"))
1618	_ = gw.sendText(ctx, adapter, msg, formatBotProjects(gw.buildProjectIndex(), query, botProjectListLimit))
1619
1620	case slashCommandVerb(msg.Text) == "/use":
1621	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1622	return
1623	}
1624	_ = gw.sendText(ctx, adapter, msg, gw.handleUseProjectCommand(key, msg.Text))
1625
1626	case slashCommandVerb(msg.Text) == "/sessions":
1627	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1628	return
1629	}
1630	_ = gw.sendText(ctx, adapter, msg, gw.handleSessionsCommand(msg.Text))
1631
1632	case slashCommandVerb(msg.Text) == "/attach":
1633	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1634	return
1635	}
1636	_ = gw.sendText(ctx, adapter, msg, gw.handleAttachSessionCommand(key, msg.Text))
1637
1638	case slashCommandVerb(msg.Text) == "/search":
1639	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1640	return
1641	}
1642	_ = gw.sendText(ctx, adapter, msg, gw.handleProjectSearchCommand(ctx, msg.Text))
1643
1644	case strings.HasPrefix(msg.Text, "/desktop"):
1645	// God view over the embedding desktop app: listing every live desktop
1646	// session and answering its approvals is strictly more power than the
1647	// per-session approver role, so gate on admin.
1648	if !gw.requireCommandRole(ctx, adapter, msg, "admin") {
1649	return
1650	}
1651	_ = gw.sendText(ctx, adapter, msg, gw.handleDesktopCommand(msg))
1652
1653	case strings.HasPrefix(msg.Text, "/status"):
1654	active := gw.sessions.ActiveCount()
1655	pending := gw.sessions.PendingCount(key)
1656	gw.mu.Lock()
1657	sessions := len(gw.controllers)
1658	gw.mu.Unlock()
1659	mode := gw.currentToolApprovalMode(key, msg)
1660	_ = gw.sendText(ctx, adapter, msg, fmt.Sprintf("活跃任务数: %d\n保留会话数: %d\n工具审批模式: %s\n队列模式: %s\n当前会话排队: %d\n连接健康: %s", active, sessions, toolApprovalModeLabel(mode), queueModeLabel(gw.queueMode(key, msg)), pending, gw.adapterHealthSummaryText()))
1661
1662	case strings.HasPrefix(msg.Text, "/help"):
1663	help := "可用命令:\n" +
1664	"/stop - 停止当前任务\n" +
1665	"/new - 开始新会话\n" +
1666	"/reset - 重置会话\n" +
1667	"/approve <id> - 批准操作\n" +
1668	"/deny <id> - 拒绝操作\n" +
1669	"/answer <id> <选项> - 回答 ask 问题\n" +
1670	"/yolo on\|off\|auto\|status - 切换或查看工具审批模式\n" +
1671	"/mode yolo\|ask\|auto - 切换工具审批模式\n" +
1672	"/queue steer\|followup\|collect\|interrupt\|status - 切换或查看队列模式\n" +
1673	"/projects [关键词] - 查看可切换项目索引\n" +
1674	"/use project <id\|名称> - 将当前远端会话切到某个项目\n" +
1675	"/sessions search <关键词> - 搜索可 attach 的历史会话\n" +
1676	"/attach session <id\|关键词> - 绑定当前远端会话到已有历史会话\n" +
1677	"/search all <关键词> - 跨已索引项目检索文件内容\n" +
1678	"/desktop status\|watch\|approve\|deny\|answer - 桌面端上帝视角(需内嵌运行)\n" +
1679	"/status - 查看状态\n" +
1680	"/help - 显示帮助"
1681	_ = gw.sendText(ctx, adapter, msg, help)
1682	}
1683	}
1684
1685	func slashCommandVerb(text string) string {
1686	parts := strings.Fields(strings.TrimSpace(text))
1687	if len(parts) == 0 {
1688	return ""
1689	}
1690	return strings.ToLower(parts[0])
1691	}
1692
1693	func (gw *BotGateway) handleUseProjectCommand(key, text string) string {
1694	selector := parseUseProjectSelector(text)
1695	if selector == "" {
1696	return "用法: /use project <项目 id\|名称\|路径>，或 /use project default 恢复默认路由。"
1697	}
1698	if isDefaultBotSelector(selector) {
1699	if !gw.setSessionRuntimeOverride(key, sessionRuntimeOverride{}, false) {
1700	return botRuntimeSwitchBusyText()
1701	}
1702	return "已恢复当前远端会话的默认项目路由。下一条消息会按 bot 配置重新选择 workspace。"
1703	}
1704	projects := gw.buildProjectIndex()
1705	project, matches := resolveBotProject(projects, selector)
1706	if project.Root == "" {
1707	if len(matches) > 0 {
1708	return "匹配到多个项目，请使用项目 id：\n" + formatBotProjects(matches, "", botProjectListLimit)
1709	}
1710	return "没有匹配的项目。可先用 /projects 查看当前索引。"
1711	}
1712	if !gw.setSessionRuntimeOverride(key, sessionRuntimeOverride{
1713	channel: ChannelConfig{WorkspaceRoot: project.Root},
1714	label: "project:" + project.ID,
1715	}, true) {
1716	return botRuntimeSwitchBusyText()
1717	}
1718	return fmt.Sprintf("已将当前远端会话切到项目 %s %s。\n下一条消息将在 %s 中运行。", project.ID, project.Name, displayBotPath(project.Root))
1719	}
1720
1721	func parseUseProjectSelector(text string) string {
1722	parts := strings.Fields(text)
1723	if len(parts) < 2 \|\| strings.ToLower(parts[0]) != "/use" {
1724	return ""
1725	}
1726	if len(parts) >= 3 && strings.EqualFold(parts[1], "project") {
1727	return strings.TrimSpace(strings.Join(parts[2:], " "))
1728	}
1729	return strings.TrimSpace(strings.Join(parts[1:], " "))
1730	}
1731
1732	func (gw *BotGateway) handleSessionsCommand(text string) string {
1733	query := parseSessionsQuery(text)
1734	projects := gw.buildProjectIndex()
1735	sessions := gw.buildSessionIndex(projects)
1736	return formatBotSessions(sessions, query, botSessionListLimit)
1737	}
1738
1739	func parseSessionsQuery(text string) string {
1740	parts := strings.Fields(text)
1741	if len(parts) <= 1 {
1742	return ""
1743	}
1744	if strings.EqualFold(parts[1], "search") {
1745	return strings.TrimSpace(strings.Join(parts[2:], " "))
1746	}
1747	return strings.TrimSpace(strings.Join(parts[1:], " "))
1748	}
1749
1750	func (gw *BotGateway) handleAttachSessionCommand(key, text string) string {
1751	selector := parseAttachSessionSelector(text)
1752	if selector == "" {
1753	return "用法: /attach session <会话 id\|关键词\|path:...>"
1754	}
1755	projects := gw.buildProjectIndex()
1756	sessions := gw.buildSessionIndex(projects)
1757	session, matches := resolveBotSession(sessions, selector)
1758	if session.ID == "" {
1759	if len(matches) > 0 {
1760	return "匹配到多个会话，请使用会话 id：\n" + formatBotSessions(matches, "", botSessionListLimit)
1761	}
1762	return "没有匹配的会话。可先用 /sessions search <关键词> 查看当前索引。"
1763	}
1764	if session.SessionPath == "" {
1765	return "这个会话没有可恢复的 path: transcript，暂时不能 attach。"
1766	}
1767	if info, err := os.Stat(session.SessionPath); err != nil \|\| info.IsDir() {
1768	return "会话文件不可用或已被移动：" + displayBotPath(session.SessionPath)
1769	}
1770	workspaceRoot := session.WorkspaceRoot
1771	if workspaceRoot == "" {
1772	project := botProjectForPath(projects, session.SessionPath)
1773	workspaceRoot = project.Root
1774	}
1775	if !gw.setSessionRuntimeOverride(key, sessionRuntimeOverride{
1776	channel: ChannelConfig{WorkspaceRoot: workspaceRoot},
1777	sessionPath: session.SessionPath,
1778	label: "session:" + session.ID,
1779	}, true) {
1780	return botRuntimeSwitchBusyText()
1781	}
1782	projectName := firstNonEmptyString(session.ProjectName, botProjectName(workspaceRoot), "global")
1783	return fmt.Sprintf("已 attach 到会话 %s（%s）。\n下一条消息会从 %s 继续。", session.ID, projectName, displayBotPath(session.SessionPath))
1784	}
1785
1786	func parseAttachSessionSelector(text string) string {
1787	parts := strings.Fields(text)
1788	if len(parts) < 3 \|\| !strings.EqualFold(parts[0], "/attach") \|\| !strings.EqualFold(parts[1], "session") {
1789	return ""
1790	}
1791	return strings.TrimSpace(strings.Join(parts[2:], " "))
1792	}
1793
1794	func (gw *BotGateway) handleProjectSearchCommand(ctx context.Context, text string) string {
1795	parts := strings.Fields(text)
1796	if len(parts) < 3 \|\| !strings.EqualFold(parts[1], "all") {
1797	return "用法: /search all <关键词>"
1798	}
1799	query := strings.TrimSpace(strings.Join(parts[2:], " "))
1800	searchCtx, cancel := context.WithTimeout(ctx, 8*time.Second)
1801	defer cancel()
1802	results, err := searchBotProjects(searchCtx, gw.buildProjectIndex(), query, botSearchListLimit)
1803	if err != nil {
1804	return "检索失败：" + err.Error()
1805	}
1806	return formatBotProjectSearchResults(results, botSearchListLimit)
1807	}
1808
1809	func botRuntimeSwitchBusyText() string {
1810	return "当前会话仍有正在运行、等待确认或后台执行的任务。请先完成或停止这些任务，再切换项目或 attach 会话。"
1811	}
1812
1813	func (gw *BotGateway) setSessionRuntimeOverride(key string, override sessionRuntimeOverride, enabled bool) bool {
1814	return gw.sessions.runIfIdle(key, func() bool {
1815	var old *sessionState
1816	gw.mu.Lock()
1817	if state, ok := gw.controllers[key]; ok {
1818	if botSessionHasActiveWork(state) {
1819	gw.mu.Unlock()
1820	return false
1821	}
1822	old = state
1823	delete(gw.controllers, key)
1824	}
1825	if enabled {
1826	override.sessionPath = canonicalBotPath(override.sessionPath)
1827	override.channel.WorkspaceRoot = canonicalBotPath(override.channel.WorkspaceRoot)
1828	gw.sessionOverrides[key] = override
1829	} else {
1830	delete(gw.sessionOverrides, key)
1831	}
1832	gw.mu.Unlock()
1833	gw.closeSessionState(old)
1834	return true
1835	})
1836	}
1837
1838	func botSessionHasActiveWork(state *sessionState) bool {
1839	if state == nil \|\| state.ctrl == nil {
1840	return false
1841	}
1842	status, ok := safeBotControllerRuntimeStatus(state.ctrl)
1843	if !ok {
1844	return true
1845	}
1846	return status.Running \|\| status.PendingPrompt \|\| status.BackgroundJobs > 0
1847	}
1848
1849	func safeBotControllerRuntimeStatus(ctrl botController) (status control.RuntimeStatus, ok bool) {
1850	if ctrl == nil {
1851	return control.RuntimeStatus{}, false
1852	}
1853	defer func() {
1854	if recover() != nil {
1855	status = control.RuntimeStatus{}
1856	ok = false
1857	}
1858	}()
1859	return ctrl.RuntimeStatus(), true
1860	}
1861
1862	func (gw *BotGateway) sessionRuntimeOverrideForMessage(msg InboundMessage) (sessionRuntimeOverride, bool) {
1863	key := BuildSessionKey(msg.Session())
1864	gw.mu.Lock()
1865	defer gw.mu.Unlock()
1866	override, ok := gw.sessionOverrides[key]
1867	return override, ok
1868	}
1869
1870	func isDefaultBotSelector(selector string) bool {
1871	switch strings.ToLower(strings.TrimSpace(selector)) {
1872	case "default", "reset", "inherit", "global", "none", "默认", "重置":
1873	return true
1874	default:
1875	return false
1876	}
1877	}
1878
1879	func parseQueueCommand(text string) (mode string, clear bool, statusOnly bool, ok bool) {
1880	parts := strings.Fields(text)
1881	if len(parts) == 0 \|\| strings.ToLower(strings.TrimSpace(parts[0])) != "/queue" {
1882	return "", false, false, false
1883	}
1884	if len(parts) == 1 {
1885	return "", false, true, true
1886	}
1887	switch strings.ToLower(strings.TrimSpace(parts[1])) {
1888	case "status", "state", "show", "状态", "查看":
1889	return "", false, true, true
1890	case "default", "reset", "inherit", "默认", "重置":
1891	return "", true, false, true
1892	default:
1893	if normalized := NormalizeOptionalQueueMode(parts[1]); normalized != "" {
1894	return normalized, false, false, true
1895	}
1896	return "", false, false, false
1897	}
1898	}
1899
1900	func (gw *BotGateway) queueStatusText(key string, msg InboundMessage) string {
1901	return fmt.Sprintf("当前队列模式：%s\n当前会话排队: %d\n全局上限: %d\n溢出策略: %s\n用法：/queue steer\|followup\|collect\|interrupt\|status\|default",
1902	queueModeLabel(gw.queueMode(key, msg)),
1903	gw.sessions.PendingCount(key),
1904	gw.cfg.QueueCap,
1905	queueDropLabel(gw.cfg.QueueDrop),
1906	)
1907	}
1908
1909	func queueModeLabel(mode string) string {
1910	switch NormalizeQueueMode(mode) {
1911	case QueueModeFollowup:
1912	return "逐条跟进"
1913	case QueueModeCollect:
1914	return "合并收集"
1915	case QueueModeInterrupt:
1916	return "打断重跑"
1917	default:
1918	return "即时补充"
1919	}
1920	}
1921
1922	func queueDropLabel(drop string) string {
1923	switch NormalizeQueueDrop(drop) {
1924	case QueueDropOld:
1925	return "丢弃最早消息"
1926	case QueueDropNew:
1927	return "拒绝新消息"
1928	default:
1929	return "压缩摘要"
1930	}
1931	}
1932
1933	func (gw *BotGateway) adapterHealthSummaryText() string {
1934	snapshots := gw.AdapterHealth()
1935	if len(snapshots) == 0 {
1936	return "未启动"
1937	}
1938	parts := make([]string, 0, len(snapshots))
1939	for _, h := range snapshots {
1940	label := strings.TrimSpace(h.ID)
1941	if label == "" {
1942	label = string(h.Platform)
1943	}
1944	status := strings.TrimSpace(h.Status)
1945	if status == "" {
1946	status = "unknown"
1947	}
1948	parts = append(parts, fmt.Sprintf("%s=%s", label, status))
1949	}
1950	return strings.Join(parts, ", ")
1951	}
1952
1953	func parseToolApprovalModeCommand(text string) (mode string, statusOnly bool, ok bool) {
1954	parts := strings.Fields(text)
1955	if len(parts) == 0 {
1956	return "", false, false
1957	}
1958	cmd := strings.ToLower(strings.TrimSpace(parts[0]))
1959	switch cmd {
1960	case "/yolo":
1961	if len(parts) == 1 {
1962	return control.ToolApprovalYolo, false, true
1963	}
1964	return parseToolApprovalModeArg(parts[1])
1965	case "/mode":
1966	if len(parts) == 1 {
1967	return "", true, true
1968	}
1969	return parseToolApprovalModeArg(parts[1])
1970	default:
1971	return "", false, false
1972	}
1973	}
1974
1975	func parseToolApprovalModeArg(arg string) (mode string, statusOnly bool, ok bool) {
1976	switch strings.ToLower(strings.TrimSpace(arg)) {
1977	case "status", "state", "show", "状态", "查看":
1978	return "", true, true
1979	case "on", "enable", "enabled", "true", "1", "yolo", "full", "full-access", "bypass", "开启", "打开":
1980	return control.ToolApprovalYolo, false, true
1981	case "off", "disable", "disabled", "false", "0", "ask", "询问", "关闭":
1982	return control.ToolApprovalAsk, false, true
1983	case "auto", "自动":
1984	return control.ToolApprovalAuto, false, true
1985	default:
1986	return "", false, false
1987	}
1988	}
1989
1990	func (gw *BotGateway) setToolApprovalModeForMessage(key string, msg InboundMessage, mode string) error {
1991	mode = normalizeBotToolApprovalMode(mode)
1992	var ctrl botController
1993
1994	gw.mu.Lock()
1995	if state, ok := gw.controllers[key]; ok {
1996	ctrl = state.ctrl
1997	}
1998	gw.updateToolApprovalModeDefaultLocked(msg, mode)
1999	gw.mu.Unlock()
2000
2001	if ctrl != nil {
2002	ctrl.SetToolApprovalMode(mode)
2003	}
2004	if gw.cfg.OnToolApprovalModeChange != nil {
2005	return gw.cfg.OnToolApprovalModeChange(msg, mode)
2006	}
2007	return nil
2008	}
2009
2010	func (gw *BotGateway) updateToolApprovalModeDefaultLocked(msg InboundMessage, mode string) {
2011	if id := strings.TrimSpace(msg.ConnectionID); id != "" {
2012	if gw.cfg.ConnectionChannels == nil {
2013	gw.cfg.ConnectionChannels = make(map[string]ChannelConfig)
2014	}
2015	channel := gw.cfg.ConnectionChannels[id]
2016	channel.ToolApprovalMode = mode
2017	gw.cfg.ConnectionChannels[id] = channel
2018	return
2019	}
2020	if msg.Platform != "" {
2021	if gw.cfg.Channels == nil {
2022	gw.cfg.Channels = make(map[Platform]ChannelConfig)
2023	}
2024	channel := gw.cfg.Channels[msg.Platform]
2025	channel.ToolApprovalMode = mode
2026	gw.cfg.Channels[msg.Platform] = channel
2027	return
2028	}
2029	gw.cfg.ToolApprovalMode = mode
2030	}
2031
2032	func (gw *BotGateway) currentToolApprovalMode(key string, msg InboundMessage) string {
2033	var ctrl botController
2034	gw.mu.Lock()
2035	if state, ok := gw.controllers[key]; ok {
2036	ctrl = state.ctrl
2037	}
2038	gw.mu.Unlock()
2039	if ctrl != nil {
2040	return ctrl.ToolApprovalMode()
2041	}
2042	_, _, mode := gw.sessionOptionsForMessage(msg)
2043	return mode
2044	}
2045
2046	func (gw *BotGateway) toolApprovalModeStatusText(key string, msg InboundMessage) string {
2047	mode := gw.currentToolApprovalMode(key, msg)
2048	return fmt.Sprintf("当前工具审批模式：%s\n用法：/yolo on\|off\|auto\|status，或 /mode yolo\|ask\|auto", toolApprovalModeLabel(mode))
2049	}
2050
2051	func toolApprovalModeChangedText(mode string) string {
2052	switch normalizeBotToolApprovalMode(mode) {
2053	case control.ToolApprovalYolo:
2054	return "已开启 YOLO：普通工具审批将自动放行；Ask 问题和计划批准仍会等待确认。"
2055	case control.ToolApprovalAuto:
2056	return "已切换为自动模式：策略允许的工具会自动放行，仍保留需要询问或拒绝的规则。"
2057	default:
2058	return "已切回询问模式：工具执行前会请求确认。"
2059	}
2060	}
2061
2062	func toolApprovalModeLabel(mode string) string {
2063	switch normalizeBotToolApprovalMode(mode) {
2064	case control.ToolApprovalYolo:
2065	return "YOLO"
2066	case control.ToolApprovalAuto:
2067	return "自动"
2068	default:
2069	return "询问"
2070	}
2071	}
2072
2073	func (gw *BotGateway) runTurn(ctx context.Context, adapter Adapter, key string, msg InboundMessage, cleanup func()) {
2074	gw.logger.Info("bot turn started", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8])
2075	defer func() {
2076	// 检查是否有等待队列中的消息
2077	next := gw.sessions.Release(key)
2078	if next != nil {
2079	if cleanup != nil {
2080	cleanup()
2081	}
2082	nextCleanup := makeReactionCleanup(gw.takeReactionCleanups(key))
2083	gw.logger.Info("bot pending message released", "platform", next.Platform, "chat_type", next.ChatType, "chat", hashID(next.ChatID), "session", key[:8])
2084	gw.runTurn(ctx, adapter, key, *next, nextCleanup)
2085	return
2086	}
2087	gw.flushReactionCleanups(key, cleanup)
2088	}()
2089
2090	// 获取或创建 Controller
2091	state := gw.getOrCreateSession(ctx, key, msg)
2092	if state == nil \|\| state.ctrl == nil {
2093	_ = gw.sendText(ctx, adapter, msg, "内部错误：无法创建会话。")
2094	return
2095	}
2096	gw.rememberSessionReady(msg, state.ctrl)
2097
2098	// 构建输入文本：群聊中在消息前加上发送者名，并把 IM 媒体保存为 @附件引用。
2099	input := gw.inputTextWithMedia(ctx, adapter, msg, state)
2100	if msg.ChatType == ChatGroup {
2101	userName := strings.TrimSpace(msg.UserName)
2102	if msg.ResolveUserName != nil {
2103	if resolved := strings.TrimSpace(msg.ResolveUserName(ctx)); resolved != "" {
2104	userName = resolved
2105	}
2106	}
2107	input = fmt.Sprintf("[%s] %s", userName, input)
2108	}
2109
2110	// 发送"正在输入"状态
2111	_ = adapter.SendTyping(ctx, msg.ChatID)
2112
2113	// 创建事件渲染 sink
2114	sink := newRenderSink(
2115	ctx,
2116	adapter,
2117	msg.ConnectionID,
2118	msg.Domain,
2119	msg.ChatID,
2120	msg.ChatType,
2121	msg.UserID,
2122	msg.MessageID,
2123	gw.logger,
2124	func(approval event.Approval) {
2125	gw.mu.Lock()
2126	if state.pendingApprovals == nil {
2127	state.pendingApprovals = make(map[string]event.Approval)
2128	}
2129	state.pendingApprovals[approval.ID] = approval
2130	state.lastApprovalID = approval.ID
2131	gw.mu.Unlock()
2132	},
2133	func(ask event.Ask) {
2134	gw.mu.Lock()
2135	if state.pendingAsks == nil {
2136	state.pendingAsks = make(map[string][]event.AskQuestion)
2137	}
2138	state.pendingAsks[ask.ID] = ask.Questions
2139	state.lastAskID = ask.ID
2140	gw.mu.Unlock()
2141	},
2142	)
2143	// Finish initializing the sink before publishing it as the live target: once
2144	// setTarget runs, other goroutines can reach this sink via state.sink.Emit.
2145	sink.ctrl = state.ctrl
2146	state.sink.setTarget(sink)
2147	defer state.sink.setTarget(nil)
2148
2149	// 创建带取消的 context
2150	turnCtx, cancel := context.WithCancel(ctx)
2151	defer cancel()
2152
2153	gw.mu.Lock()
2154	live := gw.controllers[key] == state
2155	if live {
2156	state.cancel = cancel
2157	}
2158	state.lastActive = time.Now()
2159	gw.mu.Unlock()
2160	if !live {
2161	// The session was closed (gateway stop or runtime rebuild) after this
2162	// turn picked it up; a cancel published now would never be consumed, so
2163	// abort the turn instead of running it uncancellable.
2164	cancel()
2165	}
2166
2167	// 运行一轮对话
2168	err := state.ctrl.RunTurn(turnCtx, input)
2169	sink.Emit(event.Event{Kind: event.TurnDone, Err: err})
2170	if err != nil {
2171	gw.logger.Warn("turn error", "session", key[:8], "err", err)
2172	return
2173	}
2174	gw.logger.Info("bot turn completed", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8])
2175	}
2176
2177	func (gw BotGateway) inputTextWithMedia(ctx context.Context, adapter Adapter, msg InboundMessage, state sessionState) string {
2178	input := msg.Text
2179	if len(msg.MediaURLs) == 0 && len(msg.Media) == 0 {
2180	return input
2181	}
2182	workspaceRoot := ""
2183	if state != nil && state.ctrl != nil {
2184	workspaceRoot = state.ctrl.WorkspaceRoot()
2185	}
2186	if strings.TrimSpace(workspaceRoot) == "" {
2187	_, workspaceRoot, _ = gw.sessionOptionsForMessage(msg)
2188	}
2189	refs, errs := saveInboundMedia(ctx, workspaceRoot, msg.MediaURLs)
2190	itemRefs, fallbacks, itemErrs := saveInboundMediaItems(ctx, workspaceRoot, msg.Media)
2191	refs = append(refs, itemRefs...)
2192	errs = append(errs, itemErrs...)
2193	if len(errs) > 0 {
2194	gw.logger.Warn("bot media attachment failed", "platform", msg.Platform, "chat", hashID(msg.ChatID), "errors", len(errs))
2195	_ = gw.sendText(ctx, adapter, msg, fmt.Sprintf("有 %d 个附件保存失败；我会先处理可用内容。", len(errs)))
2196	}
2197	return appendMediaRefs(appendMediaFallbacks(input, fallbacks), refs)
2198	}
2199
2200	func (gw BotGateway) getOrCreateSession(ctx context.Context, key string, msg InboundMessage) sessionState {
2201	profile := gw.sessionProfileForMessage(msg)
2202	var stale *sessionState
2203	gw.mu.Lock()
2204	if state, ok := gw.controllers[key]; ok {
2205	if !sessionStateMatchesRuntime(state, profile) {
2206	if botSessionHasActiveWork(state) {
2207	gw.mu.Unlock()
2208	safeBotSetToolApprovalMode(state.ctrl, profile.toolApprovalMode)
2209	gw.logger.Warn("bot session runtime change deferred while work is active", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8])
2210	return state
2211	}
2212	delete(gw.controllers, key)
2213	stale = state
2214	gw.mu.Unlock()
2215	gw.closeSessionState(stale)
2216	gw.logger.Warn("bot session runtime changed; rebuilding", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8], "old_workspace_set", strings.TrimSpace(stale.workspaceRoot) != "", "new_workspace_set", profile.workspaceRoot != "", "old_model", stale.model, "new_model", profile.model)
2217	} else {
2218	updateSessionStateRuntime(state, msg, profile)
2219	gw.mu.Unlock()
2220	safeBotSetToolApprovalMode(state.ctrl, profile.toolApprovalMode)
2221	gw.logger.Info("bot session reused", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8])
2222	return state
2223	}
2224	} else {
2225	gw.mu.Unlock()
2226	}
2227
2228	// Create the lease owner before the controller so automatic conflict
2229	// recovery can move ownership to the recovery branch before the controller
2230	// commits to writing it. Without this callback the bot kept guarding the
2231	// original path while continuing on an unleased recovery path.
2232	sessionSink := &sessionEventSink{}
2233	leases := control.NewSessionLeaseKeeper()
2234	state := &sessionState{
2235	sink: sessionSink,
2236	leases: leases,
2237	platform: msg.Platform,
2238	connectionID: strings.TrimSpace(msg.ConnectionID),
2239	model: profile.model,
2240	workspaceRoot: profile.workspaceRoot,
2241	toolApprovalMode: profile.toolApprovalMode,
2242	sessionPath: profile.sessionPath,
2243	pendingAsks: make(map[string][]event.AskQuestion),
2244	createdAt: time.Now(),
2245	lastActive: time.Now(),
2246	}
2247	gw.logger.Info("bot session creating", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8], "model", profile.model, "workspace_set", profile.workspaceRoot != "", "tool_approval_mode", profile.toolApprovalMode)
2248	ctrl, err := boot.Build(ctx, boot.Options{
2249	Model: profile.model,
2250	MaxSteps: gw.cfg.MaxSteps,
2251	MaxStepsKey: "bot.max_steps",
2252	RequireKey: true,
2253	Sink: sessionSink,
2254	StatsSource: "bot",
2255	WorkspaceRoot: profile.workspaceRoot,
2256	SessionDir: botSessionDir(profile.workspaceRoot),
2257	ApprovalTimeout: gw.approvalTimeout(),
2258	OnSessionRecovered: gw.botSessionRecoveredHandler(key, msg, state),
2259	})
2260	if err != nil {
2261	leases.Release()
2262	gw.logger.Error("build controller failed", "err", secrets.RedactError(err))
2263	return nil
2264	}
2265	state.ctrl = ctrl
2266	if profile.sessionPath != "" {
2267	// A mapped binding degrades to a fresh session on failure; only an
2268	// explicit /attach is allowed to hard-fail the message, because the
2269	// user named that exact session.
2270	degrade := func(reason string, err error) bool {
2271	if !profile.sessionPathOptional {
2272	return false
2273	}
2274	gw.logger.Warn("mapped bot session unavailable; starting fresh", "reason", reason, "session_path", profile.sessionPath, "err", err)
2275	profile.sessionPath = ""
2276	state.sessionPath = ""
2277	state.mappingDegraded = true
2278	return true
2279	}
2280	if err := leases.Rebind(profile.sessionPath); err != nil {
2281	if !degrade("lease held elsewhere", err) {
2282	ctrl.Close()
2283	leases.Release()
2284	gw.logger.Error("attached bot session is in use", "err", control.SessionInUseMessage(err))
2285	return nil
2286	}
2287	} else if loaded, err := agent.LoadSession(profile.sessionPath); err != nil {
2288	if !degrade("load failed", err) {
2289	ctrl.Close()
2290	leases.Release()
2291	if os.IsNotExist(err) {
2292	gw.logger.Error("attached bot session missing", "session_path", profile.sessionPath)
2293	} else {
2294	gw.logger.Error("attached bot session load failed", "session_path", profile.sessionPath, "err", err)
2295	}
2296	return nil
2297	}
2298	} else {
2299	ctrl.Resume(loaded, profile.sessionPath)
2300	}
2301	}
2302	ctrl.EnableInteractiveApproval()
2303	ctrl.SetToolApprovalMode(profile.toolApprovalMode)
2304	ctrl.EnsureSessionPath()
2305	if err := leases.Rebind(ctrl.SessionPath()); err != nil {
2306	ctrl.Close()
2307	leases.Release()
2308	gw.logger.Error("bot session lease failed", "err", control.SessionInUseMessage(err))
2309	return nil
2310	}
2311
2312	var replace *sessionState
2313	gw.mu.Lock()
2314	// Re-check under the lock: while we were off-lock in boot.Build, a second
2315	// message for the same key may have built and registered its own session.
2316	// Reuse it only when it still targets this message's runtime profile.
2317	if existing, ok := gw.controllers[key]; ok {
2318	if sessionStateMatchesRuntime(existing, profile) {
2319	updateSessionStateRuntime(existing, msg, profile)
2320	gw.mu.Unlock()
2321	ctrl.Close()
2322	leases.Release()
2323	safeBotSetToolApprovalMode(existing.ctrl, profile.toolApprovalMode)
2324	gw.logger.Info("bot session built concurrently; discarding duplicate", "platform", msg.Platform, "chat", hashID(msg.ChatID), "session", key[:8])
2325	return existing
2326	}
2327	delete(gw.controllers, key)
2328	replace = existing
2329	}
2330	gw.controllers[key] = state
2331	gw.mu.Unlock()
2332	gw.closeSessionState(replace)
2333
2334	gw.logger.Info("bot session created", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "session", key[:8])
2335	return state
2336	}
2337
2338	func updateSessionStateRuntime(state *sessionState, msg InboundMessage, profile sessionRuntimeProfile) {
2339	if state == nil {
2340	return
2341	}
2342	if state.connectionID == "" {
2343	state.connectionID = strings.TrimSpace(msg.ConnectionID)
2344	}
2345	if state.platform == "" {
2346	state.platform = msg.Platform
2347	}
2348	state.model = profile.model
2349	state.workspaceRoot = profile.workspaceRoot
2350	state.toolApprovalMode = profile.toolApprovalMode
2351	state.sessionPath = profile.sessionPath
2352	state.lastActive = time.Now()
2353	}
2354
2355	func (gw *BotGateway) sessionProfileForMessage(msg InboundMessage) sessionRuntimeProfile {
2356	model, workspaceRoot, toolApprovalMode := gw.sessionOptionsForMessage(msg)
2357	var sessionPath string
2358	sessionPathOptional := false
2359	if override, ok := gw.sessionRuntimeOverrideForMessage(msg); ok {
2360	sessionPath = override.sessionPath
2361	}
2362	// A persisted session_mappings binding is the durable chat→session link
2363	// the desktop writes into the connection config. Without consuming it
2364	// here, every gateway restart or runtime rebuild opened a brand-new
2365	// session file for the chat and the configured binding was display-only
2366	// (#6917, #6934).
2367	if sessionPath == "" {
2368	if mapped := gw.sessionMappingPathForMessage(msg); mapped != "" {
2369	sessionPath = mapped
2370	sessionPathOptional = true
2371	}
2372	}
2373	return sessionRuntimeProfile{
2374	model: strings.TrimSpace(model),
2375	workspaceRoot: strings.TrimSpace(workspaceRoot),
2376	toolApprovalMode: normalizeBotToolApprovalMode(toolApprovalMode),
2377	sessionPath: canonicalBotPath(sessionPath),
2378	sessionPathOptional: sessionPathOptional,
2379	}
2380	}
2381
2382	// sessionMappingPathForMessage resolves the persisted session_mappings entry
2383	// for a message to an existing session file. Only bindings that resolve to a
2384	// present, readable file participate — a moved or deleted target quietly
2385	// degrades to normal session creation rather than blocking the chat.
2386	func (gw *BotGateway) sessionMappingPathForMessage(msg InboundMessage) string {
2387	gw.mu.Lock()
2388	var mappings []SessionMapping
2389	if msg.ConnectionID != "" {
2390	if channel, ok := gw.cfg.ConnectionChannels[msg.ConnectionID]; ok {
2391	mappings = channel.SessionMappings
2392	}
2393	}
2394	if len(mappings) == 0 {
2395	if channel, ok := gw.cfg.Channels[msg.Platform]; ok {
2396	mappings = channel.SessionMappings
2397	}
2398	}
2399	gw.mu.Unlock()
2400	mapping, ok := matchingSessionMapping(mappings, msg)
2401	if !ok {
2402	return ""
2403	}
2404	path := botSessionPathFromTarget(mapping.SessionID)
2405	if path == "" {
2406	path = botSessionPathFromTarget(mapping.SessionSource)
2407	}
2408	if path == "" {
2409	return ""
2410	}
2411	if info, err := os.Stat(path); err != nil \|\| info.IsDir() {
2412	return ""
2413	}
2414	return path
2415	}
2416
2417	func sessionStateMatchesRuntime(state *sessionState, profile sessionRuntimeProfile) bool {
2418	if state == nil \|\| state.ctrl == nil {
2419	return false
2420	}
2421	if stateModel := strings.TrimSpace(state.model); stateModel != "" && profile.model != "" && stateModel != profile.model {
2422	return false
2423	}
2424	stateRoot := strings.TrimSpace(state.workspaceRoot)
2425	wantRoot := strings.TrimSpace(profile.workspaceRoot)
2426	if stateRoot == "" {
2427	root, ok := safeBotControllerWorkspaceRoot(state.ctrl)
2428	if ok {
2429	stateRoot = strings.TrimSpace(root)
2430	} else if wantRoot != "" {
2431	return false
2432	}
2433	}
2434	if stateRoot != wantRoot {
2435	return false
2436	}
2437	// A state that already degraded off its mapped session keeps running on
2438	// its fresh path even though the profile re-resolves the mapping each
2439	// message; rebuilding here would spawn a new session per message while the
2440	// mapped file stays unavailable.
2441	if profile.sessionPathOptional && state.mappingDegraded {
2442	return true
2443	}
2444	if canonicalBotPath(state.sessionPath) != canonicalBotPath(profile.sessionPath) {
2445	return false
2446	}
2447	if profile.sessionPath != "" && canonicalBotPath(state.ctrl.SessionPath()) != canonicalBotPath(profile.sessionPath) {
2448	return false
2449	}
2450	return true
2451	}
2452
2453	func safeBotControllerWorkspaceRoot(ctrl botController) (root string, ok bool) {
2454	if ctrl == nil {
2455	return "", false
2456	}
2457	defer func() {
2458	if recover() != nil {
2459	root = ""
2460	ok = false
2461	}
2462	}()
2463	return ctrl.WorkspaceRoot(), true
2464	}
2465
2466	func safeBotSetToolApprovalMode(ctrl botController, mode string) {
2467	if ctrl == nil {
2468	return
2469	}
2470	defer func() {
2471	_ = recover()
2472	}()
2473	ctrl.SetToolApprovalMode(mode)
2474	}
2475
2476	// defaultBotApprovalTimeout caps how long a bot session waits for a remote
2477	// user's approval/ask reply before treating it as denied, so an abandoned
2478	// prompt (or a dropped IM event) can't leave the session wedged forever
2479	// (#4626, #4402). 30 minutes is generous for a human reply yet bounded.
2480	const defaultBotApprovalTimeout = 30 * time.Minute
2481
2482	// approvalTimeout resolves the configured bot approval wait: zero uses the
2483	// bounded default; a negative value opts out (wait indefinitely).
2484	func (gw *BotGateway) approvalTimeout() time.Duration {
2485	switch {
2486	case gw.cfg.ApprovalTimeout < 0:
2487	return 0
2488	case gw.cfg.ApprovalTimeout == 0:
2489	return defaultBotApprovalTimeout
2490	default:
2491	return gw.cfg.ApprovalTimeout
2492	}
2493	}
2494
2495	func botSessionDir(workspaceRoot string) string {
2496	if strings.TrimSpace(workspaceRoot) == "" {
2497	return config.SessionDir()
2498	}
2499	if dir := config.ProjectSessionDir(workspaceRoot); dir != "" {
2500	return dir
2501	}
2502	return config.SessionDir()
2503	}
2504
2505	func (gw *BotGateway) rememberSessionReady(msg InboundMessage, ctrl botController) {
2506	if gw.cfg.OnSessionReady == nil \|\| ctrl == nil {
2507	return
2508	}
2509	gw.rememberSessionPath(msg, ctrl.SessionPath())
2510	}
2511
2512	func (gw *BotGateway) rememberSessionPath(msg InboundMessage, sessionPath string) {
2513	if gw.cfg.OnSessionReady == nil {
2514	return
2515	}
2516	sessionID := botSessionTarget(sessionPath)
2517	if sessionID == "" {
2518	return
2519	}
2520	if err := gw.cfg.OnSessionReady(msg, sessionID); err != nil {
2521	gw.logger.Warn("remember bot session failed", "platform", msg.Platform, "connection", msg.ConnectionID, "err", err)
2522	}
2523	}
2524
2525	// botSessionRecoveredHandler keeps the controller path, its writer lease, and
2526	// the remote-to-session mapping on the same recovery generation. The lease
2527	// handoff runs first and is failure-atomic: if the recovery path is already
2528	// owned, the controller stays on the original path and the old lease remains
2529	// held. Mapping updates are limited to this exact sessionState so a late
2530	// callback from a retired controller cannot overwrite its replacement.
2531	func (gw BotGateway) botSessionRecoveredHandler(key string, msg InboundMessage, state sessionState) func(control.SessionRecoveryInfo) error {
2532	return func(info control.SessionRecoveryInfo) error {
2533	if state == nil \|\| state.leases == nil {
2534	return nil
2535	}
2536	// Keep the lease handoff and mapping publication atomic with respect to
2537	// state retirement. In particular, never let a callback that outlives
2538	// Stop reacquire a lease after closeSessionState has released it.
2539	state.lifecycleMu.Lock()
2540	defer state.lifecycleMu.Unlock()
2541	if state.retired {
2542	return errBotSessionRetired
2543	}
2544	if err := state.leases.HandleSessionRecovered(info); err != nil {
2545	return err
2546	}
2547
2548	originalPath := canonicalBotPath(info.OriginalPath)
2549	recoveryPath := canonicalBotPath(info.RecoveryPath)
2550	live := false
2551	gw.mu.Lock()
2552	if gw.controllers[key] == state {
2553	live = true
2554	if canonicalBotPath(state.sessionPath) == originalPath {
2555	state.sessionPath = recoveryPath
2556	}
2557	if override, ok := gw.sessionOverrides[key]; ok && canonicalBotPath(override.sessionPath) == originalPath {
2558	override.sessionPath = recoveryPath
2559	gw.sessionOverrides[key] = override
2560	}
2561	}
2562	gw.mu.Unlock()
2563
2564	if live {
2565	gw.rememberSessionPath(msg, recoveryPath)
2566	}
2567	return nil
2568	}
2569	}
2570
2571	func botSessionTarget(sessionPath string) string {
2572	sessionPath = strings.TrimSpace(sessionPath)
2573	if sessionPath == "" {
2574	return ""
2575	}
2576	return "path:" + sessionPath
2577	}
2578
2579	func (gw *BotGateway) sessionOptionsForMessage(msg InboundMessage) (model string, workspaceRoot string, toolApprovalMode string) {
2580	// cfg.ToolApprovalMode / Channels / ConnectionChannels are rewritten under
2581	// gw.mu at runtime (/yolo, UpdateConnectionToolApprovalMode), so snapshot them
2582	// under a short lock and resolve outside it — applyRuntimeOverrideOptions
2583	// takes gw.mu itself. Copying the ChannelConfig value is enough: writers
2584	// replace whole map entries and never mutate SessionMappings in place.
2585	gw.mu.Lock()
2586	model = gw.cfg.Model
2587	workspaceRoot = gw.cfg.WorkspaceRoot
2588	toolApprovalMode = normalizeBotToolApprovalMode(gw.cfg.ToolApprovalMode)
2589	var connChannel ChannelConfig
2590	connOK := false
2591	if msg.ConnectionID != "" {
2592	connChannel, connOK = gw.cfg.ConnectionChannels[msg.ConnectionID]
2593	}
2594	platChannel, platOK := gw.cfg.Channels[msg.Platform]
2595	gw.mu.Unlock()
2596
2597	var mappings []SessionMapping
2598	if connOK {
2599	applyBotChannelOptions(connChannel, &model, &workspaceRoot, &toolApprovalMode)
2600	mappings = connChannel.SessionMappings
2601	if mapping, ok := matchingSessionMapping(mappings, msg); ok {
2602	workspaceRoot = workspaceRootForSessionMapping(mapping, workspaceRoot)
2603	}
2604	model, workspaceRoot, toolApprovalMode = gw.applyRouteOptions(msg, model, workspaceRoot, toolApprovalMode)
2605	model, workspaceRoot, toolApprovalMode = gw.applyRuntimeOverrideOptions(msg, model, workspaceRoot, toolApprovalMode)
2606	return model, workspaceRoot, toolApprovalMode
2607	}
2608	if platOK {
2609	applyBotChannelOptions(platChannel, &model, &workspaceRoot, &toolApprovalMode)
2610	mappings = platChannel.SessionMappings
2611	}
2612	if mapping, ok := matchingSessionMapping(mappings, msg); ok {
2613	workspaceRoot = workspaceRootForSessionMapping(mapping, workspaceRoot)
2614	}
2615	model, workspaceRoot, toolApprovalMode = gw.applyRouteOptions(msg, model, workspaceRoot, toolApprovalMode)
2616	model, workspaceRoot, toolApprovalMode = gw.applyRuntimeOverrideOptions(msg, model, workspaceRoot, toolApprovalMode)
2617	return model, workspaceRoot, toolApprovalMode
2618	}
2619
2620	func (gw *BotGateway) applyRuntimeOverrideOptions(msg InboundMessage, model, workspaceRoot, toolApprovalMode string) (string, string, string) {
2621	if override, ok := gw.sessionRuntimeOverrideForMessage(msg); ok {
2622	applyBotChannelOptions(override.channel, &model, &workspaceRoot, &toolApprovalMode)
2623	}
2624	return model, workspaceRoot, toolApprovalMode
2625	}
2626
2627	func (gw *BotGateway) applyRouteOptions(msg InboundMessage, model, workspaceRoot, toolApprovalMode string) (string, string, string) {
2628	for _, route := range gw.cfg.Routes {
2629	if routeMatchesMessage(route, msg) {
2630	applyBotChannelOptions(route.Channel, &model, &workspaceRoot, &toolApprovalMode)
2631	break
2632	}
2633	}
2634	return model, workspaceRoot, toolApprovalMode
2635	}
2636
2637	func applyBotChannelOptions(channel ChannelConfig, model string, workspaceRoot string, toolApprovalMode *string) {
2638	if value := strings.TrimSpace(channel.Model); value != "" {
2639	*model = value
2640	}
2641	if value := strings.TrimSpace(channel.WorkspaceRoot); value != "" {
2642	*workspaceRoot = value
2643	}
2644	if value := normalizeOptionalBotToolApprovalMode(channel.ToolApprovalMode); value != "" {
2645	*toolApprovalMode = value
2646	}
2647	}
2648
2649	func matchingSessionMapping(mappings []SessionMapping, msg InboundMessage) (SessionMapping, bool) {
2650	for i := range mappings {
2651	if sessionMappingMatches(mappings[i], msg) {
2652	return mappings[i], true
2653	}
2654	}
2655	return SessionMapping{}, false
2656	}
2657
2658	func sessionMappingMatches(mapping SessionMapping, msg InboundMessage) bool {
2659	if strings.TrimSpace(mapping.RemoteID) != strings.TrimSpace(msg.ChatID) {
2660	return false
2661	}
2662	chatType, userID, threadID := sessionMappingIdentity(msg)
2663	mappingChatType := strings.TrimSpace(mapping.ChatType)
2664	if mappingChatType == "" {
2665	return chatType == ""
2666	}
2667	if mappingChatType != chatType {
2668	return false
2669	}
2670	if strings.TrimSpace(mapping.UserID) != userID {
2671	return false
2672	}
2673	return strings.TrimSpace(mapping.ThreadID) == threadID
2674	}
2675
2676	func sessionMappingIdentity(msg InboundMessage) (chatType string, userID string, threadID string) {
2677	switch msg.ChatType {
2678	case ChatGroup, ChatGuild:
2679	chatType = string(msg.ChatType)
2680	userID = strings.TrimSpace(msg.UserID)
2681	case ChatThread:
2682	chatType = string(msg.ChatType)
2683	threadID = strings.TrimSpace(msg.ThreadID)
2684	if threadID == "" {
2685	threadID = strings.TrimSpace(msg.ChatID)
2686	}
2687	}
2688	return chatType, userID, threadID
2689	}
2690
2691	func workspaceRootForSessionMapping(mapping SessionMapping, fallback string) string {
2692	if root := strings.TrimSpace(mapping.WorkspaceRoot); root != "" {
2693	return root
2694	}
2695	if strings.EqualFold(strings.TrimSpace(mapping.Scope), "global") {
2696	return ""
2697	}
2698	return fallback
2699	}
2700
2701	func routeMatchesMessage(route RouteConfig, msg InboundMessage) bool {
2702	if value := strings.TrimSpace(route.ConnectionID); value != "" && value != strings.TrimSpace(msg.ConnectionID) {
2703	return false
2704	}
2705	if route.Platform != "" && route.Platform != msg.Platform {
2706	return false
2707	}
2708	if route.ChatType != "" && route.ChatType != msg.ChatType {
2709	return false
2710	}
2711	if value := strings.TrimSpace(route.ChatID); value != "" && value != strings.TrimSpace(msg.ChatID) {
2712	return false
2713	}
2714	if value := strings.TrimSpace(route.UserID); value != "" && value != strings.TrimSpace(msg.UserID) {
2715	return false
2716	}
2717	if value := strings.TrimSpace(route.ThreadID); value != "" && value != strings.TrimSpace(msg.ThreadID) {
2718	return false
2719	}
2720	return true
2721	}
2722
2723	func normalizeBotToolApprovalMode(mode string) string {
2724	if value := normalizeOptionalBotToolApprovalMode(mode); value != "" {
2725	return value
2726	}
2727	return control.ToolApprovalAsk
2728	}
2729
2730	func normalizeOptionalBotToolApprovalMode(mode string) string {
2731	switch strings.ToLower(strings.TrimSpace(mode)) {
2732	case control.ToolApprovalAsk:
2733	return control.ToolApprovalAsk
2734	case control.ToolApprovalAuto:
2735	return control.ToolApprovalAuto
2736	case control.ToolApprovalYolo, "full", "full-access", "bypass":
2737	return control.ToolApprovalYolo
2738	default:
2739	return ""
2740	}
2741	}
2742
2743	func (gw *BotGateway) sendText(ctx context.Context, adapter Adapter, msg InboundMessage, text string) error {
2744	out := OutboundMessage{
2745	ConnectionID: msg.ConnectionID,
2746	Domain: msg.Domain,
2747	ChatID: msg.ChatID,
2748	ChatType: msg.ChatType,
2749	Text: text,
2750	ReplyToMsgID: msg.MessageID,
2751	}
2752	binding := AdapterBinding{
2753	ID: strings.TrimSpace(msg.ConnectionID),
2754	Domain: strings.TrimSpace(msg.Domain),
2755	Platform: msg.Platform,
2756	Adapter: adapter,
2757	}
2758	if binding.Platform == "" && adapter != nil {
2759	binding.Platform = adapter.Platform()
2760	}
2761	if binding.ID == "" && adapter != nil {
2762	binding.ID = adapter.Name()
2763	}
2764	result, err := gw.sendViaAdapter(ctx, binding, out)
2765	if err != nil {
2766	gw.logger.Warn("bot send failed", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "reply_to", hashID(msg.MessageID), "err", err)
2767	return err
2768	}
2769	gw.logger.Info("bot send completed", "platform", msg.Platform, "chat_type", msg.ChatType, "chat", hashID(msg.ChatID), "reply_to", hashID(msg.MessageID), "message", hashID(result.MessageID))
2770	return err
2771	}
2772
2773	func (gw *BotGateway) sendViaAdapter(ctx context.Context, binding AdapterBinding, msg OutboundMessage) (SendResult, error) {
2774	if binding.Adapter == nil {
2775	return SendResult{}, errors.New("bot send: adapter is nil")
2776	}
2777	if strings.TrimSpace(msg.ConnectionID) == "" {
2778	msg.ConnectionID = binding.ID
2779	}
2780	if strings.TrimSpace(msg.Domain) == "" {
2781	msg.Domain = binding.Domain
2782	}
2783	result, err := binding.Adapter.Send(ctx, msg)
2784	gw.markAdapterSend(binding, err)
2785	for _, messageID := range result.DeliveredMessageIDs() {
2786	gw.rememberOutboundMessage(binding.Platform, binding.ID, binding.Domain, msg.ChatID, messageID)
2787	}
2788	return result, err
2789	}
2790
2791	func parseAskAnswers(questions []event.AskQuestion, raw string) []event.AskAnswer {
2792	raw = strings.TrimSpace(raw)
2793	if len(questions) == 0 {
2794	return []event.AskAnswer{{Selected: []string{raw}}}
2795	}
2796	byID := make(map[string]*event.AskQuestion, len(questions))
2797	for i := range questions {
2798	q := &questions[i]
2799	byID[q.ID] = q
2800	byID[fmt.Sprintf("%d", i+1)] = q
2801	}
2802	answerMap := make(map[string][]string, len(questions))
2803	if strings.Contains(raw, "=") {
2804	for _, part := range strings.Split(raw, ";") {
2805	k, v, ok := strings.Cut(part, "=")
2806	if !ok {
2807	continue
2808	}
2809	q := byID[strings.TrimSpace(k)]
2810	if q == nil {
2811	continue
2812	}
2813	answerMap[q.ID] = normalizeAskSelection(*q, strings.TrimSpace(v))
2814	}
2815	} else if len(questions) == 1 {
2816	answerMap[questions[0].ID] = normalizeAskSelection(questions[0], raw)
2817	}
2818	out := make([]event.AskAnswer, 0, len(questions))
2819	for _, q := range questions {
2820	out = append(out, event.AskAnswer{QuestionID: q.ID, Selected: answerMap[q.ID]})
2821	}
2822	return out
2823	}
2824
2825	func normalizeAskSelection(q event.AskQuestion, raw string) []string {
2826	parts := []string{raw}
2827	if q.Multi && strings.Contains(raw, ",") {
2828	parts = strings.Split(raw, ",")
2829	}
2830	out := make([]string, 0, len(parts))
2831	for _, part := range parts {
2832	part = strings.TrimSpace(part)
2833	if part == "" {
2834	continue
2835	}
2836	if idx, err := strconv.Atoi(part); err == nil && idx >= 1 && idx <= len(q.Options) {
2837	out = append(out, q.Options[idx-1].Label)
2838	continue
2839	}
2840	out = append(out, part)
2841	}
2842	return out
2843	}
2844
2845	// UpdateConnectionToolApprovalMode updates the in-memory tool approval mode for
2846	// a single bot connection without restarting the gateway. Empty mode clears the
2847	// connection override, so existing sessions inherit the current gateway default.
2848	func (gw *BotGateway) UpdateConnectionToolApprovalMode(connID, mode string) {
2849	connID = strings.TrimSpace(connID)
2850	if connID == "" {
2851	return
2852	}
2853	mode = normalizeOptionalBotToolApprovalMode(mode)
2854	type controllerMode struct {
2855	ctrl botController
2856	mode string
2857	}
2858	var updates []controllerMode
2859
2860	gw.mu.Lock()
2861	if gw.cfg.ConnectionChannels == nil {
2862	gw.cfg.ConnectionChannels = make(map[string]ChannelConfig)
2863	}
2864	ch := gw.cfg.ConnectionChannels[connID]
2865	ch.ToolApprovalMode = mode
2866	gw.cfg.ConnectionChannels[connID] = ch
2867	// Update every active session that belongs to this connection.
2868	for _, state := range gw.controllers {
2869	if state == nil \|\| state.ctrl == nil \|\| strings.TrimSpace(state.connectionID) != connID {
2870	continue
2871	}
2872	effectiveMode := mode
2873	if effectiveMode == "" {
2874	effectiveMode = normalizeBotToolApprovalMode(gw.cfg.ToolApprovalMode)
2875	}
2876	updates = append(updates, controllerMode{ctrl: state.ctrl, mode: effectiveMode})
2877	}
2878	gw.mu.Unlock()
2879
2880	for _, update := range updates {
2881	update.ctrl.SetToolApprovalMode(update.mode)
2882	}
2883	}
2884
2885	// SendToAdapter sends a message through the adapter identified by connID.
2886	// Returns an error if no matching adapter is found.
2887	func (gw *BotGateway) SendToAdapter(ctx context.Context, connID, domain string, msg OutboundMessage) (SendResult, error) {
2888	connID = strings.TrimSpace(connID)
2889	domain = strings.TrimSpace(domain)
2890	var target AdapterBinding
2891	gw.mu.Lock()
2892	for _, binding := range gw.adapters {
2893	if strings.TrimSpace(binding.ID) == connID &&
2894	(domain == "" \|\| strings.EqualFold(strings.TrimSpace(binding.Domain), domain)) {
2895	target = binding
2896	break
2897	}
2898	}
2899	gw.mu.Unlock()
2900	if target.Adapter != nil {
2901	return gw.sendViaAdapter(ctx, target, msg)
2902	}
2903	return SendResult{}, fmt.Errorf("SendToAdapter: no adapter found for connection %q (domain %q)", connID, domain)
2904	}
2905
2906	// SendTextToAdapter sends a plain text message through the adapter identified by connID.
2907	func (gw *BotGateway) SendTextToAdapter(ctx context.Context, connID, domain, chatID string, chatType ChatType, text string) (SendResult, error) {
2908	return gw.SendToAdapter(ctx, connID, domain, OutboundMessage{
2909	ChatID: chatID,
2910	ChatType: chatType,
2911	Text: text,
2912	})
2913	}
2914

2914 lines GO