gateway_test.go - DeepSeek-Reasonix

gateway_test.go

1	package bot
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"io"
8	"log/slog"
9	"net/http"
10	"net/http/httptest"
11	"os"
12	"path/filepath"
13	"strings"
14	"sync"
15	"testing"
16	"time"
17
18	"reasonix/internal/agent"
19	"reasonix/internal/control"
20	"reasonix/internal/event"
21	"reasonix/internal/provider"
22	"reasonix/internal/tool"
23	)
24
25	// fakeAdapter 是一个内存中的假适配器，用于测试 BotGateway。
26	type fakeAdapter struct {
27	mu sync.Mutex
28	stopOnce sync.Once
29	platform Platform
30	name string
31	msgCh chan InboundMessage
32	sent []OutboundMessage
33	started bool
34	startErr error
35	}
36
37	type resultAdapter struct {
38	*fakeAdapter
39	result SendResult
40	err error
41	}
42
43	func (a *resultAdapter) Send(_ context.Context, msg OutboundMessage) (SendResult, error) {
44	a.mu.Lock()
45	a.sent = append(a.sent, msg)
46	a.mu.Unlock()
47	return a.result, a.err
48	}
49
50	func newFakeAdapter(platform Platform, name string) *fakeAdapter {
51	return &fakeAdapter{
52	platform: platform,
53	name: name,
54	msgCh: make(chan InboundMessage, 16),
55	}
56	}
57
58	func (f *fakeAdapter) Platform() Platform { return f.platform }
59	func (f *fakeAdapter) Name() string { return f.name }
60	func (f *fakeAdapter) Messages() <-chan InboundMessage { return f.msgCh }
61
62	func (f *fakeAdapter) Start(ctx context.Context) error {
63	if f.startErr != nil {
64	return f.startErr
65	}
66	f.mu.Lock()
67	f.started = true
68	f.mu.Unlock()
69	return nil
70	}
71
72	func (f *fakeAdapter) Stop() error {
73	f.stopOnce.Do(func() {
74	close(f.msgCh)
75	})
76	return nil
77	}
78
79	func (f *fakeAdapter) Send(ctx context.Context, msg OutboundMessage) (SendResult, error) {
80	f.mu.Lock()
81	f.sent = append(f.sent, msg)
82	f.mu.Unlock()
83	return SendResult{MessageID: "fake_msg_1"}, nil
84	}
85
86	func (f *fakeAdapter) SendTyping(ctx context.Context, chatID string) error { return nil }
87
88	func (f *fakeAdapter) sentMessages() []OutboundMessage {
89	f.mu.Lock()
90	defer f.mu.Unlock()
91	out := make([]OutboundMessage, len(f.sent))
92	copy(out, f.sent)
93	return out
94	}
95
96	type blockingSendAdapter struct {
97	*fakeAdapter
98	entered chan struct{}
99	release chan struct{}
100	once sync.Once
101	}
102
103	func newBlockingSendAdapter(platform Platform, name string) *blockingSendAdapter {
104	return &blockingSendAdapter{
105	fakeAdapter: newFakeAdapter(platform, name),
106	entered: make(chan struct{}),
107	release: make(chan struct{}),
108	}
109	}
110
111	func (f *blockingSendAdapter) Send(ctx context.Context, msg OutboundMessage) (SendResult, error) {
112	f.once.Do(func() { close(f.entered) })
113	select {
114	case <-f.release:
115	case <-ctx.Done():
116	return SendResult{}, ctx.Err()
117	}
118	return f.fakeAdapter.Send(ctx, msg)
119	}
120
121	type fakeReactionAdapter struct {
122	*fakeAdapter
123	reactions []string
124	cleanups []string
125	}
126
127	type gatewayFakeProvider struct{}
128
129	func (gatewayFakeProvider) Name() string { return "fake" }
130
131	func (gatewayFakeProvider) Stream(context.Context, provider.Request) (<-chan provider.Chunk, error) {
132	ch := make(chan provider.Chunk)
133	close(ch)
134	return ch, nil
135	}
136
137	func (f *fakeReactionAdapter) AddPendingReaction(ctx context.Context, messageID string) (func(), error) {
138	f.mu.Lock()
139	f.reactions = append(f.reactions, messageID)
140	f.mu.Unlock()
141	return func() {
142	f.mu.Lock()
143	defer f.mu.Unlock()
144	f.cleanups = append(f.cleanups, messageID)
145	}, nil
146	}
147
148	func (f *fakeReactionAdapter) cleanupMessages() []string {
149	f.mu.Lock()
150	defer f.mu.Unlock()
151	out := make([]string, len(f.cleanups))
152	copy(out, f.cleanups)
153	return out
154	}
155
156	type queueTestController struct {
157	botController
158	mu sync.Mutex
159	steers []string
160	rejectSteer bool
161	canceled bool
162	}
163
164	func (c *queueTestController) Steer(text string) {
165	_ = c.TrySteer(text)
166	}
167
168	func (c *queueTestController) TrySteer(text string) bool {
169	c.mu.Lock()
170	defer c.mu.Unlock()
171	if c.rejectSteer {
172	return false
173	}
174	c.steers = append(c.steers, text)
175	return true
176	}
177
178	func (c *queueTestController) Cancel() {
179	c.mu.Lock()
180	defer c.mu.Unlock()
181	c.canceled = true
182	}
183
184	func (c *queueTestController) SessionPath() string { return "" }
185	func (c *queueTestController) WorkspaceRoot() string { return "" }
186
187	func (c *queueTestController) steered() []string {
188	c.mu.Lock()
189	defer c.mu.Unlock()
190	out := make([]string, len(c.steers))
191	copy(out, c.steers)
192	return out
193	}
194
195	func (c *queueTestController) wasCanceled() bool {
196	c.mu.Lock()
197	defer c.mu.Unlock()
198	return c.canceled
199	}
200
201	type rotatingBotController struct {
202	botController
203	path string
204	newPath string
205	newCalls int
206	closed bool
207	}
208
209	func (c *rotatingBotController) Running() bool { return false }
210	func (c *rotatingBotController) NewSession() error {
211	c.newCalls++
212	c.path = c.newPath
213	return nil
214	}
215	func (c *rotatingBotController) SessionPath() string { return c.path }
216	func (c *rotatingBotController) Close() { c.closed = true }
217
218	type runtimeStatusBotController struct {
219	botController
220	status control.RuntimeStatus
221	workspaceRoot string
222	sessionPath string
223	closed bool
224	}
225
226	func (c *runtimeStatusBotController) RuntimeStatus() control.RuntimeStatus { return c.status }
227	func (c *runtimeStatusBotController) WorkspaceRoot() string { return c.workspaceRoot }
228	func (c *runtimeStatusBotController) SessionPath() string { return c.sessionPath }
229	func (c *runtimeStatusBotController) Close() { c.closed = true }
230
231	type blockingApprovalController struct {
232	botController
233	emit func(event.Event)
234	emitted chan struct{}
235	approved chan struct{}
236	done chan struct{}
237	once sync.Once
238	}
239
240	func (c *blockingApprovalController) RunTurn(ctx context.Context, input string) error {
241	c.emit(event.Event{Kind: event.ApprovalRequest, Approval: event.Approval{ID: "appr-1", Tool: "bash", Subject: "sample command"}})
242	close(c.emitted)
243	select {
244	case <-c.approved:
245	close(c.done)
246	return nil
247	case <-ctx.Done():
248	return ctx.Err()
249	}
250	}
251
252	func (c *blockingApprovalController) Approve(id string, allow, session, persist bool) {
253	c.once.Do(func() { close(c.approved) })
254	}
255
256	type blockingAskController struct {
257	botController
258	emit func(event.Event)
259	emitted chan struct{}
260	answered chan []event.AskAnswer
261	done chan struct{}
262	once sync.Once
263	}
264
265	func (c *blockingAskController) RunTurn(ctx context.Context, input string) error {
266	c.emit(event.Event{Kind: event.AskRequest, Ask: event.Ask{ID: "ask-1", Questions: []event.AskQuestion{{
267	ID: "q1",
268	Header: "Planner",
269	Prompt: "Which plan?",
270	Options: []event.AskOption{
271	{Label: "Small patch"},
272	{Label: "Refactor"},
273	},
274	}}}})
275	close(c.emitted)
276	select {
277	case <-c.answered:
278	close(c.done)
279	return nil
280	case <-ctx.Done():
281	return ctx.Err()
282	}
283	}
284
285	func (c *blockingAskController) AnswerQuestion(id string, answers []event.AskAnswer) {
286	c.once.Do(func() { c.answered <- answers })
287	}
288
289	func TestFakeAdapterInterface(t *testing.T) {
290	fa := newFakeAdapter(PlatformQQ, "fake-qq")
291
292	if fa.Platform() != PlatformQQ {
293	t.Error("wrong platform")
294	}
295	if fa.Name() != "fake-qq" {
296	t.Error("wrong name")
297	}
298
299	ctx := context.Background()
300	if err := fa.Start(ctx); err != nil {
301	t.Fatal("start:", err)
302	}
303	if !fa.started {
304	t.Error("should be started")
305	}
306
307	_, err := fa.Send(ctx, OutboundMessage{ChatID: "c1", Text: "hello"})
308	if err != nil {
309	t.Fatal("send:", err)
310	}
311
312	sent := fa.sentMessages()
313	if len(sent) != 1 {
314	t.Fatalf("sent count = %d, want 1", len(sent))
315	}
316	if sent[0].Text != "hello" {
317	t.Errorf("sent text = %q, want %q", sent[0].Text, "hello")
318	}
319
320	if err := fa.Stop(); err != nil {
321	t.Fatal("stop:", err)
322	}
323	}
324
325	func TestGatewayConstructAndStop(t *testing.T) {
326	cfg := GatewayConfig{
327	Model: "test",
328	MaxSteps: 10,
329	WorkspaceRoot: ".",
330	Enabled: map[Platform]bool{PlatformQQ: true},
331	Allowlist: AllowlistConfig{Enabled: false},
332	}
333
334	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
335	gw := NewGateway(cfg, map[Platform]Adapter{
336	PlatformQQ: newFakeAdapter(PlatformQQ, "fake-qq"),
337	}, logger)
338
339	// 网关不应该 panic
340	if gw == nil {
341	t.Fatal("gateway should not be nil")
342	}
343	gw.Stop()
344	}
345
346	func TestGatewayStartsHealthyAdaptersWhenOneFails(t *testing.T) {
347	cfg := GatewayConfig{
348	Enabled: map[Platform]bool{PlatformFeishu: true, PlatformWeixin: true},
349	Allowlist: AllowlistConfig{AllowAll: true},
350	}
351	good := newFakeAdapter(PlatformFeishu, "good-feishu")
352	bad := newFakeAdapter(PlatformWeixin, "bad-weixin")
353	bad.startErr = errors.New("missing token")
354	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
355	gw := NewGatewayWithAdapterBindings(cfg, []AdapterBinding{
356	{ID: "feishu-lark", Platform: PlatformFeishu, Adapter: good},
357	{ID: "weixin-weixin", Platform: PlatformWeixin, Adapter: bad},
358	}, logger)
359
360	if err := gw.Start(context.Background()); err != nil {
361	t.Fatalf("start should keep healthy adapters running: %v", err)
362	}
363	defer gw.Stop()
364	if got := gw.AdapterCount(); got != 1 {
365	t.Fatalf("adapter count = %d, want 1", got)
366	}
367	if !good.started {
368	t.Fatal("healthy adapter was not started")
369	}
370	if bad.started {
371	t.Fatal("failing adapter should not be marked started")
372	}
373	startErr := gw.StartErrors()
374	if len(startErr) != 1 \|\| !strings.Contains(startErr[0].Error(), "weixin-weixin") {
375	t.Fatalf("start errors = %#v, want wrapped connection error", startErr)
376	}
377	}
378
379	func TestGatewaySendToAdapterReleasesLockBeforeSend(t *testing.T) {
380	adapter := newBlockingSendAdapter(PlatformFeishu, "blocking-feishu")
381	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
382	gw := NewGatewayWithAdapterBindings(GatewayConfig{}, []AdapterBinding{{
383	ID: "feishu-lark",
384	Domain: "lark",
385	Platform: PlatformFeishu,
386	Adapter: adapter,
387	}}, logger)
388
389	sendDone := make(chan error, 1)
390	go func() {
391	_, err := gw.SendToAdapter(context.Background(), "feishu-lark", "lark", OutboundMessage{ChatID: "chat", Text: "hello"})
392	sendDone <- err
393	}()
394
395	select {
396	case <-adapter.entered:
397	case <-time.After(500 * time.Millisecond):
398	t.Fatal("adapter send did not start")
399	}
400
401	updateDone := make(chan struct{})
402	go func() {
403	gw.UpdateConnectionToolApprovalMode("feishu-lark", "ask")
404	close(updateDone)
405	}()
406	select {
407	case <-updateDone:
408	case <-time.After(500 * time.Millisecond):
409	t.Fatal("UpdateConnectionToolApprovalMode blocked behind SendToAdapter")
410	}
411
412	close(adapter.release)
413	select {
414	case err := <-sendDone:
415	if err != nil {
416	t.Fatalf("SendToAdapter returned error: %v", err)
417	}
418	case <-time.After(500 * time.Millisecond):
419	t.Fatal("SendToAdapter did not finish after release")
420	}
421	}
422
423	func TestGatewayReturnsErrorWhenAllAdaptersFail(t *testing.T) {
424	cfg := GatewayConfig{
425	Enabled: map[Platform]bool{PlatformWeixin: true},
426	Allowlist: AllowlistConfig{AllowAll: true},
427	}
428	bad := newFakeAdapter(PlatformWeixin, "bad-weixin")
429	bad.startErr = errors.New("missing token")
430	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
431	gw := NewGatewayWithAdapterBindings(cfg, []AdapterBinding{
432	{ID: "weixin-weixin", Platform: PlatformWeixin, Adapter: bad},
433	}, logger)
434
435	err := gw.Start(context.Background())
436	if err == nil {
437	t.Fatal("start should fail when every adapter fails")
438	}
439	if !strings.Contains(err.Error(), "weixin-weixin") {
440	t.Fatalf("error = %v, want connection id", err)
441	}
442	if got := gw.AdapterCount(); got != 0 {
443	t.Fatalf("adapter count = %d, want 0", got)
444	}
445	if len(gw.StartErrors()) != 1 {
446	t.Fatalf("start errors = %#v, want one", gw.StartErrors())
447	}
448	}
449
450	func TestGatewayAllowlistCheck(t *testing.T) {
451	cfg := GatewayConfig{
452	Allowlist: AllowlistConfig{
453	Enabled: true,
454	Users: map[Platform][]string{
455	PlatformQQ: {"allowed_user_1"},
456	},
457	},
458	}
459
460	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
461	gw := NewGateway(cfg, nil, logger)
462
463	if !gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatDM, UserID: "allowed_user_1"}) {
464	t.Error("allowed user should pass")
465	}
466	if gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatDM, UserID: "unknown_user"}) {
467	t.Error("unknown user should not pass")
468	}
469	// 不同平台
470	if gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, UserID: "allowed_user_1"}) {
471	t.Error("QQ allowlist should not apply to feishu")
472	}
473	}
474
475	func TestGatewayRejectsBeforeInboundEnrichment(t *testing.T) {
476	adapter := newFakeAdapter(PlatformFeishu, "feishu")
477	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{
478	Enabled: true,
479	Users: map[Platform][]string{PlatformFeishu: {"allowed-user"}},
480	}}, map[Platform]Adapter{PlatformFeishu: adapter}, discardLogger())
481
482	mediaLoads := 0
483	nameLoads := 0
484	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu", Platform: PlatformFeishu, Adapter: adapter}, InboundMessage{
485	Platform: PlatformFeishu,
486	ChatType: ChatDM,
487	ChatID: "chat",
488	UserID: "blocked-user",
489	Media: []InboundMedia{{Load: func(context.Context) ([]byte, string, error) {
490	mediaLoads++
491	return []byte("payload"), "payload.txt", nil
492	}}},
493	ResolveUserName: func(context.Context) string {
494	nameLoads++
495	return "Blocked User"
496	},
497	})
498
499	if mediaLoads != 0 \|\| nameLoads != 0 {
500	t.Fatalf("pre-admission enrichment calls = media:%d name:%d, want zero", mediaLoads, nameLoads)
501	}
502	}
503
504	func TestGatewayRoleListsGrantAllowlistAdmission(t *testing.T) {
505	cfg := GatewayConfig{
506	Allowlist: AllowlistConfig{
507	Enabled: true,
508	Admins: map[Platform][]string{
509	PlatformFeishu: {"admin_user"},
510	},
511	Approvers: map[Platform][]string{
512	PlatformFeishu: {"approver_user"},
513	},
514	},
515	}
516
517	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
518	gw := NewGateway(cfg, nil, logger)
519
520	if !gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, UserID: "admin_user"}) {
521	t.Error("admin role should grant base bot admission")
522	}
523	if !gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, UserID: "approver_user"}) {
524	t.Error("approver role should grant base bot admission")
525	}
526	if gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, UserID: "unknown_user"}) {
527	t.Error("unknown user should still be rejected")
528	}
529	}
530
531	func TestGatewayApproverRoleDoesNotGrantAdminCommands(t *testing.T) {
532	cfg := GatewayConfig{
533	Allowlist: AllowlistConfig{
534	Enabled: true,
535	Approvers: map[Platform][]string{
536	PlatformFeishu: {"approver_user"},
537	},
538	},
539	}
540
541	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
542	gw := NewGateway(cfg, nil, logger)
543	msg := InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, UserID: "approver_user"}
544
545	if !gw.checkCommandRole(PlatformFeishu, msg, "approver") {
546	t.Error("approver should be allowed to run approver commands")
547	}
548	if gw.checkCommandRole(PlatformFeishu, msg, "admin") {
549	t.Error("approver should not be allowed to run admin commands")
550	}
551	}
552
553	func TestGatewayAllowlistDoesNotApplyGroupsToDirectMessages(t *testing.T) {
554	cfg := GatewayConfig{
555	Allowlist: AllowlistConfig{
556	Enabled: true,
557	Users: map[Platform][]string{
558	PlatformQQ: {"allowed_user"},
559	},
560	Groups: map[Platform][]string{
561	PlatformQQ: {"allowed_group"},
562	},
563	},
564	}
565
566	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
567	gw := NewGateway(cfg, nil, logger)
568
569	if !gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatDirect, ChatID: "guild-dm", UserID: "allowed_user"}) {
570	t.Error("direct message should not be rejected by group allowlist")
571	}
572	if gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatGroup, ChatID: "unknown_group", UserID: "allowed_user"}) {
573	t.Error("unknown group should still be rejected by group allowlist")
574	}
575	}
576
577	func TestGatewayGroupAllowlistStillNarrowsRoleAdmission(t *testing.T) {
578	cfg := GatewayConfig{
579	Allowlist: AllowlistConfig{
580	Enabled: true,
581	Admins: map[Platform][]string{
582	PlatformFeishu: {"admin_user"},
583	},
584	Groups: map[Platform][]string{
585	PlatformFeishu: {"allowed_group"},
586	},
587	},
588	}
589
590	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
591	gw := NewGateway(cfg, nil, logger)
592
593	if !gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, ChatID: "direct", UserID: "admin_user"}) {
594	t.Error("admin role admission should still allow direct messages")
595	}
596	if !gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatGroup, ChatID: "allowed_group", UserID: "admin_user"}) {
597	t.Error("admin role should pass in allowed group")
598	}
599	if gw.checkAllowlist(PlatformFeishu, InboundMessage{Platform: PlatformFeishu, ChatType: ChatGroup, ChatID: "unknown_group", UserID: "admin_user"}) {
600	t.Error("admin role should still be rejected in an unknown group")
601	}
602	}
603
604	func TestGatewayAllowlistGatesOnOperatorNotCardRequester(t *testing.T) {
605	cfg := GatewayConfig{
606	Allowlist: AllowlistConfig{
607	Enabled: true,
608	Users: map[Platform][]string{
609	PlatformFeishu: {"requester"},
610	},
611	},
612	}
613	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
614	gw := NewGateway(cfg, nil, logger)
615
616	stranger := InboundMessage{Platform: PlatformFeishu, ChatType: ChatGroup, ChatID: "chat", UserID: "requester", OperatorID: "stranger"}
617	if gw.checkAllowlist(PlatformFeishu, stranger) {
618	t.Error("a non-allowlisted operator must be rejected even when the card carries an allowlisted requester id")
619	}
620
621	allowed := InboundMessage{Platform: PlatformFeishu, ChatType: ChatGroup, ChatID: "chat", UserID: "requester", OperatorID: "requester"}
622	if !gw.checkAllowlist(PlatformFeishu, allowed) {
623	t.Error("an allowlisted operator should pass")
624	}
625	}
626
627	func TestGatewayAllowlistDisabledRejectsByDefault(t *testing.T) {
628	cfg := GatewayConfig{
629	Allowlist: AllowlistConfig{Enabled: false},
630	}
631	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
632	gw := NewGateway(cfg, nil, logger)
633
634	if gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatDM, UserID: "any_user"}) {
635	t.Error("disabled allowlist should reject unless allow_all is explicit")
636	}
637	}
638
639	func TestGatewayAllowAll(t *testing.T) {
640	cfg := GatewayConfig{
641	Allowlist: AllowlistConfig{AllowAll: true},
642	}
643	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
644	gw := NewGateway(cfg, nil, logger)
645
646	if !gw.checkAllowlist(PlatformQQ, InboundMessage{Platform: PlatformQQ, ChatType: ChatDM, UserID: "any_user"}) {
647	t.Error("allow_all should allow everyone")
648	}
649	}
650
651	func TestGatewayNormalizesNumericApprovalShortcutsOnlyWhenPending(t *testing.T) {
652	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
653	gw := NewGateway(GatewayConfig{}, nil, logger)
654	key := "session-key"
655
656	if _, ok := gw.normalizeApprovalShortcut(key, "1"); ok {
657	t.Fatal("numeric text without a pending approval should stay a normal message")
658	}
659
660	gw.controllers[key] = &sessionState{
661	pendingApprovals: map[string]event.Approval{
662	"42": {ID: "42", Tool: "explore"},
663	},
664	lastApprovalID: "42",
665	}
666
667	got, ok := gw.normalizeApprovalShortcut(key, "1")
668	if !ok \|\| got != "/approve 42" {
669	t.Fatalf("normalize 1 = %q,%v; want /approve 42,true", got, ok)
670	}
671	got, ok = gw.normalizeApprovalShortcut(key, "2")
672	if !ok \|\| got != "/deny 42" {
673	t.Fatalf("normalize 2 = %q,%v; want /deny 42,true", got, ok)
674	}
675	gw.forgetPendingApproval(key, "42")
676	if _, ok := gw.normalizeApprovalShortcut(key, "1"); ok {
677	t.Fatal("numeric text after approval is forgotten should stay a normal message")
678	}
679	}
680
681	func TestGatewayNormalizesTaskGrantRecoveryShortcuts(t *testing.T) {
682	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
683	gw := NewGateway(GatewayConfig{}, nil, logger)
684	key := "recovery-key"
685	gw.controllers[key] = &sessionState{
686	pendingApprovals: map[string]event.Approval{
687	"r1": {ID: "r1", Kind: "recovery", Recovery: &event.RecoveryApproval{CanGrantTask: true}},
688	},
689	lastApprovalID: "r1",
690	}
691	for input, want := range map[string]string{
692	"1": "/recovery-continue r1",
693	"2": "/recovery-continue-task r1",
694	"3": "/recovery-revise r1",
695	} {
696	got, ok := gw.normalizeApprovalShortcut(key, input)
697	if !ok \|\| got != want {
698	t.Fatalf("normalize %q = %q,%v; want %q,true", input, got, ok, want)
699	}
700	}
701	}
702
703	func TestGatewayNormalizesAskShortcutForPendingAsk(t *testing.T) {
704	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
705	gw := NewGateway(GatewayConfig{}, nil, logger)
706	key := "session-key"
707
708	if _, ok := gw.normalizeAskShortcut(key, "1"); ok {
709	t.Fatal("numeric text without a pending ask should stay a normal message")
710	}
711
712	gw.controllers[key] = &sessionState{
713	pendingAsks: map[string][]event.AskQuestion{
714	"ask-1": {{
715	ID: "q1",
716	Prompt: "Choose one",
717	Options: []event.AskOption{
718	{Label: "Allow once"},
719	{Label: "Deny"},
720	},
721	}},
722	},
723	lastAskID: "ask-1",
724	}
725
726	got, ok := gw.normalizeAskShortcut(key, "2")
727	if !ok \|\| got != "/answer ask-1 2" {
728	t.Fatalf("normalize 2 = %q,%v; want /answer ask-1 2,true", got, ok)
729	}
730	got, ok = gw.normalizeAskShortcut(key, "1;2")
731	if !ok \|\| got != "/answer ask-1 1;2" {
732	t.Fatalf("normalize 1;2 = %q,%v; want /answer ask-1 1;2,true", got, ok)
733	}
734	got, ok = gw.normalizeAskShortcut(key, "freeform answer")
735	if !ok \|\| got != "/answer ask-1 freeform answer" {
736	t.Fatalf("normalize freeform answer = %q,%v; want /answer ask-1 freeform answer,true", got, ok)
737	}
738
739	gw.controllers[key].pendingAsks["ask-2"] = []event.AskQuestion{
740	{ID: "q1", Prompt: "First", Options: []event.AskOption{{Label: "A"}}},
741	{ID: "q2", Prompt: "Second", Options: []event.AskOption{{Label: "B"}}},
742	}
743	gw.controllers[key].lastAskID = "ask-2"
744	got, ok = gw.normalizeAskShortcut(key, "1")
745	if !ok \|\| got != "/answer ask-2 1" {
746	t.Fatalf("normalize 1 on multi-question = %q,%v; want /answer ask-2 1,true", got, ok)
747	}
748	if _, ok := gw.normalizeAskShortcut(key, "/stop"); ok {
749	t.Fatal("slash commands should not be normalized/routed by ask shortcut")
750	}
751	}
752
753	func TestGatewaySessionOptionsUseConnectionToolApprovalOverride(t *testing.T) {
754	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
755	gw := NewGateway(GatewayConfig{
756	Model: "default-model",
757	ToolApprovalMode: "auto",
758	Channels: map[Platform]ChannelConfig{
759	PlatformFeishu: {Model: "platform-model", ToolApprovalMode: "ask"},
760	},
761	ConnectionChannels: map[string]ChannelConfig{
762	"feishu-lark": {Model: "lark-model", ToolApprovalMode: "yolo"},
763	},
764	}, nil, logger)
765
766	model, _, mode := gw.sessionOptionsForMessage(InboundMessage{
767	Platform: PlatformFeishu,
768	ConnectionID: "feishu-lark",
769	})
770	if model != "lark-model" \|\| mode != "yolo" {
771	t.Fatalf("lark session options = model %q mode %q, want lark-model/yolo", model, mode)
772	}
773
774	model, _, mode = gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformFeishu})
775	if model != "platform-model" \|\| mode != "ask" {
776	t.Fatalf("platform session options = model %q mode %q, want platform-model/ask", model, mode)
777	}
778	}
779
780	func TestGatewayNumericApprovalShortcutActiveWithoutPendingSendsGuidance(t *testing.T) {
781	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
782	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
783	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
784	binding := AdapterBinding{ID: "weixin-weixin", Domain: "weixin", Platform: PlatformWeixin, Adapter: adapter}
785	msg := InboundMessage{
786	Platform: PlatformWeixin,
787	ConnectionID: "weixin-weixin",
788	Domain: "weixin",
789	ChatType: ChatDM,
790	ChatID: "chat",
791	UserID: "user",
792	Text: "seed",
793	}
794	key := BuildSessionKey(msg.Session())
795	if acquired, _ := gw.sessions.TryAcquire(key, msg); !acquired {
796	t.Fatal("failed to mark session active")
797	}
798
799	msg.Text = "1"
800	gw.handleMessage(context.Background(), binding, msg)
801
802	sent := adapter.sentMessages()
803	if len(sent) != 1 {
804	t.Fatalf("sent count = %d, want 1", len(sent))
805	}
806	if !strings.Contains(sent[0].Text, "没有找到可匹配的待处理操作") {
807	t.Fatalf("sent text = %q, want pending operation guidance", sent[0].Text)
808	}
809	}
810
811	func TestGatewayApproveWithoutSessionSendsGuidance(t *testing.T) {
812	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
813	gw := NewGateway(GatewayConfig{}, nil, logger)
814	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
815	msg := InboundMessage{ChatType: ChatDM, ChatID: "chat", UserID: "user", Text: "/approve 1"}
816
817	gw.handleSlashCommand(context.Background(), adapter, "missing-session", msg)
818
819	sent := adapter.sentMessages()
820	if len(sent) != 1 {
821	t.Fatalf("sent count = %d, want 1", len(sent))
822	}
823	if !strings.Contains(sent[0].Text, "没有找到当前会话中的待审批操作") {
824	t.Fatalf("sent text = %q, want missing approval guidance", sent[0].Text)
825	}
826	}
827
828	func TestGatewayNewSessionRemembersRotatedSessionPath(t *testing.T) {
829	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
830	var remembered string
831	gw := NewGateway(GatewayConfig{
832	OnSessionReady: func(msg InboundMessage, sessionID string) error {
833	remembered = sessionID
834	return nil
835	},
836	}, nil, logger)
837	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
838	msg := InboundMessage{
839	Platform: PlatformWeixin,
840	ConnectionID: "weixin-weixin",
841	Domain: "weixin",
842	ChatType: ChatDM,
843	ChatID: "chat",
844	UserID: "user",
845	Text: "/new",
846	}
847	key := BuildSessionKey(msg.Session())
848	sessionDir := t.TempDir()
849	oldPath := agent.NewSessionPath(sessionDir, "old-model")
850	exec := agent.New(gatewayFakeProvider{}, tool.NewRegistry(), agent.NewSession("system"), agent.Options{}, event.Discard)
851	ctrl := control.New(control.Options{Executor: exec, SessionDir: sessionDir, SessionPath: oldPath, Label: "fake-model"})
852	leases := control.NewSessionLeaseKeeper()
853	if err := leases.Rebind(oldPath); err != nil {
854	t.Fatalf("bind old session lease: %v", err)
855	}
856	gw.controllers[key] = &sessionState{ctrl: ctrl, leases: leases, sessionPath: oldPath}
857
858	gw.handleSlashCommand(context.Background(), adapter, key, msg)
859
860	if remembered == "" \|\| !strings.HasPrefix(remembered, "path:") {
861	t.Fatalf("remembered session = %q, want path target", remembered)
862	}
863	if remembered == "path:"+oldPath {
864	t.Fatalf("remembered session = %q, want rotated path", remembered)
865	}
866	if ctrl.SessionPath() == oldPath {
867	t.Fatalf("controller session path was not rotated")
868	}
869	if got := leases.HeldPath(); got != agent.CanonicalSessionPath(ctrl.SessionPath()) {
870	t.Fatalf("held lease = %q, want rotated path %q", got, agent.CanonicalSessionPath(ctrl.SessionPath()))
871	}
872	oldLease, err := agent.TryAcquireSessionLease(oldPath)
873	if err != nil {
874	t.Fatalf("old session lease was not released: %v", err)
875	}
876	oldLease.Release()
877	gw.closeSessions()
878	}
879
880	func TestGatewayRecoveryRebindsLeaseAndRemembersSessionPath(t *testing.T) {
881	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
882	dir := t.TempDir()
883	originalPath := filepath.Join(dir, "session.jsonl")
884
885	disk := agent.NewSession("sys")
886	disk.Add(provider.Message{Role: provider.RoleUser, Content: "first"})
887	disk.Add(provider.Message{Role: provider.RoleAssistant, Content: "disk"})
888	if err := disk.Save(originalPath); err != nil {
889	t.Fatalf("save disk session: %v", err)
890	}
891
892	local := agent.NewSession("sys")
893	local.Add(provider.Message{Role: provider.RoleUser, Content: "first"})
894	local.Add(provider.Message{Role: provider.RoleAssistant, Content: "local"})
895	exec := agent.New(nil, nil, local, agent.Options{}, event.Discard)
896
897	var remembered []string
898	gw := NewGateway(GatewayConfig{
899	OnSessionReady: func(_ InboundMessage, sessionID string) error {
900	remembered = append(remembered, sessionID)
901	return nil
902	},
903	}, nil, logger)
904	msg := InboundMessage{
905	Platform: PlatformWeixin,
906	ConnectionID: "weixin-main",
907	Domain: "weixin",
908	ChatType: ChatDM,
909	ChatID: "chat",
910	UserID: "user",
911	}
912	key := BuildSessionKey(msg.Session())
913	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
914	sessionSink := &sessionEventSink{}
915	sessionSink.setTarget(newRenderSink(
916	context.Background(), adapter, msg.ConnectionID, msg.Domain, msg.ChatID,
917	msg.ChatType, msg.UserID, msg.MessageID, logger, nil, nil,
918	))
919	t.Cleanup(func() { sessionSink.setTarget(nil) })
920	leases := control.NewSessionLeaseKeeper()
921	if err := leases.Rebind(originalPath); err != nil {
922	t.Fatalf("bind original session lease: %v", err)
923	}
924	state := &sessionState{sink: sessionSink, leases: leases, sessionPath: originalPath}
925	gw.controllers[key] = state
926	gw.sessionOverrides[key] = sessionRuntimeOverride{sessionPath: originalPath, label: "session:original"}
927	ctrl := control.New(control.Options{
928	Executor: exec,
929	SessionDir: dir,
930	SessionPath: originalPath,
931	Label: "test",
932	Sink: sessionSink,
933	OnSessionRecovered: gw.botSessionRecoveredHandler(key, msg, state),
934	})
935	state.ctrl = ctrl
936	t.Cleanup(gw.closeSessions)
937
938	if err := ctrl.Snapshot(); err != nil {
939	t.Fatalf("snapshot diverged session: %v", err)
940	}
941	recoveryPath := ctrl.SessionPath()
942	if recoveryPath == "" \|\| recoveryPath == originalPath {
943	t.Fatalf("controller path = %q, want recovery path", recoveryPath)
944	}
945	if got := leases.HeldPath(); got != agent.CanonicalSessionPath(recoveryPath) {
946	t.Fatalf("held lease = %q, want recovery path %q", got, agent.CanonicalSessionPath(recoveryPath))
947	}
948	oldLease, err := agent.TryAcquireSessionLease(originalPath)
949	if err != nil {
950	t.Fatalf("original session lease was not released: %v", err)
951	}
952	oldLease.Release()
953
954	gw.mu.Lock()
955	gotStatePath := state.sessionPath
956	gotOverridePath := gw.sessionOverrides[key].sessionPath
957	gw.mu.Unlock()
958	if canonicalBotPath(gotStatePath) != canonicalBotPath(recoveryPath) {
959	t.Fatalf("state path = %q, want recovery path %q", gotStatePath, recoveryPath)
960	}
961	if canonicalBotPath(gotOverridePath) != canonicalBotPath(recoveryPath) {
962	t.Fatalf("override path = %q, want recovery path %q", gotOverridePath, recoveryPath)
963	}
964	if len(remembered) != 1 \|\| remembered[0] != botSessionTarget(recoveryPath) {
965	t.Fatalf("remembered sessions = %v, want [%q]", remembered, botSessionTarget(recoveryPath))
966	}
967
968	if err := ctrl.Snapshot(); err != nil {
969	t.Fatalf("snapshot recovered session: %v", err)
970	}
971	matches, err := filepath.Glob(filepath.Join(dir, "-recovery-.jsonl"))
972	if err != nil {
973	t.Fatalf("glob recovery sessions: %v", err)
974	}
975	transcripts := matches[:0]
976	for _, path := range matches {
977	if !strings.HasSuffix(path, ".events.jsonl") && !strings.HasSuffix(path, ".conflicts.jsonl") {
978	transcripts = append(transcripts, path)
979	}
980	}
981	if len(transcripts) != 1 \|\| transcripts[0] != recoveryPath {
982	t.Fatalf("recovery transcripts = %v, want only %q", transcripts, recoveryPath)
983	}
984	if sent := adapter.sentMessages(); len(sent) != 0 {
985	t.Fatalf("recovery maintenance leaked into IM messages: %+v", sent)
986	}
987	}
988
989	func TestGatewayRecoveryLeaseFailureKeepsOriginalGeneration(t *testing.T) {
990	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
991	dir := t.TempDir()
992	originalPath := filepath.Join(dir, "original.jsonl")
993	recoveryPath := filepath.Join(dir, "recovery.jsonl")
994	readyCalls := 0
995	gw := NewGateway(GatewayConfig{
996	OnSessionReady: func(InboundMessage, string) error {
997	readyCalls++
998	return nil
999	},
1000	}, nil, logger)
1001	msg := InboundMessage{Platform: PlatformWeixin, ChatType: ChatDM, ChatID: "chat", UserID: "user"}
1002	key := BuildSessionKey(msg.Session())
1003	leases := control.NewSessionLeaseKeeper()
1004	if err := leases.Rebind(originalPath); err != nil {
1005	t.Fatalf("bind original session lease: %v", err)
1006	}
1007	defer leases.Release()
1008	blocker, err := agent.TryAcquireSessionLease(recoveryPath)
1009	if err != nil {
1010	t.Fatalf("bind recovery blocker: %v", err)
1011	}
1012	defer blocker.Release()
1013	state := &sessionState{leases: leases, sessionPath: originalPath}
1014	gw.controllers[key] = state
1015	gw.sessionOverrides[key] = sessionRuntimeOverride{sessionPath: originalPath}
1016
1017	err = gw.botSessionRecoveredHandler(key, msg, state)(control.SessionRecoveryInfo{
1018	OriginalPath: originalPath,
1019	RecoveryPath: recoveryPath,
1020	})
1021	if err == nil {
1022	t.Fatal("recovery handoff succeeded while recovery lease was held")
1023	}
1024	if got := leases.HeldPath(); got != agent.CanonicalSessionPath(originalPath) {
1025	t.Fatalf("held lease = %q, want original path %q", got, agent.CanonicalSessionPath(originalPath))
1026	}
1027	gw.mu.Lock()
1028	gotStatePath := state.sessionPath
1029	gotOverridePath := gw.sessionOverrides[key].sessionPath
1030	gw.mu.Unlock()
1031	if gotStatePath != originalPath \|\| gotOverridePath != originalPath {
1032	t.Fatalf("paths changed after failed handoff: state=%q override=%q", gotStatePath, gotOverridePath)
1033	}
1034	if readyCalls != 0 {
1035	t.Fatalf("session-ready callback ran %d times after failed handoff", readyCalls)
1036	}
1037	}
1038
1039	func TestGatewayLateRecoveryCannotReplaceCurrentSessionMapping(t *testing.T) {
1040	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1041	dir := t.TempDir()
1042	oldPath := filepath.Join(dir, "old.jsonl")
1043	oldRecoveryPath := filepath.Join(dir, "old-recovery.jsonl")
1044	currentPath := filepath.Join(dir, "current.jsonl")
1045	readyCalls := 0
1046	gw := NewGateway(GatewayConfig{
1047	OnSessionReady: func(InboundMessage, string) error {
1048	readyCalls++
1049	return nil
1050	},
1051	}, nil, logger)
1052	msg := InboundMessage{Platform: PlatformWeixin, ChatType: ChatDM, ChatID: "chat", UserID: "user"}
1053	key := BuildSessionKey(msg.Session())
1054	oldLeases := control.NewSessionLeaseKeeper()
1055	if err := oldLeases.Rebind(oldPath); err != nil {
1056	t.Fatalf("bind old session lease: %v", err)
1057	}
1058	defer oldLeases.Release()
1059	oldState := &sessionState{leases: oldLeases, sessionPath: oldPath}
1060	currentState := &sessionState{sessionPath: currentPath}
1061	gw.controllers[key] = currentState
1062	gw.sessionOverrides[key] = sessionRuntimeOverride{sessionPath: currentPath}
1063
1064	if err := gw.botSessionRecoveredHandler(key, msg, oldState)(control.SessionRecoveryInfo{
1065	OriginalPath: oldPath,
1066	RecoveryPath: oldRecoveryPath,
1067	}); err != nil {
1068	t.Fatalf("late recovery handoff: %v", err)
1069	}
1070	if got := oldLeases.HeldPath(); got != agent.CanonicalSessionPath(oldRecoveryPath) {
1071	t.Fatalf("old generation lease = %q, want recovery path %q", got, agent.CanonicalSessionPath(oldRecoveryPath))
1072	}
1073	gw.mu.Lock()
1074	gotCurrentPath := currentState.sessionPath
1075	gotOverridePath := gw.sessionOverrides[key].sessionPath
1076	gw.mu.Unlock()
1077	if gotCurrentPath != currentPath \|\| gotOverridePath != currentPath {
1078	t.Fatalf("current mapping overwritten by late recovery: state=%q override=%q", gotCurrentPath, gotOverridePath)
1079	}
1080	if readyCalls != 0 {
1081	t.Fatalf("session-ready callback ran %d times for retired generation", readyCalls)
1082	}
1083	}
1084
1085	func TestGatewayLateRecoveryAfterRetirementDoesNotReacquireLease(t *testing.T) {
1086	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1087	dir := t.TempDir()
1088	originalPath := filepath.Join(dir, "original.jsonl")
1089	recoveryPath := filepath.Join(dir, "recovery.jsonl")
1090	gw := NewGateway(GatewayConfig{}, nil, logger)
1091	msg := InboundMessage{Platform: PlatformWeixin, ChatType: ChatDM, ChatID: "chat", UserID: "user"}
1092	key := BuildSessionKey(msg.Session())
1093	leases := control.NewSessionLeaseKeeper()
1094	if err := leases.Rebind(originalPath); err != nil {
1095	t.Fatalf("bind original session lease: %v", err)
1096	}
1097	state := &sessionState{leases: leases, sessionPath: originalPath}
1098	gw.controllers[key] = state
1099	handler := gw.botSessionRecoveredHandler(key, msg, state)
1100
1101	// Gateway shutdown retires and releases the state before waiting for every
1102	// turn goroutine. A callback already captured by the controller must not
1103	// reacquire a lease after that teardown.
1104	gw.closeSessions()
1105	if got := leases.HeldPath(); got != "" {
1106	t.Fatalf("lease after retirement = %q, want empty", got)
1107	}
1108	if err := handler(control.SessionRecoveryInfo{
1109	OriginalPath: originalPath,
1110	RecoveryPath: recoveryPath,
1111	}); !errors.Is(err, errBotSessionRetired) {
1112	t.Fatalf("late recovery error = %v, want %v", err, errBotSessionRetired)
1113	}
1114	if got := leases.HeldPath(); got != "" {
1115	t.Fatalf("late recovery reacquired lease after retirement: %q", got)
1116	}
1117	probe, err := agent.TryAcquireSessionLease(recoveryPath)
1118	if err != nil {
1119	t.Fatalf("recovery lease remained unavailable after late callback: %v", err)
1120	}
1121	probe.Release()
1122	}
1123
1124	func TestGatewayNewSessionLeaseFailureRetiresSession(t *testing.T) {
1125	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1126	readyCalls := 0
1127	gw := NewGateway(GatewayConfig{
1128	OnSessionReady: func(InboundMessage, string) error {
1129	readyCalls++
1130	return nil
1131	},
1132	}, nil, logger)
1133	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1134	msg := InboundMessage{
1135	Platform: PlatformWeixin,
1136	ConnectionID: "weixin-weixin",
1137	Domain: "weixin",
1138	ChatType: ChatDM,
1139	ChatID: "chat",
1140	UserID: "user",
1141	Text: "/new",
1142	}
1143	key := BuildSessionKey(msg.Session())
1144	sessionDir := t.TempDir()
1145	oldPath := filepath.Join(sessionDir, "old.jsonl")
1146	newPath := filepath.Join(sessionDir, "new.jsonl")
1147	ctrl := &rotatingBotController{path: oldPath, newPath: newPath}
1148	leases := control.NewSessionLeaseKeeper()
1149	if err := leases.Rebind(oldPath); err != nil {
1150	t.Fatalf("bind old session lease: %v", err)
1151	}
1152	blocker, err := agent.TryAcquireSessionLease(newPath)
1153	if err != nil {
1154	t.Fatalf("hold rotated session lease: %v", err)
1155	}
1156	defer blocker.Release()
1157	gw.controllers[key] = &sessionState{ctrl: ctrl, leases: leases, sessionPath: oldPath}
1158
1159	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1160
1161	gw.mu.Lock()
1162	_, exists := gw.controllers[key]
1163	gw.mu.Unlock()
1164	if exists {
1165	t.Fatal("lease-failed session remains registered")
1166	}
1167	if ctrl.newCalls != 1 \|\| !ctrl.closed {
1168	t.Fatalf("controller lifecycle = new calls %d closed %v, want 1/true", ctrl.newCalls, ctrl.closed)
1169	}
1170	if got := leases.HeldPath(); got != "" {
1171	t.Fatalf("old lease remained held after retirement: %q", got)
1172	}
1173	oldLease, err := agent.TryAcquireSessionLease(oldPath)
1174	if err != nil {
1175	t.Fatalf("old session lease was not released after retirement: %v", err)
1176	}
1177	oldLease.Release()
1178	if readyCalls != 0 {
1179	t.Fatalf("session-ready callback ran %d times after failed creation", readyCalls)
1180	}
1181	sent := adapter.sentMessages()
1182	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "新会话创建失败") \|\| strings.Contains(sent[0].Text, "已开始新会话") {
1183	t.Fatalf("sent messages = %+v, want a single creation-failed response", sent)
1184	}
1185	}
1186
1187	func TestGatewayCloseSessionStateReleasesSessionLease(t *testing.T) {
1188	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1189	gw := NewGateway(GatewayConfig{}, nil, logger)
1190	path := filepath.Join(t.TempDir(), "session.jsonl")
1191	leases := control.NewSessionLeaseKeeper()
1192	if err := leases.Rebind(path); err != nil {
1193	t.Fatalf("bind session lease: %v", err)
1194	}
1195	ctrl := control.New(control.Options{})
1196
1197	gw.closeSessionState(&sessionState{ctrl: ctrl, leases: leases})
1198
1199	lease, err := agent.TryAcquireSessionLease(path)
1200	if err != nil {
1201	t.Fatalf("session lease was not released after controller close: %v", err)
1202	}
1203	lease.Release()
1204	}
1205
1206	func TestGatewayYoloCommandUpdatesCurrentSessionAndConnectionDefault(t *testing.T) {
1207	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1208	var persistedMode string
1209	var persistedConnection string
1210	gw := NewGateway(GatewayConfig{
1211	ToolApprovalMode: "ask",
1212	ConnectionChannels: map[string]ChannelConfig{
1213	"feishu-lark": {ToolApprovalMode: "ask"},
1214	},
1215	OnToolApprovalModeChange: func(msg InboundMessage, mode string) error {
1216	persistedConnection = msg.ConnectionID
1217	persistedMode = mode
1218	return nil
1219	},
1220	}, nil, logger)
1221	adapter := newFakeAdapter(PlatformFeishu, "fake-lark")
1222	msg := InboundMessage{
1223	Platform: PlatformFeishu,
1224	ConnectionID: "feishu-lark",
1225	Domain: "lark",
1226	ChatType: ChatDM,
1227	ChatID: "chat",
1228	UserID: "user",
1229	Text: "/yolo on",
1230	}
1231	key := BuildSessionKey(msg.Session())
1232	ctrl := control.New(control.Options{})
1233	ctrl.SetToolApprovalMode(control.ToolApprovalAsk)
1234	gw.controllers[key] = &sessionState{ctrl: ctrl}
1235
1236	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1237
1238	if got := ctrl.ToolApprovalMode(); got != control.ToolApprovalYolo {
1239	t.Fatalf("current session mode = %q, want yolo", got)
1240	}
1241	if got := gw.cfg.ConnectionChannels["feishu-lark"].ToolApprovalMode; got != control.ToolApprovalYolo {
1242	t.Fatalf("connection default mode = %q, want yolo", got)
1243	}
1244	if persistedConnection != "feishu-lark" \|\| persistedMode != control.ToolApprovalYolo {
1245	t.Fatalf("persisted = %q/%q, want feishu-lark/yolo", persistedConnection, persistedMode)
1246	}
1247	sent := adapter.sentMessages()
1248	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "已开启 YOLO") {
1249	t.Fatalf("sent = %#v, want yolo confirmation", sent)
1250	}
1251	}
1252
1253	func TestGatewayUpdateConnectionToolApprovalModeUpdatesHashedActiveSessions(t *testing.T) {
1254	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1255	gw := NewGateway(GatewayConfig{
1256	ToolApprovalMode: "ask",
1257	ConnectionChannels: map[string]ChannelConfig{
1258	"feishu-lark": {ToolApprovalMode: "yolo"},
1259	},
1260	}, nil, logger)
1261
1262	msg := InboundMessage{
1263	Platform: PlatformFeishu,
1264	ConnectionID: "feishu-lark",
1265	Domain: "lark",
1266	ChatType: ChatDM,
1267	ChatID: "chat",
1268	UserID: "user",
1269	}
1270	key := BuildSessionKey(msg.Session())
1271	if strings.HasPrefix(key, msg.ConnectionID) {
1272	t.Fatalf("test setup expected hashed key, got %q", key)
1273	}
1274	ctrl := control.New(control.Options{})
1275	ctrl.SetToolApprovalMode(control.ToolApprovalYolo)
1276	gw.controllers[key] = &sessionState{ctrl: ctrl, platform: msg.Platform, connectionID: msg.ConnectionID}
1277
1278	gw.UpdateConnectionToolApprovalMode("feishu-lark", control.ToolApprovalAsk)
1279
1280	if got := ctrl.ToolApprovalMode(); got != control.ToolApprovalAsk {
1281	t.Fatalf("active session mode = %q, want ask", got)
1282	}
1283	if got := gw.cfg.ConnectionChannels["feishu-lark"].ToolApprovalMode; got != control.ToolApprovalAsk {
1284	t.Fatalf("connection default mode = %q, want ask", got)
1285	}
1286	}
1287
1288	func TestGatewayUpdateConnectionToolApprovalModeInheritsGatewayDefault(t *testing.T) {
1289	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1290	gw := NewGateway(GatewayConfig{
1291	ToolApprovalMode: control.ToolApprovalAuto,
1292	ConnectionChannels: map[string]ChannelConfig{
1293	"feishu-lark": {ToolApprovalMode: control.ToolApprovalYolo},
1294	"feishu-feishu": {ToolApprovalMode: control.ToolApprovalYolo},
1295	},
1296	}, nil, logger)
1297
1298	larkMsg := InboundMessage{
1299	Platform: PlatformFeishu,
1300	ConnectionID: "feishu-lark",
1301	Domain: "lark",
1302	ChatType: ChatDM,
1303	ChatID: "chat",
1304	UserID: "user",
1305	}
1306	larkKey := BuildSessionKey(larkMsg.Session())
1307	larkCtrl := control.New(control.Options{})
1308	larkCtrl.SetToolApprovalMode(control.ToolApprovalYolo)
1309	gw.controllers[larkKey] = &sessionState{ctrl: larkCtrl, platform: larkMsg.Platform, connectionID: larkMsg.ConnectionID}
1310
1311	otherCtrl := control.New(control.Options{})
1312	otherCtrl.SetToolApprovalMode(control.ToolApprovalYolo)
1313	gw.controllers["other-hashed-key"] = &sessionState{ctrl: otherCtrl, platform: PlatformFeishu, connectionID: "feishu-feishu"}
1314
1315	gw.UpdateConnectionToolApprovalMode("feishu-lark", "")
1316
1317	if got := gw.cfg.ConnectionChannels["feishu-lark"].ToolApprovalMode; got != "" {
1318	t.Fatalf("connection override = %q, want empty inherit", got)
1319	}
1320	if got := larkCtrl.ToolApprovalMode(); got != control.ToolApprovalAuto {
1321	t.Fatalf("lark active session mode = %q, want inherited auto", got)
1322	}
1323	if got := otherCtrl.ToolApprovalMode(); got != control.ToolApprovalYolo {
1324	t.Fatalf("other connection mode = %q, want unchanged yolo", got)
1325	}
1326	}
1327
1328	func TestGatewayApprovalReplyUnblocksWedgedTurn(t *testing.T) {
1329	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1330	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1331	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1332	binding := AdapterBinding{ID: "feishu", Platform: PlatformFeishu, Adapter: adapter}
1333	msg := InboundMessage{
1334	Platform: PlatformFeishu,
1335	ConnectionID: "feishu",
1336	ChatType: ChatDM,
1337	ChatID: "chat",
1338	UserID: "user",
1339	Text: "delete everything",
1340	}
1341	key := BuildSessionKey(msg.Session())
1342	sink := &sessionEventSink{}
1343	ctrl := &blockingApprovalController{
1344	emit: sink.Emit,
1345	emitted: make(chan struct{}),
1346	approved: make(chan struct{}),
1347	done: make(chan struct{}),
1348	}
1349	gw.controllers[key] = &sessionState{
1350	ctrl: ctrl,
1351	sink: sink,
1352	pendingApprovals: make(map[string]event.Approval),
1353	pendingAsks: make(map[string][]event.AskQuestion),
1354	}
1355
1356	ctx, cancel := context.WithCancel(context.Background())
1357	defer cancel()
1358	go gw.dispatchLoop(ctx, binding)
1359
1360	adapter.msgCh <- msg
1361	select {
1362	case <-ctrl.emitted:
1363	case <-time.After(2 * time.Second):
1364	t.Fatal("approval request was never emitted; turn did not start")
1365	}
1366
1367	adapter.msgCh <- InboundMessage{
1368	Platform: PlatformFeishu,
1369	ConnectionID: "feishu",
1370	ChatType: ChatDM,
1371	ChatID: "chat",
1372	UserID: "user",
1373	Text: "/approve appr-1",
1374	}
1375
1376	select {
1377	case <-ctrl.done:
1378	case <-time.After(2 * time.Second):
1379	t.Fatal("deadlock: /approve reply was not delivered while the turn blocked on approval")
1380	}
1381	}
1382
1383	func TestGatewayAskReplyUnblocksWedgedTurn(t *testing.T) {
1384	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1385	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1386	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1387	binding := AdapterBinding{ID: "feishu", Platform: PlatformFeishu, Adapter: adapter}
1388	msg := InboundMessage{
1389	Platform: PlatformFeishu,
1390	ConnectionID: "feishu",
1391	ChatType: ChatDM,
1392	ChatID: "chat",
1393	UserID: "user",
1394	Text: "choose a plan",
1395	}
1396	key := BuildSessionKey(msg.Session())
1397	sink := &sessionEventSink{}
1398	ctrl := &blockingAskController{
1399	emit: sink.Emit,
1400	emitted: make(chan struct{}),
1401	answered: make(chan []event.AskAnswer, 1),
1402	done: make(chan struct{}),
1403	}
1404	gw.controllers[key] = &sessionState{
1405	ctrl: ctrl,
1406	sink: sink,
1407	pendingApprovals: make(map[string]event.Approval),
1408	pendingAsks: make(map[string][]event.AskQuestion),
1409	}
1410
1411	ctx, cancel := context.WithCancel(context.Background())
1412	defer cancel()
1413	go gw.dispatchLoop(ctx, binding)
1414
1415	adapter.msgCh <- msg
1416	select {
1417	case <-ctrl.emitted:
1418	case <-time.After(2 * time.Second):
1419	t.Fatal("ask request was never emitted; turn did not start")
1420	}
1421
1422	adapter.msgCh <- InboundMessage{
1423	Platform: PlatformFeishu,
1424	ConnectionID: "feishu",
1425	ChatType: ChatDM,
1426	ChatID: "chat",
1427	UserID: "user",
1428	Text: "1",
1429	}
1430
1431	select {
1432	case <-ctrl.done:
1433	case <-time.After(2 * time.Second):
1434	t.Fatal("deadlock: ask reply was not delivered while the turn blocked on user choice")
1435	}
1436	}
1437
1438	func TestGatewayModeCommandSupportsAskAutoAndStatus(t *testing.T) {
1439	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1440	gw := NewGateway(GatewayConfig{
1441	ConnectionChannels: map[string]ChannelConfig{
1442	"weixin-weixin": {ToolApprovalMode: "ask"},
1443	},
1444	}, nil, logger)
1445	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1446	msg := InboundMessage{
1447	Platform: PlatformWeixin,
1448	ConnectionID: "weixin-weixin",
1449	Domain: "weixin",
1450	ChatType: ChatDM,
1451	ChatID: "chat",
1452	UserID: "user",
1453	}
1454	key := BuildSessionKey(msg.Session())
1455
1456	msg.Text = "/mode auto"
1457	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1458	if got := gw.cfg.ConnectionChannels["weixin-weixin"].ToolApprovalMode; got != control.ToolApprovalAuto {
1459	t.Fatalf("/mode auto default = %q, want auto", got)
1460	}
1461
1462	msg.Text = "/yolo off"
1463	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1464	if got := gw.cfg.ConnectionChannels["weixin-weixin"].ToolApprovalMode; got != control.ToolApprovalAsk {
1465	t.Fatalf("/yolo off default = %q, want ask", got)
1466	}
1467
1468	msg.Text = "/mode"
1469	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1470	sent := adapter.sentMessages()
1471	if len(sent) != 3 {
1472	t.Fatalf("sent count = %d, want 3", len(sent))
1473	}
1474	if !strings.Contains(sent[2].Text, "当前工具审批模式：询问") {
1475	t.Fatalf("status = %q, want ask status", sent[2].Text)
1476	}
1477	}
1478
1479	func TestGatewayHelpMentionsYoloCommands(t *testing.T) {
1480	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1481	gw := NewGateway(GatewayConfig{}, nil, logger)
1482	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1483	msg := InboundMessage{ChatType: ChatDM, ChatID: "chat", UserID: "user", Text: "/help"}
1484
1485	gw.handleSlashCommand(context.Background(), adapter, "session-key", msg)
1486
1487	sent := adapter.sentMessages()
1488	if len(sent) != 1 {
1489	t.Fatalf("sent count = %d, want 1", len(sent))
1490	}
1491	if !strings.Contains(sent[0].Text, "/yolo on\|off\|auto\|status") \|\| !strings.Contains(sent[0].Text, "/mode yolo\|ask\|auto") {
1492	t.Fatalf("help = %q, want yolo commands", sent[0].Text)
1493	}
1494	if !strings.Contains(sent[0].Text, "/projects") \|\| !strings.Contains(sent[0].Text, "/attach session") \|\| !strings.Contains(sent[0].Text, "/search all") {
1495	t.Fatalf("help = %q, want project/session commands", sent[0].Text)
1496	}
1497	}
1498
1499	func TestGatewayProjectCommandsListAndUseProjectOverride(t *testing.T) {
1500	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1501	base := t.TempDir()
1502	alpha := filepath.Join(base, "alpha-project")
1503	beta := filepath.Join(base, "beta-project")
1504	if err := os.MkdirAll(alpha, 0o755); err != nil {
1505	t.Fatal(err)
1506	}
1507	if err := os.MkdirAll(beta, 0o755); err != nil {
1508	t.Fatal(err)
1509	}
1510	gw := NewGateway(GatewayConfig{
1511	WorkspaceRoot: alpha,
1512	ConnectionChannels: map[string]ChannelConfig{
1513	"weixin-main": {WorkspaceRoot: beta},
1514	},
1515	}, nil, logger)
1516	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1517	msg := InboundMessage{
1518	Platform: PlatformWeixin,
1519	ConnectionID: "weixin-main",
1520	ChatType: ChatDM,
1521	ChatID: "chat",
1522	UserID: "user",
1523	Text: "/projects",
1524	}
1525	key := BuildSessionKey(msg.Session())
1526
1527	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1528	sent := adapter.sentMessages()
1529	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "alpha-project") \|\| !strings.Contains(sent[0].Text, "beta-project") {
1530	t.Fatalf("/projects sent = %#v, want both projects", sent)
1531	}
1532
1533	msg.Text = "/use project alpha"
1534	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1535	_, root, _ := gw.sessionOptionsForMessage(msg)
1536	if canonicalBotPath(root) != canonicalBotPath(alpha) {
1537	t.Fatalf("workspace after /use project = %q, want %q", root, alpha)
1538	}
1539
1540	msg.Text = "/use project default"
1541	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1542	_, root, _ = gw.sessionOptionsForMessage(msg)
1543	if canonicalBotPath(root) != canonicalBotPath(beta) {
1544	t.Fatalf("workspace after /use project default = %q, want connection default %q", root, beta)
1545	}
1546	}
1547
1548	func TestGatewaySessionsSearchAndAttachSessionOverride(t *testing.T) {
1549	t.Setenv("REASONIX_HOME", t.TempDir())
1550	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1551	projectRoot := filepath.Join(t.TempDir(), "attach-project")
1552	if err := os.MkdirAll(projectRoot, 0o755); err != nil {
1553	t.Fatal(err)
1554	}
1555	sessionDir := botSessionDir(projectRoot)
1556	sessionPath := filepath.Join(sessionDir, "attached.jsonl")
1557	sess := agent.NewSession("system")
1558	sess.Add(provider.Message{Role: provider.RoleUser, Content: "needle attach conversation"})
1559	if err := sess.Save(sessionPath); err != nil {
1560	t.Fatalf("Save session: %v", err)
1561	}
1562	if err := agent.UpdateSessionMeta(sessionPath, "model-a", "needle attach conversation", 1, true); err != nil {
1563	t.Fatalf("UpdateSessionMeta: %v", err)
1564	}
1565	gw := NewGateway(GatewayConfig{WorkspaceRoot: projectRoot}, nil, logger)
1566	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1567	msg := InboundMessage{
1568	Platform: PlatformFeishu,
1569	ConnectionID: "feishu-lark",
1570	ChatType: ChatDM,
1571	ChatID: "chat",
1572	UserID: "user",
1573	Text: "/sessions search needle",
1574	}
1575	key := BuildSessionKey(msg.Session())
1576
1577	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1578	sent := adapter.sentMessages()
1579	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "needle attach") \|\| !strings.Contains(sent[0].Text, "s1") {
1580	t.Fatalf("/sessions sent = %#v, want indexed session", sent)
1581	}
1582
1583	msg.Text = "/attach session s1"
1584	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1585	profile := gw.sessionProfileForMessage(msg)
1586	if canonicalBotPath(profile.sessionPath) != canonicalBotPath(sessionPath) {
1587	t.Fatalf("attached session path = %q, want %q", profile.sessionPath, sessionPath)
1588	}
1589	if canonicalBotPath(profile.workspaceRoot) != canonicalBotPath(projectRoot) {
1590	t.Fatalf("attached workspace root = %q, want %q", profile.workspaceRoot, projectRoot)
1591	}
1592	}
1593
1594	func TestGatewayRuntimeOverridePreservesControllersWithActiveWork(t *testing.T) {
1595	tests := []struct {
1596	name string
1597	status control.RuntimeStatus
1598	admittedTurn bool
1599	}{
1600	{name: "foreground turn", status: control.RuntimeStatus{Running: true}},
1601	{name: "pending prompt", status: control.RuntimeStatus{PendingPrompt: true}},
1602	{name: "background job", status: control.RuntimeStatus{BackgroundJobs: 1}},
1603	{name: "admitted turn", admittedTurn: true},
1604	}
1605	for _, tc := range tests {
1606	t.Run(tc.name, func(t *testing.T) {
1607	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1608	gw := NewGateway(GatewayConfig{}, nil, logger)
1609	msg := InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, ChatID: "chat", UserID: "user"}
1610	key := BuildSessionKey(msg.Session())
1611	ctrl := &runtimeStatusBotController{status: tc.status, workspaceRoot: "/old"}
1612	state := &sessionState{ctrl: ctrl, workspaceRoot: "/old"}
1613	gw.controllers[key] = state
1614	gw.sessionOverrides[key] = sessionRuntimeOverride{channel: ChannelConfig{WorkspaceRoot: "/old"}, label: "project:old"}
1615	if tc.admittedTurn {
1616	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{}); !result.Acquired {
1617	t.Fatalf("admit turn: %+v", result)
1618	}
1619	}
1620
1621	text := gw.handleUseProjectCommand(key, "/use project default")
1622	if !strings.Contains(text, "请先完成或停止") {
1623	t.Fatalf("busy response = %q", text)
1624	}
1625	if ctrl.closed \|\| gw.controllers[key] != state {
1626	t.Fatalf("active controller was replaced or closed: closed=%v installed=%v", ctrl.closed, gw.controllers[key] == state)
1627	}
1628	if override, ok := gw.sessionOverrides[key]; !ok \|\| override.label != "project:old" {
1629	t.Fatalf("active override changed: %+v, present=%v", override, ok)
1630	}
1631	})
1632	}
1633	}
1634
1635	func TestGatewayDefersProfileMismatchWhileBackgroundWorkIsActive(t *testing.T) {
1636	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1637	newRoot := t.TempDir()
1638	gw := NewGateway(GatewayConfig{WorkspaceRoot: newRoot}, nil, logger)
1639	msg := InboundMessage{Platform: PlatformFeishu, ChatType: ChatDM, ChatID: "chat", UserID: "user"}
1640	key := BuildSessionKey(msg.Session())
1641	ctrl := &runtimeStatusBotController{
1642	status: control.RuntimeStatus{BackgroundJobs: 1}, workspaceRoot: t.TempDir(),
1643	}
1644	state := &sessionState{ctrl: ctrl, workspaceRoot: ctrl.workspaceRoot}
1645	gw.controllers[key] = state
1646
1647	got := gw.getOrCreateSession(context.Background(), key, msg)
1648	if got != state \|\| gw.controllers[key] != state \|\| ctrl.closed {
1649	t.Fatalf("profile mismatch canceled active work: gotOld=%v installed=%v closed=%v", got == state, gw.controllers[key] == state, ctrl.closed)
1650	}
1651	if state.workspaceRoot == newRoot {
1652	t.Fatalf("deferred runtime change mutated the live profile to %q", state.workspaceRoot)
1653	}
1654	}
1655
1656	func TestGatewaySearchAllSearchesIndexedProjects(t *testing.T) {
1657	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1658	projectRoot := t.TempDir()
1659	if err := os.WriteFile(filepath.Join(projectRoot, "needle.txt"), []byte("alpha\nunique-cross-project-needle\n"), 0o644); err != nil {
1660	t.Fatal(err)
1661	}
1662	gw := NewGateway(GatewayConfig{WorkspaceRoot: projectRoot}, nil, logger)
1663
1664	text := gw.handleProjectSearchCommand(context.Background(), "/search all unique-cross-project-needle")
1665	if !strings.Contains(text, "needle.txt") \|\| !strings.Contains(text, "unique-cross-project-needle") {
1666	t.Fatalf("search text = %q, want file hit", text)
1667	}
1668	}
1669
1670	func TestSearchBotProjectsFallbackStopsAtLimit(t *testing.T) {
1671	projectRoot := t.TempDir()
1672	if err := os.WriteFile(filepath.Join(projectRoot, "one.txt"), []byte("first fallback needle\n"), 0o644); err != nil {
1673	t.Fatal(err)
1674	}
1675	if err := os.WriteFile(filepath.Join(projectRoot, "two.txt"), []byte("second fallback needle\n"), 0o644); err != nil {
1676	t.Fatal(err)
1677	}
1678	projects := []botProjectEntry{{
1679	ID: "p1",
1680	Name: "project",
1681	Root: projectRoot,
1682	}}
1683
1684	results, err := searchBotProjectsFallback(context.Background(), projects, []string{projectRoot}, "fallback needle", 1)
1685	if err != nil {
1686	t.Fatalf("fallback search: %v", err)
1687	}
1688
1689	if len(results) != 1 {
1690	t.Fatalf("fallback results = %d, want 1", len(results))
1691	}
1692	if results[0].ProjectID != "p1" \|\| !strings.Contains(results[0].Text, "fallback needle") {
1693	t.Fatalf("fallback result = %#v, want project hit", results[0])
1694	}
1695	}
1696
1697	func TestSearchBotProjectsFallbackHonorsContextCancel(t *testing.T) {
1698	projectRoot := t.TempDir()
1699	if err := os.WriteFile(filepath.Join(projectRoot, "needle.txt"), []byte("fallback needle\n"), 0o644); err != nil {
1700	t.Fatal(err)
1701	}
1702	projects := []botProjectEntry{{
1703	ID: "p1",
1704	Name: "project",
1705	Root: projectRoot,
1706	}}
1707	ctx, cancel := context.WithCancel(context.Background())
1708	cancel()
1709
1710	results, err := searchBotProjectsFallback(ctx, projects, []string{projectRoot}, "fallback needle", 1)
1711
1712	if !errors.Is(err, context.Canceled) {
1713	t.Fatalf("fallback error = %v, want context canceled", err)
1714	}
1715	if len(results) != 0 {
1716	t.Fatalf("fallback results = %d, want 0 after canceled context", len(results))
1717	}
1718	}
1719
1720	func TestGatewayAdminRoleRequiredForProjectIndexCommands(t *testing.T) {
1721	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1722	gw := NewGateway(GatewayConfig{
1723	Allowlist: AllowlistConfig{
1724	Enabled: true,
1725	Users: map[Platform][]string{PlatformWeixin: []string{"user"}},
1726	Admins: map[Platform][]string{PlatformWeixin: []string{"admin"}},
1727	},
1728	}, nil, logger)
1729	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1730	msg := InboundMessage{
1731	Platform: PlatformWeixin,
1732	ChatType: ChatDM,
1733	ChatID: "chat",
1734	UserID: "user",
1735	Text: "/projects",
1736	}
1737	key := BuildSessionKey(msg.Session())
1738
1739	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1740
1741	sent := adapter.sentMessages()
1742	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "没有执行此 bot 命令的权限") {
1743	t.Fatalf("sent = %#v, want permission denial", sent)
1744	}
1745	}
1746
1747	func TestGatewayDefaultQueueSteersActiveTurn(t *testing.T) {
1748	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1749	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1750	adapter := &fakeReactionAdapter{fakeAdapter: newFakeAdapter(PlatformFeishu, "fake-feishu")}
1751	msg := InboundMessage{
1752	Platform: PlatformFeishu,
1753	ConnectionID: "feishu-feishu",
1754	ChatType: ChatDM,
1755	ChatID: "chat",
1756	UserID: "user",
1757	Text: "please adjust the current task",
1758	MessageID: "m2",
1759	}
1760	key := BuildSessionKey(msg.Session())
1761	ctrl := &queueTestController{}
1762	gw.controllers[key] = &sessionState{ctrl: ctrl, sink: &sessionEventSink{}}
1763	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeFollowup}); !result.Acquired {
1764	t.Fatalf("failed to mark session active: %+v", result)
1765	}
1766
1767	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu-feishu", Platform: PlatformFeishu, Adapter: adapter}, msg)
1768
1769	if got := ctrl.steered(); len(got) != 1 \|\| got[0] != msg.Text {
1770	t.Fatalf("steers = %#v, want current message", got)
1771	}
1772	sent := adapter.sentMessages()
1773	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "并入当前任务") {
1774	t.Fatalf("sent = %#v, want steer acknowledgement", sent)
1775	}
1776	if pending := gw.sessions.PendingCount(key); pending != 0 {
1777	t.Fatalf("pending = %d, want 0", pending)
1778	}
1779	if cleaned := adapter.cleanupMessages(); len(cleaned) != 1 \|\| cleaned[0] != "m2" {
1780	t.Fatalf("cleanup messages = %#v, want [m2]", cleaned)
1781	}
1782	}
1783
1784	func TestGatewayRejectedSteerFallsBackToFollowupQueue(t *testing.T) {
1785	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1786	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1787	adapter := &fakeReactionAdapter{fakeAdapter: newFakeAdapter(PlatformFeishu, "fake-feishu")}
1788	msg := InboundMessage{
1789	Platform: PlatformFeishu,
1790	ConnectionID: "feishu-feishu",
1791	ChatType: ChatDM,
1792	ChatID: "chat",
1793	UserID: "user",
1794	Text: "run this after the current turn",
1795	MessageID: "m-rejected-steer",
1796	}
1797	key := BuildSessionKey(msg.Session())
1798	ctrl := &queueTestController{rejectSteer: true}
1799	gw.controllers[key] = &sessionState{ctrl: ctrl, sink: &sessionEventSink{}}
1800	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeFollowup}); !result.Acquired {
1801	t.Fatalf("failed to mark session active: %+v", result)
1802	}
1803
1804	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu-feishu", Platform: PlatformFeishu, Adapter: adapter}, msg)
1805
1806	if got := ctrl.steered(); len(got) != 0 {
1807	t.Fatalf("rejected steers = %#v, want none", got)
1808	}
1809	if pending := gw.sessions.PendingCount(key); pending != 1 {
1810	t.Fatalf("pending = %d, want rejected steer preserved as one follow-up", pending)
1811	}
1812	if sent := adapter.sentMessages(); len(sent) != 0 {
1813	t.Fatalf("sent = %#v, rejected steer must not receive an applied acknowledgement", sent)
1814	}
1815	}
1816
1817	func TestGatewayDefaultQueueSteersMediaOnlyActiveTurn(t *testing.T) {
1818	imageServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1819	w.Header().Set("Content-Type", "image/png")
1820	_, _ = w.Write([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1a, '\n'})
1821	}))
1822	defer imageServer.Close()
1823
1824	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1825	gw := NewGateway(GatewayConfig{
1826	Allowlist: AllowlistConfig{AllowAll: true},
1827	WorkspaceRoot: t.TempDir(),
1828	}, nil, logger)
1829	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1830	msg := InboundMessage{
1831	Platform: PlatformFeishu,
1832	ConnectionID: "feishu-feishu",
1833	ChatType: ChatDM,
1834	ChatID: "chat",
1835	UserID: "user",
1836	MediaURLs: []string{imageServer.URL + "/image.png"},
1837	MessageID: "m-media",
1838	}
1839	key := BuildSessionKey(msg.Session())
1840	ctrl := &queueTestController{}
1841	gw.controllers[key] = &sessionState{ctrl: ctrl, sink: &sessionEventSink{}}
1842	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeFollowup}); !result.Acquired {
1843	t.Fatalf("failed to mark session active: %+v", result)
1844	}
1845
1846	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu-feishu", Platform: PlatformFeishu, Adapter: adapter}, msg)
1847
1848	got := ctrl.steered()
1849	if len(got) != 1 \|\| !strings.Contains(got[0], "Attachments:") \|\| !strings.Contains(got[0], "@.reasonix/attachments/") {
1850	t.Fatalf("steers = %#v, want saved attachment reference", got)
1851	}
1852	}
1853
1854	func TestGatewayQueueFollowupKeepsMessagesForLaterTurns(t *testing.T) {
1855	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1856	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1857	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1858	msg := InboundMessage{
1859	Platform: PlatformWeixin,
1860	ConnectionID: "weixin-weixin",
1861	ChatType: ChatDM,
1862	ChatID: "chat",
1863	UserID: "user",
1864	Text: "first followup",
1865	}
1866	key := BuildSessionKey(msg.Session())
1867	ctrl := &queueTestController{}
1868	gw.controllers[key] = &sessionState{ctrl: ctrl, sink: &sessionEventSink{}}
1869	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeFollowup}); !result.Acquired {
1870	t.Fatalf("failed to mark session active: %+v", result)
1871	}
1872	gw.sessions.SetQueueMode(key, QueueModeFollowup)
1873
1874	gw.handleMessage(context.Background(), AdapterBinding{ID: "weixin-weixin", Platform: PlatformWeixin, Adapter: adapter}, msg)
1875	second := msg
1876	second.Text = "second followup"
1877	gw.handleMessage(context.Background(), AdapterBinding{ID: "weixin-weixin", Platform: PlatformWeixin, Adapter: adapter}, second)
1878
1879	if got := ctrl.steered(); len(got) != 0 {
1880	t.Fatalf("steers = %#v, want none in followup mode", got)
1881	}
1882	if pending := gw.sessions.PendingCount(key); pending != 2 {
1883	t.Fatalf("pending = %d, want 2", pending)
1884	}
1885	next := gw.sessions.Release(key)
1886	if next == nil \|\| next.Text != "first followup" {
1887	t.Fatalf("first release = %#v, want first followup", next)
1888	}
1889	next = gw.sessions.Release(key)
1890	if next == nil \|\| next.Text != "second followup" {
1891	t.Fatalf("second release = %#v, want second followup", next)
1892	}
1893	}
1894
1895	func TestGatewayQueueInterruptCancelsAndKeepsNewestMessage(t *testing.T) {
1896	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1897	gw := NewGateway(GatewayConfig{Allowlist: AllowlistConfig{AllowAll: true}}, nil, logger)
1898	adapter := newFakeAdapter(PlatformQQ, "fake-qq")
1899	msg := InboundMessage{
1900	Platform: PlatformQQ,
1901	ConnectionID: "qq",
1902	ChatType: ChatDM,
1903	ChatID: "chat",
1904	UserID: "user",
1905	Text: "newest request",
1906	}
1907	key := BuildSessionKey(msg.Session())
1908	ctrl := &queueTestController{}
1909	gw.controllers[key] = &sessionState{ctrl: ctrl, sink: &sessionEventSink{}}
1910	if result := gw.sessions.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeFollowup}); !result.Acquired {
1911	t.Fatalf("failed to mark session active: %+v", result)
1912	}
1913	gw.sessions.SetQueueMode(key, QueueModeInterrupt)
1914
1915	gw.handleMessage(context.Background(), AdapterBinding{ID: "qq", Platform: PlatformQQ, Adapter: adapter}, msg)
1916
1917	if !ctrl.wasCanceled() {
1918	t.Fatal("controller was not canceled")
1919	}
1920	next := gw.sessions.Release(key)
1921	if next == nil \|\| next.Text != "newest request" {
1922	t.Fatalf("release = %#v, want newest request", next)
1923	}
1924	sent := adapter.sentMessages()
1925	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "稍后处理这条新消息") {
1926	t.Fatalf("sent = %#v, want interrupt acknowledgement", sent)
1927	}
1928	}
1929
1930	func TestGatewayUnknownDMGetsPairingCode(t *testing.T) {
1931	t.Setenv("REASONIX_HOME", t.TempDir())
1932	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1933	gw := NewGateway(GatewayConfig{
1934	PairingEnabled: true,
1935	Allowlist: AllowlistConfig{Enabled: true, Users: map[Platform][]string{PlatformFeishu: nil}},
1936	}, nil, logger)
1937	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1938	msg := InboundMessage{
1939	Platform: PlatformFeishu,
1940	ConnectionID: "feishu-feishu",
1941	ChatType: ChatDM,
1942	ChatID: "chat",
1943	UserID: "user",
1944	Text: "hello",
1945	}
1946
1947	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu-feishu", Platform: PlatformFeishu, Adapter: adapter}, msg)
1948
1949	sent := adapter.sentMessages()
1950	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "配对码") \|\| !strings.Contains(sent[0].Text, "reasonix bot pairing approve") {
1951	t.Fatalf("sent = %#v, want pairing instructions", sent)
1952	}
1953	reqs, err := ListPairingRequests()
1954	if err != nil {
1955	t.Fatalf("list pairing: %v", err)
1956	}
1957	if len(reqs) != 1 \|\| reqs[0].UserID != "user" \|\| reqs[0].ChatID != "chat" {
1958	t.Fatalf("pairing requests = %+v, want one request for user/chat", reqs)
1959	}
1960	}
1961
1962	func TestGatewayAdminRoleRequiredForYoloWhenConfigured(t *testing.T) {
1963	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1964	gw := NewGateway(GatewayConfig{
1965	Allowlist: AllowlistConfig{
1966	Enabled: true,
1967	Users: map[Platform][]string{PlatformWeixin: []string{"user"}},
1968	Admins: map[Platform][]string{PlatformWeixin: []string{"admin"}},
1969	},
1970	}, nil, logger)
1971	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
1972	msg := InboundMessage{
1973	Platform: PlatformWeixin,
1974	ChatType: ChatDM,
1975	ChatID: "chat",
1976	UserID: "user",
1977	Text: "/yolo on",
1978	}
1979	key := BuildSessionKey(msg.Session())
1980
1981	gw.handleSlashCommand(context.Background(), adapter, key, msg)
1982
1983	sent := adapter.sentMessages()
1984	if len(sent) != 1 \|\| !strings.Contains(sent[0].Text, "没有执行此 bot 命令的权限") {
1985	t.Fatalf("sent = %#v, want permission denial", sent)
1986	}
1987	}
1988
1989	func TestGatewayIgnoresOutboundEchoMessageID(t *testing.T) {
1990	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
1991	gw := NewGateway(GatewayConfig{
1992	IgnoreSelfMessages: true,
1993	Allowlist: AllowlistConfig{AllowAll: true},
1994	}, nil, logger)
1995	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
1996	msg := InboundMessage{
1997	Platform: PlatformFeishu,
1998	ConnectionID: "feishu-feishu",
1999	ChatType: ChatDM,
2000	ChatID: "chat",
2001	UserID: "user",
2002	MessageID: "incoming",
2003	Text: "/status",
2004	}
2005	if err := gw.sendText(context.Background(), adapter, msg, "reply"); err != nil {
2006	t.Fatalf("sendText: %v", err)
2007	}
2008	echo := msg
2009	echo.MessageID = "fake_msg_1"
2010
2011	gw.handleMessage(context.Background(), AdapterBinding{ID: "feishu-feishu", Platform: PlatformFeishu, Adapter: adapter}, echo)
2012
2013	if sent := adapter.sentMessages(); len(sent) != 1 {
2014	t.Fatalf("sent count = %d, want only original outbound echo registration", len(sent))
2015	}
2016	}
2017
2018	func TestGatewayIgnoresConfiguredSelfUserID(t *testing.T) {
2019	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2020	gw := NewGateway(GatewayConfig{
2021	IgnoreSelfMessages: true,
2022	SelfUserIDs: map[Platform][]string{PlatformWeixin: []string{"bot-user"}},
2023	Allowlist: AllowlistConfig{AllowAll: true},
2024	}, nil, logger)
2025	adapter := newFakeAdapter(PlatformWeixin, "fake-weixin")
2026	msg := InboundMessage{
2027	Platform: PlatformWeixin,
2028	ConnectionID: "weixin-weixin",
2029	ChatType: ChatDM,
2030	ChatID: "chat",
2031	UserID: "bot-user",
2032	MessageID: "self-message",
2033	Text: "/status",
2034	}
2035
2036	gw.handleMessage(context.Background(), AdapterBinding{ID: "weixin-weixin", Platform: PlatformWeixin, Adapter: adapter}, msg)
2037
2038	if sent := adapter.sentMessages(); len(sent) != 0 {
2039	t.Fatalf("sent count = %d, want self message ignored", len(sent))
2040	}
2041	}
2042
2043	func TestGatewayAdapterHealthTracksSend(t *testing.T) {
2044	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2045	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
2046	gw := NewGatewayWithAdapterBindings(GatewayConfig{
2047	Enabled: map[Platform]bool{PlatformFeishu: true},
2048	Allowlist: AllowlistConfig{AllowAll: true},
2049	}, []AdapterBinding{{ID: "feishu-lark", Platform: PlatformFeishu, Domain: "lark", Adapter: adapter}}, logger)
2050	if err := gw.Start(context.Background()); err != nil {
2051	t.Fatalf("Start: %v", err)
2052	}
2053	defer gw.Stop()
2054
2055	if _, err := gw.SendToAdapter(context.Background(), "feishu-lark", "lark", OutboundMessage{ChatID: "chat", ChatType: ChatDM, Text: "hello"}); err != nil {
2056	t.Fatalf("SendToAdapter: %v", err)
2057	}
2058
2059	health := gw.AdapterHealth()
2060	if len(health) != 1 {
2061	t.Fatalf("health count = %d, want 1", len(health))
2062	}
2063	if health[0].ID != "feishu-lark" \|\| health[0].Status != "running" \|\| health[0].Sends != 1 \|\| health[0].SendErrors != 0 {
2064	t.Fatalf("health = %+v, want running send count", health[0])
2065	}
2066	}
2067
2068	func TestGatewayControlServerStatusAndSend(t *testing.T) {
2069	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2070	adapter := newFakeAdapter(PlatformFeishu, "fake-feishu")
2071	gw := NewGatewayWithAdapterBindings(GatewayConfig{
2072	Enabled: map[Platform]bool{PlatformFeishu: true},
2073	Allowlist: AllowlistConfig{AllowAll: true},
2074	ControlEnabled: true,
2075	ControlAddr: "127.0.0.1:0",
2076	ControlToken: "secret",
2077	}, []AdapterBinding{{ID: "feishu-lark", Platform: PlatformFeishu, Domain: "lark", Adapter: adapter}}, logger)
2078	if err := gw.Start(context.Background()); err != nil {
2079	t.Fatalf("Start: %v", err)
2080	}
2081	defer gw.Stop()
2082
2083	statusURL := "http://" + gw.ControlAddr() + "/status"
2084	resp, err := http.Get(statusURL)
2085	if err != nil {
2086	t.Fatalf("GET /status without token: %v", err)
2087	}
2088	_ = resp.Body.Close()
2089	if resp.StatusCode != http.StatusUnauthorized {
2090	t.Fatalf("GET /status without token status = %d, want 401", resp.StatusCode)
2091	}
2092
2093	req, err := http.NewRequest(http.MethodGet, statusURL, nil)
2094	if err != nil {
2095	t.Fatalf("new status request: %v", err)
2096	}
2097	req.Header.Set("Authorization", "Bearer secret")
2098	resp, err = http.DefaultClient.Do(req)
2099	if err != nil {
2100	t.Fatalf("GET /status: %v", err)
2101	}
2102	defer resp.Body.Close()
2103	if resp.StatusCode != http.StatusOK {
2104	t.Fatalf("GET /status status = %d, want 200", resp.StatusCode)
2105	}
2106	var status controlStatusResponse
2107	if err := json.NewDecoder(resp.Body).Decode(&status); err != nil {
2108	t.Fatalf("decode status: %v", err)
2109	}
2110	if status.Status != "running" \|\| len(status.Adapters) != 1 \|\| status.Adapters[0].ID != "feishu-lark" {
2111	t.Fatalf("status = %+v, want running feishu-lark", status)
2112	}
2113
2114	req, err = http.NewRequest(http.MethodPost, "http://"+gw.ControlAddr()+"/send", strings.NewReader(`{"connection_id":"feishu-lark","domain":"lark","chat_id":"chat","chat_type":"dm","text":"hello"}`))
2115	if err != nil {
2116	t.Fatalf("new send request: %v", err)
2117	}
2118	req.Header.Set("Authorization", "Bearer secret")
2119	req.Header.Set("Content-Type", "application/json")
2120	resp, err = http.DefaultClient.Do(req)
2121	if err != nil {
2122	t.Fatalf("POST /send: %v", err)
2123	}
2124	_ = resp.Body.Close()
2125	if resp.StatusCode != http.StatusOK {
2126	t.Fatalf("POST /send status = %d, want 200", resp.StatusCode)
2127	}
2128	if sent := adapter.sentMessages(); len(sent) != 1 \|\| sent[0].Text != "hello" {
2129	t.Fatalf("sent = %+v, want hello", sent)
2130	}
2131	if health := gw.AdapterHealth(); len(health) != 1 \|\| health[0].Sends != 1 {
2132	t.Fatalf("health = %+v, want one send", health)
2133	}
2134
2135	req, err = http.NewRequest(http.MethodGet, "http://"+gw.ControlAddr()+"/metrics", nil)
2136	if err != nil {
2137	t.Fatalf("new metrics request: %v", err)
2138	}
2139	req.Header.Set("Authorization", "Bearer secret")
2140	resp, err = http.DefaultClient.Do(req)
2141	if err != nil {
2142	t.Fatalf("GET /metrics: %v", err)
2143	}
2144	metricsBody, _ := io.ReadAll(resp.Body)
2145	_ = resp.Body.Close()
2146	if resp.StatusCode != http.StatusOK \|\| !strings.Contains(string(metricsBody), "reasonix_bot_adapter_sends_total") {
2147	t.Fatalf("GET /metrics status=%d body=%q, want adapter metrics", resp.StatusCode, string(metricsBody))
2148	}
2149	}
2150
2151	func TestControlSendReportsAndTracksPartialDelivery(t *testing.T) {
2152	adapter := &resultAdapter{
2153	fakeAdapter: newFakeAdapter(PlatformFeishu, "partial-feishu"),
2154	result: SendResult{
2155	MessageID: "media-2",
2156	MessageIDs: []string{"text-1", "media-1", "media-2"},
2157	},
2158	err: errors.New("media-3 failed"),
2159	}
2160	gw := NewGatewayWithAdapterBindings(GatewayConfig{
2161	IgnoreSelfMessages: true,
2162	}, []AdapterBinding{{ID: "feishu-lark", Platform: PlatformFeishu, Domain: "lark", Adapter: adapter}}, discardLogger())
2163	req := httptest.NewRequest(http.MethodPost, "/send", strings.NewReader(`{"connection_id":"feishu-lark","domain":"lark","chat_id":"chat","text":"hello"}`))
2164	recorder := httptest.NewRecorder()
2165
2166	gw.handleControlSend(recorder, req)
2167
2168	if recorder.Code != http.StatusMultiStatus {
2169	t.Fatalf("POST /send status = %d, want %d", recorder.Code, http.StatusMultiStatus)
2170	}
2171	var response controlSendResponse
2172	if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil {
2173	t.Fatalf("decode partial response: %v", err)
2174	}
2175	if !response.Partial \|\| response.Error != "media-3 failed" \|\| len(response.MessageIDs) != 3 {
2176	t.Fatalf("partial response = %+v", response)
2177	}
2178	for _, messageID := range response.MessageIDs {
2179	if !gw.isSelfMessage(InboundMessage{
2180	Platform: PlatformFeishu,
2181	ConnectionID: "feishu-lark",
2182	Domain: "lark",
2183	ChatID: "chat",
2184	MessageID: messageID,
2185	}) {
2186	t.Fatalf("delivered message %q was not registered for echo suppression", messageID)
2187	}
2188	}
2189	}
2190
2191	func TestGatewayAddsPendingReactionWhenAdapterSupportsIt(t *testing.T) {
2192	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2193	gw := NewGateway(GatewayConfig{}, nil, logger)
2194	fa := &fakeReactionAdapter{fakeAdapter: newFakeAdapter(PlatformFeishu, "fake-feishu")}
2195
2196	gw.addPendingReaction(context.Background(), PlatformFeishu, fa, InboundMessage{MessageID: "om_123"})
2197
2198	if len(fa.reactions) != 1 \|\| fa.reactions[0] != "om_123" {
2199	t.Fatalf("reactions = %#v, want [om_123]", fa.reactions)
2200	}
2201	}
2202
2203	func TestGatewayStoresQueuedReactionCleanupBeforeControllerExists(t *testing.T) {
2204	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2205	gw := NewGateway(GatewayConfig{}, nil, logger)
2206	fa := &fakeReactionAdapter{fakeAdapter: newFakeAdapter(PlatformFeishu, "fake-feishu")}
2207	first := InboundMessage{
2208	Platform: PlatformFeishu,
2209	ConnectionID: "feishu-feishu",
2210	Domain: "feishu",
2211	ChatType: ChatDM,
2212	ChatID: "chat",
2213	UserID: "user",
2214	Text: "first",
2215	MessageID: "om_first",
2216	}
2217	key := BuildSessionKey(first.Session())
2218	if acquired, merged := gw.sessions.TryAcquire(key, first); !acquired \|\| merged {
2219	t.Fatalf("first TryAcquire = (%v, %v), want acquired without merge", acquired, merged)
2220	}
2221
2222	queued := first
2223	queued.Text = "queued"
2224	queued.MessageID = "om_queued"
2225	cleanup := gw.addPendingReaction(context.Background(), PlatformFeishu, fa, queued)
2226	if acquired, merged := gw.sessions.TryAcquire(key, queued); acquired \|\| !merged {
2227	t.Fatalf("queued TryAcquire = (%v, %v), want merged while active", acquired, merged)
2228	}
2229	gw.storeReactionCleanup(key, cleanup)
2230	if _, ok := gw.controllers[key]; ok {
2231	t.Fatal("test setup expected no controller state yet")
2232	}
2233	if cleaned := fa.cleanupMessages(); len(cleaned) != 0 {
2234	t.Fatalf("cleanup messages before flush = %#v, want none", cleaned)
2235	}
2236
2237	gw.flushReactionCleanups(key, nil)
2238	cleaned := fa.cleanupMessages()
2239	if len(cleaned) != 1 \|\| cleaned[0] != "om_queued" {
2240	t.Fatalf("cleanup messages = %#v, want [om_queued]", cleaned)
2241	}
2242	}
2243
2244	func TestGatewaySessionOptionsUseChannelOverride(t *testing.T) {
2245	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2246	gw := NewGateway(GatewayConfig{
2247	Model: "global-model",
2248	WorkspaceRoot: "/global",
2249	Channels: map[Platform]ChannelConfig{
2250	PlatformFeishu: {Model: "feishu-model", WorkspaceRoot: "/feishu"},
2251	PlatformWeixin: {WorkspaceRoot: "/weixin"},
2252	},
2253	}, nil, logger)
2254
2255	model, root, mode := gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformFeishu})
2256	if model != "feishu-model" \|\| root != "/feishu" {
2257	t.Fatalf("feishu options = %q,%q; want channel override", model, root)
2258	}
2259	if mode != "ask" {
2260	t.Fatalf("feishu tool approval mode = %q, want ask", mode)
2261	}
2262
2263	model, root, mode = gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformWeixin})
2264	if model != "global-model" \|\| root != "/weixin" {
2265	t.Fatalf("weixin options = %q,%q; want global model and channel root", model, root)
2266	}
2267	if mode != "ask" {
2268	t.Fatalf("weixin tool approval mode = %q, want ask", mode)
2269	}
2270
2271	model, root, mode = gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformQQ})
2272	if model != "global-model" \|\| root != "/global" {
2273	t.Fatalf("qq options = %q,%q; want global defaults", model, root)
2274	}
2275	if mode != "ask" {
2276	t.Fatalf("qq tool approval mode = %q, want ask", mode)
2277	}
2278	}
2279
2280	func TestGatewaySessionOptionsPreferConnectionOverride(t *testing.T) {
2281	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2282	gw := NewGateway(GatewayConfig{
2283	Model: "global-model",
2284	WorkspaceRoot: "/global",
2285	Channels: map[Platform]ChannelConfig{
2286	PlatformFeishu: {Model: "feishu-model", WorkspaceRoot: "/feishu"},
2287	},
2288	ConnectionChannels: map[string]ChannelConfig{
2289	"feishu-lark": {Model: "lark-model", WorkspaceRoot: "/lark"},
2290	},
2291	}, nil, logger)
2292
2293	model, root, mode := gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformFeishu, ConnectionID: "feishu-lark"})
2294	if model != "lark-model" \|\| root != "/lark" {
2295	t.Fatalf("lark options = %q,%q; want connection override", model, root)
2296	}
2297	if mode != "ask" {
2298	t.Fatalf("lark tool approval mode = %q, want ask", mode)
2299	}
2300	}
2301
2302	func TestGatewaySessionOptionsPreferConnectionSessionMappingWorkspace(t *testing.T) {
2303	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2304	gw := NewGateway(GatewayConfig{
2305	Model: "global-model",
2306	WorkspaceRoot: "/global",
2307	ConnectionChannels: map[string]ChannelConfig{
2308	"weixin-main": {
2309	WorkspaceRoot: "/connection",
2310	SessionMappings: []SessionMapping{
2311	{RemoteID: "group-1", ChatType: string(ChatGroup), UserID: "other", Scope: "project", WorkspaceRoot: "/other"},
2312	{RemoteID: "group-1", ChatType: string(ChatGroup), UserID: "user-1", Scope: "project", WorkspaceRoot: "/mapped"},
2313	},
2314	},
2315	},
2316	}, nil, logger)
2317
2318	model, root, mode := gw.sessionOptionsForMessage(InboundMessage{
2319	Platform: PlatformWeixin,
2320	ConnectionID: "weixin-main",
2321	ChatType: ChatGroup,
2322	ChatID: "group-1",
2323	UserID: "user-1",
2324	})
2325	if model != "global-model" \|\| root != "/mapped" \|\| mode != "ask" {
2326	t.Fatalf("mapped options = %q,%q,%q; want global model, mapped workspace, ask", model, root, mode)
2327	}
2328
2329	_, root, _ = gw.sessionOptionsForMessage(InboundMessage{
2330	Platform: PlatformWeixin,
2331	ConnectionID: "weixin-main",
2332	ChatType: ChatGroup,
2333	ChatID: "group-1",
2334	UserID: "new-user",
2335	})
2336	if root != "/connection" {
2337	t.Fatalf("unmapped group user workspace = %q, want connection default", root)
2338	}
2339	}
2340
2341	func TestGatewaySessionOptionsAllowSessionMappingGlobalWorkspace(t *testing.T) {
2342	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2343	gw := NewGateway(GatewayConfig{
2344	WorkspaceRoot: "/global-default",
2345	ConnectionChannels: map[string]ChannelConfig{
2346	"weixin-main": {
2347	WorkspaceRoot: "/connection",
2348	SessionMappings: []SessionMapping{{RemoteID: "dm-1", Scope: "global"}},
2349	},
2350	},
2351	}, nil, logger)
2352
2353	_, root, _ := gw.sessionOptionsForMessage(InboundMessage{
2354	Platform: PlatformWeixin,
2355	ConnectionID: "weixin-main",
2356	ChatType: ChatDM,
2357	ChatID: "dm-1",
2358	})
2359	if root != "" {
2360	t.Fatalf("global mapping workspace = %q, want empty global workspace", root)
2361	}
2362	}
2363
2364	func TestSessionStateMatchesRuntimeRejectsWorkspaceOrModelMismatch(t *testing.T) {
2365	ctrl := control.New(control.Options{WorkspaceRoot: "/old"})
2366	defer ctrl.Close()
2367	state := &sessionState{ctrl: ctrl, model: "model-a"}
2368
2369	if !sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-a", workspaceRoot: "/old"}) {
2370	t.Fatal("session should match the controller workspace and model")
2371	}
2372	if sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-a", workspaceRoot: "/new"}) {
2373	t.Fatal("session matched a different workspace root")
2374	}
2375	if sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-b", workspaceRoot: "/old"}) {
2376	t.Fatal("session matched a different model")
2377	}
2378	}
2379
2380	func TestSessionStateMatchesRuntimeRejectsAttachedSessionPathMismatch(t *testing.T) {
2381	root := t.TempDir()
2382	pathA := filepath.Join(root, "a.jsonl")
2383	pathB := filepath.Join(root, "b.jsonl")
2384	ctrl := control.New(control.Options{WorkspaceRoot: root, SessionPath: pathA})
2385	defer ctrl.Close()
2386	state := &sessionState{ctrl: ctrl, model: "model-a", workspaceRoot: root, sessionPath: pathA}
2387
2388	if !sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-a", workspaceRoot: root, sessionPath: pathA}) {
2389	t.Fatal("attached session should match the same pinned path")
2390	}
2391	if sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-a", workspaceRoot: root, sessionPath: pathB}) {
2392	t.Fatal("attached session matched a different path")
2393	}
2394	if sessionStateMatchesRuntime(state, sessionRuntimeProfile{model: "model-a", workspaceRoot: root}) {
2395	t.Fatal("attached session matched an unpinned profile")
2396	}
2397	}
2398
2399	func TestGatewaySessionOptionsPreferRemoteRouteOverride(t *testing.T) {
2400	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
2401	gw := NewGateway(GatewayConfig{
2402	Model: "global-model",
2403	WorkspaceRoot: "/global",
2404	ToolApprovalMode: "ask",
2405	ConnectionChannels: map[string]ChannelConfig{
2406	"feishu-lark": {Model: "lark-model", WorkspaceRoot: "/lark", ToolApprovalMode: "auto"},
2407	},
2408	Routes: []RouteConfig{{
2409	ConnectionID: "feishu-lark",
2410	ChatType: ChatGroup,
2411	ChatID: "group-1",
2412	Channel: ChannelConfig{Model: "route-model", WorkspaceRoot: "/route", ToolApprovalMode: "yolo"},
2413	}},
2414	}, nil, logger)
2415
2416	model, root, mode := gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformFeishu, ConnectionID: "feishu-lark", ChatType: ChatGroup, ChatID: "group-1"})
2417	if model != "route-model" \|\| root != "/route" \|\| mode != "yolo" {
2418	t.Fatalf("route options = %q,%q,%q; want route override", model, root, mode)
2419	}
2420	model, root, mode = gw.sessionOptionsForMessage(InboundMessage{Platform: PlatformFeishu, ConnectionID: "feishu-lark", ChatType: ChatGroup, ChatID: "group-2"})
2421	if model != "lark-model" \|\| root != "/lark" \|\| mode != "auto" {
2422	t.Fatalf("non-matching options = %q,%q,%q; want connection override", model, root, mode)
2423	}
2424	}
2425
2426	func TestBotSessionDirUsesProjectWorkspaceRoot(t *testing.T) {
2427	root := t.TempDir()
2428	got := botSessionDir(root)
2429	if got == "" \|\| got == botSessionDir("") {
2430	t.Fatalf("project session dir = %q, want project-specific dir", got)
2431	}
2432	}
2433
2434	// A persisted session_mappings binding must resolve to the mapped session file
2435	// at session-profile time — before this, the binding was display-only and every
2436	// gateway restart opened a fresh session for the chat (#6917, #6934).
2437	func TestSessionProfileConsumesPersistedMapping(t *testing.T) {
2438	dir := t.TempDir()
2439	mapped := filepath.Join(dir, "chat.jsonl")
2440	if err := os.WriteFile(mapped, []byte(`{"role":"user","content":"hi"}`+"\n"), 0o644); err != nil {
2441	t.Fatal(err)
2442	}
2443	gw := &BotGateway{
2444	cfg: GatewayConfig{
2445	ConnectionChannels: map[string]ChannelConfig{
2446	"conn-1": {SessionMappings: []SessionMapping{{
2447	RemoteID: "chat-42",
2448	SessionID: "path:" + mapped,
2449	}}},
2450	},
2451	},
2452	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
2453	sessionOverrides: map[string]sessionRuntimeOverride{},
2454	}
2455	msg := InboundMessage{ConnectionID: "conn-1", ChatID: "chat-42", ChatType: ChatDM}
2456
2457	profile := gw.sessionProfileForMessage(msg)
2458	if canonicalBotPath(profile.sessionPath) != canonicalBotPath(mapped) {
2459	t.Fatalf("profile.sessionPath = %q, want mapped %q", profile.sessionPath, mapped)
2460	}
2461	if !profile.sessionPathOptional {
2462	t.Fatal("mapping-derived path must be optional (degradable), not an attach-style hard binding")
2463	}
2464
2465	// A missing target quietly degrades to normal creation.
2466	if err := os.Remove(mapped); err != nil {
2467	t.Fatal(err)
2468	}
2469	profile = gw.sessionProfileForMessage(msg)
2470	if profile.sessionPath != "" {
2471	t.Fatalf("missing mapped file should resolve to empty path, got %q", profile.sessionPath)
2472	}
2473
2474	// An /attach override outranks the mapping.
2475	gw.sessionOverrides[BuildSessionKey(msg.Session())] = sessionRuntimeOverride{sessionPath: filepath.Join(dir, "attached.jsonl")}
2476	profile = gw.sessionProfileForMessage(msg)
2477	if profile.sessionPathOptional {
2478	t.Fatal("attach override must stay a hard binding")
2479	}
2480	}
2481
2482	// A state that degraded off its unavailable mapped session must not be torn
2483	// down by the next message re-resolving the mapping — that would spawn a new
2484	// session file per message.
2485	func TestDegradedMappingStateStaysStable(t *testing.T) {
2486	s := &sessionState{mappingDegraded: true, sessionPath: "", ctrl: stubPathController{}}
2487	profile := sessionRuntimeProfile{sessionPath: "/some/mapped.jsonl", sessionPathOptional: true}
2488	if !sessionStateMatchesRuntime(s, profile) {
2489	t.Fatal("degraded state must keep matching an optional mapped profile")
2490	}
2491	hard := sessionRuntimeProfile{sessionPath: "/some/mapped.jsonl"}
2492	if sessionStateMatchesRuntime(s, hard) {
2493	t.Fatal("an explicit attach must still force a rebuild")
2494	}
2495	}
2496
2497	type stubPathController struct{ botController }
2498
2499	func (stubPathController) SessionPath() string { return "" }
2500	func (stubPathController) WorkspaceRoot() string { return "" }
2501

2501 lines GO