run/runtime/execution_test.go

package runtime

import (
	"context"
	"encoding/json"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"browser.local/run/protocol"
	"browser.local/run/spool"
)

func TestRunHelperProcess(t *testing.T) {
	if os.Getenv("RUN_TEST_HELPER") != "1" {
		return
	}
	if os.Getenv("RUN_LOG_LINES") == "1" {
		marker := os.Getenv("RUN_LOG_MARKER")
		_, _ = os.Stdout.WriteString("managed stdout ready " + marker + "\n")
		_, _ = os.Stderr.WriteString("managed stderr ready " + marker + "\n")
		return
	}
	if trigger := os.Getenv("RUN_LOG_TRIGGER_FILE"); trigger != "" {
		for {
			if _, err := os.Stat(trigger); err == nil {
				break
			}
			time.Sleep(25 * time.Millisecond)
		}
		marker := os.Getenv("RUN_LOG_MARKER")
		_, _ = os.Stdout.WriteString("managed stdout triggered " + marker + "\n")
		_, _ = os.Stderr.WriteString("managed stderr triggered " + marker + "\n")
		if os.Getenv("RUN_HOLD_AFTER_LOG") != "1" {
			return
		}
		for {
			time.Sleep(100 * time.Millisecond)
		}
	}
	if os.Getenv("RUN_LOG_TICKS") == "1" {
		marker := os.Getenv("RUN_LOG_MARKER")
		for {
			_, _ = os.Stdout.WriteString("managed stdout tick " + marker + "\n")
			_, _ = os.Stderr.WriteString("managed stderr tick " + marker + "\n")
			time.Sleep(100 * time.Millisecond)
		}
	}
	if os.Getenv("RUN_EXIT_NOW") == "1" {
		return
	}
	for {
		time.Sleep(100 * time.Millisecond)
	}
}

func TestTypedProcessStartStreamsManagedOutput(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	setupProcessWorkspace(t, root, assignment, true)
	scope := processScope(root, assignment)
	writeJSONFixture(t, filepath.Join(scope, "actions", "start.json"), map[string]any{
		"version":       1,
		"action":        "start",
		"mode":          "supervised",
		"executableKey": "bin/game-server",
		"arguments":     []string{"-test.run=TestRunHelperProcess"},
		"environment":   map[string]string{"RUN_TEST_HELPER": "1", "RUN_LOG_LINES": "1"},
	})
	logSink := &recordingLogSink{}

	result := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(logSink)).Execute(assignment)

	if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.ProcessState != "running" {
		t.Fatalf("expected managed process start, got %+v", result)
	}
	deadline := time.Now().Add(3 * time.Second)
	lines := logSink.snapshot()
	for len(lines) < 2 && time.Now().Before(deadline) {
		time.Sleep(25 * time.Millisecond)
		lines = logSink.snapshot()
	}
	joined := strings.Join(lines, "\n")
	if !strings.Contains(joined, "stdout:managed stdout ready") || !strings.Contains(joined, "stderr:managed stderr ready") {
		t.Fatalf("expected managed stdout/stderr to stream, got %+v", lines)
	}
	time.Sleep(250 * time.Millisecond)
}

func TestTypedProcessOutputCaptureResumesAfterRunRestart(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	setupProcessWorkspace(t, root, assignment, false)
	scope := processScope(root, assignment)
	writeJSONFixture(t, filepath.Join(scope, "actions", "start.json"), map[string]any{
		"version":       1,
		"action":        "start",
		"mode":          "supervised",
		"executableKey": "bin/game-server",
		"arguments":     []string{"-test.run=TestRunHelperProcess"},
		"environment":   map[string]string{"RUN_TEST_HELPER": "1", "RUN_LOG_TICKS": "1"},
	})

	started := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root)).Execute(assignment)
	if started.State != lifecycleResultStateSucceeded || started.ExecutionResult.ProcessState != "running" {
		t.Fatalf("expected managed process start, got %+v", started)
	}
	time.Sleep(250 * time.Millisecond)
	logSink := &recordingLogSink{}
	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(logSink))
	restarted.ResumeManagedProcessLogs(context.Background())

	deadline := time.Now().Add(3 * time.Second)
	lines := logSink.snapshot()
	for len(lines) < 2 && time.Now().Before(deadline) {
		time.Sleep(25 * time.Millisecond)
		lines = logSink.snapshot()
	}
	joined := strings.Join(lines, "\n")
	if !strings.Contains(joined, "stdout:managed stdout tick") || !strings.Contains(joined, "stderr:managed stderr tick") {
		t.Fatalf("expected restarted run to resume captured output, got %+v", lines)
	}
	stop := executionAssignment(protocol.RunCapabilityProcessStop)
	stop.TargetKey = "actions/stop.json"
	if stopped := restarted.Execute(stop); stopped.State != lifecycleResultStateSucceeded {
		t.Fatalf("stop resumed process: %+v", stopped)
	}
	waitForManagedTailers(t, restarted.managed.(*OSManagedProcessSupervisor))
}

