run/runtime/worker_test.go

package runtime

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

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

func TestWorkerRegistersHeartbeatsAndStoresSession(t *testing.T) {
	client := newFakeWorkerClient()
	worker, err := NewWorker(workerTestConfig(t), client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}

	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	if worker.State().SessionToken != "session-token" {
		t.Fatalf("expected session token stored, got %+v", worker.State())
	}
	if len(client.helloRequests) != 1 || client.helloRequests[0].RegistrationToken != "registration-token" || len(client.helloRequests[0].CapabilityReport.Capabilities) == 0 {
		t.Fatalf("unexpected hello request: %+v", client.helloRequests)
	}

	if err := worker.HeartbeatOnce(context.Background()); err != nil {
		t.Fatalf("heartbeat: %v", err)
	}
	if len(client.heartbeatRequests) != 1 {
		t.Fatalf("expected heartbeat request")
	}
	heartbeat := client.heartbeatRequests[0]
	if heartbeat.SessionToken != "session-token" || heartbeat.Capacity.MaxJobs != 2 || heartbeat.Capacity.RunningJobs != 0 {
		t.Fatalf("unexpected heartbeat request: %+v", heartbeat)
	}
	for _, forbidden := range []string{"/Users/", "unix://", "Bearer ", "sk-", "password=", "artifact", "log"} {
		if containsText(heartbeat.Capacity.Summary, forbidden) {
			t.Fatalf("heartbeat summary exposed forbidden fragment %q: %+v", forbidden, heartbeat)
		}
	}
}

func TestManagedProcessLogStreamIsScopedToItsGeneration(t *testing.T) {
	assignment := protocol.RunJobAssignment{RunEndpointID: "run-test", ServerInstanceID: "server-test", JobID: "job-test", LogSessionID: "generation-2"}
	if got, want := logStreamIDForAssignment(assignment, "game.console.stdout"), "run.run-test.server-test.generation-2.game.console.stdout"; got != want {
		t.Fatalf("expected session-scoped supervised stream, got %q", got)
	}
	assignment.LogSessionID = ""
	if got, want := logStreamIDForAssignment(assignment, "game.console.stdout"), "job.job-test.game.console.stdout"; got != want {
		t.Fatalf("expected non-session job stream to retain legacy ID, got %q", got)
	}
}

func TestWorkerRenewsExpiredOrUnauthorizedSession(t *testing.T) {
	client := newFakeWorkerClient()
	worker, err := NewWorker(workerTestConfig(t), client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	worker.state.SessionExpiresAt = time.Now().UTC().Add(30 * time.Second)
	if err := worker.HeartbeatOnce(context.Background()); err != nil {
		t.Fatalf("renew expiring session: %v", err)
	}
	if len(client.helloRequests) != 2 || len(client.heartbeatRequests) != 0 {
		t.Fatalf("expected proactive hello renewal, hello=%d heartbeat=%d", len(client.helloRequests), len(client.heartbeatRequests))
	}

	client.heartbeatErr = api.PlatformRequestError{Status: http.StatusUnauthorized}
	worker.state.SessionExpiresAt = time.Time{}
	if err := worker.HeartbeatOnce(context.Background()); err != nil {
		t.Fatalf("recover unauthorized session: %v", err)
	}
	if len(client.helloRequests) != 3 || worker.State().SessionToken == "" {
		t.Fatalf("expected unauthorized heartbeat to re-register: state=%+v hello=%d", worker.State(), len(client.helloRequests))
	}

	client.heartbeatErr = api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"sessionToken is invalid"}}
	if err := worker.HeartbeatOnce(context.Background()); err != nil {
		t.Fatalf("recover legacy invalid session: %v", err)
	}
	if len(client.helloRequests) != 4 || worker.State().SessionToken == "" {
		t.Fatalf("expected legacy invalid session to re-register: state=%+v hello=%d", worker.State(), len(client.helloRequests))
	}
}

func TestWorkerReRegistersWhenClaimReportsInvalidSession(t *testing.T) {
	client := newFakeWorkerClient()
	client.claimErr = api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"sessionToken is invalid"}}
	worker, err := NewWorker(workerTestConfig(t), client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	if handled, err := worker.ClaimAndRunOnce(context.Background()); err != nil || handled {
		t.Fatalf("claim session renewal handled=%v err=%v", handled, err)
	}
	if len(client.helloRequests) != 2 || len(client.reconcileRequests) != 1 {
		t.Fatalf("expected claim session renewal, hello=%d reconcile=%d", len(client.helloRequests), len(client.reconcileRequests))
	}
}

func TestWorkerSerializesConcurrentSessionRefresh(t *testing.T) {
	client := newRotatingSessionClient()
	worker, err := NewWorker(workerTestConfig(t), client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	observedToken := worker.State().SessionToken

	start := make(chan struct{})
	errs := make(chan error, 2)
	for i := 0; i < 2; i++ {
		go func() {
			<-start
			errs <- worker.reregisterAndReconcile(context.Background(), "test_concurrent_refresh", observedToken)
		}()
	}
	close(start)
	for i := 0; i < 2; i++ {
		if err := <-errs; err != nil {
			t.Fatalf("refresh %d: %v", i, err)
		}
	}
	if client.HelloCount() != 2 {
		t.Fatalf("expected initial registration plus one refresh, got %d hello calls", client.HelloCount())
	}
	if got := worker.State().SessionToken; got != "session-token-2" {
		t.Fatalf("expected refreshed session token, got %q", got)
	}
	if len(client.reconcileRequests) != 2 {
		t.Fatalf("expected both refresh callers to reconcile, got %d", len(client.reconcileRequests))
	}
}

func TestWorkerClaimsAcksProgressAndCompletesJob(t *testing.T) {
	client := newFakeWorkerClient()
	client.claimJob = workerJobAssignment(protocol.RunCapabilityProcessStart)
	client.claimJob.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{{Key: "game-console-stdout", Kind: "process.stdout", StreamKey: "game.console.stdout", CursorKind: "sequence"}}
	worker, err := NewWorker(workerTestConfig(t), client, WithProcessSupervisor(staticSupervisor{stdout: "server ready\n"}))
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}

	handled, err := worker.ClaimAndRunOnce(context.Background())
	if err != nil || !handled {
		t.Fatalf("claim/run handled=%v err=%v", handled, err)
	}
	if len(client.ackRequests) != 1 || len(client.progressRequests) != 1 || len(client.resultRequests) != 1 || len(client.cancelPollRequests) != 1 {
		t.Fatalf("expected ack/progress/result/cancel calls, got ack=%d progress=%d result=%d cancel=%d", len(client.ackRequests), len(client.progressRequests), len(client.resultRequests), len(client.cancelPollRequests))
	}
	if len(client.claimRequests) != 1 || client.claimRequests[0].WaitSeconds != jobClaimWaitSeconds {
		t.Fatalf("expected persistent claim wait, got %+v", client.claimRequests)
	}
	if client.progressRequests[0].Progress.Percent != 10 || client.resultRequests[0].State != "succeeded" || client.resultRequests[0].ResultRef == "" {
		t.Fatalf("unexpected job channel payloads: progress=%+v result=%+v", client.progressRequests[0], client.resultRequests[0])
	}
	if worker.journal.ActiveCount() != 0 {
		t.Fatalf("expected terminal job removed from journal")
	}
}

