package automation

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/automation/blob"
	"github.com/tidbcloud/AgentTeam/internal/automation/dsl"
	"github.com/tidbcloud/AgentTeam/internal/service/automation/runtime"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
	"gorm.io/gorm"
)

func (s *AutomationsService) RegisterAutomationRunner(ctx context.Context, workspaceID string, subject string, input AutomationRunnerRegisterMutation) (AutomationRunnerRegistration, error) {
	if err := s.validateScope(workspaceID, subject); err != nil {
		return AutomationRunnerRegistration{}, err
	}
	input.RunnerID = strings.TrimSpace(input.RunnerID)
	if input.RunnerID == "" {
		id, err := newAutomationRuntimeID("wfrn_")
		if err != nil {
			return AutomationRunnerRegistration{}, fmt.Errorf("generate automation runner id: %w", err)
		}
		input.RunnerID = id
	}
	if err := s.automationAuthorizer().CanManageRunner(ctx, s.automationAuthorizationRequest(ctx, workspaceID, subject, AutomationAuthorizationRequest{RunnerID: input.RunnerID})); err != nil {
		return AutomationRunnerRegistration{}, err
	}
	labels := runtime.NormalizeRunnerLabels(input.Labels)
	if len(labels) > s.automationLimits().MaxRunnerLabels {
		return AutomationRunnerRegistration{}, invalidArgument("Automation runner labels exceed maximum count")
	}
	runnerGroup := strings.TrimSpace(input.RunnerGroup)
	if err := validateAutomationRunnerMatchKeys(runnerGroup, labels); err != nil {
		return AutomationRunnerRegistration{}, err
	}
	token, err := newAutomationRunnerToken()
	if err != nil {
		return AutomationRunnerRegistration{}, fmt.Errorf("generate automation runner registration token: %w", err)
	}
	now := s.now()
	expiresAt := now.Add(automationRunnerRegistrationTokenTTL)
	runner := automationstore.AutomationRunner{
		WorkspaceID:  strings.TrimSpace(workspaceID),
		RunnerID:     input.RunnerID,
		Name:         strings.TrimSpace(input.Name),
		RunnerGroup:  runnerGroup,
		LabelsJSON:   automationRunnerJSON(labels),
		Status:       automationstore.AutomationRunnerStatusActive,
		RegisteredAt: now,
		CreatedAt:    now,
		UpdatedAt:    now,
	}
	tokenRow := automationstore.AutomationRunnerToken{
		TokenHash:   automationRunnerTokenHash(token),
		WorkspaceID: runner.WorkspaceID,
		RunnerID:    runner.RunnerID,
		TokenKind:   automationstore.AutomationRunnerTokenKindRegistration,
		ExpiresAt:   expiresAt,
		CreatedAt:   now,
		UpdatedAt:   now,
	}
	if err := s.store.RegisterRunnerWithToken(ctx, runner, tokenRow, now); err != nil {
		return AutomationRunnerRegistration{}, err
	}
	return AutomationRunnerRegistration{RunnerID: input.RunnerID, Name: runner.Name, RunnerGroup: runner.RunnerGroup, Labels: labels, RegistrationToken: token, TokenExpiresAt: expiresAt, RegisteredAt: now}, nil
}

// LookupAutomationRunnerRegistrationToken validates a one-time runner
// registration token without consuming it.
func (s *AutomationsService) LookupAutomationRunnerRegistrationToken(ctx context.Context, runnerToken string) (AutomationRunnerRegistrationLookup, error) {
	if err := s.ensureStore(); err != nil {
		return AutomationRunnerRegistrationLookup{}, err
	}
	if strings.TrimSpace(runnerToken) == "" {
		return AutomationRunnerRegistrationLookup{}, unauthorized("Automation runner registration token is required")
	}
	tokenHash := automationRunnerTokenHash(runnerToken)
	runner, _, err := s.store.GetActiveRunnerRegistrationByTokenHash(ctx, tokenHash, s.now())
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerRegistrationLookup{}, unauthorized("Automation runner registration token is invalid or expired")
	}
	if err != nil {
		return AutomationRunnerRegistrationLookup{}, err
	}
	return AutomationRunnerRegistrationLookup{
		WorkspaceID: runner.WorkspaceID,
		RunnerID:    runner.RunnerID,
		Name:        runner.Name,
	}, nil
}

// RegisterAutomationActionsRunnerCredential atomically consumes a one-time runner
// registration token and stores official actions/runner OAuth client metadata.
func (s *AutomationsService) RegisterAutomationActionsRunnerCredential(ctx context.Context, runnerToken string, credential automationstore.AutomationActionsRunnerCredential) error {
	if err := s.ensureStore(); err != nil {
		return err
	}
	if strings.TrimSpace(runnerToken) == "" {
		return unauthorized("Automation runner registration token is required")
	}
	_, err := s.store.UpsertActionsRunnerCredentialWithRegistrationTokenHash(ctx, automationRunnerTokenHash(runnerToken), credential, s.now())
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return unauthorized("Automation runner registration token is invalid or expired")
	}
	return err
}

