FEATURE: SSE Resumption in execd

.github/workflows/execd-test.yml

@@ -109,6 +109,8 @@ jobs:
 
           sleep 5
           python3 tests/smoke_api.py
+          python3 tests/command_resume_test.py
+
       - name: SIGTERM forward test
         if: matrix.os == 'ubuntu-latest'
         working-directory: components/execd

components/execd/RELEASE_NOTES.md

@@ -109,8 +109,8 @@ Thanks to these contributors ❤️
 - @csdbianhua
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.8
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.8
 
 # components/execd 1.0.7
 
@@ -144,8 +144,8 @@ Thanks to these contributors ❤️
 - @dependabot
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.7
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.7
 
 # components/execd 1.0.6
 
@@ -167,8 +167,8 @@ Thanks to these contributors ❤️
 - @dependabot
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.6
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.6
 
 # components/execd 1.0.5
 
@@ -185,8 +185,8 @@ Thanks to these contributors ❤️
 - @Pangjiping
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.5
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.5
 
 # components/execd 1.0.4
 
@@ -209,8 +209,8 @@ Thanks to these contributors ❤️
 - @ninan-nn
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.4
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.4
 
 # components/execd 1.0.3
 
@@ -233,8 +233,8 @@ Thanks to these contributors ❤️
 - @jwx0925
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.3
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.3
 
 # components/execd 1.0.2
 
@@ -262,8 +262,8 @@ Thanks to these contributors ❤️
 - @ninan-nn
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.2
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.2
 
 # components/execd 1.0.1
 
@@ -295,8 +295,8 @@ Thanks to these contributors ❤️
 - @jwx0925
 
 ---
-- Docker Hub: opensandbox/execd:v1.0.9
-- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.9
+- Docker Hub: opensandbox/execd:v1.0.1
+- Aliyun Registry: sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/execd:v1.0.1
 
 # components/execd 1.0.0
 

components/execd/pkg/runtime/bash_session.go

@@ -252,7 +252,7 @@ func (s *bashSession) run(ctx context.Context, request *ExecuteCodeRequest) erro
 				continue
 			}
 			if request.Hooks.OnExecuteStdout != nil {
-				request.Hooks.OnExecuteStdout(line)
+				request.Hooks.OnExecuteStdout(request.NextEventID(), line)
 			}
 		}
 	}

components/execd/pkg/runtime/bash_session_test.go

