package runtime

import (
	"context"
	"encoding/json"
	"fmt"
	"testing"
	"time"

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

func TestAutomationStateMachine(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 17, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "000000000000000" + string(rune('0'+ids)), nil
	}

	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{
		"plan":   {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{ID: "first", Run: "echo checkout"}, {ID: "second", Run: "echo ok"}}},
		"review": {Needs: []string{"plan"}, RunsOn: map[string]any{"group": "trusted", "labels": []any{"self-hosted", "macos"}}, Steps: []dsl.StepSpec{{ID: "review", Run: "echo review"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_runtime", WorkspaceID: "ws-1", DefinitionID: "wfd_1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	require.Len(t, runtime.JobRuns, 2)
	require.Equal(t, "plan", runtime.JobRuns[0].JobID)
	require.Equal(t, "github_actions_job", runtime.JobRuns[0].JobType)
	require.Equal(t, `"ubuntu-latest"`, runtime.JobRuns[0].RunsOnJSON)
	require.Equal(t, automation.AutomationJobRunStatusQueued, runtime.JobRuns[0].Status)
	require.Equal(t, "review", runtime.JobRuns[1].JobID)
	require.JSONEq(t, `{"group":"trusted","labels":["self-hosted","macos"]}`, runtime.JobRuns[1].RunsOnJSON)
	require.Equal(t, automation.AutomationJobRunStatusPending, runtime.JobRuns[1].Status)
	require.Len(t, runtime.Jobs, 1)
	require.Equal(t, `["ubuntu-latest"]`, runtime.Jobs[0].LabelsJSON)
	created, err := store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusRunning, created.Status)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Minute) }, nil)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Labels: []string{"ubuntu-latest"}, Now: now.Add(time.Second)})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	require.NoError(t, worker.CompleteLeasedJob(ctx, claimed[0], "runner-1", `{"ok":true}`))
	jobRuns, err := store.ListJobRuns(ctx, "ws-1", "wfr_runtime")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status)
	require.Equal(t, automation.AutomationJobRunStatusQueued, jobRuns[1].Status)

	claimed, err = store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, RunnerGroup: "trusted", Labels: []string{"self-hosted", "macos"}, Now: now.Add(2 * time.Minute)})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	require.Equal(t, "trusted", claimed[0].RunnerGroup)
	require.Equal(t, `["self-hosted","macos"]`, claimed[0].LabelsJSON)
	require.NoError(t, worker.CompleteLeasedJob(ctx, claimed[0], "runner-1", `{"ok":true}`))
	finished, err := store.GetRun(ctx, "ws-1", "wfr_runtime")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusSucceeded, finished.Status)
	require.NotNil(t, finished.CompletedAt)
}

func TestAutomationCheckoutJobsRequireServerManagedRunnerGroup(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%scheckoutgroup%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Checkout", Jobs: map[string]dsl.JobSpec{
		"test": {RunsOn: "ubuntu-latest", Steps: []dsl.StepSpec{{Uses: "agentteam/checkout"}, {Run: "test -f go.mod"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_checkout_group", WorkspaceID: "ws-1", Name: "Checkout", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	require.Len(t, runtime.Jobs, 1)
	require.Equal(t, ServerManagedRunnerGroup, runtime.Jobs[0].RunnerGroup)
	require.Equal(t, `["ubuntu-latest"]`, runtime.Jobs[0].LabelsJSON)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)

	externalClaim, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "external-runner", LeaseDuration: time.Minute, Limit: 1, Labels: []string{"ubuntu-latest"}, Now: now.Add(time.Second)})
	require.NoError(t, err)
	require.Empty(t, externalClaim)

	serverClaim, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "server-runner", LeaseDuration: time.Minute, Limit: 1, RunnerGroup: ServerManagedRunnerGroup, Labels: []string{"ubuntu-latest"}, Now: now.Add(2 * time.Second)})
	require.NoError(t, err)
	require.Len(t, serverClaim, 1)
	require.Equal(t, runtime.JobRuns[0].ID, serverClaim[0].JobRunID)
}

