diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index e167e85d..417ea130 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -243,10 +243,9 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents } // Intent classify fast-path: when agent is busy on DM, classify user intent - // to detect status queries, cancel requests, etc. without queueing. + // to detect status queries, cancel requests, or steer/new_task for mid-run injection. // 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 maxConcurrent == 1 && agents.IsSessionBusy(sessionKey) { if loop, ok := agentLoop.(*agent.Loop); ok && loop.Provider() != nil { locale := msg.Metadata["locale"] if locale == "" { @@ -277,8 +276,26 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } return - default: - // steer / new_task → queue as normal + case agent.IntentSteer, agent.IntentNewTask: + // Mid-run injection: inject into the running loop instead of queueing. + injected := agents.InjectMessage(sessionKey, agent.InjectedMessage{ + Content: msg.Content, + UserID: userID, + }) + if injected { + slog.Info("inbound: injected mid-run message", + "intent", string(intent), "session", sessionKey) + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: msg.Channel, + ChatID: msg.ChatID, + Content: i18n.T(locale, i18n.MsgInjectedAck), + Metadata: outMeta, + }) + return + } + // Fallback: injection failed (channel full) → fall through to scheduler queue + slog.Info("inbound: injection failed, queueing as normal", + "intent", string(intent), "session", sessionKey) } } } diff --git a/cmd/gateway_consumer_process.go b/cmd/gateway_consumer_process.go index 4b5f5439..d699e4bd 100644 --- a/cmd/gateway_consumer_process.go +++ b/cmd/gateway_consumer_process.go @@ -40,10 +40,11 @@ func makeSchedulerRunFunc(agents *agent.Router, cfg *config.Config) scheduler.Ru // 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) + injectCh := agents.RegisterRun(req.RunID, req.SessionKey, agentID, cancel) defer agents.UnregisterRun(req.RunID) defer cancel() + req.InjectCh = injectCh return loop.Run(runCtx, req) } } diff --git a/internal/agent/inject.go b/internal/agent/inject.go new file mode 100644 index 00000000..9d8adeeb --- /dev/null +++ b/internal/agent/inject.go @@ -0,0 +1,108 @@ +package agent + +import ( + "fmt" + "log/slog" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// InjectedMessage represents a user message injected into a running agent loop +// at the turn boundary (after tool results, before next LLM call). +type InjectedMessage struct { + Content string + UserID string +} + +// processedInjection holds the two message forms: one for the LLM (with context wrapper) +// and one for session persistence (original content). +type processedInjection struct { + forLLM providers.Message // wrapped with "[User sent a follow-up...]" prefix + forSession providers.Message // original content for session history +} + +// injectBufferSize is the capacity of the per-run injection channel. +const injectBufferSize = 5 + +// processInjectedMessage validates and wraps an injected message for the LLM. +// Returns nil, false if the message should be skipped (blocked by input guard). +func (l *Loop) processInjectedMessage(injected InjectedMessage, emitRun func(AgentEvent)) (*processedInjection, bool) { + // Security: scan injected content with input guard + if l.inputGuard != nil { + if matches := l.inputGuard.Scan(injected.Content); len(matches) > 0 { + matchStr := strings.Join(matches, ",") + if l.injectionAction == "block" { + slog.Warn("security.injection_blocked_midrun", + "agent", l.id, "user", injected.UserID, + "patterns", matchStr) + return nil, false + } + slog.Warn("security.injection_detected_midrun", + "agent", l.id, "user", injected.UserID, + "patterns", matchStr) + } + } + + // Truncate oversized content + content := injected.Content + maxChars := l.maxMessageChars + if maxChars <= 0 { + maxChars = 32000 + } + if len(content) > maxChars { + content = content[:maxChars] + "\n[Message truncated]" + } + + // Wrap with context hint so LLM knows this is a mid-run follow-up + wrapped := fmt.Sprintf("[User sent a follow-up message while you were working]\n%s", content) + + slog.Info("mid-run injection", + "agent", l.id, "user", injected.UserID, + "msg_len", len(content)) + + // Emit activity event so UI/channels know about the injection + if emitRun != nil { + emitRun(AgentEvent{ + Type: protocol.AgentEventActivity, + AgentID: l.id, + Payload: map[string]any{ + "phase": "injected_message", + "content": truncateForLog(content, 200), + }, + }) + } + + return &processedInjection{ + forLLM: providers.Message{Role: "user", Content: wrapped}, + forSession: providers.Message{Role: "user", Content: content}, + }, true +} + +// drainInjectChannel reads all available messages from the injection channel +// without blocking. Returns processed messages ready to append to the loop. +func (l *Loop) drainInjectChannel(ch <-chan InjectedMessage, emitRun func(AgentEvent)) (forLLM, forSession []providers.Message) { + if ch == nil { + return nil, nil + } + for { + select { + case injected := <-ch: + if result, ok := l.processInjectedMessage(injected, emitRun); ok { + forLLM = append(forLLM, result.forLLM) + forSession = append(forSession, result.forSession) + } + default: + return forLLM, forSession + } + } +} + +// truncateForLog truncates a string for log/event payloads. +func truncateForLog(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + return s[:maxLen] + "..." +} diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 4920c909..910dbc19 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -582,6 +582,17 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } } + // Mid-run injection (Point B): drain all buffered user follow-up messages + // before exiting. If found, save current assistant response and continue + // the loop so the LLM can respond to the injected messages. + if forLLM, forSession := l.drainInjectChannel(req.InjectCh, emitRun); len(forLLM) > 0 { + messages = append(messages, providers.Message{Role: "assistant", Content: resp.Content}) + messages = append(messages, forLLM...) + pendingMsgs = append(pendingMsgs, providers.Message{Role: "assistant", Content: resp.Content}) + pendingMsgs = append(pendingMsgs, forSession...) + continue + } + finalContent = resp.Content finalThinking = resp.Thinking break @@ -912,6 +923,14 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) break } } + + // Mid-run injection (Point A): drain any user follow-up messages + // that arrived during tool execution. Append them after tool results + // so the next LLM call sees: [tool results...] + [user follow-ups...]. + if forLLM, forSession := l.drainInjectChannel(req.InjectCh, emitRun); len(forLLM) > 0 { + messages = append(messages, forLLM...) + pendingMsgs = append(pendingMsgs, forSession...) + } } // 4. Full sanitization pipeline (matching TS extractAssistantText + sanitizeUserFacingText) diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index ab9d5c98..43c8c406 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -334,6 +334,11 @@ type RunRequest struct { HideInput bool // don't persist input message in session history (announce runs) ContentSuffix string // appended to assistant response before saving (e.g. image markdown for WS) + // Mid-run message injection channel (nil = disabled). + // When set, the loop drains this channel at turn boundaries to inject + // user follow-up messages into the running conversation. + InjectCh <-chan InjectedMessage + // Delegation context (set when running as a delegate agent) DelegationID string // delegation ID for event correlation TeamID string // team ID (if delegation is team-scoped) diff --git a/internal/agent/router.go b/internal/agent/router.go index 118d7f4a..f3e251b7 100644 --- a/internal/agent/router.go +++ b/internal/agent/router.go @@ -153,25 +153,31 @@ func (r *Router) IsRunning(agentID string) bool { // --- Active Run Tracking (matching TS chat-abort.ts) --- -// ActiveRun tracks a running agent invocation so it can be aborted via chat.abort. +// ActiveRun tracks a running agent invocation so it can be aborted via chat.abort +// and supports mid-run message injection via InjectCh. type ActiveRun struct { RunID string SessionKey string AgentID string Cancel context.CancelFunc StartedAt time.Time + InjectCh chan InjectedMessage // buffered channel for mid-run user message injection } // RegisterRun records an active run so it can be aborted later. -func (r *Router) RegisterRun(runID, sessionKey, agentID string, cancel context.CancelFunc) { +// Returns a receive-only channel for mid-run message injection. +func (r *Router) RegisterRun(runID, sessionKey, agentID string, cancel context.CancelFunc) <-chan InjectedMessage { + injectCh := make(chan InjectedMessage, injectBufferSize) r.activeRuns.Store(runID, &ActiveRun{ RunID: runID, SessionKey: sessionKey, AgentID: agentID, Cancel: cancel, StartedAt: time.Now(), + InjectCh: injectCh, }) r.sessionRuns.Store(sessionKey, runID) + return injectCh } // UnregisterRun removes a completed/cancelled run from tracking. @@ -204,6 +210,26 @@ func (r *Router) AbortRun(runID, sessionKey string) bool { return true } +// InjectMessage sends a user message to the running loop for a session. +// Returns true if the message was accepted, false if no active run or channel full. +func (r *Router) InjectMessage(sessionKey string, msg InjectedMessage) bool { + runIDVal, ok := r.sessionRuns.Load(sessionKey) + if !ok { + return false + } + runVal, ok := r.activeRuns.Load(runIDVal) + if !ok { + return false + } + run := runVal.(*ActiveRun) + select { + case run.InjectCh <- msg: + return true + default: + return false // channel full + } +} + // InvalidateUserWorkspace clears the cached workspace for a user across all cached agent loops. // Used when user_agent_profiles.workspace changes (e.g. admin reassignment). func (r *Router) InvalidateUserWorkspace(userID string) { diff --git a/internal/config/config.go b/internal/config/config.go index 0a097fc9..73b80b28 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -124,10 +124,6 @@ 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. diff --git a/internal/gateway/methods/chat.go b/internal/gateway/methods/chat.go index fbc875ae..6cda9c1d 100644 --- a/internal/gateway/methods/chat.go +++ b/internal/gateway/methods/chat.go @@ -128,9 +128,25 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re runCtxBase = store.WithUserID(runCtxBase, userID) } + // Mid-run injection: if session already has an active run, inject the message + // into the running loop instead of starting a new concurrent run. + if m.agents.IsSessionBusy(sessionKey) { + injected := m.agents.InjectMessage(sessionKey, agent.InjectedMessage{ + Content: params.Message, + UserID: userID, + }) + if injected { + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "injected": true, + })) + return + } + // Fallback: injection failed (channel full), proceed with new run + } + // Create cancellable context for abort support (matching TS AbortController pattern). runCtx, cancel := context.WithCancel(runCtxBase) - m.agents.RegisterRun(runID, sessionKey, params.AgentID, cancel) + injectCh := m.agents.RegisterRun(runID, sessionKey, params.AgentID, cancel) // Run agent asynchronously - events are broadcast via the event system go func() { @@ -175,6 +191,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re RunID: runID, UserID: userID, Stream: params.Stream, + InjectCh: injectCh, }) if err != nil { diff --git a/internal/i18n/catalog_en.go b/internal/i18n/catalog_en.go index 51478852..346a0321 100644 --- a/internal/i18n/catalog_en.go +++ b/internal/i18n/catalog_en.go @@ -131,6 +131,7 @@ func init() { MsgStatusPhaseCompact: "Phase: Compacting context...", MsgStatusPhaseDefault: "Phase: Processing...", MsgCancelledReply: "✋ Cancelled. What would you like to do next?", + MsgInjectedAck: "Got it, I'll incorporate that into what I'm working on.", // Knowledge Graph MsgEntityIDRequired: "entity_id is required", diff --git a/internal/i18n/catalog_vi.go b/internal/i18n/catalog_vi.go index 5c23c41d..76666b17 100644 --- a/internal/i18n/catalog_vi.go +++ b/internal/i18n/catalog_vi.go @@ -131,6 +131,7 @@ func init() { 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?", + MsgInjectedAck: "Đã nhận, tôi sẽ xử lý trong tác vụ hiện tại.", // Knowledge Graph MsgEntityIDRequired: "entity_id là bắt buộc", diff --git a/internal/i18n/catalog_zh.go b/internal/i18n/catalog_zh.go index 00f4d8ef..524779cb 100644 --- a/internal/i18n/catalog_zh.go +++ b/internal/i18n/catalog_zh.go @@ -131,6 +131,7 @@ func init() { MsgStatusPhaseCompact: "阶段：压缩上下文中...", MsgStatusPhaseDefault: "阶段：处理中...", MsgCancelledReply: "✋ 已取消。您接下来想做什么？", + MsgInjectedAck: "收到，我会在当前任务中处理。", // Knowledge Graph MsgEntityIDRequired: "entity_id 是必填项", diff --git a/internal/i18n/keys.go b/internal/i18n/keys.go index 390aaa84..dd32f34b 100644 --- a/internal/i18n/keys.go +++ b/internal/i18n/keys.go @@ -132,6 +132,7 @@ const ( MsgStatusPhaseCompact = "status.phase_compact" // "Phase: Compacting context..." MsgStatusPhaseDefault = "status.phase_default" // "Phase: Processing..." MsgCancelledReply = "status.cancelled" // "✋ Cancelled. What would you like to do next?" + MsgInjectedAck = "status.injected_ack" // "Got it, I'll incorporate that into what I'm working on." // --- Knowledge Graph --- MsgEntityIDRequired = "error.entity_id_required" // "entity_id is required" diff --git a/ui/web/src/components/chat/chat-input.tsx b/ui/web/src/components/chat/chat-input.tsx index a5fd070f..730510f9 100644 --- a/ui/web/src/components/chat/chat-input.tsx +++ b/ui/web/src/components/chat/chat-input.tsx @@ -37,11 +37,10 @@ export function ChatInput({ onSend, onAbort, isRunning, disabled }: ChatInputPro (e: KeyboardEvent) => { if (e.key === "Enter" && !e.shiftKey) { e.preventDefault(); - if (isRunning) return; handleSend(); } }, - [handleSend, isRunning], + [handleSend], ); const handleInput = useCallback(() => { @@ -127,14 +126,24 @@ export function ChatInput({ onSend, onAbort, isRunning, disabled }: ChatInputPro className="flex-1 resize-none rounded-lg border bg-background px-4 py-2.5 text-base md:text-sm placeholder:text-muted-foreground focus:outline-none focus:ring-1 focus:ring-ring disabled:opacity-50" /> {isRunning ? ( - +
+ + +
) : (