package runtime

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/automation/blob"
	"github.com/tidbcloud/AgentTeam/internal/automation/dsl"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/store/automation"
	"gorm.io/gorm"
)

const (
	defaultServerRunnerID                    = "server-default"
	defaultServerRunnerConcurrency           = 1
	maxServerRunnerConcurrency               = 32
	defaultServerRunnerPollInterval          = time.Second
	defaultServerRunnerJobTimeout            = 30 * time.Minute
	defaultServerRunnerCheckoutTimeout       = 5 * time.Minute
	defaultServerRunnerCheckoutMaxBytes      = int64(512 * 1024 * 1024)
	defaultServerRunnerCheckoutMaxFiles      = 100000
	defaultServerRunnerRecoveryWorkspaceScan = 20
	maxServerRunnerRecoveryWorkspaceScan     = 100
	maxServerRunnerLogChunksPerStream        = 256
	defaultServerRunnerMaxLogChunkBytes      = 64 * 1024
	defaultServerRunnerMaxRunnerLabels       = 8
	serverRunnerLeaseGrace                   = time.Minute
	serverRunnerLogContentType               = "text/plain; charset=utf-8"
	serverRunnerCancelPollInterval           = 500 * time.Millisecond
	serverRunnerFailureKindJobTimeout        = "job_timeout"
	serverRunnerFailureKindLogWrite          = "log_write_failed"
)

var errAutomationJobCanceled = errors.New("automation job canceled")

// ServerRunnerConfig configures the MeshX server-managed automation runner.
type ServerRunnerConfig struct {
	RunnerID               string
	QueueName              string
	Labels                 []string
	Concurrency            int
	WorkDir                string
	PollInterval           time.Duration
	JobTimeout             time.Duration
	CheckoutMaterializer   CheckoutMaterializer
	WikiPublisher          WikiPublisher
	TreePublisher          TreePublisher
	CheckoutTimeout        time.Duration
	CheckoutMaxBytes       int64
	CheckoutMaxFiles       int
	RecoveryWorkspaceLimit int
	MaxLogChunkBytes       int
	MaxRunnerLabels        int
	Executor               JobExecutor
	CredentialKey          []byte
}

// ServerRunner claims ready automation jobs and delegates validated shell-only
// execution to its configured executor. It is an internal background runner,
// not the public runner registration API.
type ServerRunner struct {
	store          *automation.Store
	blobStore      blob.Store
	cfg            ServerRunnerConfig
	executor       JobExecutor
	leaseOwner     string
	workspaces     chan string
	now            func() time.Time
	logger         *slog.Logger
	shardBuckets   []int
	workspaceScope func(context.Context) ([]string, error)
	closeOnce      sync.Once
	closeErr       error
	claimMu        sync.Mutex
	claiming       bool
	inFlight       sync.WaitGroup
}

// NewServerRunner constructs a server-managed automation runner and verifies that
// its working directory can be created before the server advertises Automation as
// runnable.
func NewServerRunner(store *automation.Store, blobStore blob.Store, cfg ServerRunnerConfig, now func() time.Time, logger *slog.Logger) (*ServerRunner, error) {
	if store == nil {
		return nil, fmt.Errorf("automation server runner store is nil")
	}
	if blobStore == nil {
		return nil, fmt.Errorf("automation server runner blob store is nil")
	}
	cfg = normalizeServerRunnerConfig(cfg)
	if len(cfg.Labels) > cfg.MaxRunnerLabels {
		return nil, fmt.Errorf("automation server runner labels exceed maximum count")
	}
	if cfg.Executor == nil {
		return nil, fmt.Errorf("automation server runner executor is nil")
	}
	if err := os.MkdirAll(cfg.WorkDir, 0o755); err != nil {
		return nil, fmt.Errorf("create automation server runner workdir: %w", err)
	}
	if now == nil {
		now = func() time.Time { return time.Now().UTC() }
	}
	queueCapacity := cfg.Concurrency * cfg.RecoveryWorkspaceLimit * 2
	if queueCapacity < 64 {
		queueCapacity = 64
	}
	shards := make([]int, 0, automation.AutomationClaimShardCount)
	for shard := 0; shard < automation.AutomationClaimShardCount; shard++ {
		shards = append(shards, shard)
	}
	runner := &ServerRunner{
		store:        store,
		blobStore:    blobStore,
		cfg:          cfg,
		executor:     cfg.Executor,
		leaseOwner:   serverRunnerLeaseOwner(cfg.RunnerID),
		workspaces:   make(chan string, queueCapacity),
		now:          now,
		logger:       logger,
		shardBuckets: shards,
	}
	runner.claiming = true
	return runner, nil
}