func TestManagedProcessOutputAfterCanceledRunContextIsSpooledOnRestart(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	assignment.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{
		{Key: "console-stdout", Kind: "process.stdout", StreamKey: "game.console.stdout", CursorKind: "sequence"},
		{Key: "console-stderr", Kind: "process.stderr", StreamKey: "game.console.stderr", CursorKind: "sequence"},
	}
	setupProcessWorkspace(t, root, assignment, false)
	scope := processScope(root, assignment)
	trigger := filepath.Join(root, "emit-after-cancel")
	writeJSONFixture(t, filepath.Join(scope, "actions", "start.json"), map[string]any{
		"version":       1,
		"action":        "start",
		"mode":          "supervised",
		"executableKey": "bin/game-server",
		"arguments":     []string{"-test.run=TestRunHelperProcess"},
		"environment": map[string]string{
			"RUN_TEST_HELPER":      "1",
			"RUN_LOG_TRIGGER_FILE": trigger,
			"RUN_LOG_MARKER":       "after-cancel",
		},
	})
	logSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("new log spool: %v", err)
	}
	oldCtx, cancelOldRun := context.WithCancel(context.Background())
	oldSink := &contextRejectingLogSink{delegate: &SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "old-session", Spool: logSpool}}
	oldExecutor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(oldSink))
	started := oldExecutor.ExecuteContext(oldCtx, assignment)
	if started.State != lifecycleResultStateSucceeded {
		t.Fatalf("start managed process: %+v", started)
	}
	oldManaged := oldExecutor.managed.(*OSManagedProcessSupervisor)
	identity := oldManaged.Status(ProcessIdentity{Scope: scope})
	if identity.LogSessionID == "" {
		t.Fatalf("expected managed log session: %+v", identity)
	}
	t.Cleanup(func() {
		_, _ = oldManaged.Stop(context.Background(), identity)
	})

	cancelOldRun()
	if err := os.WriteFile(trigger, []byte("emit"), 0o600); err != nil {
		t.Fatalf("trigger post-cancel output: %v", err)
	}
	waitForFileText(t, filepath.Join(root, "state", "process-output", identity.StdoutLogRef), "after-cancel")
	beforeRestart := waitForManagedState(t, oldManaged, scope, "exited")
	if beforeRestart.StdoutOffset != identity.StdoutOffset || beforeRestart.StderrOffset != identity.StderrOffset {
		t.Fatalf("canceled sink advanced output offsets: before=%+v after=%+v", identity, beforeRestart)
	}
	if pending, err := logSpool.Pending(); err != nil || len(pending) != 0 {
		t.Fatalf("canceled sink unexpectedly committed spool entries: pending=%+v err=%v", pending, err)
	}
	oldManaged.mu.Lock()
	oldManaged.stopTailersLocked(identity)
	oldManaged.mu.Unlock()

	restartedSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("reopen log spool: %v", err)
	}
	restarted := NewLifecycleExecutor(
		WithLifecycleWorkspaceRoot(root),
		WithProcessLogSink(&SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "new-session", Spool: restartedSpool}),
	)
	restarted.ResumeManagedProcessLogs(context.Background())
	waitForSpooledText(t, restartedSpool, "managed stdout triggered after-cancel", "managed stderr triggered after-cancel")
	restartedManaged := restarted.managed.(*OSManagedProcessSupervisor)
	resumed := waitForManagedOffsets(t, restartedManaged, scope, identity.StdoutOffset, identity.StderrOffset)
	if resumed.LogSessionID != identity.LogSessionID || resumed.StdoutOffset <= identity.StdoutOffset || resumed.StderrOffset <= identity.StderrOffset {
		t.Fatalf("restart did not retain session and commit offsets: before=%+v after=%+v", identity, resumed)
	}
	for _, batch := range mustPendingLogs(t, restartedSpool) {
		if batch.LogSessionID != identity.LogSessionID || !batch.SessionStartedAt.Equal(identity.StartedAt) {
			t.Fatalf("spooled batch lost process session metadata: %+v", batch)
		}
		if batch.FirstSeq != 1 || len(batch.Entries) == 0 || batch.Entries[0].Seq != 1 {
			t.Fatalf("fresh generation stream did not start durably at sequence 1: %+v", batch)
		}
		for index, entry := range batch.Entries {
			if entry.Seq != uint64(index+1) {
				t.Fatalf("fresh generation stream sequence is not contiguous: %+v", batch)
			}
		}
	}
	stop := executionAssignment(protocol.RunCapabilityProcessStop)
	stop.TargetKey = "actions/stop.json"
	if stopped := restarted.Execute(stop); stopped.State != lifecycleResultStateSucceeded {
		t.Fatalf("stop resumed process: %+v", stopped)
	}
	waitForManagedTailers(t, restartedManaged)
}

