package runtime

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/automation/blob"
	"github.com/tidbcloud/AgentTeam/internal/automation/dsl"
	"github.com/tidbcloud/AgentTeam/internal/store/automation"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

func TestServerRunnerExecutesJobWithExecutorAndWritesLogs(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 0, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Server runner", Jobs: map[string]dsl.JobSpec{
		"build": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{
				{ID: "one", Run: "printf 'hello-stdout\\n'"},
				{ID: "two", Run: "printf 'hello-stderr\\n' >&2"},
			},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_server_success", WorkspaceID: "ws-server", Name: "Server runner", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		if _, err := req.Stdout.Write([]byte("hello-stdout\n")); err != nil {
			return ExecuteJobResult{}, err
		}
		if _, err := req.Stderr.Write([]byte("hello-stderr\n")); err != nil {
			return ExecuteJobResult{}, err
		}
		return fakeSuccessfulJobResult(req), nil
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), MaxLogChunkBytes: 5, JobTimeout: time.Minute, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusSucceeded, finished.Status)
	require.JSONEq(t, `{"runner_id":"server-default","steps":[{"id":"one","index":0,"exit_code":0,"started_at":"2026-06-12T10:00:00Z","completed_at":"2026-06-12T10:00:00Z"},{"id":"two","index":1,"exit_code":0,"started_at":"2026-06-12T10:00:00Z","completed_at":"2026-06-12T10:00:00Z"}]}`, jobRuns[0].OutputJSON)

	stdout, err := store.ListLogChunks(ctx, "ws-server", run.ID, jobRuns[0].ID, "stdout", 1, 20)
	require.NoError(t, err)
	require.Greater(t, len(stdout), 1, "small MaxLogChunkBytes should chunk stdout")
	require.Equal(t, "hello-stdout\n", readServerRunnerLogChunks(t, ctx, blobStore, stdout))
	stderr, err := store.ListLogChunks(ctx, "ws-server", run.ID, jobRuns[0].ID, "stderr", 1, 20)
	require.NoError(t, err)
	require.Equal(t, "hello-stderr\n", readServerRunnerLogChunks(t, ctx, blobStore, stderr))
}

func TestServerRunnerSecretMaskerMasksAcrossWrites(t *testing.T) {
	masker := newServerRunnerSecretMasker(newServerRunnerSecretMatcher(map[string]string{"token": "super-secret"}))
	require.Empty(t, string(masker.Mask([]byte("super"), false)))
	require.Equal(t, "***", string(masker.Mask([]byte("-secret"), false)))
	require.Empty(t, string(masker.Mask(nil, true)))
}

func TestServerRunnerFailsUnsupportedUsesStepClosed(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 30, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Unsupported", Jobs: map[string]dsl.JobSpec{
		"checkout": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Uses: "actions/checkout@v4"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_unsupported", WorkspaceID: "ws-server", Name: "Unsupported", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: time.Minute}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusFailed, finished.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "uses is not supported")
}

