feat(execd): add command/session resume api

Author: Pangjiping

components/execd/pkg/runtime/types.go

@@ -24,8 +24,6 @@ import (
 )
 
 // ExecuteResultHook groups execution callbacks.
-// Eid is only assigned for stdout/stderr on run command (pipe tail) and run-in-session (bash stdout pipe);
-// other paths pass eid=0. Ids are allocated in runtime at pipe sync, not in HTTP/SSE writers.
 type ExecuteResultHook struct {
 	OnExecuteInit     func(context string)
 	OnExecuteResult   func(result map[string]any, count int)
@@ -48,20 +46,16 @@ type ExecuteCodeRequest struct {
 	Gid      *uint32           `json:"gid,omitempty"`
 	Hooks    ExecuteResultHook
 
-	// eventSeq assigns monotonic eids (1-based) for stdout/stderr on run command and bash session only.
 	eventSeq atomic.Uint64
 }
 
-// nextStdoutStderrEventID returns the next eid for stdout/stderr lines. Used only from run command
-// pipe tailers and bash session stdout; other callers should pass 0 into OnExecuteStdout/Stderr.
 func (req *ExecuteCodeRequest) nextStdoutStderrEventID() int64 {
 	if req == nil {
 		return 0
 	}
 	return int64(req.eventSeq.Add(1))
 }
 
-// wrapStdoutPipeHook wraps stdout delivery so eid is assigned when a line is flushed from the pipe tailer, not in SSE writes.
 func (req *ExecuteCodeRequest) wrapStdoutPipeHook() func(string) {
 	return func(text string) {
 		if text == "" || req.Hooks.OnExecuteStdout == nil {
@@ -72,7 +66,6 @@ func (req *ExecuteCodeRequest) wrapStdoutPipeHook() func(string) {
 	}
 }
 
-// wrapStderrPipeHook wraps stderr delivery so eid is assigned when a line is flushed from the pipe tailer, not in SSE writes.
 func (req *ExecuteCodeRequest) wrapStderrPipeHook() func(string) {
 	return func(text string) {
 		if text == "" || req.Hooks.OnExecuteStderr == nil {

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

@@ -321,6 +321,25 @@ func (c *CodeInterpretingController) RunInSession() {
 	time.Sleep(flag.ApiGracefulShutdownTimeout)
 }
 
+func (c *CodeInterpretingController) ResumeSessionStream() {
+	sessionID := c.ctx.Param("sessionId")
+	if sessionID == "" {
+		c.RespondError(
+			http.StatusBadRequest,
+			model.ErrorCodeMissingQuery,
+			"missing path parameter 'sessionId'",
+		)
+		return
+	}
+	_ = c.QueryInt64(c.ctx.Query(model.SessionResumeAfterEidQuery), 0)
+
+	c.RespondError(
+		http.StatusNotImplemented,
+		model.ErrorCodeNotImplemented,
+		"session stream resume is not implemented yet",
+	)
+}
+
 // DeleteSession deletes a bash session (delete_session API).
 func (c *CodeInterpretingController) DeleteSession() {
 	sessionID := c.ctx.Param("sessionId")

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

@@ -125,6 +125,21 @@ func (c *CodeInterpretingController) GetBackgroundCommandOutput() {
 	c.ctx.String(http.StatusOK, "%s", output)
 }
 
+func (c *CodeInterpretingController) ResumeCommandStream() {
+	commandID := c.ctx.Param("id")
+	if commandID == "" {
+		c.RespondError(http.StatusBadRequest, model.ErrorCodeInvalidRequest, "missing command execution id")
+		return
+	}
+	_ = c.QueryInt64(c.ctx.Query(model.CommandResumeAfterEidQuery), 0)
+
+	c.RespondError(
+		http.StatusNotImplemented,
+		model.ErrorCodeNotImplemented,
+		"command stream resume is not implemented yet",
+	)
+}
+
 func (c *CodeInterpretingController) buildExecuteCommandRequest(request model.RunCommandRequest) *runtime.ExecuteCodeRequest {
 	timeout := time.Duration(request.TimeoutMs) * time.Millisecond
 	if request.Background {

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

@@ -21,6 +21,9 @@ import (
 	"github.com/go-playground/validator/v10"
 )
 
+const CommandResumeAfterEidQuery = "after_eid"
+const SessionResumeAfterEidQuery = CommandResumeAfterEidQuery
+
 // RunCommandRequest represents a shell command execution request.
 type RunCommandRequest struct {
 	Command    string `json:"command" validate:"required"`

components/execd/pkg/web/model/error.go

@@ -26,6 +26,7 @@ const (
 	ErrorCodeFileNotFound        ErrorCode = "FILE_NOT_FOUND"
 	ErrorCodeUnknown             ErrorCode = "UNKNOWN"
 	ErrorCodeContextNotFound     ErrorCode = "CONTEXT_NOT_FOUND"
+	ErrorCodeNotImplemented      ErrorCode = "NOT_IMPLEMENTED"
 )
 
 type ErrorResponse struct {

components/execd/pkg/web/router.go

@@ -66,6 +66,7 @@ func NewRouter(accessToken string) *gin.Engine {
 	{
 		session.POST("", withCode(func(c *controller.CodeInterpretingController) { c.CreateSession() }))
 		session.POST("/:sessionId/run", withCode(func(c *controller.CodeInterpretingController) { c.RunInSession() }))
+		session.GET("/:sessionId/resume", withCode(func(c *controller.CodeInterpretingController) { c.ResumeSessionStream() }))
 		session.DELETE("/:sessionId", withCode(func(c *controller.CodeInterpretingController) { c.DeleteSession() }))
 	}
 
@@ -74,6 +75,7 @@ func NewRouter(accessToken string) *gin.Engine {
 		command.POST("", withCode(func(c *controller.CodeInterpretingController) { c.RunCommand() }))
 		command.DELETE("", withCode(func(c *controller.CodeInterpretingController) { c.InterruptCommand() }))
 		command.GET("/status/:id", withCode(func(c *controller.CodeInterpretingController) { c.GetCommandStatus() }))
+		command.GET("/:id/resume", withCode(func(c *controller.CodeInterpretingController) { c.ResumeCommandStream() }))
 		command.GET("/:id/logs", withCode(func(c *controller.CodeInterpretingController) { c.GetBackgroundCommandOutput() }))
 	}