func TestImmediateManagedProcessRestartTailsNewGeneration(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	assignment.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{{Key: "console-stdout", Kind: "process.stdout", StreamKey: "game.console.stdout", CursorKind: "sequence"}}
	setupProcessWorkspace(t, root, assignment, false)
	scope := processScope(root, assignment)
	writeTickAction := func(marker string) {
		writeJSONFixture(t, filepath.Join(scope, "actions", "start.json"), map[string]any{
			"version":       1,
			"action":        "start",
			"mode":          "supervised",
			"executableKey": "bin/game-server",
			"arguments":     []string{"-test.run=TestRunHelperProcess"},
			"environment":   map[string]string{"RUN_TEST_HELPER": "1", "RUN_LOG_TICKS": "1", "RUN_LOG_MARKER": marker},
		})
	}
	logSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("new log spool: %v", err)
	}
	executor := NewLifecycleExecutor(
		WithLifecycleWorkspaceRoot(root),
		WithProcessLogSink(&SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "session-token", Spool: logSpool}),
	)
	writeTickAction("generation-a")
	if started := executor.Execute(assignment); started.State != lifecycleResultStateSucceeded {
		t.Fatalf("start generation A: %+v", started)
	}
	managed := executor.managed.(*OSManagedProcessSupervisor)
	first := managed.Status(ProcessIdentity{Scope: scope})
	waitForSpooledText(t, logSpool, "generation-a")
	stop := executionAssignment(protocol.RunCapabilityProcessStop)
	stop.TargetKey = "actions/stop.json"
	if stopped := executor.Execute(stop); stopped.State != lifecycleResultStateSucceeded {
		t.Fatalf("stop generation A: %+v", stopped)
	}
	writeTickAction("generation-b")
	if started := executor.Execute(assignment); started.State != lifecycleResultStateSucceeded {
		t.Fatalf("start generation B: %+v", started)
	}
	second := managed.Status(ProcessIdentity{Scope: scope})
	if second.LogSessionID == "" || second.LogSessionID == first.LogSessionID {
		t.Fatalf("expected a new process log session: first=%+v second=%+v", first, second)
	}
	waitForSpooledText(t, logSpool, "generation-b")
	time.Sleep(managedProcessOutputDrainDelay + 300*time.Millisecond)
	waitForSpooledTextCount(t, logSpool, "generation-b", 4)
	sessions := map[string]bool{}
	for _, batch := range mustPendingLogs(t, logSpool) {
		sessions[batch.LogSessionID] = true
	}
	if !sessions[first.LogSessionID] || !sessions[second.LogSessionID] {
		t.Fatalf("durable spool did not retain both process generations: sessions=%+v", sessions)
	}
	current := managed.Status(ProcessIdentity{Scope: scope})
	if err := managed.updateOutputOffset(first, "stdout", current.StdoutOffset+1_000_000); err != nil {
		t.Fatalf("update stale generation offset: %v", err)
	}
	if afterOldDrain := managed.Status(ProcessIdentity{Scope: scope}); afterOldDrain.StdoutOffset != current.StdoutOffset {
		t.Fatalf("old generation corrupted current output offset: before=%+v after=%+v", current, afterOldDrain)
	}
	if stopped := executor.Execute(stop); stopped.State != lifecycleResultStateSucceeded {
		t.Fatalf("stop generation B: %+v", stopped)
	}
	waitForManagedTailers(t, managed)
}

func TestManagedProcessRestartRetainsUndrainedRetiredGeneration(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	assignment.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "game.console.stdout"}, {Kind: "process.stderr", StreamKey: "game.console.stderr"}}
	setupProcessWorkspace(t, root, assignment, false)
	scope := processScope(root, assignment)
	writeAction := func(marker string) {
		writeJSONFixture(t, filepath.Join(scope, "actions", "start.json"), map[string]any{
			"version":       1,
			"action":        "start",
			"mode":          "supervised",
			"executableKey": "bin/game-server",
			"arguments":     []string{"-test.run=TestRunHelperProcess"},
			"environment":   map[string]string{"RUN_TEST_HELPER": "1", "RUN_LOG_LINES": "1", "RUN_LOG_MARKER": marker},
		})
	}
	oldExecutor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(alwaysRejectingLogSink{}))
	writeAction("retired-a")
	if started := oldExecutor.Execute(assignment); started.State != lifecycleResultStateSucceeded {
		t.Fatalf("start generation A: %+v", started)
	}
	oldManaged := oldExecutor.managed.(*OSManagedProcessSupervisor)
	first := waitForManagedState(t, oldManaged, scope, "exited")
	waitForFileText(t, filepath.Join(root, "state", "process-output", first.StdoutLogRef), "retired-a")

	writeAction("current-b")
	if started := oldExecutor.Execute(assignment); started.State != lifecycleResultStateSucceeded {
		t.Fatalf("start generation B: %+v", started)
	}
	second := waitForManagedState(t, oldManaged, scope, "exited")
	waitForFileText(t, filepath.Join(root, "state", "process-output", second.StdoutLogRef), "current-b")
	if first.LogSessionID == second.LogSessionID {
		t.Fatalf("process restart reused log session: first=%+v second=%+v", first, second)
	}
	oldManaged.mu.Lock()
	if len(oldManaged.retired) != 1 {
		oldManaged.mu.Unlock()
		t.Fatalf("expected one durable retired generation, got %+v", oldManaged.retired)
	}
	oldManaged.stopTailersLocked(first)
	oldManaged.stopTailersLocked(second)
	oldManaged.mu.Unlock()

	logSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("open restart spool: %v", err)
	}
	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "restart-session", Spool: logSpool}))
	restarted.ResumeManagedProcessLogs(context.Background())
	waitForSpooledText(t, logSpool, "retired-a", "current-b")
	restartedManaged := restarted.managed.(*OSManagedProcessSupervisor)
	waitForManagedTailers(t, restartedManaged)
	restartedManaged.mu.Lock()
	retiredCount := len(restartedManaged.retired)
	restartedManaged.mu.Unlock()
	if retiredCount != 0 {
		t.Fatalf("drained retired generation remained in journal: count=%d", retiredCount)
	}
}

