Add provider steering and fix prompt queue behavior

README.md

@@ -176,24 +176,28 @@ Form requests support strings, numbers, integers, booleans, single choices, and
 
 | Key | Action |
 | --- | --- |
-| Enter | Send prompt or select a list entry. |
+| Enter | Send a prompt; steer the active turn when supported, otherwise queue it. Select a list entry or save an edited queue entry. |
 | Alt+Enter / Ctrl+J | Insert a newline. Shift+Enter also works in compatible terminals. |
 | Ctrl+P | Compact command search. |
 | Ctrl+O | Expand full transcript and tool details; press again or Esc to return. |
 | Ctrl+G | Agent picker. |
 | Ctrl+S | Session picker. |
 | Ctrl+N | New session. |
-| Tab | Complete the selected command, setting, or `@file` suggestion. |
-| Up / Alt+Up, Down / Alt+Down | Recall prompt history; Up works on the first input line. |
+| Tab | Complete the selected suggestion; otherwise queue a follow-up while the agent works. |
+| Alt+Up | Edit the latest queued prompt, keeping the current draft for later; otherwise recall prompt history. |
+| Up, Down / Alt+Down | Recall prompt history; Up works on the first input line. |
 | Terminal scrollback / mouse wheel | Scroll completed output; mouse selection is available. |
 | Page Up / Page Down | Scroll expanded details and dialog explanations. |
-| Esc | Dismiss suggestions/dialogs first; otherwise stop the active turn. |
+| Esc | Restore an edited queue entry and the previous draft; otherwise dismiss suggestions/dialogs or stop the active turn. |
 | Ctrl+C | Stop an active turn; otherwise clear a draft, or press twice on an empty prompt to exit. |
 | Ctrl+D | Exit from an empty, idle prompt. |
 | Type in a selector | Filter immediately; one Enter selects, Esc returns to the draft. |
 | Ctrl+D in the session picker | Confirm deletion, or local removal if remote deletion is unsupported. |
+| Ctrl+D in the queue picker | Remove the selected queued prompt. |
 
-You can compose the next prompt while the agent works. Enter queues it, and queued prompts run in order after a successful turn. Stopping or failing a turn pauses the queue; `/queue` lets you edit an entry, and `/queue send` continues it. Queued prompts remain in memory and stay tied to their original session; they are not saved across restarts. Multiline paste stays in the draft until you send it.
+You can compose while the agent works. **Enter steers** an active turn when the agent advertises steering, including current Codex and Claude ACP adapters. **Tab queues** a separate follow-up, after any visible completion suggestion is handled. Agents without steering support queue Enter submissions too. Pending prompts appear above the composer and run in order after a successful turn.
+
+**Alt+Up** edits the latest queued prompt; `/queue` selects any entry. Editing pauses dispatch, retains the entry's position and attachments, and saves the draft you were typing. Enter saves the edit; Esc restores the original entry. Both return to your previous draft. Use Ctrl+D in `/queue` to remove an entry, `/queue clear` to remove queued work, and `/queue send` to resume after a stop or failure. An unconfirmed steering delivery stays in a paused queue for review instead of being resent automatically. Queues stay tied to their original session and remain in memory; they are not saved across restarts. Multiline paste stays in the draft until you send it.
 
 Other commands: `/agents`, `/help`, `/info`, `/auth`, `/logout`, `/refresh`, `/quit`. `/info` displays advertised capabilities and configuration. `/logs` shows recent agent stderr and the last error; `/reconnect` restarts the agent and reloads the saved session when supported. `/logout` requires the agent's logout capability.
 
@@ -226,7 +230,7 @@ Under `MICRO_ACP_HOME`, the paths are `config.json`, `data/sessions/`, and
 then overrides only the configuration file. Demo history lives under `demo/`
 in the state directory (`data/demo/` with `MICRO_ACP_HOME`).
 