func TestServerRunnerCheckoutMaterializesBeforeExecutorRunStep(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 45, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Checkout", Jobs: map[string]dsl.JobSpec{
		"test": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{
				{Uses: "agentteam/checkout", With: map[string]any{"path": "src"}},
				{Run: "test -f src/go.mod"},
			},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_server_checkout", WorkspaceID: "ws-server", Name: "Checkout", Status: automation.AutomationRunStatusRunning, RepositoryFullName: "pingcap/agentteam", CommitSHA: "0123456789abcdef0123456789abcdef01234567", CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	materializer := fakeCheckoutMaterializer{materialize: func(ctx context.Context, req CheckoutRequest) (CheckoutResult, error) {
		require.Equal(t, "pingcap/agentteam", req.RepositoryFullName)
		require.Equal(t, "0123456789abcdef0123456789abcdef01234567", req.CommitSHA)
		require.True(t, req.Clean)
		require.EqualValues(t, 1000, req.MaxBytes)
		require.Equal(t, 100, req.MaxFiles)
		require.NoError(t, os.MkdirAll(req.Destination, 0o755))
		require.NoError(t, os.WriteFile(filepath.Join(req.Destination, "go.mod"), []byte("module checkout.test\n"), 0o644))
		return CheckoutResult{Strategy: "test_archive", Files: 1, Bytes: 21}, nil
	}}
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		result := ExecuteJobResult{Steps: make([]JobStepResult, 0, len(req.Spec.Steps))}
		for idx, step := range req.Spec.Steps {
			started := req.Now().Format(time.RFC3339Nano)
			exitCode := 0
			var err error
			if strings.TrimSpace(step.Uses) != "" {
				exitCode, err = req.UsesStepHandler(ctx, UsesStepRequest{Step: step, Index: idx})
			} else {
				require.FileExists(t, filepath.Join(req.WorkDir, "src", "go.mod"))
			}
			result.Steps = append(result.Steps, JobStepResult{
				Index:       idx,
				ExitCode:    exitCode,
				StartedAt:   started,
				CompletedAt: req.Now().Format(time.RFC3339Nano),
			})
			if err != nil {
				return result, err
			}
		}
		return result, nil
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: time.Minute, CheckoutMaterializer: materializer, CheckoutMaxBytes: 1000, CheckoutMaxFiles: 100, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	require.JSONEq(t, `{"runner_id":"server-default","steps":[{"index":0,"exit_code":0,"started_at":"2026-06-12T10:45:00Z","completed_at":"2026-06-12T10:45:00Z"},{"index":1,"exit_code":0,"started_at":"2026-06-12T10:45:00Z","completed_at":"2026-06-12T10:45:00Z"}]}`, jobRuns[0].OutputJSON)
	events, err := store.ListEvents(ctx, "ws-server", run.ID, 1, 20)
	require.NoError(t, err)
	requireEventTypes(t, events, "run_started", "checkout_started", "checkout_succeeded", "job_succeeded", "run_reconciled")
	require.Contains(t, eventByType(t, events, "checkout_succeeded").MetadataJSON, `"strategy":"test_archive"`)
}

func TestServerRunnerCheckoutFailureFailsJobAndWritesEvent(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 50, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Checkout failure", Jobs: map[string]dsl.JobSpec{
		"test": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Uses: "agentteam/checkout"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_checkout_failure", WorkspaceID: "ws-server", Name: "Checkout failure", Status: automation.AutomationRunStatusRunning, RepositoryFullName: "pingcap/agentteam", CommitSHA: "fedcba9876543210fedcba9876543210fedcba98", CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{
		WorkDir:    t.TempDir(),
		JobTimeout: time.Minute,
		CheckoutMaterializer: fakeCheckoutMaterializer{materialize: func(context.Context, CheckoutRequest) (CheckoutResult, error) {
			return CheckoutResult{}, errors.New("archive unavailable")
		}},
	}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusFailed, finished.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "archive unavailable")
	events, err := store.ListEvents(ctx, "ws-server", run.ID, 1, 20)
	require.NoError(t, err)
	requireEventTypes(t, events, "run_started", "checkout_started", "checkout_failed", "job_failed", "run_failed")
	require.Contains(t, eventByType(t, events, "checkout_failed").MetadataJSON, `"error":"archive unavailable"`)
}