func TestManagedProcessSourceCursorDeduplicatesCommittedLineWithStaleJournalOffset(t *testing.T) {
	root := t.TempDir()
	stateDir := filepath.Join(root, "state")
	outputDir := filepath.Join(stateDir, "process-output")
	if err := os.MkdirAll(outputDir, 0o700); err != nil {
		t.Fatalf("create process output dir: %v", err)
	}
	line := "committed before offset\n"
	identity := ProcessIdentity{Scope: filepath.Join(root, "instances", "server-execution", "local"), ServerInstanceID: "server-execution", RunEndpointID: "run-execution", JobID: "execution-job", Capability: protocol.RunCapabilityProcessStart, ProfileKey: "local", LogSessionID: "session-stale-offset", PID: 12345, StartedAt: time.Now().UTC(), State: "exited", StdoutLogRef: "stale.stdout.log", StderrLogRef: "stale.stderr.log", StdoutStreamKey: "game.console.stdout", StderrStreamKey: "game.console.stderr"}
	if err := os.WriteFile(filepath.Join(outputDir, identity.StdoutLogRef), []byte(line), 0o600); err != nil {
		t.Fatalf("write stale stdout: %v", err)
	}
	if err := os.WriteFile(filepath.Join(outputDir, identity.StderrLogRef), nil, 0o600); err != nil {
		t.Fatalf("write stale stderr: %v", err)
	}
	body, err := json.Marshal(processJournal{Version: managedProcessJournalVersion, Items: map[string]ProcessIdentity{identity.Scope: identity}, Retired: map[string]ProcessIdentity{}})
	if err != nil {
		t.Fatalf("marshal process journal: %v", err)
	}
	if err := os.WriteFile(filepath.Join(stateDir, "processes.json"), body, 0o600); err != nil {
		t.Fatalf("write process journal: %v", err)
	}
	logSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("new log spool: %v", err)
	}
	assignment := assignmentFromProcessIdentity(identity)
	sink := &SpoolLogSink{RunEndpointID: identity.RunEndpointID, SessionToken: "old-run-session", Spool: logSpool}
	if err := sink.AppendWithCursor(context.Background(), assignment, "stdout", strings.TrimSpace(line), ProcessLogCursor{StartOffset: 0, EndOffset: int64(len(line))}); err != nil {
		t.Fatalf("commit line before offset: %v", err)
	}
	streamID := logStreamIDForAssignment(assignment, identity.StdoutStreamKey)
	if err := logSpool.Ack(protocol.LogBatchIngestResponse{LogStreamID: streamID, AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
		t.Fatalf("ack committed line: %v", err)
	}
	restartedSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
	if err != nil {
		t.Fatalf("restart log spool: %v", err)
	}
	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&SpoolLogSink{RunEndpointID: identity.RunEndpointID, SessionToken: "new-run-session", Spool: restartedSpool}))
	restarted.ResumeManagedProcessLogs(context.Background())
	managed := restarted.managed.(*OSManagedProcessSupervisor)
	resumed := waitForManagedOffsets(t, managed, identity.Scope, 0, -1)
	waitForManagedTailers(t, managed)
	if resumed.StdoutOffset != int64(len(line)) {
		t.Fatalf("stale journal offset was not advanced: %+v", resumed)
	}
	if pending := mustPendingLogs(t, restartedSpool); len(pending) != 0 {
		t.Fatalf("committed source cursor was enqueued twice: %+v", pending)
	}
}

