diff --git a/docs/design/steering-spec.md b/docs/design/steering-spec.md index 7cd8dc452..0951bf864 100644 --- a/docs/design/steering-spec.md +++ b/docs/design/steering-spec.md @@ -19,30 +19,62 @@ The user's intent reaches the model **as soon as the current tool finishes**, no ```mermaid graph TD subgraph External Callers - CH[Channel Handler] - API[HTTP API] - WS[WebSocket] + TG[Telegram] + DC[Discord] + SL[Slack] end subgraph AgentLoop + BUS[MessageBus] + DRAIN[drainBusToSteering goroutine] SQ[steeringQueue] RLI[runLLMIteration] TE[Tool Execution Loop] LLM[LLM Call] end - CH -->|Steer| SQ - API -->|Steer| SQ - WS -->|Steer| SQ + TG -->|PublishInbound| BUS + DC -->|PublishInbound| BUS + SL -->|PublishInbound| BUS + + BUS -->|ConsumeInbound while busy| DRAIN + DRAIN -->|Steer| SQ RLI -->|1. initial poll| SQ TE -->|2. poll after each tool| SQ - TE -->|3. poll after last tool| SQ SQ -->|pendingMessages| RLI RLI -->|inject into context| LLM ``` +### Bus drain mechanism + +Channels (Telegram, Discord, etc.) publish messages to the `MessageBus` via `PublishInbound`. Without additional wiring, these messages would sit in the bus buffer until the current `processMessage` finishes — meaning steering would never work for real users. + +The solution: when `Run()` starts processing a message, it spawns a **drain goroutine** (`drainBusToSteering`) that keeps consuming from the bus and calling `Steer()`. When `processMessage` returns, the drain is canceled and normal consumption resumes. + +```mermaid +sequenceDiagram + participant Bus + participant Run + participant Drain + participant AgentLoop + + Run->>Bus: ConsumeInbound() → msg + Run->>Drain: spawn drainBusToSteering(ctx) + Run->>Run: processMessage(msg) + + Note over Drain: running concurrently + + Bus-->>Drain: ConsumeInbound() → newMsg + Drain->>AgentLoop: al.transcribeAudioInMessage(ctx, newMsg) + Drain->>AgentLoop: Steer(providers.Message{Content: newMsg.Content}) + + Run->>Run: processMessage returns + Run->>Drain: cancel context + Note over Drain: exits +``` + ## Data Structures ### steeringQueue @@ -59,7 +91,7 @@ A thread-safe FIFO queue, private to the `agent` package. | Method | Description | |--------|-------------| -| `push(msg)` | Appends a message to the queue | +| `push(msg) error` | Appends a message to the queue. Returns an error if the queue is full (`MaxQueueSize`) | | `dequeue() []Message` | Removes and returns messages according to `mode`. Returns `nil` if empty | | `len() int` | Returns the current queue length | | `setMode(mode)` | Updates the dequeue strategy | @@ -86,7 +118,7 @@ A new field was added to `processOptions`: | Method | Signature | Description | |--------|-----------|-------------| -| `Steer` | `Steer(msg providers.Message)` | Enqueues a steering message. Thread-safe, can be called from any goroutine. | +| `Steer` | `Steer(msg providers.Message) error` | Enqueues a steering message. Returns an error if the queue is full or not initialized. Thread-safe, can be called from any goroutine. | | `SteeringMode` | `SteeringMode() SteeringMode` | Returns the current dequeue mode. | | `SetSteeringMode` | `SetSteeringMode(mode SteeringMode)` | Changes the dequeue mode at runtime. | | `Continue` | `Continue(ctx, sessionKey, channel, chatID) (string, error)` | Resumes an idle agent using pending steering messages. Returns `""` if queue is empty. | @@ -134,24 +166,18 @@ sequenceDiagram LLM-->>runLLMIteration: response with toolCalls[0..N] loop for each tool call (sequential) - alt i > 0 - ToolExecution->>AgentLoop: dequeueSteeringMessages() - AgentLoop-->>ToolExecution: steeringMessages - - alt steering found - Note over ToolExecution: Mark tool[i..N] as
"Skipped due to queued user message." - ToolExecution-->>runLLMIteration: steeringAfterTools = steeringMessages - Note over ToolExecution: break out of tool loop - end - end - ToolExecution->>ToolExecution: execute tool[i] ToolExecution->>ToolExecution: process result,
append to messages[] - alt last tool (i == N-1) - ToolExecution->>AgentLoop: dequeueSteeringMessages() - AgentLoop-->>ToolExecution: steeringMessages (may be empty) + ToolExecution->>AgentLoop: dequeueSteeringMessages() + AgentLoop-->>ToolExecution: steeringMessages + + alt steering found + opt remaining tools > 0 + Note over ToolExecution: Mark tool[i+1..N-1] as
"Skipped due to queued user message." + end Note over ToolExecution: steeringAfterTools = steeringMessages + Note over ToolExecution: break out of tool loop end end @@ -168,12 +194,11 @@ sequenceDiagram | # | Location | When | Purpose | |---|----------|------|---------| | 1 | Top of `runLLMIteration`, before first LLM call | Once, at loop entry | Catch messages enqueued while the agent was still setting up context | -| 2 | Between tool calls, before tool `[i]` where `i > 0` | After each tool finishes | Interrupt mid-batch if the user sent a steering message | -| 3 | After the last tool in the batch | After tool `[N-1]` finishes | Catch messages that arrived during the last tool's execution | +| 2 | After every tool completes (including the first and the last) | Immediately after each tool's result is processed | Interrupt the batch as early as possible — if steering is found and there are remaining tools, they are all skipped | ### What happens to skipped tools -When steering interrupts a tool batch at index `i`, all tools from `i` to `N-1` are **not executed**. Instead, a tool result message is generated for each: +When steering interrupts a tool batch after tool `[i]` completes, all tools from `[i+1]` to `[N-1]` are **not executed**. Instead, a tool result message is generated for each: ```json { @@ -213,6 +238,27 @@ This allows **one extra iteration** when steering arrives right at the max itera > **Trade-off:** This introduces latency when the LLM requests multiple independent tools in a single turn. In practice, most batches contain 1-2 tools, so the impact is minimal. The benefit of being able to interrupt outweighs the cost. +### Why skip remaining tools (instead of letting them finish) + +Two strategies were considered when a steering message is detected mid-batch: + +1. **Skip remaining tools** (chosen) — stop executing, mark the rest as skipped, inject steering +2. **Finish all tools, then inject** — let everything run, append steering afterwards + +Strategy 2 was rejected for three reasons: + +**Irreversible side effects.** Tools can send emails, write files, spawn subagents, or call external APIs. If the user says "stop" or "change direction", those actions have already happened and cannot be undone. + +| Tool batch | Steering | Skip (1) | Finish (2) | +|---|---|---|---| +| `[search, send_email]` | "don't send it" | Email not sent | Email sent | +| `[query, write_file, spawn]` | "wrong database" | Only query runs | File + subagent wasted | +| `[fetch₁, fetch₂, fetch₃, write]` | topic change | 1 fetch | 3 fetches + write, all discarded | + +**Wasted latency.** Tools like web fetches and API calls take seconds each. In a 3-tool batch averaging 3-4s per tool, the user would wait 10+ seconds for work that gets thrown away. + +**The LLM retains full awareness.** Skipped tools receive an explicit `"Skipped due to queued user message."` result, so the model knows what was not done and can decide whether to re-execute with the new context or take a different path. + ## The Continue() method `Continue` handles the case where the agent is **idle** (its last message was from the assistant) and the user has enqueued steering messages in the meantime. @@ -255,3 +301,6 @@ flowchart TD | Skipped tools get explicit error results | The LLM protocol requires a tool result for every tool call in the assistant message. Omitting them would cause API errors. The skip message also informs the model about what was not done. | | `Continue()` uses `SkipInitialSteeringPoll` | Prevents race conditions and double-dequeuing when resuming an idle agent. | | Queue stored on `AgentLoop`, not `AgentInstance` | Steering is a loop-level concern (it affects the iteration flow), not a per-agent concern. All agents share the same steering queue since `processMessage` is sequential. | +| Bus drain goroutine in `Run()` | Channels (Telegram, Discord, etc.) publish to the bus via `PublishInbound`. Without the drain, messages would queue in the bus channel buffer and only be consumed after `processMessage` returns — defeating the purpose of steering. The drain goroutine bridges the gap by consuming new bus messages and calling `Steer()` while the agent is busy. | +| Audio transcription before steering | The drain goroutine calls `al.transcribeAudioInMessage(ctx, msg)` before steering, so voice messages are converted to text before the agent sees them. If transcription fails, the error is silently discarded and the original message is steered as-is. | +| `MaxQueueSize = 10` | Prevents unbounded memory growth if a user sends many messages while the agent is busy. Excess messages are dropped with a warning. | diff --git a/docs/steering.md b/docs/steering.md index 6f4cbd27b..ad08f8425 100644 --- a/docs/steering.md +++ b/docs/steering.md @@ -49,13 +49,16 @@ The environment variable `PICOCLAW_AGENTS_DEFAULTS_STEERING_MODE` can be used as ### Steer — Send a steering message ```go -agentLoop.Steer(providers.Message{ +err := agentLoop.Steer(providers.Message{ Role: "user", Content: "change direction, focus on X instead", }) +if err != nil { + // Queue is full (MaxQueueSize=10) or not initialized +} ``` -The message is enqueued in a thread-safe manner. It will be picked up at the next polling point (after the current tool finishes). +The message is enqueued in a thread-safe manner. Returns an error if the queue is full or not initialized. It will be picked up at the next polling point (after the current tool finishes). ### SteeringMode / SetSteeringMode @@ -73,6 +76,9 @@ When the agent is idle (it has finished processing and its last message was from ```go response, err := agentLoop.Continue(ctx, sessionKey, channel, chatID) +if err != nil { + // Error (e.g. "no default agent available") +} if response == "" { // No steering messages in queue, the agent stays idle } @@ -82,21 +88,48 @@ if response == "" { ## Polling points in the loop -Steering is checked at **three points** in the agent cycle: +Steering is checked at **two points** in the agent cycle: 1. **At loop start** — before the first LLM call, to catch messages enqueued during setup -2. **After each tool** — between tool calls within the same batch -3. **After the last tool** — to catch messages that arrived while the last tool was executing +2. **After every tool completes** — including the first and the last. If steering is found and there are remaining tools, they are all skipped immediately -## Skipped tool behavior +## Why remaining tools are skipped -When steering interrupts a batch of tool calls, the tools that were not yet executed receive a `tool` result with: +When a steering message is detected, all remaining tools in the batch are skipped rather than executed. The alternative — let all tools finish and inject the steering message afterwards — was considered and rejected. Here is why. + +### Preventing unwanted side effects + +Tools can have **irreversible side effects**. If the user says "no, wait" while the agent is mid-batch, executing the remaining tools means those side effects happen anyway: + +| Tool batch | Steering message | With skip | Without skip | +|---|---|---|---| +| `[web_search, send_email]` | "don't send it" | Email **not** sent | Email sent, damage done | +| `[query_db, write_file, spawn_agent]` | "use another database" | Only the query runs | File written + subagent spawned, all wasted | +| `[search₁, search₂, search₃, write_file]` | user changes topic entirely | 1 search | 3 searches + file write, all irrelevant | + +### Avoiding wasted time + +Tools that take seconds (web fetches, API calls, database queries) would all run to completion before the agent sees the user's correction. In a batch of 3 tools each taking 3-4 seconds, that's 10+ seconds of work that will be discarded. + +With skipping, the agent reacts as soon as the current tool finishes — typically within a few seconds instead of waiting for the entire batch. + +### The LLM gets full context + +Skipped tools receive an explicit error result (`"Skipped due to queued user message."`), so the model knows exactly which actions were not performed. It can then decide whether to re-execute them with the new context, or take a different path entirely. + +### Trade-off: sequential execution + +Skipping requires tools to run **sequentially** (the previous implementation ran them in parallel). This introduces latency when the LLM requests multiple independent tools in a single turn. In practice, most batches contain 1-2 tools, so the impact is minimal compared to the benefit of being able to stop unwanted actions. + +## Skipped tool result format + +When steering interrupts a batch, each tool that was not executed receives a `tool` result with: ``` Content: "Skipped due to queued user message." ``` -This is saved to the session and sent to the model, so it is aware that some requested actions were not performed. +This is saved to the session via `AddFullMessage` and sent to the model, so it is aware that some requested actions were not performed. ## Full flow example @@ -117,8 +150,17 @@ This is saved to the session and sent to the model, so it is aware that some req 7. LLM receives the full updated context and responds accordingly ``` +## Automatic bus drain + +When the agent loop (`Run()`) starts processing a message, it spawns a background goroutine that keeps consuming new inbound messages from the bus. These messages are automatically redirected into the steering queue via `Steer()`. This means: + +- Users on any channel (Telegram, Discord, etc.) don't need to do anything special — their messages are automatically captured as steering when the agent is busy +- Audio messages are transcribed before being steered, so the agent receives text. If transcription fails, the original (non-transcribed) message is steered as-is +- When `processMessage` finishes, the drain goroutine is canceled and normal message consumption resumes + ## Notes - Steering **does not interrupt** a tool that is currently executing. It waits for the current tool to finish, then checks the queue. - With `one-at-a-time` mode, if multiple messages are enqueued rapidly, they will be processed one per iteration. This gives the model the opportunity to react to each message individually. - With `all` mode, all pending messages are combined into a single injection. Useful when you want the agent to receive all the context at once. +- The steering queue has a maximum capacity of 10 messages (`MaxQueueSize`). `Steer()` returns an error when the queue is full. In the bus drain path, the error is logged as a warning and the message is effectively dropped. diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f3ed3ca6a..eb132c2db 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -260,6 +260,13 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } + // Start a goroutine that drains the bus while processMessage is + // running. Any inbound messages that arrive during processing are + // redirected into the steering queue so the agent loop can pick + // them up between tool calls. + drainCtx, drainCancel := context.WithCancel(ctx) + go al.drainBusToSteering(drainCtx) + // Process message func() { // TODO: Re-enable media cleanup after inbound media is properly consumed by the agent. @@ -275,6 +282,8 @@ func (al *AgentLoop) Run(ctx context.Context) error { // } // }() + defer drainCancel() + response, err := al.processMessage(ctx, msg) if err != nil { response = fmt.Sprintf("Error processing message: %v", err) @@ -321,6 +330,39 @@ func (al *AgentLoop) Run(ctx context.Context) error { return nil } +// drainBusToSteering continuously consumes inbound messages and redirects +// them into the steering queue. It runs in a goroutine while processMessage +// is active and stops when drainCtx is canceled (i.e., processMessage returns). +func (al *AgentLoop) drainBusToSteering(ctx context.Context) { + for { + msg, ok := al.bus.ConsumeInbound(ctx) + if !ok { + return + } + + // Transcribe audio if needed before steering, so the agent sees text. + msg, _ = al.transcribeAudioInMessage(ctx, msg) + + logger.InfoCF("agent", "Redirecting inbound message to steering queue", + map[string]any{ + "channel": msg.Channel, + "sender_id": msg.SenderID, + "content_len": len(msg.Content), + }) + + if err := al.Steer(providers.Message{ + Role: "user", + Content: msg.Content, + }); err != nil { + logger.WarnCF("agent", "Failed to steer message, will be lost", + map[string]any{ + "error": err.Error(), + "channel": msg.Channel, + }) + } + } +} + func (al *AgentLoop) Stop() { al.running.Store(false) } @@ -1285,33 +1327,6 @@ func (al *AgentLoop) runLLMIteration( var steeringAfterTools []providers.Message for i, tc := range normalizedToolCalls { - // Check for steering before executing (except for the first tool) - if i > 0 { - if steerMsgs := al.dequeueSteeringMessages(); len(steerMsgs) > 0 { - steeringAfterTools = steerMsgs - logger.InfoCF("agent", "Steering interrupt: skipping remaining tools", - map[string]any{ - "agent_id": agent.ID, - "skipped_from": i, - "total_tools": len(normalizedToolCalls), - "steering_count": len(steerMsgs), - }) - - // Mark remaining tool calls as skipped - for j := i; j < len(normalizedToolCalls); j++ { - skippedTC := normalizedToolCalls[j] - toolResultMsg := providers.Message{ - Role: "tool", - Content: "Skipped due to queued user message.", - ToolCallID: skippedTC.ID, - } - messages = append(messages, toolResultMsg) - agent.Sessions.AddFullMessage(opts.SessionKey, toolResultMsg) - } - break - } - } - argsJSON, _ := json.Marshal(tc.Arguments) argsPreview := utils.Truncate(string(argsJSON), 200) logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), @@ -1414,11 +1429,35 @@ func (al *AgentLoop) runLLMIteration( messages = append(messages, toolResultMsg) agent.Sessions.AddFullMessage(opts.SessionKey, toolResultMsg) - // After the last tool, also check for steering messages. - if i == len(normalizedToolCalls)-1 { - if steerMsgs := al.dequeueSteeringMessages(); len(steerMsgs) > 0 { - steeringAfterTools = steerMsgs + // After EVERY tool (including the first and last), check for + // steering messages. If found and there are remaining tools, + // skip them all. + if steerMsgs := al.dequeueSteeringMessages(); len(steerMsgs) > 0 { + remaining := len(normalizedToolCalls) - i - 1 + if remaining > 0 { + logger.InfoCF("agent", "Steering interrupt: skipping remaining tools", + map[string]any{ + "agent_id": agent.ID, + "completed": i + 1, + "skipped": remaining, + "total_tools": len(normalizedToolCalls), + "steering_count": len(steerMsgs), + }) + + // Mark remaining tool calls as skipped + for j := i + 1; j < len(normalizedToolCalls); j++ { + skippedTC := normalizedToolCalls[j] + toolResultMsg := providers.Message{ + Role: "tool", + Content: "Skipped due to queued user message.", + ToolCallID: skippedTC.ID, + } + messages = append(messages, toolResultMsg) + agent.Sessions.AddFullMessage(opts.SessionKey, toolResultMsg) + } } + steeringAfterTools = steerMsgs + break } }