func TestWorkerDispatchesSelfUpdateJob(t *testing.T) {
	client := newFakeWorkerClient()
	assignment := workerJobAssignment(protocol.RunCapabilityRunSelfUpdate)
	assignment.TargetKey = "run/update"
	assignment.InputRef = "artifact://artifact-run-latest"
	client.claimJob = assignment
	executableName := "run"
	if runtime.GOOS == "windows" {
		executableName = "run.exe"
	}
	client.updatePayload = []byte("staged run binary")
	client.updateInput = protocol.RunUpdateInputResponse{JobID: assignment.JobID, ServerInstanceID: assignment.ServerInstanceID, RunEndpointID: assignment.RunEndpointID, ArtifactID: "artifact-run-latest", Checksum: bytesChecksum(client.updatePayload), SizeBytes: int64(len(client.updatePayload)), TargetOS: runtime.GOOS, TargetArch: runtime.GOARCH, PackageFormat: "raw-executable", ExecutableName: executableName, TargetRelease: "run-release-test", ChunkSizeBytes: 64}
	activator := &recordingSelfUpdateActivator{}
	worker, err := NewWorker(workerTestConfig(t), client, WithSelfUpdateActivator(activator))
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}

	handled, err := worker.ClaimAndRunOnce(context.Background())
	if err != nil || !handled {
		t.Fatalf("claim/run handled=%v err=%v", handled, err)
	}
	if len(client.resultRequests) != 1 || client.resultRequests[0].State != "succeeded" || !strings.Contains(client.resultRequests[0].ResultRef, "run-update-staged") {
		t.Fatalf("expected self-update result, got %+v", client.resultRequests)
	}
	if activator.manifestPath == "" {
		t.Fatal("expected activation only after accepted terminal result")
	}
}

func TestWorkerFileReadLargeFileUploadsArtifact(t *testing.T) {
	cfg := workerTestConfig(t)
	client := newFakeWorkerClient()
	assignment := workerJobAssignment(protocol.RunCapabilityFilesRead)
	assignment.TargetKey = "logs/big.log"
	assignment.ExecutionInput.WorkspaceScope = "run-local"
	assignment.ExecutionInput.MaxReadBytes = 4
	client.claimJob = assignment
	client.buildInput = protocol.DistributionBuildInputResponse{JobID: assignment.JobID}
	scope, err := NewWorkspaceResolver(cfg.WorkspaceRoot).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 fixture: %v", err)
	}
	payload := []byte(strings.Repeat("A", fileArtifactChunkSize) + "tail")
	if err := os.WriteFile(filepath.Join(scope, assignment.TargetKey), payload, 0o600); err != nil {
		t.Fatalf("write fixture: %v", err)
	}
	worker, err := NewWorker(cfg, client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}

	handled, err := worker.ClaimAndRunOnce(context.Background())
	if err != nil || !handled {
		t.Fatalf("claim/run handled=%v err=%v", handled, err)
	}
	if len(client.artifactOpenRequests) != 1 {
		t.Fatalf("expected artifact transfer, got %+v", client.artifactOpenRequests)
	}
	opened := client.artifactOpenRequests[0]
	if opened.OwnerID != assignment.JobID || opened.SizeBytes != int64(len(payload)) || opened.Checksum != bytesChecksum(payload) || opened.ChunkSizeBytes != fileArtifactChunkSize {
		t.Fatalf("unexpected artifact open: %+v", opened)
	}
	if string(client.artifactPayload) != string(payload) {
		t.Fatalf("uploaded artifact payload mismatch")
	}
	if len(client.resultRequests) != 1 || client.resultRequests[0].State != "succeeded" || !strings.HasPrefix(client.resultRequests[0].ResultRef, "artifact://artifact-job-worker-file-read") {
		t.Fatalf("expected artifact result ref, got %+v", client.resultRequests)
	}
	if client.resultRequests[0].ExecutionResult.Content != "" || client.resultRequests[0].ExecutionResult.SizeBytes != int64(len(payload)) {
		t.Fatalf("large file result must not inline content: %+v", client.resultRequests[0].ExecutionResult)
	}
}

func TestWorkerRegistersPackageIdentity(t *testing.T) {
	client := newFakeWorkerClient()
	cfg := workerTestConfig(t)
	cfg.RegistrationToken = "current-run-key"
	cfg.ServerInstanceID = "server-worker"
	cfg.PluginID = "game.minecraft"
	cfg.ComponentKind = "run"
	cfg.KeyGeneration = 7
	worker, err := NewWorker(cfg, client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}

	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	hello := client.helloRequests[0]
	if hello.RegistrationToken != "current-run-key" || hello.ServerInstanceID != "server-worker" || hello.ComponentKind != "run" || hello.KeyGeneration != 7 {
		t.Fatalf("expected package identity in hello request, got %+v", hello)
	}
}

func TestWorkerReportsSelfUpdateHealthOnlyAfterRegistrationAndReconciliation(t *testing.T) {
	client := newFakeWorkerClient()
	cfg := workerTestConfig(t)
	cfg.Version = "run-release-2"
	cfg.UpdateJobID = "job-update-health"
	cfg.UpdateOutcome = "succeeded"
	cfg.UpdateAttempt = 2
	cfg.UpdateLeaseToken = "lease-update-health"
	worker, err := NewWorker(cfg, client)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	if len(client.helloRequests) != 1 || client.helloRequests[0].UpdateOutcome != "" || len(client.updateHealthReports) != 0 {
		t.Fatalf("hello must not report update success before reconciliation: hello=%+v reports=%+v", client.helloRequests, client.updateHealthReports)
	}
	if err := worker.ReconcileOnce(context.Background()); err != nil {
		t.Fatalf("reconcile: %v", err)
	}
	if err := worker.reportRunUpdateHealth(context.Background()); err != nil {
		t.Fatalf("report reconciled update health: %v", err)
	}
	if len(client.updateHealthReports) != 1 {
		t.Fatalf("expected one update health report, got %+v", client.updateHealthReports)
	}
	report := client.updateHealthReports[0]
	if report.SessionToken != "session-token" || report.JobID != cfg.UpdateJobID || report.Attempt != cfg.UpdateAttempt || report.LeaseToken != cfg.UpdateLeaseToken || report.Version != cfg.Version || report.Outcome != "succeeded" {
		t.Fatalf("unexpected update health report: %+v", report)
	}
}

func TestWorkerHandlesCancellationAndReconcile(t *testing.T) {
	client := newFakeWorkerClient()
	client.claimJob = workerJobAssignment(protocol.RunCapabilityProcessStart)
	client.cancelResponse = protocol.RunJobCancelPollResponse{Accepted: true, RunEndpointID: "run-test", HasCancel: true, JobID: "job-worker", Reason: "operator requested", ServerTime: workerTestTime()}
	worker, err := NewWorker(workerTestConfig(t), client, WithProcessSupervisor(blockingSupervisor{}))
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}

	handled, err := worker.ClaimAndRunOnce(context.Background())
	if err != nil || !handled {
		t.Fatalf("claim/run handled=%v err=%v", handled, err)
	}
	if len(client.resultRequests) != 1 || client.resultRequests[0].State != "cancelled" || client.resultRequests[0].ErrorCode != "lifecycle_cancelled" {
		t.Fatalf("expected cancelled terminal result, got %+v", client.resultRequests)
	}

	worker.journal.MarkActive(workerJobAssignment(protocol.RunCapabilityProcessStart))
	client.reconcileResponse = protocol.RunJobReconcileResponse{
		Accepted:      true,
		RunEndpointID: "run-test",
		ConfirmedJobs: []protocol.RunJobAssignment{workerJobAssignment(protocol.RunCapabilityProcessStop)},
		DiscardJobIDs: []string{"job-worker"},
		ServerTime:    workerTestTime(),
	}
	if err := worker.ReconcileOnce(context.Background()); err != nil {
		t.Fatalf("reconcile: %v", err)
	}
	if ids := worker.journal.ActiveJobIDs(); !reflect.DeepEqual(ids, []string{"job-worker-stop"}) {
		t.Fatalf("expected reconcile to replace active job ids, got %+v", ids)
	}
}

