package runtime

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"time"

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

type WorkerClient interface {
	Hello(context.Context, protocol.RunHelloRequest) (protocol.RunHelloResponse, error)
	Heartbeat(context.Context, protocol.RunHeartbeatRequest) (protocol.RunHeartbeatResponse, error)
	ReportLifecycle(context.Context, protocol.RunLifecycleReportRequest) (protocol.RunLifecycleReportResponse, error)
	GetRunLogStreamProgress(context.Context, protocol.RunLogStreamProgressRequest) (protocol.RunLogStreamProgressResponse, error)
	ClaimJob(context.Context, protocol.RunJobClaimRequest) (protocol.RunJobClaimResponse, error)
	AckJob(context.Context, protocol.RunJobAckRequest) (protocol.RunJobAckResponse, error)
	UpdateJobProgress(context.Context, protocol.RunJobProgressRequest) (protocol.RunJobProgressResponse, error)
	CompleteJob(context.Context, protocol.RunJobResultRequest) (protocol.RunJobResultResponse, error)
	GetDistributionBuildInput(context.Context, protocol.DistributionBuildInputRequest) (protocol.DistributionBuildInputResponse, error)
	GetDependencyExecutionInput(context.Context, protocol.DependencyExecutionInputRequest) (protocol.DependencyExecutionInputResponse, error)
	GetSourceRCONExecutionInput(context.Context, protocol.SourceRCONExecutionInputRequest) (protocol.SourceRCONExecutionInputResponse, error)
	GetProtectedRequestExecutionInput(context.Context, protocol.ProtectedRequestExecutionInputRequest) (protocol.ProtectedRequestExecutionInputResponse, error)
	GetRunUpdateInput(context.Context, protocol.RunUpdateInputRequest) (protocol.RunUpdateInputResponse, error)
	ReadRunUpdateChunk(context.Context, protocol.RunUpdateChunkRequest) (protocol.RunUpdateChunkResponse, error)
	ReportRunUpdateHealth(context.Context, protocol.RunUpdateHealthRequest) (protocol.RunUpdateHealthResponse, error)
	IngestMetricBatch(context.Context, protocol.MetricBatchIngestRequest) (protocol.MetricBatchIngestResponse, error)
	OpenArtifactTransfer(context.Context, protocol.ArtifactTransferOpenRequest) (protocol.ArtifactTransferOpenResponse, error)
	UploadArtifactChunk(context.Context, protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error)
	CompleteArtifactTransfer(context.Context, protocol.ArtifactTransferCompleteRequest) (protocol.ArtifactTransferCompleteResponse, error)
	PollJobCancel(context.Context, protocol.RunJobCancelPollRequest) (protocol.RunJobCancelPollResponse, error)
	ReconcileJobs(context.Context, protocol.RunJobReconcileRequest) (protocol.RunJobReconcileResponse, error)
}

const (
	jobActivePollInterval       = 10 * time.Second
	jobClaimWaitSeconds         = 25
	durableUploaderFlushTimeout = 5 * time.Second
	fileArtifactChunkSize       = 1024 * 1024
	maxFileArtifactBytes        = int64(512 * 1024 * 1024)
)

type Worker struct {
	cfg                  config.Config
	client               WorkerClient
	executor             LifecycleExecutor
	state                WorkerState
	journal              *JobJournal
	stateMu              sync.RWMutex
	sessionRefreshMu     sync.Mutex
	sequenceMu           sync.Mutex
	restartMu            sync.Mutex
	restartRequested     bool
	observationMu        sync.Mutex
	reportedObservations map[string]uint64
	metricCollector      MetricCollector
}

type WorkerState struct {
	RunEndpointID    string
	SessionToken     string
	SessionExpiresAt time.Time
	Capabilities     []string
	Capacity         protocol.RunCapacityReport
	LastHeartbeat    time.Time
	Sequence         uint64
}

func NewWorker(cfg config.Config, client WorkerClient, options ...LifecycleExecutorOption) (*Worker, error) {
	if client == nil {
		return nil, fmt.Errorf("worker client is required")
	}
	if cfg.RunEndpointID == "" {
		cfg.RunEndpointID = config.DefaultEndpointID
	}
	if cfg.DisplayName == "" {
		cfg.DisplayName = config.DefaultDisplayName
	}
	if cfg.Version == "" {
		cfg.Version = config.DefaultVersion
	}
	if cfg.MaxJobs <= 0 {
		cfg.MaxJobs = 1
	}
	executorOptions := append([]LifecycleExecutorOption{
		WithLifecycleWorkspaceRoot(cfg.WorkspaceRoot),
		WithManagedProcessStateRoot(managedProcessStateRoot(cfg)),
		WithManagedProcessOutputRoot(cfg.WorkspaceRoot),
		WithLocalStartupDiagnostics(cfg.LocalStartupDiagnostics),
	}, options...)
	if err := migrateManagedProcessState(cfg, managedProcessStateRoot(cfg)); err != nil {
		return nil, err
	}
	journal, err := NewPersistentJobJournal(cfg.WorkspaceRoot)
	if err != nil {
		return nil, err
	}
	worker := &Worker{
		cfg:      cfg,
		client:   client,
		executor: NewLifecycleExecutor(executorOptions...),
		state: WorkerState{
			RunEndpointID: cfg.RunEndpointID,
			Capabilities:  SupportedRunCapabilitiesForComponent(cfg.ComponentKind),
			Capacity:      protocol.RunCapacityReport{MaxJobs: cfg.MaxJobs},
		},
		journal:              journal,
		reportedObservations: map[string]uint64{},
		metricCollector:      defaultMetricCollector{},
	}
	if worker.executor.metricCollector != nil {
		worker.metricCollector = worker.executor.metricCollector
	}
	return worker, nil
}

func migrateLegacyManagedProcessState(cfg config.Config, stateRoot string) error {
	workspaceRoot := cfg.WorkspaceRoot
	if strings.TrimSpace(workspaceRoot) == "" {
		workspaceRoot = filepath.Join(".", ".run-workspace")
	}
	legacyPath := filepath.Join(workspaceRoot, "state", "processes.json")
	newPath := filepath.Join(stateRoot, "state", "processes.json")
	if filepath.Clean(legacyPath) == filepath.Clean(newPath) {
		return nil
	}
	if _, err := os.Stat(newPath); err == nil {
		return nil
	} else if !os.IsNotExist(err) {
		return err
	}
	body, err := os.ReadFile(legacyPath)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("read legacy managed process state: %w", err)
	}
	var legacy processJournal
	if err := json.Unmarshal(body, &legacy); err != nil {
		return fmt.Errorf("decode legacy managed process state: %w", err)
	}
	filtered := processJournal{Version: managedProcessJournalVersion, Items: map[string]ProcessIdentity{}, Retired: map[string]ProcessIdentity{}}
	for key, item := range legacy.Items {
		if managedProcessBelongsToRun(item, cfg) {
			filtered.Items[key] = item
		}
	}
	for key, item := range legacy.Retired {
		if managedProcessBelongsToRun(item, cfg) {
			filtered.Retired[key] = item
		}
	}
	if len(filtered.Items) == 0 && len(filtered.Retired) == 0 {
		return nil
	}
	if err := ensureDirectory(filepath.Dir(newPath)); err != nil {
		return fmt.Errorf("create isolated managed process state: %w", err)
	}
	encoded, err := json.Marshal(filtered)
	if err != nil {
		return fmt.Errorf("encode isolated managed process state: %w", err)
	}
	if err := os.WriteFile(newPath, encoded, 0o600); err != nil {
		return fmt.Errorf("write isolated managed process state: %w", err)
	}
	log.Printf("RUN phase=process.managed status=legacy_state_migrated items=%d retired=%d", len(filtered.Items), len(filtered.Retired))
	return nil
}

