fix(execd): preserve blank lines in command stdout SSE stream (#902)

* fix(execd): preserve blank lines in command stdout SSE stream

readFromPos previously skipped consecutive line terminators when the
buffer was empty, dropping standalone newline lines from the SSE output.
Emit "\n" for blank lines while keeping the existing line-content emit
behavior (terminator is still stripped from non-empty lines). \r\n pairs
are coalesced via lastWasCR to avoid duplicate blank emits.

Adds a unit test covering blank lines, leading blank, and CRLF, and an
end-to-end smoke test that runs `printf 'a\n\nb\n\n\nc\n'` and asserts
the stdout event sequence preserves the blanks.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(execd): wait for stdout/stderr tail to flush before complete on Windows

Windows runCommand fired OnExecuteComplete immediately after closing the
done channel, racing the tail goroutines that emit pending stdout/stderr
SSE events. Clients that break on execution_complete then missed final
output, e.g. the blank-line smoke assertion saw an empty sequence while
the server log showed the events were emitted.

Mirror the Linux path: track the tail goroutines with a sync.WaitGroup
and wg.Wait() after close(done) so all buffered output drains before
the completion event is sent. Also covers the cmd.Start failure path.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(execd): persist CRLF state across tail polls

readFromPos previously declared lastWasCR as a per-call local, so the
CRLF coalescing only worked when \r and \n were read in the same
invocation. When tailStdPipe polls a writer that flushes \r before \n
(common on Windows/cmd), the \r and \n can land in separate polls; the
second call starts with lastWasCR=false and emits a spurious "\n" blank
line for the trailing \n. A bare blank \r\n line split across polls
would surface as two blanks.

Hoist the state into tailStdPipe and thread it through readFromPos so
the CR detection survives between polls. Add regression tests covering
split CRLF after content and split blank CRLF.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(execd): make blank-lines smoke command cross-platform

Replace POSIX printf with single-quoted format string with a python -c
one-liner. cmd /C does not strip single quotes, so the previous command
only worked on Windows runners that happened to have Git for Windows in
PATH (MSYS2 argv pre-processing strips the quotes); on a bare Windows
sandbox the smoke would fail before reaching the filesystem checks.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(execd): branch blank-lines smoke per platform

The previous python -c attempt produced a SyntaxError on Windows because
Go's syscall.EscapeArg wraps the cmd /C argument in quotes, escaping the
inner quotes as \". cmd /C strips the outer quotes (rule 2 of its parser)
but leaves the literal \" inside, and MSVCRT's argv parser then treats
\" as a literal double-quote character without toggling quote state, so
the first embedded space terminates argv[2] and python sees an
unterminated string literal.

Use a cmd-native echo chain on Windows (no inner quotes, & is sequential)
and keep POSIX printf on Linux/macOS. The execd reader collapses CRLF to
LF, so both platforms yield the same event sequence.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: Pangjiping

components/execd/pkg/runtime/command_common.go

@@ -32,13 +32,14 @@ func (c *Controller) tailStdPipe(file string, onExecute func(text string), done
 	defer ticker.Stop()
 
 	mutex := &sync.Mutex{}
+	var lastWasCR bool
 	for {
 		select {
 		case <-done:
-			c.readFromPos(mutex, file, lastPos, onExecute, true)
+			c.readFromPos(mutex, file, lastPos, onExecute, true, &lastWasCR)
 			return
 		case <-ticker.C:
-			newPos := c.readFromPos(mutex, file, lastPos, onExecute, false)
+			newPos := c.readFromPos(mutex, file, lastPos, onExecute, false, &lastWasCR)
 			lastPos = newPos
 		}
 	}
@@ -104,7 +105,9 @@ func (c *Controller) combinedOutputFileName(session string) string {
 }
 
 // readFromPos streams new content from a file starting at startPos.
-func (c *Controller) readFromPos(mutex *sync.Mutex, filepath string, startPos int64, onExecute func(string), flushIncomplete bool) int64 {
+// lastWasCR persists CRLF detection across calls so a \r\n pair split between
+// two polls does not surface a spurious blank line for the trailing \n.
+func (c *Controller) readFromPos(mutex *sync.Mutex, filepath string, startPos int64, onExecute func(string), flushIncomplete bool, lastWasCR *bool) int64 {
 	if !mutex.TryLock() {
 		return -1
 	}
@@ -121,6 +124,15 @@ func (c *Controller) readFromPos(mutex *sync.Mutex, filepath string, startPos in
 	reader := bufio.NewReader(file)
 	var buffer bytes.Buffer
 	var currentPos int64 = startPos
+	cr := false
+	if lastWasCR != nil {
+		cr = *lastWasCR
+	}
+	defer func() {
+		if lastWasCR != nil {
+			*lastWasCR = cr
+		}
+	}()
 
 	for {
 		b, err := reader.ReadByte()
@@ -138,15 +150,22 @@ func (c *Controller) readFromPos(mutex *sync.Mutex, filepath string, startPos in
 
 		// Check if it's a line terminator (\n or \r)
 		if b == '\n' || b == '\r' {
-			// If buffer has content, output this line
-			if buffer.Len() > 0 {
+			switch {
+			case buffer.Len() > 0:
+				// Flush the line content without the terminator
 				onExecute(buffer.String())
 				buffer.Reset()
+			case b == '\n' && cr:
+				// Second half of a \r\n pair; already emitted on \r
+			default:
+				// Standalone blank line; surface it so callers see the gap
+				onExecute("\n")
 			}
-			// Skip line terminator
+			cr = (b == '\r')
 			continue
 		}
 
+		cr = false
 		buffer.WriteByte(b)
 	}
 

components/execd/pkg/runtime/command_test.go

@@ -42,7 +42,7 @@ func TestReadFromPos_SplitsOnCRAndLF(t *testing.T) {
 
 	var got []string
 	c := &Controller{}
-	nextPos := c.readFromPos(mutex, logFile, 0, func(s string) { got = append(got, s) }, false)
+	nextPos := c.readFromPos(mutex, logFile, 0, func(s string) { got = append(got, s) }, false, nil)
 
 	want := []string{"line1", "prog 10%", "prog 20%", "prog 30%", "last"}
 	require.Len(t, got, len(want))
@@ -59,7 +59,7 @@ func TestReadFromPos_SplitsOnCRAndLF(t *testing.T) {
 	_ = f.Close()
 
 	got = got[:0]
-	c.readFromPos(mutex, logFile, nextPos, func(s string) { got = append(got, s) }, false)
+	c.readFromPos(mutex, logFile, nextPos, func(s string) { got = append(got, s) }, false, nil)
 	want = []string{"tail1", "tail2"}
 	require.Len(t, got, len(want))
 	for i := range want {
@@ -77,7 +77,7 @@ func TestReadFromPos_LongLine(t *testing.T) {
 
 	var got []string
 	c := &Controller{}
-	c.readFromPos(&sync.Mutex{}, logFile, 0, func(s string) { got = append(got, s) }, false)
+	c.readFromPos(&sync.Mutex{}, logFile, 0, func(s string) { got = append(got, s) }, false, nil)
 
 	require.Len(t, got, 1, "expected one token")
 	require.Equal(t, strings.TrimSuffix(longLine, "\n"), got[0], "long line mismatch")
@@ -98,15 +98,86 @@ func TestReadFromPos_FlushesTrailingLine(t *testing.T) {
 	}
 
 	// First read: should only get complete lines with newlines
-	pos := c.readFromPos(mutex, file, 0, onExecute, false)
+	pos := c.readFromPos(mutex, file, 0, onExecute, false, nil)
 	assert.GreaterOrEqual(t, pos, int64(0))
 	assert.Equal(t, []string{"line1"}, lines)
 
 	// Flush at end: should output the last line (without newline)
-	c.readFromPos(mutex, file, pos, onExecute, true)
+	c.readFromPos(mutex, file, pos, onExecute, true, nil)
 	assert.Equal(t, []string{"line1", "lastline-without-newline"}, lines)
 }
 
+func TestReadFromPos_PreservesBlankLines(t *testing.T) {
+	tmp := t.TempDir()
+	logFile := filepath.Join(tmp, "stdout.log")
+
+	// Mix of single newlines, consecutive blank lines, leading blank, and CRLF.
+	initial := "a\n\nb\n\n\nc\n\r\nd\n"
+	require.NoError(t, os.WriteFile(logFile, []byte(initial), 0o644))
+
+	var got []string
+	c := &Controller{}
+	c.readFromPos(&sync.Mutex{}, logFile, 0, func(s string) { got = append(got, s) }, false, nil)
+
+	want := []string{"a", "\n", "b", "\n", "\n", "c", "\n", "d"}
+	require.Equal(t, want, got)
+}
+
+// TestReadFromPos_CRLFAcrossPolls ensures a \r\n pair that arrives in two
+// successive polls does not emit a spurious blank line for the trailing \n.
+// Reproduces the regression on Windows/cmd writers that flush \r before \n.
+func TestReadFromPos_CRLFAcrossPolls(t *testing.T) {
+	tmp := t.TempDir()
+	logFile := filepath.Join(tmp, "stdout.log")
+
+	require.NoError(t, os.WriteFile(logFile, []byte("a\r"), 0o644))
+
+	var got []string
+	c := &Controller{}
+	mutex := &sync.Mutex{}
+	var lastWasCR bool
+	pos := c.readFromPos(mutex, logFile, 0, func(s string) { got = append(got, s) }, false, &lastWasCR)
+	require.Equal(t, []string{"a"}, got)
+	require.True(t, lastWasCR, "CR state must persist for next poll")
+
+	f, err := os.OpenFile(logFile, os.O_APPEND|os.O_WRONLY, 0o644)
+	require.NoError(t, err)
+	_, err = f.WriteString("\nb\n")
+	require.NoError(t, err)
+	_ = f.Close()
+
+	got = got[:0]
+	c.readFromPos(mutex, logFile, pos, func(s string) { got = append(got, s) }, false, &lastWasCR)
+	require.Equal(t, []string{"b"}, got, "trailing \\n of split CRLF must not emit a blank line")
+}
+
+// TestReadFromPos_BlankCRLFAcrossPolls ensures a blank \r\n line split across
+// polls is emitted as a single blank, not duplicated.
+func TestReadFromPos_BlankCRLFAcrossPolls(t *testing.T) {
+	tmp := t.TempDir()
+	logFile := filepath.Join(tmp, "stdout.log")
+
+	require.NoError(t, os.WriteFile(logFile, []byte("\r"), 0o644))
+
+	var got []string
+	c := &Controller{}
+	mutex := &sync.Mutex{}
+	var lastWasCR bool
+	pos := c.readFromPos(mutex, logFile, 0, func(s string) { got = append(got, s) }, false, &lastWasCR)
+	require.Equal(t, []string{"\n"}, got)
+	require.True(t, lastWasCR)
+
+	f, err := os.OpenFile(logFile, os.O_APPEND|os.O_WRONLY, 0o644)
+	require.NoError(t, err)
+	_, err = f.WriteString("\n")
+	require.NoError(t, err)
+	_ = f.Close()
+
+	got = got[:0]
+	c.readFromPos(mutex, logFile, pos, func(s string) { got = append(got, s) }, false, &lastWasCR)
+	require.Empty(t, got, "trailing \\n of split blank CRLF must not emit a second blank")
+}
+
 func TestRunCommand_Echo(t *testing.T) {
 	if goruntime.GOOS == "windows" {
 		t.Skip("bash not available on windows")

components/execd/pkg/runtime/command_windows.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"os/exec"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/alibaba/opensandbox/execd/pkg/jupyter/execute"
@@ -57,15 +58,21 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
 	cmd.Env = mergeEnvs(os.Environ(), extraEnv)
 
 	done := make(chan struct{}, 1)
+	var wg sync.WaitGroup
+	wg.Add(2)
 	safego.Go(func() {
+		defer wg.Done()
 		c.tailStdPipe(c.stdoutFileName(session), request.Hooks.OnExecuteStdout, done)
 	})
 	safego.Go(func() {
+		defer wg.Done()
 		c.tailStdPipe(c.stderrFileName(session), request.Hooks.OnExecuteStderr, done)
 	})
 
 	err = cmd.Start()
 	if err != nil {
+		close(done)
+		wg.Wait()
 		request.Hooks.OnExecuteError(&execute.ErrorOutput{EName: "CommandExecError", EValue: err.Error()})
 		log.Error("CommandExecError: error starting commands: %v", err)
 		return nil
@@ -80,6 +87,7 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
 
 	err = cmd.Wait()
 	close(done)
+	wg.Wait()
 	if err != nil {
 		var eName, eValue string
 		var traceback []string

components/execd/tests/smoke_api.py

@@ -91,6 +91,58 @@ def fetch_logs(cmd_id: str, cursor: int = 0):
     return r.text, r.headers.get("EXECD-COMMANDS-TAIL-CURSOR")
 
 
+def run_command_blank_lines():
+    """
+    Foreground command whose stdout contains consecutive newlines must surface
+    blank-line events instead of dropping them. Regression test for the
+    readFromPos fix that preserves empty lines (a\n\nb -> ["a", "\n", "b"]).
+    """
+    url = f"{BASE_URL}/command"
+    # Pick a shell-native command per platform so the regression covers both
+    # POSIX (LF-only) and Windows cmd (CRLF) byte streams without depending on
+    # Git for Windows / MSYS argv mangling. The execd reader collapses CRLF to
+    # LF, so both produce ["a", "\n", "b", "\n", "\n", "c"].
+    if os.name == "nt":
+        # cmd /C echo chain: each segment writes "<text>\r\n"; "echo." writes
+        # a bare "\r\n". Order is deterministic because "&" is sequential.
+        command = "echo a&echo.&echo b&echo.&echo.&echo c"
+    else:
+        # printf emits exact bytes: a\n\nb\n\n\nc\n
+        command = "printf 'a\\n\\nb\\n\\n\\nc\\n'"
+    payload = {
+        "command": command,
+        "background": False,
+    }
+
+    stdout_texts = []
+    saw_complete = False
+    with session.post(url, json=payload, stream=True, timeout=15) as resp:
+        expect(resp.status_code == 200, f"SSE start failed: {resp.status_code} {resp.text}")
+        for line in resp.iter_lines():
+            if not line:
+                continue
+            try:
+                if line.startswith(b"data:"):
+                    data = json.loads(line[len(b"data:") :].decode())
+                else:
+                    data = json.loads(line.decode())
+            except Exception:
+                continue
+            event_type = data.get("type")
+            if event_type == "stdout":
+                stdout_texts.append(data.get("text", ""))
+            elif event_type == "execution_complete":
+                saw_complete = True
+                break
+
+    expect(saw_complete, "did not observe execution_complete")
+    want = ["a", "\n", "b", "\n", "\n", "c"]
+    expect(
+        stdout_texts == want,
+        f"blank-line stdout sequence mismatch: got {stdout_texts!r}, want {want!r}",
+    )
+
+
 def sse_disconnect_should_stop_ping():
     """
     Open an SSE stream for a long-running command, receive init, then close the
@@ -248,6 +300,9 @@ def main():
     sse_disconnect_should_stop_ping()
     print("[+] SSE disconnect handled")
 
+    run_command_blank_lines()
+    print("[+] run_command preserves blank lines")
+
     cmd_id = sse_get_command_id()
     print(f"[+] command id: {cmd_id}")