fix: stream platform builder progress

platform/service/distribution_build_execution.go

@@ -88,7 +88,16 @@ func (svc *CoreService) executeDistributionBuild(job domain.Job) error {
 	if isTerminalJobState(job.State) {
 		return nil
 	}
-	payload, buildErr := svc.configuredDistributionBuilder().Build(input)
+	builder := svc.configuredDistributionBuilder()
+	var payload []byte
+	var buildErr error
+	if progressBuilder, ok := builder.(distributionBuilderWithProgress); ok {
+		payload, buildErr = progressBuilder.BuildWithProgress(input, func(progress DistributionBuildProgress) {
+			_ = svc.updateDistributionBuildProgress(job, progress)
+		})
+	} else {
+		payload, buildErr = builder.Build(input)
+	}
 	if buildErr != nil {
 		return svc.failDistributionBuildJob(job, builderJobFailureMessage(buildErr))
 	}
@@ -432,7 +441,7 @@ func (svc *CoreService) markDistributionBuildRunning(job *domain.Job) error {
 	}
 	current.State = domain.JobStateRunning
 	current.Attempt = maxInt(current.Attempt, 1)
-	current.Progress = domain.JobProgress{Percent: 5, Phase: current.Progress.Phase, Message: "platform builder started"}
+	current.Progress = domain.JobProgress{Percent: 5, Phase: current.Progress.Phase, Message: "env_check: platform builder started"}
 	current.UpdatedAt = stamp
 	if err := svc.updateScheduledJob(current); err != nil {
 		return err
@@ -444,6 +453,29 @@ func (svc *CoreService) markDistributionBuildRunning(job *domain.Job) error {
 	return nil
 }
 
+func (svc *CoreService) updateDistributionBuildProgress(job domain.Job, progress DistributionBuildProgress) error {
+	stamp := svc.now()
+	svc.jobMu.Lock()
+	defer svc.jobMu.Unlock()
+
+	current, err := svc.store.Jobs().Get(job.ID)
+	if err != nil {
+		return err
+	}
+	if isTerminalJobState(current.State) {
+		return nil
+	}
+	if progress.Percent < current.Progress.Percent {
+		progress.Percent = current.Progress.Percent
+	}
+	current.Progress = domain.JobProgress{Percent: progress.Percent, Phase: current.Progress.Phase, Message: strings.TrimSpace(progress.Message)}
+	current.UpdatedAt = stamp
+	if err := svc.updateScheduledJob(current); err != nil {
+		return err
+	}
+	return svc.projectDistributionBuildProgress(current, stamp)
+}
+
 func (svc *CoreService) succeedDistributionBuildJob(job domain.Job, artifactID string) error {
 	stamp := svc.now()
 	svc.jobMu.Lock()

platform/service/distribution_build_execution_test.go

@@ -19,6 +19,12 @@ type captureDistributionBuilder struct {
 	payload []byte
 }
 
+type progressDistributionBuilder struct {
+	started chan struct{}
+	release <-chan struct{}
+	payload []byte
+}
+
 func (builder captureDistributionBuilder) Readiness() (bool, string) {
 	return true, ""
 }
@@ -31,6 +37,27 @@ func (builder captureDistributionBuilder) Build(input domain.DistributionBuildIn
 	return domain.CopyBytes(builder.payload), nil
 }
 
+func (builder progressDistributionBuilder) Readiness() (bool, string) {
+	return true, ""
+}
+
+func (builder progressDistributionBuilder) Build(input domain.DistributionBuildInput) ([]byte, error) {
+	return builder.BuildWithProgress(input, nil)
+}
+
+func (builder progressDistributionBuilder) BuildWithProgress(_ domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error) {
+	if progress != nil {
+		progress(DistributionBuildProgress{Percent: 72, Message: "build_compile: compiling run target executable"})
+	}
+	if builder.started != nil {
+		builder.started <- struct{}{}
+	}
+	if builder.release != nil {
+		<-builder.release
+	}
+	return domain.CopyBytes(builder.payload), nil
+}
+
 func TestCoreServiceKeepsPlatformBuildKeyOffMachineJobChannel(t *testing.T) {
 	svc, session, instance := newDistributionTestFixture(t)
 	plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
@@ -319,6 +346,39 @@ func TestCoreServiceDoesNotDuplicateInFlightPlatformBuild(t *testing.T) {
 	completeDistributionBuild(t, svc, first, nil)
 }
 
+func TestCoreServiceProjectsPlatformBuilderProgressBeforeCompletion(t *testing.T) {
+	svc, session, instance := newDistributionTestFixture(t)
+	started := make(chan struct{}, 1)
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+	svc.ConfigureDistributionBuilder(progressDistributionBuilder{started: started, release: release, payload: []byte("progress-platform-build")})
+
+	distribution, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{
+		ServerInstanceID: instance.ID,
+		TargetOS:         "linux",
+		TargetArch:       "amd64",
+		IdempotencyKey:   "platform-builder-progress",
+	})
+	if err != nil {
+		t.Fatalf("generate platform distribution: %v", err)
+	}
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("platform builder did not report progress")
+	}
+	job, err := svc.GetJob(distribution.BuildJobID)
+	if err != nil {
+		t.Fatalf("get progress job: %v", err)
+	}
+	if job.State != domain.JobStateRunning || job.Progress.Percent != 72 || job.Progress.Message != "build_compile: compiling run target executable" {
+		t.Fatalf("expected in-flight platform builder progress, got %+v", job)
+	}
+	releaseOnce.Do(func() { close(release) })
+	completeDistributionBuild(t, svc, distribution, nil)
+}
+
 func TestCoreServiceDiscardsBuildCompletedAfterKeyReset(t *testing.T) {
 	svc, session, instance := newDistributionTestFixture(t)
 	inputs := make(chan domain.DistributionBuildInput, 1)

platform/service/distribution_builder.go

@@ -3,6 +3,7 @@ package service
 import (
 	"archive/tar"
 	"archive/zip"
+	"bufio"
 	"bytes"
 	"compress/gzip"
 	"context"
@@ -11,10 +12,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"browser.local/platform/domain"
@@ -33,6 +37,15 @@ type DistributionBuilder interface {
 	Build(input domain.DistributionBuildInput) ([]byte, error)
 }
 
+type DistributionBuildProgress struct {
+	Percent int
+	Message string
+}
+
+type distributionBuilderWithProgress interface {
+	BuildWithProgress(input domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error)
+}
+
 // DockerDistributionBuilderConfig configures a container-per-build builder.
 type DockerDistributionBuilderConfig struct {
 	DockerBinary  string
@@ -42,6 +55,7 @@ type DockerDistributionBuilderConfig struct {
 	Timeout       time.Duration
 	PlatformURL   string
 	CommandRunner func(ctx context.Context, name string, args ...string) ([]byte, error)
+	CommandStream func(ctx context.Context, name string, args []string, onLine func(string)) ([]byte, error)
 }
 
 // DockerDistributionBuilder runs each build in a container from a pinned image,
@@ -52,6 +66,7 @@ type DockerDistributionBuilder struct {
 }
 
 func NewDockerDistributionBuilder(config DockerDistributionBuilderConfig) *DockerDistributionBuilder {
+	customRunner := config.CommandRunner != nil
 	if strings.TrimSpace(config.DockerBinary) == "" {
 		config.DockerBinary = "docker"
 	}
@@ -61,6 +76,9 @@ func NewDockerDistributionBuilder(config DockerDistributionBuilderConfig) *Docke
 	if config.CommandRunner == nil {
 		config.CommandRunner = runCommandCombined
 	}
+	if config.CommandStream == nil && !customRunner {
+		config.CommandStream = runCommandStreamCombined
+	}
 	return &DockerDistributionBuilder{config: config}
 }
 
@@ -68,6 +86,46 @@ func runCommandCombined(ctx context.Context, name string, args ...string) ([]byt
 	return exec.CommandContext(ctx, name, args...).CombinedOutput()
 }
 
+func runCommandStreamCombined(ctx context.Context, name string, args []string, onLine func(string)) ([]byte, error) {
+	cmd := exec.CommandContext(ctx, name, args...)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, err
+	}
+	if err := cmd.Start(); err != nil {
+		return nil, err
+	}
+
+	var mu sync.Mutex
+	var output bytes.Buffer
+	var wg sync.WaitGroup
+	capture := func(reader io.Reader) {
+		defer wg.Done()
+		scanner := bufio.NewScanner(reader)
+		scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+		for scanner.Scan() {
+			line := scanner.Text()
+			mu.Lock()
+			output.WriteString(line)
+			output.WriteByte('\n')
+			mu.Unlock()
+			if onLine != nil {
+				onLine(line)
+			}
+		}
+	}
+	wg.Add(2)
+	go capture(stdout)
+	go capture(stderr)
+	err = cmd.Wait()
+	wg.Wait()
+	return output.Bytes(), err
+}
+
 func (builder *DockerDistributionBuilder) Readiness() (bool, string) {
 	if strings.TrimSpace(builder.config.Image) == "" {
 		return false, "platform builder image is not configured"
@@ -124,9 +182,14 @@ func pinnedBuilderImage(image string) bool {
 }
 
 func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildInput) ([]byte, error) {
+	return builder.BuildWithProgress(input, nil)
+}
+
+func (builder *DockerDistributionBuilder) BuildWithProgress(input domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error) {
 	if ready, reason := builder.Readiness(); !ready {
 		return nil, validationError(reason)
 	}
+	reportBuilderProgress(progress, 8, "env_check: platform builder readiness verified")
 	sourceDir, err := filepath.Abs(strings.TrimSpace(builder.config.SourceDir))
 	if err != nil {
 		return nil, validationError("platform builder run source directory is invalid")
@@ -150,6 +213,7 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn
 		}
 	}
 	defer func() { _ = os.RemoveAll(jobDir) }()
+	reportBuilderProgress(progress, 14, "env_check: platform builder workspace prepared")
 
 	// The auth key reaches the container through a per-job input file, never
 	// through a job-channel response to a machine-side endpoint or a container
@@ -193,12 +257,15 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn
 	ctx, cancel := context.WithTimeout(context.Background(), builder.config.Timeout)
 	defer cancel()
 	args := builder.containerArgs(input, sourceDir, inputDir, buildDir, outputDir, outputName)
-	if output, err := builder.config.CommandRunner(ctx, builder.config.DockerBinary, args...); err != nil {
+	reportBuilderProgress(progress, 18, "git_sync: platform builder container starting")
+	output, err := builder.runBuildCommand(ctx, args, progress)
+	if err != nil {
 		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
 			return nil, validationError("platform builder timed out while building the distribution")
 		}
 		return nil, validationError("platform builder failed: " + safeBuilderFailure(output, input.AuthKey, builder.config.SourceDir, sourceDir, jobDir))
 	}
+	reportBuilderProgress(progress, 92, "package_finalize: reading platform builder output")
 	binary, err := os.ReadFile(filepath.Join(outputDir, outputName))
 	if err != nil {
 		return nil, validationError("platform builder did not produce a distribution executable")
@@ -216,14 +283,74 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn
 	return packageClientManagerDistribution(input.PackageFormat, outputName, binary, configPayload)
 }
 
+func (builder *DockerDistributionBuilder) runBuildCommand(ctx context.Context, args []string, progress func(DistributionBuildProgress)) ([]byte, error) {
+	if builder.config.CommandStream == nil {
+		return builder.config.CommandRunner(ctx, builder.config.DockerBinary, args...)
+	}
+	return builder.config.CommandStream(ctx, builder.config.DockerBinary, args, func(line string) {
+		parsed, ok := parseBuilderProgressLine(line)
+		if ok {
+			reportBuilderProgress(progress, parsed.Percent, parsed.Message)
+		}
+	})
+}
+
+const builderProgressMarker = "__platform_builder_progress__|"
+
+func parseBuilderProgressLine(line string) (DistributionBuildProgress, bool) {
+	line = strings.TrimSpace(line)
+	if !strings.HasPrefix(line, builderProgressMarker) {
+		return DistributionBuildProgress{}, false
+	}
+	rest := strings.TrimPrefix(line, builderProgressMarker)
+	parts := strings.SplitN(rest, "|", 2)
+	if len(parts) != 2 {
+		return DistributionBuildProgress{}, false
+	}
+	percent, err := strconv.Atoi(strings.TrimSpace(parts[0]))
+	if err != nil {
+		return DistributionBuildProgress{}, false
+	}
+	message := strings.TrimSpace(parts[1])
+	if message == "" {
+		return DistributionBuildProgress{}, false
+	}
+	if percent < 1 {
+		percent = 1
+	}
+	if percent > 99 {
+		percent = 99
+	}
+	return DistributionBuildProgress{Percent: percent, Message: message}, true
+}
+
+func reportBuilderProgress(progress func(DistributionBuildProgress), percent int, message string) {
+	if progress == nil {
+		return
+	}
+	if percent < 1 {
+		percent = 1
+	}
+	if percent > 99 {
+		percent = 99
+	}
+	progress(DistributionBuildProgress{Percent: percent, Message: message})
+}
+
 const distributionBuildScript = `#!/bin/sh
 set -eu
 
+progress() {
+  printf '__platform_builder_progress__|%s|%s\n' "$1" "$2"
+}
+
 auth_key="$(cat /workspace/input/auth-key)"
 if [ "$COMPONENT_KIND" = "run" ]; then
+  progress 24 'git_sync: preparing run source snapshot'
   rm -rf /workspace/build/run-source
   mkdir -p /workspace/build/run-source
   cp -R /workspace/source/. /workspace/build/run-source/
+  progress 34 'env_check: injecting run build metadata'
   seed_b64="$(base64 /workspace/input/workspace-seed.json | tr -d '\n')"
   cat > /workspace/build/run-source/config/workspace_seed_generated.go <<EOF
 package config
@@ -243,8 +370,11 @@ EOF
   ldflags="$ldflags -X browser.local/run/config.BuildComponentKey=$PROFILE_KEY"
   ldflags="$ldflags -X browser.local/run/config.BuildKeyGeneration=$KEY_GENERATION"
   ldflags="$ldflags -X browser.local/run/config.BuildVersion=$TARGET_RELEASE"
+  progress 48 'build_compile: downloading Go modules'
   go mod download
+  progress 72 'build_compile: compiling run target executable'
   go build -trimpath -ldflags "$ldflags" -o "/workspace/output/$OUTPUT_FILENAME" ./cmd/run
+  progress 88 'package_finalize: run executable written'
   exit 0
 fi
 
@@ -257,10 +387,12 @@ case "$REPOSITORY_URL" in
   *) printf 'client-manager repository must use https\n' >&2; exit 2 ;;
 esac
 cd /workspace/build
+progress 24 'git_sync: fetching client-manager source'
 git init --quiet
 git remote add origin "$REPOSITORY_URL"
 git fetch --quiet --depth 1 origin "$SOURCE_REVISION"
 git checkout --quiet --detach FETCH_HEAD
+progress 36 'env_check: writing client-manager configuration'
 {
   printf 'server_url: "%s"\n' "$PLATFORM_URL"
   printf 'server_instance_id: "%s"\n' "$SERVER_INSTANCE_ID"
@@ -270,9 +402,12 @@ git checkout --quiet --detach FETCH_HEAD
   printf 'scum_client_machine_label: "managed-client"\n'
   printf 'ftp_provider: 3\n'
 } > config.yaml
+progress 52 'deps_download: downloading Go modules'
 go mod download
+progress 76 'build_compile: compiling client-manager executable'
 go build -trimpath -ldflags '-s -w' -o "/workspace/output/$OUTPUT_FILENAME" .
 cp config.yaml /workspace/output/config.yaml
+progress 88 'package_finalize: client-manager package inputs written'
 `
 
 func (builder *DockerDistributionBuilder) containerArgs(input domain.DistributionBuildInput, sourceDir string, inputDir string, buildDir string, outputDir string, outputName string) []string {
@@ -391,7 +526,7 @@ func safeBuilderFailure(output []byte, sensitiveValues ...string) string {
 	kept := make([]string, 0, len(lines))
 	for index := len(lines) - 1; index >= 0 && len(kept) < 3; index-- {
 		line := strings.TrimSpace(lines[index])
-		if line == "" || strings.Contains(line, "/workspace/input") || strings.Contains(line, "auth-key") {
+		if line == "" || strings.HasPrefix(line, builderProgressMarker) || strings.Contains(line, "/workspace/input") || strings.Contains(line, "auth-key") {
 			continue
 		}
 		line = redactBuilderHostPaths(line)

platform/service/distribution_builder_test.go

@@ -185,6 +185,58 @@ func TestDockerDistributionBuilderKeepsSecretInIsolatedInput(t *testing.T) {
 	}
 }
 
+func TestDockerDistributionBuilderStreamsProgressMarkers(t *testing.T) {
+	sourceDir := createBuilderSource(t)
+	workspaceDir := t.TempDir()
+	var progress []DistributionBuildProgress
+	builder := NewDockerDistributionBuilder(DockerDistributionBuilderConfig{
+		DockerBinary: "docker-test",
+		Image:        "browser-platform-distribution-builder:1.0.0",
+		SourceDir:    sourceDir,
+		WorkspaceDir: workspaceDir,
+		CommandRunner: func(_ context.Context, _ string, args ...string) ([]byte, error) {
+			if len(args) > 0 && (args[0] == "version" || args[0] == "image") {
+				return []byte("27.0.0"), nil
+			}
+			return nil, errors.New("unexpected non-stream command")
+		},
+		CommandStream: func(_ context.Context, _ string, args []string, onLine func(string)) ([]byte, error) {
+			onLine("__platform_builder_progress__|72|build_compile: compiling run target executable")
+			outputDir := builderMountHostPath(t, args, "/workspace/output")
+			if err := os.WriteFile(filepath.Join(outputDir, "run"), []byte("compiled-run"), 0o700); err != nil {
+				return nil, err
+			}
+			return []byte("builder finished"), nil
+		},
+	})
+	payload, err := builder.BuildWithProgress(domain.DistributionBuildInput{
+		JobID:          "job-progress",
+		ComponentKind:  domain.DistributionComponentRun,
+		PluginID:       "game.scum",
+		TargetOS:       "linux",
+		TargetArch:     "amd64",
+		OutputFilename: "run",
+		AuthKey:        "component-key",
+	}, func(item DistributionBuildProgress) {
+		progress = append(progress, item)
+	})
+	if err != nil {
+		t.Fatalf("build with progress: %v", err)
+	}
+	if string(payload) != "compiled-run" {
+		t.Fatalf("unexpected built payload %q", payload)
+	}
+	found := false
+	for _, item := range progress {
+		if item.Percent == 72 && item.Message == "build_compile: compiling run target executable" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("streamed progress marker was not reported: %+v", progress)
+	}
+}
+
 func TestDockerDistributionBuilderRedactsFailureAndTimeout(t *testing.T) {
 	sourceDir := createBuilderSource(t)
 	workspaceDir := t.TempDir()