func TestServerRunnerPublishesStagedWikiPages(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 55, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Publish wiki", Jobs: map[string]dsl.JobSpec{
		"publish": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{
				{Run: "stage files"},
				{Uses: dsl.AgentTeamPublishWikiUse, With: map[string]any{
					dsl.AgentTeamPublishWikiWithRepository:  "pingcap/tidb.wiki",
					dsl.AgentTeamPublishWikiWithSourcePath:  "staged-wiki",
					dsl.AgentTeamPublishWikiWithDescription: "Generated wiki",
					dsl.AgentTeamPublishWikiWithLabel:       "automation-preset-generated",
					dsl.AgentTeamPublishWikiWithMaxPages:    "7",
				}},
			},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_server_publish_wiki", WorkspaceID: "ws-server", Name: "Publish wiki", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	publisher := &fakeWikiPublisher{publish: func(ctx context.Context, req WikiPublishRequest) (WikiPublishResult, error) {
		require.Equal(t, "ws-server", req.WorkspaceID)
		require.Equal(t, "wfr_server_publish_wiki", req.RunID)
		require.Equal(t, "pingcap/tidb.wiki", req.Repository)
		require.Equal(t, "Generated wiki", req.Description)
		require.Equal(t, "automation-preset-generated", req.Label)
		require.Equal(t, 7, req.MaxPages)
		require.FileExists(t, filepath.Join(req.SourcePath, "overview.md"))
		require.NotContains(t, req.SourcePath, "..")
		return WikiPublishResult{Repository: req.Repository, Pages: 1, Bytes: 12, IndexedCommitSHA: "wiki-sha", Reconciled: true}, nil
	}}
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		require.NoError(t, os.MkdirAll(filepath.Join(req.WorkDir, "staged-wiki"), 0o755))
		require.NoError(t, os.WriteFile(filepath.Join(req.WorkDir, "staged-wiki", "overview.md"), []byte("# Overview\n"), 0o644))
		return fakeExecuteJob(ctx, req)
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: time.Minute, WikiPublisher: publisher, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	require.Len(t, publisher.requests, 1)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	events, err := store.ListEvents(ctx, "ws-server", run.ID, 1, 20)
	require.NoError(t, err)
	requireEventTypes(t, events, "run_started", "wiki_publish_started", "wiki_publish_succeeded", "job_succeeded", "run_reconciled")
	require.Contains(t, eventByType(t, events, "wiki_publish_succeeded").MetadataJSON, `"pages":1`)
}

func TestServerRunnerPublishesStagedRepositoryTree(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 10, 58, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Publish tree", Jobs: map[string]dsl.JobSpec{
		"publish": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{
				{Run: "stage files"},
				{Uses: dsl.AgentTeamPublishTreeUse, With: map[string]any{
					dsl.AgentTeamPublishTreeWithRepository:  "pingcap/tidb.insights",
					dsl.AgentTeamPublishTreeWithSourcePath:  "mounted-insights",
					dsl.AgentTeamPublishTreeWithDescription: "Generated insights",
					dsl.AgentTeamPublishTreeWithMaxFiles:    "17",
				}},
			},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_server_publish_tree", WorkspaceID: "ws-server", Name: "Publish tree", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	publisher := &fakeTreePublisher{publish: func(ctx context.Context, req TreePublishRequest) (TreePublishResult, error) {
		require.Equal(t, "ws-server", req.WorkspaceID)
		require.Equal(t, "wfr_server_publish_tree", req.RunID)
		require.Equal(t, "pingcap/tidb.insights", req.Repository)
		require.Equal(t, "Generated insights", req.Description)
		require.Equal(t, 17, req.MaxFiles)
		require.FileExists(t, filepath.Join(req.SourcePath, "br", "run.sh.md"))
		require.NotContains(t, req.SourcePath, "..")
		return TreePublishResult{Repository: req.Repository, Files: 1, Bytes: 12, CommitSHA: "tree-sha"}, nil
	}}
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		require.NoError(t, os.MkdirAll(filepath.Join(req.WorkDir, "mounted-insights", "br"), 0o755))
		require.NoError(t, os.WriteFile(filepath.Join(req.WorkDir, "mounted-insights", "br", "run.sh.md"), []byte("# Run\n"), 0o644))
		return fakeExecuteJob(ctx, req)
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: time.Minute, TreePublisher: publisher, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	require.Len(t, publisher.requests, 1)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	events, err := store.ListEvents(ctx, "ws-server", run.ID, 1, 20)
	require.NoError(t, err)
	requireEventTypes(t, events, "run_started", "tree_publish_started", "tree_publish_succeeded", "job_succeeded", "run_reconciled")
	require.Contains(t, eventByType(t, events, "tree_publish_succeeded").MetadataJSON, `"files":1`)
	require.Contains(t, eventByType(t, events, "tree_publish_succeeded").MetadataJSON, `"commit_sha":"tree-sha"`)
}

func TestNewServerRunnerRejectsTooManyLabels(t *testing.T) {
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)

	_, err = NewServerRunner(store, blobStore, ServerRunnerConfig{
		WorkDir:         t.TempDir(),
		Labels:          []string{"one", "two"},
		MaxRunnerLabels: 1,
		Executor:        fakeJobExecutor{},
	}, nil, nil)
	require.Error(t, err)
	require.Contains(t, err.Error(), "labels exceed")
}