func (s *AutomationsService) CreateAutomationRunnerSession(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerSessionMutation) (AutomationRunnerSession, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return AutomationRunnerSession{}, err
	}
	return s.createAutomationRunnerSessionWithToken(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// CreateOrReuseAutomationRunnerSessionForRunner returns the active session for
// an already-authenticated runner, creating one only when no active session
// exists.
func (s *AutomationsService) CreateOrReuseAutomationRunnerSessionForRunner(ctx context.Context, workspaceID string, input AutomationRunnerSessionMutation) (AutomationRunnerSession, error) {
	if err := s.ensureStore(); err != nil {
		return AutomationRunnerSession{}, err
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return AutomationRunnerSession{}, invalidArgument("Workspace id is required")
	}
	runnerID := strings.TrimSpace(input.RunnerID)
	if runnerID == "" {
		return AutomationRunnerSession{}, invalidArgument("Automation runner id is required")
	}
	now := s.now()
	reused, token, err := s.store.GetActiveRunnerSessionForRunner(ctx, workspaceID, runnerID, now)
	if err == nil {
		return automationRunnerSessionResponse(reused, "", token.TokenHash), nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerSession{}, err
	}
	input.RunnerID = runnerID
	session, sessionTokenRow, sessionToken, err := s.newAutomationRunnerSessionRows(workspaceID, input, now)
	if err != nil {
		return AutomationRunnerSession{}, err
	}
	created, err := s.store.CreateRunnerSessionForRunner(ctx, session, sessionTokenRow, now)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerSession{}, unauthorized("Automation runner is not active")
	}
	if err != nil {
		return AutomationRunnerSession{}, err
	}
	return automationRunnerSessionResponse(created, sessionToken, sessionTokenRow.TokenHash), nil
}

func (s *AutomationsService) createAutomationRunnerSessionWithToken(ctx context.Context, workspaceID string, registrationTokenHash string, input AutomationRunnerSessionMutation) (AutomationRunnerSession, error) {
	if strings.TrimSpace(registrationTokenHash) == "" {
		return AutomationRunnerSession{}, unauthorized("Automation runner registration token is required")
	}
	now := s.now()
	session, sessionTokenRow, sessionToken, err := s.newAutomationRunnerSessionRows(workspaceID, input, now)
	if err != nil {
		return AutomationRunnerSession{}, err
	}
	created, err := s.store.CreateRunnerSessionWithToken(ctx, registrationTokenHash, session, sessionTokenRow, now)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerSession{}, unauthorized("Automation runner registration token is invalid or expired")
	}
	if err != nil {
		return AutomationRunnerSession{}, err
	}
	return automationRunnerSessionResponse(created, sessionToken, sessionTokenRow.TokenHash), nil
}

func (s *AutomationsService) newAutomationRunnerSessionRows(workspaceID string, input AutomationRunnerSessionMutation, now time.Time) (automationstore.AutomationRunnerSession, automationstore.AutomationRunnerToken, string, error) {
	runnerID := strings.TrimSpace(input.RunnerID)
	if runnerID == "" {
		return automationstore.AutomationRunnerSession{}, automationstore.AutomationRunnerToken{}, "", invalidArgument("Automation runner id is required")
	}
	sessionID := strings.TrimSpace(input.SessionID)
	if sessionID == "" {
		id, err := newAutomationRuntimeID("wfs_")
		if err != nil {
			return automationstore.AutomationRunnerSession{}, automationstore.AutomationRunnerToken{}, "", fmt.Errorf("generate automation runner session id: %w", err)
		}
		sessionID = id
	}
	sessionToken, err := newAutomationRunnerToken()
	if err != nil {
		return automationstore.AutomationRunnerSession{}, automationstore.AutomationRunnerToken{}, "", fmt.Errorf("generate automation runner session token: %w", err)
	}
	expiresAt := now.Add(automationRunnerSessionTTL)
	session := automationstore.AutomationRunnerSession{
		WorkspaceID: strings.TrimSpace(workspaceID),
		RunnerID:    runnerID,
		SessionID:   sessionID,
		LeaseOwner:  automationRunnerLeaseOwner(runnerID, sessionID),
		Status:      automationstore.AutomationRunnerSessionStatusActive,
		HeartbeatAt: now,
		ExpiresAt:   expiresAt,
		CreatedAt:   now,
		UpdatedAt:   now,
	}
	token := automationstore.AutomationRunnerToken{
		TokenHash:   automationRunnerTokenHash(sessionToken),
		WorkspaceID: session.WorkspaceID,
		RunnerID:    session.RunnerID,
		TokenKind:   automationstore.AutomationRunnerTokenKindSession,
		SessionID:   session.SessionID,
		ExpiresAt:   expiresAt,
		CreatedAt:   now,
		UpdatedAt:   now,
	}
	return session, token, sessionToken, nil
}

func automationRunnerSessionResponse(created automationstore.AutomationRunnerSession, sessionToken string, sessionTokenHash string) AutomationRunnerSession {
	sessionTokenHash = strings.TrimSpace(sessionTokenHash)
	if sessionTokenHash == "" && strings.TrimSpace(sessionToken) != "" {
		sessionTokenHash = automationRunnerTokenHash(sessionToken)
	}
	return AutomationRunnerSession{RunnerID: created.RunnerID, SessionID: created.SessionID, LeaseOwner: created.LeaseOwner, SessionToken: sessionToken, SessionTokenHash: sessionTokenHash, HeartbeatAt: created.HeartbeatAt, ExpiresAt: created.ExpiresAt, CreatedAt: created.CreatedAt}
}

func (s *AutomationsService) AcquireAutomationRunnerJobs(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerAcquireMutation) ([]AutomationRunnerJob, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return nil, err
	}
	return s.acquireAutomationRunnerJobsByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// AcquireAutomationRunnerJobsByTokenHash claims automation jobs for an already