func TestWorkerSpoolHooksUseRegisteredSession(t *testing.T) {
	client := newFakeWorkerClient()
	client.claimJob = workerJobAssignment(protocol.RunCapabilityProcessStart)
	client.claimJob.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{{Key: "game-console-stdout", Kind: "process.stdout", StreamKey: "game.console.stdout", CursorKind: "sequence"}}
	logSpool, err := spool.NewLogSpool(t.TempDir())
	if err != nil {
		t.Fatalf("log spool: %v", err)
	}
	artifactQueue, err := spool.NewArtifactQueue(t.TempDir())
	if err != nil {
		t.Fatalf("artifact queue: %v", err)
	}
	worker, err := NewWorker(
		workerTestConfig(t),
		client,
		WithProcessSupervisor(staticSupervisor{stdout: "started password=hidden\n"}),
		WithProcessLogSink(&SpoolLogSink{Spool: logSpool}),
		WithLifecycleArtifactHook(&QueueArtifactHook{Queue: artifactQueue}),
	)
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	if _, err := worker.ClaimAndRunOnce(context.Background()); err != nil {
		t.Fatalf("claim/run: %v", err)
	}
	logs, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("pending logs: %v", err)
	}
	if len(logs) != 1 || logs[0].RunEndpointID != "run-test" || logs[0].SessionToken != "session-token" || logs[0].StreamKey != "game.console.stdout" || logs[0].Entries[0].Line != "started password=hidden" || logs[0].Entries[0].Redacted {
		t.Fatalf("unexpected spooled logs: %+v", logs)
	}
	chunks, err := artifactQueue.Pending()
	if err != nil {
		t.Fatalf("pending artifact chunks: %v", err)
	}
	if len(chunks) != 1 || chunks[0].RunEndpointID != "run-test" || chunks[0].SessionToken != "session-token" {
		t.Fatalf("unexpected artifact chunks: %+v", chunks)
	}
}

func TestLiveLogSinkQueuesCurrentBatchWithoutDurableSpool(t *testing.T) {
	client := &recordingLiveLogClient{received: make(chan protocol.LogBatchIngestRequest, 1)}
	sink := NewLiveLogSink(client)
	sink.SetSession("run-test", "session-token")
	t.Cleanup(sink.Close)
	assignment := protocol.RunJobAssignment{
		JobID:            "execution-job",
		RunEndpointID:    "run-test",
		ServerInstanceID: "server-worker",
		Capability:       protocol.RunCapabilityProcessStart,
		LogSessionID:     "generation-current",
		SessionStartedAt: workerTestTime(),
		ExecutionInput:   protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "game.console.stdout"}}},
	}
	if err := sink.Append(context.Background(), assignment, "stdout", "current output"); err != nil {
		t.Fatalf("append live log: %v", err)
	}
	select {
	case batch := <-client.received:
		if batch.LogStreamID != "run.run-test.server-worker.generation-current.game.console.stdout" || batch.FirstSeq != 1 || batch.LastSeq != 1 || batch.Checksum == "" || len(batch.Entries) != 1 || batch.Entries[0].Line != "current output" {
			t.Fatalf("unexpected live batch: %+v", batch)
		}
	case <-time.After(time.Second):
		t.Fatal("live log was not dispatched")
	}
}

func TestSpoolLogSinkUsesRunScopedStreamForAutonomousLifecycle(t *testing.T) {
	logSpool, err := spool.NewLogSpool(t.TempDir())
	if err != nil {
		t.Fatalf("log spool: %v", err)
	}
	sink := &SpoolLogSink{Spool: logSpool}
	assignment := protocol.RunJobAssignment{
		JobID:            "autonomous-bootstrap-start",
		RunEndpointID:    "run-test",
		ServerInstanceID: "server-worker",
		Capability:       protocol.RunCapabilityProcessStart,
		IdempotencyKey:   "autonomous:release:start",
		LeaseToken:       "local-autonomous-bootstrap",
		ExecutionInput: protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{
			{Key: "game-console-stdout", Kind: "process.stdout", StreamKey: "game.console.stdout", CursorKind: "sequence"},
		}},
	}

	if err := sink.Append(context.Background(), assignment, "stdout", "autonomous output"); err != nil {
		t.Fatalf("append autonomous log: %v", err)
	}
	logs, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("pending logs: %v", err)
	}
	if len(logs) != 1 || logs[0].LogStreamID != "run.run-test.server-worker.game.console.stdout" || logs[0].StreamKey != "game.console.stdout" {
		t.Fatalf("expected run-scoped autonomous stream, got %+v", logs)
	}
}

func TestSpoolLogSinkDurablyAppendsFreshSessionWithCanceledWorkerContext(t *testing.T) {
	logSpool, err := spool.NewLogSpool(t.TempDir())
	if err != nil {
		t.Fatalf("log spool: %v", err)
	}
	progressCalls := 0
	sink := &SpoolLogSink{RunEndpointID: "run-test", SessionToken: "session-token", Spool: logSpool, Progress: func(context.Context, string, string) (uint64, error) {
		progressCalls++
		return 0, context.Canceled
	}}
	assignment := protocol.RunJobAssignment{
		JobID:            "managed-start",
		RunEndpointID:    "run-test",
		ServerInstanceID: "server-worker",
		Capability:       protocol.RunCapabilityProcessStart,
		LogSessionID:     "generation-current",
		SessionStartedAt: workerTestTime(),
		ExecutionInput: protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{
			{Kind: "process.stdout", StreamKey: "game.console.stdout"},
		}},
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if err := sink.Append(ctx, assignment, "stdout", "output during shutdown"); err != nil {
		t.Fatalf("append fresh session with canceled worker context: %v", err)
	}
	if progressCalls != 0 {
		t.Fatalf("fresh generation unexpectedly depended on remote progress: calls=%d", progressCalls)
	}
	pending, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("pending logs: %v", err)
	}
	if len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].LogSessionID != assignment.LogSessionID || pending[0].Entries[0].Line != "output during shutdown" {
		t.Fatalf("expected durable session batch despite canceled context, got %+v", pending)
	}
}