@@ -56,7 +56,7 @@ func TestBashSession_NonZeroExitEmitsError(t *testing.T) {
 		Timeout:  5 * time.Second,
 		Hooks: ExecuteResultHook{
 			OnExecuteInit:   func(s string) { sessionID = s },
-			OnExecuteStdout: func(s string) { stdoutLine = s },
+			OnExecuteStdout: func(_ int64, s string) { stdoutLine = s },
 			OnExecuteError:  func(err *execute.ErrorOutput) { errCh <- err },
 			OnExecuteComplete: func(_ time.Duration) {
 				completeCh <- struct{}{}
@@ -105,7 +105,7 @@ func TestBashSession_envAndExitCode(t *testing.T) {
 			require.Equal(t, session.config.Session, ctx, "unexpected session in OnExecuteInit")
 			initCalls++
 		},
-		OnExecuteStdout: func(text string) {
+		OnExecuteStdout: func(_ int64, text string) {
 			t.Log(text)
 			stdoutLines = append(stdoutLines, text)
 		},
@@ -178,7 +178,7 @@ func TestBashSession_envLargeOutputChained(t *testing.T) {
 			require.Equal(t, session.config.Session, ctx, "unexpected session in OnExecuteInit")
 			initCalls++
 		},
-		OnExecuteStdout: func(text string) {
+		OnExecuteStdout: func(_ int64, text string) {
 			t.Log(text)
 			stdoutLines = append(stdoutLines, text)
 		},
@@ -223,7 +223,7 @@ func TestBashSession_cwdPersistsWithoutOverride(t *testing.T) {
 	targetDir := t.TempDir()
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}
@@ -265,7 +265,7 @@ func TestBashSession_requestCwdOverridesAfterCd(t *testing.T) {
 
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}
@@ -307,7 +307,7 @@ func TestBashSession_envDumpNotLeakedWhenNoTrailingNewline(t *testing.T) {
 
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}
@@ -335,7 +335,7 @@ func TestBashSession_envDumpNotLeakedWhenNoOutput(t *testing.T) {
 
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}
@@ -366,7 +366,7 @@ func TestBashSession_heredoc(t *testing.T) {
 	t.Cleanup(func() { _ = controller.DeleteBashSession(sessionID) })
 
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			fmt.Printf("[stdout] %s\n", line)
 		},
 		OnExecuteComplete: func(d time.Duration) {
@@ -419,7 +419,7 @@ func TestBashSession_execReplacesShell(t *testing.T) {
 
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}
@@ -459,7 +459,7 @@ func TestBashSession_complexExec(t *testing.T) {
 
 	var stdoutLines []string
 	hooks := ExecuteResultHook{
-		OnExecuteStdout: func(line string) {
+		OnExecuteStdout: func(_ int64, line string) {
 			stdoutLines = append(stdoutLines, line)
 		},
 	}

components/execd/pkg/runtime/command.go

@@ -134,11 +134,11 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
 	wg.Add(2)
 	safego.Go(func() {
 		defer wg.Done()
-		c.tailStdPipe(stdoutPath, request.Hooks.OnExecuteStdout, done)
+		c.tailStdPipe(stdoutPath, request.wrapStdoutPipeHook(), done)
 	})
 	safego.Go(func() {
 		defer wg.Done()
-		c.tailStdPipe(stderrPath, request.Hooks.OnExecuteStderr, done)
+		c.tailStdPipe(stderrPath, request.wrapStderrPipeHook(), done)
 	})
 
 	err = cmd.Start()

components/execd/pkg/runtime/command_test.go

@@ -133,10 +133,10 @@ func TestRunCommand_Echo(t *testing.T) {
 		Timeout: 5 * time.Second,
 		Hooks: ExecuteResultHook{
 			OnExecuteInit: func(s string) { sessionID = s },
-			OnExecuteStdout: func(s string) {
+			OnExecuteStdout: func(_ int64, s string) {
 				stdoutLines = append(stdoutLines, s)
 			},
-			OnExecuteStderr: func(s string) {
+			OnExecuteStderr: func(_ int64, s string) {
 				stderrLines = append(stderrLines, s)
 			},
 			OnExecuteError: func(err *execute.ErrorOutput) {
@@ -188,8 +188,8 @@ func TestRunCommand_Error(t *testing.T) {
 		Timeout: 5 * time.Second,
 		Hooks: ExecuteResultHook{
 			OnExecuteInit:   func(s string) { sessionID = s },
-			OnExecuteStdout: func(s string) { stdoutLines = append(stdoutLines, s) },
-			OnExecuteStderr: func(s string) { stderrLines = append(stderrLines, s) },
+			OnExecuteStdout: func(_ int64, s string) { stdoutLines = append(stdoutLines, s) },
+			OnExecuteStderr: func(_ int64, s string) { stderrLines = append(stderrLines, s) },
 			OnExecuteError: func(err *execute.ErrorOutput) {
 				gotErr = err
 				completeCh <- struct{}{}
@@ -241,8 +241,8 @@ func TestRunCommand_ExpandsHomeInCwd(t *testing.T) {
 		Timeout: 5 * time.Second,
 		Hooks: ExecuteResultHook{
 			OnExecuteInit:   func(_ string) {},
-			OnExecuteStdout: func(s string) { stdoutLines = append(stdoutLines, s) },
-			OnExecuteStderr: func(_ string) {},
+			OnExecuteStdout: func(_ int64, s string) { stdoutLines = append(stdoutLines, s) },
+			OnExecuteStderr: func(_ int64, _ string) {},
 			OnExecuteError: func(err *execute.ErrorOutput) {
 				require.Failf(t, "unexpected error hook", "%+v", err)
 			},
@@ -303,8 +303,8 @@ func TestRunCommand_ExpandsCwdFromRequestEnvWithHigherPriority(t *testing.T) {
 		},
 		Hooks: ExecuteResultHook{
 			OnExecuteInit:   func(_ string) {},
-			OnExecuteStdout: func(s string) { stdoutLines = append(stdoutLines, s) },
-			OnExecuteStderr: func(_ string) {},
+			OnExecuteStdout: func(_ int64, s string) { stdoutLines = append(stdoutLines, s) },
+			OnExecuteStderr: func(_ int64, _ string) {},
 			OnExecuteError: func(err *execute.ErrorOutput) {
 				gotErr = err
 			},

components/execd/pkg/runtime/command_windows.go

@@ -58,10 +58,10 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
 
 	done := make(chan struct{}, 1)
 	safego.Go(func() {
-		c.tailStdPipe(c.stdoutFileName(session), request.Hooks.OnExecuteStdout, done)
+		c.tailStdPipe(c.stdoutFileName(session), request.wrapStdoutPipeHook(), done)
 	})
 	safego.Go(func() {
-		c.tailStdPipe(c.stderrFileName(session), request.Hooks.OnExecuteStderr, done)
+		c.tailStdPipe(c.stderrFileName(session), request.wrapStderrPipeHook(), done)
 	})
 
 	err = cmd.Start()

components/execd/pkg/runtime/jupyter.go

@@ -122,9 +122,16 @@ func dispatchExecutionResultHooks(request *ExecuteCodeRequest, result *execute.E
 	for _, stream := range result.Stream {
 		switch stream.Name {
 		case execute.StreamStdout:
-			request.Hooks.OnExecuteStdout(stream.Text)
+			if stream.Text != "" && request.Hooks.OnExecuteStdout != nil {
+				eid := request.NextEventID()
+				request.Hooks.OnExecuteStdout(eid, stream.Text)
+			}
 		case execute.StreamStderr:
-			request.Hooks.OnExecuteStderr(stream.Text)
+			if stream.Text != "" && request.Hooks.OnExecuteStderr != nil {
+				eid := request.NextEventID()
+				request.Hooks.OnExecuteStderr(eid, stream.Text)
+			}
+		default:
 		}
 	}
 }

components/execd/pkg/runtime/types.go

@@ -17,6 +17,7 @@ package runtime
 import (
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/alibaba/opensandbox/execd/pkg/jupyter/execute"
@@ -27,8 +28,8 @@ type ExecuteResultHook struct {
 	OnExecuteInit     func(context string)
 	OnExecuteResult   func(result map[string]any, count int)
 	OnExecuteStatus   func(status string)
-	OnExecuteStdout   func(stdout string) //nolint:predeclared
-	OnExecuteStderr   func(stderr string) //nolint:predeclared
+	OnExecuteStdout   func(eid int64, stdout string) //nolint:predeclared
+	OnExecuteStderr   func(eid int64, stderr string) //nolint:predeclared
 	OnExecuteError    func(err *execute.ErrorOutput)
 	OnExecuteComplete func(executionTime time.Duration)
 }
@@ -44,6 +45,35 @@ type ExecuteCodeRequest struct {
 	Uid      *uint32           `json:"uid,omitempty"`
 	Gid      *uint32           `json:"gid,omitempty"`
 	Hooks    ExecuteResultHook
+
+	eventSeq atomic.Uint64
+}
+
+func (req *ExecuteCodeRequest) NextEventID() int64 {
+	if req == nil {
+		return 0
+	}
+	return int64(req.eventSeq.Add(1))
+}
+
+func (req *ExecuteCodeRequest) wrapStdoutPipeHook() func(string) {
+	return func(text string) {
+		if text == "" || req.Hooks.OnExecuteStdout == nil {
+			return
+		}
+		eid := req.NextEventID()
+		req.Hooks.OnExecuteStdout(eid, text)
+	}
+}
+
+func (req *ExecuteCodeRequest) wrapStderrPipeHook() func(string) {
+	return func(text string) {
+		if text == "" || req.Hooks.OnExecuteStderr == nil {
+			return
+		}
+		eid := req.NextEventID()
+		req.Hooks.OnExecuteStderr(eid, text)
+	}
 }
 
 // SetDefaultHooks installs stdout logging fallbacks for unset hooks.
@@ -55,10 +85,10 @@ func (req *ExecuteCodeRequest) SetDefaultHooks() {
 		req.Hooks.OnExecuteStatus = func(status string) { fmt.Printf("OnExecuteStatus: %s\n", status) }
 	}
 	if req.Hooks.OnExecuteStdout == nil {
-		req.Hooks.OnExecuteStdout = func(stdout string) { fmt.Printf("OnExecuteStdout: %s\n", stdout) }
+		req.Hooks.OnExecuteStdout = func(eid int64, stdout string) { fmt.Printf("OnExecuteStdout: eid=%d %s\n", eid, stdout) }
 	}
 	if req.Hooks.OnExecuteStderr == nil {
-		req.Hooks.OnExecuteStderr = func(stderr string) { fmt.Printf("OnExecuteStderr: %s\n", stderr) }
+		req.Hooks.OnExecuteStderr = func(eid int64, stderr string) { fmt.Printf("OnExecuteStderr: eid=%d %s\n", eid, stderr) }
 	}
 	if req.Hooks.OnExecuteError == nil {
 		req.Hooks.OnExecuteError = func(err *execute.ErrorOutput) { fmt.Printf("OnExecuteError: %++v\n", err) }

components/execd/pkg/sbuf/config.go

@@ -0,0 +1,37 @@
+// 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 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 sbuf
+
+// Config controls per-stream bounds and append policy.
+type Config struct {
+	// MaxEvents is the maximum number of events retained per stream. Oldest events are dropped when exceeded.
+	// Zero defaults to DefaultMaxEvents.
+	MaxEvents int
+	// MaxBytes is the approximate upper bound on total payload bytes per stream (sum of len(Payload)).
+	// Oldest events are dropped until under the limit. Zero means no byte limit.
+	MaxBytes int64
+	// StrictMonotonic rejects Append when eid <= last eid for that stream. Recommended for execd SSE eids.
+	StrictMonotonic bool
+}
+
+const DefaultMaxEvents = 1024
+
+func (c *Config) normalized() Config {
+	out := *c
+	if out.MaxEvents <= 0 {
+		out.MaxEvents = DefaultMaxEvents
+	}
+	return out
+}