返回 DeepSeek-Reasonix
bot_runtime_app.go
根目录 / desktop / bot_runtime_app.go
1 package main
2
3 import (
4 "context"
5 "fmt"
6 "log/slog"
7 "os"
8 "strings"
9 "sync"
10 "time"
11
12 "reasonix/internal/bot"
13 "reasonix/internal/botruntime"
14 "reasonix/internal/config"
15 )
16
17 type BotRuntimeStatusView struct {
18 Running bool `json:"running"`
19 Status string `json:"status"`
20 Message string `json:"message"`
21 Connections int `json:"connections"`
22 StartedAt string `json:"startedAt"`
23 Platforms map[string]string `json:"platforms,omitempty"`
24 }
25
26 type desktopBotRuntime struct {
27 // lifecycleMu serializes start/stop transitions so two apply/stop calls
28 // can't race a gateway into existence. The slow work (gw.Stop teardown,
29 // gw.Start dials) runs while holding it but NOT r.mu, so status/send reads
30 // never block on a restart.
31 lifecycleMu sync.Mutex
32 mu sync.Mutex
33 cancel context.CancelFunc
34 gw *bot.BotGateway
35 status BotRuntimeStatusView
36 }
37
38 func newDesktopBotRuntime() *desktopBotRuntime {
39 return &desktopBotRuntime{status: BotRuntimeStatusView{Status: "stopped", Message: "bot runtime is not started"}}
40 }
41
42 func desktopBotChannelsWithLegacyQQ(qq config.QQBotConfig, channels map[bot.Platform]bot.ChannelConfig, connectionChannels map[string]bot.ChannelConfig) (map[bot.Platform]bot.ChannelConfig, map[string]bot.ChannelConfig) {
43 channel := bot.ChannelConfig{
44 Model: strings.TrimSpace(qq.Model),
45 ToolApprovalMode: normalizeBotConnectionToolApprovalMode(qq.ToolApprovalMode),
46 WorkspaceRoot: strings.TrimSpace(qq.WorkspaceRoot),
47 }
48 if channel.Model == "" && channel.ToolApprovalMode == "" && channel.WorkspaceRoot == "" {
49 return channels, connectionChannels
50 }
51 if channels == nil {
52 channels = make(map[bot.Platform]bot.ChannelConfig)
53 }
54 if _, ok := channels[bot.PlatformQQ]; !ok {
55 channels[bot.PlatformQQ] = channel
56 }
57 if connectionChannels == nil {
58 connectionChannels = make(map[string]bot.ChannelConfig)
59 }
60 if _, ok := connectionChannels[string(bot.PlatformQQ)]; !ok {
61 connectionChannels[string(bot.PlatformQQ)] = channel
62 }
63 return channels, connectionChannels
64 }
65
66 // desktopBotChannelsWithLegacyDingtalk 把 legacy [bot.dingtalk] 的模型/权限/
67 // 工作目录合成进 Channels 与 ConnectionChannels,使直配(无 connection)的
68 // 钉钉 bot 也能从设置面板配置这些运行选项(与 legacy QQ 同路径)。
69 func desktopBotChannelsWithLegacyDingtalk(dt config.DingtalkBotConfig, channels map[bot.Platform]bot.ChannelConfig, connectionChannels map[string]bot.ChannelConfig) (map[bot.Platform]bot.ChannelConfig, map[string]bot.ChannelConfig) {
70 channel := bot.ChannelConfig{
71 Model: strings.TrimSpace(dt.Model),
72 ToolApprovalMode: normalizeBotConnectionToolApprovalMode(dt.ToolApprovalMode),
73 WorkspaceRoot: strings.TrimSpace(dt.WorkspaceRoot),
74 SessionMappings: botruntime.SessionMappings(dt.SessionMappings),
75 }
76 if channel.Model == "" && channel.ToolApprovalMode == "" && channel.WorkspaceRoot == "" && len(channel.SessionMappings) == 0 {
77 return channels, connectionChannels
78 }
79 if channels == nil {
80 channels = make(map[bot.Platform]bot.ChannelConfig)
81 }
82 if _, ok := channels[bot.PlatformDingtalk]; !ok {
83 channels[bot.PlatformDingtalk] = channel
84 }
85 if connectionChannels == nil {
86 connectionChannels = make(map[string]bot.ChannelConfig)
87 }
88 if _, ok := connectionChannels[string(bot.PlatformDingtalk)]; !ok {
89 connectionChannels[string(bot.PlatformDingtalk)] = channel
90 }
91 return channels, connectionChannels
92 }
93
94 func (a *App) refreshBotRuntimeAsync() {
95 if a.ctx == nil {
96 return
97 }
98 a.goSafe("refreshBotRuntime", a.refreshBotRuntime)
99 }
100
101 func (a *App) refreshBotRuntime() {
102 // NewApp always pre-fills botRuntime; a nil here means a test-constructed
103 // App with no bot runtime, which must not lazily create one from a
104 // background goroutine (that would race a concurrent refresh).
105 if a.botRuntime == nil {
106 return
107 }
108 var watcherVersion uint64
109 if a.botBridge != nil {
110 watcherVersion = a.botBridge.watcherVersion()
111 }
112 cfg, err := a.loadDesktopBotConfig()
113 if err != nil {
114 a.botRuntime.stop("error", err.Error())
115 return
116 }
117 // Assign through a typed local so a nil *botBridgeHub never becomes a
118 // non-nil bot.DesktopBridge interface inside the gateway config.
119 var bridge bot.DesktopBridge
120 if a.botBridge != nil {
121 // 配置是订阅的持久化事实源:每次运行时重算前重新种子,桌面重启后
122 // /desktop watch 的订阅继续生效。
123 a.botBridge.seedWatchers(bridgeRoutesFromConfig(cfg.Bot.DesktopWatchers), watcherVersion)
124 bridge = a.botBridge
125 }
126 _ = a.botRuntime.apply(a.bootContext(), cfg, globalTabWorkspaceRoot(), a.persistRemoteBotToolApprovalMode, bridge)
127 }
128
129 func (a *App) loadDesktopBotConfig() (*config.Config, error) {
130 // Read-only load feeding the bot runtime and connection diagnostics. It
131 // must load credentials: the runtime resolves app secrets and control
132 // tokens from the process env (AppSecretEnv, Control.TokenEnv), which the
133 // credential-free view load would leave unset on a fresh process.
134 cfg, _, err := a.loadDesktopUserConfigForViewWithCredentials()
135 if err != nil {
136 return nil, err
137 }
138 return cfg, nil
139 }
140
141 func (a *App) stopBotRuntime() {
142 if a.botRuntime != nil {
143 a.botRuntime.stop("stopped", "bot runtime stopped")
144 }
145 }
146
147 func (a *App) BotRuntimeStatus() BotRuntimeStatusView {
148 if a.botRuntime == nil {
149 return BotRuntimeStatusView{Status: "stopped", Message: "bot runtime is not started"}
150 }
151 return a.botRuntime.snapshot()
152 }
153
154 func (r *desktopBotRuntime) apply(parent context.Context, cfg *config.Config, workspaceRoot string, onToolApprovalModeChange func(bot.InboundMessage, string) error, bridge bot.DesktopBridge) error {
155 if r == nil {
156 return nil
157 }
158 if parent == nil {
159 parent = context.Background()
160 }
161 plan := desktopBotRuntimePlan(cfg)
162 r.lifecycleMu.Lock()
163 defer r.lifecycleMu.Unlock()
164 r.stopCurrent()
165 if !plan.Start {
166 r.setStatus(BotRuntimeStatusView{Status: plan.Status, Message: plan.Message})
167 return nil
168 }
169
170 logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
171 ctx, cancel := context.WithCancel(parent)
172 modelName := botruntime.ModelName(cfg, "")
173 channels := botruntime.ChannelConfigs(cfg.Bot.Connections, true, true)
174 connectionChannels := botruntime.ConnectionChannelConfigs(cfg.Bot.Connections, true, true)
175 channels, connectionChannels = desktopBotChannelsWithLegacyQQ(cfg.Bot.QQ, channels, connectionChannels)
176 channels, connectionChannels = desktopBotChannelsWithLegacyDingtalk(cfg.Bot.Dingtalk, channels, connectionChannels)
177 gwCfg := bot.GatewayConfig{
178 Model: modelName,
179 ToolApprovalMode: cfg.Bot.ToolApprovalMode,
180 MaxSteps: cfg.Bot.MaxSteps,
181 QueueMode: cfg.Bot.QueueMode,
182 QueueCap: cfg.Bot.QueueCap,
183 QueueDrop: cfg.Bot.QueueDrop,
184 PairingEnabled: cfg.Bot.Pairing.Enabled,
185 PairingTTL: time.Duration(cfg.Bot.Pairing.RequestTTLMinutes) * time.Minute,
186 PairingMaxPending: cfg.Bot.Pairing.MaxPendingPerPlatform,
187 IgnoreSelfMessages: cfg.Bot.IgnoreSelfMessages,
188 SelfUserIDs: map[bot.Platform][]string{
189 bot.PlatformQQ: cfg.Bot.SelfUserIDs.QQ,
190 bot.PlatformFeishu: cfg.Bot.SelfUserIDs.Feishu,
191 bot.PlatformWeixin: cfg.Bot.SelfUserIDs.Weixin,
192 bot.PlatformDingtalk: cfg.Bot.SelfUserIDs.Dingtalk,
193 },
194 ControlEnabled: cfg.Bot.Control.Enabled,
195 ControlAddr: cfg.Bot.Control.Addr,
196 ControlToken: os.Getenv(strings.TrimSpace(cfg.Bot.Control.TokenEnv)),
197 WorkspaceRoot: workspaceRoot,
198 Channels: channels,
199 ConnectionChannels: connectionChannels,
200 Routes: botruntime.RouteConfigs(cfg.Bot.Routes, true, true),
201 ConnectionAccess: botruntime.ConnectionAccessConfigs(cfg),
202 Enabled: plan.Enabled,
203 Allowlist: bot.AllowlistConfig{
204 Enabled: cfg.Bot.Allowlist.Enabled,
205 AllowAll: cfg.Bot.Allowlist.AllowAll,
206 Users: map[bot.Platform][]string{
207 bot.PlatformQQ: cfg.Bot.Allowlist.QQUsers,
208 bot.PlatformFeishu: cfg.Bot.Allowlist.FeishuUsers,
209 bot.PlatformWeixin: cfg.Bot.Allowlist.WeixinUsers,
210 bot.PlatformDingtalk: cfg.Bot.Allowlist.DingtalkUsers,
211 },
212 Approvers: map[bot.Platform][]string{
213 bot.PlatformQQ: cfg.Bot.Allowlist.QQApprovers,
214 bot.PlatformFeishu: cfg.Bot.Allowlist.FeishuApprovers,
215 bot.PlatformWeixin: cfg.Bot.Allowlist.WeixinApprovers,
216 bot.PlatformDingtalk: cfg.Bot.Allowlist.DingtalkApprovers,
217 },
218 Admins: map[bot.Platform][]string{
219 bot.PlatformQQ: cfg.Bot.Allowlist.QQAdmins,
220 bot.PlatformFeishu: cfg.Bot.Allowlist.FeishuAdmins,
221 bot.PlatformWeixin: cfg.Bot.Allowlist.WeixinAdmins,
222 bot.PlatformDingtalk: cfg.Bot.Allowlist.DingtalkAdmins,
223 },
224 Groups: map[bot.Platform][]string{
225 bot.PlatformQQ: cfg.Bot.Allowlist.QQGroups,
226 bot.PlatformFeishu: cfg.Bot.Allowlist.FeishuGroups,
227 bot.PlatformWeixin: cfg.Bot.Allowlist.WeixinGroups,
228 bot.PlatformDingtalk: cfg.Bot.Allowlist.DingtalkGroups,
229 },
230 },
231 Debounce: time.Duration(cfg.Bot.DebounceMs) * time.Millisecond,
232 ModelResolver: botruntime.ModelResolver(cfg),
233 OnInbound: botruntime.NewRemoteRememberer(logger),
234 OnSessionReady: botruntime.NewSessionRemembererWithWorkspace(logger, workspaceRoot),
235 OnToolApprovalModeChange: onToolApprovalModeChange,
236 Desktop: bridge,
237 }
238 bindings := botruntime.AdapterBindings(cfg, plan.Enabled, nil, logger)
239 if len(bindings) == 0 {
240 cancel()
241 r.setStatus(BotRuntimeStatusView{Status: "stopped", Message: "no bot adapters configured"})
242 return nil
243 }
244 gw := bot.NewGatewayWithAdapterBindings(gwCfg, bindings, logger)
245 if err := gw.Start(ctx); err != nil {
246 cancel()
247 gw.Stop()
248 r.setStatus(BotRuntimeStatusView{Status: "error", Message: err.Error(), Connections: gw.AdapterCount()})
249 return err
250 }
251 runningConnections := gw.AdapterCount()
252 startErrors := gw.StartErrors()
253 status := "running"
254 message := fmt.Sprintf("%d bot connection(s) running", runningConnections)
255 if len(startErrors) > 0 {
256 status = "degraded"
257 message = fmt.Sprintf("%d bot connection(s) running; %d failed to start: %s", runningConnections, len(startErrors), summarizeBotRuntimeErrors(startErrors))
258 }
259 r.mu.Lock()
260 r.cancel = cancel
261 r.gw = gw
262 r.status = BotRuntimeStatusView{
263 Running: true,
264 Status: status,
265 Message: message,
266 Connections: runningConnections,
267 StartedAt: time.Now().UTC().Format(time.RFC3339),
268 }
269 r.mu.Unlock()
270 return nil
271 }
272
273 func (a *App) persistRemoteBotToolApprovalMode(msg bot.InboundMessage, mode string) error {
274 mode = normalizeBotConnectionToolApprovalMode(mode)
275 if mode == "" {
276 return nil
277 }
278 return a.applyConfigOnly(func(c *config.Config) error {
279 id := strings.TrimSpace(msg.ConnectionID)
280 now := time.Now().UTC().Format(time.RFC3339)
281 if id != "" {
282 for i := range c.Bot.Connections {
283 if c.Bot.Connections[i].ID == id || botruntime.ConnectionRuntimeID(c.Bot.Connections[i]) == id {
284 c.Bot.Connections[i].ToolApprovalMode = mode
285 c.Bot.Connections[i].UpdatedAt = now
286 return nil
287 }
288 }
289 }
290 c.Bot.ToolApprovalMode = mode
291 return nil
292 })
293 }
294
295 func summarizeBotRuntimeErrors(errs []error) string {
296 parts := make([]string, 0, len(errs))
297 for _, err := range errs {
298 if err == nil {
299 continue
300 }
301 parts = append(parts, err.Error())
302 }
303 if len(parts) == 0 {
304 return ""
305 }
306 if len(parts) > 3 {
307 hidden := len(parts) - 3
308 parts = append(parts[:3], fmt.Sprintf("%d more", hidden))
309 }
310 return strings.Join(parts, "; ")
311 }
312
313 type botRuntimePlan struct {
314 Start bool
315 Status string
316 Message string
317 Enabled map[bot.Platform]bool
318 }
319
320 func desktopBotRuntimePlan(cfg *config.Config) botRuntimePlan {
321 if cfg == nil {
322 return botRuntimePlan{Status: "error", Message: "config is unavailable"}
323 }
324 if !cfg.Bot.Enabled {
325 return botRuntimePlan{Status: "stopped", Message: "bot is disabled"}
326 }
327 if !botruntime.BotConfigHasAccessControl(cfg.Bot) {
328 return botRuntimePlan{Status: "blocked", Message: "bot requires an allowlist, pairing, per-bot access, or allow_all=true"}
329 }
330 enabled, unknown := botruntime.EnabledPlatforms(cfg, nil)
331 if len(unknown) > 0 {
332 return botRuntimePlan{Status: "error", Message: "unknown bot channel: " + strings.Join(unknown, ", ")}
333 }
334 if !botruntime.HasEnabledPlatform(enabled) {
335 return botRuntimePlan{Status: "stopped", Message: "no bot channels enabled"}
336 }
337 return botRuntimePlan{Start: true, Status: "running", Message: "bot runtime can start", Enabled: enabled}
338 }
339
340 func (r *desktopBotRuntime) stop(status, message string) {
341 r.lifecycleMu.Lock()
342 defer r.lifecycleMu.Unlock()
343 r.stopCurrent()
344 r.setStatus(BotRuntimeStatusView{Status: status, Message: message})
345 }
346
347 // stopCurrent detaches the running gateway under r.mu, then tears it down
348 // off-lock: gw.Stop() closes every session controller (up to the jobs teardown
349 // grace each) and must not stall status/send readers. Callers hold lifecycleMu.
350 func (r *desktopBotRuntime) stopCurrent() {
351 r.mu.Lock()
352 cancel := r.cancel
353 gw := r.gw
354 r.cancel = nil
355 r.gw = nil
356 r.mu.Unlock()
357 if cancel != nil {
358 cancel()
359 }
360 if gw != nil {
361 gw.Stop()
362 }
363 }
364
365 func (r *desktopBotRuntime) setStatus(status BotRuntimeStatusView) {
366 r.mu.Lock()
367 r.status = status
368 r.mu.Unlock()
369 }
370
371 func (r *desktopBotRuntime) snapshot() BotRuntimeStatusView {
372 r.mu.Lock()
373 defer r.mu.Unlock()
374 s := r.status
375 if r.gw != nil {
376 s.Platforms = botAdapterPlatformStatuses(r.gw.AdapterHealth())
377 }
378 return s
379 }
380
381 // botAdapterPlatformStatuses 把 gateway 的适配器健康快照收敛为
382 // platform → status 映射(如 dingtalk → running),供设置面板显示在线状态。
383 func botAdapterPlatformStatuses(health []bot.AdapterHealthSnapshot) map[string]string {
384 out := make(map[string]string, len(health))
385 for _, h := range health {
386 if strings.TrimSpace(string(h.Platform)) != "" {
387 out[string(h.Platform)] = h.Status
388 }
389 }
390 return out
391 }
392
393 // updateConnectionToolApprovalMode updates a connection's tool approval mode
394 // on the running gateway without restarting. Returns true if updated, false if
395 // the gateway is not running or the connection is unknown.
396 func (r *desktopBotRuntime) updateConnectionToolApprovalMode(connID, mode string) bool {
397 r.mu.Lock()
398 defer r.mu.Unlock()
399 if r.gw == nil {
400 return false
401 }
402 mode = normalizeBotConnectionToolApprovalMode(mode)
403 // Update ConnectionChannels in the internal GatewayConfig so new sessions
404 // pick up the mode. Existing sessions are updated by the gateway directly.
405 r.gw.UpdateConnectionToolApprovalMode(connID, mode)
406 return true
407 }
408
409 // SendToAdapter sends a message through the running gateway's adapter
410 // identified by connID. Returns an error if the gateway is not running
411 // or no matching adapter is found.
412 func (r *desktopBotRuntime) SendToAdapter(ctx context.Context, connID, domain string, msg bot.OutboundMessage) (bot.SendResult, error) {
413 r.mu.Lock()
414 gw := r.gw
415 r.mu.Unlock()
416 if gw == nil {
417 return bot.SendResult{}, nil // gateway not running — silent no-op
418 }
419 return gw.SendToAdapter(ctx, connID, domain, msg)
420 }
421
422 // TestSendToAdapter sends a test message through the running gateway's adapter
423 // identified by connID. The adapter must implement bot.TestSender (dingtalk).
424 func (r *desktopBotRuntime) TestSendToAdapter(ctx context.Context, connID, domain, text string) (bot.SendResult, error) {
425 r.mu.Lock()
426 gw := r.gw
427 r.mu.Unlock()
428 if gw == nil {
429 return bot.SendResult{}, fmt.Errorf("bot runtime is not running")
430 }
431 return gw.TestSendToAdapter(ctx, connID, domain, text)
432 }
433
434 // Running returns true if the bot gateway is currently active.
435 func (r *desktopBotRuntime) Running() bool {
436 r.mu.Lock()
437 defer r.mu.Unlock()
438 return r.gw != nil
439 }
440
441 // ForwardTargets returns the list of bot forward targets derived from the
442 // current config's bot connections and their session mappings. Each mapping
443 // produces one target (connID + chatID + chatType) for event forwarding.
444 func (r *desktopBotRuntime) ForwardTargets(cfg *config.Config) []botForwardTarget {
445 if cfg == nil {
446 return nil
447 }
448 var targets []botForwardTarget
449 seen := make(map[botForwardTarget]bool)
450 for _, conn := range cfg.Bot.Connections {
451 if !conn.Enabled {
452 continue
453 }
454 connID := botruntime.ConnectionRuntimeID(conn)
455 domain := strings.TrimSpace(conn.Domain)
456 for _, sm := range conn.SessionMappings {
457 remoteID := strings.TrimSpace(sm.RemoteID)
458 if remoteID == "" {
459 continue
460 }
461 chatType := bot.ChatDM
462 if sm.ChatType != "" {
463 chatType = bot.ChatType(sm.ChatType)
464 }
465 target := botForwardTarget{
466 ConnID: connID,
467 Domain: domain,
468 ChatID: remoteID,
469 ChatType: chatType,
470 }
471 if seen[target] {
472 continue
473 }
474 seen[target] = true
475 targets = append(targets, target)
476 }
477 }
478 return targets
479 }
480
480 lines GO