func TestGenericWorkerJobLogDoesNotUseRunStreamProgress(t *testing.T) {
	client := newFakeWorkerClient()
	cfg := workerTestConfig(t)
	cfg.ServerInstanceID = ""
	logSpool, err := spool.NewLogSpool(t.TempDir())
	if err != nil {
		t.Fatalf("new log spool: %v", err)
	}
	sink := &SpoolLogSink{Spool: logSpool}
	worker, err := NewWorker(cfg, client, WithProcessLogSink(sink))
	if err != nil {
		t.Fatalf("new generic worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register generic worker: %v", err)
	}
	assignment := protocol.RunJobAssignment{JobID: "job-generic", RunEndpointID: cfg.RunEndpointID, ServerInstanceID: "server-from-job", Capability: protocol.RunCapabilityProcessStart, ExecutionInput: protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "game.console.stdout"}}}}
	if err := sink.Append(context.Background(), assignment, "stdout", "generic job output"); err != nil {
		t.Fatalf("append generic job output: %v", err)
	}
	if len(client.logProgressRequests) != 0 {
		t.Fatalf("job-scoped stream must not use run-scoped progress recovery: %+v", client.logProgressRequests)
	}
	pending, err := logSpool.Pending()
	if err != nil || len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].ServerInstanceID != assignment.ServerInstanceID {
		t.Fatalf("generic job output did not enter the local spool: pending=%+v err=%v", pending, err)
	}
}

func TestManagedProcessObservationIDBindsLogSession(t *testing.T) {
	identity := ProcessIdentity{LogSessionID: "generation-current", RunEndpointID: "run-test", ServerInstanceID: "server-worker"}
	if got := managedProcessObservationID(identity); got != "log-session:generation-current" {
		t.Fatalf("managed process observation did not bind the log session: %q", got)
	}
}

func TestSpoolLogSinkReconcilesStableStreamBeforeAllocating(t *testing.T) {
	logSpool, err := spool.NewLogSpool(t.TempDir())
	if err != nil {
		t.Fatalf("log spool: %v", err)
	}
	client := newFakeWorkerClient()
	client.progressResponse.LatestSeq = 3816
	sink := &SpoolLogSink{Spool: logSpool, RunEndpointID: "run-test", SessionToken: "session-token", Progress: func(ctx context.Context, serverInstanceID string, streamID string) (uint64, error) {
		response, err := client.GetRunLogStreamProgress(ctx, protocol.RunLogStreamProgressRequest{RunEndpointID: "run-test", SessionToken: "session-token", ServerInstanceID: serverInstanceID, LogStreamID: streamID})
		return response.LatestSeq, err
	}}
	assignment := protocol.RunJobAssignment{
		JobID:            "autonomous-bootstrap-start",
		RunEndpointID:    "run-test",
		ServerInstanceID: "server-worker",
		Capability:       protocol.RunCapabilityProcessStart,
		IdempotencyKey:   "autonomous:release:start",
		LeaseToken:       "local-autonomous-bootstrap",
		ExecutionInput:   protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "scum.console.stdout"}}},
	}

	if err := sink.Append(context.Background(), assignment, "stdout", "current output"); err != nil {
		t.Fatalf("append: %v", err)
	}
	pending, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("pending logs: %v", err)
	}
	if len(pending) != 1 || pending[0].FirstSeq != 3817 || pending[0].LastSeq != 3817 {
		t.Fatalf("expected reconciled sequence 3817, got %+v", pending)
	}
	if len(client.logProgressRequests) != 1 || client.logProgressRequests[0].LogStreamID != "run.run-test.server-worker.scum.console.stdout" || client.logProgressRequests[0].ServerInstanceID != assignment.ServerInstanceID {
		t.Fatalf("expected one scoped progress query, got %+v", client.logProgressRequests)
	}
}

func TestSpoolLogSinkKeepsSequencesIndependentAndDurable(t *testing.T) {
	root := t.TempDir()
	logSpool, err := spool.NewLogSpool(root)
	if err != nil {
		t.Fatalf("log spool: %v", err)
	}
	sink := &SpoolLogSink{Spool: logSpool}
	assignment := protocol.RunJobAssignment{JobID: "autonomous-bootstrap-start", RunEndpointID: "run-test", ServerInstanceID: "server-worker", Capability: protocol.RunCapabilityProcessStart, IdempotencyKey: "autonomous:release:start", LeaseToken: "local-autonomous-bootstrap", ExecutionInput: protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "stdout"}, {Kind: "process.stderr", StreamKey: "stderr"}}}}
	for _, item := range []struct {
		stream string
		line   string
	}{{"stdout", "out-1"}, {"stderr", "err-1"}, {"stdout", "out-2"}} {
		if err := sink.Append(context.Background(), assignment, item.stream, item.line); err != nil {
			t.Fatalf("append %s: %v", item.stream, err)
		}
	}
	pending, err := logSpool.Pending()
	if err != nil {
		t.Fatalf("pending logs: %v", err)
	}
	pendingByStream := map[string]protocol.LogBatchIngestRequest{}
	for _, batch := range pending {
		pendingByStream[batch.StreamKey] = batch
	}
	stdoutBatch := pendingByStream["stdout"]
	stderrBatch := pendingByStream["stderr"]
	if len(pending) != 2 || stderrBatch.StreamKey != "stderr" || stderrBatch.FirstSeq != 1 || stderrBatch.LastSeq != 1 || stdoutBatch.StreamKey != "stdout" || stdoutBatch.FirstSeq != 1 || stdoutBatch.LastSeq != 2 || len(stdoutBatch.Entries) != 2 {
		t.Fatalf("expected independent per-stream sequences, got %+v", pending)
	}
	client := &recordingDurableLogClient{}
	if _, err := logSpool.Flush(context.Background(), client); err != nil {
		t.Fatalf("flush logs: %v", err)
	}
	restarted, err := spool.NewLogSpool(root)
	if err != nil {
		t.Fatalf("reopen log spool: %v", err)
	}
	restartedSink := &SpoolLogSink{Spool: restarted}
	if err := restartedSink.Append(context.Background(), assignment, "stdout", "out-3"); err != nil {
		t.Fatalf("append after restart: %v", err)
	}
	pending, err = restarted.Pending()
	if err != nil {
		t.Fatalf("pending restarted logs: %v", err)
	}
	if len(pending) != 1 || pending[0].FirstSeq != 3 {
		t.Fatalf("expected durable stdout sequence 3 after restart, got %+v", pending)
	}
}

func TestSessionLogBatchClientOverridesSpooledIdentityAndChecksum(t *testing.T) {
	recorder := &recordingDurableLogClient{}
	entry := protocol.LogEntry{Seq: 7, Timestamp: workerTestTime(), Level: "info", Line: "server ready", Redacted: false}
	batch := protocol.LogBatchIngestRequest{
		RunEndpointID:    "old-endpoint",
		SessionToken:     "old-token",
		LogStreamID:      "job.job-1.stdout",
		ServerInstanceID: "server-worker",
		StreamKey:        "stdout",
		Source:           "process",
		FirstSeq:         7,
		LastSeq:          7,
		Checksum:         "sha256:0000000000000000000000000000000000000000000000000000000000000000",
		Entries:          []protocol.LogEntry{entry},
	}
	client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	response, err := client.IngestLogBatch(context.Background(), batch)
	if err != nil || !response.Accepted {
		t.Fatalf("ingest through session client accepted=%t err=%v", response.Accepted, err)
	}
	if recorder.batch.RunEndpointID != "run-current" || recorder.batch.SessionToken != "token-current" {
		t.Fatalf("expected current identity, got %+v", recorder.batch)
	}
	expectedChecksum, err := checksumForLogEntries([]protocol.LogEntry{entry})
	if err != nil {
		t.Fatalf("checksum: %v", err)
	}
	if recorder.batch.Checksum != expectedChecksum {
		t.Fatalf("expected recomputed checksum %s, got %s", expectedChecksum, recorder.batch.Checksum)
	}
}

