orchestra/internal/operations/retry.go

package operations

import (
	"encoding/json"
	"errors"
	"fmt"

	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"orchestra/internal/store"
)

// ErrNotRetryable reports that a task cannot be given another retry budget.
// Retry is deliberately narrow: it recovers a task the router gave up on, and
// nothing else.
var ErrNotRetryable = errors.New("task is not retryable")

// RetryTask gives a retry-exhausted task a fresh attempt budget.
//
// A task that reaches the router's MaxAttempts is terminal, and nothing in the
// log resets Attempt, so before this the only way to work an exhausted issue
// again was to invent a second task for it. That defeats (source, external_id)
// dedupe and loses the task's own history. This keeps the task: its id, its
// source pair, its goal, acceptance, decisions, work phase and artifact refs
// all stay as they are, and the original failure events stay in the log.
//
// This is not an escape hatch for a live lease. An owned task still has to be
// released by its owner (F9); retrying a terminal task is a different
// operation and stays separate on purpose.
//
// operationID makes the call idempotent. A retry that already ran under the
// same id returns its event instead of resetting an attempt that has since
// started running.
func RetryTask(s *store.Store, surface authz.Surface, taskID, operationID string) (domain.Event, error) {
	if err := authz.AuthorizeEvent(surface, "TaskCorrected"); err != nil {
		return domain.Event{}, err
	}
	if operationID == "" {
		return domain.Event{}, fmt.Errorf("%w: operation_id required", domain.ErrInvalid)
	}
	if e, ok := retryOperation(s, taskID, operationID); ok {
		return e, nil
	}
	t, ok := s.Task(taskID)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if t.State != domain.StateFailed {
		return domain.Event{}, fmt.Errorf("%w: task %s is %s, not failed", ErrNotRetryable, taskID, t.State)
	}
	if t.Lease != nil {
		return domain.Event{}, fmt.Errorf("%w: task %s still holds a lease owned by %s", ErrNotRetryable, taskID, t.Lease.HarnessID)
	}
	failure, reason := lastFailure(s, taskID)
	if failure.ID == "" {
		return domain.Event{}, fmt.Errorf("%w: task %s has no failure to correct", ErrNotRetryable, taskID)
	}
	if reason != "retry_limit" {
		// Retry restores a retry budget. A failure that was not the budget
		// running out needs a decision about the failure itself, not another
		// identical attempt.
		return domain.Event{}, fmt.Errorf("%w: task %s failed with %q, not retry exhaustion", ErrNotRetryable, taskID, reason)
	}
	b, err := json.Marshal(map[string]any{
		"corrects":      failure.ID,
		"operation_id":  operationID,
		"state":         string(domain.StateQueued),
		"attempt":       0,
		"next_retry_at": "",
		"failure_class": "",
		"last_error":    "",
	})
	if err != nil {
		return domain.Event{}, err
	}
	e := domain.Event{ID: domain.NewID(), Type: "TaskCorrected", TaskID: taskID, Version: t.Version + 1, Payload: b, Surface: string(surface)}
	if err := s.Append(e); err != nil {
		return domain.Event{}, err
	}
	return e, nil
}

// retryOperation finds a retry already recorded under this operation id.
func retryOperation(s *store.Store, taskID, operationID string) (domain.Event, bool) {
	for _, e := range s.Events(0) {
		if e.TaskID != taskID || e.Type != "TaskCorrected" {
			continue
		}
		var p struct {
			OperationID string `json:"operation_id"`
		}
		if json.Unmarshal(e.Payload, &p) == nil && p.OperationID != "" && p.OperationID == operationID {
			return e, true
		}
	}
	return domain.Event{}, false
}

// lastFailure returns the most recent TaskFailed event and its reason.
func lastFailure(s *store.Store, taskID string) (domain.Event, string) {
	var found domain.Event
	var reason string
	for _, e := range s.Events(0) {
		if e.TaskID != taskID || e.Type != "TaskFailed" {
			continue
		}
		var p struct {
			Reason string `json:"reason"`
		}
		_ = json.Unmarshal(e.Payload, &p)
		found, reason = e, p.Reason
	}
	return found, reason
}