func TestManagedProcessSupervisorMigratesLiveLegacySession(t *testing.T) {
	root := t.TempDir()
	stateDir := filepath.Join(root, "state")
	outputDir := filepath.Join(stateDir, "process-output")
	if err := os.MkdirAll(outputDir, 0o700); err != nil {
		t.Fatalf("create state dirs: %v", err)
	}
	stdout, err := os.OpenFile(filepath.Join(outputDir, "legacy.stdout.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		t.Fatalf("open legacy stdout: %v", err)
	}
	stderr, err := os.OpenFile(filepath.Join(outputDir, "legacy.stderr.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		_ = stdout.Close()
		t.Fatalf("open legacy stderr: %v", err)
	}
	cmd := exec.Command(mustExecutable(t), "-test.run=TestRunHelperProcess")
	cmd.Env = append(os.Environ(), "RUN_TEST_HELPER=1", "RUN_LOG_TICKS=1", "RUN_LOG_MARKER=legacy")
	cmd.Stdout = stdout
	cmd.Stderr = stderr
	if err := cmd.Start(); err != nil {
		_ = stdout.Close()
		_ = stderr.Close()
		t.Fatalf("start legacy managed process: %v", err)
	}
	t.Cleanup(func() {
		_ = cmd.Process.Kill()
		_ = cmd.Wait()
		_ = stdout.Close()
		_ = stderr.Close()
	})
	startedAt := time.Now().UTC()
	identity := ProcessIdentity{Scope: filepath.Join(root, "instances", "legacy-server", "local"), ServerInstanceID: "legacy-server", RunEndpointID: "run-execution", JobID: "legacy-job", Capability: protocol.RunCapabilityProcessStart, ProfileKey: "local", PID: cmd.Process.Pid, StartedAt: startedAt, CommandFingerprint: "sha256:legacy", State: "running", ObservationSeq: 1, StdoutLogRef: "legacy.stdout.log", StderrLogRef: "legacy.stderr.log", UpdatedAt: startedAt}
	body, err := json.Marshal(processJournal{Version: 1, Items: map[string]ProcessIdentity{identity.Scope: identity}})
	if err != nil {
		t.Fatalf("marshal legacy journal: %v", err)
	}
	if err := os.WriteFile(filepath.Join(stateDir, "processes.json"), body, 0o600); err != nil {
		t.Fatalf("write legacy journal: %v", err)
	}
	supervisor, err := NewOSManagedProcessSupervisor(root)
	if err != nil {
		t.Fatalf("migrate legacy supervisor: %v", err)
	}
	migrated := supervisor.Status(ProcessIdentity{Scope: identity.Scope})
	if migrated.LogSessionID == "" || migrated.StartedAt.IsZero() {
		t.Fatalf("live legacy process did not receive a persisted session: %+v", migrated)
	}
	persistedBody, err := os.ReadFile(filepath.Join(stateDir, "processes.json"))
	if err != nil {
		t.Fatalf("read migrated journal: %v", err)
	}
	var persisted processJournal
	if err := json.Unmarshal(persistedBody, &persisted); err != nil {
		t.Fatalf("decode migrated journal: %v", err)
	}
	if persisted.Version != managedProcessJournalVersion || persisted.Items[identity.Scope].LogSessionID != migrated.LogSessionID {
		t.Fatalf("legacy session migration was not durable: %+v", persisted)
	}
}

func TestTypedProcessStartStopIsIdempotentAndReconciles(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	setupProcessWorkspace(t, root, assignment, false)
	executor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root))
	started := executor.Execute(assignment)
	if started.State != lifecycleResultStateSucceeded || started.ExecutionResult.ProcessState != "running" {
		t.Fatalf("expected running process, got %+v", started)
	}
	managed := executor.managed.(*OSManagedProcessSupervisor)
	first := managed.Status(ProcessIdentity{Scope: processScope(root, assignment)})
	if first.LogSessionID == "" {
		t.Fatalf("expected a persisted supervised log session, got %+v", first)
	}
	second := executor.Execute(assignment)
	current := managed.Status(ProcessIdentity{Scope: processScope(root, assignment)})
	if second.State != lifecycleResultStateSucceeded || current.PID != first.PID || current.LogSessionID != first.LogSessionID {
		t.Fatalf("expected idempotent start, first=%+v second=%+v", first, second)
	}

	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root))
	status := executionAssignment(protocol.RunCapabilityProcessStatus)
	status.TargetKey = "actions/status.json"
	statusResult := restarted.Execute(status)
	if statusResult.State != lifecycleResultStateSucceeded || statusResult.ExecutionResult.ProcessState != "running" {
		t.Fatalf("expected restart reconciliation to retain process, got %+v", statusResult)
	}
	resumed := restarted.managed.(*OSManagedProcessSupervisor).Status(ProcessIdentity{Scope: processScope(root, assignment)})
	if resumed.LogSessionID != first.LogSessionID {
		t.Fatalf("expected Run restart to retain session, before=%q after=%q", first.LogSessionID, resumed.LogSessionID)
	}
	beforeResumeSeq := resumed.ObservationSeq
	beforeResumeUpdatedAt := resumed.UpdatedAt
	restarted.ResumeManagedProcessLogs(context.Background())
	resumed = restarted.managed.(*OSManagedProcessSupervisor).Status(ProcessIdentity{Scope: processScope(root, assignment)})
	if resumed.ObservationSeq <= beforeResumeSeq || !resumed.UpdatedAt.After(beforeResumeUpdatedAt) {
		t.Fatalf("expected resume to refresh process observation, beforeSeq=%d after=%+v", beforeResumeSeq, resumed)
	}
	stop := executionAssignment(protocol.RunCapabilityProcessStop)
	stop.TargetKey = "actions/stop.json"
	stopped := restarted.Execute(stop)
	if stopped.State != lifecycleResultStateSucceeded || stopped.ExecutionResult.ProcessState != "stopped" {
		t.Fatalf("expected stopped process, got %+v", stopped)
	}
	if again := restarted.Execute(stop); again.State != lifecycleResultStateSucceeded {
		t.Fatalf("expected idempotent stop, got %+v", again)
	}
}

func TestTypedProcessUnexpectedExitIsReported(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	setupProcessWorkspace(t, root, assignment, true)
	executor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root))
	if result := executor.Execute(assignment); result.State != lifecycleResultStateSucceeded {
		t.Fatalf("start unexpected-exit fixture: %+v", result)
	}
	status := executionAssignment(protocol.RunCapabilityProcessStatus)
	status.TargetKey = "actions/status.json"
	result := executor.Execute(status)
	deadline := time.Now().Add(3 * time.Second)
	for result.ExecutionResult.ProcessState == "running" && time.Now().Before(deadline) {
		time.Sleep(50 * time.Millisecond)
		result = executor.Execute(status)
	}
	if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.ProcessState != "exited" || result.ExecutionResult.ExitClassification == "" {
		t.Fatalf("expected unexpected exit evidence, got %+v", result)
	}
}