func TestAutomationPublishWikiJobsRequireServerManagedRunnerGroup(t *testing.T) {
	now := time.Date(2026, 6, 12, 12, 10, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%spublishgroup%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Publish wiki", Jobs: map[string]dsl.JobSpec{
		"publish": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{{
				Uses: dsl.AgentTeamPublishWikiUse,
				With: map[string]any{
					dsl.AgentTeamPublishWikiWithRepository: "pingcap/tidb.wiki",
					dsl.AgentTeamPublishWikiWithSourcePath: "staged-wiki",
				},
			}},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_publish_group", WorkspaceID: "ws-1", Name: "Publish wiki", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}

	runtime, err := BuildInitialRuntime(run, spec, now, newID)

	require.NoError(t, err)
	require.Len(t, runtime.Jobs, 1)
	require.Equal(t, ServerManagedRunnerGroup, runtime.Jobs[0].RunnerGroup)
	require.Equal(t, `["ubuntu-latest"]`, runtime.Jobs[0].LabelsJSON)
}

func TestAutomationPublishTreeJobsRequireServerManagedRunnerGroup(t *testing.T) {
	now := time.Date(2026, 6, 12, 12, 11, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%spublishtreegroup%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Publish tree", Jobs: map[string]dsl.JobSpec{
		"publish": {
			RunsOn: "ubuntu-latest",
			Steps: []dsl.StepSpec{{
				Uses: dsl.AgentTeamPublishTreeUse,
				With: map[string]any{
					dsl.AgentTeamPublishTreeWithRepository: "pingcap/tidb.insights",
					dsl.AgentTeamPublishTreeWithSourcePath: "mounted-insights",
				},
			}},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_publish_tree_group", WorkspaceID: "ws-1", Name: "Publish tree", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}

	runtime, err := BuildInitialRuntime(run, spec, now, newID)

	require.NoError(t, err)
	require.Len(t, runtime.Jobs, 1)
	require.Equal(t, ServerManagedRunnerGroup, runtime.Jobs[0].RunnerGroup)
	require.Equal(t, `["ubuntu-latest"]`, runtime.Jobs[0].LabelsJSON)
}

func TestAutomationAuthoredReservedRunnerGroupIsNotPersisted(t *testing.T) {
	now := time.Date(2026, 6, 12, 12, 15, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%sreservedgroup%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Reserved", Jobs: map[string]dsl.JobSpec{
		"test": {
			RunsOn: map[string]any{"group": ServerManagedRunnerGroup, "labels": []any{"ubuntu-latest"}},
			Steps:  []dsl.StepSpec{{Run: "echo ordinary"}},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_reserved_group", WorkspaceID: "ws-1", Name: "Reserved", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}

	runtime, err := BuildInitialRuntime(run, spec, now, newID)

	require.NoError(t, err)
	require.Len(t, runtime.Jobs, 1)
	require.Empty(t, runtime.Jobs[0].RunnerGroup)
	require.Equal(t, `["ubuntu-latest"]`, runtime.Jobs[0].LabelsJSON)
}

func TestBuildInitialRuntimePreservesEnvReferences(t *testing.T) {
	now := time.Date(2026, 6, 14, 11, 0, 0, 0, time.UTC)
	spec := dsl.DefinitionSpec{
		Name: "Env runtime",
		Env: map[string]any{
			"NODE_ENV": "production",
			"SHARED":   "automation",
			"GLOBAL":   "${{ inputs.global }}",
		},
		Jobs: map[string]dsl.JobSpec{
			"deploy": {
				Env: map[string]any{
					"SHARED":   "job",
					"GH_TOKEN": "${{ secrets.github_publish_token }}",
				},
				Steps: []dsl.StepSpec{{
					Run: "echo deploy",
					Env: map[string]any{"SLACK_TOKEN": "${{ secrets.deploy_slack_token }}"},
				}},
			},
		},
	}
	run := automation.AutomationRun{ID: "wfr_env_runtime", WorkspaceID: "ws-1", Name: "Env runtime", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, sequentialRuntimeIDs())
	require.NoError(t, err)
	require.Len(t, runtime.JobRuns, 1)

	var job dsl.JobSpec
	require.NoError(t, json.Unmarshal([]byte(runtime.JobRuns[0].InputJSON), &job))
	require.Equal(t, "production", job.Env["NODE_ENV"])
	require.Equal(t, "job", job.Env["SHARED"])
	require.Equal(t, "${{ inputs.global }}", job.Env["GLOBAL"])
	require.Equal(t, "${{ secrets.github_publish_token }}", job.Env["GH_TOKEN"])
	require.Equal(t, "${{ secrets.deploy_slack_token }}", job.Steps[0].Env["SLACK_TOKEN"])
}

func sequentialRuntimeIDs() IDFactory {
	ids := 0
	return func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%stest%03d", prefix, ids), nil
	}
}

func TestAutomationWorkerRejectsStaleLeaseCommit(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 18, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "stale000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{"plan": {Steps: []dsl.StepSpec{{ID: "first", Run: "echo stale"}}}}}
	run := automation.AutomationRun{ID: "wfr_stale", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "stale-worker", LeaseDuration: time.Second, Limit: 1, Now: now})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	_, err = store.RecoverExpiredJobs(ctx, now.Add(2*time.Second), 10)
	require.NoError(t, err)

	worker := NewWorker(store, func() time.Time { return now.Add(3 * time.Second) }, nil)
	require.Error(t, worker.CompleteLeasedJob(ctx, claimed[0], "stale-worker", `{"ok":true}`))
	jobRuns, err := store.ListJobRuns(ctx, "ws-1", "wfr_stale")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobRunStatusQueued, jobRuns[0].Status)
}

func TestAutomationWorkerCompleteCallbackIsIdempotent(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 18, 30, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "idem0000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{"plan": {Steps: []dsl.StepSpec{{ID: "first", Run: "echo idempotent"}}}}}
	run := automation.AutomationRun{ID: "wfr_idempotent_success", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Second) }, nil)
	require.NoError(t, worker.CompleteLeasedJob(ctx, claimed[0], "runner-1", `{"ok":true}`))
	require.NoError(t, worker.CompleteLeasedJob(ctx, claimed[0], "runner-1", `{"ok":true}`))
	events, err := store.ListEvents(ctx, "ws-1", "wfr_idempotent_success", 1, 20)
	require.NoError(t, err)
	var succeededEvents, reconciledEvents int
	for _, event := range events {
		if event.EventType == "job_succeeded" {
			succeededEvents++
		}
		if event.EventType == "run_reconciled" {
			reconciledEvents++
		}
	}
	require.Equal(t, 1, succeededEvents)
	require.Equal(t, 1, reconciledEvents)
}