// migrateManagedProcessState also imports matching journals from prior Run
// state namespaces. A generated package can change its component metadata
// while retaining the same endpoint/server/profile; that must not create a
// second autonomous process for the same server during a Run update.
func migrateManagedProcessState(cfg config.Config, stateRoot string) error {
	if err := migrateLegacyManagedProcessState(cfg, stateRoot); err != nil {
		return err
	}
	workspaceRoot := cfg.WorkspaceRoot
	if strings.TrimSpace(workspaceRoot) == "" {
		workspaceRoot = filepath.Join(".", ".run-workspace")
	}
	targetPath := filepath.Join(stateRoot, "state", "processes.json")
	target := processJournal{Version: managedProcessJournalVersion, Items: map[string]ProcessIdentity{}, Retired: map[string]ProcessIdentity{}}
	if body, err := os.ReadFile(targetPath); err == nil {
		if err := json.Unmarshal(body, &target); err != nil {
			return fmt.Errorf("decode managed process state: %w", err)
		}
		if target.Items == nil {
			target.Items = map[string]ProcessIdentity{}
		}
		if target.Retired == nil {
			target.Retired = map[string]ProcessIdentity{}
		}
	} else if !os.IsNotExist(err) {
		return fmt.Errorf("read managed process state: %w", err)
	}
	changed := false
	err := filepath.WalkDir(workspaceRoot, func(path string, entry os.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if entry.IsDir() || entry.Name() != "processes.json" {
			return nil
		}
		if filepath.Clean(path) == filepath.Clean(targetPath) {
			return nil
		}
		body, err := os.ReadFile(path)
		if err != nil {
			return err
		}
		var source processJournal
		if err := json.Unmarshal(body, &source); err != nil {
			return fmt.Errorf("decode managed process state candidate: %w", err)
		}
		for key, item := range source.Items {
			if !managedProcessBelongsToRun(item, cfg) {
				continue
			}
			existing, exists := target.Items[key]
			if !exists || (!processAlivePID(existing.PID) && processAlivePID(item.PID)) {
				target.Items[key] = item
				changed = true
			}
		}
		for key, item := range source.Retired {
			if managedProcessBelongsToRun(item, cfg) {
				if _, exists := target.Retired[key]; !exists {
					target.Retired[key] = item
					changed = true
				}
			}
		}
		return nil
	})
	if err != nil {
		return fmt.Errorf("scan managed process state: %w", err)
	}
	if !changed {
		return nil
	}
	if err := ensureDirectory(filepath.Dir(targetPath)); err != nil {
		return fmt.Errorf("create managed process state directory: %w", err)
	}
	body, err := json.Marshal(target)
	if err != nil {
		return fmt.Errorf("encode managed process state: %w", err)
	}
	temporary := targetPath + ".tmp"
	if err := os.WriteFile(temporary, body, 0o600); err != nil {
		return fmt.Errorf("write managed process state: %w", err)
	}
	if err := os.Rename(temporary, targetPath); err != nil {
		_ = os.Remove(temporary)
		return fmt.Errorf("replace managed process state: %w", err)
	}
	log.Printf("RUN phase=process.managed status=state_namespaces_merged items=%d retired=%d", len(target.Items), len(target.Retired))
	return nil
}

func managedProcessBelongsToRun(item ProcessIdentity, cfg config.Config) bool {
	if item.RunEndpointID == "" || item.ServerInstanceID == "" || item.RunEndpointID != cfg.RunEndpointID || item.ServerInstanceID != cfg.ServerInstanceID {
		return false
	}
	return cfg.ComponentKey == "" || item.ProfileKey == "" || item.ProfileKey == cfg.ComponentKey
}

func managedProcessStateRoot(cfg config.Config) string {
	workspaceRoot := cfg.WorkspaceRoot
	if strings.TrimSpace(workspaceRoot) == "" {
		workspaceRoot = filepath.Join(".", ".run-workspace")
	}
	identity := strings.Join([]string{
		"run-process-state-v1",
		cfg.RunEndpointID,
		cfg.ServerInstanceID,
		cfg.PluginID,
		cfg.ComponentKind,
		cfg.ComponentKey,
	}, "\x00")
	digest := sha256.Sum256([]byte(identity))
	return filepath.Join(workspaceRoot, "run-services", hex.EncodeToString(digest[:]))
}

func (worker *Worker) Register(ctx context.Context) error {
	worker.sessionRefreshMu.Lock()
	defer worker.sessionRefreshMu.Unlock()
	return worker.registerUnlocked(ctx)
}

func (worker *Worker) registerUnlocked(ctx context.Context) error {
	state := worker.State()
	log.Printf("RUN phase=register status=starting endpoint=%s version=%s server=%s plugin=%s component=%s componentKey=%s capabilities=%d maxJobs=%d", worker.cfg.RunEndpointID, worker.cfg.Version, worker.cfg.ServerInstanceID, worker.cfg.PluginID, worker.cfg.ComponentKind, safeOptional(worker.cfg.ComponentKey), len(state.Capabilities), worker.cfg.MaxJobs)
	response, err := worker.client.Hello(ctx, protocol.RunHelloRequest{
		RegistrationToken: worker.cfg.RegistrationToken,
		RunEndpointID:     worker.cfg.RunEndpointID,
		ServerInstanceID:  worker.cfg.ServerInstanceID,
		PluginID:          worker.cfg.PluginID,
		ComponentKind:     worker.cfg.ComponentKind,
		ComponentKey:      worker.cfg.ComponentKey,
		KeyGeneration:     worker.cfg.KeyGeneration,
		DisplayName:       worker.cfg.DisplayName,
		Version:           worker.cfg.Version,
		Status:            "online",
		Platform:          runtime.GOOS,
		Architecture:      runtime.GOARCH,
		CapabilityReport: protocol.RunCapabilityReport{
			Capabilities: state.Capabilities,
			Fingerprint:  capabilityFingerprint(state.Capabilities),
		},
		Capacity: worker.capacityReportFor(state),
	})
	if err != nil {
		log.Printf("RUN phase=register status=failed endpoint=%s error=%s", worker.cfg.RunEndpointID, RedactText(err.Error()))
		return err
	}
	if !response.Accepted || response.SessionToken == "" {
		log.Printf("RUN phase=register status=rejected endpoint=%s accepted=%t sessionTokenPresent=%t", worker.cfg.RunEndpointID, response.Accepted, response.SessionToken != "")
		return fmt.Errorf("run hello was not accepted")
	}
	worker.stateMu.Lock()
	worker.state.SessionToken = response.SessionToken
	worker.state.SessionExpiresAt = response.SessionExpiresAt
	state = worker.state
	state.Capabilities = append([]string(nil), state.Capabilities...)
	worker.stateMu.Unlock()
	if sink, ok := worker.executor.logSink.(*SpoolLogSink); ok {
		sink.mu.Lock()
		sink.RunEndpointID = state.RunEndpointID
		sink.SessionToken = state.SessionToken
		sink.Progress = func(ctx context.Context, serverInstanceID string, streamID string) (uint64, error) {
			current, err := worker.registeredState()
			if err != nil {
				return 0, err
			}
			response, err := worker.client.GetRunLogStreamProgress(ctx, protocol.RunLogStreamProgressRequest{RunEndpointID: current.RunEndpointID, SessionToken: current.SessionToken, ServerInstanceID: serverInstanceID, LogStreamID: streamID})
			if err != nil {
				return 0, err
			}
			if !response.Accepted || response.LogStreamID != streamID {
				return 0, fmt.Errorf("run log stream progress was not accepted")
			}
			return response.LatestSeq, nil
		}
		sink.mu.Unlock()
	}
	if hook, ok := worker.executor.artifactHook.(*QueueArtifactHook); ok {
		hook.RunEndpointID = state.RunEndpointID
		hook.SessionToken = state.SessionToken
	}
	worker.executor.ResumeManagedProcessLogs(ctx)
	log.Printf("RUN phase=register status=accepted endpoint=%s sessionExpiresAt=%s heartbeatSeconds=%d", state.RunEndpointID, response.SessionExpiresAt.Format(time.RFC3339), response.HeartbeatIntervalSeconds)
	return nil
}

