fix(docs): update WeCom AI Bot timeout duration in README and improve streamTask comments
This commit is contained in:
parent
e88b39f21e
commit
81f6787dd5
2 changed files with 33 additions and 27 deletions
|
|
@ -75,9 +75,9 @@ PicoClaw 立即返回 {finish: false}(Agent 开始处理)
|
||||||
└─ Agent 完成 → 返回 {finish: true, content: "回答内容"}
|
└─ Agent 完成 → 返回 {finish: true, content: "回答内容"}
|
||||||
```
|
```
|
||||||
|
|
||||||
**超时处理**(任务超过 5 分 30 秒):
|
**超时处理**(任务超过 30 秒):
|
||||||
|
|
||||||
若 Agent 处理时间超过约 5 分 30 秒(企业微信最大轮询窗口为 6 分钟),PicoClaw 会:
|
若 Agent 处理时间超过约 30 秒(企业微信最大轮询窗口为 6 分钟),PicoClaw 会:
|
||||||
|
|
||||||
1. 立即关闭流,向用户显示「⏳ 正在处理中,请稍候,结果将稍后发送。」
|
1. 立即关闭流,向用户显示「⏳ 正在处理中,请稍候,结果将稍后发送。」
|
||||||
2. Agent 继续在后台运行
|
2. Agent 继续在后台运行
|
||||||
|
|
|
||||||
|
|
@ -37,21 +37,28 @@ type WeComAIBotChannel struct {
|
||||||
taskMu sync.RWMutex
|
taskMu sync.RWMutex
|
||||||
}
|
}
|
||||||
|
|
||||||
// streamTask represents a streaming task for AI Bot
|
// streamTask represents a streaming task for AI Bot.
|
||||||
|
//
|
||||||
|
// Mutable fields (Finished, StreamClosed, StreamClosedAt) must be read/written
|
||||||
|
// while holding WeComAIBotChannel.taskMu. Immutable fields (StreamID, ChatID,
|
||||||
|
// ResponseURL, Question, CreatedTime, Deadline, answerCh, ctx, cancel) are set
|
||||||
|
// once at creation and never modified, so they are safe to read without a lock.
|
||||||
type streamTask struct {
|
type streamTask struct {
|
||||||
|
// immutable after creation
|
||||||
StreamID string
|
StreamID string
|
||||||
ChatID string // used by Send() to find this task
|
ChatID string // used by Send() to find this task
|
||||||
ResponseURL string // temporary URL for proactive reply (valid 1 hour, use once)
|
ResponseURL string // temporary URL for proactive reply (valid 1 hour, use once)
|
||||||
Question string
|
Question string
|
||||||
CreatedTime time.Time
|
CreatedTime time.Time
|
||||||
Deadline time.Time // ~30s, we close the stream here and switch to response_url
|
Deadline time.Time // ~30s, we close the stream here and switch to response_url
|
||||||
StreamClosed bool // stream returned finish:true; waiting for agent to reply via response_url
|
|
||||||
StreamClosedAt time.Time // set when StreamClosed becomes true; used for accelerated cleanup
|
|
||||||
Finished bool // fully done
|
|
||||||
mu sync.Mutex
|
|
||||||
answerCh chan string // receives agent reply from Send()
|
answerCh chan string // receives agent reply from Send()
|
||||||
ctx context.Context // canceled when task is removed; used to interrupt the agent goroutine
|
ctx context.Context // canceled when task is removed; used to interrupt the agent goroutine
|
||||||
cancel context.CancelFunc // call on task removal to cancel ctx
|
cancel context.CancelFunc // call on task removal to cancel ctx
|
||||||
|
|
||||||
|
// mutable — guarded by WeComAIBotChannel.taskMu
|
||||||
|
StreamClosed bool // stream returned finish:true; waiting for agent to reply via response_url
|
||||||
|
StreamClosedAt time.Time // set when StreamClosed becomes true; used for accelerated cleanup
|
||||||
|
Finished bool // fully done
|
||||||
}
|
}
|
||||||
|
|
||||||
// WeComAIBotMessage represents the decrypted JSON message from WeCom AI Bot
|
// WeComAIBotMessage represents the decrypted JSON message from WeCom AI Bot
|
||||||
|
|
@ -194,8 +201,13 @@ func (c *WeComAIBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) e
|
||||||
}
|
}
|
||||||
c.chatTasks[msg.ChatID] = queue
|
c.chatTasks[msg.ChatID] = queue
|
||||||
var task *streamTask
|
var task *streamTask
|
||||||
|
var streamClosed bool
|
||||||
|
var responseURL string
|
||||||
if len(queue) > 0 {
|
if len(queue) > 0 {
|
||||||
task = queue[0]
|
task = queue[0]
|
||||||
|
// Read mutable fields while holding c.taskMu to avoid data races.
|
||||||
|
streamClosed = task.StreamClosed
|
||||||
|
responseURL = task.ResponseURL
|
||||||
}
|
}
|
||||||
c.taskMu.Unlock()
|
c.taskMu.Unlock()
|
||||||
|
|
||||||
|
|
@ -210,13 +222,9 @@ func (c *WeComAIBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) e
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
task.mu.Lock()
|
|
||||||
streamClosed := task.StreamClosed
|
|
||||||
responseURL := task.ResponseURL
|
|
||||||
task.mu.Unlock()
|
|
||||||
|
|
||||||
if streamClosed {
|
if streamClosed {
|
||||||
// Stream already ended with a "please wait" notice; send the real reply via response_url.
|
// Stream already ended with a "please wait" notice; send the real reply via response_url.
|
||||||
|
// Note: task.StreamID and task.ChatID are immutable, safe to read without a lock.
|
||||||
logger.InfoCF("wecom_aibot", "Sending reply via response_url", map[string]any{
|
logger.InfoCF("wecom_aibot", "Sending reply via response_url", map[string]any{
|
||||||
"stream_id": task.StreamID,
|
"stream_id": task.StreamID,
|
||||||
"chat_id": msg.ChatID,
|
"chat_id": msg.ChatID,
|
||||||
|
|
@ -779,13 +787,11 @@ func (c *WeComAIBotChannel) getStreamResponse(task *streamTask, timestamp, nonce
|
||||||
// Normal finish: remove from all maps.
|
// Normal finish: remove from all maps.
|
||||||
c.removeTask(task)
|
c.removeTask(task)
|
||||||
} else if closeStreamOnly {
|
} else if closeStreamOnly {
|
||||||
// Only mark stream as closed; keep in chatTasks for Send() to find.
|
// Mark stream as closed and remove from streamTasks under a single lock
|
||||||
task.mu.Lock()
|
// to keep StreamClosed/StreamClosedAt consistent with map membership.
|
||||||
|
c.taskMu.Lock()
|
||||||
task.StreamClosed = true
|
task.StreamClosed = true
|
||||||
task.StreamClosedAt = time.Now()
|
task.StreamClosedAt = time.Now()
|
||||||
task.mu.Unlock()
|
|
||||||
// Remove from streamTasks (no more stream polls expected).
|
|
||||||
c.taskMu.Lock()
|
|
||||||
delete(c.streamTasks, task.StreamID)
|
delete(c.streamTasks, task.StreamID)
|
||||||
c.taskMu.Unlock()
|
c.taskMu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
@ -816,12 +822,12 @@ func (c *WeComAIBotChannel) getStreamResponse(task *streamTask, timestamp, nonce
|
||||||
// removeTask removes a task from both streamTasks and chatTasks, marks it finished,
|
// removeTask removes a task from both streamTasks and chatTasks, marks it finished,
|
||||||
// and cancels its context to interrupt the associated agent goroutine.
|
// and cancels its context to interrupt the associated agent goroutine.
|
||||||
func (c *WeComAIBotChannel) removeTask(task *streamTask) {
|
func (c *WeComAIBotChannel) removeTask(task *streamTask) {
|
||||||
task.mu.Lock()
|
// Cancel first so the agent goroutine stops as soon as possible,
|
||||||
task.Finished = true
|
// before we acquire the write lock.
|
||||||
task.mu.Unlock()
|
task.cancel()
|
||||||
task.cancel() // interrupt agent goroutine bound to this task
|
|
||||||
|
|
||||||
c.taskMu.Lock()
|
c.taskMu.Lock()
|
||||||
|
task.Finished = true // written under c.taskMu, consistent with all readers
|
||||||
delete(c.streamTasks, task.StreamID)
|
delete(c.streamTasks, task.StreamID)
|
||||||
queue := c.chatTasks[task.ChatID]
|
queue := c.chatTasks[task.ChatID]
|
||||||
for i, t := range queue {
|
for i, t := range queue {
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue