fix(e2e): demultiplex act container stream for reliable CI output parsing

e2e/harness/act.go

@@ -3,11 +3,13 @@ package harness
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"io"
 	"strings"
 
+	"github.com/moby/moby/api/pkg/stdcopy"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/api/types/mount"
 	"github.com/testcontainers/testcontainers-go"
@@ -269,23 +271,22 @@ func (a *ActRunner) RunWorkflow(ctx context.Context, opts RunOpts) (*ExtendedWor
 		return nil, fmt.Errorf("failed to run act: %w", err)
 	}
 
-	// Read logs from the reader
-	var logs bytes.Buffer
-	if reader != nil {
-		_, err = io.Copy(&logs, reader)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read act output: %w", err)
-		}
+	// Read and demultiplex the container's hijacked-attach stream. Without
+	// demuxing, a TTY-less exec (CI Linux) prefixes each chunk with Docker's
+	// 8-byte stream header, which corrupts the JSON the parser consumes.
+	logs, err := readDemuxedStream(reader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read act output: %w", err)
 	}
 
 	// Parse JSON output
-	result, err := ParseActOutput(logs.String())
+	result, err := ParseActOutput(logs)
 	if err != nil {
 		// Fall back to basic result if parsing fails
 		result = &ExtendedWorkflowResult{
 			Conclusion: "success",
 			Jobs:       make(map[string]*JobResultExtended),
-			Logs:       logs.String(),
+			Logs:       logs,
 		}
 	}
 
@@ -352,21 +353,21 @@ func (a *ActRunner) RunWorkflowFromRepo(ctx context.Context, opts RunOpts) (*Ext
 		return nil, fmt.Errorf("failed to run act: %w", err)
 	}
 
