fix loop

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]
+        TG[Telegram]
-        API[HTTP API]
+        DC[Discord]
-        WS[WebSocket]
+        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
+    TG -->|PublishInbound| BUS
-    API -->|Steer| SQ
+    DC -->|PublishInbound| BUS
-    WS -->|Steer| SQ
+    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->>ToolExecution: execute tool[i]
+        ToolExecution->>ToolExecution: process result,<br/>append to messages[]
+
         ToolExecution->>AgentLoop: dequeueSteeringMessages()
         AgentLoop-->>ToolExecution: steeringMessages
 
         alt steering found
-                Note over ToolExecution: Mark tool[i..N] as<br/>"Skipped due to queued user message."
+            opt remaining tools > 0
-                ToolExecution-->>runLLMIteration: steeringAfterTools = steeringMessages
+                Note over ToolExecution: Mark tool[i+1..N-1] as<br/>"Skipped due to queued user message."
-                Note over ToolExecution: break out of tool loop
             end
-        end
-
-        ToolExecution->>ToolExecution: execute tool[i]
-        ToolExecution->>ToolExecution: process result,<br/>append to messages[]
-
-        alt last tool (i == N-1)
-            ToolExecution->>AgentLoop: dequeueSteeringMessages()
-            AgentLoop-->>ToolExecution: steeringMessages (may be empty)
             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 |
+| 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 |
-| 3 | After the last tool in the batch | After tool `[N-1]` finishes | Catch messages that arrived during the last tool's execution |
 
 ### 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. |

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
+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
-3. **After the last tool** — to catch messages that arrived while the last tool was executing
 
-## 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.

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.
+			// After EVERY tool (including the first and last), check for
-			if i == len(normalizedToolCalls)-1 {
+			// steering messages. If found and there are remaining tools,
+			// skip them all.
 			if steerMsgs := al.dequeueSteeringMessages(); len(steerMsgs) > 0 {
-					steeringAfterTools = steerMsgs
+				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
 			}
 		}