func normalizeServerRunnerConfig(cfg ServerRunnerConfig) ServerRunnerConfig {
	cfg.RunnerID = strings.TrimSpace(cfg.RunnerID)
	if cfg.RunnerID == "" {
		cfg.RunnerID = defaultServerRunnerID
	}
	cfg.QueueName = strings.TrimSpace(cfg.QueueName)
	if cfg.QueueName == "" {
		cfg.QueueName = DefaultQueueName
	}
	cfg.Labels = NormalizeRunnerLabels(cfg.Labels)
	if len(cfg.Labels) == 0 {
		cfg.Labels = []string{"ubuntu-latest"}
	}
	if cfg.Concurrency <= 0 {
		cfg.Concurrency = defaultServerRunnerConcurrency
	} else if cfg.Concurrency > maxServerRunnerConcurrency {
		cfg.Concurrency = maxServerRunnerConcurrency
	}
	if cfg.WorkDir == "" {
		cfg.WorkDir = ".data/meshx/automation-runner"
	}
	if cfg.PollInterval <= 0 {
		cfg.PollInterval = defaultServerRunnerPollInterval
	}
	if cfg.JobTimeout <= 0 {
		cfg.JobTimeout = defaultServerRunnerJobTimeout
	}
	if cfg.CheckoutTimeout <= 0 {
		cfg.CheckoutTimeout = defaultServerRunnerCheckoutTimeout
	}
	if cfg.CheckoutMaxBytes <= 0 {
		cfg.CheckoutMaxBytes = defaultServerRunnerCheckoutMaxBytes
	}
	if cfg.CheckoutMaxFiles <= 0 {
		cfg.CheckoutMaxFiles = defaultServerRunnerCheckoutMaxFiles
	}
	if cfg.RecoveryWorkspaceLimit <= 0 {
		cfg.RecoveryWorkspaceLimit = defaultServerRunnerRecoveryWorkspaceScan
	} else if cfg.RecoveryWorkspaceLimit > maxServerRunnerRecoveryWorkspaceScan {
		cfg.RecoveryWorkspaceLimit = maxServerRunnerRecoveryWorkspaceScan
	}
	if cfg.MaxLogChunkBytes <= 0 {
		cfg.MaxLogChunkBytes = defaultServerRunnerMaxLogChunkBytes
	}
	if cfg.MaxRunnerLabels <= 0 || cfg.MaxRunnerLabels > defaultServerRunnerMaxRunnerLabels {
		cfg.MaxRunnerLabels = defaultServerRunnerMaxRunnerLabels
	}
	if len(cfg.CredentialKey) > 0 {
		cfg.CredentialKey = append([]byte(nil), cfg.CredentialKey...)
	}
	return cfg
}

// SignalWorkspace wakes the runner for one workspace with newly ready work. If
// the in-memory queue is full, the bounded recovery scan will pick it up.
func (r *ServerRunner) SignalWorkspace(workspaceID string) {
	if r == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	select {
	case r.workspaces <- workspaceID:
	default:
	}
}

// Run starts server-runner workers and returns after ctx is canceled.
func (r *ServerRunner) Run(ctx context.Context) {
	if r == nil {
		return
	}
	var workers sync.WaitGroup
	for i := 0; i < r.cfg.Concurrency; i++ {
		workers.Add(1)
		logging.Go(nil, func() {
			defer workers.Done()
			r.workerLoop(ctx)
		})
	}
	r.enqueueReadyWorkspaces(ctx)
	ticker := time.NewTicker(r.cfg.PollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			workers.Wait()
			return
		case <-ticker.C:
			r.enqueueReadyWorkspaces(ctx)
		}
	}
}

func (r *ServerRunner) workerLoop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case workspaceID := <-r.workspaces:
			if err := r.processWorkspace(ctx, workspaceID); err != nil && ctx.Err() == nil {
				r.log(ctx, slog.LevelWarn, "automation server runner workspace failed", "workspace_id", workspaceID, "error", err)
			}
		}
	}
}

func (r *ServerRunner) enqueueReadyWorkspaces(ctx context.Context) {
	workspaceIDs, err := r.currentWorkspaceScope(ctx)
	if err != nil {
		if ctx.Err() == nil {
			r.log(ctx, slog.LevelWarn, "automation server runner workspace scope failed", "error", err)
		}
		return
	}
	workspaces, err := r.store.ListReadyWorkspaces(ctx, r.cfg.QueueName, r.currentTime(), r.cfg.RecoveryWorkspaceLimit, workspaceIDs)
	if err != nil {
		if ctx.Err() == nil {
			r.log(ctx, slog.LevelWarn, "automation server runner recovery scan failed", "error", err)
		}
		return
	}
	for _, workspaceID := range workspaces {
		select {
		case <-ctx.Done():
			return
		case r.workspaces <- workspaceID:
		default:
			return
		}
	}
}

