| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "crypto/sha256" |
| 5 | "encoding/hex" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "time" |
| 10 | ) |
| 11 | |
| 12 | const ( |
| 13 | QueueModeSteer = "steer" |
| 14 | QueueModeFollowup = "followup" |
| 15 | QueueModeCollect = "collect" |
| 16 | QueueModeInterrupt = "interrupt" |
| 17 | |
| 18 | QueueDropSummarize = "summarize" |
| 19 | QueueDropOld = "old" |
| 20 | QueueDropNew = "new" |
| 21 | |
| 22 | DefaultQueueCap = 20 |
| 23 | ) |
| 24 | |
| 25 | type QueueOptions struct { |
| 26 | Mode string |
| 27 | Cap int |
| 28 | Drop string |
| 29 | } |
| 30 | |
| 31 | type QueueResult struct { |
| 32 | Acquired bool |
| 33 | Queued bool |
| 34 | Rejected bool |
| 35 | Dropped bool |
| 36 | Pending int |
| 37 | Mode string |
| 38 | } |
| 39 | |
| 40 | type QueueSnapshot struct { |
| 41 | Active int |
| 42 | Pending int |
| 43 | Dropped int |
| 44 | Sessions int |
| 45 | } |
| 46 | |
| 47 | func NormalizeQueueMode(mode string) string { |
| 48 | switch strings.ToLower(strings.TrimSpace(mode)) { |
| 49 | case QueueModeSteer: |
| 50 | return QueueModeSteer |
| 51 | case QueueModeFollowup: |
| 52 | return QueueModeFollowup |
| 53 | case QueueModeCollect: |
| 54 | return QueueModeCollect |
| 55 | case QueueModeInterrupt: |
| 56 | return QueueModeInterrupt |
| 57 | default: |
| 58 | return QueueModeSteer |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | func NormalizeOptionalQueueMode(mode string) string { |
| 63 | switch strings.ToLower(strings.TrimSpace(mode)) { |
| 64 | case QueueModeSteer: |
| 65 | return QueueModeSteer |
| 66 | case QueueModeFollowup: |
| 67 | return QueueModeFollowup |
| 68 | case QueueModeCollect: |
| 69 | return QueueModeCollect |
| 70 | case QueueModeInterrupt: |
| 71 | return QueueModeInterrupt |
| 72 | default: |
| 73 | return "" |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | func NormalizeQueueDrop(drop string) string { |
| 78 | switch strings.ToLower(strings.TrimSpace(drop)) { |
| 79 | case QueueDropOld: |
| 80 | return QueueDropOld |
| 81 | case QueueDropNew: |
| 82 | return QueueDropNew |
| 83 | default: |
| 84 | return QueueDropSummarize |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | // BuildSessionKey 根据 Hermes 模式生成稳定的 session key： |
| 89 | // - DM：按 chat 隔离（同一 DM 会话共享历史） |
| 90 | // - 群聊：按 user 隔离（每人独立会话） |
| 91 | // - thread：共享（thread 内所有人共享上下文） |
| 92 | func BuildSessionKey(src SessionSource) string { |
| 93 | var scope string |
| 94 | source := sessionSourceID(src) |
| 95 | switch src.ChatType { |
| 96 | case ChatDM: |
| 97 | scope = fmt.Sprintf("%s:dm:%s", source, src.ChatID) |
| 98 | case ChatGroup: |
| 99 | scope = fmt.Sprintf("%s:group:%s:%s", source, src.ChatID, src.UserID) |
| 100 | case ChatGuild: |
| 101 | scope = fmt.Sprintf("%s:guild:%s:%s", source, src.ChatID, src.UserID) |
| 102 | case ChatDirect: |
| 103 | scope = fmt.Sprintf("%s:direct:%s", source, src.ChatID) |
| 104 | case ChatThread: |
| 105 | threadID := src.ThreadID |
| 106 | if threadID == "" { |
| 107 | threadID = src.ChatID |
| 108 | } |
| 109 | scope = fmt.Sprintf("%s:thread:%s", source, threadID) |
| 110 | default: |
| 111 | scope = fmt.Sprintf("%s:%s:%s:%s", source, src.ChatType, src.ChatID, src.UserID) |
| 112 | } |
| 113 | h := sha256.Sum256([]byte(scope)) |
| 114 | return hex.EncodeToString(h[:])[:16] |
| 115 | } |
| 116 | |
| 117 | func sessionSourceID(src SessionSource) string { |
| 118 | if src.ConnectionID != "" { |
| 119 | return src.ConnectionID |
| 120 | } |
| 121 | if src.Domain != "" { |
| 122 | return fmt.Sprintf("%s:%s", src.Platform, src.Domain) |
| 123 | } |
| 124 | return string(src.Platform) |
| 125 | } |
| 126 | |
| 127 | // slashCommands 是绕过忙碌队列的命令集合。 |
| 128 | var slashCommands = map[string]bool{ |
| 129 | "/stop": true, |
| 130 | "/new": true, |
| 131 | "/reset": true, |
| 132 | "/approve": true, |
| 133 | "/approve-session": true, |
| 134 | "/deny": true, |
| 135 | "/answer": true, |
| 136 | "/mode": true, |
| 137 | "/model": true, |
| 138 | "/queue": true, |
| 139 | "/projects": true, |
| 140 | "/use": true, |
| 141 | "/sessions": true, |
| 142 | "/attach": true, |
| 143 | "/search": true, |
| 144 | "/desktop": true, |
| 145 | "/status": true, |
| 146 | "/help": true, |
| 147 | } |
| 148 | |
| 149 | // IsSlashBypass 判断消息是否为绕过队列的斜杠命令。 |
| 150 | func IsSlashBypass(text string) bool { |
| 151 | if len(text) == 0 { |
| 152 | return false |
| 153 | } |
| 154 | cmd := text |
| 155 | for i, r := range text { |
| 156 | if r == ' ' { |
| 157 | cmd = text[:i] |
| 158 | break |
| 159 | } |
| 160 | } |
| 161 | return slashCommands[cmd] |
| 162 | } |
| 163 | |
| 164 | // pendingTurn 是等待执行的一轮对话。 |
| 165 | type pendingTurn struct { |
| 166 | msg InboundMessage |
| 167 | timestamp time.Time |
| 168 | mode string |
| 169 | } |
| 170 | |
| 171 | // SessionManager 管理 session 级别的并发控制：同一 session 同时只跑一个任务。 |
| 172 | type SessionManager struct { |
| 173 | mu sync.Mutex |
| 174 | active map[string]bool // session key -> 是否正在运行 |
| 175 | pending map[string][]pendingTurn // session key -> 等待队列 |
| 176 | debounce time.Duration |
| 177 | modeOverrides map[string]string |
| 178 | dropped map[string][]string |
| 179 | } |
| 180 | |
| 181 | // NewSessionManager 创建一个新的 session 管理器。debounce 是消息合并窗口。 |
| 182 | func NewSessionManager(debounce time.Duration) *SessionManager { |
| 183 | if debounce <= 0 { |
| 184 | debounce = 1500 * time.Millisecond |
| 185 | } |
| 186 | return &SessionManager{ |
| 187 | active: make(map[string]bool), |
| 188 | pending: make(map[string][]pendingTurn), |
| 189 | debounce: debounce, |
| 190 | modeOverrides: make(map[string]string), |
| 191 | dropped: make(map[string][]string), |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | // TryAcquire 尝试获取 session 锁。如果 session 正忙且消息非绕过命令，返回 false。 |
| 196 | // 返回 (acquired, merged) — merged 为 true 表示消息已合并到等待队列。 |
| 197 | func (sm *SessionManager) TryAcquire(key string, msg InboundMessage) (acquired bool, merged bool) { |
| 198 | result := sm.TryAcquireWithQueue(key, msg, QueueOptions{Mode: QueueModeCollect, Cap: DefaultQueueCap, Drop: QueueDropSummarize}) |
| 199 | return result.Acquired, result.Queued |
| 200 | } |
| 201 | |
| 202 | func (sm *SessionManager) TryAcquireWithQueue(key string, msg InboundMessage, opts QueueOptions) QueueResult { |
| 203 | sm.mu.Lock() |
| 204 | defer sm.mu.Unlock() |
| 205 | |
| 206 | mode := NormalizeQueueMode(opts.Mode) |
| 207 | if mode == QueueModeSteer || mode == QueueModeInterrupt { |
| 208 | mode = QueueModeFollowup |
| 209 | } |
| 210 | cap := opts.Cap |
| 211 | if cap <= 0 { |
| 212 | cap = DefaultQueueCap |
| 213 | } |
| 214 | drop := NormalizeQueueDrop(opts.Drop) |
| 215 | |
| 216 | if sm.active[key] { |
| 217 | // 绕过命令立即返回 true（让调用方直接处理） |
| 218 | if IsSlashBypass(msg.Text) { |
| 219 | return QueueResult{Acquired: true, Mode: mode} |
| 220 | } |
| 221 | queue := sm.pending[key] |
| 222 | if len(queue) >= cap { |
| 223 | switch drop { |
| 224 | case QueueDropNew: |
| 225 | return QueueResult{Rejected: true, Pending: len(queue), Mode: mode} |
| 226 | case QueueDropOld, QueueDropSummarize: |
| 227 | removed := queue[0] |
| 228 | queue = queue[1:] |
| 229 | if drop == QueueDropSummarize { |
| 230 | sm.dropped[key] = append(sm.dropped[key], queueSummary(removed.msg.Text)) |
| 231 | } |
| 232 | } |
| 233 | } |
| 234 | if mode == QueueModeCollect && len(queue) > 0 { |
| 235 | last := &queue[len(queue)-1] |
| 236 | if msg.Text != "" && time.Since(last.timestamp) < sm.debounce { |
| 237 | if last.msg.Text != "" { |
| 238 | last.msg.Text = last.msg.Text + "\n" + msg.Text |
| 239 | } else { |
| 240 | last.msg.Text = msg.Text |
| 241 | } |
| 242 | last.timestamp = time.Now() |
| 243 | last.mode = mode |
| 244 | sm.pending[key] = queue |
| 245 | return QueueResult{Queued: true, Dropped: len(sm.dropped[key]) > 0, Pending: len(queue), Mode: mode} |
| 246 | } |
| 247 | } |
| 248 | queue = append(queue, pendingTurn{msg: msg, timestamp: time.Now(), mode: mode}) |
| 249 | sm.pending[key] = queue |
| 250 | return QueueResult{Queued: true, Dropped: len(sm.dropped[key]) > 0, Pending: len(queue), Mode: mode} |
| 251 | } |
| 252 | |
| 253 | sm.active[key] = true |
| 254 | return QueueResult{Acquired: true, Mode: mode} |
| 255 | } |
| 256 | |
| 257 | // TryAcquireIdle acquires an idle session without enqueueing when another |
| 258 | // goroutine wins the race. Durable inbox items already live on disk, so adding |
| 259 | // a second in-memory copy on a failed claim would execute them twice. |
| 260 | func (sm *SessionManager) TryAcquireIdle(key string) bool { |
| 261 | sm.mu.Lock() |
| 262 | defer sm.mu.Unlock() |
| 263 | if sm.active[key] { |
| 264 | return false |
| 265 | } |
| 266 | sm.active[key] = true |
| 267 | return true |
| 268 | } |
| 269 | |
| 270 | func (sm *SessionManager) ReplacePending(key string, msg InboundMessage) QueueResult { |
| 271 | sm.mu.Lock() |
| 272 | defer sm.mu.Unlock() |
| 273 | if !sm.active[key] { |
| 274 | sm.active[key] = true |
| 275 | return QueueResult{Acquired: true, Mode: QueueModeInterrupt} |
| 276 | } |
| 277 | sm.pending[key] = []pendingTurn{{msg: msg, timestamp: time.Now(), mode: QueueModeFollowup}} |
| 278 | delete(sm.dropped, key) |
| 279 | return QueueResult{Queued: true, Pending: 1, Mode: QueueModeInterrupt} |
| 280 | } |
| 281 | |
| 282 | // Release 释放 session 锁，返回等待队列中的下一条消息（合并后）。 |
| 283 | func (sm *SessionManager) Release(key string) *InboundMessage { |
| 284 | sm.mu.Lock() |
| 285 | defer sm.mu.Unlock() |
| 286 | |
| 287 | queue := sm.pending[key] |
| 288 | if len(queue) == 0 { |
| 289 | delete(sm.active, key) |
| 290 | delete(sm.pending, key) |
| 291 | delete(sm.dropped, key) |
| 292 | return nil |
| 293 | } |
| 294 | |
| 295 | mode := NormalizeQueueMode(queue[0].mode) |
| 296 | var merged *InboundMessage |
| 297 | if mode == QueueModeFollowup { |
| 298 | m := queue[0].msg |
| 299 | merged = &m |
| 300 | if len(queue) == 1 { |
| 301 | delete(sm.pending, key) |
| 302 | } else { |
| 303 | sm.pending[key] = queue[1:] |
| 304 | } |
| 305 | merged.Text = sm.consumeDroppedPrefixLocked(key, merged.Text) |
| 306 | return merged |
| 307 | } |
| 308 | |
| 309 | // collect 模式取出等待队列，并合并其中所有消息。 |
| 310 | for i := range queue { |
| 311 | if merged == nil { |
| 312 | m := queue[i].msg |
| 313 | merged = &m |
| 314 | } else { |
| 315 | if queue[i].msg.Text != "" { |
| 316 | merged.Text = merged.Text + "\n" + queue[i].msg.Text |
| 317 | } |
| 318 | } |
| 319 | } |
| 320 | delete(sm.pending, key) |
| 321 | merged.Text = sm.consumeDroppedPrefixLocked(key, merged.Text) |
| 322 | // active 保持 true，因为调用方会立即用 merged 消息开始新 turn |
| 323 | return merged |
| 324 | } |
| 325 | |
| 326 | func (sm *SessionManager) consumeDroppedPrefixLocked(key, text string) string { |
| 327 | dropped := sm.dropped[key] |
| 328 | if len(dropped) == 0 { |
| 329 | return text |
| 330 | } |
| 331 | delete(sm.dropped, key) |
| 332 | var b strings.Builder |
| 333 | fmt.Fprintf(&b, "[Queue note: %d older pending message(s) were dropped because this bot session reached its queue cap.", len(dropped)) |
| 334 | if len(dropped) > 0 { |
| 335 | b.WriteString(" Dropped summaries:") |
| 336 | limit := min(len(dropped), 3) |
| 337 | for i := range limit { |
| 338 | fmt.Fprintf(&b, "\n- %s", dropped[i]) |
| 339 | } |
| 340 | if len(dropped) > limit { |
| 341 | fmt.Fprintf(&b, "\n- ... and %d more", len(dropped)-limit) |
| 342 | } |
| 343 | } |
| 344 | b.WriteString("]\n\n") |
| 345 | b.WriteString(text) |
| 346 | return b.String() |
| 347 | } |
| 348 | |
| 349 | func queueSummary(text string) string { |
| 350 | text = strings.TrimSpace(text) |
| 351 | if text == "" { |
| 352 | return "(empty message)" |
| 353 | } |
| 354 | runes := []rune(text) |
| 355 | if len(runes) <= 180 { |
| 356 | return text |
| 357 | } |
| 358 | return string(runes[:180]) + "..." |
| 359 | } |
| 360 | |
| 361 | // IsActive 返回 session 是否有正在运行的任务。 |
| 362 | func (sm *SessionManager) IsActive(key string) bool { |
| 363 | sm.mu.Lock() |
| 364 | defer sm.mu.Unlock() |
| 365 | return sm.active[key] |
| 366 | } |
| 367 | |
| 368 | // Debounce returns the collect-mode merge window. |
| 369 | func (sm *SessionManager) Debounce() time.Duration { |
| 370 | if sm == nil { |
| 371 | return 1500 * time.Millisecond |
| 372 | } |
| 373 | return sm.debounce |
| 374 | } |
| 375 | |
| 376 | // runIfIdle holds the per-gateway admission lock while fn switches runtime |
| 377 | // ownership for key. A normal message cannot become active between the idle |
| 378 | // check and the controller unlink/close sequence. |
| 379 | func (sm *SessionManager) runIfIdle(key string, fn func() bool) bool { |
| 380 | sm.mu.Lock() |
| 381 | defer sm.mu.Unlock() |
| 382 | if sm.active[key] || fn == nil { |
| 383 | return false |
| 384 | } |
| 385 | if !fn() { |
| 386 | return false |
| 387 | } |
| 388 | delete(sm.pending, key) |
| 389 | delete(sm.dropped, key) |
| 390 | return true |
| 391 | } |
| 392 | |
| 393 | // ActiveCount 返回当前活跃 session 数。 |
| 394 | func (sm *SessionManager) ActiveCount() int { |
| 395 | sm.mu.Lock() |
| 396 | defer sm.mu.Unlock() |
| 397 | return len(sm.active) |
| 398 | } |
| 399 | |
| 400 | func (sm *SessionManager) PendingCount(key string) int { |
| 401 | sm.mu.Lock() |
| 402 | defer sm.mu.Unlock() |
| 403 | return len(sm.pending[key]) |
| 404 | } |
| 405 | |
| 406 | func (sm *SessionManager) Snapshot() QueueSnapshot { |
| 407 | sm.mu.Lock() |
| 408 | defer sm.mu.Unlock() |
| 409 | var pending int |
| 410 | var dropped int |
| 411 | for _, queue := range sm.pending { |
| 412 | pending += len(queue) |
| 413 | } |
| 414 | for _, summaries := range sm.dropped { |
| 415 | dropped += len(summaries) |
| 416 | } |
| 417 | return QueueSnapshot{ |
| 418 | Active: len(sm.active), |
| 419 | Pending: pending, |
| 420 | Dropped: dropped, |
| 421 | Sessions: len(sm.active) + len(sm.pending), |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | func (sm *SessionManager) QueueMode(key, fallback string) string { |
| 426 | sm.mu.Lock() |
| 427 | defer sm.mu.Unlock() |
| 428 | if mode := sm.modeOverrides[key]; mode != "" { |
| 429 | return mode |
| 430 | } |
| 431 | return NormalizeQueueMode(fallback) |
| 432 | } |
| 433 | |
| 434 | func (sm *SessionManager) SetQueueMode(key, mode string) { |
| 435 | sm.mu.Lock() |
| 436 | defer sm.mu.Unlock() |
| 437 | if normalized := NormalizeOptionalQueueMode(mode); normalized != "" { |
| 438 | sm.modeOverrides[key] = normalized |
| 439 | } |
| 440 | } |
| 441 | |
| 442 | func (sm *SessionManager) ClearQueueMode(key string) { |
| 443 | sm.mu.Lock() |
| 444 | defer sm.mu.Unlock() |
| 445 | delete(sm.modeOverrides, key) |
| 446 | } |
| 447 | |
| 448 | // ForceRelease 强制释放 session（用于 session 关闭或错误恢复）。 |
| 449 | func (sm *SessionManager) ForceRelease(key string) { |
| 450 | sm.mu.Lock() |
| 451 | defer sm.mu.Unlock() |
| 452 | delete(sm.active, key) |
| 453 | delete(sm.pending, key) |
| 454 | delete(sm.dropped, key) |
| 455 | } |
| 456 |