func TestAutomationWorkerFailureRetryIsBackoffAndIdempotent(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 18, 45, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "retry000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{"plan": {Steps: []dsl.StepSpec{{ID: "first", Run: "echo ok"}}}}}
	run := automation.AutomationRun{ID: "wfr_retry", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	runtime.JobRuns[0].MaxAttempts = 3
	runtime.Jobs[0].MaxAttempts = 3
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Second) }, nil)
	require.NoError(t, worker.FailLeasedJob(ctx, claimed[0], "runner-1", "runner failed", ""))
	require.NoError(t, worker.FailLeasedJob(ctx, claimed[0], "runner-1", "runner failed", ""))
	jobRun, err := store.GetJobRun(ctx, "ws-1", runtime.JobRuns[0].ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobRunStatusQueued, jobRun.Status)
	require.Nil(t, jobRun.CompletedAt)
	claimedTooEarly, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-2", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(2 * time.Second)})
	require.NoError(t, err)
	require.Empty(t, claimedTooEarly)
	claimedRetry, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-2", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(31 * time.Second)})
	require.NoError(t, err)
	require.Len(t, claimedRetry, 1)
	require.Equal(t, 2, claimedRetry[0].Attempt)
	events, err := store.ListEvents(ctx, "ws-1", "wfr_retry", 1, 20)
	require.NoError(t, err)
	var retryEvents int
	for _, event := range events {
		if event.EventType == "job_retry_scheduled" {
			retryEvents++
		}
	}
	require.Equal(t, 1, retryEvents)
}

func TestAutomationWorkerExhaustedFailureFailsRun(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 19, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "fail0000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{"plan": {Steps: []dsl.StepSpec{{ID: "http", Run: "echo failed"}}}}}
	run := automation.AutomationRun{ID: "wfr_failed", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	runtime.JobRuns[0].MaxAttempts = 1
	runtime.Jobs[0].MaxAttempts = 1
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Minute) }, nil)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(time.Second)})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	staleJobRun, err := store.GetJobRun(ctx, "ws-1", runtime.JobRuns[0].ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobRunStatusQueued, staleJobRun.Status)
	require.NoError(t, worker.FailLeasedJob(ctx, claimed[0], "runner-1", "runner failed", ""))
	require.True(t, worker.failureAlreadyApplied(ctx, claimed[0]), "stale concurrent loser must re-read current failed job run")
	require.NoError(t, worker.FailLeasedJob(ctx, claimed[0], "runner-1", "runner failed", ""))
	failed, err := store.GetRun(ctx, "ws-1", "wfr_failed")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusFailed, failed.Status)
	require.NotEmpty(t, failed.ErrorJSON)
	events, err := store.ListEvents(ctx, "ws-1", "wfr_failed", 1, 20)
	require.NoError(t, err)
	var jobFailedEvents, runFailedEvents int
	for _, event := range events {
		if event.EventType == "job_failed" {
			jobFailedEvents++
		}
		if event.EventType == "run_failed" {
			runFailedEvents++
		}
	}
	require.Equal(t, 1, jobFailedEvents)
	require.Equal(t, 1, runFailedEvents)
}