func TestServerRunnerJobTimeoutCancelsExecutorAndLogWriters(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 2, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Timeout log writer", Jobs: map[string]dsl.JobSpec{
		"slow": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "sleep 2"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_timeout_logs", WorkspaceID: "ws-server", Name: "Timeout log writer", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	executor := fakeJobExecutor{execute: func(execCtx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		_, ok := execCtx.Deadline()
		require.True(t, ok, "server runner should pass a deadline-bound execution context")
		select {
		case <-execCtx.Done():
		case <-time.After(time.Second):
			return ExecuteJobResult{}, fmt.Errorf("executor context was not timed out")
		}
		_, writeErr := req.Stdout.Write([]byte("late output"))
		require.ErrorIs(t, writeErr, context.DeadlineExceeded)
		return fakeSuccessfulJobResult(req), execCtx.Err()
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: 10 * time.Millisecond, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "automation job timed out after 10ms")
	require.Contains(t, jobRuns[0].ErrorJSON, `"failure_kind":"job_timeout"`)
	stdout, err := store.ListLogChunks(ctx, "ws-server", run.ID, jobRuns[0].ID, "stdout", 1, 20)
	require.NoError(t, err)
	require.Empty(t, stdout)
}

func TestServerRunnerPersistsExecutorFailureKind(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 3, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Executor failure", Jobs: map[string]dsl.JobSpec{
		"build": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "echo ok"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_executor_failure_kind", WorkspaceID: "ws-server", Name: "Executor failure", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	executor := fakeJobExecutor{execute: func(context.Context, ExecuteJobRequest) (ExecuteJobResult, error) {
		return ExecuteJobResult{}, ExecutorError{Kind: ExecutorFailureImageMissing, Err: fmt.Errorf("No such image: ubuntu:missing")}
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), JobTimeout: time.Minute, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "No such image")
	require.Contains(t, jobRuns[0].ErrorJSON, `"failure_kind":"image_missing"`)
	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusFailed, finished.Status)
	require.Contains(t, finished.ErrorJSON, `"failure_kind":"image_missing"`)
}

func TestServerRunnerCompletesJobWhenDockerCleanupFinishesAfterDeadline(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 4, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Slow cleanup", Jobs: map[string]dsl.JobSpec{
		"cleanup": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "echo done"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_slow_cleanup", WorkspaceID: "ws-server", Name: "Slow cleanup", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	dockerClient := &fakeDockerClient{
		attachOutputs: []fakeDockerAttachOutput{{stdout: "done\n"}},
		removeDelay:   25 * time.Millisecond,
	}
	executor, err := newDockerExecutorWithClient(dockerClient, DockerExecutorConfig{Images: map[string]string{"ubuntu-latest": "ubuntu:24.04"}})
	require.NoError(t, err)
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{
		WorkDir:    t.TempDir(),
		JobTimeout: 5 * time.Millisecond,
		Executor:   executor,
	}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusSucceeded, finished.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	require.True(t, dockerClient.removedValue())
}

func TestServerRunnerLogsCleanupErrorAfterCanceledJob(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 5, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Cancel", Jobs: map[string]dsl.JobSpec{
		"cancel": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "sleep 60"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_cancel_cleanup", WorkspaceID: "ws-server", Name: "Cancel", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	var queueJobID int64
	executor := fakeJobExecutor{execute: func(execCtx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		queueJobID = req.Job.QueueJobID
		worker := NewWorker(store, func() time.Time { return now.Add(time.Second) }, nil)
		require.NoError(t, worker.CancelRun(ctx, req.Job.WorkspaceID, req.Job.RunID, "user-1", "stop requested"))
		select {
		case <-execCtx.Done():
			return ExecuteJobResult{}, markExecutorCleanupError(fmt.Errorf("remove denied"))
		case <-time.After(2 * time.Second):
			return ExecuteJobResult{}, fmt.Errorf("executor context was not canceled")
		}
	}}
	var logs bytes.Buffer
	logger := slog.New(slog.NewTextHandler(&logs, nil))
	runner, err := NewServerRunner(store, blobStore, ServerRunnerConfig{
		WorkDir:    t.TempDir(),
		JobTimeout: time.Minute,
		Executor:   executor,
	}, func() time.Time { return now }, logger)
	require.NoError(t, err)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	finished, err := store.GetRun(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusCanceled, finished.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusCanceled, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "stop requested")
	require.NotContains(t, jobRuns[0].ErrorJSON, "remove denied")
	queueJob, err := store.GetJob(ctx, queueJobID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobStatusCanceled, queueJob.Status)
	require.Contains(t, logs.String(), "automation server runner canceled job cleanup failed")
	require.Contains(t, logs.String(), "remove denied")
}