func (r *ServerRunner) processWorkspace(ctx context.Context, workspaceID string) error {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return nil
	}
	workspaceIDs, err := r.currentWorkspaceScope(ctx)
	if err != nil {
		return fmt.Errorf("load automation server runner workspace scope: %w", err)
	}
	if workspaceIDs != nil && !containsWorkspaceID(workspaceIDs, workspaceID) {
		return nil
	}
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		r.claimMu.Lock()
		if !r.claiming {
			r.claimMu.Unlock()
			return nil
		}
		jobs, err := r.store.ClaimJobs(ctx, automation.ClaimJobsRequest{
			WorkspaceID:     workspaceID,
			QueueName:       r.cfg.QueueName,
			ShardBuckets:    r.shardBuckets,
			LeaseOwner:      r.leaseOwner,
			LeaseDuration:   r.cfg.JobTimeout + serverRunnerLeaseGrace,
			Limit:           1,
			RunnerGroup:     ServerManagedRunnerGroup,
			Labels:          r.cfg.Labels,
			MaxRunnerLabels: r.cfg.MaxRunnerLabels,
			Now:             r.currentTime(),
		})
		if err != nil {
			r.claimMu.Unlock()
			return err
		}
		if len(jobs) == 0 {
			r.claimMu.Unlock()
			return nil
		}
		r.inFlight.Add(len(jobs))
		r.claimMu.Unlock()
		for _, job := range jobs {
			err := func() error {
				defer r.inFlight.Done()
				return r.runJob(ctx, job)
			}()
			if err != nil {
				if ctx.Err() != nil {
					return ctx.Err()
				}
				r.log(ctx, slog.LevelWarn, "automation server runner job failed", "workspace_id", job.WorkspaceID, "run_id", job.RunID, "job_run_id", job.JobRunID, "queue_job_id", job.ID, "error", err)
			}
		}
	}
}

func (r *ServerRunner) runJob(ctx context.Context, job automation.AutomationJob) error {
	worker := NewWorker(r.store, r.currentTime, nil)
	jobRun, err := worker.AcceptLeasedJob(ctx, job, r.leaseOwner)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil
		}
		return err
	}
	spec, err := serverRunnerShellJob(jobRun)
	if err != nil {
		return worker.FailLeasedJobTerminal(ctx, job, r.leaseOwner, err.Error(), serverRunnerOutputJSON(r.cfg.RunnerID, nil))
	}
	resolvedEnv, err := r.resolveShellJobEnv(ctx, jobRun.WorkspaceID, jobRun.RunID, spec)
	if err != nil {
		return worker.FailLeasedJobTerminal(ctx, job, r.leaseOwner, err.Error(), serverRunnerOutputJSON(r.cfg.RunnerID, nil))
	}
	redactor := NewSecretRedactor(resolvedEnv.secrets)
	workDir, err := r.jobWorkDir(jobRun)
	if err != nil {
		return worker.FailLeasedJobTerminal(ctx, job, r.leaseOwner, err.Error(), serverRunnerOutputJSON(r.cfg.RunnerID, nil))
	}
	if err := os.MkdirAll(workDir, 0o755); err != nil {
		return worker.FailLeasedJobTerminal(ctx, job, r.leaseOwner, fmt.Sprintf("create automation job workdir: %v", err), serverRunnerOutputJSON(r.cfg.RunnerID, nil))
	}
	defer func() {
		if err := os.RemoveAll(workDir); err != nil {
			r.log(ctx, slog.LevelWarn, "automation server runner workdir cleanup failed", "workspace_id", job.WorkspaceID, "run_id", job.RunID, "job_run_id", job.JobRunID, "workdir", workDir, "error", err)
		}
	}()
	execCtx, stopExecution := r.jobExecutionContext(ctx, job, r.cfg.JobTimeout)
	defer stopExecution(nil)
	cancelLogs := func() { stopExecution(nil) }
	logs := newServerRunnerJobLogs(resolvedEnv.secrets)
	stdoutWriter := r.newLogWriter(execCtx, job, jobRun, "stdout", &logs.stdout, cancelLogs)
	stderrWriter := r.newLogWriter(execCtx, job, jobRun, "stderr", &logs.stderr, cancelLogs)
	output, err := r.executor.ExecuteJob(execCtx, ExecuteJobRequest{
		RunnerID: r.cfg.RunnerID,
		Job: JobExecutionIdentity{
			QueueJobID:  job.ID,
			WorkspaceID: job.WorkspaceID,
			RunID:       job.RunID,
			JobRunID:    jobRun.ID,
			JobID:       jobRun.JobID,
		},
		Spec:    spec,
		WorkDir: workDir,
		Env:     r.runnerEnv(jobRun, nil),
		StepEnv: r.stepEnvs(jobRun, resolvedEnv),
		Stdout:  stdoutWriter,
		Stderr:  stderrWriter,
		Now:     r.currentTime,
		UsesStepHandler: func(ctx context.Context, req UsesStepRequest) (int, error) {
			return r.executeUsesStep(ctx, job, jobRun, req.Step, req.Index, workDir)
		},
	})
	_ = stdoutWriter.Flush()
	_ = stderrWriter.Flush()
	if errors.Is(context.Cause(execCtx), errAutomationJobCanceled) {
		r.logExecutorCleanupFailure(ctx, "automation server runner canceled job cleanup failed", job, err, redactor)
		return nil
	}
	if err != nil {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if errors.Is(context.Cause(execCtx), context.DeadlineExceeded) || errors.Is(err, context.DeadlineExceeded) {
			r.logExecutorCleanupFailure(ctx, "automation server runner timed out job cleanup failed", job, err, redactor)
			return worker.FailLeasedJobTerminalWithMetadata(ctx, job, r.leaseOwner, serverRunnerJobTimeoutError(r.cfg.JobTimeout).Error(), serverRunnerFailureMetadata(serverRunnerFailureKindJobTimeout), serverRunnerOutputJSON(r.cfg.RunnerID, output.Steps))
		}
		return worker.FailLeasedJobTerminalWithMetadata(ctx, job, r.leaseOwner, redactor.RedactString(err.Error()), serverRunnerFailureMetadataForError(err, logs.firstError()), serverRunnerOutputJSON(r.cfg.RunnerID, output.Steps))
	}
	if logErr := logs.firstError(); logErr != nil {
		return worker.FailLeasedJobTerminalWithMetadata(ctx, job, r.leaseOwner, redactor.RedactString(fmt.Sprintf("write automation job log: %v", logErr)), serverRunnerFailureMetadata(serverRunnerFailureKindLogWrite), serverRunnerOutputJSON(r.cfg.RunnerID, output.Steps))
	}
	if err := worker.CompleteLeasedJob(ctx, job, r.leaseOwner, serverRunnerOutputJSON(r.cfg.RunnerID, output.Steps)); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil
		}
		return err
	}
	r.log(ctx, slog.LevelInfo, "automation server runner job completed", "workspace_id", job.WorkspaceID, "run_id", job.RunID, "job_run_id", job.JobRunID, "queue_job_id", job.ID)
	return nil
}