func TestScopedFileExecutorRejectsEscapesAndWritesAtomically(t *testing.T) {
	root := t.TempDir()
	executor, err := NewFileExecutor(root)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesWrite)
	assignment.TargetKey = "config/server.properties"
	assignment.ExecutionInput.Content = "name=alpha\n"
	assignment.ExecutionInput.MaxReadBytes = 64 * 1024
	first := executor.Execute(context.Background(), assignment)
	if first.State != lifecycleResultStateSucceeded || first.ExecutionResult.Version != 1 {
		t.Fatalf("expected first atomic write, got %+v", first)
	}
	assignment.ExecutionInput.ExpectedVersion = 99
	conflict := executor.Execute(context.Background(), assignment)
	if conflict.State != lifecycleResultStateFailed || conflict.ErrorCode != "file_version_conflict" {
		t.Fatalf("expected version conflict, got %+v", conflict)
	}
	assignment.ExecutionInput.ExpectedVersion = 1
	assignment.ExecutionInput.ExpectedChecksum = first.ExecutionResult.Checksum
	assignment.ExecutionInput.Content = "name=beta\n"
	second := executor.Execute(context.Background(), assignment)
	if second.State != lifecycleResultStateSucceeded || second.ExecutionResult.Version != 2 {
		t.Fatalf("expected compare-and-swap write, got %+v", second)
	}

	symlinkTarget := filepath.Join(root, "outside.txt")
	if err := os.WriteFile(symlinkTarget, []byte("outside"), 0o600); err != nil {
		t.Fatalf("write outside fixture: %v", err)
	}
	scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	if err != nil {
		t.Fatalf("scope: %v", err)
	}
	if err := os.MkdirAll(filepath.Join(scope, "config"), 0o700); err != nil {
		t.Fatalf("mkdir config: %v", err)
	}
	if err := os.Symlink(symlinkTarget, filepath.Join(scope, "config", "link")); err != nil {
		t.Fatalf("symlink fixture: %v", err)
	}
	assignment.TargetKey = "config/link"
	if result := executor.Execute(context.Background(), assignment); result.State != lifecycleResultStateFailed {
		t.Fatalf("expected symlink rejection, got %+v", result)
	}
	assignment.TargetKey = "../outside"
	if result := executor.Execute(context.Background(), assignment); result.State != lifecycleResultStateFailed {
		t.Fatalf("expected traversal rejection, got %+v", result)
	}
}

func TestScopedFileExecutorBoundsReads(t *testing.T) {
	root := t.TempDir()
	executor, err := NewFileExecutor(root)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesRead)
	assignment.TargetKey = "logs/latest.log"
	assignment.ExecutionInput.MaxReadBytes = 4
	scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	if err != nil {
		t.Fatalf("scope: %v", err)
	}
	if err := os.MkdirAll(filepath.Join(scope, "logs"), 0o700); err != nil {
		t.Fatalf("mkdir logs: %v", err)
	}
	if err := os.WriteFile(filepath.Join(scope, assignment.TargetKey), []byte("too large"), 0o600); err != nil {
		t.Fatalf("write log fixture: %v", err)
	}
	result := executor.Execute(context.Background(), assignment)
	if result.State != lifecycleResultStateFailed || result.ErrorCode != "file_read_too_large" {
		t.Fatalf("expected bounded read failure, got %+v", result)
	}
}

func TestScopedFileExecutorListsDirectories(t *testing.T) {
	root := t.TempDir()
	executor, err := NewFileExecutor(root)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesList)
	assignment.TargetKey = "server-root"
	assignment.ExecutionInput.Inputs = map[string]string{"path": "", "recursive": "false", "query": ""}
	scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	if err != nil {
		t.Fatalf("scope: %v", err)
	}
	if err := os.MkdirAll(filepath.Join(scope, "config", "nested"), 0o700); err != nil {
		t.Fatalf("mkdir fixture: %v", err)
	}
	if err := os.WriteFile(filepath.Join(scope, "config", "server.properties"), []byte("name=example\n"), 0o600); err != nil {
		t.Fatalf("write fixture: %v", err)
	}
	result := executor.Execute(context.Background(), assignment)
	if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.Kind != "file.list" {
		t.Fatalf("expected directory listing, got %+v", result)
	}
	var envelope fileListEnvelope
	if err := json.Unmarshal([]byte(result.ExecutionResult.Content), &envelope); err != nil {
		t.Fatalf("decode listing: %v", err)
	}
	if envelope.DirectoryKey != "server-root" || len(envelope.Entries) != 1 || envelope.Entries[0].LogicalKey != "server-root/config" || envelope.Entries[0].Kind != "directory" {
		t.Fatalf("unexpected root listing: %+v", envelope)
	}
	assignment.ExecutionInput.Inputs["path"] = "config"
	assignment.ExecutionInput.Inputs["recursive"] = "true"
	result = executor.Execute(context.Background(), assignment)
	if result.State != lifecycleResultStateSucceeded {
		t.Fatalf("expected recursive listing, got %+v", result)
	}
	if err := json.Unmarshal([]byte(result.ExecutionResult.Content), &envelope); err != nil || len(envelope.Entries) != 2 {
		t.Fatalf("unexpected recursive listing: %+v err=%v", envelope, err)
	}
}

func TestDeploymentFileExecutorUsesServerRoot(t *testing.T) {
	workspaceRoot := t.TempDir()
	serverRoot := t.TempDir()
	if err := os.WriteFile(filepath.Join(workspaceRoot, ".platform"), []byte("workspace"), 0o600); err != nil {
		t.Fatalf("write workspace marker: %v", err)
	}
	if err := os.WriteFile(filepath.Join(serverRoot, "SCUMServer.exe"), []byte("server"), 0o600); err != nil {
		t.Fatalf("write server marker: %v", err)
	}
	executor, err := NewFileExecutor(workspaceRoot)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesList)
	assignment.TargetKey = "server-root"
	assignment.ExecutionInput.Inputs = map[string]string{"path": "", "recursive": "false", "query": ""}
	assignment.ExecutionInput.Deployment = &protocol.ServerDeploymentExecution{SchemaVersion: "1", Mode: "guided-install", ServerRoot: serverRoot, Revision: 1}
	result := executor.Execute(context.Background(), assignment)
	if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.Kind != "file.list" {
		t.Fatalf("expected deployment root listing, got %+v", result)
	}
	var envelope fileListEnvelope
	if err := json.Unmarshal([]byte(result.ExecutionResult.Content), &envelope); err != nil {
		t.Fatalf("decode deployment listing: %v", err)
	}
	if len(envelope.Entries) != 1 || envelope.Entries[0].Name != "SCUMServer.exe" {
		t.Fatalf("expected only deployment root entry, got %+v", envelope.Entries)
	}
}