// authenticated runner session token hash.
func (s *AutomationsService) AcquireAutomationRunnerJobsByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerAcquireMutation) ([]AutomationRunnerJob, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return nil, err
	}
	return s.acquireAutomationRunnerJobsByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) acquireAutomationRunnerJobsByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerAcquireMutation) ([]AutomationRunnerJob, error) {
	runnerID := strings.TrimSpace(input.RunnerID)
	if runnerID == "" {
		return nil, invalidArgument("Automation runner id is required")
	}
	if strings.TrimSpace(input.SessionID) == "" {
		return nil, invalidArgument("Automation runner session id is required")
	}
	if input.ShardBucket != nil && (*input.ShardBucket < 0 || *input.ShardBucket >= automationstore.AutomationClaimShardCount) {
		return nil, invalidArgument("Automation runner shard_bucket is invalid")
	}
	queueName := strings.TrimSpace(input.QueueName)
	if queueName == "" {
		queueName = runtime.DefaultQueueName
	}
	limit := input.Limit
	if limit <= 0 {
		limit = 1
	}
	if limit > 20 {
		limit = 20
	}
	leaseSeconds := input.LeaseSeconds
	if leaseSeconds <= 0 {
		leaseSeconds = automationRunnerDefaultLeaseSeconds
	}
	if leaseSeconds > 3600 {
		leaseSeconds = 3600
	}
	waitSeconds := input.WaitSeconds
	if waitSeconds < 0 {
		return nil, invalidArgument("Automation runner wait_seconds is invalid")
	}
	if waitSeconds > 30 {
		waitSeconds = 30
	}
	sessionID := strings.TrimSpace(input.SessionID)
	shards := automationRunnerAcquireShards(input.ShardBucket)
	claim := func(now time.Time) ([]automationstore.AutomationJob, error) {
		session, err := s.activeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, runnerID, sessionID)
		if err != nil {
			return nil, err
		}
		if dsl.IsReservedRunnerGroup(session.RunnerGroup) {
			return nil, invalidArgument("Automation runner group is reserved")
		}
		sessionLabels := decodeAutomationRunnerLabels(session.LabelsJSON)
		leaseOwner := session.LeaseOwner
		jobs, err := s.store.ClaimJobs(ctx, automationstore.ClaimJobsRequest{WorkspaceID: strings.TrimSpace(workspaceID), QueueName: queueName, ShardBuckets: shards, LeaseOwner: leaseOwner, LeaseDuration: time.Duration(leaseSeconds) * time.Second, Limit: limit, RunnerGroup: session.RunnerGroup, Labels: sessionLabels, MaxRunnerLabels: s.automationLimits().MaxRunnerLabels, Now: now})
		if err != nil {
			return nil, err
		}
		return jobs, nil
	}
	claimed, err := s.acquireAutomationRunnerJobsLongPoll(ctx, strings.TrimSpace(workspaceID), waitSeconds, claim)
	if err != nil {
		return nil, err
	}
	out := make([]AutomationRunnerJob, 0, len(claimed))
	for _, job := range claimed {
		runnerJob, err := s.automationRunnerJob(ctx, strings.TrimSpace(workspaceID), job)
		if err != nil {
			return nil, err
		}
		out = append(out, runnerJob)
	}
	return out, nil
}

func (s *AutomationsService) acquireAutomationRunnerJobsLongPoll(ctx context.Context, workspaceID string, waitSeconds int, claim func(time.Time) ([]automationstore.AutomationJob, error)) ([]automationstore.AutomationJob, error) {
	if waitSeconds == 0 {
		return claim(s.now())
	}
	deadline := time.NewTimer(time.Duration(waitSeconds) * time.Second)
	defer deadline.Stop()
	backoff := automationRunnerAcquireInitialBackoff
	for {
		wake, cancelWait := s.runnerWorkspaceWait(workspaceID)
		claimed, err := claim(s.now())
		if err != nil || len(claimed) > 0 {
			cancelWait()
			return claimed, err
		}
		idle := time.NewTimer(backoff)
		select {
		case <-ctx.Done():
			stopAutomationRunnerTimer(idle)
			cancelWait()
			return nil, ctx.Err()
		case <-deadline.C:
			stopAutomationRunnerTimer(idle)
			cancelWait()
			return []automationstore.AutomationJob{}, nil
		case <-wake:
			stopAutomationRunnerTimer(idle)
			cancelWait()
			backoff = automationRunnerAcquireInitialBackoff
		case <-idle.C:
			cancelWait()
			backoff *= 2
			if backoff > automationRunnerAcquireMaxBackoff {
				backoff = automationRunnerAcquireMaxBackoff
			}
		}
	}
}

func stopAutomationRunnerTimer(timer *time.Timer) {
	if timer == nil {
		return
	}
	if timer.Stop() {
		return
	}
	select {
	case <-timer.C:
	default:
	}
}

func (s *AutomationsService) runnerWorkspaceWait(workspaceID string) (<-chan struct{}, func()) {
	if s == nil {
		return nil, func() {}
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return nil, func() {}
	}
	s.runnerWaitMu.Lock()
	defer s.runnerWaitMu.Unlock()
	if s.runnerWaiters == nil {
		s.runnerWaiters = make(map[string]map[chan struct{}]struct{})
	}
	waiters := s.runnerWaiters[workspaceID]
	if waiters == nil {
		waiters = make(map[chan struct{}]struct{})
		s.runnerWaiters[workspaceID] = waiters
	}
	wait := make(chan struct{})
	waiters[wait] = struct{}{}
	cancel := func() {
		s.runnerWaitMu.Lock()
		defer s.runnerWaitMu.Unlock()
		waiters := s.runnerWaiters[workspaceID]
		if waiters == nil {
			return
		}
		delete(waiters, wait)
		if len(waiters) == 0 {
			delete(s.runnerWaiters, workspaceID)
		}
	}
	return wait, cancel
}