func (r *ServerRunner) logExecutorCleanupFailure(ctx context.Context, message string, job automation.AutomationJob, err error, redactor *SecretRedactor) {
	if cleanupErr := executorCleanupFailure(err); cleanupErr != nil {
		r.log(ctx, slog.LevelWarn, message, "workspace_id", job.WorkspaceID, "run_id", job.RunID, "job_run_id", job.JobRunID, "queue_job_id", job.ID, "error", redactor.RedactString(cleanupErr.Error()))
	}
}

func (r *ServerRunner) jobExecutionContext(ctx context.Context, job automation.AutomationJob, timeout time.Duration) (context.Context, func(error)) {
	if timeout <= 0 {
		timeout = defaultServerRunnerJobTimeout
	}
	timeoutCtx, cancelTimeout := context.WithTimeout(ctx, timeout)
	execCtx, cancel := context.WithCancelCause(timeoutCtx)
	done := make(chan struct{})
	var stopOnce sync.Once
	stop := func(err error) {
		stopOnce.Do(func() {
			close(done)
			cancel(err)
			cancelTimeout()
		})
	}
	logging.Go(nil, func() {
		ticker := time.NewTicker(serverRunnerCancelPollInterval)
		defer ticker.Stop()
		for {
			select {
			case <-done:
				return
			case <-execCtx.Done():
				return
			case <-ticker.C:
				canceled, err := r.store.RunnerJobCancelRequested(execCtx, job.WorkspaceID, job.ID, r.leaseOwner)
				if err != nil {
					r.log(execCtx, slog.LevelWarn, "automation server runner cancel check failed", "workspace_id", job.WorkspaceID, "run_id", job.RunID, "job_run_id", job.JobRunID, "queue_job_id", job.ID, "error", err)
					continue
				}
				if canceled {
					stop(errAutomationJobCanceled)
					return
				}
			}
		}
	})
	return execCtx, stop
}

func serverRunnerJobTimeoutError(timeout time.Duration) error {
	if timeout <= 0 {
		timeout = defaultServerRunnerJobTimeout
	}
	return fmt.Errorf("automation job timed out after %s", timeout)
}

func serverRunnerFailureMetadata(kind string) map[string]any {
	kind = strings.TrimSpace(kind)
	if kind == "" {
		return nil
	}
	return map[string]any{"failure_kind": kind}
}

func serverRunnerFailureMetadataForError(err error, logErr error) map[string]any {
	if logErr != nil {
		return serverRunnerFailureMetadata(serverRunnerFailureKindLogWrite)
	}
	if kind := ExecutorFailureKindFor(err); kind != "" {
		return serverRunnerFailureMetadata(string(kind))
	}
	return nil
}

