fix(orchestrator): retry on_agent_error recovery after a failed commit instead of giving up ↗

Engine defers the mark for EvaluateOnly transitions

The engine now skips MarkOperationApplied when EvaluateOnly produced a transition and tells the caller it owns the mark.

type HandleResult struct {
  Transitioned bool
  FromStepID   string
  ToStepID     string
  Guards      []QuorumGuardState
  DataPatch   map[string]any
  Idempotent  bool
  ActionCount int
  TransitionAbandoned bool
  OperationMarkDeferred bool
}

  result, err := e.processActions(ctx, in, state, step, actions, filter)
  if err != nil {
    return HandleResult{}, err
  }

  // A deferred transition is work this call evaluated but did not commit
  // (processActions skips ApplyTransition under EvaluateOnly). Marking the
  // operation applied here would claim a commit the caller still owes, so
  // ownership of the marker passes to the caller instead.
  if in.EvaluateOnly && result.Transitioned {
    result.OperationMarkDeferred = in.OperationID != ""
    return result, nil
  }

  return result, e.markOperationAppliedForInput(ctx, in)

Per-operation lock for on_agent_error

The lock serializes the full load-evaluate-commit-mark window for one operation id so a blocked racer reloads fresh state.

type agentErrorOperationLock struct {
  mu   sync.Mutex
  refs int
}

func (s *Service) lockAgentErrorOperation(operationID string) func() {
  s.agentErrorOperationLocksMu.Lock()
  if s.agentErrorOperationLocks == nil {
    s.agentErrorOperationLocks = make(map[string]*agentErrorOperationLock)
  }
  entry := s.agentErrorOperationLocks[operationID]
  if entry == nil {
    entry = &agentErrorOperationLock{}
    s.agentErrorOperationLocks[operationID] = entry
  }
  entry.refs++
  s.agentErrorOperationLocksMu.Unlock()

  entry.mu.Lock()
  return func() {
    s.agentErrorOperationLocksMu.Lock()
    entry.refs--
    if entry.refs == 0 {
      delete(s.agentErrorOperationLocks, operationID)
    }
    s.agentErrorOperationLocksMu.Unlock()
    entry.mu.Unlock()
  }
}

Dispatch now commits before it marks

The dispatch acquires the lock before any load, captures the commit result, and marks only after a successful commit.

func (s *Service) dispatchKanbanAgentErrorTrigger(ctx context.Context, data watcher.AgentEventData) {
  deps := s.agentErrorDeps.Load()
  if deps == nil || deps.engine == nil {
    return
  }
  if data.SessionID == "" {
    return
  }
  if data.UserInitiated {
    s.logger.Debug(msgAgentErrorUserInitiated,
      zap.String("task_id", data.TaskID),
      zap.String("session_id", data.SessionID))
    return
  }

  operationID := agentErrorOperationID(data.SessionID, data.AgentExecutionID)
  unlock := s.lockAgentErrorOperation(operationID)
  defer unlock()

  session, task, ok := s.resolveAgentErrorDispatchTarget(ctx, data)
  if !ok {
    return
  }

  state := s.buildMachineState(ctx, task, session)

  if result.Transitioned {
    applied := s.applyEngineTransition(ctx, data.TaskID, session, result, engine.TriggerOnAgentError, task.Description, true)
    if applied && result.OperationMarkDeferred {
      if markErr := deps.store.MarkOperationApplied(ctx, operationID); markErr != nil {
        s.logger.Warn(msgAgentErrorMarkFailed,
          zap.String("task_id", data.TaskID),
          zap.String("session_id", data.SessionID),
          zap.String("operation_id", operationID),
          zap.Error(markErr))
      }
    }
  }

Separate lock map on Service

A dedicated map keeps on_agent_error contention separate from on_children_completed contention.

  // agentErrorOperationLocks serializes concurrent on_agent_error dispatches
  // that carry the same operation id — the load -> evaluate -> commit ->
  // mark window, held from before the task/session/MachineState load so a
  // blocked racer always reloads state rather than evaluating a
  // PreloadedState built before another racer's commit.
  agentErrorOperationLocksMu sync.Mutex
  agentErrorOperationLocks   map[string]*agentErrorOperationLock

Pin test for future EvaluateOnly call sites

The test scans the backend for HandleInput{EvaluateOnly:true, OperationID:non-empty} and fails when a new site is not in the allowlist.

var registeredEvaluateOnlyOperationMarkingSites = []string{
  "internal/orchestrator/Service.dispatchKanbanAgentErrorTrigger",
  "internal/orchestrator/Service.evaluateChildrenCompleted",
}

func functionPairsEvaluateOnlyWithOperationID(body *ast.BlockStmt, engineAlias string, isEnginePackage bool) bool {
  pairs := false
  ast.Inspect(body, func(n ast.Node) bool {
    lit, ok := n.(*ast.CompositeLit)
    if !ok {
      return true
    }
    if !isHandleInputLitType(lit.Type, engineAlias, isEnginePackage) {
      return true
    }
    hasEvaluateOnlyTrue := false
    hasNonEmptyOperationID := false
    for _, elt := range lit.Elts {
      kv, ok := elt.(*ast.KeyValueExpr)
      if !ok {
        continue
      }
      key, ok := kv.Key.(*ast.Ident)
      if !ok {
        continue
      }
      switch key.Name {
      case "EvaluateOnly":
        if ident, ok := kv.Value.(*ast.Ident); ok && ident.Name == "true" {
          hasEvaluateOnlyTrue = true
        }
      case "OperationID":
        if !isEmptyStringLiteral(kv.Value) {
          hasNonEmptyOperationID = true
        }
      }
    }
    if hasEvaluateOnlyTrue && hasNonEmptyOperationID {
      pairs = true
    }
    return true
  })
  return pairs
}