func TestAutomationWorkerExhaustedFailureBlocksDownstream(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 19, 30, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "block000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{
		"build":  {Steps: []dsl.StepSpec{{ID: "build", Run: "make"}}},
		"test":   {Needs: []string{"build"}, Steps: []dsl.StepSpec{{ID: "test", Run: "go test"}}},
		"deploy": {Needs: []string{"test"}, Steps: []dsl.StepSpec{{ID: "deploy", Run: "make deploy"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_blocked_dependency", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	runtime.JobRuns[0].MaxAttempts = 1
	runtime.Jobs[0].MaxAttempts = 1
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(time.Second)})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Minute) }, nil)
	require.NoError(t, worker.FailLeasedJob(ctx, claimed[0], "runner-1", "runner failed", ""))

	run, err = store.GetRun(ctx, "ws-1", "wfr_blocked_dependency")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusFailed, run.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-1", "wfr_blocked_dependency")
	require.NoError(t, err)
	statusByJobID := map[string]string{}
	for _, jobRun := range jobRuns {
		statusByJobID[jobRun.JobID] = jobRun.Status
	}
	require.Equal(t, automation.AutomationJobRunStatusFailed, statusByJobID["build"])
	require.Equal(t, automation.AutomationJobRunStatusBlocked, statusByJobID["test"])
	require.Equal(t, automation.AutomationJobRunStatusBlocked, statusByJobID["deploy"])
	claimedAfterFailure, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-2", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(2 * time.Minute)})
	require.NoError(t, err)
	require.Empty(t, claimedAfterFailure)
	events, err := store.ListEvents(ctx, "ws-1", "wfr_blocked_dependency", 1, 20)
	require.NoError(t, err)
	var blockedEvents, runFailedEvents int
	for _, event := range events {
		if event.EventType == "job_blocked" {
			blockedEvents++
		}
		if event.EventType == "run_failed" {
			runFailedEvents++
		}
	}
	require.Equal(t, 2, blockedEvents)
	require.Equal(t, 1, runFailedEvents)
}