func TestServerRunnerLogsCleanupErrorAfterTimedOutJob(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 10, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Timeout cleanup", Jobs: map[string]dsl.JobSpec{
		"timeout": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "sleep 60"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_timeout_cleanup", WorkspaceID: "ws-server", Name: "Timeout cleanup", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	dockerClient := &fakeDockerClient{
		attachOutputs: []fakeDockerAttachOutput{{blockUntilContextDone: true}},
		removeErr:     fmt.Errorf("remove denied"),
	}
	executor, err := newDockerExecutorWithClient(dockerClient, DockerExecutorConfig{Images: map[string]string{"ubuntu-latest": "ubuntu:24.04"}})
	require.NoError(t, err)
	var logs bytes.Buffer
	logger := slog.New(slog.NewTextHandler(&logs, nil))
	runner, err := NewServerRunner(store, blobStore, ServerRunnerConfig{
		WorkDir:    t.TempDir(),
		JobTimeout: 5 * time.Millisecond,
		Executor:   executor,
	}, func() time.Time { return now }, logger)
	require.NoError(t, err)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "automation job timed out after 5ms")
	require.Contains(t, jobRuns[0].ErrorJSON, `"failure_kind":"job_timeout"`)
	require.NotContains(t, jobRuns[0].ErrorJSON, "remove denied")
	require.Contains(t, logs.String(), "automation server runner timed out job cleanup failed")
	require.Contains(t, logs.String(), "remove denied")
	require.True(t, dockerClient.removedValue())
}

func TestServerRunnerFailsJobWhenLogOutputExceedsBudget(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 15, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Log cap", Jobs: map[string]dsl.JobSpec{
		"noisy": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "i=0; while [ $i -lt 1100 ]; do printf x; i=$((i+1)); done"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_log_cap", WorkspaceID: "ws-server", Name: "Log cap", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		_, err := req.Stdout.Write([]byte(strings.Repeat("x", 1100)))
		return fakeSuccessfulJobResult(req), err
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), MaxLogChunkBytes: 4, JobTimeout: time.Minute, Executor: executor}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "log output exceeded")
	require.Contains(t, jobRuns[0].ErrorJSON, `"failure_kind":"log_write_failed"`)
	stdout, err := store.ListLogChunks(ctx, "ws-server", run.ID, jobRuns[0].ID, "stdout", 1, maxServerRunnerLogChunksPerStream+1)
	require.NoError(t, err)
	require.LessOrEqual(t, len(stdout), maxServerRunnerLogChunksPerStream)
	require.LessOrEqual(t, len(readServerRunnerLogChunks(t, ctx, blobStore, stdout)), int(runner.effectiveMaxLogBytesPerStream()))
}