func serverRunnerShellJob(jobRun automation.AutomationJobRun) (dsl.JobSpec, error) {
	var spec dsl.JobSpec
	if err := json.Unmarshal([]byte(jobRun.InputJSON), &spec); err != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation server runner job input is invalid")
	}
	if strings.TrimSpace(spec.If) != "" {
		return dsl.JobSpec{}, fmt.Errorf("automation job if conditions are not supported by the server runner")
	}
	if spec.Strategy != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation job strategy and matrix are not supported by the server runner")
	}
	if spec.Services != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation job services are not supported by the server runner")
	}
	if spec.Container != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation job containers are not supported by the server runner")
	}
	if spec.Concurrency != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation job concurrency is not supported by the server runner")
	}
	if spec.Environment != nil {
		return dsl.JobSpec{}, fmt.Errorf("automation job environments are not supported by the server runner")
	}
	if len(spec.Outputs) > 0 {
		return dsl.JobSpec{}, fmt.Errorf("automation job outputs are not supported by the server runner")
	}
	if len(spec.Defaults) > 0 {
		return dsl.JobSpec{}, fmt.Errorf("automation job defaults are not supported by the server runner")
	}
	if len(spec.Steps) == 0 {
		return dsl.JobSpec{}, fmt.Errorf("automation job must contain at least one step")
	}
	for idx, step := range spec.Steps {
		uses := strings.TrimSpace(step.Uses)
		run := strings.TrimSpace(step.Run)
		if strings.TrimSpace(step.If) != "" {
			return dsl.JobSpec{}, fmt.Errorf("automation step if conditions are not supported by the server runner")
		}
		if uses != "" {
			if run != "" {
				return dsl.JobSpec{}, fmt.Errorf("automation step %d cannot define both uses and run", idx)
			}
			if uses != dsl.AgentTeamCheckoutUse && uses != dsl.AgentTeamPublishWikiUse && uses != dsl.AgentTeamPublishTreeUse {
				return dsl.JobSpec{}, fmt.Errorf("automation step uses is not supported by the server runner")
			}
			continue
		}
		if run == "" {
			return dsl.JobSpec{}, fmt.Errorf("automation step %d must define run", idx)
		}
	}
	return spec, nil
}

func (r *ServerRunner) executeUsesStep(ctx context.Context, job automation.AutomationJob, jobRun automation.AutomationJobRun, step dsl.StepSpec, stepIndex int, workDir string) (int, error) {
	switch strings.TrimSpace(step.Uses) {
	case dsl.AgentTeamCheckoutUse:
		if err := r.executeCheckoutStep(ctx, job, jobRun, step, stepIndex, workDir); err != nil {
			return -1, err
		}
		return 0, nil
	case dsl.AgentTeamPublishWikiUse:
		if err := r.executePublishWikiStep(ctx, job, jobRun, step, stepIndex, workDir); err != nil {
			return -1, err
		}
		return 0, nil
	case dsl.AgentTeamPublishTreeUse:
		if err := r.executePublishTreeStep(ctx, job, jobRun, step, stepIndex, workDir); err != nil {
			return -1, err
		}
		return 0, nil
	default:
		return -1, fmt.Errorf("automation step uses is not supported by the server runner")
	}
}

func (r *ServerRunner) executeCheckoutStep(ctx context.Context, job automation.AutomationJob, jobRun automation.AutomationJobRun, step dsl.StepSpec, stepIndex int, workDir string) error {
	if r.cfg.CheckoutMaterializer == nil {
		return fmt.Errorf("automation checkout materializer is not configured")
	}
	options, optionErrs := dsl.ParseAgentTeamCheckoutOptions(step.With)
	if len(optionErrs) > 0 {
		return optionErrs[0]
	}
	destination, err := serverRunnerCheckoutDestination(workDir, options.Path)
	if err != nil {
		return err
	}
	run, err := r.store.GetRun(ctx, jobRun.WorkspaceID, jobRun.RunID)
	if err != nil {
		return fmt.Errorf("load automation run for checkout: %w", err)
	}
	repositoryFullName := strings.TrimSpace(run.RepositoryFullName)
	commitSHA := strings.TrimSpace(run.CommitSHA)
	contentID := strings.TrimSpace(run.ContentID)
	contentVersionID := strings.TrimSpace(run.ContentVersionID)
	if contentID == "" || contentVersionID == "" {
		if repositoryFullName == "" || commitSHA == "" {
			return fmt.Errorf("automation checkout requires exact content version provenance")
		}
	}
	request := CheckoutRequest{
		WorkspaceID: jobRun.WorkspaceID, RunID: jobRun.RunID, JobRunID: jobRun.ID,
		RepositoryFullName: repositoryFullName, CommitSHA: commitSHA,
		ContentID: contentID, ContentVersionID: contentVersionID,
		PrincipalID: strings.TrimSpace(run.OwnerPrincipalID),
		Destination: destination, Clean: options.Clean,
		MaxBytes: r.cfg.CheckoutMaxBytes, MaxFiles: r.cfg.CheckoutMaxFiles,
	}
	if err := r.appendCheckoutEvent(ctx, job, jobRun, stepIndex, "checkout_started", "Automation checkout started", checkoutEventMetadata(request, options.Path, nil)); err != nil {
		return err
	}
	checkoutTimeout := r.cfg.CheckoutTimeout
	checkoutCtx, cancel := context.WithTimeout(ctx, checkoutTimeout)
	defer cancel()
	started := time.Now()
	result, err := r.cfg.CheckoutMaterializer.Materialize(checkoutCtx, request)
	if err != nil {
		failedMetadata := checkoutEventMetadata(request, options.Path, nil)
		failedMetadata["error"] = err.Error()
		eventErr := r.appendCheckoutEvent(ctx, job, jobRun, stepIndex, "checkout_failed", "Automation checkout failed", failedMetadata)
		if eventErr != nil {
			return errors.Join(fmt.Errorf("automation checkout failed: %w", err), eventErr)
		}
		if errors.Is(checkoutCtx.Err(), context.DeadlineExceeded) {
			return fmt.Errorf("automation checkout timed out after %s", checkoutTimeout)
		}
		return fmt.Errorf("automation checkout failed: %w", err)
	}
	if result.Duration <= 0 {
		result.Duration = time.Since(started)
	}
	if err := r.appendCheckoutEvent(ctx, job, jobRun, stepIndex, "checkout_succeeded", "Automation checkout succeeded", checkoutEventMetadata(request, options.Path, &result)); err != nil {
		return err
	}
	return nil
}