-The client targets stable **ACP v1** using `BrokkAi/acp-go`, pinned to commit `5b2c77c673e0` (v0.10.0 plus the cancellation transport fix). Native forks additionally use the SDK's opt-in unstable v1 schema. Draft ACP v2 and editor-specific experimental extensions are not advertised. The [support matrix](docs/acp-support.md) maps protocol methods to UI flows and tests.
+The client targets stable **ACP v1** using `BrokkAi/acp-go`, pinned to commit `5b2c77c673e0` (v0.10.0 plus the cancellation transport fix). Native forks additionally use the SDK's opt-in unstable v1 schema. Steering uses the advertised `_session/steering` extension. Draft ACP v2 and editor-specific experimental capabilities are not advertised. The [support matrix](docs/acp-support.md) maps protocol methods to UI flows and tests.
 
 ## Development
 

docs/acp-support.md

@@ -47,7 +47,7 @@ Terminal authentication is a separate process flow rather than an `authenticate`
 | `available_commands_update` | Refreshes inline slash suggestions; resolves client command collisions through `/agent`. |
 | `current_mode_update` | Updates the current legacy mode. |
 | `config_option_update` | Refreshes open selectors, completions, and the persistent status line, including custom options, On/Off values, and model-dependent choices. |
-| `session_info_update` | Updates title and timestamp. |
+| `session_info_update` | Updates title and timestamp; Codex thread-status metadata also tracks detached steering continuations. |
 | `usage_update` | Stores used/capacity and reported cost; the status line displays the percentage of context remaining. |
 
 Prompts can carry text, resource links, images, audio, embedded text resources and embedded binary resources. Capability checks happen before sending. Full content blocks persist in local session files. The TUI renders text/resources and labels images, audio and binary data; it does not render bitmap images or play audio. Tool content supports text/resources, diffs and terminal references. `TestRichContentRoundTripAndPersistence` verifies these survive the actual stdio transport and disk storage.
@@ -56,6 +56,7 @@ Form fields support the restricted flat schema: strings, integers, numbers, bool
 
 ## Extensions and boundaries
 
+- `_session/steering`: enabled only by top-level `initialize._meta.steering.supported`. Enter injects guidance; Tab retains a local FIFO follow-up. The client requests `idleBehavior: promptRequired`, so Claude can return unconsumed late input for one normal `session/prompt`. Codex's `startedNewTurn` is already consumed: its `_meta.codex.threadStatus` updates keep the turn busy until completion. Accepted steering is recorded once; rejected or ambiguous deliveries remain in a paused queue. See the [Claude example](https://github.com/agentclientprotocol/claude-agent-acp/blob/main/examples/steering.ts) and [Codex extension](https://github.com/agentclientprotocol/codex-acp/blob/main/src/AcpExtensions.ts). In-memory ACP transport tests cover acceptance, non-delivery, unknown outcomes, and detached completion before/after acknowledgment; TUI tests cover queue order, editing, deletion, draft/context retention, and fallback.
 - `session/fork`: implemented through the SDK's unstable v1 schema only when advertised. Replayed history replaces the inherited transcript; agents that do not replay retain the saved history. Tested by `TestNativeForkUsesAdvertisedCapability` and `TestNativeForkReplayReplacesInheritedHistory`. `/fork --context` is a separately labeled text-context fallback.
 - MCP configuration: stdio, HTTP and SSE definitions are forwarded during session operations. The agent connects to and operates those servers. This client is not an MCP server or an MCP-over-ACP proxy.
 - Additional directories: passed only through supported session operations, restored with saved sessions, and included in client callback routing.

internal/client/client.go

