orchestra/internal/human/reconcile_test.go

package human

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

type fakeSource struct {
	inputs []Input
	next   string
	err    error
	calls  int
	seen   []store.SourceCursor
}

func (f *fakeSource) FetchAfter(_ context.Context, task domain.Task, cursor store.SourceCursor) ([]Input, store.SourceCursor, error) {
	f.calls++
	f.seen = append(f.seen, cursor)
	if f.err != nil {
		return nil, store.SourceCursor{}, f.err
	}
	return f.inputs, store.SourceCursor{TaskID: task.ID, Provider: "gitea", Cursor: f.next}, nil
}

func input(id, body string) Input {
	return Input{Provider: "gitea", ExternalID: id, Author: "kami", At: time.Unix(1700000000, 0).UTC(), Body: body}
}

func setup(t *testing.T) (string, *store.Store, domain.Task) {
	t.Helper()
	dir := t.TempDir()
	s, err := store.Open(dir)
	if err != nil {
		t.Fatal(err)
	}
	// Ingested directly: provider imports this package, so the test cannot.
	created := []byte(`{"source":"gitea","external_id":"381","project":"p"}`)
	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: domain.NewID(), Version: 1, Payload: created, Surface: string(authz.System)}); err != nil {
		t.Fatal(err)
	}
	tasks := s.Tasks()
	if len(tasks) != 1 {
		t.Fatalf("tasks=%d", len(tasks))
	}
	return dir, s, tasks[0]
}

func reconciler(s *store.Store, src Source) *Reconciler {
	return &Reconciler{Store: s, Sources: map[string]Source{"gitea": src}}
}