func serverRunnerCheckoutDestination(workDir string, checkoutPath string) (string, error) {
	root, err := filepath.Abs(workDir)
	if err != nil {
		return "", fmt.Errorf("resolve automation job workdir: %w", err)
	}
	path := strings.TrimSpace(checkoutPath)
	if path == "" || path == "." {
		return root, nil
	}
	if filepath.IsAbs(path) {
		return "", fmt.Errorf("automation checkout path must be relative")
	}
	cleanedPath := filepath.Clean(filepath.FromSlash(path))
	if cleanedPath == "." {
		return root, nil
	}
	if cleanedPath == ".." || strings.HasPrefix(cleanedPath, ".."+string(os.PathSeparator)) {
		return "", fmt.Errorf("automation checkout path escapes job workdir")
	}
	target := filepath.Join(root, cleanedPath)
	cleanedTarget := filepath.Clean(target)
	if cleanedTarget != root && !strings.HasPrefix(cleanedTarget, root+string(os.PathSeparator)) {
		return "", fmt.Errorf("automation checkout path escapes job workdir")
	}
	return cleanedTarget, nil
}

func (r *ServerRunner) appendCheckoutEvent(ctx context.Context, job automation.AutomationJob, jobRun automation.AutomationJobRun, stepIndex int, eventType string, summary string, metadata map[string]any) error {
	now := r.currentTime()
	event := automation.AutomationEvent{
		ID:               serverRunnerCheckoutEventID(jobRun.ID, job.Attempt, stepIndex, eventType, now),
		WorkspaceID:      jobRun.WorkspaceID,
		RunID:            jobRun.RunID,
		JobRunID:         jobRun.ID,
		EventType:        eventType,
		ActorPrincipalID: r.leaseOwner,
		ResourceType:     "automation_job",
		ResourceID:       strconv.FormatInt(job.ID, 10),
		Summary:          summary,
		MetadataJSON:     jsonString(metadata),
		CreatedAt:        now,
	}
	_, err := r.store.CreateLeasedJobEvent(ctx, job.ID, r.leaseOwner, event, now)
	return err
}

func checkoutEventMetadata(request CheckoutRequest, checkoutPath string, result *CheckoutResult) map[string]any {
	checkoutPath = strings.TrimSpace(checkoutPath)
	if checkoutPath == "" {
		checkoutPath = "."
	}
	metadata := map[string]any{"strategy": "content_version", "path": checkoutPath}
	if contentID := strings.TrimSpace(request.ContentID); contentID != "" {
		metadata["content_id"] = contentID
		metadata["content_version_id"] = strings.TrimSpace(request.ContentVersionID)
	} else {
		metadata["strategy"] = "ags_archive"
		metadata["repository"] = strings.TrimSpace(request.RepositoryFullName)
		metadata["commit_sha"] = strings.TrimSpace(request.CommitSHA)
	}
	if result == nil {
		return metadata
	}
	if strings.TrimSpace(result.Strategy) != "" {
		metadata["strategy"] = strings.TrimSpace(result.Strategy)
	}
	metadata["files"] = result.Files
	metadata["bytes"] = result.Bytes
	metadata["duration_ms"] = result.Duration.Milliseconds()
	return metadata
}

func serverRunnerCheckoutEventID(jobRunID string, attempt int, stepIndex int, eventType string, now time.Time) string {
	seed := fmt.Sprintf("%s:%d:%d:%s:%d", jobRunID, attempt, stepIndex, eventType, now.UnixNano())
	return "wfe_" + blob.ShortHash(seed, 48)
}