func TestSessionLogBatchClientQuarantinesSequenceGap(t *testing.T) {
	recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"log batch firstSeq must follow latest acknowledged sequence"}}}
	client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	_, err := client.IngestLogBatch(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "job.example.stdout", FirstSeq: 2256, LastSeq: 2256, Entries: []protocol.LogEntry{{Seq: 2256, Timestamp: workerTestTime(), Line: "line"}}})
	var permanent spool.PermanentLogBatchError
	if !errors.As(err, &permanent) || permanent.Reason != "platform_sequence_gap" {
		t.Fatalf("expected permanent platform_sequence_gap rejection, got %#v", err)
	}
}

func TestSessionLogBatchClientRecoversLatestSeqThroughProgressEndpoint(t *testing.T) {
	workerClient := newFakeWorkerClient()
	workerClient.logStreamProgress = 736
	client := sessionLogBatchClient{client: &recordingDurableLogClient{}, progressClient: workerClient, runEndpointID: "run-current", sessionToken: "token-current"}
	latestSeq, err := client.LogStreamLatestSeq(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.run-current.server-worker.session.stdout", ServerInstanceID: "server-worker", FirstSeq: 10413})
	if err != nil || latestSeq != 736 {
		t.Fatalf("recover latest sequence: latestSeq=%d err=%v", latestSeq, err)
	}
	if len(workerClient.logProgressRequests) != 1 || workerClient.logProgressRequests[0].RunEndpointID != "run-current" || workerClient.logProgressRequests[0].SessionToken != "token-current" || workerClient.logProgressRequests[0].ServerInstanceID != "server-worker" || workerClient.logProgressRequests[0].LogStreamID != "run.run-current.server-worker.session.stdout" {
		t.Fatalf("unexpected progress request: %+v", workerClient.logProgressRequests)
	}
}

func TestSessionLogBatchClientQuarantinesSequenceConflict(t *testing.T) {
	recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"log batch conflicts with acknowledged range"}}}
	client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	_, err := client.IngestLogBatch(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.run-current.server-worker.scum.console.stdout", FirstSeq: 1, LastSeq: 1, Entries: []protocol.LogEntry{{Seq: 1, Timestamp: workerTestTime(), Line: "line"}}})
	var permanent spool.PermanentLogBatchError
	if !errors.As(err, &permanent) || permanent.Reason != "platform_acknowledged_range_conflict" {
		t.Fatalf("expected permanent platform_acknowledged_range_conflict rejection, got %#v", err)
	}
}

func TestSessionLogBatchClientQuarantinesPlatformNotFound(t *testing.T) {
	recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusNotFound, Code: "not_found"}}
	client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	_, err := client.IngestLogBatch(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.stale-endpoint.stale-server.session.stdout", FirstSeq: 1, LastSeq: 1, Entries: []protocol.LogEntry{{Seq: 1, Timestamp: workerTestTime(), Line: "line"}}})
	var permanent spool.PermanentLogBatchError
	if !errors.As(err, &permanent) || permanent.Reason != "platform_not_found" {
		t.Fatalf("expected permanent platform_not_found rejection, got %#v", err)
	}
}

func TestSessionLogBatchClientQuarantinesSessionMetadataMismatch(t *testing.T) {
	recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"log session metadata must match stream"}}}
	client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	_, err := client.IngestLogBatch(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.run-current.server-worker.session.stdout", LogSessionID: "session-old", SessionStartedAt: workerTestTime(), FirstSeq: 1, LastSeq: 1, Entries: []protocol.LogEntry{{Seq: 1, Timestamp: workerTestTime(), Line: "line"}}})
	var permanent spool.PermanentLogBatchError
	if !errors.As(err, &permanent) || permanent.Reason != "session_metadata_mismatch" {
		t.Fatalf("expected permanent session_metadata_mismatch rejection, got %#v", err)
	}
}

func TestSessionArtifactChunkClientOverridesSpooledIdentity(t *testing.T) {
	recorder := &recordingDurableArtifactClient{}
	chunk := protocol.ArtifactChunkUploadRequest{RunEndpointID: "old-endpoint", SessionToken: "old-token", TransferID: "transfer-1", ArtifactID: "artifact-1", ChunkIndex: 2}
	client := sessionArtifactChunkClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	response, err := client.UploadArtifactChunk(context.Background(), chunk)
	if err != nil || !response.Accepted {
		t.Fatalf("upload through session client accepted=%t err=%v", response.Accepted, err)
	}
	if recorder.chunk.RunEndpointID != "run-current" || recorder.chunk.SessionToken != "token-current" {
		t.Fatalf("expected current identity, got %+v", recorder.chunk)
	}
}

func TestSessionArtifactChunkClientDropsPlatformMissingTransfer(t *testing.T) {
	recorder := &recordingDurableArtifactClient{err: api.PlatformRequestError{Status: http.StatusNotFound, Code: "not_found"}}
	chunk := protocol.ArtifactChunkUploadRequest{RunEndpointID: "old-endpoint", SessionToken: "old-token", TransferID: "transfer-stale", ArtifactID: "artifact-stale", ChunkIndex: 3}
	client := sessionArtifactChunkClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
	response, err := client.UploadArtifactChunk(context.Background(), chunk)
	if err != nil || !response.Accepted || response.TransferID != chunk.TransferID || response.ChunkIndex != chunk.ChunkIndex {
		t.Fatalf("expected stale chunk ack for queue cleanup, response=%+v err=%v", response, err)
	}
	if recorder.chunk.RunEndpointID != "run-current" || recorder.chunk.SessionToken != "token-current" {
		t.Fatalf("expected current identity before stale drop, got %+v", recorder.chunk)
	}
}

func TestWorkerRetryBackoffIsBounded(t *testing.T) {
	if got := boundedRetryBackoff(75 * time.Millisecond); got != 75*time.Millisecond {
		t.Fatalf("expected configured backoff, got %s", got)
	}
	if got := boundedRetryBackoff(time.Minute); got != 30*time.Second {
		t.Fatalf("expected capped backoff, got %s", got)
	}
}

