Add mid-turn message steering for running agent sessions

[trungutt]

Summary

Addresses #2223.

• Add POST /sessions/:id/steer API endpoint that injects user messages into a running agent session mid-turn
• Add POST /sessions/:id/followup API endpoint that queues messages for end-of-turn processing (one per turn)
• Agent loop drains steered messages after tool execution and before the stop-condition check, so the LLM sees them on the next iteration
• Follow-up messages are popped one at a time after the model stops and stop-hooks run, giving each a full undivided agent turn
• Emits user_message SSE events when messages are picked up, so API clients can confirm delivery to the user
• Guard RunSession against concurrent streams on the same session

This is a common pattern in agentic coding tools where the user can steer or provide follow-up context while the agent is executing tool calls. No behavioral change to the TUI — the existing client-side message queue continues to work as before. The TUI can adopt mid-turn steering in a follow-up PR.

Current flow (before this PR)

Messages sent while the agent is busy are held client-side and only processed after the entire turn finishes. Each queued message starts a completely new turn.

sequenceDiagram
    participant User
    participant Client
    participant Agent Loop

    User->>Client: "fix the bug"
    Client->>Agent Loop: POST /sessions/:id/agent (start stream)
    activate Agent Loop

    Note over Agent Loop: LLM call → tool calls → LLM call → ...

    User->>Client: "also update the tests"
    Note over Client: Message held in<br/>client-side queue

    User->>Client: "use pytest not unittest"
    Note over Client: Message held in<br/>client-side queue

    Agent Loop-->>Client: StreamStopped
    deactivate Agent Loop

    Client->>Agent Loop: POST /sessions/:id/agent ("also update the tests")
    activate Agent Loop
    Agent Loop-->>Client: StreamStopped
    deactivate Agent Loop

    Client->>Agent Loop: POST /sessions/:id/agent ("use pytest not unittest")
    activate Agent Loop
    Agent Loop-->>Client: StreamStopped
    deactivate Agent Loop

Loading

The problem

• The agent finishes its work without knowing the user wanted tests updated or pytest used — it may have already written unittest-based tests
• Each queued message starts a separate turn, so the agent does redundant work (e.g. rewrites tests it just wrote)
• The user has no way to course-correct the agent while it's working

New flow (this PR)

Two distinct mechanisms for two distinct intents:

Steer — urgent mid-turn injection

sequenceDiagram
    participant User
    participant Client
    participant Steer Queue
    participant Agent Loop

    User->>Client: "fix the bug"
    Client->>Agent Loop: POST /sessions/:id/agent (start stream)
    activate Agent Loop

    Note over Agent Loop: LLM call → tool calls executing...

    User->>Client: "use pytest not unittest"
    Client->>Steer Queue: POST /sessions/:id/steer
    Note over Steer Queue: Message enqueued

    Note over Agent Loop: Tool calls finish

    Steer Queue-->>Agent Loop: Drain all (mid-turn)
    Note over Agent Loop: Injects message in<br/>system-reminder tags,<br/>forces loop to continue

    Note over Agent Loop: LLM call (sees steered message) → ...

    Agent Loop-->>Client: StreamStopped
    deactivate Agent Loop

Loading

The agent sees "use pytest not unittest" before writing any tests — no wasted work.

Follow-up — end-of-turn queue

sequenceDiagram
    participant User
    participant Client
    participant FollowUp Queue
    participant Agent Loop

    User->>Client: "fix the bug"
    Client->>Agent Loop: POST /sessions/:id/agent (start stream)
    activate Agent Loop

    User->>Client: "then write a README"
    Client->>FollowUp Queue: POST /sessions/:id/followup
    Note over FollowUp Queue: Message enqueued

    User->>Client: "and add a changelog entry"
    Client->>FollowUp Queue: POST /sessions/:id/followup
    Note over FollowUp Queue: Message enqueued

    Note over Agent Loop: Model stops (bug fixed)

    FollowUp Queue-->>Agent Loop: Pop one ("write a README")
    Note over Agent Loop: New turn starts<br/>(plain user message)
    Note over Agent Loop: Model stops (README written)

    FollowUp Queue-->>Agent Loop: Pop one ("add a changelog entry")
    Note over Agent Loop: New turn starts
    Note over Agent Loop: Model stops (changelog added)

    Agent Loop-->>Client: StreamStopped
    deactivate Agent Loop

Loading

Each follow-up gets a full undivided turn — the agent finishes one task before starting the next.

Steer vs Follow-up — two distinct scenarios

	Steer (POST /steer)	Follow-up (POST /followup)
When	Injected at the next safe point mid-turn (after tool execution, before the next LLM call)	Processed after the current turn finishes completely
Drain semantics	Drain all pending messages every check	Pop exactly one message per turn boundary
Injection format	Wrapped in <system-reminder> tags (mid-stream context)	Plain user message (fresh turn)
Effect on loop	Forces res.Stopped = false — loop continues	continue — starts a new iteration of the loop
Use case	"The agent needs to see this NOW" — course corrections, additional context	"Run this next" — follow-up questions that can wait

MessageQueue interface with Lock + Confirm/Cancel

The queue storage is behind a MessageQueue interface so callers can provide their own implementation (e.g. persistent/distributed store). The interface uses a Lock + Confirm/Cancel pattern for safe dequeue: Dequeue locks a message, Confirm permanently removes it after sess.AddMessage succeeds, and Cancel releases it back to the queue on failure. This prevents message loss in persistent implementations when the process crashes between dequeue and session persistence.

type MessageQueue interface {
    Enqueue(ctx context.Context, msg QueuedMessage) bool
    Dequeue(ctx context.Context) (QueuedMessage, bool)  // locks the message
    Confirm(ctx context.Context) error                   // removes after successful processing
    Cancel(ctx context.Context) error                    // releases back to queue on failure
    Drain(ctx context.Context) []QueuedMessage           // lock + auto-confirm all
    Len(ctx context.Context) int
}

The default in-memory implementation (backed by a buffered channel) treats Confirm/Cancel as no-ops. Custom implementations can be injected via WithSteerQueue and WithFollowUpQueue options:

rt, err := runtime.New(team,
    runtime.WithSteerQueue(myPersistentQueue),
    runtime.WithFollowUpQueue(myPersistentQueue),
)