Added test for handling stray output in headerPrefixPipe and refactored header parsing logic to skip non-header lines

Author: m-messer

internal/execution/supervisor/adapter_rpc_pipe.go

@@ -48,53 +48,40 @@ func (h *headerPrefixPipe) Read(p []byte) (int, error) {
 		h.reader = bufio.NewReader(h.stdio)
 	}
 
-	// read headers
-	headers := ""
+	// Scan lines until we find Content-Length:, skipping any stray output.
+	var contentLength int
 	for {
 		line, err := h.reader.ReadString('\n')
 		if err != nil {
 			return 0, err
 		}
-
-		headers += line
-
-		// Detect the end of headers with double CRLF
-		if strings.HasSuffix(headers, "\r\n\r\n") {
-			break
+		line = strings.TrimRight(line, "\r\n")
+		if !strings.HasPrefix(line, "Content-Length:") {
+			continue
+		}
+		parts := strings.SplitN(line, ":", 2)
+		if len(parts) != 2 {
+			return 0, fmt.Errorf("malformed Content-Length header")
 		}
+		v, err := strconv.Atoi(strings.TrimSpace(parts[1]))
+		if err != nil {
+			return 0, fmt.Errorf("invalid Content-Length value: %s", parts[1])
+		}
+		contentLength = v
+		break
 	}
-	headers = strings.TrimSpace(headers)
-
-	// get content-length value
-	var contentLength int
-	lines := strings.Split(headers, "\r\n")
-	for _, line := range lines {
-		line = strings.TrimSpace(line)
-
-		if strings.HasPrefix(line, "Content-Length:") {
-			parts := strings.SplitN(line, ":", 2)
-			if len(parts) != 2 {
-				return 0, fmt.Errorf("malformed Content-Length header")
-			}
 
-			lengthStr := strings.TrimSpace(parts[1])
-
-			if lengthValue, err := strconv.Atoi(lengthStr); err != nil {
-				return 0, fmt.Errorf("invalid Content-Length value: %s", lengthStr)
-			} else {
-				contentLength = lengthValue
-			}
-
-			// found the content-length
+	// Drain remaining header lines until the blank separator.
+	for {
+		line, err := h.reader.ReadString('\n')
+		if err != nil {
+			return 0, err
+		}
+		if strings.TrimRight(line, "\r\n") == "" {
 			break
-
 		}
 	}
 
-	if contentLength == 0 {
-		return 0, fmt.Errorf("Content-Length header not found or zero")
-	}
-
 	// Read exactly contentLength bytes
 	content := make([]byte, contentLength)
 	n, err := io.ReadFull(h.reader, content)

internal/execution/supervisor/adapter_rpc_pipe_test.go

@@ -107,6 +107,26 @@ func TestHeaderPrefixPipe_MultipleMessages(t *testing.T) {
 	assert.Equal(t, msg2, readBuf[:n])
 }
 
+func TestHeaderPrefixPipe_SkipsStrayOutputBeforeContentLength(t *testing.T) {
+	buf := newRwc()
+	pipe := &headerPrefixPipe{stdio: buf}
+
+	msg := []byte(`{"jsonrpc":"2.0","id":1,"result":"ok"}`)
+
+	// Stray lines an evaluation function might emit on stdout before the
+	// server loop starts (e.g. model-loading logs, library init messages).
+	stray := "Loading model...\nReady.\n"
+	header := fmt.Sprintf("Content-Length: %d\r\n\r\n", len(msg))
+	buf.(*rwc).Buffer.Write([]byte(stray))
+	buf.(*rwc).Buffer.Write([]byte(header))
+	buf.(*rwc).Buffer.Write(msg)
+
+	readBuf := make([]byte, 512)
+	n, err := pipe.Read(readBuf)
+	require.NoError(t, err)
+	assert.Equal(t, msg, readBuf[:n])
+}
+
 func TestHeaderPrefixPipe_ConcurrentReadWrite(t *testing.T) {
 	// Use two separate pipes: one for the write direction, one for the read
 	// direction. This mirrors the real pipe topology where read and write are