func (s *AutomationsService) AcceptAutomationRunnerJob(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.acceptAutomationRunnerJobByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// AcceptAutomationRunnerJobByTokenHash accepts a leased automation job for an
// already authenticated runner session token hash.
func (s *AutomationsService) AcceptAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.acceptAutomationRunnerJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) acceptAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	job, leaseOwner, err := s.automationRunnerLeasedJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
	if err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	jobRun, err := runtime.NewWorker(s.store, s.now, nil).AcceptLeasedJob(ctx, job, leaseOwner)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return automationstore.AutomationJobRun{}, notFound("Automation runner job lease not found or expired")
	}
	if err == nil {
		slog.InfoContext(ctx, "automation runner job accepted", "workspace_id", workspaceID, "runner_id", input.RunnerID, "queue_job_id", input.QueueJobID, "job_run_id", jobRun.ID)
	}
	return jobRun, err
}

func (s *AutomationsService) CompleteAutomationRunnerJob(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.completeAutomationRunnerJobByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// CompleteAutomationRunnerJobByTokenHash completes a leased automation job for an
// already authenticated runner session token hash.
func (s *AutomationsService) CompleteAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.completeAutomationRunnerJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) completeAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	job, leaseOwner, err := s.automationRunnerLeasedJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
	if err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	if err := validateOptionalJSON(input.OutputJSON, "Automation runner output_json must be valid JSON"); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	redactor, err := s.automationRunnerCallbackRedactor(ctx, workspaceID, job)
	if err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	input.OutputJSON, err = redactAutomationRunnerJSON(input.OutputJSON, redactor)
	if err != nil {
		return automationstore.AutomationJobRun{}, invalidArgument("Automation runner output_json must be valid JSON")
	}
	worker := runtime.NewWorker(s.store, s.now, nil)
	if err := worker.CompleteLeasedJob(ctx, job, leaseOwner, input.OutputJSON); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return automationstore.AutomationJobRun{}, notFound("Automation runner job lease not found or expired")
		}
		return automationstore.AutomationJobRun{}, err
	}
	slog.InfoContext(ctx, "automation runner job completed", "workspace_id", workspaceID, "runner_id", input.RunnerID, "queue_job_id", input.QueueJobID, "job_run_id", job.JobRunID)
	return s.store.GetJobRun(ctx, strings.TrimSpace(workspaceID), job.JobRunID)
}

func (s *AutomationsService) FailAutomationRunnerJob(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.failAutomationRunnerJobByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// FailAutomationRunnerJobByTokenHash fails a leased automation job for an already
// authenticated runner session token hash.
func (s *AutomationsService) FailAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	return s.failAutomationRunnerJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) failAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJobRun, error) {
	job, leaseOwner, err := s.automationRunnerLeasedJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
	if err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	if err := validateOptionalJSON(input.OutputJSON, "Automation runner output_json must be valid JSON"); err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	message := strings.TrimSpace(input.Message)
	if message == "" {
		message = "automation runner reported failure"
	}
	redactor, err := s.automationRunnerCallbackRedactor(ctx, workspaceID, job)
	if err != nil {
		return automationstore.AutomationJobRun{}, err
	}
	message = redactor.RedactString(message)
	input.OutputJSON, err = redactAutomationRunnerJSON(input.OutputJSON, redactor)
	if err != nil {
		return automationstore.AutomationJobRun{}, invalidArgument("Automation runner output_json must be valid JSON")
	}
	worker := runtime.NewWorker(s.store, s.now, nil)
	var failErr error
	if input.TerminalFailure {
		failErr = worker.FailLeasedJobTerminal(ctx, job, leaseOwner, message, input.OutputJSON)
	} else {
		failErr = worker.FailLeasedJob(ctx, job, leaseOwner, message, input.OutputJSON)
	}
	if failErr != nil {
		if errors.Is(failErr, gorm.ErrRecordNotFound) {
			return automationstore.AutomationJobRun{}, notFound("Automation runner job lease not found or expired")
		}
		return automationstore.AutomationJobRun{}, failErr
	}
	slog.InfoContext(ctx, "automation runner job failed", "workspace_id", workspaceID, "runner_id", input.RunnerID, "queue_job_id", input.QueueJobID, "job_run_id", job.JobRunID)
	return s.store.GetJobRun(ctx, strings.TrimSpace(workspaceID), job.JobRunID)
}

