Single mutability evaluator
apps/backend/internal/task/models/runner_mutability.go ↗One function implements the ten ordered conditions so projection and enforcement cannot disagree.
Verdict
func EvaluateRunnerMutability(s RunnerMutabilitySignals) RunnerMutabilityVerdict {
switch {
case s.Archived:
return RunnerMutabilityVerdict{false, RunnerReasonTaskArchived}
case s.RepositoryCount == 0:
return RunnerMutabilityVerdict{false, RunnerReasonNoRepository}
case s.RepositoryCount > 1:
return RunnerMutabilityVerdict{false, RunnerReasonMultipleRepositories}
case s.HasSession:
return RunnerMutabilityVerdict{false, RunnerReasonSessionExists}
case s.HasEnvironment:
return RunnerMutabilityVerdict{false, RunnerReasonEnvironmentExists}
case s.HasExecutorRunning:
return RunnerMutabilityVerdict{false, RunnerReasonExecutorRunning}
case s.HasWorkspaceFolder:
return RunnerMutabilityVerdict{false, RunnerReasonWorkspaceFolderAttached}
case strings.TrimSpace(s.WorkspacePath) != "":
return RunnerMutabilityVerdict{false, RunnerReasonWorkspacePathSet}
case s.HasActiveGroupMembership:
return RunnerMutabilityVerdict{false, RunnerReasonWorkspaceGroupMember}
case !runnerWorkspaceBindingIndependent(s.HasParent, s.WorkspaceMode):
return RunnerMutabilityVerdict{false, RunnerReasonWorkspaceBindingNotIndependent}
default:
return RunnerMutabilityVerdict{true, RunnerReasonEligible}
}
}
Signals
type RunnerMutabilitySignals struct {
Archived bool
RepositoryCount int
HasSession bool
HasEnvironment bool
HasExecutorRunning bool
HasWorkspaceFolder bool
WorkspacePath string
HasActiveGroupMembership bool
HasParent bool
WorkspaceMode string
}
Service: batched projection and switch
apps/backend/internal/task/service/service_runner_switch.go ↗The service batches mutability reads for lists and runs the compatibility gate outside the transaction before it delegates to the repository.
Batched views
func (s *Service) BuildRunnerMutabilityViews(ctx context.Context, tasks []*models.Task) map[string]RunnerMutabilityView {
out := make(map[string]RunnerMutabilityView, len(tasks))
if len(tasks) == 0 {
return out
}
batch, err := s.loadRunnerMutabilitySignalBatch(ctx, ids)
if err != nil {
s.logger.Warn("failed to load signals for runner mutability", zap.Error(err))
return unavailable()
}
for _, t := range tasks {
out[t.ID] = batch.verdictFor(t)
}
return out
}
Switch entry
func (s *Service) SwitchTaskRunner(ctx context.Context, taskID, executorProfileID string) (*models.Task, error) {
if strings.TrimSpace(taskID) == "" || strings.TrimSpace(executorProfileID) == "" {
return nil, ErrRunnerSwitchMalformed
}
task, err := s.tasks.GetTask(ctx, taskID)
if err != nil {
return nil, err
}
if err := s.authorizeTaskScope(ctx, taskID, authz.ScopeTaskWrite); err != nil {
return nil, err
}
executor, err := s.resolveExecutorForProfile(ctx, executorProfileID)
if err != nil {
return nil, err
}
compat := s.resolveRunnerCompatibility(ctx, taskID, executor)
req := models.RunnerSwitchRequest{
TaskID: taskID, ExecutorProfileID: executorProfileID,
CompatibilityApplicable: compat.applicable,
CompatibilityChecked: compat.checked,
CompatibilityCloneURLFound: compat.cloneURLFound,
ResolvedRepositoryID: compat.repositoryID,
GroupMembershipChecker: s.runnerGroupMembershipChecker,
}
result, err := s.tasks.SwitchTaskRunner(ctx, req)
if err != nil {
return nil, err
}
if result.Changed {
s.PublishTaskUpdated(ctx, result.Task)
}
return result.Task, nil
}
Compatibility gate
func (s *Service) resolveRunnerCompatibility(ctx context.Context, taskID string, executor *models.Executor) runnerCompatibilityResolution {
if s.executorCapabilityProber == nil || !s.executorCapabilityProber.RequiresCloneURL(string(executor.Type)) {
return runnerCompatibilityResolution{}
}
links, err := s.taskRepos.ListTaskRepositories(ctx, taskID)
if err != nil {
return runnerCompatibilityResolution{applicable: true, resolutionFailed: true}
}
if len(links) != 1 {
return runnerCompatibilityResolution{applicable: true}
}
found, err := runnerRepositoryHasCloneURL(ctx, repo)
if err != nil {
return runnerCompatibilityResolution{applicable: true, resolutionFailed: true}
}
return runnerCompatibilityResolution{applicable: true, checked: true, cloneURLFound: found, repositoryID: link.RepositoryID}
}
Repository: serialized transaction
apps/backend/internal/task/repository/sqlite/runner_switch.go ↗The repository takes the task row lock, re-evaluates mutability, confirms the compatibility snapshot, and writes one metadata key atomically.
Transaction
func (r *Repository) SwitchTaskRunner(ctx context.Context, req models.RunnerSwitchRequest) (*models.RunnerSwitchResult, error) {
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return nil, fmt.Errorf("%w: %v", repoerrors.ErrRunnerEvaluationUnavailable, err)
}
defer func() { _ = tx.Rollback() }()
if err := kandevdb.LockTaskRowInTx(ctx, tx, r.db.DriverName(), req.TaskID); err != nil {
return nil, err
}
task, err := r.runnerSwitchReadTaskTx(ctx, tx, req.TaskID)
verdict, repoSnapshot, err := r.runnerSwitchEvaluate(ctx, req, task)
if !verdict.Editable {
return nil, &repoerrors.ErrRunnerMutabilityConflict{Reason: verdict.Reason}
}
switch {
case req.CompatibilityResolutionFailed:
return nil, fmt.Errorf("%w: compatibility resolution failed", repoerrors.ErrRunnerEvaluationUnavailable)
case req.CompatibilityChecked:
if err := runnerSwitchConfirmCompatibility(req, repoSnapshot); err != nil {
return nil, err
}
case req.CompatibilityApplicable:
return nil, fmt.Errorf("%w: repository shape changed", repoerrors.ErrRunnerEvaluationUnavailable)
}
result, err := r.runnerSwitchApply(ctx, tx, task, req.ExecutorProfileID)
if err := tx.Commit(); err != nil {
return nil, fmt.Errorf("%w: %v", repoerrors.ErrRunnerEvaluationUnavailable, err)
}
return result, nil
}
Row lock
func LockTaskRowInTx(ctx context.Context, tx *sqlx.Tx, driverName, taskID string) error {
if driverName != pgxDriverName {
return nil
}
var locked string
if err := tx.QueryRowContext(ctx, tx.Rebind(`SELECT id FROM tasks WHERE id = ? FOR UPDATE`), taskID).Scan(&locked); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("%w: %s", ErrTaskRowNotFound, taskID)
}
return fmt.Errorf("lock task row: %w", err)
}
return nil
}
Projection plumbing
apps/backend/internal/task/dto/task_runner_mutability.go ↗Every task read path stamps the derived verdict onto the DTO so the client never computes it.
DTO enrichment
func EnrichTaskRunnerMutability(dto *TaskDTO, projection TaskRunnerMutabilityProjection) {
if dto == nil {
return
}
dto.RunnerEditable = projection.Editable
dto.RunnerIneligibleReason = projection.Reason
}
Boot payload
runnerViews := b.p.taskSvc.BuildRunnerMutabilityViews(ctx, tasks)
// ...
taskdto.EnrichTaskRunnerMutability(&dto, bootRunnerMutabilityProjection(runnerViews[task.ID]))
taskdto.EnrichTaskStatusSummary(&dto, task.ID, statusSummaries)
WS handler
func (h *TaskHandlers) wsUpdateTaskRunner(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsUpdateTaskRunnerRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
task, err := h.service.SwitchTaskRunner(ctx, req.ID, req.ExecutorProfileID)
if err != nil {
return runnerSwitchWSError(msg, err, h.logger)
}
dtos, err := buildTaskDTOsWithSessionInfo(ctx, h.service, h.logger, h.foregroundActivity, h.taskParkedProjection, []*models.Task{task})
return ws.NewResponse(msg.ID, msg.Action, dtos[0])
}
Frontend: store and dialog
apps/web/lib/kanban/map-task.ts ↗The kanban mapper fails closed on missing fields and the edit dialog tracks whether the user touched the runner selector.
Kanban mapper
function runnerMutabilityProjection(source: TaskLike) {
return {
runnerEditable: source.runner_editable ?? false,
runnerIneligibleReason: source.runner_ineligible_reason ?? "evaluation_unavailable",
};
}
Edit target
export type SidebarTaskEditTarget = {
id: string;
title: string;
workflowId: string;
workflowStepId: string;
primaryExecutorProfileId?: string;
runnerEditable?: boolean;
runnerIneligibleReason?: string;
};
WS API
export async function switchTaskRunner(taskId: string, executorProfileId: string): Promise<Task> {
const client = getWebSocketClient();
if (!client) throw new Error(WS_CLIENT_UNAVAILABLE);
const response = await client.request("task.runner", {
id: taskId,
executor_profile_id: executorProfileId,
});
return response as Task;
}
Next launch uses the new runner
apps/backend/internal/orchestrator/executor/task_runner_context.go ↗Session preparation detects a concurrent runner switch, reloads the task, and retries once so the new runner takes effect.
Retry on runner change
func (e *Executor) prepareSession(ctx context.Context, task *v1.Task, agentProfileID string, executorID string, executorProfileID string, workflowStepID string, bindWorkspace bool, taskEnvironmentID string, workflowRoute *models.WorkflowSessionRoute) (string, error) {
runnerProfileExplicit := taskRunnerProfileExplicit(ctx, strings.TrimSpace(executorProfileID) != "")
currentTask := task
currentExecutorProfileID := executorProfileID
for attempt := 0; attempt < 2; attempt++ {
sessionID, err := e.prepareSessionAttempt(ctx, currentTask, agentProfileID, executorID, currentExecutorProfileID, workflowStepID, bindWorkspace, taskEnvironmentID, workflowRoute)
if err == nil || !errors.Is(err, models.ErrTaskRunnerChanged) || runnerProfileExplicit || attempt == 1 {
return sessionID, err
}
refreshed, refreshedExecutorProfileID, refreshErr := e.reloadTaskForRunnerRetry(ctx, task.ID)
currentTask = refreshed
currentExecutorProfileID = refreshedExecutorProfileID
}
return "", models.ErrTaskRunnerChanged
}
Context flag
func WithTaskRunnerProfileExplicit(ctx context.Context, explicit bool) context.Context {
return context.WithValue(ctx, taskRunnerProfileExplicitKey{}, explicit)
}
func taskRunnerResolution(task *v1.Task, executorProfileID string, explicit bool) (bool, string) {
resolvedProfileID := taskRunnerProfileID(task)
return !explicit && executorProfileID != "" && executorProfileID == resolvedProfileID, resolvedProfileID
}