func TestAutomationWorkerCancelRunAggregatesState(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 20, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "cancel0000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Runtime", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{
		"build": {Steps: []dsl.StepSpec{{ID: "build", Run: "make"}}},
		"test":  {Needs: []string{"build"}, Steps: []dsl.StepSpec{{ID: "test", Run: "go test"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_cancel", WorkspaceID: "ws-1", Name: "Runtime", Status: automation.AutomationRunStatusRunning, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs)
	require.NoError(t, err)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: runtime.Jobs[0].ShardBucket, LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	worker := NewWorker(store, func() time.Time { return now.Add(time.Second) }, nil)
	require.NoError(t, worker.CancelRun(ctx, "ws-1", "wfr_cancel", "user-1", "no longer needed"))
	canceled, err := store.GetRun(ctx, "ws-1", "wfr_cancel")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationRunStatusCanceled, canceled.Status)
	jobRuns, err := store.ListJobRuns(ctx, "ws-1", "wfr_cancel")
	require.NoError(t, err)
	for _, jobRun := range jobRuns {
		require.Equal(t, automation.AutomationJobRunStatusCanceled, jobRun.Status)
	}
	job, err := store.GetJob(ctx, claimed[0].ID)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobStatusCanceled, job.Status)
	require.Error(t, worker.CompleteLeasedJob(ctx, claimed[0], "runner-1", `{"ok":true}`))
	jobRuns, err = store.ListJobRuns(ctx, "ws-1", "wfr_cancel")
	require.NoError(t, err)
	for _, jobRun := range jobRuns {
		require.Equal(t, automation.AutomationJobRunStatusCanceled, jobRun.Status)
	}
	events, err := store.ListEvents(ctx, "ws-1", "wfr_cancel", 1, 20)
	require.NoError(t, err)
	var sawCancel bool
	for _, event := range events {
		sawCancel = sawCancel || event.EventType == "run_canceled"
	}
	require.True(t, sawCancel)
}

func TestAutomationAgentTeamNativeWaitResumeSchedulesDownstream(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 21, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "wait0000000000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Waits", On: map[string]any{"manual": map[string]any{}}, Jobs: map[string]dsl.JobSpec{
		"approve": {Steps: []dsl.StepSpec{{ID: "approval", Uses: "agentteam/wait-event", With: map[string]any{"event_type": "automation_event_received", "resource_type": "event", "resource_id": "ship-approval"}}}},
		"deploy":  {Needs: []string{"approve"}, Steps: []dsl.StepSpec{{ID: "deploy", Run: "make deploy"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_wait", WorkspaceID: "ws-1", Name: "Waits", Status: automation.AutomationRunStatusWaiting, CreatedByPrincipalID: "user-1", OwnerPrincipalID: "user-1", CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	require.Len(t, runtime.Jobs, 0)
	require.Len(t, runtime.Waits, 1)
	require.Equal(t, automation.AutomationWaitKindWaitEvent, runtime.Waits[0].WaitKind)
	require.Equal(t, "event", runtime.Waits[0].ResourceType)
	require.Equal(t, "ship-approval", runtime.Waits[0].ResourceID)
	require.Equal(t, automation.AutomationJobRunStatusWaiting, runtime.JobRuns[0].Status)
	_, err = store.CreateRunWithRuntime(ctx, run, runtime.JobRuns, runtime.Events, runtime.Jobs, runtime.Waits)
	require.NoError(t, err)

	eventer := NewEventer(store, func() time.Time { return now.Add(time.Minute) })
	wait, err := eventer.ResumeWait(ctx, "ws-1", runtime.Waits[0].ID, "approver-1", `{"approved":true}`)
	require.NoError(t, err)
	wait, err = eventer.ResumeWait(ctx, "ws-1", runtime.Waits[0].ID, "approver-1", `{"approved":true}`)
	require.NoError(t, err)
	require.Equal(t, automation.AutomationWaitStatusResolved, wait.Status)
	require.Nil(t, wait.ActiveKey)
	jobRuns, err := store.ListJobRuns(ctx, "ws-1", "wfr_wait")
	require.NoError(t, err)
	require.Equal(t, automation.AutomationJobRunStatusSucceeded, jobRuns[0].Status)
	require.Equal(t, automation.AutomationJobRunStatusQueued, jobRuns[1].Status)
	claimed, err := store.ClaimJobs(ctx, automation.ClaimJobsRequest{WorkspaceID: "ws-1", QueueName: DefaultQueueName, ShardBucket: shardForRun("wfr_wait"), LeaseOwner: "runner-1", LeaseDuration: time.Minute, Limit: 1, Now: now.Add(2 * time.Minute)})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	require.Equal(t, jobRuns[1].ID, claimed[0].JobRunID)
	events, err := store.ListEvents(ctx, "ws-1", "wfr_wait", 1, 20)
	require.NoError(t, err)
	var waitResolvedEvents int
	for _, event := range events {
		if event.EventType == "wait_resolved" {
			waitResolvedEvents++
		}
	}
	require.Equal(t, 1, waitResolvedEvents)
}

func TestAutomationAgentTeamNativeWaitMalformedJobsAreNotClassifiedAsWaits(t *testing.T) {
	now := time.Date(2026, 6, 2, 21, 15, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return prefix + "malformed00000" + string(rune('0'+ids)), nil
	}
	spec := dsl.DefinitionSpec{Name: "Malformed waits", Jobs: map[string]dsl.JobSpec{
		"mixed": {
			Steps: []dsl.StepSpec{
				{Uses: "agentteam/wait-event"},
				{Run: "echo should not be skipped"},
			},
		},
		"conditional": {
			If:    "${{ inputs.ready }}",
			Steps: []dsl.StepSpec{{Uses: "agentteam/wait-event"}},
		},
	}}
	run := automation.AutomationRun{ID: "wfr_malformed_wait", WorkspaceID: "ws-1", Name: "Malformed waits", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	require.Len(t, runtime.Waits, 0)
	require.Len(t, runtime.Jobs, 2)
	for _, jobRun := range runtime.JobRuns {
		require.Equal(t, automation.AutomationJobRunStatusQueued, jobRun.Status)
	}
}

func TestAutomationAgentTeamNativeAgentTaskRequiresTaskIdentity(t *testing.T) {
	now := time.Date(2026, 6, 7, 14, 0, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%smissingtask%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Missing task", Jobs: map[string]dsl.JobSpec{
		"task": {Steps: []dsl.StepSpec{{Uses: "agentteam/agent-task"}}},
	}}
	run := automation.AutomationRun{ID: "wfr_missing_task", WorkspaceID: "ws-1", Name: "Missing task", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}

	_, err := BuildInitialRuntime(run, spec, now, newID)
	require.ErrorContains(t, err, "agentteam/agent-task wait requires task_id or resource_id")
}

func TestAutomationAgentTeamNativeAgentTaskUsesFixedSignalIdentity(t *testing.T) {
	now := time.Date(2026, 6, 7, 14, 5, 0, 0, time.UTC)
	ids := 0
	newID := func(prefix string) (string, error) {
		ids++
		return fmt.Sprintf("%sfixedtask%03d", prefix, ids), nil
	}
	spec := dsl.DefinitionSpec{Name: "Fixed task signal", Jobs: map[string]dsl.JobSpec{
		"task": {Steps: []dsl.StepSpec{{Uses: "agentteam/agent-task", With: map[string]any{"task_id": "tsk_fixed", "resource_type": "custom", "event_type": "custom.done"}}}},
	}}
	run := automation.AutomationRun{ID: "wfr_fixed_task", WorkspaceID: "ws-1", Name: "Fixed task signal", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}

	runtime, err := BuildInitialRuntime(run, spec, now, newID)
	require.NoError(t, err)
	require.Len(t, runtime.Waits, 1)
	require.Equal(t, "agent_task", runtime.Waits[0].ResourceType)
	require.Equal(t, "tsk_fixed", runtime.Waits[0].ResourceID)
	require.Equal(t, "agent_task_completed", runtime.Waits[0].ResumeEventType)
	require.Equal(t, automation.AutomationWaitSignalKey("ws-1", automation.AutomationWaitSignal{WaitKind: automation.AutomationWaitKindAgentTask, EventType: "agent_task_completed", ResourceType: "agent_task", ResourceID: "tsk_fixed"}), runtime.Waits[0].SignalKey)
}

func TestAutomationAgentTeamNativeWaitActiveKeyIsUnique(t *testing.T) {
	ctx := context.Background()
	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{}))
	store := automation.New(db)
	now := time.Date(2026, 6, 2, 21, 30, 0, 0, time.UTC)
	activeKey := "ws-1:wfr_unique:wfj_wait"
	run := automation.AutomationRun{ID: "wfr_unique", WorkspaceID: "ws-1", Name: "Waits", Status: automation.AutomationRunStatusRunning, CreatedAt: now, UpdatedAt: now}
	_, err = store.CreateRun(ctx, run)
	require.NoError(t, err)
	jobRun := automation.AutomationJobRun{ID: "wfj_wait", WorkspaceID: "ws-1", RunID: "wfr_unique", JobID: "wait", JobType: "github_actions_job", Status: automation.AutomationJobRunStatusWaiting, InputJSON: `{"steps":[{"uses":"agentteam/wait-event"}]}`, ScheduledAt: now, CreatedAt: now, UpdatedAt: now}
	wait := automation.AutomationWait{ID: "wfw_one", WorkspaceID: "ws-1", RunID: "wfr_unique", JobRunID: "wfj_wait", WaitKind: automation.AutomationWaitKindWaitEvent, Status: automation.AutomationWaitStatusActive, ActiveKey: &activeKey, ResourceType: "event", ResourceID: "topic-1", CreatedAt: now, UpdatedAt: now}
	created, err := store.CreateWait(ctx, wait, jobRun, automation.AutomationEvent{ID: "wfe_wait_one", WorkspaceID: "ws-1", RunID: "wfr_unique", JobRunID: "wfj_wait", EventType: "wait_started", CreatedAt: now})
	require.NoError(t, err)
	duplicate := wait
	duplicate.ID = "wfw_two"
	again, err := store.CreateWait(ctx, duplicate, jobRun, automation.AutomationEvent{ID: "wfe_wait_two", WorkspaceID: "ws-1", RunID: "wfr_unique", JobRunID: "wfj_wait", EventType: "wait_started", CreatedAt: now})
	require.NoError(t, err)
	require.Equal(t, created.ID, again.ID)
	waits, err := store.ListWaits(ctx, "ws-1", "wfr_unique", automation.AutomationWaitStatusActive)
	require.NoError(t, err)
	require.Len(t, waits, 1)
	events, err := store.ListEvents(ctx, "ws-1", "wfr_unique", 1, 20)
	require.NoError(t, err)
	require.Len(t, events, 1)
}