func (r *ServerRunner) stepEnvs(jobRun automation.AutomationJobRun, resolved serverRunnerResolvedEnv) [][]string {
	envs := make([][]string, len(resolved.stepEnv))
	for idx := range envs {
		envs[idx] = r.runnerEnv(jobRun, resolved.envForStep(idx))
	}
	return envs
}

func (r *ServerRunner) runnerEnv(jobRun automation.AutomationJobRun, extra map[string]string) []string {
	env := map[string]string{
		"CI":                          "true",
		"MESHX_AUTOMATION_RUN_ID":     jobRun.RunID,
		"MESHX_AUTOMATION_JOB_RUN_ID": jobRun.ID,
		"MESHX_AUTOMATION_JOB_ID":     jobRun.JobID,
		"TMPDIR":                      "/tmp",
	}
	for key, value := range extra {
		env[key] = value
	}
	keys := make([]string, 0, len(env))
	for key := range env {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	out := make([]string, 0, len(env))
	for _, key := range keys {
		out = append(out, key+"="+env[key])
	}
	return out
}

func (logs *serverRunnerJobLogs) firstError() error {
	if logs == nil {
		return nil
	}
	if err := logs.stdout.state.err; err != nil && !errors.Is(err, context.Canceled) {
		return err
	}
	if err := logs.stderr.state.err; err != nil && !errors.Is(err, context.Canceled) {
		return err
	}
	return nil
}

type serverRunnerLogWriter struct {
	ctx     context.Context
	runner  *ServerRunner
	job     automation.AutomationJob
	jobRun  automation.AutomationJobRun
	stream  string
	state   *serverRunnerLogState
	masker  *serverRunnerSecretMasker
	maxByte int64
	cancel  context.CancelFunc
}

type serverRunnerLogState struct {
	nextSeq      int64
	writtenBytes int64
	err          error
}

type serverRunnerJobLogs struct {
	stdout serverRunnerStreamLog
	stderr serverRunnerStreamLog
}

type serverRunnerStreamLog struct {
	state  serverRunnerLogState
	masker *serverRunnerSecretMasker
}

func newServerRunnerJobLogs(secrets map[string]string) *serverRunnerJobLogs {
	matcher := newServerRunnerSecretMatcher(secrets)
	return &serverRunnerJobLogs{
		stdout: serverRunnerStreamLog{masker: newServerRunnerSecretMasker(matcher)},
		stderr: serverRunnerStreamLog{masker: newServerRunnerSecretMasker(matcher)},
	}
}

func (r *ServerRunner) newLogWriter(ctx context.Context, job automation.AutomationJob, jobRun automation.AutomationJobRun, stream string, log *serverRunnerStreamLog, cancel context.CancelFunc) *serverRunnerLogWriter {
	if log == nil {
		log = &serverRunnerStreamLog{}
	}
	return &serverRunnerLogWriter{ctx: ctx, runner: r, job: job, jobRun: jobRun, stream: stream, state: &log.state, masker: log.masker, maxByte: r.effectiveMaxLogBytesPerStream(), cancel: cancel}
}

func (r *ServerRunner) effectiveMaxLogChunkBytes() int {
	if r == nil || r.cfg.MaxLogChunkBytes <= 0 {
		return defaultServerRunnerMaxLogChunkBytes
	}
	return r.cfg.MaxLogChunkBytes
}

func (r *ServerRunner) effectiveMaxLogBytesPerStream() int64 {
	return int64(r.effectiveMaxLogChunkBytes()) * maxServerRunnerLogChunksPerStream
}

func (w *serverRunnerLogWriter) Write(payload []byte) (int, error) {
	if len(payload) == 0 {
		return 0, nil
	}
	masked := payload
	if w.masker != nil {
		masked = w.masker.Mask(payload, false)
	}
	if len(masked) == 0 {
		return len(payload), nil
	}
	if err := w.writeMasked(masked); err != nil {
		return 0, err
	}
	return len(payload), nil
}

func (w *serverRunnerLogWriter) Flush() error {
	if w == nil || w.masker == nil {
		return nil
	}
	if w.state != nil && w.state.err != nil {
		return w.state.err
	}
	masked := w.masker.Mask(nil, true)
	if len(masked) == 0 {
		return nil
	}
	return w.writeMasked(masked)
}

func (w *serverRunnerLogWriter) writeMasked(payload []byte) error {
	chunkSize := w.runner.effectiveMaxLogChunkBytes()
	for len(payload) > 0 {
		if err := w.ctx.Err(); err != nil {
			return err
		}
		if w.state.nextSeq >= maxServerRunnerLogChunksPerStream {
			return w.fail(w.logLimitError())
		}
		remaining := w.maxByte - w.state.writtenBytes
		if remaining <= 0 {
			return w.fail(w.logLimitError())
		}
		writeSize := chunkSize
		if writeSize > len(payload) {
			writeSize = len(payload)
		}
		if int64(writeSize) > remaining {
			writeSize = int(remaining)
		}
		w.state.nextSeq++
		chunk := payload[:writeSize]
		if err := w.runner.appendLogChunk(w.ctx, w.job, w.jobRun, w.stream, w.state.nextSeq, chunk); err != nil {
			return w.fail(err)
		}
		w.state.writtenBytes += int64(writeSize)
		payload = payload[writeSize:]
		if len(payload) > 0 && w.state.writtenBytes >= w.maxByte {
			return w.fail(w.logLimitError())
		}
	}
	return nil
}

func (w *serverRunnerLogWriter) logLimitError() error {
	return fmt.Errorf("automation job log output exceeded %d bytes or %d chunks for %s", w.maxByte, maxServerRunnerLogChunksPerStream, w.stream)
}

func (w *serverRunnerLogWriter) fail(err error) error {
	if w.state != nil && w.state.err == nil {
		w.state.err = err
	}
	if w.cancel != nil {
		w.cancel()
	}
	return err
}

func (r *ServerRunner) appendLogChunk(ctx context.Context, job automation.AutomationJob, jobRun automation.AutomationJobRun, stream string, sequence int64, payload []byte) error {
	if len(payload) == 0 {
		return nil
	}
	now := r.currentTime()
	objectKey, err := blob.LogChunkObjectKey(jobRun.WorkspaceID, jobRun.RunID, jobRun.ID, stream, sequence)
	if err != nil {
		return err
	}
	if err := r.blobStore.Put(ctx, objectKey, bytes.NewReader(payload), int64(len(payload))); err != nil {
		return err
	}
	sha := blob.SHA256Hex(payload)
	chunk := automation.AutomationLogChunk{
		ID:          blob.LogChunkID(jobRun.ID, stream, sequence),
		WorkspaceID: jobRun.WorkspaceID,
		RunID:       jobRun.RunID,
		JobRunID:    jobRun.ID,
		Stream:      stream,
		Sequence:    sequence,
		ObjectKey:   objectKey,
		SizeBytes:   int64(len(payload)),
		SHA256:      sha,
		ContentType: serverRunnerLogContentType,
		CreatedAt:   now,
	}
	event := automation.AutomationEvent{
		ID:               blob.LogEventID(jobRun.ID, stream, sequence),
		WorkspaceID:      jobRun.WorkspaceID,
		RunID:            jobRun.RunID,
		JobRunID:         jobRun.ID,
		EventType:        "job_log_appended",
		ActorPrincipalID: r.leaseOwner,
		ResourceType:     "automation_job",
		ResourceID:       strconv.FormatInt(job.ID, 10),
		Summary:          "Automation job log appended",
		MetadataJSON: jsonString(map[string]any{
			"stream":       stream,
			"sequence":     sequence,
			"object_key":   objectKey,
			"size":         len(payload),
			"sha256":       sha,
			"content_type": serverRunnerLogContentType,
		}),
		CreatedAt: now,
	}
	if _, _, err := r.store.CreateLeasedJobLogChunk(ctx, job.ID, r.leaseOwner, chunk, event, now); err != nil {
		_ = r.blobStore.Delete(ctx, objectKey)
		return err
	}
	return nil
}

func (r *ServerRunner) jobWorkDir(jobRun automation.AutomationJobRun) (string, error) {
	root, err := filepath.Abs(r.cfg.WorkDir)
	if err != nil {
		return "", err
	}
	target := filepath.Join(root, serverRunnerPathSegment(jobRun.WorkspaceID), serverRunnerPathSegment(jobRun.RunID), serverRunnerPathSegment(jobRun.ID))
	cleaned := filepath.Clean(target)
	if cleaned != root && !strings.HasPrefix(cleaned, root+string(os.PathSeparator)) {
		return "", fmt.Errorf("automation job workdir escapes runner root")
	}
	return cleaned, nil
}

func serverRunnerLeaseOwner(runnerID string) string {
	return "wfsr_" + blob.ShortHash(strings.TrimSpace(runnerID), 48)
}

func serverRunnerOutputJSON(runnerID string, steps []JobStepResult) string {
	return jsonString(map[string]any{"runner_id": strings.TrimSpace(runnerID), "steps": steps})
}

func serverRunnerPathSegment(value string) string {
	trimmed := strings.TrimSpace(value)
	if trimmed == "" {
		return "_"
	}
	if trimmed == "." || trimmed == ".." {
		return "h-" + blob.ShortHash(trimmed, 32)
	}
	for _, r := range trimmed {
		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' {
			continue
		}
		return "h-" + blob.ShortHash(trimmed, 32)
	}
	return trimmed
}

func (r *ServerRunner) currentTime() time.Time {
	if r.now != nil {
		return r.now().UTC()
	}
	return time.Now().UTC()
}

func (r *ServerRunner) log(ctx context.Context, level slog.Level, message string, args ...any) {
	logger := r.logger
	if logger == nil {
		logger = slog.Default()
	}
	logger.Log(ctx, level, message, args...)
}