func TestServerRunnerReportsDockerLogBudgetExceeded(t *testing.T) {
	ctx := t.Context()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 20, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{Name: "Docker log cap", Jobs: map[string]dsl.JobSpec{
		"noisy": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "printf many"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_docker_log_cap", WorkspaceID: "ws-server", Name: "Docker log cap", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	dockerClient := &fakeDockerClient{attachOutputs: []fakeDockerAttachOutput{{stdout: strings.Repeat("x", 1100)}}}
	executor, err := newDockerExecutorWithClient(dockerClient, DockerExecutorConfig{Images: map[string]string{"ubuntu-latest": "ubuntu:24.04"}})
	require.NoError(t, err)
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{
		WorkDir:          t.TempDir(),
		MaxLogChunkBytes: 4,
		JobTimeout:       time.Minute,
		Executor:         executor,
	}, now)

	require.NoError(t, runner.processWorkspace(ctx, "ws-server"))

	jobRuns, err := store.ListJobRuns(ctx, "ws-server", run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automation.AutomationJobRunStatusFailed, jobRuns[0].Status)
	require.Contains(t, jobRuns[0].ErrorJSON, "log output exceeded")
	require.Contains(t, jobRuns[0].ErrorJSON, `"failure_kind":"log_write_failed"`)
	require.NotContains(t, jobRuns[0].ErrorJSON, "context canceled")
	require.True(t, dockerClient.removedValue())
}

func TestServerRunnerConcurrencyOneDoesNotRunJobsInParallel(t *testing.T) {
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	store := newServerRunnerTestStore(t)
	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	now := time.Date(2026, 6, 12, 11, 30, 0, 0, time.UTC)
	markerDir := t.TempDir()
	activePath := filepath.Join(markerDir, "active")
	overlapPath := filepath.Join(markerDir, "overlap")
	spec := dsl.DefinitionSpec{Name: "Concurrency", Jobs: map[string]dsl.JobSpec{
		"first":  {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "first"}}},
		"second": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Run: "second"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_server_concurrency", WorkspaceID: "ws-server", Name: "Concurrency", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	createServerRunnerRuntime(t, ctx, store, run, spec, now)
	executor := fakeJobExecutor{execute: func(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
		if _, err := os.Stat(activePath); err == nil {
			if err := os.WriteFile(overlapPath, []byte("overlap"), 0o644); err != nil {
				return ExecuteJobResult{}, err
			}
		}
		if err := os.WriteFile(activePath, []byte(req.Job.JobRunID), 0o644); err != nil {
			return ExecuteJobResult{}, err
		}
		defer func() { _ = os.Remove(activePath) }()
		select {
		case <-ctx.Done():
			return ExecuteJobResult{}, ctx.Err()
		case <-time.After(150 * time.Millisecond):
		}
		return fakeSuccessfulJobResult(req), nil
	}}
	runner := newServerRunnerForTest(t, store, blobStore, ServerRunnerConfig{WorkDir: t.TempDir(), Concurrency: 1, PollInterval: 10 * time.Millisecond, JobTimeout: time.Minute, Executor: executor}, now)

	go runner.Run(ctx)
	runner.SignalWorkspace("ws-server")
	runner.SignalWorkspace("ws-server")
	require.Eventually(t, func() bool {
		finished, err := store.GetRun(ctx, "ws-server", run.ID)
		return err == nil && finished.Status == automation.AutomationRunStatusSucceeded
	}, 3*time.Second, 20*time.Millisecond)
	cancel()

	require.NoFileExists(t, overlapPath)
}

func newServerRunnerTestStore(t *testing.T) *automation.Store {
	t.Helper()
	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(&automation.AutomationRun{}, &automation.AutomationRepoStatusOutbox{}, &automation.AutomationRunSlot{}, &automation.AutomationJobRun{}, &automation.AutomationEvent{}, &automation.AutomationJob{}, &automation.AutomationWait{}, &automation.AutomationLogChunk{}, &automation.AutomationArtifact{}, &automation.AutomationSecret{}))
	return automation.New(db)
}

func newServerRunnerForTest(t *testing.T, store *automation.Store, blobStore blob.Store, cfg ServerRunnerConfig, now time.Time) *ServerRunner {
	t.Helper()
	if cfg.Executor == nil {
		cfg.Executor = fakeJobExecutor{}
	}
	runner, err := NewServerRunner(store, blobStore, cfg, func() time.Time { return now }, nil)
	require.NoError(t, err)
	return runner
}

func createServerRunnerRuntime(t *testing.T, ctx context.Context, store *automation.Store, run automation.AutomationRun, spec dsl.DefinitionSpec, now time.Time) {
	t.Helper()
	nextID := 0
	runtime, err := BuildInitialRuntime(run, spec, now, func(prefix string) (string, error) {
		nextID++
		return fmt.Sprintf("%s%016d", prefix, nextID), nil
	})
	require.NoError(t, err)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs, runtime.Waits)
	require.NoError(t, err)
}