func (worker *Worker) registeredState() (WorkerState, error) {
	state := worker.State()
	if state.SessionToken == "" {
		return state, fmt.Errorf("worker is not registered")
	}
	return state, nil
}

func (worker *Worker) HeartbeatOnce(ctx context.Context) error {
	state, err := worker.registeredState()
	if err != nil {
		return err
	}
	if !state.SessionExpiresAt.IsZero() && !time.Now().UTC().Add(time.Minute).Before(state.SessionExpiresAt) {
		log.Printf("RUN phase=heartbeat status=session_expiring endpoint=%s sessionExpiresAt=%s", state.RunEndpointID, state.SessionExpiresAt.Format(time.RFC3339))
		return worker.reregisterAndReconcile(ctx, "heartbeat_session_expiring", state.SessionToken)
	}
	log.Printf("RUN phase=heartbeat status=starting endpoint=%s activeJobs=%d", state.RunEndpointID, worker.journal.ActiveCount())
	response, err := worker.client.Heartbeat(ctx, protocol.RunHeartbeatRequest{
		RunEndpointID:         state.RunEndpointID,
		SessionToken:          state.SessionToken,
		Version:               worker.cfg.Version,
		Status:                "online",
		CapabilityFingerprint: capabilityFingerprint(state.Capabilities),
		Capacity:              worker.capacityReportFor(state),
	})
	if err != nil {
		if sessionInvalidError(err) {
			log.Printf("RUN phase=heartbeat status=session_invalid endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error()))
			return worker.reregisterAndReconcile(ctx, "heartbeat_session_invalid", state.SessionToken)
		}
		log.Printf("RUN phase=heartbeat status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error()))
		return err
	}
	if !response.Accepted {
		log.Printf("RUN phase=heartbeat status=rejected endpoint=%s", state.RunEndpointID)
		return fmt.Errorf("heartbeat was not accepted")
	}
	worker.stateMu.Lock()
	if worker.state.SessionToken == state.SessionToken {
		worker.state.LastHeartbeat = response.ServerTime
	}
	worker.stateMu.Unlock()
	log.Printf("RUN phase=heartbeat status=accepted endpoint=%s serverTime=%s nextSeconds=%d", state.RunEndpointID, response.ServerTime.Format(time.RFC3339), response.NextHeartbeatSeconds)
	return nil
}

func (worker *Worker) reregisterAndReconcile(ctx context.Context, reason string, observedToken string) error {
	worker.sessionRefreshMu.Lock()
	defer worker.sessionRefreshMu.Unlock()
	state := worker.State()
	if observedToken != "" && state.SessionToken != "" && state.SessionToken != observedToken {
		log.Printf("RUN phase=register status=already_refreshed reason=%s endpoint=%s", safeOptional(reason), state.RunEndpointID)
		return worker.ReconcileOnce(ctx)
	}
	log.Printf("RUN phase=register status=refreshing reason=%s endpoint=%s", safeOptional(reason), state.RunEndpointID)
	if err := worker.registerUnlocked(ctx); err != nil {
		return err
	}
	return worker.ReconcileOnce(ctx)
}

func (worker *Worker) ClaimAndRunOnce(ctx context.Context) (bool, error) {
	state, err := worker.registeredState()
	if err != nil {
		return false, err
	}
	log.Printf("RUN phase=claim status=polling endpoint=%s activeJobs=%d", state.RunEndpointID, worker.journal.ActiveCount())
	claim, err := worker.client.ClaimJob(ctx, protocol.RunJobClaimRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		Capabilities:  state.Capabilities,
		Capacity:      worker.capacityReportFor(state),
		WaitSeconds:   jobClaimWaitSeconds,
	})
	if err != nil {
		if sessionInvalidError(err) {
			log.Printf("RUN phase=claim status=session_invalid endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error()))
			return false, worker.reregisterAndReconcile(ctx, "claim_session_invalid", state.SessionToken)
		}
		log.Printf("RUN phase=claim status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error()))
		return false, err
	}
	if !claim.Accepted || !claim.HasJob || claim.Job == nil {
		log.Printf("RUN phase=claim status=idle endpoint=%s accepted=%t hasJob=%t", state.RunEndpointID, claim.Accepted, claim.HasJob)
		return false, nil
	}
	assignment := *claim.Job
	log.Printf("RUN phase=claim status=assigned job=%s capability=%s target=%s attempt=%d server=%s workspaceScope=%s", assignment.JobID, assignment.Capability, safeOptional(assignment.TargetKey), assignment.Attempt, assignment.ServerInstanceID, safeOptional(assignment.ExecutionInput.WorkspaceScope))
	return true, worker.runAssignment(ctx, assignment)
}

func sessionInvalidError(err error) bool {
	var sessionError interface{ SessionInvalid() bool }
	return errors.As(err, &sessionError) && sessionError.SessionInvalid()
}

func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.RunJobAssignment) error {
	state, err := worker.registeredState()
	if err != nil {
		return err
	}
	if assignment.RunEndpointID != state.RunEndpointID {
		return fmt.Errorf("job assignment endpoint does not match registered Run endpoint")
	}
	log.Printf("RUN phase=job status=journal_store job=%s capability=%s attempt=%d", assignment.JobID, assignment.Capability, assignment.Attempt)
	if err := worker.journal.Store(assignment); err != nil {
		log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	log.Printf("RUN phase=job status=ack_start job=%s capability=%s", assignment.JobID, assignment.Capability)
	ack, err := worker.client.AckJob(ctx, protocol.RunJobAckRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		JobID:         assignment.JobID,
		LeaseToken:    assignment.LeaseToken,
		Attempt:       assignment.Attempt,
		Message:       "job accepted by run worker",
	})
	if err != nil {
		log.Printf("RUN phase=job status=ack_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	assignment = ack.Job
	if !ack.Accepted {
		log.Printf("RUN phase=job status=ack_rejected job=%s", assignment.JobID)
		return fmt.Errorf("job acknowledgement was not accepted")
	}
	log.Printf("RUN phase=job status=ack_accepted job=%s attempt=%d", assignment.JobID, assignment.Attempt)
	if err := worker.journal.Store(assignment); err != nil {
		log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	progressSequence := worker.nextProgressSequence(assignment.ProgressSequence)
	state, err = worker.registeredState()
	if err != nil {
		return err
	}
	log.Printf("RUN phase=job status=progress_start job=%s sequence=%d", assignment.JobID, progressSequence)
	progress, err := worker.client.UpdateJobProgress(ctx, protocol.RunJobProgressRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		JobID:         assignment.JobID,
		LeaseToken:    assignment.LeaseToken,
		Attempt:       assignment.Attempt,
		Progress:      protocol.RunJobProgressReport{Percent: 10, Message: "lifecycle execution started"},
		Sequence:      progressSequence,
	})
	if err != nil {
		log.Printf("RUN phase=job status=progress_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	if !progress.Accepted {
		log.Printf("RUN phase=job status=progress_rejected job=%s", assignment.JobID)
		return fmt.Errorf("job progress was not accepted")
	}
	assignment = progress.Job
	log.Printf("RUN phase=job status=progress_accepted job=%s percent=%d", assignment.JobID, assignment.Progress.Percent)
	if err := worker.journal.Store(assignment); err != nil {
		log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	log.Printf("RUN phase=job status=execute_start job=%s capability=%s", assignment.JobID, assignment.Capability)
	execution, assignment, cancelledByPlatform, err := worker.executeWithJobPolling(ctx, assignment)
	if err != nil {
		log.Printf("RUN phase=job status=execute_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	log.Printf("RUN phase=job status=execute_done job=%s state=%s errorCode=%s message=%s cancelledByPlatform=%t", assignment.JobID, execution.State, safeOptional(execution.ErrorCode), safeOptional(execution.Message), cancelledByPlatform)
	if cancelledByPlatform && execution.State == lifecycleResultStateSucceeded {
		execution = LifecycleExecutionResult{
			State:     lifecycleResultStateCancelled,
			Progress:  protocol.RunJobProgressReport{Percent: 100, Message: "cancelled by platform"},
			Message:   "cancelled by platform",
			ErrorCode: "lifecycle_cancelled",
		}
	}
	state, err = worker.registeredState()
	if err != nil {
		return err
	}
	resultRequest := LifecycleResultRequest(assignment, state.SessionToken, execution)
	log.Printf("RUN phase=job status=result_store job=%s state=%s", assignment.JobID, resultRequest.State)
	if err := worker.journal.StorePendingResult(resultRequest, execution.ActivationManifest); err != nil {
		log.Printf("RUN phase=job status=result_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	log.Printf("RUN phase=job status=result_submit job=%s state=%s", assignment.JobID, resultRequest.State)
	result, err := worker.client.CompleteJob(ctx, resultRequest)
	if err != nil {
		log.Printf("RUN phase=job status=result_submit_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	if !result.Accepted {
		log.Printf("RUN phase=job status=result_rejected job=%s", assignment.JobID)
		return fmt.Errorf("job result was not accepted")
	}
	log.Printf("RUN phase=job status=result_accepted job=%s state=%s", assignment.JobID, resultRequest.State)
	if err := worker.journal.Delete(assignment.JobID); err != nil {
		log.Printf("RUN phase=job status=journal_delete_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
		return err
	}
	log.Printf("RUN phase=job status=complete job=%s state=%s", assignment.JobID, resultRequest.State)
	if execution.ActivationManifest != "" {
		log.Printf("RUN phase=self_update status=activate_start job=%s", assignment.JobID)
		if err := worker.executor.selfUpdateActivator.Activate(execution.ActivationManifest); err != nil {
			log.Printf("RUN phase=self_update status=activate_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
			return fmt.Errorf("launch self-update helper: %w", err)
		}
		worker.restartMu.Lock()
		worker.restartRequested = true
		worker.restartMu.Unlock()
	}
	return nil
}

func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment protocol.RunJobAssignment) (LifecycleExecutionResult, protocol.RunJobAssignment, bool, error) {
	jobCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	cancelledByPlatform := false
	pollCancel := func() {
		state, err := worker.registeredState()
		if err != nil {
			log.Printf("RUN phase=job.cancel_poll status=skipped_unregistered job=%s error=%s", assignment.JobID, RedactText(err.Error()))
			return
		}
		log.Printf("RUN phase=job.cancel_poll status=starting job=%s", assignment.JobID)
		response, err := worker.client.PollJobCancel(ctx, protocol.RunJobCancelPollRequest{
			RunEndpointID: state.RunEndpointID,
			SessionToken:  state.SessionToken,
			JobID:         assignment.JobID,
			LeaseToken:    assignment.LeaseToken,
			Attempt:       assignment.Attempt,
		})
		if err != nil {
			log.Printf("RUN phase=job.cancel_poll status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
			return
		}
		if err == nil && response.HasCancel {
			log.Printf("RUN phase=job.cancel_poll status=cancel_requested job=%s", assignment.JobID)
			cancelledByPlatform = true
			cancel()
			return
		}
		log.Printf("RUN phase=job.cancel_poll status=clear job=%s", assignment.JobID)
	}
	pollCancel()

	executionCh := make(chan LifecycleExecutionResult, 1)
	executionAssignment := assignment
	go func() {
		executionCh <- worker.executeAssignment(jobCtx, executionAssignment)
	}()
	log.Printf("RUN phase=job.execute status=worker_started job=%s pollSeconds=%d", assignment.JobID, int(jobActivePollInterval/time.Second))
	ticker := time.NewTicker(jobActivePollInterval)
	defer ticker.Stop()
	for {
		select {
		case execution := <-executionCh:
			log.Printf("RUN phase=job.execute status=worker_finished job=%s state=%s", assignment.JobID, execution.State)
			return execution, assignment, cancelledByPlatform, nil
		case <-ctx.Done():
			log.Printf("RUN phase=job.execute status=context_done job=%s error=%s", assignment.JobID, RedactText(ctx.Err().Error()))
			cancel()
			execution := <-executionCh
			return execution, assignment, cancelledByPlatform, ctx.Err()
		case <-ticker.C:
			log.Printf("RUN phase=job.execute status=active job=%s percent=%d", assignment.JobID, assignment.Progress.Percent)
			if !cancelledByPlatform {
				pollCancel()
			}
			if cancelledByPlatform {
				continue
			}
			state, err := worker.registeredState()
			if err != nil {
				cancel()
				log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
				return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err
			}
			progress, err := worker.client.UpdateJobProgress(ctx, protocol.RunJobProgressRequest{
				RunEndpointID: state.RunEndpointID,
				SessionToken:  state.SessionToken,
				JobID:         assignment.JobID,
				LeaseToken:    assignment.LeaseToken,
				Attempt:       assignment.Attempt,
				Progress:      protocol.RunJobProgressReport{Percent: assignment.Progress.Percent, Message: "lifecycle execution active"},
				Sequence:      worker.nextProgressSequence(assignment.ProgressSequence),
			})
			if err != nil || !progress.Accepted {
				cancel()
				if err == nil {
					err = fmt.Errorf("job lease renewal was not accepted")
				}
				log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
				return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err
			}
			assignment = progress.Job
			log.Printf("RUN phase=job.execute status=lease_renewed job=%s percent=%d", assignment.JobID, assignment.Progress.Percent)
			if err := worker.journal.Store(assignment); err != nil {
				cancel()
				log.Printf("RUN phase=job.execute status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
				return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err
			}
		}
	}
}

func (worker *Worker) nextProgressSequence(minimum uint64) uint64 {
	worker.sequenceMu.Lock()
	defer worker.sequenceMu.Unlock()
	worker.stateMu.Lock()
	defer worker.stateMu.Unlock()
	if worker.state.Sequence < minimum {
		worker.state.Sequence = minimum
	}
	worker.state.Sequence++
	return worker.state.Sequence
}

func (worker *Worker) executeAssignment(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult {
	log.Printf("RUN phase=job.dispatch status=select job=%s capability=%s inputRef=%s target=%s", assignment.JobID, assignment.Capability, safeOptional(assignment.InputRef), safeOptional(assignment.TargetKey))
	if protocol.IsProtectedRequestCapability(assignment.Capability) {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=protected_request", assignment.JobID)
		return worker.executeProtectedRequestJob(ctx, assignment)
	}
	if assignment.ExecutionInput.SourceRCON != nil {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=source_rcon", assignment.JobID)
		return worker.executeSourceRCONJob(ctx, assignment)
	}
	if assignment.Capability == protocol.RunCapabilityDistributionBuild {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=distribution_build", assignment.JobID)
		return worker.executeDistributionBuild(ctx, assignment)
	}
	if assignment.Capability == protocol.RunCapabilityDependenciesCheck || assignment.Capability == protocol.RunCapabilityDependenciesInstall {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=dependencies", assignment.JobID)
		return worker.executeDependencyJob(ctx, assignment)
	}
	if assignment.Capability == protocol.RunCapabilityRunSelfUpdate {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=self_update", assignment.JobID)
		return worker.executeRunSelfUpdate(ctx, assignment)
	}
	if assignment.Capability == protocol.RunCapabilityLogsBackfill {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=logs_backfill", assignment.JobID)
		return worker.executor.ExecuteLogBackfill(ctx, assignment)
	}
	if assignment.Capability == protocol.RunCapabilityRemoteRunDBSQLiteProbe {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=sqlite_schema_probe", assignment.JobID)
		if worker.executor.sqliteSchemaProbe == nil {
			return lifecycleFailure("sqlite_probe_unavailable", "SQLite schema probe executor is unavailable")
		}
		if assignment.ExecutionInput.SQLiteSchemaProbe != nil {
			targetKey, err := worker.materializeSQLiteProbeDataTarget(ctx, assignment)
			if err != nil {
				return sqliteProbeFailureForDataTarget(assignment, err)
			}
			assignment.TargetKey = targetKey
		}
		return worker.executor.sqliteSchemaProbe.Execute(ctx, assignment)
	}
	if isSupportedLifecycleCapability(assignment.Capability) {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=lifecycle", assignment.JobID)
		return worker.executor.ExecuteContext(ctx, assignment)
	}
	if supportedCapability(SupportedFileCapabilities(), assignment.Capability) {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=file", assignment.JobID)
		return worker.executeFileJob(ctx, assignment)
	}
	if isSupportedDistributionCapability(assignment.Capability) {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=distribution", assignment.JobID)
		return ExecuteDistributionJob(ctx, assignment)
	}
	if isSupportedRemoteCapability(assignment.Capability) {
		log.Printf("RUN phase=job.dispatch status=selected job=%s executor=remote", assignment.JobID)
		return ExecuteRemoteAccessJob(ctx, assignment)
	}
	log.Printf("RUN phase=job.dispatch status=unsupported job=%s capability=%s", assignment.JobID, assignment.Capability)
	return lifecycleFailure("unsupported_run_capability", "unsupported run capability")
}

func (worker *Worker) executeFileJob(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult {
	if assignment.Capability != protocol.RunCapabilityFilesRead {
		return worker.executor.ExecuteContext(ctx, assignment)
	}
	fileExecutor := worker.executor.fileExecutor
	if fileExecutor == nil {
		return lifecycleExecutionFailure("file_executor_unavailable", "file executor is unavailable", false)
	}
	scope, targetPath, deploymentRoot, err := fileExecutor.existingReadTargetForAssignment(assignment)
	if err != nil {
		return lifecycleExecutionFailure("file_read_failed", err.Error(), false)
	}
	info, err := os.Stat(targetPath)
	if err != nil {
		return lifecycleExecutionFailure("file_read_failed", err.Error(), false)
	}
	limit := assignment.ExecutionInput.MaxReadBytes
	if limit <= 0 || limit > maxExecutionContentBytes {
		limit = maxExecutionContentBytes
	}
	if info.Size() <= int64(limit) {
		return fileExecutor.read(ctx, scope, deploymentRoot, assignment)
	}
	if info.Size() > maxFileArtifactBytes {
		return lifecycleExecutionFailure("file_read_too_large", "file exceeds artifact transfer limit", false)
	}
	checksum, err := checksumServerFileForArtifact(ctx, targetPath)
	if err != nil {
		if errors.Is(err, context.Canceled) {
			return lifecycleExecutionFailure("file_cancelled", "file read cancelled", false)
		}
		return lifecycleExecutionFailure("file_read_failed", err.Error(), false)
	}
	artifactID := "artifact-" + safeWorkspaceName(assignment.JobID) + "-file-read"
	if err := worker.uploadFileArtifact(ctx, assignment, artifactID, targetPath, info.Size(), checksum); err != nil {
		if errors.Is(err, context.Canceled) {
			return lifecycleExecutionFailure("file_cancelled", "file read cancelled", false)
		}
		return lifecycleExecutionFailure("file_artifact_upload_failed", "file artifact upload failed", false)
	}
	metadata := fileExecutor.metadata(scope, assignment.TargetKey, checksum, info.Size())
	return LifecycleExecutionResult{State: lifecycleResultStateSucceeded, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "file read completed"}, ResultRef: "artifact://" + artifactID, Message: "file read completed", ExecutionResult: protocol.RunJobExecutionResult{Kind: "file.read", Version: metadata.Version, Checksum: checksum, SizeBytes: info.Size(), Summary: "large file transferred as artifact"}}
}

func (worker *Worker) executeProtectedRequestJob(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult {
	if err := protocol.ValidateRunJobAssignment(assignment); err != nil {
		return protectedRequestFailure(assignment.Capability, ProtectedRequestStatusFailed, "protected_request_assignment_invalid")
	}
	state, err := worker.registeredState()
	if err != nil {
		return protectedRequestFailure(assignment.Capability, ProtectedRequestStatusFailed, "protected_request_unregistered")
	}
	input, err := worker.client.GetProtectedRequestExecutionInput(ctx, protocol.ProtectedRequestExecutionInputRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		JobID:         assignment.JobID,
		LeaseToken:    assignment.LeaseToken,
		Attempt:       assignment.Attempt,
		FencingToken:  assignment.FencingToken,
	})
	if err != nil || !protectedRequestInputMatchesAssignment(input, assignment, state.RunEndpointID) {
		return protectedRequestFailure(assignment.Capability, ProtectedRequestStatusFailed, "protected_request_input_unavailable")
	}
	return worker.executor.ExecuteProtectedRequest(ctx, assignment, input)
}

func protectedRequestInputMatchesAssignment(input protocol.ProtectedRequestExecutionInputResponse, assignment protocol.RunJobAssignment, endpointID string) bool {
	return protocol.ValidProtectedRequestExecutionInput(input) && input.JobID == assignment.JobID && input.ServerInstanceID == assignment.ServerInstanceID && input.RunEndpointID == endpointID && input.FencingToken == assignment.FencingToken && input.TargetKey == assignment.TargetKey && input.TransportKey == assignment.ExecutionInput.RemoteAdapterKey && input.Kind == protectedRequestKindForCapability(assignment.Capability)
}

func (worker *Worker) executeSourceRCONJob(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult {
	if err := protocol.ValidateRunJobAssignment(assignment); err != nil {
		return lifecycleFailure("unsafe_source_rcon_plan", "Source RCON plan is invalid")
	}
	state, err := worker.registeredState()
	if err != nil {
		return lifecycleFailure("source_rcon_unregistered", "Run worker is not registered")
	}
	input, err := worker.client.GetSourceRCONExecutionInput(ctx, protocol.SourceRCONExecutionInputRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		JobID:         assignment.JobID,
		LeaseToken:    assignment.LeaseToken,
		Attempt:       assignment.Attempt,
	})
	if err != nil || input.JobID != assignment.JobID || input.ServerInstanceID != assignment.ServerInstanceID || input.RunEndpointID != state.RunEndpointID {
		return lifecycleFailure("source_rcon_input_unavailable", "Source RCON command input is unavailable")
	}
	return worker.executor.ExecuteSourceRCON(ctx, assignment, input.Command)
}

func supportedCapability(capabilities []string, target string) bool {
	for _, capability := range capabilities {
		if capability == target {
			return true
		}
	}
	return false
}

func (worker *Worker) ReconcileOnce(ctx context.Context) error {
	state, err := worker.registeredState()
	if err != nil {
		return err
	}
	activeBefore := worker.journal.ActiveCount()
	log.Printf("RUN phase=reconcile status=starting endpoint=%s activeJobs=%d", state.RunEndpointID, activeBefore)
	response, err := worker.client.ReconcileJobs(ctx, protocol.RunJobReconcileRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		ActiveJobs:    worker.journal.ReconcileEntries(),
	})
	if err != nil {
		log.Printf("RUN phase=reconcile status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error()))
		return err
	}
	if !response.Accepted {
		log.Printf("RUN phase=reconcile status=rejected endpoint=%s", state.RunEndpointID)
		return fmt.Errorf("job reconciliation was not accepted")
	}
	confirmed := map[string]struct{}{}
	for _, job := range response.ConfirmedJobs {
		if job.RunEndpointID != state.RunEndpointID {
			return fmt.Errorf("reconciled job endpoint does not match registered Run endpoint")
		}
		if err := worker.journal.Store(job); err != nil {
			return err
		}
		confirmed[job.JobID] = struct{}{}
	}
	for _, jobID := range response.DiscardJobIDs {
		if err := worker.journal.Delete(jobID); err != nil {
			return err
		}
		confirmed[jobID] = struct{}{}
	}
	for _, job := range worker.journal.ActiveJobs() {
		if _, accounted := confirmed[job.JobID]; !accounted {
			if err := worker.journal.Delete(job.JobID); err != nil {
				return err
			}
		}
	}
	log.Printf("RUN phase=reconcile status=accepted endpoint=%s confirmed=%d discarded=%d activeBefore=%d activeAfter=%d", state.RunEndpointID, len(response.ConfirmedJobs), len(response.DiscardJobIDs), activeBefore, worker.journal.ActiveCount())
	return nil
}

func (worker *Worker) RecoverActiveJobs(ctx context.Context) error {
	activeJobs := worker.journal.ActiveJobs()
	log.Printf("RUN phase=recover status=starting activeJobs=%d", len(activeJobs))
	for _, assignment := range activeJobs {
		if pending, ok := worker.journal.PendingResult(assignment.JobID); ok {
			log.Printf("RUN phase=recover status=pending_result job=%s", assignment.JobID)
			activationManifest := worker.journal.PendingActivation(assignment.JobID)
			state, err := worker.registeredState()
			if err != nil {
				return err
			}
			pending.SessionToken = state.SessionToken
			result, err := worker.client.CompleteJob(ctx, pending)
			if err != nil {
				log.Printf("RUN phase=recover status=result_submit_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
				return err
			}
			if !result.Accepted {
				log.Printf("RUN phase=recover status=result_rejected job=%s", assignment.JobID)
				return fmt.Errorf("recovered job result was not accepted")
			}
			if err := worker.journal.Delete(assignment.JobID); err != nil {
				return err
			}
			if activationManifest != "" {
				if err := worker.executor.selfUpdateActivator.Activate(activationManifest); err != nil {
					return fmt.Errorf("launch recovered self-update helper: %w", err)
				}
				worker.restartMu.Lock()
				worker.restartRequested = true
				worker.restartMu.Unlock()
			}
			continue
		}
		log.Printf("RUN phase=recover status=rerun_active_job job=%s capability=%s", assignment.JobID, assignment.Capability)
		if err := worker.runAssignment(ctx, assignment); err != nil {
			log.Printf("RUN phase=recover status=rerun_failed job=%s error=%s", assignment.JobID, RedactText(err.Error()))
			return err
		}
	}
	log.Printf("RUN phase=recover status=complete activeJobs=%d", worker.journal.ActiveCount())
	return nil
}

func (worker *Worker) Run(ctx context.Context) error {
	log.Printf("RUN phase=run status=starting endpoint=%s", worker.cfg.RunEndpointID)
	if err := worker.Register(ctx); err != nil {
		return err
	}
	if err := worker.ReconcileOnce(ctx); err != nil {
		return err
	}
	if err := worker.RecoverActiveJobs(ctx); err != nil {
		return err
	}
	if err := worker.RunAutonomousLifecycleOnce(ctx); err != nil {
		return err
	}
	if err := worker.reportAutonomousProcessObservations(ctx); err != nil {
		log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", RedactText(err.Error()))
	}
	worker.reportMetricsDegraded(ctx, "startup")
	if err := MarkSelfUpdateHealthy(worker.cfg.UpdateHealthFile); err != nil {
		log.Printf("RUN phase=self_update_health status=mark_failed error=%s", RedactText(err.Error()))
		return err
	}
	if err := worker.reportRunUpdateHealth(ctx); err != nil {
		return err
	}
	heartbeatInterval := durationOrDefault(worker.cfg.HeartbeatInterval, 15*time.Second)
	jobInterval := durationOrDefault(worker.cfg.PollInterval, 2*time.Second)
	state := worker.State()
	log.Printf("RUN phase=run status=ready endpoint=%s heartbeatSeconds=%d jobPollSeconds=%d", state.RunEndpointID, int(heartbeatInterval/time.Second), int(jobInterval/time.Second))
	heartbeatTicker := time.NewTicker(heartbeatInterval)
	defer heartbeatTicker.Stop()
	workerCtx, cancelWorker := context.WithCancel(ctx)
	uploaderDone := make(chan struct{})
	go worker.runDurableUploaders(workerCtx, uploaderDone)
	jobDone := make(chan error, 1)
	go func() { jobDone <- worker.runJobLoop(workerCtx, jobInterval) }()
	defer func() {
		cancelWorker()
		<-uploaderDone
	}()
	for {
		select {
		case <-ctx.Done():
			log.Printf("RUN phase=run status=context_done error=%s", RedactText(ctx.Err().Error()))
			return ctx.Err()
		case err := <-jobDone:
			log.Printf("RUN phase=run status=job_loop_done error=%s", errorSummary(err))
			return err
		case <-heartbeatTicker.C:
			if err := worker.HeartbeatOnce(ctx); err != nil {
				log.Printf("RUN phase=heartbeat status=retry_scheduled backoffMs=%d", boundedRetryBackoff(worker.cfg.RetryBackoff).Milliseconds())
				heartbeatTicker.Reset(boundedRetryBackoff(worker.cfg.RetryBackoff))
				continue
			}
			if err := worker.reportAutonomousProcessObservations(ctx); err != nil {
				log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", RedactText(err.Error()))
			}
			worker.reportMetricsDegraded(ctx, "heartbeat")
			heartbeatTicker.Reset(heartbeatInterval)
		}
	}
}

func (worker *Worker) reportRunUpdateHealth(ctx context.Context) error {
	if worker.cfg.UpdateJobID == "" && worker.cfg.UpdateOutcome == "" && worker.cfg.UpdateAttempt == 0 && worker.cfg.UpdateLeaseToken == "" {
		log.Printf("RUN phase=self_update_health status=skipped")
		return nil
	}
	if worker.cfg.UpdateJobID == "" || (worker.cfg.UpdateOutcome != "succeeded" && worker.cfg.UpdateOutcome != "rolled-back") || worker.cfg.UpdateAttempt <= 0 || worker.cfg.UpdateLeaseToken == "" {
		log.Printf("RUN phase=self_update_health status=invalid_config")
		return fmt.Errorf("self-update health report configuration is incomplete")
	}
	log.Printf("RUN phase=self_update_health status=reporting job=%s outcome=%s attempt=%d", worker.cfg.UpdateJobID, worker.cfg.UpdateOutcome, worker.cfg.UpdateAttempt)
	state, err := worker.registeredState()
	if err != nil {
		return err
	}
	response, err := worker.client.ReportRunUpdateHealth(ctx, protocol.RunUpdateHealthRequest{
		RunEndpointID: state.RunEndpointID,
		SessionToken:  state.SessionToken,
		JobID:         worker.cfg.UpdateJobID,
		LeaseToken:    worker.cfg.UpdateLeaseToken,
		Attempt:       worker.cfg.UpdateAttempt,
		Outcome:       worker.cfg.UpdateOutcome,
		Version:       worker.cfg.Version,
	})
	if err != nil {
		log.Printf("RUN phase=self_update_health status=failed job=%s error=%s", worker.cfg.UpdateJobID, RedactText(err.Error()))
		return err
	}
	if !response.Accepted || response.JobID != worker.cfg.UpdateJobID {
		log.Printf("RUN phase=self_update_health status=rejected job=%s", worker.cfg.UpdateJobID)
		return fmt.Errorf("self-update health report was not accepted")
	}
	log.Printf("RUN phase=self_update_health status=accepted job=%s", worker.cfg.UpdateJobID)
	return nil
}

func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration) error {
	log.Printf("RUN phase=job_loop status=starting fallbackPollMs=%d claimWaitSeconds=%d", interval.Milliseconds(), jobClaimWaitSeconds)
	for {
		select {
		case <-ctx.Done():
			log.Printf("RUN phase=job_loop status=context_done error=%s", RedactText(ctx.Err().Error()))
			return ctx.Err()
		default:
		}
		if worker.journal.ActiveCount() > 0 {
			log.Printf("RUN phase=job_loop status=active_jobs activeJobs=%d", worker.journal.ActiveCount())
			if err := worker.ReconcileOnce(ctx); err != nil {
				log.Printf("RUN phase=job_loop status=reconcile_failed error=%s", RedactText(err.Error()))
				if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil {
					return err
				}
				continue
			}
			if err := worker.RecoverActiveJobs(ctx); err != nil {
				log.Printf("RUN phase=job_loop status=recover_failed error=%s", RedactText(err.Error()))
				if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil {
					return err
				}
				continue
			}
		}
		claimStartedAt := time.Now()
		handled, err := worker.ClaimAndRunOnce(ctx)
		if err != nil {
			log.Printf("RUN phase=job_loop status=claim_failed error=%s", RedactText(err.Error()))
			if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil {
				return err
			}
			continue
		}
		worker.restartMu.Lock()
		restartRequested := worker.restartRequested
		worker.restartMu.Unlock()
		if restartRequested {
			log.Printf("RUN phase=job_loop status=restart_requested")
			return ErrSelfUpdateRestartRequested
		}
		if !handled && time.Since(claimStartedAt) < interval {
			if err := waitWorkerLoop(ctx, interval-time.Since(claimStartedAt)); err != nil {
				return err
			}
		}
	}
}

func waitWorkerLoop(ctx context.Context, delay time.Duration) error {
	if delay <= 0 {
		return nil
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

type durableLogClient interface {
	IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error)
}

type durableArtifactClient interface {
	UploadArtifactChunk(context.Context, protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error)
}

type sessionLogBatchClient struct {
	client        durableLogClient
	runEndpointID string
	sessionToken  string
}

type sessionLogStreamProgressClient struct {
	client interface {
		GetRunLogStreamProgress(context.Context, protocol.RunLogStreamProgressRequest) (protocol.RunLogStreamProgressResponse, error)
	}
	runEndpointID string
	sessionToken  string
	serverID      string
}

func (client sessionLogStreamProgressClient) GetRunLogStreamProgress(ctx context.Context, streamID string) (uint64, error) {
	response, err := client.client.GetRunLogStreamProgress(ctx, protocol.RunLogStreamProgressRequest{
		RunEndpointID:    client.runEndpointID,
		SessionToken:     client.sessionToken,
		ServerInstanceID: client.serverID,
		LogStreamID:      streamID,
	})
	if err != nil {
		return 0, err
	}
	if !response.Accepted || response.LogStreamID != streamID {
		return 0, fmt.Errorf("platform log stream progress response is invalid")
	}
	return response.LatestSeq, nil
}

func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
	batch.RunEndpointID = client.runEndpointID
	batch.SessionToken = client.sessionToken
	if checksum, err := checksumForLogEntries(batch.Entries); err == nil {
		batch.Checksum = checksum
	}
	response, err := client.client.IngestLogBatch(ctx, batch)
	if err != nil && (logBatchSequenceGapError(err) || logBatchAcknowledgedRangeConflict(err)) {
		reason := "platform_sequence_gap"
		if logBatchAcknowledgedRangeConflict(err) {
			reason = "platform_acknowledged_range_conflict"
		}
		log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=%s error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, reason, RedactText(err.Error()))
		return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection(reason, err)
	}
	if err != nil && logBatchLegacySessionMetadataError(err) && strings.TrimSpace(batch.LogSessionID) == "" && !batch.SessionStartedAt.IsZero() {
		log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=legacy_session_metadata error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error()))
		return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("legacy_session_metadata", err)
	}
	if err != nil && logBatchSessionMetadataMismatchError(err) {
		log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=session_metadata_mismatch error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error()))
		return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("session_metadata_mismatch", err)
	}
	return response, err
}

func logBatchSequenceGapError(err error) bool {
	var gapError interface{ LogBatchSequenceGap() bool }
	return errors.As(err, &gapError) && gapError.LogBatchSequenceGap()
}

func logBatchAcknowledgedRangeConflict(err error) bool {
	var conflict interface{ LogBatchAcknowledgedRangeConflict() bool }
	return errors.As(err, &conflict) && conflict.LogBatchAcknowledgedRangeConflict()
}

func logBatchLegacySessionMetadataError(err error) bool {
	var legacyMetadata interface{ LogBatchLegacySessionMetadata() bool }
	return errors.As(err, &legacyMetadata) && legacyMetadata.LogBatchLegacySessionMetadata()
}

func logBatchSessionMetadataMismatchError(err error) bool {
	var mismatch interface{ LogBatchSessionMetadataMismatch() bool }
	return errors.As(err, &mismatch) && mismatch.LogBatchSessionMetadataMismatch()
}

func artifactChunkMissingOnPlatform(err error) bool {
	var httpErr interface{ HTTPStatus() int }
	return errors.As(err, &httpErr) && httpErr.HTTPStatus() == http.StatusNotFound
}

type sessionArtifactChunkClient struct {
	client        durableArtifactClient
	runEndpointID string
	sessionToken  string
}

func (client sessionArtifactChunkClient) UploadArtifactChunk(ctx context.Context, chunk protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) {
	chunk.RunEndpointID = client.runEndpointID
	chunk.SessionToken = client.sessionToken
	response, err := client.client.UploadArtifactChunk(ctx, chunk)
	if err != nil {
		if artifactChunkMissingOnPlatform(err) {
			log.Printf("RUN phase=durable_uploaders.artifacts status=drop_stale transfer=%s artifact=%s chunk=%d reason=platform_transfer_missing", safeOptional(chunk.TransferID), safeOptional(chunk.ArtifactID), chunk.ChunkIndex)
			return protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: chunk.TransferID, ArtifactID: chunk.ArtifactID, ChunkIndex: chunk.ChunkIndex}, nil
		}
		return response, err
	}
	return response, nil
}

func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struct{}) {
	defer close(done)
	logSink, hasLogSink := worker.executor.logSink.(*SpoolLogSink)
	artifactHook, hasArtifactHook := worker.executor.artifactHook.(*QueueArtifactHook)
	logClient, hasLogClient := worker.client.(durableLogClient)
	artifactClient, hasArtifactClient := worker.client.(durableArtifactClient)
	if (!hasLogSink || !hasLogClient) && (!hasArtifactHook || !hasArtifactClient) {
		log.Printf("RUN phase=durable_uploaders status=disabled logs=%t artifacts=%t", hasLogSink && hasLogClient, hasArtifactHook && hasArtifactClient)
		return
	}
	log.Printf("RUN phase=durable_uploaders status=starting logs=%t artifacts=%t", hasLogSink && hasLogClient, hasArtifactHook && hasArtifactClient)
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Printf("RUN phase=durable_uploaders status=stopping")
			return
		case <-ticker.C:
			state, err := worker.registeredState()
			if err != nil {
				log.Printf("RUN phase=durable_uploaders status=skipped_unregistered error=%s", RedactText(err.Error()))
				continue
			}
			if hasLogSink && hasLogClient {
				startedAt := time.Now()
				log.Printf("RUN phase=durable_uploaders.logs status=flush_start timeoutMs=%d", durableUploaderFlushTimeout.Milliseconds())
				flushCtx, cancel := context.WithTimeout(ctx, durableUploaderFlushTimeout)
				client := sessionLogBatchClient{client: logClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken}
				flushed, err := logSink.Spool.Flush(flushCtx, client)
				if err != nil {
					log.Printf("RUN phase=durable_uploaders.logs status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), RedactText(err.Error()))
				} else if flushed > 0 {
					log.Printf("RUN phase=durable_uploaders.logs status=flushed count=%d durationMs=%d", flushed, time.Since(startedAt).Milliseconds())
				}
				cancel()
			}
			if hasArtifactHook && hasArtifactClient {
				startedAt := time.Now()
				log.Printf("RUN phase=durable_uploaders.artifacts status=flush_start timeoutMs=%d", durableUploaderFlushTimeout.Milliseconds())
				flushCtx, cancel := context.WithTimeout(ctx, durableUploaderFlushTimeout)
				client := sessionArtifactChunkClient{client: artifactClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken}
				flushed, err := artifactHook.Queue.Flush(flushCtx, client)
				if err != nil {
					log.Printf("RUN phase=durable_uploaders.artifacts status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), RedactText(err.Error()))
				} else if flushed > 0 {
					log.Printf("RUN phase=durable_uploaders.artifacts status=flushed count=%d durationMs=%d", flushed, time.Since(startedAt).Milliseconds())
				}
				cancel()
			}
		}
	}
}

func checksumForLogEntries(entries []protocol.LogEntry) (string, error) {
	stable := make([]logEntryChecksumBody, len(entries))
	for i, entry := range entries {
		stable[i] = logEntryChecksumBody{
			Seq:       entry.Seq,
			Timestamp: entry.Timestamp.UTC().Format("2006-01-02T15:04:05.000000000Z07:00"),
			Level:     entry.Level,
			Line:      entry.Line,
			Fields:    entry.Fields,
			Redacted:  entry.Redacted,
		}
	}
	encoded, err := json.Marshal(stable)
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(encoded)
	return "sha256:" + hex.EncodeToString(sum[:]), nil
}

type logEntryChecksumBody struct {
	Seq       uint64            `json:"seq"`
	Timestamp string            `json:"timestamp"`
	Level     string            `json:"level,omitempty"`
	Line      string            `json:"line"`
	Fields    map[string]string `json:"fields,omitempty"`
	Redacted  bool              `json:"redacted"`
}

func (worker *Worker) capacityReport() protocol.RunCapacityReport {
	return worker.capacityReportFor(worker.State())
}

func (worker *Worker) capacityReportFor(state WorkerState) protocol.RunCapacityReport {
	return protocol.RunCapacityReport{
		MaxJobs:     state.Capacity.MaxJobs,
		RunningJobs: worker.journal.ActiveCount(),
		QueuedJobs:  0,
		Summary:     "worker control active; job capacity reported separately",
	}
}

func (worker *Worker) State() WorkerState {
	worker.stateMu.RLock()
	defer worker.stateMu.RUnlock()
	state := worker.state
	state.Capabilities = append([]string(nil), state.Capabilities...)
	return state
}

type SpoolLogSink struct {
	RunEndpointID string
	SessionToken  string
	Spool         spool.LogSpool
	mu            sync.Mutex
	Progress      func(context.Context, string, string) (uint64, error)
}

func (sink *SpoolLogSink) Append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string) error {
	return sink.append(ctx, assignment, stream, line, nil)
}

func (sink *SpoolLogSink) AppendWithCursor(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string, cursor ProcessLogCursor) error {
	return sink.append(ctx, assignment, stream, line, &spool.LogSourceCursor{StartOffset: cursor.StartOffset, EndOffset: cursor.EndOffset})
}

func (sink *SpoolLogSink) append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string, cursor *spool.LogSourceCursor) error {
	sink.mu.Lock()
	defer sink.mu.Unlock()
	redactedLine := RedactText(line)
	redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ")
	streamKey := declaredProcessStreamKey(assignment, stream)
	logStreamID := logStreamIDForAssignment(assignment, streamKey)
	entry := protocol.LogEntry{Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted}
	source := "process"
	if strings.HasPrefix(stream, "management-program.") {
		source = "management-program"
	} else if assignment.Capability == protocol.RunCapabilityLogsBackfill {
		source = "file"
	}
	var recoverProgress func(context.Context, string) (uint64, error)
	runStreamPrefix := "run." + assignment.RunEndpointID + "." + assignment.ServerInstanceID + "."
	if sink.Progress != nil && strings.HasPrefix(logStreamID, runStreamPrefix) {
		recoverProgress = func(ctx context.Context, streamID string) (uint64, error) {
			return sink.Progress(ctx, assignment.ServerInstanceID, streamID)
		}
	}
	if strings.TrimSpace(assignment.LogSessionID) != "" {
		// A generation-scoped stream is globally fresh, so its first durable
		// append must not depend on platform availability or the worker ctx.
		recoverProgress = nil
	}
	_, _, err := sink.Spool.EnqueueNextAggregated(ctx, protocol.LogBatchIngestRequest{
		RunEndpointID:    sink.RunEndpointID,
		SessionToken:     sink.SessionToken,
		LogStreamID:      logStreamID,
		ServerInstanceID: assignment.ServerInstanceID,
		StreamKey:        streamKey,
		Source:           source,
		LogSessionID:     assignment.LogSessionID,
		SessionStartedAt: assignment.SessionStartedAt,
		Entries:          []protocol.LogEntry{entry},
	}, cursor, recoverProgress, checksumForLogEntries)
	return err
}

func logStreamIDForAssignment(assignment protocol.RunJobAssignment, streamKey string) string {
	if strings.TrimSpace(assignment.LogSessionID) != "" {
		return fmt.Sprintf("run.%s.%s.%s.%s", assignment.RunEndpointID, assignment.ServerInstanceID, assignment.LogSessionID, streamKey)
	}
	if autonomousLifecycleLogAssignment(assignment) {
		return fmt.Sprintf("run.%s.%s.%s", assignment.RunEndpointID, assignment.ServerInstanceID, streamKey)
	}
	return fmt.Sprintf("job.%s.%s", assignment.JobID, streamKey)
}

func autonomousLifecycleLogAssignment(assignment protocol.RunJobAssignment) bool {
	return strings.HasPrefix(assignment.IdempotencyKey, "autonomous:") || strings.HasPrefix(assignment.LeaseToken, "local-autonomous-") || strings.HasPrefix(assignment.JobID, "autonomous-")
}

func declaredProcessStreamKey(assignment protocol.RunJobAssignment, stream string) string {
	kind := ""
	if stream == "stdout" {
		kind = "process.stdout"
	} else if stream == "stderr" {
		kind = "process.stderr"
	}
	if kind != "" {
		for _, source := range assignment.ExecutionInput.LogSources {
			if source.Kind == kind && strings.TrimSpace(source.StreamKey) != "" {
				return source.StreamKey
			}
		}
	}
	return stream
}

type QueueArtifactHook struct {
	RunEndpointID string
	SessionToken  string
	Queue         spool.ArtifactQueue
}

func (hook QueueArtifactHook) QueueLifecycleResult(_ context.Context, assignment protocol.RunJobAssignment, result ProcessResult) (string, error) {
	ref := fmt.Sprintf("artifact://jobs/%s/lifecycle-result", assignment.JobID)
	payload := []byte(RedactText(result.Stdout + result.Stderr))
	if len(payload) == 0 {
		payload = []byte("lifecycle result metadata")
	}
	artifactID := "artifact-" + assignment.JobID + "-lifecycle"
	if err := hook.Queue.Enqueue(protocol.ArtifactChunkUploadRequest{
		RunEndpointID: hook.RunEndpointID,
		SessionToken:  hook.SessionToken,
		TransferID:    "transfer-" + assignment.JobID,
		ArtifactID:    artifactID,
		ChunkIndex:    0,
		Offset:        0,
		SizeBytes:     len(payload),
		Checksum:      checksumForText(string(payload)),
		Payload:       payload,
	}); err != nil {
		return "", err
	}
	return ref, nil
}

func capabilityFingerprint(capabilities []string) string {
	return checksumForText(strings.Join(capabilities, ","))
}

func durationOrDefault(value time.Duration, fallback time.Duration) time.Duration {
	if value <= 0 {
		return fallback
	}
	return value
}

func boundedRetryBackoff(value time.Duration) time.Duration {
	value = durationOrDefault(value, time.Second)
	if value > 30*time.Second {
		return 30 * time.Second
	}
	return value
}