func TestDeploymentFileExecutorListsNestedPathsFromServerRoot(t *testing.T) {
	workspaceRoot := t.TempDir()
	serverRoot := t.TempDir()
	if err := os.MkdirAll(filepath.Join(serverRoot, "SCUM", "Saved"), 0o700); err != nil {
		t.Fatalf("mkdir server root fixture: %v", err)
	}
	if err := os.WriteFile(filepath.Join(serverRoot, "SCUM", "Saved", "ServerSettings.ini"), []byte("[/Script/SCUM.ServerSettings]\n"), 0o600); err != nil {
		t.Fatalf("write server root fixture: %v", err)
	}
	executor, err := NewFileExecutor(workspaceRoot)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesList)
	assignment.TargetKey = "server-root"
	assignment.ExecutionInput.Deployment = &protocol.ServerDeploymentExecution{SchemaVersion: "1", Mode: "guided-install", ServerRoot: serverRoot, Revision: 7}
	assignment.ExecutionInput.Inputs = map[string]string{"path": "SCUM", "recursive": "true", "query": ""}
	result := executor.Execute(context.Background(), assignment)
	if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.Kind != "file.list" {
		t.Fatalf("expected deployment root listing, got %+v", result)
	}
	var envelope fileListEnvelope
	if err := json.Unmarshal([]byte(result.ExecutionResult.Content), &envelope); err != nil {
		t.Fatalf("decode listing: %v", err)
	}
	if envelope.Path != "SCUM" || len(envelope.Entries) != 2 || envelope.Entries[0].RelativePath != "SCUM/Saved" || envelope.Entries[1].RelativePath != "SCUM/Saved/ServerSettings.ini" {
		t.Fatalf("expected deployment-root relative paths, got %+v", envelope)
	}
}

func TestScopedFileExecutorCancellationLeavesTargetUnchanged(t *testing.T) {
	root := t.TempDir()
	executor, err := NewFileExecutor(root)
	if err != nil {
		t.Fatalf("new file executor: %v", err)
	}
	assignment := executionAssignment(protocol.RunCapabilityFilesWrite)
	assignment.TargetKey = "config/server.properties"
	assignment.ExecutionInput.Content = "cancelled=true\n"
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	result := executor.Execute(ctx, assignment)
	if result.State != lifecycleResultStateFailed || result.ErrorCode != "file_cancelled" {
		t.Fatalf("expected cancelled write, got %+v", result)
	}
	scope, _ := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	if _, err := os.Stat(filepath.Join(scope, assignment.TargetKey)); !os.IsNotExist(err) {
		t.Fatalf("cancelled write changed target: %v", err)
	}
}

func executionAssignment(capability string) protocol.RunJobAssignment {
	return protocol.RunJobAssignment{JobID: "execution-job", ServerInstanceID: "server-execution", RunEndpointID: "run-local", Capability: capability, TargetKey: "actions/start.json", InputRef: "input://server-execution/execution", LeaseToken: "lease", Attempt: 1, ExecutionInput: protocol.RunJobExecutionInput{WorkspaceScope: "local", MaxReadBytes: maxExecutionContentBytes}}
}

func processScope(root string, assignment protocol.RunJobAssignment) string {
	scope, _ := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	return scope
}

func setupProcessWorkspace(t *testing.T, root string, assignment protocol.RunJobAssignment, exits bool) {
	t.Helper()
	scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
	if err != nil {
		t.Fatalf("scope process fixture: %v", err)
	}
	if err := os.MkdirAll(filepath.Join(scope, "actions"), 0o700); err != nil {
		t.Fatalf("mkdir process fixture: %v", err)
	}
	if err := os.MkdirAll(filepath.Join(scope, "bin"), 0o700); err != nil {
		t.Fatalf("mkdir executable fixture: %v", err)
	}
	binary, err := os.Open(filepath.Join(filepath.Dir(mustExecutable(t)), filepath.Base(mustExecutable(t))))
	if err != nil {
		t.Fatalf("open test binary: %v", err)
	}
	defer binary.Close()
	target := filepath.Join(scope, "bin", "game-server")
	output, err := os.Create(target)
	if err != nil {
		t.Fatalf("create test binary: %v", err)
	}
	if _, err := io.Copy(output, binary); err != nil {
		t.Fatalf("copy test binary: %v", err)
	}
	if err := output.Chmod(0o700); err != nil {
		t.Fatalf("chmod test binary: %v", err)
	}
	if err := output.Close(); err != nil {
		t.Fatalf("close test binary: %v", err)
	}
	actionDir := filepath.Join(scope, "actions")
	start := map[string]any{"version": 1, "action": "start", "mode": "supervised", "executableKey": "bin/game-server", "arguments": []string{"-test.run=TestRunHelperProcess"}, "environment": map[string]string{"RUN_TEST_HELPER": "1"}}
	if exits {
		start["environment"] = map[string]string{"RUN_TEST_HELPER": "1", "RUN_EXIT_NOW": "1"}
	}
	writeJSONFixture(t, filepath.Join(actionDir, "start.json"), start)
	writeJSONFixture(t, filepath.Join(actionDir, "stop.json"), map[string]any{"version": 1, "action": "stop", "mode": "control"})
	writeJSONFixture(t, filepath.Join(actionDir, "status.json"), map[string]any{"version": 1, "action": "status", "mode": "control"})
}

