fix(execd): flush resume buffer tail after SSE attach

Author: Pangjiping

components/execd/pkg/web/controller/command.go

@@ -165,8 +165,12 @@ func (c *CodeInterpretingController) ResumeCommandStream() {
 	}
 
 	c.setupSSEResponse()
+	lastReplayMaxEid := afterEid
 	for _, ev := range events {
 		c.writeSingleEvent("ResumeBuffer", ev.Payload, false, fmt.Sprintf("buffer eid=%d", ev.EID), 0)
+		if ev.EID > lastReplayMaxEid {
+			lastReplayMaxEid = ev.EID
+		}
 	}
 
 	st2, _ := codeRunner.GetCommandStatus(commandID)
@@ -187,6 +191,9 @@ func (c *CodeInterpretingController) ResumeCommandStream() {
 		return
 	}
 
+	// Catch up events appended while the snapshot slice was replayed (holder still nil); same mutex as writeFrame.
+	h.flushResumeTail(commandID, lastReplayMaxEid)
+
 	select {
 	case <-h.waitDone():
 	case <-c.ctx.Request.Context().Done():

components/execd/pkg/web/controller/command_stream.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Alibaba Group Holding Ltd.
+// Copyright 2026 Alibaba Group Holding Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -167,6 +167,37 @@ func (r *streamRegistry) writeSSE(id string, data []byte, bufEid int64, handler,
 	h.writeFrame(data, bufEid, handler, summary)
 }
 
+// flushResumeTail writes all buffered events with EID > afterEid to the current holder while holding h.mu.
+// Live writeFrame calls block on the same mutex, so chunks appended only to the ring during the initial
+// snapshot replay cannot be missed on this connection (see ResumeCommandStream).
+func (h *streamHub) flushResumeTail(commandID string, afterEid int64) {
+	if h == nil || h.holder == nil {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	tail, ok := resumeBuffer.EventsAfter(commandID, afterEid)
+	if !ok || len(tail) == 0 {
+		return
+	}
+	writer := h.holder.writer
+	for _, ev := range tail {
+		payload := append(append([]byte(nil), ev.Payload...), '\n', '\n')
+		n, err := writer.Write(payload)
+		if err == nil && n != len(payload) {
+			err = io.ErrShortWrite
+		}
+		if err != nil {
+			log.Error("flushResumeTail: write eid=%d: %v", ev.EID, err)
+			return
+		}
+		if flusher, ok := writer.(http.Flusher); ok {
+			flusher.Flush()
+		}
+	}
+}
+
 func (h *streamHub) writeFrame(data []byte, bufEid int64, handler, summary string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()

components/execd/pkg/web/controller/command_stream_test.go

@@ -0,0 +1,49 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controller
+
+import (
+	"context"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFlushResumeTail_NilHubNoPanic(t *testing.T) {
+	var h *streamHub
+	h.flushResumeTail("any", 0)
+}
+
+func TestFlushResumeTail_WritesBufferedEvents(t *testing.T) {
+	cmdID := "flush-resume-test-cmd"
+	payload := []byte(`{"type":"stdout","eid":1}`)
+	require.NoError(t, resumeBuffer.Append(cmdID, 1, payload))
+
+	w := httptest.NewRecorder()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	h := &streamHub{
+		streamID: cmdID,
+		done:     make(chan struct{}),
+		holder:   &streamHolder{writer: w, ctx: ctx},
+	}
+	h.flushResumeTail(cmdID, 0)
+
+	body := w.Body.String()
+	require.Contains(t, body, "stdout")
+	require.Contains(t, body, `"eid":1`)
+}