func TestWorkerIntegrationWithPlatformLikeServer(t *testing.T) {
	assignment := workerJobAssignment(protocol.RunCapabilityProcessStart)
	seen := []string{}
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		seen = append(seen, r.URL.Path)
		switch r.URL.Path {
		case "/api/v1/run/control/hello":
			var request protocol.RunHelloRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.RunEndpointID != "run-test" || request.Capacity.MaxJobs != 2 {
				t.Fatalf("unexpected hello: %+v", request)
			}
			writeWorkerTestJSON(t, w, protocol.RunHelloResponse{Accepted: true, RunEndpointID: request.RunEndpointID, SessionToken: "session-token", HeartbeatIntervalSeconds: 15, ServerTime: workerTestTime()})
		case "/api/v1/run/control/heartbeat":
			var request protocol.RunHeartbeatRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.SessionToken != "session-token" || request.Capacity.RunningJobs != 0 {
				t.Fatalf("unexpected heartbeat: %+v", request)
			}
			writeWorkerTestJSON(t, w, protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: workerTestTime()})
		case "/api/v1/run/jobs/claim":
			var request protocol.RunJobClaimRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.SessionToken != "session-token" || len(request.Capabilities) == 0 || request.WaitSeconds != jobClaimWaitSeconds {
				t.Fatalf("unexpected claim: %+v", request)
			}
			writeWorkerTestJSON(t, w, protocol.RunJobClaimResponse{Accepted: true, RunEndpointID: request.RunEndpointID, HasJob: true, Job: &assignment, NextPollSeconds: 2, ServerTime: workerTestTime()})
		case "/api/v1/run/jobs/ack":
			var request protocol.RunJobAckRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.JobID != assignment.JobID || request.LeaseToken != assignment.LeaseToken {
				t.Fatalf("unexpected ack: %+v", request)
			}
			assignment.State = "running"
			writeWorkerTestJSON(t, w, protocol.RunJobAckResponse{Accepted: true, Job: assignment, ServerTime: workerTestTime()})
		case "/api/v1/run/jobs/progress":
			var request protocol.RunJobProgressRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.Progress.Percent != 10 {
				t.Fatalf("unexpected progress: %+v", request)
			}
			assignment.Progress = request.Progress
			writeWorkerTestJSON(t, w, protocol.RunJobProgressResponse{Accepted: true, Job: assignment, ServerTime: workerTestTime()})
		case "/api/v1/run/jobs/cancel":
			var request protocol.RunJobCancelPollRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.JobID != assignment.JobID {
				t.Fatalf("unexpected cancel poll: %+v", request)
			}
			writeWorkerTestJSON(t, w, protocol.RunJobCancelPollResponse{Accepted: true, RunEndpointID: request.RunEndpointID, ServerTime: workerTestTime()})
		case "/api/v1/run/jobs/result":
			var request protocol.RunJobResultRequest
			decodeWorkerTestJSON(t, r, &request)
			if request.State != "succeeded" || request.ResultRef == "" {
				t.Fatalf("unexpected result: %+v", request)
			}
			assignment.State = request.State
			assignment.ResultRef = request.ResultRef
			writeWorkerTestJSON(t, w, protocol.RunJobResultResponse{Accepted: true, Job: assignment, ServerTime: workerTestTime()})
		default:
			t.Fatalf("unexpected path: %s", r.URL.Path)
		}
	}))
	defer server.Close()

	client, err := api.NewPlatformClient(server.URL)
	if err != nil {
		t.Fatalf("platform client: %v", err)
	}
	cfg := workerTestConfig(t)
	cfg.PlatformURL = server.URL
	worker, err := NewWorker(cfg, client, WithProcessSupervisor(staticSupervisor{stdout: "integration ok\n"}))
	if err != nil {
		t.Fatalf("new worker: %v", err)
	}
	if err := worker.Register(context.Background()); err != nil {
		t.Fatalf("register: %v", err)
	}
	if err := worker.HeartbeatOnce(context.Background()); err != nil {
		t.Fatalf("heartbeat: %v", err)
	}
	if handled, err := worker.ClaimAndRunOnce(context.Background()); err != nil || !handled {
		t.Fatalf("claim/run handled=%v err=%v", handled, err)
	}
	expected := []string{
		"/api/v1/run/control/hello",
		"/api/v1/run/control/heartbeat",
		"/api/v1/run/jobs/claim",
		"/api/v1/run/jobs/ack",
		"/api/v1/run/jobs/progress",
		"/api/v1/run/jobs/cancel",
		"/api/v1/run/jobs/result",
	}
	if !reflect.DeepEqual(seen, expected) {
		t.Fatalf("unexpected platform flow: %+v", seen)
	}
}

type fakeWorkerClient struct {
	helloRequests        []protocol.RunHelloRequest
	heartbeatRequests    []protocol.RunHeartbeatRequest
	lifecycleReports     []protocol.RunLifecycleReportRequest
	claimRequests        []protocol.RunJobClaimRequest
	ackRequests          []protocol.RunJobAckRequest
	progressRequests     []protocol.RunJobProgressRequest
	resultRequests       []protocol.RunJobResultRequest
	cancelPollRequests   []protocol.RunJobCancelPollRequest
	reconcileRequests    []protocol.RunJobReconcileRequest
	logProgressRequests  []protocol.RunLogStreamProgressRequest
	metricRequests       []protocol.MetricBatchIngestRequest
	claimJob             protocol.RunJobAssignment
	claimErr             error
	cancelResponse       protocol.RunJobCancelPollResponse
	reconcileResponse    protocol.RunJobReconcileResponse
	progressResponse     protocol.RunLogStreamProgressResponse
	buildInput           protocol.DistributionBuildInputResponse
	dependencyInput      protocol.DependencyExecutionInputResponse
	sourceRCONInput      protocol.SourceRCONExecutionInputResponse
	sourceRCONRequests   []protocol.SourceRCONExecutionInputRequest
	sourceRCONInputErr   error
	protectedInput       protocol.ProtectedRequestExecutionInputResponse
	protectedRequests    []protocol.ProtectedRequestExecutionInputRequest
	protectedInputErr    error
	updateInput          protocol.RunUpdateInputResponse
	updatePayload        []byte
	updateChunkOffsets   []int64
	updateHealthReports  []protocol.RunUpdateHealthRequest
	artifactPayload      []byte
	artifactOpenRequests []protocol.ArtifactTransferOpenRequest
	artifactTransfer     string
	heartbeatErr         error
	lifecycleReportErr   error
	logStreamProgress    uint64
	resultErr            error
	metricErr            error
}

type recordingSelfUpdateActivator struct {
	manifestPath string
	err          error
}

func (activator *recordingSelfUpdateActivator) Activate(path string) error {
	activator.manifestPath = path
	return activator.err
}

func newFakeWorkerClient() *fakeWorkerClient {
	return &fakeWorkerClient{
		cancelResponse:    protocol.RunJobCancelPollResponse{Accepted: true, RunEndpointID: "run-test", ServerTime: workerTestTime()},
		reconcileResponse: protocol.RunJobReconcileResponse{Accepted: true, RunEndpointID: "run-test", ServerTime: workerTestTime()},
		progressResponse:  protocol.RunLogStreamProgressResponse{Accepted: true, RunEndpointID: "run-test", ServerInstanceID: "server-worker", ServerTime: workerTestTime()},
	}
}

type rotatingSessionClient struct {
	*fakeWorkerClient
	mu         sync.Mutex
	helloCount int
}

func newRotatingSessionClient() *rotatingSessionClient {
	return &rotatingSessionClient{fakeWorkerClient: newFakeWorkerClient()}
}

func (client *rotatingSessionClient) Hello(_ context.Context, request protocol.RunHelloRequest) (protocol.RunHelloResponse, error) {
	client.mu.Lock()
	defer client.mu.Unlock()
	client.helloCount++
	client.helloRequests = append(client.helloRequests, request)
	return protocol.RunHelloResponse{Accepted: true, RunEndpointID: request.RunEndpointID, SessionToken: fmt.Sprintf("session-token-%d", client.helloCount), ServerTime: workerTestTime(), HeartbeatIntervalSeconds: 15}, nil
}

func (client *rotatingSessionClient) HelloCount() int {
	client.mu.Lock()
	defer client.mu.Unlock()
	return client.helloCount
}