func (s *AutomationsService) AppendAutomationRunnerJobLog(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationEvent, error) {
	if strings.TrimSpace(input.LogContent) == "" {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner log content is required")
	}
	payload := []byte(input.LogContent)
	if len(payload) > s.automationLimits().MaxLogChunkBytes {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner log chunk exceeds maximum size")
	}
	stream := normalizeAutomationLogStream(input.LogStream)
	if len(stream) > automationLogStreamMaxLength {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner log stream exceeds maximum length")
	}
	if input.LogSequence <= 0 {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner log sequence is required")
	}
	contentType, err := automationBlobContentType(input.LogContentType, "text/plain; charset=utf-8")
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	actualSHA256 := blob.SHA256Hex(payload)
	if err := validateAutomationBlobSHA256(input.LogSHA256, actualSHA256, false, "Automation runner log sha256 is invalid"); err != nil {
		return automationstore.AutomationEvent{}, err
	}
	job, leaseOwner, err := s.automationRunnerLeasedJob(ctx, workspaceID, runnerToken, input)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	jobRun, err := s.store.GetJobRun(ctx, strings.TrimSpace(workspaceID), job.JobRunID)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	redactor, err := s.automationRunnerCallbackRedactorForJobRun(ctx, workspaceID, jobRun)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	payload = redactor.RedactBytes(payload)
	if len(payload) > s.automationLimits().MaxLogChunkBytes {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner log chunk exceeds maximum size")
	}
	actualSHA256 = blob.SHA256Hex(payload)
	now := s.now()
	objectKey, err := automationLogChunkObjectKey(jobRun.WorkspaceID, jobRun.RunID, jobRun.ID, stream, input.LogSequence)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	if err := s.store.VerifyLeasedJob(ctx, job.ID, leaseOwner, now); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return automationstore.AutomationEvent{}, notFound("Automation runner job lease not found or expired")
		}
		return automationstore.AutomationEvent{}, err
	}
	if err := s.putAutomationBlob(ctx, objectKey, payload); err != nil {
		if errors.Is(err, blob.ErrAlreadyExists) {
			return automationstore.AutomationEvent{}, invalidArgument("Automation runner log chunk object already exists with different content")
		}
		return automationstore.AutomationEvent{}, err
	}
	chunk := automationstore.AutomationLogChunk{ID: automationLogChunkID(jobRun.ID, stream, input.LogSequence), WorkspaceID: jobRun.WorkspaceID, RunID: jobRun.RunID, JobRunID: jobRun.ID, Stream: stream, Sequence: input.LogSequence, ObjectKey: objectKey, SizeBytes: int64(len(payload)), SHA256: actualSHA256, ContentType: contentType, CreatedAt: now}
	metadata := automationRunnerJSON(map[string]any{"stream": stream, "sequence": input.LogSequence, "object_key": objectKey, "size": len(payload), "sha256": actualSHA256, "content_type": contentType})
	event := automationstore.AutomationEvent{ID: automationRunnerLogEventID(jobRun.ID, stream, input.LogSequence), WorkspaceID: jobRun.WorkspaceID, RunID: jobRun.RunID, JobRunID: jobRun.ID, EventType: "job_log_appended", ActorPrincipalID: leaseOwner, ResourceType: "automation_job", ResourceID: fmt.Sprint(job.ID), Summary: "Automation job log appended", MetadataJSON: metadata, CreatedAt: now}
	_, created, err := s.store.CreateLeasedJobLogChunk(ctx, job.ID, leaseOwner, chunk, event, now)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		s.deleteAutomationLogBlobIfMetadataMismatch(ctx, chunk)
		return automationstore.AutomationEvent{}, notFound("Automation runner job lease not found or expired")
	}
	if errors.Is(err, automationstore.ErrAutomationBlobMetadataConflict) {
		s.deleteAutomationLogBlobIfMetadataMismatch(ctx, chunk)
	}
	if err = mapAutomationBlobMetadataError(err); err != nil {
		return automationstore.AutomationEvent{}, err
	}
	slog.InfoContext(ctx, "automation runner job log appended", "workspace_id", workspaceID, "runner_id", input.RunnerID, "queue_job_id", input.QueueJobID, "job_run_id", job.JobRunID, "sequence", input.LogSequence)
	return created, err
}

