Merge pull request #7708 from skyzhao1223/fix/serve-replay-pending-on-sse

fix(serve): replay pending prompts when SSE clients attach

internal/control/approval.go

@@ -57,6 +57,10 @@ type approvalManager struct {
 	// requestApproval): Sink implementations must not block and must not call
 	// back into Ask or the tool-approval chain, or they deadlock the prompt.
 	promptMu sync.Mutex
+	// promptEmitMu serializes prompt registration and emission with an SSE
+	// attach handoff. It is separate from promptMu because promptMu remains
+	// held while waiting for the user's answer.
+	promptEmitMu sync.Mutex
 }
 
 func newApprovalManager(policy permission.Policy, mode string, timeout time.Duration) approvalManager {

internal/control/controller.go

@@ -2479,8 +2479,10 @@ func (c *Controller) Ask(ctx context.Context, questions []event.AskQuestion) ([]
 	c.approval.promptMu.Lock()
 	defer c.approval.promptMu.Unlock()
 
+	c.approval.promptEmitMu.Lock()
 	id, reply := c.approval.registerAsk(questions)
 	c.sink.Emit(event.Event{Kind: event.AskRequest, Ask: event.Ask{ID: id, Questions: questions}})
+	c.approval.promptEmitMu.Unlock()
 
 	waitCtx, cancelWait := c.approval.waitContext(ctx)
 	defer cancelWait()
@@ -2571,16 +2573,52 @@ func askAnswersHaveSelection(answers []event.AskAnswer) bool {
 // requestApproval, so in practice at most one prompt is outstanding; the loops
 // stay general so a future concurrent prompt would still replay correctly.
 func (c *Controller) ReplayPendingPrompts() {
+	c.approval.promptEmitMu.Lock()
+	noApprovals := c.replayPendingPromptsTo(c.sink)
+	c.approval.promptEmitMu.Unlock()
+	if noApprovals {
+		// Retained compatibility hook; live Auto Guard cards are ordinary approvals.
+		c.ReplayUnresolvedRecoveries()
+	}
+}
+
+// ReplayPendingPromptsTo re-emits pending prompts to one frontend sink. Serve
+// uses this for a newly attached SSE client so existing browsers do not receive
+// duplicate approval/ask cards when another client reconnects.
+func (c *Controller) ReplayPendingPromptsTo(sink event.Sink) {
+	c.approval.promptEmitMu.Lock()
+	defer c.approval.promptEmitMu.Unlock()
+	c.replayPendingPromptsTo(sink)
+}
+
+// ReplayPendingPromptsWith performs an SSE connection handoff while prompt
+// registration and emission are paused. The factory must subscribe the new
+// client and return a sink that targets it; this closes the attach race where
+// the original prompt could otherwise land between Subscribe and replay.
+func (c *Controller) ReplayPendingPromptsWith(sinkFactory func() event.Sink) {
+	if sinkFactory == nil {
+		return
+	}
+	c.approval.promptEmitMu.Lock()
+	defer c.approval.promptEmitMu.Unlock()
+	c.replayPendingPromptsTo(sinkFactory())
+}
+
+func (c *Controller) replayPendingPromptsTo(sink event.Sink) bool {
 	approvals, asks := c.approval.snapshotPrompts()
+	c.emitPendingPrompts(sink, approvals, asks)
+	return len(approvals) == 0
+}
+
+func (c *Controller) emitPendingPrompts(sink event.Sink, approvals []event.Approval, asks []event.Ask) {
+	if sink == nil {
+		return
+	}
 	for _, a := range approvals {
-		c.sink.Emit(c.approvalRequestEvent(a))
+		sink.Emit(c.approvalRequestEvent(a))
 	}
 	for _, a := range asks {
-		c.sink.Emit(event.Event{Kind: event.AskRequest, Ask: a})
-	}
-	// Retained compatibility hook; live Auto Guard cards are ordinary approvals.
-	if len(approvals) == 0 {
-		c.ReplayUnresolvedRecoveries()
+		sink.Emit(event.Event{Kind: event.AskRequest, Ask: a})
 	}
 }
 
@@ -6628,6 +6666,7 @@ func (c *Controller) requestApprovalDecisionWithOptions(ctx context.Context, too
 		}
 	}
 
+	c.approval.promptEmitMu.Lock()
 	var id string
 	var reply chan approvalReply
 	if opts.fresh || opts.requireHuman || tool == planApprovalTool {
@@ -6641,6 +6680,7 @@ func (c *Controller) requestApprovalDecisionWithOptions(ctx context.Context, too
 	}
 
 	c.sink.Emit(c.approvalRequestEvent(event.Approval{ID: id, Tool: tool, Subject: subject, Reason: reason, RawInput: append(json.RawMessage(nil), args...), Fresh: opts.fresh}))
+	c.approval.promptEmitMu.Unlock()
 	// The agent now needs the user's attention; a Notification hook can ping an
 	// external channel (desktop notice, phone) while the run blocks on the reply.
 	go c.hooks.Notification(ctx, approvalNotificationText(tool, subject), "permission_prompt")

internal/control/port.go

@@ -83,6 +83,8 @@ type Approvals interface {
 	AnswerQuestion(id string, answers []event.AskAnswer)
 	Ask(ctx context.Context, questions []event.AskQuestion) ([]event.AskAnswer, error)
 	ReplayPendingPrompts()
+	ReplayPendingPromptsTo(sink event.Sink)
+	ReplayPendingPromptsWith(sinkFactory func() event.Sink)
 	PendingPrompt() bool
 	EnableInteractiveApproval()
 	ToolApprovalMode() string

internal/control/recovery.go

@@ -297,6 +297,7 @@ func (c *Controller) emitRecoveryPrompt(ctx context.Context, taskID string, pend
 	// Strict fresh decision: never session/persist grants, never auto-drain on
 	// mode switch.
 	c.approval.promptMu.Lock()
+	c.approval.promptEmitMu.Lock()
 	// Hold promptMu for the duration of registration+emit only; waiting happens
 	// in the recovery gate on its own channel. We deliberately do not block here
 	// on the approval reply — ResolveRecovery unblocks the gate.
@@ -330,6 +331,7 @@ func (c *Controller) emitRecoveryPrompt(ctx context.Context, taskID string, pend
 	}()
 
 	c.sink.Emit(c.approvalRequestEvent(ev))
+	c.approval.promptEmitMu.Unlock()
 	c.approval.promptMu.Unlock()
 
 	if c.hooks != nil {

internal/serve/broadcaster.go

@@ -41,6 +41,29 @@ func (b *Broadcaster) Emit(e event.Event) {
 	}
 }
 
+// EmitTo delivers an event only to the supplied subscriber. It is used for
+// connection-local recovery frames, such as replaying a prompt to a browser
+// that attached after the original event was emitted. Normal runtime events
+// should continue to use Emit so every subscriber receives them.
+func (b *Broadcaster) EmitTo(target <-chan []byte, e event.Event) {
+	data, err := json.Marshal(eventwire.ToWire(e))
+	if err != nil {
+		return
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	for ch := range b.subs {
+		if (<-chan []byte)(ch) != target {
+			continue
+		}
+		select {
+		case ch <- data:
+		default: // subscriber is behind; drop this frame rather than blocking.
+		}
+		return
+	}
+}
+
 // Subscribe registers a new SSE client and returns its channel plus an
 // unsubscribe func the handler must call (defer) when the client disconnects.
 func (b *Broadcaster) Subscribe() (<-chan []byte, func()) {

internal/serve/serve.go

@@ -653,7 +653,16 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
 
-	ch, unsubscribe := s.bc.Subscribe()
+	var ch <-chan []byte
+	var unsubscribe func()
+	// Subscribe and replay as one handoff. Prompt producers are serialized with
+	// this operation, so no original event can land between the two steps.
+	s.ctl().ReplayPendingPromptsWith(func() event.Sink {
+		ch, unsubscribe = s.bc.Subscribe()
+		return event.FuncSink(func(e event.Event) {
+			s.bc.EmitTo(ch, e)
+		})
+	})
 	defer unsubscribe()
 
 	fmt.Fprint(w, ": connected\n\n") // open the stream immediately

internal/serve/serve_test.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -19,7 +20,9 @@ import (
 	"reasonix/internal/event"
 	"reasonix/internal/eventwire"
 	"reasonix/internal/jobs"
+	"reasonix/internal/permission"
 	"reasonix/internal/provider"
+	"reasonix/internal/tool"
 )
 
 func TestTitlePromptRequiresUserMessageLanguage(t *testing.T) {
@@ -66,6 +69,43 @@ type fakeRunner struct{ got chan string }
 
 func (f fakeRunner) Run(_ context.Context, input string) error { f.got <- input; return nil }
 
+type serveApprovalWriter struct{}
+
+func (serveApprovalWriter) Name() string        { return "serve_write" }
+func (serveApprovalWriter) Description() string { return "write a test file" }
+func (serveApprovalWriter) Schema() json.RawMessage {
+	return json.RawMessage(`{"type":"object","properties":{"path":{"type":"string"}}}`)
+}
+func (serveApprovalWriter) ReadOnly() bool { return false }
+func (serveApprovalWriter) Execute(context.Context, json.RawMessage) (string, error) {
+	return "ok", nil
+}
+
+type serveApprovalProvider struct {
+	mu   sync.Mutex
+	turn int
+}
+
+func (p *serveApprovalProvider) Name() string { return "serve-approval-test" }
+func (p *serveApprovalProvider) Stream(context.Context, provider.Request) (<-chan provider.Chunk, error) {
+	p.mu.Lock()
+	turn := p.turn
+	p.turn++
+	p.mu.Unlock()
+
+	ch := make(chan provider.Chunk, 2)
+	if turn == 0 {
+		ch <- provider.Chunk{Type: provider.ChunkToolCall, ToolCall: &provider.ToolCall{
+			ID: "serve-approval-1", Name: "serve_write", Arguments: `{"path":"a.txt"}`,
+		}}
+	} else {
+		ch <- provider.Chunk{Type: provider.ChunkText, Text: "done"}
+	}
+	ch <- provider.Chunk{Type: provider.ChunkDone}
+	close(ch)
+	return ch, nil
+}
+
 func TestServeSubmitRunsAndBroadcastsTurnDone(t *testing.T) {
 	bc := NewBroadcaster()
 	got := make(chan string, 1)
@@ -957,3 +997,246 @@ func TestServeContextEndpoint(t *testing.T) {
 		t.Errorf("used = %d, want 0", body["used"])
 	}
 }
+
+// TestServeEventsReplaysPendingAskOnAttach proves a late /events subscriber
+// receives a still-blocked ask_request. Without replay, the browser attaches to
+// a healthy-looking session that never surfaces the parked prompt (#7643).
+func TestServeEventsReplaysPendingAskOnAttach(t *testing.T) {
+	bc := NewBroadcaster()
+	ctrl := control.New(control.Options{Sink: bc})
+	ctrl.EnableInteractiveApproval()
+	srv := httptest.NewServer(New(ctrl, bc, config.ServeConfig{}).Handler())
+	defer srv.Close()
+
+	firstSub, cancelFirst := bc.Subscribe()
+	defer cancelFirst()
+
+	askCtx, cancelAsk := context.WithCancel(context.Background())
+	askDone := make(chan error, 1)
+	go func() {
+		_, err := ctrl.Ask(askCtx, []event.AskQuestion{{
+			ID: "q1", Prompt: "pick one", Options: []event.AskOption{{Label: "A"}, {Label: "B"}},
+		}})
+		askDone <- err
+	}()
+
+	select {
+	case data := <-firstSub:
+		if !strings.Contains(string(data), `"kind":"ask_request"`) {
+			t.Fatalf("initial subscriber got %s, want ask_request", data)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for initial ask_request")
+	}
+
+	resp, err := http.Get(srv.URL + "/events")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("/events status = %d", resp.StatusCode)
+	}
+
+	replayed := make(chan string, 1)
+	go func() {
+		buf := make([]byte, 0, 4096)
+		tmp := make([]byte, 512)
+		for {
+			n, readErr := resp.Body.Read(tmp)
+			if n > 0 {
+				buf = append(buf, tmp[:n]...)
+				if strings.Contains(string(buf), `"kind":"ask_request"`) {
+					replayed <- string(buf)
+					return
+				}
+			}
+			if readErr != nil {
+				return
+			}
+		}
+	}()
+
+	select {
+	case <-replayed:
+	case <-time.After(2 * time.Second):
+		t.Fatal("late SSE attach never received replayed ask_request")
+	}
+
+	select {
+	case err := <-askDone:
+		t.Fatalf("ask resolved before the late client answered: %v", err)
+	default:
+	}
+
+	// Reconnect recovery must be connection-local: the existing subscriber
+	// must not receive the same prompt a second time.
+	select {
+	case data := <-firstSub:
+		t.Fatalf("existing subscriber got duplicate replay: %s", data)
+	default:
+	}
+
+	cancelAsk()
+	select {
+	case <-askDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("blocked ask did not exit after test cancellation")
+	}
+}
+
+// TestServeEventsReplayHandoffSerializesPromptEmission proves the controller's
+// attach handoff can register a subscriber and replay while prompt emission is
+// serialized, so a prompt cannot land between those two operations.
+func TestServeEventsReplayHandoffSerializesPromptEmission(t *testing.T) {
+	bc := NewBroadcaster()
+	ctrl := control.New(control.Options{Sink: bc})
+	ctrl.EnableInteractiveApproval()
+
+	askCtx, cancelAsk := context.WithCancel(context.Background())
+	defer cancelAsk()
+	taskDone := make(chan struct{})
+	var sub <-chan []byte
+	var cancelSub func()
+	ctrl.ReplayPendingPromptsWith(func() event.Sink {
+		sub, cancelSub = bc.Subscribe()
+		go func() {
+			_, _ = ctrl.Ask(askCtx, []event.AskQuestion{{
+				ID: "q1", Prompt: "pick one", Options: []event.AskOption{{Label: "A"}, {Label: "B"}},
+			}})
+			close(taskDone)
+		}()
+		return event.FuncSink(func(e event.Event) { bc.EmitTo(sub, e) })
+	})
+	defer cancelSub()
+
+	select {
+	case data := <-sub:
+		if !strings.Contains(string(data), `"kind":"ask_request"`) {
+			t.Fatalf("handoff subscriber got %s, want ask_request", data)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("handoff subscriber never received ask_request")
+	}
+	select {
+	case data := <-sub:
+		t.Fatalf("handoff subscriber got duplicate ask_request: %s", data)
+	default:
+	}
+
+	cancelAsk()
+	select {
+	case <-taskDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handoff ask did not exit after cancellation")
+	}
+}
+
+// TestServeEventsReplaysPendingApprovalOnAttach covers the actual approval
+// surface from #7643: a late browser must receive a parked ApprovalRequest and
+// be able to answer it through the serve HTTP endpoint.
+func TestServeEventsReplaysPendingApprovalOnAttach(t *testing.T) {
+	reg := tool.NewRegistry()
+	reg.Add(serveApprovalWriter{})
+	ag := agent.New(&serveApprovalProvider{}, reg, agent.NewSession(""), agent.Options{}, event.Discard)
+	bc := NewBroadcaster()
+	ctrl := control.New(control.Options{
+		Runner:   ag,
+		Executor: ag,
+		Sink:     bc,
+		Policy:   permission.New("ask", nil, nil, nil),
+	})
+	ctrl.EnableInteractiveApproval()
+	srv := httptest.NewServer(New(ctrl, bc, config.ServeConfig{}).Handler())
+	defer srv.Close()
+
+	runDone := make(chan error, 1)
+	go func() { runDone <- ctrl.Executor().Run(context.Background(), "write a file") }()
+
+	deadline := time.After(2 * time.Second)
+	for !ctrl.PendingPrompt() {
+		select {
+		case <-deadline:
+			t.Fatal("timed out waiting for parked approval")
+		default:
+			time.Sleep(5 * time.Millisecond)
+		}
+	}
+
+	resp, err := http.Get(srv.URL + "/events")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("/events status = %d", resp.StatusCode)
+	}
+
+	replayed := make(chan eventwire.Event, 1)
+	go func() {
+		buf := make([]byte, 0, 4096)
+		tmp := make([]byte, 512)
+		for {
+			n, readErr := resp.Body.Read(tmp)
+			if n > 0 {
+				buf = append(buf, tmp[:n]...)
+				if strings.Contains(string(buf), `"kind":"approval_request"`) {
+					frame := string(buf)
+					start := strings.Index(frame, "data: ")
+					if start < 0 {
+						return
+					}
+					end := strings.IndexByte(frame[start:], '\n')
+					if end < 0 {
+						end = len(frame) - start
+					}
+					var wire eventwire.Event
+					if json.Unmarshal([]byte(strings.TrimSpace(frame[start+len("data: "):start+end])), &wire) == nil {
+						replayed <- wire
+					}
+					return
+				}
+			}
+			if readErr != nil {
+				return
+			}
+		}
+	}()
+
+	var approval eventwire.Event
+	select {
+	case approval = <-replayed:
+	case <-time.After(2 * time.Second):
+		t.Fatal("late SSE attach never received replayed approval_request")
+	}
+	if approval.Kind != "approval_request" || approval.Approval == nil || approval.Approval.Tool != "serve_write" {
+		t.Fatalf("replayed approval = %+v, want serve_write approval_request", approval)
+	}
+
+	payload, err := json.Marshal(map[string]any{"id": approval.Approval.ID, "allow": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	req, err := http.NewRequest(http.MethodPost, srv.URL+"/approve", strings.NewReader(string(payload)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	answer, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	answer.Body.Close()
+	if answer.StatusCode != http.StatusNoContent {
+		t.Fatalf("/approve status = %d", answer.StatusCode)
+	}
+
+	select {
+	case err := <-runDone:
+		if err != nil {
+			t.Fatalf("executor run after approval: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("executor did not finish after approval")
+	}
+}