@@ -54,6 +54,8 @@ type Client struct {
 	shutdownOnce     sync.Once
 	shutdownErr      error
 	op               sync.Mutex
+	steerOp          sync.Mutex
+	steering         steeringState
 	mu               sync.Mutex
 	current          store.Session
 	wire             acp.Session
@@ -214,6 +216,13 @@ func (c *Client) Save() error {
 	if s.ID == "" || s.RemoteID == "" {
 		return nil
 	}
+	messages := s.Messages[:0]
+	for _, message := range s.Messages {
+		if !message.Pending {
+			messages = append(messages, message)
+		}
+	}
+	s.Messages = messages
 	return c.store.Save(s)
 }
 
@@ -545,6 +554,7 @@ func (c *Client) PromptContent(blocks []acp.Content) (reason schema.StopReason,
 	c.mu.Lock()
 	c.turnCancel = cancel
 	c.turnDone = done
+	c.steering = steeringState{wake: make(chan struct{}), disabled: c.steering.disabled, sequence: c.steering.sequence}
 	c.permissionCtx = permissions
 	c.permissionCancel = stopPermissions
 	c.cancelRequested = false
@@ -574,12 +584,18 @@ func (c *Client) PromptContent(blocks []acp.Content) (reason schema.StopReason,
 		}
 	}()
 	if err = c.Save(); err != nil {
+		// A follow-up can already be waiting for this prompt's first activity.
+		// Release it as unconsumed even when the initial save prevents dispatch.
+		err = errors.Join(err, c.finishSteering(ctx))
 		return "", err
 	}
 	if s.PendingContext != "" {
 		blocks = append([]acp.Content{acp.NewTextContent(s.PendingContext)}, blocks...)
 	}
 	reason, err = c.conn.PromptContent(ctx, c.Init, c.session(), blocks)
+	if steerErr := c.finishSteering(ctx); steerErr != nil {
+		err = errors.Join(err, steerErr)
+	}
 	c.mu.Lock()
 	cancelled := c.cancelRequested
 	if reason == schema.StopReasonCancelled {
@@ -607,6 +623,7 @@ func (c *Client) Cancel() error {
 	stopPermissions := c.permissionCancel
 	if c.turnDone != nil {
 		c.cancelRequested = true
+		c.wakeSteering()
 		c.cancelTurnTools()
 	}
 	id := c.wire.SessionID

internal/client/steering.go

@@ -0,0 +1,207 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	acp "github.com/BrokkAi/acp-go"
+	"github.com/BrokkAi/acp-go/schema"
+	"github.com/BrokkAi/micro-acp/internal/store"
+)
+
+type SteeringOutcome string
+
+const (
+	SteeringInjected       SteeringOutcome = "injected"
+	SteeringStartedNewTurn SteeringOutcome = "startedNewTurn"
+	SteeringPromptRequired SteeringOutcome = "promptRequired"
+)
+
+type steeringState struct {
+	pending                        int
+	sequence                       uint64
+	closing, disabled              bool
+	ready, promptReturned          bool
+	wake                           chan struct{}
+	status                         string
+	generation, detachedGeneration uint64
+}
+
+func (c *Client) CanSteer() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	capability, _ := c.Init.Meta["steering"].(map[string]any)
+	return capability["supported"] == true && !c.steering.disabled
+}
+
+// SteerContent joins an active prompt without cancelling it or taking the
+// session-operation lock. Only explicit non-delivery permits automatic fallback.
+func (c *Client) SteerContent(sessionID string, blocks []acp.Content) (outcome SteeringOutcome, err error) {
+	c.steerOp.Lock()
+	defer c.steerOp.Unlock()
+	if err := c.validateContent(blocks); err != nil {
+		return "", err
+	}
+	if !c.CanSteer() {
+		return SteeringPromptRequired, nil
+	}
+	c.mu.Lock()
+	if c.current.ID != sessionID {
+		c.mu.Unlock()
+		return "", errors.New("steering belongs to another session")
+	}
+	if c.turnDone == nil || c.steering.closing {
+		c.mu.Unlock()
+		return SteeringPromptRequired, nil
+	}
+	if c.cancelRequested {
+		c.mu.Unlock()
+		return "", errors.New("turn is stopping; follow-up retained in queue")
+	}
+	c.steering.pending++
+	c.steering.sequence++
+	id := fmt.Sprintf("steer-%d", c.steering.sequence)
+	c.mu.Unlock()
+	defer func() {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		for i := range c.current.Messages {
+			message := &c.current.Messages[i]
+			if message.Pending && message.ID == id {
+				if err == nil && (outcome == SteeringInjected || outcome == SteeringStartedNewTurn) {
+					message.Pending = false
+				} else {
+					c.current.Messages = append(c.current.Messages[:i], c.current.Messages[i+1:]...)
+				}
+				break
+			}
+		}
+		c.steering.pending--
+		c.current.UpdatedAt = time.Now().UTC()
+		c.revision++
+		c.wakeSteering()
+	}()
+	// The prompt may still be saving its initial history before writing the RPC.
+	// Wait for turn activity or its response so steering cannot overtake it.
+	for {
+		c.mu.Lock()
+		s := c.steering
+		stopping := c.cancelRequested
+		if s.ready || s.promptReturned || stopping {
+			c.mu.Unlock()
+			if stopping {
+				return "", errors.New("turn is stopping; follow-up retained in queue")
+			}
+			if !s.ready || (s.promptReturned && s.detachedGeneration == 0) {
+				return SteeringPromptRequired, nil
+			}
+			break
+		}
+		c.mu.Unlock()
+		select {
+		case <-s.wake:
+		case <-c.conn.Done():
+			return "", errors.New("agent disconnected before steering could be sent")
+		case <-c.ctx.Done():
+			return "", c.ctx.Err()
+		}
+	}
+	c.mu.Lock()
+	nextGeneration := max(uint64(1), c.steering.generation) + 1
+	var text []string
+	for _, block := range blocks {
+		text = append(text, ContentText(block))
+	}
+	c.current.Messages = append(c.current.Messages, store.Message{Role: "user", ID: id, Text: strings.Join(text, "\n"), Content: blocks, Pending: true})
+	remoteID := c.wire.SessionID
+	c.revision++
+	c.mu.Unlock()
+	request := schema.PromptRequest{SessionID: remoteID, Prompt: blocks, Meta: schema.Meta{"steering": map[string]any{"idleBehavior": "promptRequired"}}}
+	var response struct {
+		Outcome SteeringOutcome `json:"outcome"`
+	}
+	if err = c.conn.Call(c.ctx, "_session/steering", request, &response); err != nil {
+		var rpcErr *acp.RPCError
+		if errors.As(err, &rpcErr) && rpcErr.Code == -32601 {
+			c.mu.Lock()
+			c.steering.disabled = true
+			c.mu.Unlock()
+			return SteeringPromptRequired, nil
+		}
+		return "", fmt.Errorf("steering delivery was not confirmed; check the transcript before retrying: %w", err)
+	}
+	switch response.Outcome {
+	case SteeringInjected, SteeringPromptRequired:
+		return response.Outcome, nil
+	case SteeringStartedNewTurn:
+		// Codex starts a detached continuation if the original turn settled.
+		// Its thread-status updates own completion; never send the input again.
+		c.mu.Lock()
+		c.steering.detachedGeneration = nextGeneration
+		stopping := c.cancelRequested
+		c.mu.Unlock()
+		if stopping {
+			_ = c.Cancel()
+		}
+		return response.Outcome, nil
+	case "failed":
+		return "", errors.New("agent rejected steering; follow-up retained in queue")
+	default:
+		return "", fmt.Errorf("unknown steering outcome %q; check the transcript before retrying", response.Outcome)
+	}
+}
+
+// Called with mu held. These metadata updates describe Codex's detached turn
+// lifecycle; Claude honors the promptRequired opt-in instead.
+func (c *Client) steeringStatus(meta schema.Meta) {
+	codex, _ := meta["codex"].(map[string]any)
+	status, _ := codex["threadStatus"].(map[string]any)
+	value, _ := status["type"].(string)
+	if value != "active" && value != "idle" && value != "systemError" {
+		return
+	}
+	if value == "active" && c.steering.status != "active" {
+		c.steering.generation++
+		c.steering.ready = true
+	}
+	c.steering.status = value
+	c.wakeSteering()
+}
+
+func (c *Client) wakeSteering() {
+	if c.steering.wake != nil {
+		close(c.steering.wake)
+		c.steering.wake = make(chan struct{})
+	}
+}
+
+func (c *Client) finishSteering(ctx context.Context) error {
+	c.mu.Lock()
+	c.steering.promptReturned = true
+	c.wakeSteering()
+	c.mu.Unlock()
+	for {
+		c.mu.Lock()
+		s := c.steering
+		detachedDone := s.detachedGeneration == 0 || (s.generation >= s.detachedGeneration && (s.status == "idle" || s.status == "systemError"))
+		if s.pending == 0 && detachedDone {
+			c.steering.closing = true
+			c.mu.Unlock()
+			if s.detachedGeneration != 0 && s.status == "systemError" {
+				return errors.New("steered continuation failed")
+			}
+			return nil
+		}
+		c.mu.Unlock()
+		select {
+		case <-s.wake:
+		case <-c.conn.Done():
+			return fmt.Errorf("agent disconnected during steering: %w", c.conn.Err())
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}