func (s *AutomationsService) AppendAutomationRunnerJobArtifact(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationEvent, error) {
	limits := s.automationLimits()
	artifactName := strings.TrimSpace(input.ArtifactName)
	if artifactName == "" {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact name is required")
	}
	if len(artifactName) > automationArtifactNameMaxLength {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact name exceeds maximum length")
	}
	if input.ArtifactSize < 0 {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact size is invalid")
	}
	if input.ArtifactSize > limits.MaxArtifactBytes {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact exceeds maximum size")
	}
	encodedContent := strings.TrimSpace(input.ArtifactContentBase64)
	if encodedContent == "" {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact content_base64 is required")
	}
	encodedLength := int64(len(encodedContent))
	if encodedLength > automationBase64EncodedLen(input.ArtifactSize) {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact size does not match content")
	}
	if encodedLength > automationBase64EncodedLen(limits.MaxArtifactBytes) {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact exceeds maximum size")
	}
	job, leaseOwner, err := s.automationRunnerLeasedJob(ctx, workspaceID, runnerToken, input)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	now := s.now()
	if err := s.store.VerifyLeasedJob(ctx, job.ID, leaseOwner, now); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return automationstore.AutomationEvent{}, notFound("Automation runner job lease not found or expired")
		}
		return automationstore.AutomationEvent{}, err
	}
	payload, err := base64.StdEncoding.DecodeString(encodedContent)
	if err != nil {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact content_base64 is invalid")
	}
	if int64(len(payload)) != input.ArtifactSize {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact size does not match content")
	}
	if int64(len(payload)) > limits.MaxArtifactBytes {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact exceeds maximum size")
	}
	actualSHA256 := blob.SHA256Hex(payload)
	if err := validateAutomationBlobSHA256(input.ArtifactSHA256, actualSHA256, true, "Automation runner artifact sha256 is invalid"); err != nil {
		return automationstore.AutomationEvent{}, err
	}
	contentType, err := automationBlobContentType(input.ArtifactContentType, "application/octet-stream")
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	expiresAt, err := parseAutomationExpiresAt(input.ArtifactExpiresAt)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	if len(input.MetadataJSON) > limits.MaxJobInputBytes {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact metadata_json exceeds maximum size")
	}
	if err := validateOptionalJSON(input.MetadataJSON, "Automation runner artifact metadata_json must be valid JSON"); err != nil {
		return automationstore.AutomationEvent{}, err
	}
	jobRun, err := s.store.GetJobRun(ctx, strings.TrimSpace(workspaceID), job.JobRunID)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	redactor, err := s.automationRunnerCallbackRedactorForJobRun(ctx, workspaceID, jobRun)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	artifactName = redactor.RedactString(artifactName)
	input.MetadataJSON, err = redactAutomationRunnerJSON(input.MetadataJSON, redactor)
	if err != nil {
		return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact metadata_json must be valid JSON")
	}
	artifactID, err := newAutomationRuntimeID("wfa_")
	if err != nil {
		return automationstore.AutomationEvent{}, fmt.Errorf("generate automation artifact id: %w", err)
	}
	objectKey, err := automationArtifactObjectKey(jobRun.WorkspaceID, jobRun.RunID, jobRun.ID, artifactID, artifactName)
	if err != nil {
		return automationstore.AutomationEvent{}, err
	}
	if err := s.putAutomationBlob(ctx, objectKey, payload); err != nil {
		if errors.Is(err, blob.ErrAlreadyExists) {
			return automationstore.AutomationEvent{}, invalidArgument("Automation runner artifact object already exists with different content")
		}
		return automationstore.AutomationEvent{}, err
	}
	metadata := map[string]any{"name": artifactName, "object_key": objectKey, "size": input.ArtifactSize, "sha256": actualSHA256, "content_type": contentType}
	if strings.TrimSpace(input.MetadataJSON) != "" {
		var extra map[string]any
		if err := json.Unmarshal([]byte(input.MetadataJSON), &extra); err == nil {
			metadata["metadata"] = extra
		}
	}
	artifact := automationstore.AutomationArtifact{ID: artifactID, WorkspaceID: jobRun.WorkspaceID, RunID: jobRun.RunID, JobRunID: jobRun.ID, Name: artifactName, ObjectKey: objectKey, SizeBytes: input.ArtifactSize, SHA256: actualSHA256, ContentType: contentType, ExpiresAt: expiresAt, CreatedAt: now, UpdatedAt: now}
	event := automationstore.AutomationEvent{ID: automationRunnerArtifactEventID(artifactID), WorkspaceID: jobRun.WorkspaceID, RunID: jobRun.RunID, JobRunID: jobRun.ID, EventType: "job_artifact_recorded", ActorPrincipalID: leaseOwner, ResourceType: "automation_job", ResourceID: fmt.Sprint(job.ID), Summary: "Automation job artifact recorded", MetadataJSON: automationRunnerJSON(metadata), CreatedAt: now}
	_, created, err := s.store.CreateLeasedJobArtifact(ctx, job.ID, leaseOwner, artifact, event, now)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		s.deleteAutomationBlob(ctx, objectKey)
		return automationstore.AutomationEvent{}, notFound("Automation runner job lease not found or expired")
	}
	if err != nil {
		s.deleteAutomationBlob(ctx, objectKey)
		return automationstore.AutomationEvent{}, err
	}
	slog.InfoContext(ctx, "automation runner job artifact recorded", "workspace_id", workspaceID, "runner_id", input.RunnerID, "queue_job_id", input.QueueJobID, "job_run_id", job.JobRunID, "artifact_name", artifactName)
	return created, err
}

func (s *AutomationsService) HeartbeatAutomationRunnerSession(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerHeartbeatMutation) (AutomationRunnerHeartbeat, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return AutomationRunnerHeartbeat{}, err
	}
	return s.heartbeatAutomationRunnerSessionByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// HeartbeatAutomationRunnerSessionByTokenHash renews a runner session for an
// already authenticated runner session token hash.
func (s *AutomationsService) HeartbeatAutomationRunnerSessionByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerHeartbeatMutation) (AutomationRunnerHeartbeat, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return AutomationRunnerHeartbeat{}, err
	}
	return s.heartbeatAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) heartbeatAutomationRunnerSessionByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerHeartbeatMutation) (AutomationRunnerHeartbeat, error) {
	session, err := s.activeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input.RunnerID, input.SessionID)
	if err != nil {
		return AutomationRunnerHeartbeat{}, err
	}
	now := s.now()
	expiresAt := now.Add(automationRunnerSessionTTL)
	if !automationRunnerHeartbeatDue(session, now) {
		cancelRequested, err := s.store.RunnerJobCancelRequested(ctx, strings.TrimSpace(workspaceID), input.QueueJobID, session.LeaseOwner)
		if err != nil {
			return AutomationRunnerHeartbeat{}, err
		}
		return AutomationRunnerHeartbeat{RunnerID: session.RunnerID, SessionID: session.SessionID, LeaseOwner: session.LeaseOwner, HeartbeatAt: session.HeartbeatAt, ExpiresAt: session.ExpiresAt, CancelRequested: cancelRequested}, nil
	}
	renewed, err := s.store.HeartbeatRunnerSession(ctx, strings.TrimSpace(runnerTokenHash), strings.TrimSpace(workspaceID), session.RunnerID, session.SessionID, now, expiresAt)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerHeartbeat{}, unauthorized("Automation runner session is not active")
	}
	if err != nil {
		return AutomationRunnerHeartbeat{}, err
	}
	cancelRequested, err := s.store.RunnerJobCancelRequested(ctx, strings.TrimSpace(workspaceID), input.QueueJobID, renewed.LeaseOwner)
	if err != nil {
		return AutomationRunnerHeartbeat{}, err
	}
	return AutomationRunnerHeartbeat{RunnerID: renewed.RunnerID, SessionID: renewed.SessionID, LeaseOwner: renewed.LeaseOwner, HeartbeatAt: renewed.HeartbeatAt, ExpiresAt: renewed.ExpiresAt, CancelRequested: cancelRequested}, nil
}

