orchestra/internal/human/reconcile.go

// Package human turns external human utterances into durable Orchestra
// decisions. It runs immediately before ownership of a task begins, so an
// agent can never resume from an older intent while newer human input is
// waiting in a configured source.
package human

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

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

// Input is one human utterance as the provider found it. It carries no
// Orchestra semantics on purpose: classifying it is this package's job, so
// provider code never has to know what a decision is.
type Input struct {
	Provider   string
	ExternalID string
	Author     string
	At         time.Time
	Body       string
}

// Source fetches the inputs a task has received after a cursor. It returns
// the inputs in the order the human wrote them, plus the cursor that covers
// them. The returned cursor is only persisted once every derived event is
// durable, so a Source must tolerate being asked for the same range twice.
type Source interface {
	FetchAfter(ctx context.Context, task domain.Task, cursor store.SourceCursor) ([]Input, store.SourceCursor, error)
}

// Reconciler is the pre-launch step. Wire it to Store.PreLease.
type Reconciler struct {
	Store *store.Store
	// Sources is keyed by provider name, which is also the provider half of
	// the (provider, external_id) provenance key.
	Sources map[string]Source
	Timeout time.Duration
	// Now exists for tests. Reconciliation stamps nothing itself, but the
	// classifier records when Orchestra observed the input.
	Now func() time.Time
}

// operatorInstructionSubject is the single subject every imported comment
// lands under until extraction exists. Crude, and mechanically correct: the
// text is preserved verbatim and outranks handoff prose because it is a
// decision and the handoff is not.
const operatorInstructionSubject = "operator_instruction"

// Reconcile imports every input newer than the stored cursor, then advances
// the cursor. It fails closed: any provider or append error returns an error
// and leaves the cursor where it was, so the caller refuses the launch and a
// later attempt refetches the same range.
func (r *Reconciler) Reconcile(ctx context.Context, taskID string) error {
	if r == nil || r.Store == nil || len(r.Sources) == 0 {
		return nil
	}
	task, ok := r.Store.Task(taskID)
	if !ok {
		return domain.ErrNotFound
	}
	if r.Timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, r.Timeout)
		defer cancel()
	}
	// Only the source this task came from may reconcile it. A source is
	// identified by the same string the ingest stamped on the task
	// (provider:project, e.g. "gitea:test-e2e"), which binds provider,
	// instance and repository together.
	//
	// Iterating every configured source was wrong and not merely noisy: a
	// task's external id was looked up in whatever repository each source
	// happened to point at, so once two repositories used the same issue
	// number, an unrelated human comment became an authoritative decision for
	// the wrong task. Found during burn-in with three correx tasks being
	// reconciled against kami/test-e2e.
	//
	// A source that cannot prove it owns the task is skipped, not guessed at.
	// Nothing to import is not the same as a failure to read, so a task with no
	// matching source reconciles to nothing and the launch proceeds.
	src, ok := r.Sources[task.Source]
	if !ok {
		return nil
	}
	if err := r.reconcileSource(ctx, task, task.Source, src); err != nil {
		return fmt.Errorf("%s: %w", task.Source, err)
	}
	return nil
}

func (r *Reconciler) reconcileSource(ctx context.Context, task domain.Task, provider string, src Source) error {
	cursor, _ := r.Store.SourceCursor(task.ID, provider)
	inputs, next, err := src.FetchAfter(ctx, task, cursor)
	if err != nil {
		return err
	}
	for _, in := range inputs {
		if in.ExternalID == "" {
			return fmt.Errorf("%w: input without external id", domain.ErrInvalid)
		}
		// An empty utterance decides nothing. Skipping it still advances the
		// cursor past it, so it is read once and never again.
		if strings.TrimSpace(in.Body) == "" {
			continue
		}
		// A refetch after a lost cursor write must not duplicate the
		// decision. The store rejects it too; checking first keeps the
		// ordinary resume path free of expected errors.
		if _, exists := r.Store.DecisionForSource(provider, in.ExternalID); exists {
			continue
		}
		if err := r.record(task, provider, in); err != nil && !errors.Is(err, domain.ErrDuplicate) {
			return err
		}
	}
	// Only now: every event derived from this range is durable.
	next.TaskID, next.Provider = task.ID, provider
	if next.Cursor == "" || next.Cursor == cursor.Cursor {
		return nil
	}
	return r.Store.SetSourceCursor(next)
}

func (r *Reconciler) record(task domain.Task, provider string, in Input) error {
	at := in.At
	if at.IsZero() {
		at = r.now()
	}
	current, ok := r.Store.Task(task.ID)
	if !ok {
		return domain.ErrNotFound
	}
	payload := map[string]any{
		"decision_id": domain.NewID(),
		"kind":        string(domain.HumanDecisionCorrection),
		"subject":     operatorInstructionSubject,
		"value":       in.Body,
		"source":      map[string]any{"provider": provider, "external_id": in.ExternalID},
		"author":      in.Author,
	}
	b, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	return r.Store.Append(domain.Event{
		ID: domain.NewID(), Type: domain.EventHumanDecisionRecorded, TaskID: task.ID,
		Version: current.Version + 1, At: at, Payload: b, Surface: string(authz.System),
	})
}

func (r *Reconciler) now() time.Time {
	if r.Now != nil {
		return r.Now()
	}
	return time.Now().UTC()
}