type recordingDurableLogClient struct {
	batch protocol.LogBatchIngestRequest
	err   error
}

type recordingLiveLogClient struct {
	received chan protocol.LogBatchIngestRequest
}

func (client *recordingLiveLogClient) RelayLiveLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
	client.received <- batch
	return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
}

func (client *recordingDurableLogClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
	client.batch = batch
	if client.err != nil {
		return protocol.LogBatchIngestResponse{}, client.err
	}
	return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
}

type recordingDurableArtifactClient struct {
	chunk protocol.ArtifactChunkUploadRequest
	err   error
}

func (client *recordingDurableArtifactClient) UploadArtifactChunk(_ context.Context, chunk protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) {
	client.chunk = chunk
	if client.err != nil {
		return protocol.ArtifactChunkUploadResponse{}, client.err
	}
	return protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: chunk.TransferID, ArtifactID: chunk.ArtifactID, ChunkIndex: chunk.ChunkIndex}, nil
}

func (client *fakeWorkerClient) Hello(_ context.Context, request protocol.RunHelloRequest) (protocol.RunHelloResponse, error) {
	client.helloRequests = append(client.helloRequests, request)
	return protocol.RunHelloResponse{Accepted: true, RunEndpointID: request.RunEndpointID, SessionToken: "session-token", ServerTime: workerTestTime(), HeartbeatIntervalSeconds: 15}, nil
}

func (client *fakeWorkerClient) Heartbeat(_ context.Context, request protocol.RunHeartbeatRequest) (protocol.RunHeartbeatResponse, error) {
	client.heartbeatRequests = append(client.heartbeatRequests, request)
	if client.heartbeatErr != nil {
		err := client.heartbeatErr
		client.heartbeatErr = nil
		return protocol.RunHeartbeatResponse{}, err
	}
	return protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) StreamControlEvents(ctx context.Context, request protocol.RunControlStreamRequest, handle func(protocol.RunControlEvent) error) error {
	if handle != nil {
		if err := handle(protocol.RunControlEvent{RunEndpointID: request.RunEndpointID, Type: protocol.RunControlEventTypeReady, Sequence: request.LastEventSeq, ServerTime: workerTestTime()}); err != nil {
			return err
		}
	}
	<-ctx.Done()
	return ctx.Err()
}

