feat(agent): add mid-run message injection for DM and WebSocket

Inject user follow-up messages into the running agent loop at turn
boundaries instead of queueing them for a new run. This preserves
context so the LLM sees both tool results and user follow-ups together.

- Add InjectedMessage type and drainInjectChannel helper
- Add InjectCh to ActiveRun with buffered channel (cap=5)
- Drain injection channel at two points in agent loop (after tool
  results and before no-tool-calls exit)
- Route steer/new_task intents to InjectMessage with scheduler fallback
- WebSocket: inject into running loop when session is busy
- Remove IntentClassify config toggle (always on)
- Web UI: show send + stop buttons side by side during agent run
- i18n: add injection acknowledgment messages (en/vi/zh)
This commit is contained in:
viettranx committed 2026-03-13 11:55:55 +07:00
1 parent d7ce93d91d
commit 4c7db6e09b
17 files changed
+240 -26

No files matched your search

+22 -5
View File
@@ -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)
}
}
}
+2 -1
View File
@@ -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)
}
}
+108
View File
@@ -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] + "..."
}
+19
View File
@@ -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)
+5
View File
@@ -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)
+28 -2
View File
@@ -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) {
-4
View File
@@ -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.
+18 -1
View File
@@ -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 {
+1
View File
@@ -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",
+1
View File
@@ -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",
+1
View File
@@ -131,6 +131,7 @@ func init() {
MsgStatusPhaseCompact: "阶段:压缩上下文中...",
MsgStatusPhaseDefault: "阶段:处理中...",
MsgCancelledReply: "✋ 已取消。您接下来想做什么?",
MsgInjectedAck: "收到,我会在当前任务中处理。",
// Knowledge Graph
MsgEntityIDRequired: "entity_id 是必填项",
+1
View File
@@ -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"
+19 -10
View File
@@ -37,11 +37,10 @@ export function ChatInput({ onSend, onAbort, isRunning, disabled }: ChatInputPro
(e: KeyboardEvent<HTMLTextAreaElement>) => {
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 ? (
<Button
variant="destructive"
size="icon-lg"
onClick={onAbort}
title={t("stopGeneration")}
>
<Square className="h-4 w-4" />
</Button>
<div className="flex gap-1">
<Button
size="icon-lg"
onClick={handleSend}
disabled={!value.trim() || disabled}
title={t("sendFollowUp")}
>
<Send className="h-4 w-4" />
</Button>
<Button
variant="destructive"
size="icon-lg"
onClick={onAbort}
title={t("stopGeneration")}
>
<Square className="h-4 w-4" />
</Button>
</div>
) : (
<Button
size="icon-lg"
+1
View File
@@ -76,6 +76,7 @@
"sendMessage": "Send a message... (Shift+Enter for new line)",
"stopGeneration": "Stop generation",
"sendMessageTitle": "Send message",
"sendFollowUp": "Send follow-up message",
"attachFile": "Attach file",
"thinking": "Thinking",
"thinkingStreaming": "Thinking...",
+1
View File
@@ -76,6 +76,7 @@
"selectProvider": "Chọn provider",
"sendMessage": "Gửi tin nhắn... (Shift+Enter để xuống dòng)",
"sendMessageTitle": "Gửi tin nhắn",
"sendFollowUp": "Gửi tin nhắn bổ sung",
"attachFile": "Đính kèm tệp",
"skillActivated": "Đã kích hoạt",
"skillActivating": "Đang kích hoạt...",
+1
View File
@@ -76,6 +76,7 @@
"selectProvider": "选择Provider",
"sendMessage": "发送消息... (Shift+Enter 换行)",
"sendMessageTitle": "发送消息",
"sendFollowUp": "发送后续消息",
"attachFile": "附加文件",
"skillActivated": "已激活",
"skillActivating": "激活中...",
+12 -3
View File
@@ -54,8 +54,6 @@ export function useChatSend({
timestamp: Date.now(),
});
onExpectRun();
try {
// Upload files first, then pass path+filename to chat.send
let mediaItems: { path: string; filename: string }[] | undefined;
@@ -70,7 +68,12 @@ export function useChatSend({
mediaItems = uploads.map((u) => ({ path: u.path, filename: u.filename }));
}
await ws.call<{ runId: string; content: string }>(
// Expect a new run BEFORE the call — run.started event fires during ws.call,
// not after it returns. For injected sends (agent already running), this is
// a no-op since events are already being listened to.
onExpectRun();
const res = await ws.call<{ runId?: string; content?: string; injected?: boolean }>(
Methods.CHAT_SEND,
{
agentId,
@@ -81,6 +84,12 @@ export function useChatSend({
},
600_000,
);
// If message was injected into running agent, no new run was started.
// The running agent will naturally process the injected message.
if (res?.injected) {
return;
}
} catch (err) {
setError(err instanceof Error ? err.message : "Failed to send message");
} finally {