-	var logs bytes.Buffer
-	if reader != nil {
-		_, _ = io.Copy(&logs, reader)
+	// Read and demultiplex the container's hijacked-attach stream so the JSON
+	// the parser consumes is free of Docker's per-chunk stream headers. On a
+	// TTY-less exec (CI Linux) those headers would otherwise corrupt the
+	// conclusion, job-failure, infra-saturation, and crash detection.
+	cleanedLogs, err := readDemuxedStream(reader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read act output: %w", err)
 	}
 
-	// Clean the output: strip Docker's multiplexed stream headers
-	// Docker exec output contains 8-byte headers for each chunk
-	cleanedLogs := stripDockerStreamHeaders(logs.String())
-
 	result, err := ParseActOutput(cleanedLogs)
 	if err != nil {
 		result = &ExtendedWorkflowResult{
 			Conclusion: "success",
 			Jobs:       make(map[string]*JobResultExtended),
-			Logs:       logs.String(),
+			Logs:       cleanedLogs,
 		}
 		// The parse fallback bypasses ParseActOutput's crash detection, so run
 		// it directly here against the raw logs: a crash dump is the most likely
@@ -491,41 +492,68 @@ func (a *ActRunner) buildActArgs(opts RunOpts, eventPath string) string {
 	return args
 }
 
-// stripDockerStreamHeaders removes Docker's multiplexed stream headers from output.
-// Docker exec output contains 8-byte headers for each chunk:
-// - 1 byte: stream type (0=stdin, 1=stdout, 2=stderr)
-// - 3 bytes: padding
-// - 4 bytes: payload size (big endian)
-func stripDockerStreamHeaders(input string) string {
-	var result bytes.Buffer
-	data := []byte(input)
-
-	for i := 0; i < len(data); {
-		// Look for the start of a JSON object
-		if data[i] == '{' {
-			// Find the end of this JSON line (newline or end of data)
-			end := i
-			braceCount := 0
-			for end < len(data) {
-				if data[end] == '{' {
-					braceCount++
-				} else if data[end] == '}' {
-					braceCount--
-					if braceCount == 0 {
-						end++
-						break
-					}
-				}
-				end++
-			}
-			result.Write(data[i:end])
-			result.WriteByte('\n')
-			i = end
-		} else {
-			// Skip non-JSON bytes (headers, control chars, etc.)
-			i++
-		}
+// dockerStreamHeaderLen is the size, in bytes, of the header Docker prepends to
+// each frame of a multiplexed (non-TTY) hijacked-attach stream:
+//   - 1 byte: stream type (0=stdin, 1=stdout, 2=stderr)
+//   - 3 bytes: padding (always zero)
+//   - 4 bytes: payload size (big-endian uint32)
+const dockerStreamHeaderLen = 8
+
+// readDemuxedStream drains a container exec's hijacked-attach reader and returns
+// clean text with Docker's per-frame stream headers removed.
+//
+// A container created without a TTY (every act run here, and notably CI Linux
+// where Docker allocates no TTY) returns a MULTIPLEXED stream: each chunk is
+// prefixed with an 8-byte header (stream type + big-endian length). A raw
+// io.Copy leaves those headers interspersed in the captured logs, which lands a
+// `\x02\x00...` prefix on JSON lines and corrupts conclusion, job-failure,
+// infra-saturation, and crash parsing. Docker Desktop happens to demux for us,
+// which is why the corruption never reproduces locally and only bites in CI.
+//
+// We demultiplex with stdcopy.StdCopy (the same primitive testcontainers'
+// exec.Multiplexed uses), merging stdout and stderr into a single buffer; act
+// writes its --json log to stderr, so both streams must be captured. If the
+// stream is NOT framed (a raw TTY attach), it is returned unchanged.
+func readDemuxedStream(reader io.Reader) (string, error) {
+	if reader == nil {
+		return "", nil
 	}
 
-	return result.String()
+	raw, err := io.ReadAll(reader)
+	if err != nil {
+		return "", fmt.Errorf("reading container stream: %w", err)
+	}
+
+	if !looksMultiplexed(raw) {
+		// Raw (TTY) attach: no framing to strip.
+		return string(raw), nil
+	}
+
+	var out bytes.Buffer
+	// destOut and destErr both target out so stdout and stderr merge into one
+	// clean log buffer (act emits its JSON event stream on stderr).
+	if _, err := stdcopy.StdCopy(&out, &out, bytes.NewReader(raw)); err != nil {
+		return "", fmt.Errorf("demultiplexing container stream: %w", err)
+	}
+	return out.String(), nil
+}
+
+// looksMultiplexed reports whether data begins with a well-formed Docker stream
+// frame header: a known stream type, zeroed padding, and a length that does not
+// overrun the buffer. This distinguishes a multiplexed (non-TTY) attach from a
+// raw (TTY) attach so demuxing is only applied when there is framing to strip.
+func looksMultiplexed(data []byte) bool {
+	if len(data) < dockerStreamHeaderLen {
+		return false
+	}
+	switch data[0] {
+	case byte(stdcopy.Stdin), byte(stdcopy.Stdout), byte(stdcopy.Stderr):
+	default:
+		return false
+	}
+	if data[1] != 0 || data[2] != 0 || data[3] != 0 {
+		return false
+	}
+	frameSize := binary.BigEndian.Uint32(data[4:dockerStreamHeaderLen])
+	return uint64(dockerStreamHeaderLen)+uint64(frameSize) <= uint64(len(data))
 }

e2e/harness/act_test.go

@@ -1,14 +1,120 @@
 package harness
 
 import (
+	"bytes"
 	"context"
+	"encoding/binary"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/moby/moby/api/pkg/stdcopy"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// frameDockerStream builds a single Docker hijacked-attach frame: an 8-byte
+// header (1-byte stream type, 3 zero padding bytes, big-endian uint32 length)
+// followed by the payload. This is the exact framing a TTY-less container exec
+// produces on CI Linux, and the framing a raw io.Copy would leave interspersed
+// in the captured logs.
+func frameDockerStream(t *testing.T, stream stdcopy.StdType, payload string) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	header := make([]byte, dockerStreamHeaderLen)
+	header[0] = byte(stream)
+	binary.BigEndian.PutUint32(header[4:], uint32(len(payload)))
+	buf.Write(header)
+	buf.WriteString(payload)
+	return buf.Bytes()
+}
+
+// TestReadDemuxedStream_MultiplexedFramesAreCleaned is the load-bearing proof
+// for the CI-only act-output corruption: it feeds a hand-constructed MULTIPLEXED
+// stream (stdout + stderr frames, each carrying Docker's 8-byte header) through
+// readDemuxedStream and asserts the result is clean text with the streams merged
+// and NO header bytes left behind. Before the demux fix the capture used a raw
+// io.Copy, so these header bytes would remain interspersed and corrupt the JSON
+// the act parser consumes.
+func TestReadDemuxedStream_MultiplexedFramesAreCleaned(t *testing.T) {
+	t.Parallel()
+
+	stdoutPayload := "hello world\n"
+	stderrPayload := `{"level":"info","msg":"Using docker host..."}` + "\n"
+
+	var raw bytes.Buffer
+	raw.Write(frameDockerStream(t, stdcopy.Stdout, stdoutPayload))
+	raw.Write(frameDockerStream(t, stdcopy.Stderr, stderrPayload))
+
+	got, err := readDemuxedStream(bytes.NewReader(raw.Bytes()))
+	require.NoError(t, err)
+
+	// stdout and stderr are merged into a single clean log buffer.
+	assert.Contains(t, got, "hello world")
+	assert.Contains(t, got, `{"level":"info","msg":"Using docker host..."}`)
+
+	// No Docker stream-header bytes survive. \x02\x00 is the stderr-frame
+	// header signature seen leading the corrupted CI logs; \x01\x00 is stdout.
+	assert.NotContains(t, got, "\x02\x00")
+	assert.NotContains(t, got, "\x01\x00")
+	assert.False(t, strings.ContainsRune(got, '\x00'), "no NUL padding bytes should remain")
+}
+
+// TestReadDemuxedStream_RawStreamPassesThrough verifies a raw (TTY) attach with
+// no Docker framing is returned unchanged, so the demux path never mangles an
+// already-clean stream (the Docker Desktop / TTY case).
+func TestReadDemuxedStream_RawStreamPassesThrough(t *testing.T) {
+	t.Parallel()
+
+	raw := `{"level":"info","msg":"already clean"}` + "\nplain line\n"
+	got, err := readDemuxedStream(strings.NewReader(raw))
+	require.NoError(t, err)
+	assert.Equal(t, raw, got)
+}
+
+// TestReadDemuxedStream_NilReader confirms a nil reader yields empty output
+// rather than panicking (the Exec call can return a nil reader).
+func TestReadDemuxedStream_NilReader(t *testing.T) {
+	t.Parallel()
+
+	got, err := readDemuxedStream(nil)
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
+// TestLooksMultiplexed table-checks the frame-header detector that gates demux:
+// a valid stdout/stderr frame is multiplexed; raw JSON, short input, bad
+// padding, an unknown stream type, and an overrunning length are not.
+func TestLooksMultiplexed(t *testing.T) {
+	t.Parallel()
+
+	validFrame := frameDockerStream(t, stdcopy.Stdout, "hello world\n")
+	badPadding := append([]byte{1, 1, 0, 0, 0, 0, 0, 3}, []byte("abc")...)
+	unknownType := append([]byte{9, 0, 0, 0, 0, 0, 0, 3}, []byte("abc")...)
+	overrun := []byte{1, 0, 0, 0, 0, 0, 0, 255}
+
+	tests := []struct {
+		name string
+		data []byte
+		want bool
+	}{
+		{name: "valid stdout frame", data: validFrame, want: true},
+		{name: "raw json is not framed", data: []byte(`{"level":"info"}`), want: false},
+		{name: "shorter than header", data: []byte{1, 0, 0}, want: false},
+		{name: "non-zero padding", data: badPadding, want: false},
+		{name: "unknown stream type", data: unknownType, want: false},
+		{name: "length overruns buffer", data: overrun, want: false},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tt.want, looksMultiplexed(tt.data))
+		})
+	}
+}
+
 func TestActRunner_Start(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")