func (client *fakeWorkerClient) ReportLifecycle(_ context.Context, request protocol.RunLifecycleReportRequest) (protocol.RunLifecycleReportResponse, error) {
	client.lifecycleReports = append(client.lifecycleReports, request)
	if client.lifecycleReportErr != nil {
		return protocol.RunLifecycleReportResponse{}, client.lifecycleReportErr
	}
	return protocol.RunLifecycleReportResponse{Accepted: true, RunEndpointID: request.RunEndpointID, ServerInstanceID: request.ServerInstanceID, ProjectedState: request.ExecutionResult.ProcessState, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) ClaimJob(_ context.Context, request protocol.RunJobClaimRequest) (protocol.RunJobClaimResponse, error) {
	client.claimRequests = append(client.claimRequests, request)
	if client.claimErr != nil {
		return protocol.RunJobClaimResponse{}, client.claimErr
	}
	if client.claimJob.JobID == "" {
		return protocol.RunJobClaimResponse{Accepted: true, RunEndpointID: request.RunEndpointID, HasJob: false, NextPollSeconds: 2, ServerTime: workerTestTime()}, nil
	}
	job := client.claimJob
	return protocol.RunJobClaimResponse{Accepted: true, RunEndpointID: request.RunEndpointID, HasJob: true, Job: &job, NextPollSeconds: 2, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) AckJob(_ context.Context, request protocol.RunJobAckRequest) (protocol.RunJobAckResponse, error) {
	client.ackRequests = append(client.ackRequests, request)
	job := client.claimJob
	job.State = "running"
	return protocol.RunJobAckResponse{Accepted: true, Job: job, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) UpdateJobProgress(_ context.Context, request protocol.RunJobProgressRequest) (protocol.RunJobProgressResponse, error) {
	client.progressRequests = append(client.progressRequests, request)
	job := client.claimJob
	job.Progress = request.Progress
	job.ProgressSequence = request.Sequence
	return protocol.RunJobProgressResponse{Accepted: true, Job: job, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) CompleteJob(_ context.Context, request protocol.RunJobResultRequest) (protocol.RunJobResultResponse, error) {
	client.resultRequests = append(client.resultRequests, request)
	if client.resultErr != nil {
		return protocol.RunJobResultResponse{}, client.resultErr
	}
	job := client.claimJob
	job.State = request.State
	job.Progress = request.Progress
	job.ResultRef = request.ResultRef
	return protocol.RunJobResultResponse{Accepted: true, Job: job, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) GetDistributionBuildInput(_ context.Context, request protocol.DistributionBuildInputRequest) (protocol.DistributionBuildInputResponse, error) {
	if client.buildInput.JobID == "" {
		return protocol.DistributionBuildInputResponse{}, fmt.Errorf("distribution build input is not configured")
	}
	if request.JobID != client.buildInput.JobID {
		return protocol.DistributionBuildInputResponse{}, fmt.Errorf("unexpected build job")
	}
	return client.buildInput, nil
}

func (client *fakeWorkerClient) GetDependencyExecutionInput(_ context.Context, request protocol.DependencyExecutionInputRequest) (protocol.DependencyExecutionInputResponse, error) {
	if client.dependencyInput.JobID == "" || request.JobID != client.dependencyInput.JobID {
		return protocol.DependencyExecutionInputResponse{}, fmt.Errorf("dependency input is not configured")
	}
	return client.dependencyInput, nil
}

func (client *fakeWorkerClient) GetSourceRCONExecutionInput(_ context.Context, request protocol.SourceRCONExecutionInputRequest) (protocol.SourceRCONExecutionInputResponse, error) {
	client.sourceRCONRequests = append(client.sourceRCONRequests, request)
	if client.sourceRCONInputErr != nil {
		return protocol.SourceRCONExecutionInputResponse{}, client.sourceRCONInputErr
	}
	if client.sourceRCONInput.JobID == "" || request.JobID != client.sourceRCONInput.JobID {
		return protocol.SourceRCONExecutionInputResponse{}, fmt.Errorf("Source RCON input is not configured")
	}
	return client.sourceRCONInput, nil
}

func (client *fakeWorkerClient) GetProtectedRequestExecutionInput(_ context.Context, request protocol.ProtectedRequestExecutionInputRequest) (protocol.ProtectedRequestExecutionInputResponse, error) {
	client.protectedRequests = append(client.protectedRequests, request)
	if client.protectedInputErr != nil {
		return protocol.ProtectedRequestExecutionInputResponse{}, client.protectedInputErr
	}
	if client.protectedInput.JobID == "" || request.JobID != client.protectedInput.JobID {
		return protocol.ProtectedRequestExecutionInputResponse{}, fmt.Errorf("protected request input is not configured")
	}
	return client.protectedInput, nil
}

func (client *fakeWorkerClient) GetRunUpdateInput(_ context.Context, request protocol.RunUpdateInputRequest) (protocol.RunUpdateInputResponse, error) {
	if client.updateInput.JobID == "" || request.JobID != client.updateInput.JobID {
		return protocol.RunUpdateInputResponse{}, fmt.Errorf("Run update input is not configured")
	}
	return client.updateInput, nil
}

func (client *fakeWorkerClient) ReadRunUpdateChunk(_ context.Context, request protocol.RunUpdateChunkRequest) (protocol.RunUpdateChunkResponse, error) {
	client.updateChunkOffsets = append(client.updateChunkOffsets, request.Offset)
	if client.updateInput.JobID == "" || request.JobID != client.updateInput.JobID || request.Offset < 0 || request.Offset >= int64(len(client.updatePayload)) {
		return protocol.RunUpdateChunkResponse{}, fmt.Errorf("Run update chunk is not configured")
	}
	end := request.Offset + int64(request.Length)
	if end > int64(len(client.updatePayload)) {
		end = int64(len(client.updatePayload))
	}
	return protocol.RunUpdateChunkResponse{JobID: request.JobID, ArtifactID: client.updateInput.ArtifactID, Offset: request.Offset, TotalBytes: int64(len(client.updatePayload)), Checksum: client.updateInput.Checksum, Payload: append([]byte(nil), client.updatePayload[request.Offset:end]...), Complete: end == int64(len(client.updatePayload))}, nil
}

func (client *fakeWorkerClient) ReportRunUpdateHealth(_ context.Context, request protocol.RunUpdateHealthRequest) (protocol.RunUpdateHealthResponse, error) {
	client.updateHealthReports = append(client.updateHealthReports, request)
	return protocol.RunUpdateHealthResponse{Accepted: true, JobID: request.JobID, Phase: request.Outcome, ServerTime: workerTestTime()}, nil
}

func (client *fakeWorkerClient) OpenArtifactTransfer(_ context.Context, request protocol.ArtifactTransferOpenRequest) (protocol.ArtifactTransferOpenResponse, error) {
	if client.buildInput.JobID == "" {
		return protocol.ArtifactTransferOpenResponse{}, fmt.Errorf("artifact transfer is not configured")
	}
	client.artifactOpenRequests = append(client.artifactOpenRequests, request)
	client.artifactTransfer = "transfer-build"
	return protocol.ArtifactTransferOpenResponse{Accepted: true, TransferID: client.artifactTransfer, Artifact: protocol.ArtifactMetadata{ID: request.ArtifactID, OwnerKind: request.OwnerKind, OwnerID: request.OwnerID, State: "uploading"}, ChunkSizeBytes: request.ChunkSizeBytes}, nil
}

func (client *fakeWorkerClient) UploadArtifactChunk(_ context.Context, request protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) {
	if request.TransferID != client.artifactTransfer {
		return protocol.ArtifactChunkUploadResponse{}, fmt.Errorf("unexpected artifact transfer")
	}
	client.artifactPayload = append(client.artifactPayload, request.Payload...)
	return protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: request.TransferID, ArtifactID: request.ArtifactID, ChunkIndex: request.ChunkIndex}, nil
}

func (client *fakeWorkerClient) CompleteArtifactTransfer(_ context.Context, request protocol.ArtifactTransferCompleteRequest) (protocol.ArtifactTransferCompleteResponse, error) {
	if request.TransferID != client.artifactTransfer || request.SizeBytes != int64(len(client.artifactPayload)) {
		return protocol.ArtifactTransferCompleteResponse{}, fmt.Errorf("unexpected artifact completion")
	}
	return protocol.ArtifactTransferCompleteResponse{Accepted: true, TransferID: request.TransferID, Artifact: protocol.ArtifactMetadata{ID: request.ArtifactID, OwnerKind: "job", OwnerID: client.buildInput.JobID, SizeBytes: request.SizeBytes, Checksum: request.Checksum, State: "available"}, Completed: true}, nil
}

func (client *fakeWorkerClient) PollJobCancel(_ context.Context, request protocol.RunJobCancelPollRequest) (protocol.RunJobCancelPollResponse, error) {
	client.cancelPollRequests = append(client.cancelPollRequests, request)
	return client.cancelResponse, nil
}

func (client *fakeWorkerClient) ReconcileJobs(_ context.Context, request protocol.RunJobReconcileRequest) (protocol.RunJobReconcileResponse, error) {
	client.reconcileRequests = append(client.reconcileRequests, request)
	return client.reconcileResponse, nil
}

func (client *fakeWorkerClient) GetRunLogStreamProgress(_ context.Context, request protocol.RunLogStreamProgressRequest) (protocol.RunLogStreamProgressResponse, error) {
	client.logProgressRequests = append(client.logProgressRequests, request)
	response := client.progressResponse
	response.RunEndpointID = request.RunEndpointID
	response.ServerInstanceID = request.ServerInstanceID
	response.LogStreamID = request.LogStreamID
	if response.LatestSeq == 0 {
		response.LatestSeq = client.logStreamProgress
	}
	return response, nil
}

func (client *fakeWorkerClient) IngestMetricBatch(_ context.Context, request protocol.MetricBatchIngestRequest) (protocol.MetricBatchIngestResponse, error) {
	client.metricRequests = append(client.metricRequests, request)
	if client.metricErr != nil {
		return protocol.MetricBatchIngestResponse{}, client.metricErr
	}
	return protocol.MetricBatchIngestResponse{Accepted: true, AcceptedCount: len(request.Samples), ServerTime: workerTestTime()}, nil
}

type staticSupervisor struct {
	stdout string
	stderr string
	err    error
}

func (supervisor staticSupervisor) Run(context.Context, ProcessCommand) (ProcessResult, error) {
	return ProcessResult{ExitCode: 0, Stdout: supervisor.stdout, Stderr: supervisor.stderr}, supervisor.err
}

func workerTestConfig(t *testing.T) config.Config {
	t.Helper()
	return config.Config{
		Mode:              "worker",
		PlatformURL:       "http://platform.test",
		RunEndpointID:     "run-test",
		DisplayName:       "Run Test",
		Version:           "0.1.0-test",
		RegistrationToken: "registration-token",
		WorkspaceRoot:     t.TempDir(),
		SpoolRoot:         t.TempDir(),
		MaxJobs:           2,
		HeartbeatInterval: time.Second,
		PollInterval:      time.Second,
		RetryBackoff:      time.Millisecond,
	}
}

func workerJobAssignment(capability string) protocol.RunJobAssignment {
	job := lifecycleAssignment(capability)
	job.JobID = "job-worker"
	if capability == protocol.RunCapabilityProcessStop {
		job.JobID = "job-worker-stop"
	}
	job.RunEndpointID = "run-test"
	job.ServerInstanceID = "server-worker"
	return job
}

func workerTestTime() time.Time {
	return time.Date(2026, 7, 6, 12, 0, 0, 0, time.UTC)
}

func containsText(value string, needle string) bool {
	return strings.Contains(value, needle)
}

func decodeWorkerTestJSON(t *testing.T, r *http.Request, target any) {
	t.Helper()
	if r.Method != http.MethodPost {
		t.Fatalf("expected POST, got %s", r.Method)
	}
	if err := json.NewDecoder(r.Body).Decode(target); err != nil {
		t.Fatalf("decode request: %v", err)
	}
}

func writeWorkerTestJSON(t *testing.T, w http.ResponseWriter, value any) {
	t.Helper()
	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(value); err != nil {
		t.Fatalf("encode response: %v", err)
	}
}