func writeJSONFixture(t *testing.T, path string, value any) {
	t.Helper()
	body, err := json.Marshal(value)
	if err != nil {
		t.Fatalf("marshal fixture: %v", err)
	}
	if err := os.WriteFile(path, body, 0o600); err != nil {
		t.Fatalf("write fixture: %v", err)
	}
}

type contextRejectingLogSink struct {
	delegate ProcessLogSink
}

type alwaysRejectingLogSink struct{}

func (alwaysRejectingLogSink) Append(context.Context, protocol.RunJobAssignment, string, string) error {
	return context.Canceled
}

func (sink *contextRejectingLogSink) Append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	return sink.delegate.Append(ctx, assignment, stream, line)
}

func waitForFileText(t *testing.T, path string, expected string) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		body, err := os.ReadFile(path)
		if err == nil && strings.Contains(string(body), expected) {
			return
		}
		time.Sleep(25 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %q in managed output file", expected)
}

func waitForSpooledText(t *testing.T, logSpool spool.LogSpool, expected ...string) {
	t.Helper()
	deadline := time.Now().Add(4 * time.Second)
	for time.Now().Before(deadline) {
		joined := ""
		for _, batch := range mustPendingLogs(t, logSpool) {
			for _, entry := range batch.Entries {
				joined += entry.Line + "\n"
			}
		}
		matched := true
		for _, value := range expected {
			matched = matched && strings.Contains(joined, value)
		}
		if matched {
			return
		}
		time.Sleep(25 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for durable spool entries %q", expected)
}

func mustPendingLogs(t *testing.T, logSpool spool.LogSpool) []protocol.LogBatchIngestRequest {
	t.Helper()
	pending, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("read pending log spool: %v", err)
	}
	return pending
}

func waitForSpooledTextCount(t *testing.T, logSpool spool.LogSpool, expected string, minimum int) {
	t.Helper()
	deadline := time.Now().Add(4 * time.Second)
	for time.Now().Before(deadline) {
		count := 0
		for _, batch := range mustPendingLogs(t, logSpool) {
			for _, entry := range batch.Entries {
				if strings.Contains(entry.Line, expected) {
					count++
				}
			}
		}
		if count >= minimum {
			return
		}
		time.Sleep(25 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %d durable lines containing %q: %+v", minimum, expected, mustPendingLogs(t, logSpool))
}

func waitForManagedOffsets(t *testing.T, supervisor *OSManagedProcessSupervisor, scope string, stdoutAfter int64, stderrAfter int64) ProcessIdentity {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		identity := supervisor.Status(ProcessIdentity{Scope: scope})
		if identity.StdoutOffset > stdoutAfter && identity.StderrOffset > stderrAfter {
			return identity
		}
		time.Sleep(25 * time.Millisecond)
	}
	identity := supervisor.Status(ProcessIdentity{Scope: scope})
	t.Fatalf("timed out waiting for managed output offsets: %+v", identity)
	return ProcessIdentity{}
}

func waitForManagedState(t *testing.T, supervisor *OSManagedProcessSupervisor, scope string, expected string) ProcessIdentity {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		identity := supervisor.Status(ProcessIdentity{Scope: scope})
		if identity.State == expected {
			return identity
		}
		time.Sleep(25 * time.Millisecond)
	}
	identity := supervisor.Status(ProcessIdentity{Scope: scope})
	t.Fatalf("timed out waiting for managed process state %q: %+v", expected, identity)
	return ProcessIdentity{}
}

func waitForManagedTailers(t *testing.T, supervisor *OSManagedProcessSupervisor) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		supervisor.mu.Lock()
		count := len(supervisor.tailers)
		supervisor.mu.Unlock()
		if count == 0 {
			return
		}
		time.Sleep(25 * time.Millisecond)
	}
	supervisor.mu.Lock()
	count := len(supervisor.tailers)
	supervisor.mu.Unlock()
	t.Fatalf("timed out waiting for managed output tailers to drain: %d active", count)
}

func mustExecutable(t *testing.T) string {
	t.Helper()
	path, err := os.Executable()
	if err != nil {
		t.Fatalf("find test executable: %v", err)
	}
	return path
}

func TestExecutionResultDoesNotContainPrivateIdentity(t *testing.T) {
	root := t.TempDir()
	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
	setupProcessWorkspace(t, root, assignment, true)
	result := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root)).Execute(assignment)
	encoded, _ := json.Marshal(result)
	for _, forbidden := range []string{"\"pid\"", "/Users/", "lease-token", "session-token"} {
		if strings.Contains(string(encoded), forbidden) {
			t.Fatalf("execution result exposed %q: %s", forbidden, encoded)
		}
	}
}

func TestWorkerAdvertisesAndRoutesExecutionCapabilities(t *testing.T) {
	capabilities := SupportedRunCapabilities()
	for _, capability := range []string{protocol.RunCapabilityProcessStart, protocol.RunCapabilityProcessStop, protocol.RunCapabilityProcessStatus, protocol.RunCapabilityConfigWrite, protocol.RunCapabilityFilesRead, protocol.RunCapabilityFilesWrite} {
		if !supportedCapability(capabilities, capability) {
			t.Fatalf("expected worker capability %s, got %v", capability, capabilities)
		}
	}
}