func automationRunnerHeartbeatDue(session automationstore.AutomationRunnerSession, now time.Time) bool {
	return !session.ExpiresAt.After(now.Add(automationRunnerHeartbeatRefresh))
}

// RevokeAutomationRunnerSession revokes one active runner session and its token.
func (s *AutomationsService) RevokeAutomationRunnerSession(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerSessionMutation) error {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return err
	}
	return s.revokeAutomationRunnerSessionByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// RevokeAutomationRunnerSessionByTokenHash revokes a runner session for an
// already authenticated runner session token hash.
func (s *AutomationsService) RevokeAutomationRunnerSessionByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerSessionMutation) error {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return err
	}
	return s.revokeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) revokeAutomationRunnerSessionByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerSessionMutation) error {
	session, err := s.activeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input.RunnerID, input.SessionID)
	if err != nil {
		return err
	}
	if err := s.store.RevokeRunnerSession(ctx, strings.TrimSpace(runnerTokenHash), strings.TrimSpace(workspaceID), session.RunnerID, session.SessionID, s.now()); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return unauthorized("Automation runner session is not active")
		}
		return err
	}
	return nil
}

func (s *AutomationsService) RenewAutomationRunnerJob(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobRenewMutation) (AutomationRunnerJobLease, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return AutomationRunnerJobLease{}, err
	}
	return s.renewAutomationRunnerJobByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

// RenewAutomationRunnerJobByTokenHash renews a leased automation job for an already
// authenticated runner session token hash.
func (s *AutomationsService) RenewAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobRenewMutation) (AutomationRunnerJobLease, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return AutomationRunnerJobLease{}, err
	}
	return s.renewAutomationRunnerJobByTokenHash(ctx, workspaceID, runnerTokenHash, input)
}

func (s *AutomationsService) renewAutomationRunnerJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobRenewMutation) (AutomationRunnerJobLease, error) {
	session, err := s.activeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input.RunnerID, input.SessionID)
	if err != nil {
		return AutomationRunnerJobLease{}, err
	}
	if input.QueueJobID <= 0 {
		return AutomationRunnerJobLease{}, invalidArgument("Automation runner queue job id is required")
	}
	leaseSeconds := input.LeaseSeconds
	if leaseSeconds <= 0 {
		leaseSeconds = automationRunnerDefaultLeaseSeconds
	}
	if leaseSeconds > 3600 {
		leaseSeconds = 3600
	}
	now := s.now()
	leaseUntil := now.Add(time.Duration(leaseSeconds) * time.Second)
	job, err := s.store.RenewLeasedJob(ctx, strings.TrimSpace(workspaceID), input.QueueJobID, session.LeaseOwner, now, leaseUntil)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return AutomationRunnerJobLease{}, notFound("Automation runner job lease not found or expired")
	}
	if err != nil {
		return AutomationRunnerJobLease{}, err
	}
	return AutomationRunnerJobLease{QueueJobID: job.ID, LeaseOwner: job.LeaseOwner, LeaseUntil: job.LeaseUntil}, nil
}

func (s *AutomationsService) CancelAutomationRun(ctx context.Context, workspaceID string, subject string, runID string, reason string) (automationstore.AutomationRun, error) {
	if err := s.validateScope(workspaceID, subject); err != nil {
		return automationstore.AutomationRun{}, err
	}
	if err := s.automationAuthorizer().CanCancelRun(ctx, s.automationAuthorizationRequest(ctx, workspaceID, subject, AutomationAuthorizationRequest{})); err != nil {
		return automationstore.AutomationRun{}, err
	}
	run, err := s.store.GetRun(ctx, strings.TrimSpace(workspaceID), strings.TrimSpace(runID))
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return automationstore.AutomationRun{}, notFound("Automation run not found")
	}
	if err != nil {
		return automationstore.AutomationRun{}, err
	}
	if err := s.automationAuthorizer().CanCancelRun(ctx, s.automationAuthorizationRequest(ctx, workspaceID, subject, AutomationAuthorizationRequest{Run: run})); err != nil {
		return automationstore.AutomationRun{}, err
	}
	if err := runtime.NewWorker(s.store, s.now, nil).CancelRun(ctx, strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(subject), strings.TrimSpace(reason)); err != nil {
		return automationstore.AutomationRun{}, err
	}
	slog.InfoContext(ctx, "automation run canceled", "workspace_id", workspaceID, "run_id", runID, "actor", subject)
	return s.store.GetRun(ctx, strings.TrimSpace(workspaceID), strings.TrimSpace(runID))
}

func (s *AutomationsService) automationRunnerLeasedJob(ctx context.Context, workspaceID string, runnerToken string, input AutomationRunnerJobMutation) (automationstore.AutomationJob, string, error) {
	if err := s.validateRunnerScope(workspaceID, runnerToken); err != nil {
		return automationstore.AutomationJob{}, "", err
	}
	return s.automationRunnerLeasedJobByTokenHash(ctx, workspaceID, automationRunnerTokenHash(runnerToken), input)
}

