mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 04:16:06 +00:00
feat(channels): real-time agent activity status & intent classification
- Add tool status display on channels during tool execution (streaming preview + reactions) - Emit agent.activity events at phase transitions (thinking, tool_exec, compacting) - Enrich delegation progress with per-member activity and tool info - Add LLM-based intent classifier for DM status queries when agent is busy - Keyword fast-path for cancel/status patterns (no LLM cost) - Falls back to LLM classification with 5s timeout - Supports status_query (immediate reply) and cancel (abort run) intents - Register/unregister runs in makeSchedulerRunFunc for channel inbound tracking - Add sessionRuns secondary index in Router for O(1) IsSessionBusy lookups - Add intent_classify config toggle (global default + per-agent override) - Add tool_status config toggle for channel tool status display - Add i18n keys and translations (en/vi/zh) for status messages - Add web UI config toggles for intent_classify and tool_status
This commit is contained in:
1 parent
e63ff014cb
commit
e593b9cf22
24 files changed
+593
-11
No files matched your search
@@ -38,6 +38,7 @@ deploy-*.sh
|
||||
docs/technical-diaries
|
||||
plans
|
||||
release-manifest.json
|
||||
k8s/
|
||||
|
||||
# Test credentials (never commit)
|
||||
tests/**/creds.json
|
||||
@@ -768,6 +768,34 @@ func runGateway() {
|
||||
return
|
||||
}
|
||||
channelMgr.HandleAgentEvent(agentEvent.Type, agentEvent.RunID, agentEvent.Payload)
|
||||
|
||||
// Route activity events to Router (status registry) and DelegateManager (progress tracking).
|
||||
if agentEvent.Type == protocol.AgentEventActivity {
|
||||
payloadMap, _ := agentEvent.Payload.(map[string]interface{})
|
||||
phase, _ := payloadMap["phase"].(string)
|
||||
tool, _ := payloadMap["tool"].(string)
|
||||
iteration := 0
|
||||
if v, ok := payloadMap["iteration"].(int); ok {
|
||||
iteration = v
|
||||
}
|
||||
|
||||
// Update Router activity registry (for status queries via LLM classify)
|
||||
if sessionKey := agentRouter.SessionKeyForRun(agentEvent.RunID); sessionKey != "" {
|
||||
agentRouter.UpdateActivity(sessionKey, agentEvent.RunID, phase, tool, iteration)
|
||||
}
|
||||
|
||||
// Update DelegateManager activity tracking (for enriched progress notifications)
|
||||
if delegateMgr != nil && agentEvent.DelegationID != "" {
|
||||
delegateMgr.HandleActivityEvent(agentEvent.DelegationID, phase, tool)
|
||||
}
|
||||
}
|
||||
|
||||
// Clear activity on terminal events
|
||||
if agentEvent.Type == protocol.AgentEventRunCompleted || agentEvent.Type == protocol.AgentEventRunFailed {
|
||||
if sessionKey := agentRouter.SessionKeyForRun(agentEvent.RunID); sessionKey != "" {
|
||||
agentRouter.ClearActivity(sessionKey)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// Start inbound message consumer (channel → scheduler → agent → channel)
|
||||
|
||||
+44
-1
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/sessions"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
@@ -185,8 +186,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
chatIDForRun = lk
|
||||
}
|
||||
blockReply := channelMgr != nil && channelMgr.ResolveBlockReply(msg.Channel, cfg.Gateway.BlockReply)
|
||||
toolStatus := cfg.Gateway.ToolStatus == nil || *cfg.Gateway.ToolStatus // default true
|
||||
if channelMgr != nil {
|
||||
channelMgr.RegisterRun(runID, msg.Channel, chatIDForRun, messageID, outMeta, enableStream, blockReply)
|
||||
channelMgr.RegisterRun(runID, msg.Channel, chatIDForRun, messageID, outMeta, enableStream, blockReply, toolStatus)
|
||||
}
|
||||
|
||||
// Group-aware system prompt: help the LLM adapt tone and behavior for group chats.
|
||||
@@ -222,6 +224,47 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
reqMedia = msg.Media
|
||||
}
|
||||
|
||||
// Intent classify fast-path: when agent is busy on DM, classify user intent
|
||||
// to detect status queries, cancel requests, etc. without queueing.
|
||||
// Only for DM (maxConcurrent=1) where messages queue behind the active run.
|
||||
intentClassifyEnabled := cfg.Agents.Defaults.IntentClassify == nil || *cfg.Agents.Defaults.IntentClassify
|
||||
if intentClassifyEnabled && maxConcurrent == 1 && agents.IsSessionBusy(sessionKey) {
|
||||
if loop, ok := agentLoop.(*agent.Loop); ok && loop.Provider() != nil {
|
||||
locale := msg.Metadata["locale"]
|
||||
if locale == "" {
|
||||
locale = "en"
|
||||
}
|
||||
intent := agent.ClassifyIntent(ctx, loop.Provider(), loop.Model(), msg.Content)
|
||||
switch intent {
|
||||
case agent.IntentStatusQuery:
|
||||
status := agents.GetActivity(sessionKey)
|
||||
reply := agent.FormatStatusReply(status, locale)
|
||||
msgBus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: msg.Channel,
|
||||
ChatID: msg.ChatID,
|
||||
Content: reply,
|
||||
Metadata: outMeta,
|
||||
})
|
||||
return
|
||||
case agent.IntentCancel:
|
||||
aborted := agents.AbortRunsForSession(sessionKey)
|
||||
if len(aborted) > 0 {
|
||||
slog.Info("inbound: cancelled runs via intent classify",
|
||||
"session", sessionKey, "aborted", aborted)
|
||||
msgBus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: msg.Channel,
|
||||
ChatID: msg.ChatID,
|
||||
Content: i18n.T(locale, i18n.MsgCancelledReply),
|
||||
Metadata: outMeta,
|
||||
})
|
||||
}
|
||||
return
|
||||
default:
|
||||
// steer / new_task → queue as normal
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Schedule through main lane (per-session concurrency controlled by maxConcurrent)
|
||||
outCh := sched.ScheduleWithOpts(ctx, "main", agent.RunRequest{
|
||||
SessionKey: sessionKey,
|
||||
|
||||
@@ -24,6 +24,16 @@ func makeSchedulerRunFunc(agents *agent.Router, cfg *config.Config) scheduler.Ru
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("agent %s not found: %w", agentID, err)
|
||||
}
|
||||
return loop.Run(ctx, req)
|
||||
|
||||
// Register run with the agent router so IsSessionBusy() and AbortRunsForSession()
|
||||
// work correctly for inbound channel runs (Telegram DM intent classifier, /stop, etc.).
|
||||
// The ctx from the scheduler is already cancellable; we create a child so the router's
|
||||
// cancel func is independent from the scheduler's cancel func. Calling cancel twice is safe.
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
agents.RegisterRun(req.RunID, req.SessionKey, agentID, cancel)
|
||||
defer agents.UnregisterRun(req.RunID)
|
||||
defer cancel()
|
||||
|
||||
return loop.Run(runCtx, req)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||||
)
|
||||
|
||||
// IntentType represents the classified intent of a user message.
|
||||
type IntentType string
|
||||
|
||||
const (
|
||||
IntentStatusQuery IntentType = "status_query"
|
||||
IntentCancel IntentType = "cancel"
|
||||
IntentSteer IntentType = "steer"
|
||||
IntentNewTask IntentType = "new_task"
|
||||
)
|
||||
|
||||
const intentClassifyTimeout = 5 * time.Second
|
||||
|
||||
const intentSystemPrompt = `You are an intent classifier. The user has sent a message while the AI assistant is busy processing a previous request.
|
||||
|
||||
Classify the user's intent into exactly ONE of these categories:
|
||||
- status_query: The user is asking about progress, status, or what the assistant is currently doing (e.g., "what are you doing?", "status?", "how far along?", "đang làm gì?", "bao giờ xong?")
|
||||
- cancel: The user wants to stop or cancel the current task (e.g., "stop", "cancel", "nevermind", "thôi", "dừng lại")
|
||||
- steer: The user wants to add instructions or redirect the current task (e.g., "also check X", "focus on Y instead", "thêm phần Z nữa")
|
||||
- new_task: The user is sending a new unrelated request or message
|
||||
|
||||
Respond with ONLY the category name, nothing else.`
|
||||
|
||||
// intentPatterns provides regex-free fast-path detection for common keywords.
|
||||
// Avoids LLM call for obvious patterns, saving cost and latency.
|
||||
var statusKeywords = []string{
|
||||
"status", "progress", "đang làm gì", "bao giờ xong", "做什么", "进度",
|
||||
"what are you doing", "how far", "?",
|
||||
}
|
||||
var cancelKeywords = []string{
|
||||
"stop", "cancel", "abort", "thôi", "dừng", "hủy", "取消", "停",
|
||||
"nevermind", "never mind",
|
||||
}
|
||||
|
||||
// quickClassify attempts keyword-based classification before calling the LLM.
|
||||
// Returns (intent, true) if a match is found, or ("", false) to fall through to LLM.
|
||||
func quickClassify(msg string) (IntentType, bool) {
|
||||
lower := strings.ToLower(strings.TrimSpace(msg))
|
||||
// Short messages (< 30 chars) are more likely to be simple intents.
|
||||
if len(lower) > 60 {
|
||||
return "", false
|
||||
}
|
||||
for _, kw := range cancelKeywords {
|
||||
if strings.Contains(lower, kw) {
|
||||
return IntentCancel, true
|
||||
}
|
||||
}
|
||||
for _, kw := range statusKeywords {
|
||||
if strings.Contains(lower, kw) {
|
||||
return IntentStatusQuery, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// ClassifyIntent determines the intent of a user message sent while the agent is busy.
|
||||
// Uses keyword fast-path first, then falls back to LLM classification.
|
||||
// Falls back to IntentNewTask on any error.
|
||||
func ClassifyIntent(ctx context.Context, provider providers.Provider, model, userMessage string) IntentType {
|
||||
// Fast-path: keyword matching for obvious patterns (no LLM cost).
|
||||
if intent, ok := quickClassify(userMessage); ok {
|
||||
return intent
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, intentClassifyTimeout)
|
||||
defer cancel()
|
||||
|
||||
resp, err := provider.Chat(ctx, providers.ChatRequest{
|
||||
Messages: []providers.Message{
|
||||
{Role: "system", Content: intentSystemPrompt},
|
||||
{Role: "user", Content: userMessage},
|
||||
},
|
||||
Model: model,
|
||||
Options: map[string]interface{}{
|
||||
providers.OptMaxTokens: 20,
|
||||
providers.OptTemperature: 0.0,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return IntentNewTask
|
||||
}
|
||||
|
||||
result := strings.TrimSpace(strings.ToLower(resp.Content))
|
||||
switch {
|
||||
case strings.Contains(result, "status_query"):
|
||||
return IntentStatusQuery
|
||||
case strings.Contains(result, "cancel"):
|
||||
return IntentCancel
|
||||
case strings.Contains(result, "steer"):
|
||||
return IntentSteer
|
||||
default:
|
||||
return IntentNewTask
|
||||
}
|
||||
}
|
||||
|
||||
// FormatStatusReply builds a user-friendly status response from the current agent activity.
|
||||
func FormatStatusReply(status *AgentActivityStatus, locale string) string {
|
||||
if status == nil {
|
||||
return i18n.T(locale, i18n.MsgStatusWorking)
|
||||
}
|
||||
|
||||
elapsed := time.Since(status.StartedAt).Round(time.Second)
|
||||
phase := formatPhase(status.Phase, status.Tool, locale)
|
||||
|
||||
return i18n.T(locale, i18n.MsgStatusDetailed, phase, status.Iteration, elapsed)
|
||||
}
|
||||
|
||||
// formatPhase returns a human-readable phase description.
|
||||
func formatPhase(phase, tool, locale string) string {
|
||||
switch phase {
|
||||
case "thinking":
|
||||
return i18n.T(locale, i18n.MsgStatusPhaseThinking)
|
||||
case "tool_exec":
|
||||
if tool != "" {
|
||||
return i18n.T(locale, i18n.MsgStatusPhaseToolExec, formatToolLabel(tool))
|
||||
}
|
||||
return i18n.T(locale, i18n.MsgStatusPhaseTools)
|
||||
case "compacting":
|
||||
return i18n.T(locale, i18n.MsgStatusPhaseCompact)
|
||||
default:
|
||||
return i18n.T(locale, i18n.MsgStatusPhaseDefault)
|
||||
}
|
||||
}
|
||||
|
||||
// formatToolLabel returns a user-friendly label for a tool name.
|
||||
func formatToolLabel(tool string) string {
|
||||
switch {
|
||||
case strings.HasPrefix(tool, "web"):
|
||||
return "web search"
|
||||
case tool == "exec":
|
||||
return "code execution"
|
||||
case tool == "browser":
|
||||
return "browser"
|
||||
case tool == "spawn" || tool == "delegate":
|
||||
return "delegation"
|
||||
case strings.HasPrefix(tool, "memory"):
|
||||
return "memory"
|
||||
case strings.HasPrefix(tool, "file"):
|
||||
return "file operations"
|
||||
default:
|
||||
return tool
|
||||
}
|
||||
}
|
||||
@@ -374,6 +374,14 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
|
||||
|
||||
slog.Debug("agent iteration", "agent", l.id, "iteration", iteration, "messages", len(messages))
|
||||
|
||||
// Emit activity event: thinking phase
|
||||
emitRun(AgentEvent{
|
||||
Type: protocol.AgentEventActivity,
|
||||
AgentID: l.id,
|
||||
RunID: req.RunID,
|
||||
Payload: map[string]interface{}{"phase": "thinking", "iteration": iteration},
|
||||
})
|
||||
|
||||
// Build provider request with policy-filtered tools
|
||||
var toolDefs []providers.ToolDefinition
|
||||
var allowedTools map[string]bool
|
||||
@@ -493,6 +501,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
|
||||
|
||||
if promptTokens >= threshold {
|
||||
midLoopCompacted = true
|
||||
emitRun(AgentEvent{
|
||||
Type: protocol.AgentEventActivity,
|
||||
AgentID: l.id,
|
||||
RunID: req.RunID,
|
||||
Payload: map[string]interface{}{"phase": "compacting", "iteration": iteration},
|
||||
})
|
||||
if compacted := l.compactMessagesInPlace(ctx, messages); compacted != nil {
|
||||
messages = compacted
|
||||
}
|
||||
@@ -593,6 +607,25 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
|
||||
continue // one more LLM call for summarization, then loop exits (no tool calls)
|
||||
}
|
||||
|
||||
// Emit activity event: tool execution phase
|
||||
if len(resp.ToolCalls) > 0 {
|
||||
toolNames := make([]string, len(resp.ToolCalls))
|
||||
for i, tc := range resp.ToolCalls {
|
||||
toolNames[i] = tc.Name
|
||||
}
|
||||
emitRun(AgentEvent{
|
||||
Type: protocol.AgentEventActivity,
|
||||
AgentID: l.id,
|
||||
RunID: req.RunID,
|
||||
Payload: map[string]interface{}{
|
||||
"phase": "tool_exec",
|
||||
"tool": toolNames[0],
|
||||
"tools": toolNames,
|
||||
"iteration": iteration,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// Execute tool calls (parallel when multiple, sequential when single)
|
||||
if len(resp.ToolCalls) == 1 {
|
||||
// Single tool: sequential — no goroutine overhead
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"log/slog"
|
||||
"strings"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
)
|
||||
|
||||
@@ -44,6 +45,10 @@ func (l *Loop) InvalidateUserWorkspace(userID string) {
|
||||
l.userWorkspaces.Delete(userID)
|
||||
}
|
||||
|
||||
// Provider returns the LLM provider for this agent loop.
|
||||
// Used by intent classifier to make lightweight LLM calls with the agent's own provider.
|
||||
func (l *Loop) Provider() providers.Provider { return l.provider }
|
||||
|
||||
// ProviderName returns the name of this agent's LLM provider (e.g. "anthropic", "openai").
|
||||
func (l *Loop) ProviderName() string {
|
||||
if l.provider == nil {
|
||||
|
||||
@@ -19,15 +19,26 @@ type agentEntry struct {
|
||||
cachedAt time.Time
|
||||
}
|
||||
|
||||
// AgentActivityStatus tracks the current phase of a running agent for status queries.
|
||||
type AgentActivityStatus struct {
|
||||
RunID string
|
||||
Phase string // "thinking", "tool_exec", "compacting"
|
||||
Tool string // current tool name (when Phase == "tool_exec")
|
||||
Iteration int
|
||||
StartedAt time.Time
|
||||
}
|
||||
|
||||
// Router manages multiple agent Loop instances.
|
||||
// Each agent has a unique ID and its own provider/model/tools config.
|
||||
// Cached Loops expire after TTL (safety net for multi-instance).
|
||||
type Router struct {
|
||||
agents map[string]*agentEntry
|
||||
mu sync.RWMutex
|
||||
activeRuns sync.Map // runID → *ActiveRun
|
||||
resolver ResolverFunc // optional: lazy creation from DB
|
||||
ttl time.Duration
|
||||
agents map[string]*agentEntry
|
||||
mu sync.RWMutex
|
||||
activeRuns sync.Map // runID → *ActiveRun
|
||||
sessionRuns sync.Map // sessionKey → runID (secondary index for O(1) IsSessionBusy)
|
||||
agentActivity sync.Map // sessionKey → *AgentActivityStatus
|
||||
resolver ResolverFunc // optional: lazy creation from DB
|
||||
ttl time.Duration
|
||||
}
|
||||
|
||||
func NewRouter() *Router {
|
||||
@@ -160,10 +171,15 @@ func (r *Router) RegisterRun(runID, sessionKey, agentID string, cancel context.C
|
||||
Cancel: cancel,
|
||||
StartedAt: time.Now(),
|
||||
})
|
||||
r.sessionRuns.Store(sessionKey, runID)
|
||||
}
|
||||
|
||||
// UnregisterRun removes a completed/cancelled run from tracking.
|
||||
func (r *Router) UnregisterRun(runID string) {
|
||||
if val, ok := r.activeRuns.Load(runID); ok {
|
||||
run := val.(*ActiveRun)
|
||||
r.sessionRuns.Delete(run.SessionKey)
|
||||
}
|
||||
r.activeRuns.Delete(runID)
|
||||
}
|
||||
|
||||
@@ -183,6 +199,7 @@ func (r *Router) AbortRun(runID, sessionKey string) bool {
|
||||
}
|
||||
|
||||
run.Cancel()
|
||||
r.sessionRuns.Delete(run.SessionKey)
|
||||
r.activeRuns.Delete(runID)
|
||||
return true
|
||||
}
|
||||
@@ -199,6 +216,48 @@ func (r *Router) InvalidateUserWorkspace(userID string) {
|
||||
}
|
||||
}
|
||||
|
||||
// SessionKeyForRun returns the session key associated with a run ID, or "" if not found.
|
||||
func (r *Router) SessionKeyForRun(runID string) string {
|
||||
val, ok := r.activeRuns.Load(runID)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return val.(*ActiveRun).SessionKey
|
||||
}
|
||||
|
||||
// UpdateActivity records the current phase of a running agent for status queries.
|
||||
// Called from the bus subscriber on agent.activity events.
|
||||
func (r *Router) UpdateActivity(sessionKey, runID, phase, tool string, iteration int) {
|
||||
r.agentActivity.Store(sessionKey, &AgentActivityStatus{
|
||||
RunID: runID,
|
||||
Phase: phase,
|
||||
Tool: tool,
|
||||
Iteration: iteration,
|
||||
StartedAt: time.Now(),
|
||||
})
|
||||
}
|
||||
|
||||
// ClearActivity removes the activity status for a session (on run completion).
|
||||
func (r *Router) ClearActivity(sessionKey string) {
|
||||
r.agentActivity.Delete(sessionKey)
|
||||
}
|
||||
|
||||
// GetActivity returns the current activity status for a session, or nil if idle.
|
||||
func (r *Router) GetActivity(sessionKey string) *AgentActivityStatus {
|
||||
val, ok := r.agentActivity.Load(sessionKey)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return val.(*AgentActivityStatus)
|
||||
}
|
||||
|
||||
// IsSessionBusy returns true if there's an active run for the given session key.
|
||||
// O(1) via sessionRuns secondary index.
|
||||
func (r *Router) IsSessionBusy(sessionKey string) bool {
|
||||
_, ok := r.sessionRuns.Load(sessionKey)
|
||||
return ok
|
||||
}
|
||||
|
||||
// AbortRunsForSession cancels all active runs for a session key.
|
||||
// Returns the list of aborted run IDs.
|
||||
func (r *Router) AbortRunsForSession(sessionKey string) []string {
|
||||
@@ -208,6 +267,7 @@ func (r *Router) AbortRunsForSession(sessionKey string) []string {
|
||||
if run.SessionKey == sessionKey {
|
||||
run.Cancel()
|
||||
r.activeRuns.Delete(key)
|
||||
r.sessionRuns.Delete(sessionKey)
|
||||
aborted = append(aborted, run.RunID)
|
||||
}
|
||||
return true
|
||||
|
||||
+113
-1
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
||||
@@ -46,6 +47,20 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload interface{})
|
||||
if err := sc.OnStreamEnd(ctx, rc.ChatID, ""); err != nil {
|
||||
slog.Debug("stream tool-phase end failed", "channel", rc.ChannelName, "error", err)
|
||||
}
|
||||
|
||||
// Show tool status in streaming preview (edit placeholder with tool name).
|
||||
toolName := extractPayloadString(payload, "name")
|
||||
if toolName != "" && rc.ToolStatusEnabled {
|
||||
statusText := formatToolStatus(toolName)
|
||||
outMeta := copyRoutingMeta(rc.Metadata)
|
||||
outMeta["placeholder_update"] = "true"
|
||||
m.bus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: rc.ChannelName,
|
||||
ChatID: rc.ChatID,
|
||||
Content: statusText,
|
||||
Metadata: outMeta,
|
||||
})
|
||||
}
|
||||
case protocol.ChatEventChunk:
|
||||
// Accumulate chunk deltas into full text.
|
||||
// When entering a new LLM iteration (first chunk after tool.call),
|
||||
@@ -149,7 +164,10 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload interface{})
|
||||
case protocol.AgentEventRunStarted:
|
||||
status = "thinking"
|
||||
case protocol.AgentEventToolCall:
|
||||
status = "tool"
|
||||
// Use tool-specific reaction statuses to activate existing variants
|
||||
// (web → ⚡, coding → 👨💻) that are already defined in channel reaction maps.
|
||||
toolName := extractPayloadString(payload, "name")
|
||||
status = resolveToolReactionStatus(toolName)
|
||||
case protocol.AgentEventRunCompleted:
|
||||
status = "done"
|
||||
case protocol.AgentEventRunFailed:
|
||||
@@ -180,3 +198,97 @@ func extractPayloadString(payload interface{}, key string) string {
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// copyRoutingMeta copies channel routing metadata (thread_id, local_key, group_id)
|
||||
// from RunContext.Metadata into a new map suitable for outbound messages.
|
||||
func copyRoutingMeta(src map[string]string) map[string]string {
|
||||
out := make(map[string]string)
|
||||
for _, k := range []string{"message_thread_id", "local_key", "group_id"} {
|
||||
if v := src[k]; v != "" {
|
||||
out[k] = v
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// toolStatusMap maps builtin tool names to user-friendly status messages.
|
||||
var toolStatusMap = map[string]string{
|
||||
// Filesystem
|
||||
"read_file": "📝 Reading file...",
|
||||
"write_file": "📝 Writing file...",
|
||||
"list_files": "📝 Listing files...",
|
||||
"edit": "📝 Editing file...",
|
||||
// Runtime
|
||||
"exec": "⚡ Running code...",
|
||||
// Web
|
||||
"web_search": "🔍 Searching the web...",
|
||||
"web_fetch": "🔍 Fetching web content...",
|
||||
// Memory
|
||||
"memory_search": "🧠 Searching memory...",
|
||||
"memory_get": "🧠 Retrieving memory...",
|
||||
"knowledge_graph_search": "🧠 Querying knowledge graph...",
|
||||
// Media
|
||||
"read_image": "👁 Analyzing image...",
|
||||
"read_document": "📄 Reading document...",
|
||||
"read_audio": "🎧 Processing audio...",
|
||||
"read_video": "🎬 Processing video...",
|
||||
"create_image": "🎨 Creating image...",
|
||||
"create_video": "🎬 Creating video...",
|
||||
"create_audio": "🎵 Creating audio...",
|
||||
"tts": "🔊 Generating speech...",
|
||||
// Browser
|
||||
"browser": "🌐 Browsing...",
|
||||
// Delegation & teams
|
||||
"spawn": "👥 Delegating task...",
|
||||
"handoff": "🔄 Handing off...",
|
||||
"team_tasks": "📋 Managing team tasks...",
|
||||
"team_message": "💬 Sending team message...",
|
||||
// Sessions
|
||||
"sessions_list": "📋 Listing sessions...",
|
||||
"session_status": "📋 Checking session...",
|
||||
"sessions_history": "📋 Reading history...",
|
||||
"sessions_send": "📤 Sending message...",
|
||||
// Other
|
||||
"message": "📤 Sending message...",
|
||||
"cron": "⏰ Managing schedule...",
|
||||
"skill_search": "🔍 Searching skills...",
|
||||
"use_skill": "🧩 Using skill...",
|
||||
"delegate_search": "🔍 Searching delegates...",
|
||||
"evaluate_loop": "🔄 Evaluating...",
|
||||
"mcp_tool_search": "🔌 Searching MCP tools...",
|
||||
}
|
||||
|
||||
// toolPrefixStatus maps tool name prefixes to status messages (fallback for dynamic tools).
|
||||
var toolPrefixStatus = []struct {
|
||||
prefix string
|
||||
status string
|
||||
}{
|
||||
{"mcp_", "🔌 Using external tool..."},
|
||||
}
|
||||
|
||||
// formatToolStatus returns a user-friendly status message for a tool name.
|
||||
func formatToolStatus(toolName string) string {
|
||||
if s, ok := toolStatusMap[toolName]; ok {
|
||||
return s
|
||||
}
|
||||
for _, p := range toolPrefixStatus {
|
||||
if strings.HasPrefix(toolName, p.prefix) {
|
||||
return p.status
|
||||
}
|
||||
}
|
||||
return "🔧 Running " + toolName + "..."
|
||||
}
|
||||
|
||||
// resolveToolReactionStatus maps a tool name to a reaction status string.
|
||||
// Returns tool-specific statuses ("web", "coding") that activate existing
|
||||
// but previously unused reaction variants in channel implementations.
|
||||
func resolveToolReactionStatus(toolName string) string {
|
||||
switch {
|
||||
case strings.HasPrefix(toolName, "web") || toolName == "browser":
|
||||
return "web"
|
||||
case toolName == "exec":
|
||||
return "coding"
|
||||
default:
|
||||
return "tool"
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,7 @@ type RunContext struct {
|
||||
Metadata map[string]string // outbound routing metadata (thread_id, local_key, group_id)
|
||||
Streaming bool // whether run uses streaming (to avoid double-delivery of block replies)
|
||||
BlockReplyEnabled bool // whether block.reply delivery is enabled for this run (resolved at RegisterRun time)
|
||||
ToolStatusEnabled bool // whether tool name shows in streaming preview during tool execution
|
||||
mu sync.Mutex
|
||||
streamBuffer string // accumulated streaming text (chunks are deltas)
|
||||
inToolPhase bool // true after tool.call, reset on next chunk (new LLM iteration)
|
||||
|
||||
@@ -4,7 +4,7 @@ package channels
|
||||
|
||||
// RegisterRun associates a run ID with a channel context so agent events
|
||||
// (chunks, tool calls, completion) can be forwarded to the originating channel.
|
||||
func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, metadata map[string]string, streaming, blockReply bool) {
|
||||
func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, metadata map[string]string, streaming, blockReply, toolStatus bool) {
|
||||
m.runs.Store(runID, &RunContext{
|
||||
ChannelName: channelName,
|
||||
ChatID: chatID,
|
||||
@@ -12,6 +12,7 @@ func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, meta
|
||||
Metadata: metadata,
|
||||
Streaming: streaming,
|
||||
BlockReplyEnabled: blockReply,
|
||||
ToolStatusEnabled: toolStatus,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -124,6 +124,10 @@ type AgentDefaults struct {
|
||||
// Bootstrap context truncation limits (matching TS bootstrapMaxChars / bootstrapTotalMaxChars)
|
||||
BootstrapMaxChars int `json:"bootstrapMaxChars,omitempty"` // per-file max before truncation (default 20000)
|
||||
BootstrapTotalMaxChars int `json:"bootstrapTotalMaxChars,omitempty"` // total budget across all files (default 24000)
|
||||
// IntentClassify enables LLM-based intent classification for messages sent while agent is busy.
|
||||
// When enabled, status queries ("what are you doing?") get immediate replies instead of queueing.
|
||||
// Default: true (nil = true).
|
||||
IntentClassify *bool `json:"intent_classify,omitempty"`
|
||||
}
|
||||
|
||||
// CompactionConfig configures session compaction behaviour.
|
||||
|
||||
@@ -254,6 +254,7 @@ type GatewayConfig struct {
|
||||
InboundDebounceMs int `json:"inbound_debounce_ms,omitempty"` // merge rapid messages from same sender (default 1000ms, -1 = disabled)
|
||||
Quota *QuotaConfig `json:"quota,omitempty"` // per-user/group request quotas
|
||||
BlockReply *bool `json:"block_reply,omitempty"` // deliver intermediate text during tool iterations (default false)
|
||||
ToolStatus *bool `json:"tool_status,omitempty"` // show tool name in streaming preview during tool execution (default true)
|
||||
}
|
||||
|
||||
// ToolsConfig controls tool availability, policy, and web search.
|
||||
|
||||
@@ -122,6 +122,16 @@ func init() {
|
||||
MsgNoPendingOAuth: "no pending OAuth flow",
|
||||
MsgFailedToSaveToken: "failed to save token",
|
||||
|
||||
// Intent Classify
|
||||
MsgStatusWorking: "🔄 I'm working on your request... Please wait.",
|
||||
MsgStatusDetailed: "🔄 I'm currently working on your request...\n%s (iteration %d)\nRunning for: %s\n\nPlease wait — I'll respond when done.",
|
||||
MsgStatusPhaseThinking: "Phase: Thinking...",
|
||||
MsgStatusPhaseToolExec: "Phase: Running %s",
|
||||
MsgStatusPhaseTools: "Phase: Executing tools...",
|
||||
MsgStatusPhaseCompact: "Phase: Compacting context...",
|
||||
MsgStatusPhaseDefault: "Phase: Processing...",
|
||||
MsgCancelledReply: "✋ Cancelled. What would you like to do next?",
|
||||
|
||||
// Knowledge Graph
|
||||
MsgEntityIDRequired: "entity_id is required",
|
||||
MsgEntityFieldsRequired: "external_id, name, and entity_type are required",
|
||||
|
||||
@@ -122,6 +122,16 @@ func init() {
|
||||
MsgNoPendingOAuth: "không có luồng OAuth đang chờ",
|
||||
MsgFailedToSaveToken: "không thể lưu token",
|
||||
|
||||
// Intent Classify
|
||||
MsgStatusWorking: "🔄 Mình đang xử lý yêu cầu của bạn... Vui lòng chờ.",
|
||||
MsgStatusDetailed: "🔄 Mình đang xử lý yêu cầu của bạn...\n%s (lần lặp %d)\nĐã chạy: %s\n\nVui lòng chờ — mình sẽ phản hồi khi xong.",
|
||||
MsgStatusPhaseThinking: "Giai đoạn: Đang suy nghĩ...",
|
||||
MsgStatusPhaseToolExec: "Giai đoạn: Đang chạy %s",
|
||||
MsgStatusPhaseTools: "Giai đoạn: Đang thực thi công cụ...",
|
||||
MsgStatusPhaseCompact: "Giai đoạn: Đang nén ngữ cảnh...",
|
||||
MsgStatusPhaseDefault: "Giai đoạn: Đang xử lý...",
|
||||
MsgCancelledReply: "✋ Đã hủy. Bạn muốn làm gì tiếp?",
|
||||
|
||||
// Knowledge Graph
|
||||
MsgEntityIDRequired: "entity_id là bắt buộc",
|
||||
MsgEntityFieldsRequired: "external_id, name và entity_type là bắt buộc",
|
||||
|
||||
@@ -122,6 +122,16 @@ func init() {
|
||||
MsgNoPendingOAuth: "没有待处理的 OAuth 流程",
|
||||
MsgFailedToSaveToken: "保存令牌失败",
|
||||
|
||||
// Intent Classify
|
||||
MsgStatusWorking: "🔄 我正在处理您的请求...请稍候。",
|
||||
MsgStatusDetailed: "🔄 我正在处理您的请求...\n%s(第 %d 次迭代)\n已运行:%s\n\n请稍候——完成后我会回复您。",
|
||||
MsgStatusPhaseThinking: "阶段:思考中...",
|
||||
MsgStatusPhaseToolExec: "阶段:正在运行 %s",
|
||||
MsgStatusPhaseTools: "阶段:执行工具中...",
|
||||
MsgStatusPhaseCompact: "阶段:压缩上下文中...",
|
||||
MsgStatusPhaseDefault: "阶段:处理中...",
|
||||
MsgCancelledReply: "✋ 已取消。您接下来想做什么?",
|
||||
|
||||
// Knowledge Graph
|
||||
MsgEntityIDRequired: "entity_id 是必填项",
|
||||
MsgEntityFieldsRequired: "external_id、name 和 entity_type 是必填项",
|
||||
|
||||
@@ -123,6 +123,16 @@ const (
|
||||
MsgNoPendingOAuth = "error.no_pending_oauth" // "no pending OAuth flow"
|
||||
MsgFailedToSaveToken = "error.failed_to_save_token" // "failed to save token"
|
||||
|
||||
// --- Intent Classify (channel-facing status replies) ---
|
||||
MsgStatusWorking = "status.working" // "🔄 I'm working on your request... Please wait."
|
||||
MsgStatusDetailed = "status.detailed" // "🔄 I'm currently working on your request...\n%s (iteration %d)\nRunning for: %s\n\nPlease wait — I'll respond when done."
|
||||
MsgStatusPhaseThinking = "status.phase_thinking" // "Phase: Thinking..."
|
||||
MsgStatusPhaseToolExec = "status.phase_tool_exec" // "Phase: Running %s"
|
||||
MsgStatusPhaseTools = "status.phase_tools" // "Phase: Executing tools..."
|
||||
MsgStatusPhaseCompact = "status.phase_compact" // "Phase: Compacting context..."
|
||||
MsgStatusPhaseDefault = "status.phase_default" // "Phase: Processing..."
|
||||
MsgCancelledReply = "status.cancelled" // "✋ Cancelled. What would you like to do next?"
|
||||
|
||||
// --- Knowledge Graph ---
|
||||
MsgEntityIDRequired = "error.entity_id_required" // "entity_id is required"
|
||||
MsgEntityFieldsRequired = "error.entity_fields_required" // "external_id, name, and entity_type are required"
|
||||
|
||||
@@ -47,10 +47,32 @@ type DelegationTask struct {
|
||||
TeamID uuid.UUID `json:"-"` // from link.TeamID (for delegation history)
|
||||
TeamTaskID uuid.UUID `json:"-"`
|
||||
|
||||
// Activity tracking (updated via UpdateActivity on agent.activity events)
|
||||
LastActivity string `json:"-"` // "thinking", "tool_exec", "compacting"
|
||||
LastTool string `json:"-"` // current tool name (when LastActivity == "tool_exec")
|
||||
activityMu sync.RWMutex `json:"-"`
|
||||
|
||||
cancelFunc context.CancelFunc `json:"-"`
|
||||
progressEnabled bool `json:"-"` // resolved from team settings or global default
|
||||
}
|
||||
|
||||
// UpdateActivity sets the current phase and tool for this delegation.
|
||||
func (t *DelegationTask) UpdateActivity(phase, tool string) {
|
||||
t.activityMu.Lock()
|
||||
t.LastActivity = phase
|
||||
t.LastTool = tool
|
||||
t.activityMu.Unlock()
|
||||
}
|
||||
|
||||
// GetActivity returns the current phase and tool for this delegation.
|
||||
func (t *DelegationTask) GetActivity() (phase, tool string) {
|
||||
t.activityMu.RLock()
|
||||
phase = t.LastActivity
|
||||
tool = t.LastTool
|
||||
t.activityMu.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
// originKey returns a composite key scoping this delegation to its origin conversation.
|
||||
// Used for sibling counting and artifact accumulation so that delegations from
|
||||
// different channels/chats are NOT treated as siblings of each other.
|
||||
@@ -207,3 +229,18 @@ func (dm *DelegateManager) SetProgressEnabled(enabled bool) {
|
||||
dm.progressEnabled = enabled
|
||||
}
|
||||
|
||||
// HandleActivityEvent updates the activity tracking for a delegation.
|
||||
// Called from the bus event subscriber when an agent.activity event arrives.
|
||||
// delegationID is extracted from the event's DelegationID field.
|
||||
func (dm *DelegateManager) HandleActivityEvent(delegationID, phase, tool string) {
|
||||
if delegationID == "" {
|
||||
return
|
||||
}
|
||||
// O(1) lookup: active map is keyed by delegationID.
|
||||
val, ok := dm.active.Load(delegationID)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
val.(*DelegationTask).UpdateActivity(phase, tool)
|
||||
}
|
||||
|
||||
@@ -272,10 +272,17 @@ func (dm *DelegateManager) sendProgressNotification(task *DelegationTask) {
|
||||
if t.TargetDisplayName != "" {
|
||||
label = fmt.Sprintf("%s (%s)", t.TargetDisplayName, t.TargetAgentKey)
|
||||
}
|
||||
lines = append(lines, fmt.Sprintf("- %s: %s", label, elapsed))
|
||||
// Include current activity if available
|
||||
phase, tool := t.GetActivity()
|
||||
activityStr := formatDelegateActivity(phase, tool)
|
||||
if activityStr != "" {
|
||||
lines = append(lines, fmt.Sprintf("- %s %s — %s", activityStr, label, elapsed))
|
||||
} else {
|
||||
lines = append(lines, fmt.Sprintf("- %s: %s", label, elapsed))
|
||||
}
|
||||
}
|
||||
|
||||
content := fmt.Sprintf("⏳ Your team is working on it...\n%s", strings.Join(lines, "\n"))
|
||||
content := fmt.Sprintf("🏗 Your team is working on it...\n%s", strings.Join(lines, "\n"))
|
||||
|
||||
dm.msgBus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: task.OriginChannel,
|
||||
@@ -290,11 +297,14 @@ func (dm *DelegateManager) sendProgressNotification(task *DelegationTask) {
|
||||
// Emit WS progress event alongside the outbound channel message.
|
||||
var progressItems []protocol.DelegationProgressItem
|
||||
for _, t := range active {
|
||||
phase, tool := t.GetActivity()
|
||||
item := protocol.DelegationProgressItem{
|
||||
DelegationID: t.ID,
|
||||
TargetAgentKey: t.TargetAgentKey,
|
||||
TargetDisplayName: t.TargetDisplayName,
|
||||
ElapsedMS: int(time.Since(t.CreatedAt).Milliseconds()),
|
||||
Activity: phase,
|
||||
Tool: tool,
|
||||
}
|
||||
if t.TeamTaskID != uuid.Nil {
|
||||
item.TeamTaskID = t.TeamTaskID.String()
|
||||
@@ -353,6 +363,29 @@ func (dm *DelegateManager) buildRunRequest(task *DelegationTask, message string)
|
||||
return req
|
||||
}
|
||||
|
||||
// formatDelegateActivity returns an emoji+label for the delegate's current phase.
|
||||
func formatDelegateActivity(phase, tool string) string {
|
||||
switch phase {
|
||||
case "thinking":
|
||||
return "💭"
|
||||
case "tool_exec":
|
||||
if strings.HasPrefix(tool, "web") {
|
||||
return "🔍"
|
||||
}
|
||||
if tool == "exec" {
|
||||
return "⚡"
|
||||
}
|
||||
if tool == "spawn" || tool == "delegate" {
|
||||
return "👥"
|
||||
}
|
||||
return "🔧"
|
||||
case "compacting":
|
||||
return "📦"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// resolveParentMedia loads image media files from the parent session's recent MediaRefs.
|
||||
func (dm *DelegateManager) resolveParentMedia(parentSessionKey string) []bus.MediaFile {
|
||||
history := dm.sessionStore.GetHistory(parentSessionKey)
|
||||
|
||||
@@ -76,6 +76,7 @@ const (
|
||||
AgentEventToolCall = "tool.call"
|
||||
AgentEventToolResult = "tool.result"
|
||||
AgentEventBlockReply = "block.reply"
|
||||
AgentEventActivity = "activity" // agent phase transitions: thinking, tool_exec, compacting
|
||||
)
|
||||
|
||||
// Chat event subtypes (in payload.type)
|
||||
|
||||
@@ -30,6 +30,8 @@ type DelegationProgressItem struct {
|
||||
TargetDisplayName string `json:"target_display_name,omitempty"`
|
||||
ElapsedMS int `json:"elapsed_ms"`
|
||||
TeamTaskID string `json:"team_task_id,omitempty"`
|
||||
Activity string `json:"activity,omitempty"` // "thinking", "tool_exec", "compacting"
|
||||
Tool string `json:"tool,omitempty"` // current tool name (when Activity == "tool_exec")
|
||||
}
|
||||
|
||||
// DelegationProgressPayload is emitted periodically for async delegations.
|
||||
|
||||
@@ -147,6 +147,14 @@ export function AgentsDefaultsSection({ data, onSave, saving }: Props) {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="flex items-center justify-between">
|
||||
<InfoLabel tip={t("agents.intentClassifyTip")}>{t("agents.intentClassify")}</InfoLabel>
|
||||
<Switch
|
||||
checked={defaults.intent_classify !== false}
|
||||
onCheckedChange={(v) => updateDefaults({ intent_classify: v })}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* Collapsible sub-sections */}
|
||||
<SubSection
|
||||
title={t("agents.subagents.title")}
|
||||
|
||||
@@ -20,6 +20,7 @@ interface GatewayData {
|
||||
injection_action?: string;
|
||||
inbound_debounce_ms?: number;
|
||||
block_reply?: boolean;
|
||||
tool_status?: boolean;
|
||||
}
|
||||
|
||||
const DEFAULT: GatewayData = {};
|
||||
@@ -169,6 +170,12 @@ export function GatewaySection({ data, onSave, saving }: Props) {
|
||||
<Switch checked={draft.block_reply ?? false} onCheckedChange={(v) => update({ block_reply: v })} />
|
||||
</div>
|
||||
</div>
|
||||
<div className="grid gap-1.5">
|
||||
<InfoLabel tip={t("gateway.toolStatusTip")}>{t("gateway.toolStatus")}</InfoLabel>
|
||||
<div className="flex items-center h-9">
|
||||
<Switch checked={draft.tool_status !== false} onCheckedChange={(v) => update({ tool_status: v })} />
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{dirty && (
|
||||
|
||||
@@ -29,6 +29,8 @@ export interface DelegationProgressItem {
|
||||
target_display_name?: string;
|
||||
elapsed_ms: number;
|
||||
team_task_id?: string;
|
||||
activity?: string; // "thinking", "tool_exec", "compacting"
|
||||
tool?: string; // current tool name (when activity == "tool_exec")
|
||||
}
|
||||
|
||||
export interface DelegationProgressPayload {
|
||||
|
||||
Reference in new issue
Block a user