diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 8cefdfea7..5ea4a2089 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -479,12 +479,13 @@ func (al *AgentLoop) llmWorker(ctx context.Context, queue <-chan bus.InboundMess continue } - al.llmWorkerNormal(ctx, msg) + al.llmWorkerNormal(ctx, msg, queue) } } // llmWorkerNormal processes a single non-PDF message. -func (al *AgentLoop) llmWorkerNormal(ctx context.Context, msg bus.InboundMessage) { +// queue is optional (nil when called outside Run loop). +func (al *AgentLoop) llmWorkerNormal(ctx context.Context, msg bus.InboundMessage, queue <-chan bus.InboundMessage) { al.activeRequests.Add(1) defer al.activeRequests.Done() @@ -508,9 +509,66 @@ func (al *AgentLoop) llmWorkerNormal(ctx context.Context, msg bus.InboundMessage response = fmt.Sprintf("Error processing message: %v", err) } + // Auto-continue: if inbound messages arrived in the queue while this + // turn was running, treat them as steering continuations so the agent + // responds with the full context instead of two separate turns. + if queue != nil { + if drained := al.drainQueueAsSteering(queue, msg); len(drained) > 0 { + agent := al.agentForSession(msg.SessionKey) + if agent == nil { + agent = al.registry.GetDefaultAgent() + } + if agent != nil { + sessionKey := msg.SessionKey + if sessionKey == "" { + route, _, _ := al.resolveMessageRoute(msg) + sessionKey = resolveScopeKey(route, msg.SessionKey) + } + contResp, contErr := al.continueWithSteeringMessages( + ctx, agent, sessionKey, msg.Channel, msg.ChatID, drained, + ) + if contErr == nil && contResp != "" { + response = contResp + } + } + } + } + al.sendResponseIfNeeded(ctx, msg, response) } +// drainQueueAsSteering non-blocking drains pending messages from the +// llmQueue that belong to the same chat as the original message and +// returns them as steering-style provider messages. +func (al *AgentLoop) drainQueueAsSteering( + queue <-chan bus.InboundMessage, + orig bus.InboundMessage, +) []providers.Message { + var msgs []providers.Message + for { + select { + case m, ok := <-queue: + if !ok { + return msgs + } + if m.Channel == orig.Channel && m.ChatID == orig.ChatID { + msgs = append(msgs, providers.Message{ + Role: "user", + Content: m.Content, + }) + } else { + // Re-queue by pushing to steering for later processing + al.enqueueSteeringMessage("", "", providers.Message{ + Role: "user", + Content: m.Content, + }) + } + default: + return msgs + } + } +} + // llmWorkerPDF handles a bare-PDF message with two-phase follow-up collection. func (al *AgentLoop) llmWorkerPDF(ctx context.Context, msg bus.InboundMessage, queue <-chan bus.InboundMessage) { // Phase 1: wait for OCR keywords (figures/図版) — up to 5 seconds. @@ -553,13 +611,13 @@ func (al *AgentLoop) llmWorkerPDF(ctx context.Context, msg bus.InboundMessage, q Content: followUpText, Metadata: msg.Metadata, } - al.llmWorkerNormal(ctx, followUpMsg) + al.llmWorkerNormal(ctx, followUpMsg, nil) } // Re-queue messages from other chats that were buffered. otherMsgs := extractNonChatMessages(buffered, msg.ChatID) for _, other := range otherMsgs { - al.llmWorkerNormal(ctx, other) + al.llmWorkerNormal(ctx, other, nil) } } @@ -1179,7 +1237,31 @@ func (al *AgentLoop) ProcessDirectWithChannel( }, } - return al.processMessage(ctx, msg) + response, err := al.processMessage(ctx, msg) + if err != nil { + return response, err + } + + // If steering messages arrived during the LLM call, the initial + // response is stale. Discard it and do a continuation turn that + // includes the steering messages for a fresh response. + steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(sessionKey) + if len(steeringMsgs) > 0 { + agent := al.agentForSession(sessionKey) + if agent == nil { + agent = al.registry.GetDefaultAgent() + } + if agent != nil { + contResp, contErr := al.continueWithSteeringMessages( + ctx, agent, sessionKey, channel, chatID, steeringMsgs, + ) + if contErr == nil && contResp != "" { + return contResp, nil + } + } + } + + return response, nil } // ProcessHeartbeat processes a heartbeat request without session history. diff --git a/pkg/agent/loop_commands.go b/pkg/agent/loop_commands.go index 151d11d50..e452b3098 100644 --- a/pkg/agent/loop_commands.go +++ b/pkg/agent/loop_commands.go @@ -62,6 +62,14 @@ func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance, sessionKey strin } old := agent.Model agent.Model = value + // Rebuild candidates so subsequent LLM calls use the new + // model's provider/endpoint instead of the old one. + agent.Candidates = resolveModelCandidates( + cfg, + cfg.Agents.Defaults.Provider, + value, + nil, + ) return old, nil }, SwitchChannel: func(value string) error { diff --git a/pkg/agent/loop_run.go b/pkg/agent/loop_run.go index df8a9da4f..442ee44f0 100644 --- a/pkg/agent/loop_run.go +++ b/pkg/agent/loop_run.go @@ -33,6 +33,10 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt scope := al.newTurnEventScope(agent.ID, opts.SessionKey) turnStart := time.Now() + // Create a cancelable context for hard abort support + turnCtx, turnCancelFn := context.WithCancel(ctx) + defer turnCancelFn() + // Register a turnState so the interrupt API can find this turn ts := &turnState{ turnID: scope.turnID, @@ -44,6 +48,12 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt phase: TurnPhaseRunning, startedAt: turnStart, agent: agent, + turnCancel: turnCancelFn, + } + // Bind session store and capture initial history length for rollback + if agent.Sessions != nil { + ts.session = agent.Sessions + ts.initialHistoryLength = len(agent.Sessions.GetHistory(opts.SessionKey)) } al.registerActiveTurn(ts) defer al.clearActiveTurn(ts) @@ -65,7 +75,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt // -0. Create cancelable child context and register active task - taskCtx, taskCancel := context.WithCancel(ctx) + taskCtx, taskCancel := context.WithCancel(turnCtx) defer taskCancel() @@ -354,7 +364,9 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt historyMsg = opts.HistoryMessage } - agent.Sessions.AddMessage(opts.SessionKey, "user", historyMsg) + if historyMsg != "" { + agent.Sessions.AddMessage(opts.SessionKey, "user", historyMsg) + } // 4. Record user prompt for stats @@ -413,6 +425,26 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt finalContent, iteration, err = al.runLLMIteration(ctx, agent, messages, opts, task, curPlanStatus, scope) if err != nil { + // Check for hard abort: if the turn was hard-aborted, restore + // session history and return empty response. + ts.mu.RLock() + isHardAbort := ts.hardAbort + ts.mu.RUnlock() + if isHardAbort { + if ts.session != nil { + history := ts.session.GetHistory(opts.SessionKey) + if ts.initialHistoryLength < len(history) { + ts.session.SetHistory(opts.SessionKey, history[:ts.initialHistoryLength]) + } + } + al.emitEvent(EventKindTurnEnd, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + TurnEndPayload{ + Status: TurnEndStatusAborted, + Duration: time.Since(turnStart), + }) + return "", nil + } return "", err } @@ -868,10 +900,24 @@ func (al *AgentLoop) runLLMIteration( // Also inject any initial steering messages from the process options if len(opts.InitialSteeringMessages) > 0 { - messages = append(messages, opts.InitialSteeringMessages...) + // Persist original refs in session history + for _, sm := range opts.InitialSteeringMessages { + agent.Sessions.AddFullMessage(opts.SessionKey, sm) + } + // Resolve media refs for the provider call while keeping + // the originals in session history with raw refs. + cfg := al.GetConfig() + maxMedia := cfg.Agents.Defaults.GetMaxMediaSize() + resolved := resolveMediaRefs(opts.InitialSteeringMessages, al.mediaStore, maxMedia) + messages = append(messages, resolved...) } for iteration < agent.MaxIterations { + // Check for context cancellation (e.g. hard abort) before each iteration + if ctx.Err() != nil { + return "", iteration, ctx.Err() + } + iteration++ if msg := hooks.OnIterationStart(iteration); msg != "" { @@ -1014,7 +1060,11 @@ func (al *AgentLoop) runLLMIteration( HasReasoning: response.Reasoning != "", }) - go al.handleReasoning(ctx, response.Reasoning, opts.Channel, al.targetReasoningChannelID(opts.Channel)) + reasoningText := response.Reasoning + if reasoningText == "" { + reasoningText = response.ReasoningContent + } + go al.handleReasoning(ctx, reasoningText, opts.Channel, al.targetReasoningChannelID(opts.Channel)) logger.DebugCF("agent", "LLM response", map[string]any{ @@ -1090,12 +1140,36 @@ func (al *AgentLoop) runLLMIteration( agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg) // Execute tool calls and collect results - lastBlocker := al.executeToolCalls(ctx, agent, normalizedToolCalls, &messages, opts, hooks, iteration, scope) + execResult := al.executeToolCalls(ctx, agent, normalizedToolCalls, &messages, opts, hooks, iteration, scope) + + // Graceful interrupt: make a terminal LLM call with no tool + // definitions so the agent can produce a final summary. + if execResult.gracefulInterrupt { + hintMsg := "Interrupt requested. Stop scheduling tools and provide a short final summary." + if execResult.gracefulHint != "" { + hintMsg += "\n\nInterrupt hint: " + execResult.gracefulHint + } + messages = append(messages, providers.Message{ + Role: "user", + Content: hintMsg, + }) + + terminalResp, termErr := al.callLLMWithRetry( + ctx, agent, &messages, opts, + nil, // no tool definitions + al.selectCandidates(agent, "", messages), + agent.Model, nil, iteration+1, scope, + ) + if termErr == nil && terminalResp != nil { + finalContent = terminalResp.Content + } + break + } // Tick TTL-based tool expiry after execution agent.Tools.TickTTL() - hooks.InjectReminders(iteration, &messages, lastBlocker) + hooks.InjectReminders(iteration, &messages, execResult.lastBlocker) hooks.RefreshSystemPrompt(messages) } @@ -1107,6 +1181,13 @@ func (al *AgentLoop) runLLMIteration( return finalContent, iteration, nil } +// toolExecResult holds results from executeToolCalls. +type toolExecResult struct { + lastBlocker string + gracefulInterrupt bool + gracefulHint string +} + // executeToolCalls runs each tool call sequentially, publishes results, // and returns the last blocker (error content) for reminder injection. func (al *AgentLoop) executeToolCalls( @@ -1118,12 +1199,26 @@ func (al *AgentLoop) executeToolCalls( hooks iterationHooks, iteration int, scope turnEventScope, -) string { - var lastBlocker string +) toolExecResult { + var result toolExecResult steered := false + gracefulSkip := false for i, tc := range toolCalls { + // Check for graceful interrupt between tool calls + if !gracefulSkip { + if ts := al.getActiveTurnState(opts.SessionKey); ts != nil { + ts.mu.RLock() + if ts.gracefulInterrupt { + gracefulSkip = true + result.gracefulInterrupt = true + result.gracefulHint = ts.gracefulInterruptHint + } + ts.mu.RUnlock() + } + } + // Check for pending steering messages between tool calls - if i > 0 && !steered { + if i > 0 && !steered && !gracefulSkip { steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(opts.SessionKey) if len(steeringMsgs) > 0 { steered = true @@ -1149,6 +1244,16 @@ func (al *AgentLoop) executeToolCalls( } } + // Skip remaining tools if graceful interrupt was requested + if gracefulSkip { + *messages = append(*messages, providers.Message{ + Role: "tool", + Content: "Skipped due to graceful interrupt.", + ToolCallID: tc.ID, + }) + continue + } + // Skip remaining tools if steering was injected if steered { skipMeta := EventMeta{ @@ -1312,7 +1417,7 @@ func (al *AgentLoop) executeToolCalls( contentForLLM = toolResult.Err.Error() } if toolResult.IsError || toolResult.Err != nil { - lastBlocker = contentForLLM + result.lastBlocker = contentForLLM } toolResultMsg := providers.Message{ @@ -1323,7 +1428,21 @@ func (al *AgentLoop) executeToolCalls( *messages = append(*messages, toolResultMsg) agent.Sessions.AddFullMessage(opts.SessionKey, toolResultMsg) } - return lastBlocker + + // Re-check graceful interrupt after all tool execution (may have been + // set during the last tool's execution). + if !result.gracefulInterrupt { + if ts := al.getActiveTurnState(opts.SessionKey); ts != nil { + ts.mu.RLock() + if ts.gracefulInterrupt { + result.gracefulInterrupt = true + result.gracefulHint = ts.gracefulInterruptHint + } + ts.mu.RUnlock() + } + } + + return result } // forceTextResponse makes a final LLM call without tools when max iterations