func (s *AutomationsService) automationRunnerLeasedJobByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, input AutomationRunnerJobMutation) (automationstore.AutomationJob, string, error) {
	if strings.TrimSpace(input.RunnerID) == "" {
		return automationstore.AutomationJob{}, "", invalidArgument("Automation runner id is required")
	}
	if strings.TrimSpace(input.SessionID) == "" {
		return automationstore.AutomationJob{}, "", invalidArgument("Automation runner session id is required")
	}
	if input.QueueJobID <= 0 {
		return automationstore.AutomationJob{}, "", invalidArgument("Automation runner queue job id is required")
	}
	session, err := s.activeAutomationRunnerSessionByTokenHash(ctx, workspaceID, runnerTokenHash, input.RunnerID, input.SessionID)
	if err != nil {
		return automationstore.AutomationJob{}, "", err
	}
	job, err := s.store.GetJob(ctx, input.QueueJobID)
	if errors.Is(err, gorm.ErrRecordNotFound) || (err == nil && job.WorkspaceID != strings.TrimSpace(workspaceID)) {
		return automationstore.AutomationJob{}, "", notFound("Automation runner job not found")
	}
	if err != nil {
		return automationstore.AutomationJob{}, "", err
	}
	return job, session.LeaseOwner, nil
}

func (s *AutomationsService) automationRunnerJob(ctx context.Context, workspaceID string, job automationstore.AutomationJob) (AutomationRunnerJob, error) {
	jobRun, err := s.store.GetJobRun(ctx, workspaceID, job.JobRunID)
	if err != nil {
		return AutomationRunnerJob{}, err
	}
	run, err := s.store.GetRun(ctx, workspaceID, job.RunID)
	if err != nil {
		return AutomationRunnerJob{}, err
	}
	return AutomationRunnerJob{QueueJobID: job.ID, WorkspaceID: job.WorkspaceID, RunID: job.RunID, JobRunID: job.JobRunID, JobID: jobRun.JobID, QueueName: job.QueueName, Status: job.Status, Attempt: job.Attempt, MaxAttempts: job.MaxAttempts, LeaseOwner: job.LeaseOwner, LeaseUntil: job.LeaseUntil, RepositoryOwner: run.RepositoryOwner, RepositoryName: run.RepositoryName, RepositoryFullName: run.RepositoryFullName, DefinitionPath: run.DefinitionPath, GitRef: run.GitRef, CommitSHA: run.CommitSHA, ContentID: run.ContentID, ContentVersionID: run.ContentVersionID, AutomationName: run.Name, RunsOnJSON: jobRun.RunsOnJSON, RunnerGroup: job.RunnerGroup, Labels: decodeAutomationRunnerLabels(job.LabelsJSON), InputJSON: jobRun.InputJSON, DependsOnJSON: jobRun.DependsOnJSON, TriggerContextJSON: automationRunTriggerContextJSON(run.Metadata)}, nil
}

func (s *AutomationsService) validateRunnerScope(workspaceID string, runnerToken string) error {
	if err := s.ensureStore(); err != nil {
		return err
	}
	if strings.TrimSpace(workspaceID) == "" {
		return invalidArgument("Workspace id is required")
	}
	if strings.TrimSpace(runnerToken) == "" {
		return unauthorized("Automation runner token is required")
	}
	return nil
}

func (s *AutomationsService) validateRunnerTokenHashScope(workspaceID string, runnerTokenHash string) error {
	if err := s.ensureStore(); err != nil {
		return err
	}
	if strings.TrimSpace(workspaceID) == "" {
		return invalidArgument("Workspace id is required")
	}
	if strings.TrimSpace(runnerTokenHash) == "" {
		return unauthorized("Automation runner token is required")
	}
	return nil
}

func (s *AutomationsService) activeAutomationRunnerSessionByTokenHash(ctx context.Context, workspaceID string, runnerTokenHash string, runnerID string, sessionID string) (automationstore.AutomationRunnerSession, error) {
	if err := s.validateRunnerTokenHashScope(workspaceID, runnerTokenHash); err != nil {
		return automationstore.AutomationRunnerSession{}, err
	}
	runnerID = strings.TrimSpace(runnerID)
	if runnerID == "" {
		return automationstore.AutomationRunnerSession{}, invalidArgument("Automation runner id is required")
	}
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return automationstore.AutomationRunnerSession{}, invalidArgument("Automation runner session id is required")
	}
	session, err := s.store.GetActiveRunnerSessionByToken(ctx, strings.TrimSpace(runnerTokenHash), strings.TrimSpace(workspaceID), runnerID, sessionID, s.now())
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return automationstore.AutomationRunnerSession{}, unauthorized("Automation runner session is not active")
	}
	if err != nil {
		return automationstore.AutomationRunnerSession{}, err
	}
	return session, nil
}

func automationRunnerLeaseOwner(runnerID string, sessionID string) string {
	if session := strings.TrimSpace(sessionID); session != "" {
		return "wfle_" + blob.ShortHash(strings.TrimSpace(runnerID)+"|"+session, 48)
	}
	return "wfle_" + blob.ShortHash(strings.TrimSpace(runnerID), 48)
}

func automationRunnerAcquireShards(shardBucket *int) []int {
	if shardBucket != nil {
		if *shardBucket < 0 {
			return nil
		}
		return []int{*shardBucket}
	}
	shards := make([]int, 0, automationstore.AutomationClaimShardCount)
	for shard := 0; shard < automationstore.AutomationClaimShardCount; shard++ {
		shards = append(shards, shard)
	}
	return shards
}

func decodeAutomationRunnerLabels(raw string) []string {
	if strings.TrimSpace(raw) == "" {
		return nil
	}
	var labels []string
	if err := json.Unmarshal([]byte(raw), &labels); err != nil {
		return nil
	}
	return runtime.NormalizeRunnerLabels(labels)
}