Files
修雨 c151756168 feat(connect): Q&A-bot hardening for the stream bridge
Five operational gaps that bite when a connector-backed bot answers
questions in real groups:

- Single-instance lock: one connector per robot per machine (pid file,
  stale-lock takeover). Duplicate Stream connections on one clientId get
  messages load-balanced between them, so the bot answers intermittently
  (verified live).
- Access policy: --allowed-users / --allowed-groups allowlists plus a
  per-sender rate limit (--user-rate-limit, default 20/min). Every
  message is an LLM call; without this anyone who can reach the bot can
  spend on its behalf.
- Per-conversation FIFO queue: same-chat messages run in arrival order
  (follow-ups need the previous turn's --resume session; parallel CLIs
  racing one session corrupt it); different chats stay parallel.
- Picture messages: resolve the callback downloadCode via robot
  messageFiles/download to a local temp file and hand the path to the
  agent CLI — error screenshots are the top Q&A inbound and were
  previously dropped silently. The claudecode persona now permits
  reading attached files.
- --knowledge-dir: lightweight local retrieval (markdown chunking,
  latin-token + CJK-bigram scoring) prepends top-k matching chunks to
  each question while the agent keeps running from the clean scratch
  dir, so replies stay inside DingTalk's AI-assistant response window.

All knobs have env mirrors (DWS_ALLOWED_USERS / DWS_ALLOWED_GROUPS /
DWS_USER_RATE_LIMIT / DWS_KNOWLEDGE_DIR) for service-style runs.

go build ./... clean; go test ./internal/helpers -race passes with new
tests for lock/gate/queue/knowledge/media. test/cli_compat, test/scripts
and test/unit fail identically on the unmodified base (pre-existing
fixture issues).
2026-06-11 12:18:25 +08:00

83 lines
2.8 KiB
Go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package helpers
import (
"errors"
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"syscall"
)
// connectLockDir is a var so tests can isolate lock files.
var connectLockDir = os.TempDir()
// acquireConnectLock guards against two local connectors holding Stream
// connections for the same robot. DingTalk load-balances pushes across
// connections sharing one clientId, so a duplicate connector makes the bot
// answer intermittently — the hardest-to-diagnose failure mode this command
// has (verified live). The lock is a pid file; a stale file whose owning
// process is gone is taken over silently.
func acquireConnectLock(clientID string) (release func(), err error) {
path := filepath.Join(connectLockDir, "dws-connect-"+sanitizeLockID(clientID)+".pid")
for attempt := 0; attempt < 2; attempt++ {
f, openErr := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
if openErr == nil {
fmt.Fprintf(f, "%d", os.Getpid())
f.Close()
return func() { _ = os.Remove(path) }, nil
}
if !os.IsExist(openErr) {
return nil, openErr
}
raw, readErr := os.ReadFile(path)
if readErr == nil {
if pid, perr := strconv.Atoi(strings.TrimSpace(string(raw))); perr == nil && pid > 0 && processAlive(pid) {
return nil, fmt.Errorf("机器人 %s 已有连接器在本机运行 (pid %d)。同一机器人的多个 Stream 连接会随机分流消息,表现为机器人时灵时不灵;请先停掉旧进程,或确认后删除 %s", clientID, pid, path)
}
}
// Stale lock (owner gone or file unreadable): take over.
_ = os.Remove(path)
}
return nil, fmt.Errorf("无法获取连接器锁 %s", path)
}
// sanitizeLockID keeps the lock filename filesystem-safe.
func sanitizeLockID(s string) string {
return strings.Map(func(r rune) rune {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
return r
}
return '_'
}, s)
}
// processAlive reports whether pid refers to a live process (unix signal-0
// probe; a permission error still means "alive").
func processAlive(pid int) bool {
proc, err := os.FindProcess(pid)
if err != nil {
return false
}
err = proc.Signal(syscall.Signal(0))
if err == nil {
return true
}
return errors.Is(err, syscall.EPERM)
}