func readServerRunnerLogChunks(t *testing.T, ctx context.Context, blobStore blob.Store, chunks []automation.AutomationLogChunk) string {
	t.Helper()
	sort.Slice(chunks, func(i int, j int) bool { return chunks[i].Sequence < chunks[j].Sequence })
	var out strings.Builder
	for _, chunk := range chunks {
		body, err := blobStore.Get(ctx, chunk.ObjectKey)
		require.NoError(t, err)
		payload, err := io.ReadAll(body)
		require.NoError(t, err)
		require.NoError(t, body.Close())
		out.Write(payload)
	}
	return out.String()
}

type fakeJobExecutor struct {
	execute func(context.Context, ExecuteJobRequest) (ExecuteJobResult, error)
}

func (e fakeJobExecutor) ExecuteJob(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
	if e.execute != nil {
		return e.execute(ctx, req)
	}
	return fakeExecuteJob(ctx, req)
}

func fakeExecuteJob(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
	result := ExecuteJobResult{Steps: make([]JobStepResult, 0, len(req.Spec.Steps))}
	for idx, step := range req.Spec.Steps {
		now := req.Now().Format(time.RFC3339Nano)
		exitCode := 0
		var err error
		if strings.TrimSpace(step.Uses) != "" {
			if req.UsesStepHandler == nil {
				exitCode = -1
				err = fmt.Errorf("automation step uses is not supported by the configured executor")
			} else {
				exitCode, err = req.UsesStepHandler(ctx, UsesStepRequest{Step: step, Index: idx})
			}
		}
		result.Steps = append(result.Steps, JobStepResult{
			ID:          strings.TrimSpace(step.ID),
			Name:        strings.TrimSpace(step.Name),
			Index:       idx,
			ExitCode:    exitCode,
			StartedAt:   now,
			CompletedAt: now,
		})
		if err != nil {
			return result, err
		}
	}
	return result, nil
}

func fakeSuccessfulJobResult(req ExecuteJobRequest) ExecuteJobResult {
	steps := make([]JobStepResult, 0, len(req.Spec.Steps))
	for idx, step := range req.Spec.Steps {
		now := req.Now().Format(time.RFC3339Nano)
		steps = append(steps, JobStepResult{
			ID:          strings.TrimSpace(step.ID),
			Name:        strings.TrimSpace(step.Name),
			Index:       idx,
			ExitCode:    0,
			StartedAt:   now,
			CompletedAt: now,
		})
	}
	return ExecuteJobResult{Steps: steps}
}

type fakeCheckoutMaterializer struct {
	materialize func(context.Context, CheckoutRequest) (CheckoutResult, error)
}

func (m fakeCheckoutMaterializer) Materialize(ctx context.Context, req CheckoutRequest) (CheckoutResult, error) {
	return m.materialize(ctx, req)
}

type fakeWikiPublisher struct {
	requests []WikiPublishRequest
	publish  func(context.Context, WikiPublishRequest) (WikiPublishResult, error)
}

func (p *fakeWikiPublisher) PublishWikiPages(ctx context.Context, req WikiPublishRequest) (WikiPublishResult, error) {
	p.requests = append(p.requests, req)
	if p.publish != nil {
		return p.publish(ctx, req)
	}
	return WikiPublishResult{Repository: req.Repository, Reconciled: true}, nil
}

type fakeTreePublisher struct {
	requests []TreePublishRequest
	publish  func(context.Context, TreePublishRequest) (TreePublishResult, error)
}

func (p *fakeTreePublisher) PublishRepositoryTree(ctx context.Context, req TreePublishRequest) (TreePublishResult, error) {
	p.requests = append(p.requests, req)
	if p.publish != nil {
		return p.publish(ctx, req)
	}
	return TreePublishResult{Repository: req.Repository}, nil
}

func requireEventTypes(t *testing.T, events []automation.AutomationEvent, want ...string) {
	t.Helper()
	got := make([]string, 0, len(events))
	for _, event := range events {
		got = append(got, event.EventType)
	}
	require.ElementsMatch(t, want, got)
}

func eventByType(t *testing.T, events []automation.AutomationEvent, eventType string) automation.AutomationEvent {
	t.Helper()
	for _, event := range events {
		if event.EventType == eventType {
			return event
		}
	}
	t.Fatalf("eventByType(%s) not found in %#v", eventType, events)
	return automation.AutomationEvent{}
}