func TestNewCommentBecomesStandingDecision(t *testing.T) {
	_, s, task := setup(t)
	src := &fakeSource{inputs: []Input{input("918", "no, use b")}, next: "918"}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	intent, err := s.EffectiveIntent(task.ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(intent.Decisions) != 1 {
		t.Fatalf("standing set = %+v", intent.Decisions)
	}
	d := intent.Decisions[0]
	if d.Value != "no, use b" || d.Kind != domain.HumanDecisionCorrection || d.Subject != "operator_instruction" {
		t.Fatalf("decision = %+v", d)
	}
	if d.Source.Provider != "gitea" || d.Source.ExternalID != "918" {
		t.Fatalf("provenance = %+v", d.Source)
	}
	c, ok := s.SourceCursor(task.ID, "gitea")
	if !ok || c.Cursor != "918" {
		t.Fatalf("cursor = %+v ok=%v", c, ok)
	}
}

func TestNoNewInputIsANoOp(t *testing.T) {
	_, s, task := setup(t)
	before, _ := s.Task(task.ID)
	src := &fakeSource{}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	after, _ := s.Task(task.ID)
	if after.Version != before.Version {
		t.Fatalf("version moved %d -> %d", before.Version, after.Version)
	}
	if _, ok := s.SourceCursor(task.ID, "gitea"); ok {
		t.Fatal("cursor advanced with no input")
	}
}

func TestNoConfiguredSourcesProceeds(t *testing.T) {
	_, s, task := setup(t)
	r := &Reconciler{Store: s}
	if err := r.Reconcile(context.Background(), task.ID); err != nil {
		t.Fatalf("a deployment with no source configured must not be blocked: %v", err)
	}
}

func TestSameCommentTwiceYieldsOneDecision(t *testing.T) {
	_, s, task := setup(t)
	src := &fakeSource{inputs: []Input{input("918", "no, use b")}, next: "918"}
	r := reconciler(s, src)
	for i := 0; i < 3; i++ {
		if err := r.Reconcile(context.Background(), task.ID); err != nil {
			t.Fatal(err)
		}
	}
	intent, err := s.EffectiveIntent(task.ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(intent.Decisions) != 1 {
		t.Fatalf("want 1 decision after 3 reconciles, got %d", len(intent.Decisions))
	}
	if src.seen[1].Cursor != "918" {
		t.Fatalf("second fetch did not resume from the cursor: %+v", src.seen[1])
	}
}

func TestProviderFailureFailsClosed(t *testing.T) {
	_, s, task := setup(t)
	src := &fakeSource{err: errors.New("gitea unreachable")}
	err := reconciler(s, src).Reconcile(context.Background(), task.ID)
	if err == nil {
		t.Fatal("provider failure must not be swallowed")
	}
	if !strings.Contains(err.Error(), "gitea unreachable") {
		t.Fatalf("err = %v", err)
	}
	if _, ok := s.SourceCursor(task.ID, "gitea"); ok {
		t.Fatal("cursor advanced despite fetch failure")
	}
}

// A durable append is the precondition for advancing the cursor. If the log
// write fails, the input must be refetched on the next attempt.
func TestAppendFailureLeavesCursorInPlace(t *testing.T) {
	dir, s, task := setup(t)
	log := filepath.Join(dir, "events.jsonl")
	if err := os.Chmod(log, 0400); err != nil {
		t.Fatal(err)
	}
	src := &fakeSource{inputs: []Input{input("918", "no, use b")}, next: "918"}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err == nil {
		t.Fatal("append failure must fail reconciliation")
	}
	if _, ok := s.SourceCursor(task.ID, "gitea"); ok {
		t.Fatal("cursor advanced despite append failure")
	}
	if err := os.Chmod(log, 0644); err != nil {
		t.Fatal(err)
	}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	intent, _ := s.EffectiveIntent(task.ID)
	if len(intent.Decisions) != 1 || intent.Decisions[0].Value != "no, use b" {
		t.Fatalf("retry did not record the decision: %+v", intent.Decisions)
	}
}

// The reverse crash window: the decision is durable but the cursor write
// fails. Provenance uniqueness, not the cursor, is what stops the refetch
// from becoming a second copy of the same instruction.
func TestCursorWriteFailureDoesNotDuplicateDecision(t *testing.T) {
	dir, s, task := setup(t)
	// Occupying the cursor path with a directory makes the atomic rename fail.
	if err := os.Mkdir(filepath.Join(dir, "source-cursors.json"), 0755); err != nil {
		t.Fatal(err)
	}
	src := &fakeSource{inputs: []Input{input("918", "no, use b")}, next: "918"}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err == nil {
		t.Fatal("cursor write failure must be reported")
	}
	if _, ok := s.DecisionForSource("gitea", "918"); !ok {
		t.Fatal("decision should already be durable")
	}
	if err := os.Remove(filepath.Join(dir, "source-cursors.json")); err != nil {
		t.Fatal(err)
	}
	// Same range refetched, because the cursor never advanced.
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	intent, _ := s.EffectiveIntent(task.ID)
	if len(intent.Decisions) != 1 {
		t.Fatalf("want 1 decision, got %d", len(intent.Decisions))
	}
	if c, ok := s.SourceCursor(task.ID, "gitea"); !ok || c.Cursor != "918" {
		t.Fatalf("cursor = %+v ok=%v", c, ok)
	}
}

func TestBatchRecordsEveryInputInOrder(t *testing.T) {
	_, s, task := setup(t)
	src := &fakeSource{next: "920", inputs: []Input{
		{Provider: "gitea", ExternalID: "918", At: time.Unix(1700000000, 0).UTC(), Body: "use b"},
		{Provider: "gitea", ExternalID: "919", At: time.Unix(1700000060, 0).UTC(), Body: "   "},
		{Provider: "gitea", ExternalID: "920", At: time.Unix(1700000120, 0).UTC(), Body: "and keep the old flag"},
	}}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	intent, _ := s.EffectiveIntent(task.ID)
	if len(intent.Decisions) != 2 {
		t.Fatalf("want 2 decisions, blank comment skipped: %+v", intent.Decisions)
	}
	if intent.Decisions[0].Value != "use b" || intent.Decisions[1].Value != "and keep the old flag" {
		t.Fatalf("order = %+v", intent.Decisions)
	}
}

func TestReconcileUnknownTask(t *testing.T) {
	_, s, _ := setup(t)
	src := &fakeSource{}
	if err := reconciler(s, src).Reconcile(context.Background(), "nope"); !errors.Is(err, domain.ErrNotFound) {
		t.Fatalf("want ErrNotFound, got %v", err)
	}
	if src.calls != 0 {
		t.Fatal("must not fetch for an unknown task")
	}
}

func TestInputWithoutExternalIDRejected(t *testing.T) {
	_, s, task := setup(t)
	src := &fakeSource{inputs: []Input{{Provider: "gitea", Body: "no id"}}}
	if err := reconciler(s, src).Reconcile(context.Background(), task.ID); !errors.Is(err, domain.ErrInvalid) {
		t.Fatalf("want ErrInvalid, got %v", err)
	}
}