orchestra/internal/human/reconcile_test.go

package human

import (
	"context"
	"encoding/json"
	"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)
	}
}

// Only the source a task came from may reconcile it. Iterating every configured
// source looked up a task's external id in whatever repository each source
// pointed at, so a colliding issue number would turn an unrelated human comment
// into an authoritative decision for the wrong task.
func TestReconcileUsesOnlyTheTaskOwnSource(t *testing.T) {
	s, err := store.Open(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	b, _ := json.Marshal(map[string]any{"source": "gitea:correx", "external_id": "17", "project": "correx"})
	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: "t1", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil {
		t.Fatal(err)
	}
	owner := &fakeSource{inputs: []Input{{Provider: "gitea:correx", ExternalID: "c1", Author: "kami", Body: "use b"}}, next: "c1"}
	other := &fakeSource{inputs: []Input{{Provider: "gitea:test-e2e", ExternalID: "x9", Author: "kami", Body: "delete everything"}}, next: "x9"}
	r := &Reconciler{Store: s, Sources: map[string]Source{"gitea:correx": owner, "gitea:test-e2e": other}}
	if err := r.Reconcile(context.Background(), "t1"); err != nil {
		t.Fatal(err)
	}
	if other.calls != 0 {
		t.Fatalf("a foreign source was asked about this task %d times", other.calls)
	}
	if owner.calls != 1 {
		t.Fatalf("the owning source was called %d times", owner.calls)
	}
	intent, err := s.EffectiveIntent("t1")
	if err != nil {
		t.Fatal(err)
	}
	if len(intent.Decisions) != 1 || intent.Decisions[0].Value != "use b" {
		t.Fatalf("standing set = %+v", intent.Decisions)
	}

	// A task whose source is not configured reconciles to nothing. Nothing to
	// import is not a failure to read, so the launch must not be refused.
	b2, _ := json.Marshal(map[string]any{"source": "jsonl", "external_id": "5", "project": "correx"})
	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: "t2", Version: 1, Payload: b2, Surface: string(authz.System)}); err != nil {
		t.Fatal(err)
	}
	if err := r.Reconcile(context.Background(), "t2"); err != nil {
		t.Fatalf("unconfigured source refused the launch: %v", err)
	}
	if other.calls != 0 || owner.calls != 1 {
		t.Fatalf("an unowned task reached a source: owner %d, other %d", owner.calls, other.calls)
	}
}

// A manual plan-phase gate is satisfied by a keyed comment and by nothing
// else. Without a producer for this subject the gate was unreachable: every
// imported comment landed under operator_instruction, so a phase with a
// manual step could never become verified.
func TestOnlyAKeyedCommentApprovesAPlanPhase(t *testing.T) {
	task := domain.Task{ID: "t1", PlanRef: "plan-ref-a"}
	want := domain.PlanPhaseSubject("plan-ref-a", "phase-2")
	for _, c := range []struct {
		body    string
		subject string
	}{
		{"looks good to me", operatorInstructionSubject},
		{"orchestra verify phase-2", want},
		{"Orchestra Verify phase-2", want},
		{"orchestra verify phase-2\n\nthe json line read right", want},
		{"orchestra verify ", operatorInstructionSubject},
		{"please orchestra verify phase-2", operatorInstructionSubject},
	} {
		if got := decisionSubject(task, c.body); got != c.subject {
			t.Errorf("decisionSubject(%q) = %q, want %q", c.body, got, c.subject)
		}
	}
	// No accepted plan means no phase to approve.
	if got := decisionSubject(domain.Task{ID: "t1"}, "orchestra verify phase-2"); got != operatorInstructionSubject {
		t.Errorf("without a plan ref: got %q", got)
	}
	// The subject binds the plan, so the same words under a later plan do not
	// satisfy a gate earned under the earlier one.
	if decisionSubject(domain.Task{ID: "t1", PlanRef: "plan-ref-b"}, "orchestra verify phase-2") == want {
		t.Error("a different plan ref produced the same subject")
	}
}