iop/packages/go/agenttask/integration_queue_test.go

package agenttask

import (
	"context"
	"reflect"
	"testing"
	"time"
)

func TestIntegrationOutOfOrderWorkerCompletionUsesDispatchOrdinal(t *testing.T) {
	snapshot := testSnapshot(
		"project", "workspace",
		testUnit("a-first", WriteSetDisjoint),
		testUnit("b-second", WriteSetDisjoint),
	)
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	harness.invoker.delays["a-first"] = 35 * time.Millisecond
	harness.invoker.delays["b-second"] = 2 * time.Millisecond
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if ordinals := harness.integrator.ordinals(); !reflect.DeepEqual(
		ordinals, []DispatchOrdinal{1, 2},
	) {
		t.Fatalf("integration ordinals = %v, want [1 2]", ordinals)
	}
}

func TestIntegrationTerminalDeferredAdvancesIndependentQueue(t *testing.T) {
	snapshot := testSnapshot(
		"project", "workspace",
		testUnit("a-blocked", WriteSetOverlap),
		testUnit("b-independent", WriteSetOverlap),
	)
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	harness.integrator.outcomes["a-blocked"] = IntegrationOutcomeTerminalDeferred
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	project := harness.store.snapshot().Projects["project"]
	if project.Works["a-blocked"].State != WorkStateTerminalDeferred {
		t.Fatalf("blocked work state = %s", project.Works["a-blocked"].State)
	}
	if project.Works["b-independent"].State != WorkStateCompleted {
		t.Fatalf("independent work state = %s", project.Works["b-independent"].State)
	}
	if ordinals := harness.integrator.ordinals(); !reflect.DeepEqual(
		ordinals, []DispatchOrdinal{1, 2},
	) {
		t.Fatalf("integration ordinals = %v", ordinals)
	}
}

func TestIntegrationCandidateWaitsForLowerOrdinalUntilTerminal(t *testing.T) {
	candidate := integrationCandidate{
		ProjectID: "project", WorkUnitID: "second",
		Workspace: "workspace", Ordinal: 2,
	}
	state := ManagerState{Projects: map[ProjectID]ProjectRecord{
		"project": {
			ProjectID: "project", WorkspaceID: "workspace",
			Works: map[WorkUnitID]WorkRecord{
				"first": {
					Unit: WorkUnit{ID: "first"}, State: WorkStateReviewing,
					DispatchOrdinal: 1,
				},
				"second": {
					Unit: WorkUnit{ID: "second"}, State: WorkStatePendingIntegration,
					DispatchOrdinal: 2,
				},
			},
		},
	}}
	if integrationCandidateReady(state, candidate) {
		t.Fatal("later ordinal became eligible while the first review was active")
	}
	project := state.Projects["project"]
	first := project.Works["first"]
	first.State = WorkStateTerminalDeferred
	project.Works["first"] = first
	state.Projects["project"] = project
	if !integrationCandidateReady(state, candidate) {
		t.Fatal("terminal-deferred predecessor did not release the later ordinal")
	}
}

func TestIntegrationCandidateRequiresTerminalLowerOrdinal(t *testing.T) {
	cases := []struct {
		name              string
		lowerState        WorkState
		lowerHasChangeSet bool
		wantReady         bool
	}{
		{"reviewing barrier", WorkStateReviewing, false, false},
		{"pending barrier", WorkStatePendingIntegration, true, false},
		{"blocked without change set barrier", WorkStateBlocked, false, false},
		{"blocked with change set barrier", WorkStateBlocked, true, false},
		{"stopped without change set barrier", WorkStateStopped, false, false},
		{"stopped with change set barrier", WorkStateStopped, true, false},
		{"completed releases", WorkStateCompleted, false, true},
		{"terminal-deferred releases", WorkStateTerminalDeferred, true, true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			first := WorkRecord{
				Unit: WorkUnit{ID: "first"}, State: tc.lowerState, DispatchOrdinal: 1,
			}
			if tc.lowerHasChangeSet {
				first.ChangeSet = &ChangeSetIdentity{
					ID: "change-first", Revision: "change-r1", ArtifactID: "art-first",
				}
			}
			state := ManagerState{Projects: map[ProjectID]ProjectRecord{
				"project": {
					ProjectID: "project", WorkspaceID: "workspace",
					Works: map[WorkUnitID]WorkRecord{
						"first": first,
						"second": {
							Unit:            WorkUnit{ID: "second"},
							State:           WorkStatePendingIntegration,
							DispatchOrdinal: 2,
						},
					},
				},
			}}
			candidate := integrationCandidate{
				ProjectID: "project", WorkUnitID: "second",
				Workspace: "workspace", Ordinal: 2,
			}
			if got := integrationCandidateReady(state, candidate); got != tc.wantReady {
				t.Fatalf("integrationCandidateReady(%s) = %v, want %v", tc.lowerState, got, tc.wantReady)
			}
		})
	}
}

func TestIntegrationStopResumePreservesOrdinal(t *testing.T) {
	snapshot := testSnapshot(
		"project", "workspace",
		testUnit("a-first", WriteSetDisjoint),
		testUnit("b-second", WriteSetDisjoint),
	)
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	harness.start("project", "workspace", nil)

	firstChangeSet := ChangeSetIdentity{
		ID: "change-a-first", Revision: "change-r1", ArtifactID: "art-a-first",
	}
	secondChangeSet := ChangeSetIdentity{
		ID: "change-b-second", Revision: "change-r1", ArtifactID: "art-b-second",
	}
	harness.store.edit(func(state *ManagerState) {
		state.NextOrdinal = 2
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning

		// The lower ordinal was halted mid-flight before producing a change set.
		first := project.Works["a-first"]
		first.Unit = testUnit("a-first", WriteSetDisjoint)
		first.State = WorkStateBlocked
		first.Attempt = 1
		first.AttemptID = attemptID("a-first", 1)
		first.DispatchOrdinal = 1
		first.Blocker = &Blocker{Code: BlockerProviderCapacity, Message: "halted", Retryable: true}
		project.Works["a-first"] = first

		// The higher ordinal already reviewed and is waiting to integrate.
		second := project.Works["b-second"]
		second.Unit = testUnit("b-second", WriteSetDisjoint)
		second.State = WorkStatePendingIntegration
		second.Attempt = 1
		second.AttemptID = attemptID("b-second", 1)
		second.DispatchOrdinal = 2
		second.ChangeSet = &secondChangeSet
		if second.Locators == nil {
			second.Locators = make(map[LocatorKind]LocatorRecord)
		}
		second.Locators[LocatorChangeSet] = locatorForChangeSet(project, second, secondChangeSet)
		project.Works["b-second"] = second

		state.Projects["project"] = project
	})

	// Phase 1: the higher ordinal must not integrate while the lower ordinal is
	// a non-terminal barrier, even though it currently holds no change set.
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("barrier Reconcile: %v", err)
	}
	if calls := harness.integrator.callCount(); calls != 0 {
		t.Fatalf("higher ordinal integrated ahead of a halted lower ordinal: %d calls", calls)
	}
	if state := harness.store.snapshot(); state.Projects["project"].Works["b-second"].State != WorkStatePendingIntegration {
		t.Fatalf(
			"higher ordinal left pending_integration = %s",
			state.Projects["project"].Works["b-second"].State,
		)
	}

	// Phase 2: the lower ordinal resumes and reaches pending integration; it must
	// integrate before the higher ordinal that was already waiting.
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		first := project.Works["a-first"]
		first.State = WorkStatePendingIntegration
		first.Blocker = nil
		first.ChangeSet = &firstChangeSet
		if first.Locators == nil {
			first.Locators = make(map[LocatorKind]LocatorRecord)
		}
		first.Locators[LocatorChangeSet] = locatorForChangeSet(project, first, firstChangeSet)
		project.Works["a-first"] = first
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("resume Reconcile: %v", err)
	}
	if ordinals := harness.integrator.ordinals(); !reflect.DeepEqual(
		ordinals, []DispatchOrdinal{1, 2},
	) {
		t.Fatalf("integration ordinals = %v, want [1 2]", ordinals)
	}
	final := harness.store.snapshot().Projects["project"].Works
	if final["a-first"].State != WorkStateCompleted ||
		final["b-second"].State != WorkStateCompleted {
		t.Fatalf(
			"final states a-first/b-second = %s/%s",
			final["a-first"].State,
			final["b-second"].State,
		)
	}
}

func TestRevisedChangeSetUsesNextIntegrationAttempt(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.integrator.outcomes["work"] = IntegrationOutcomeTerminalDeferred
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("first Reconcile: %v", err)
	}
	first := harness.store.snapshot().Projects["project"].Works["work"]
	if first.State != WorkStateTerminalDeferred ||
		first.IntegrationAttempt != 1 {
		t.Fatalf(
			"first state/attempt = %s/%d",
			first.State,
			first.IntegrationAttempt,
		)
	}

	harness.integrator.outcomes["work"] = IntegrationOutcomeIntegrated
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		work := project.Works["work"]
		work.State = WorkStatePendingIntegration
		revised := ChangeSetIdentity{
			ID: "change-revised", Revision: "change-r2",
			ArtifactID: work.ChangeSet.ArtifactID,
		}
		work.ChangeSet = &revised
		work.Locators[LocatorChangeSet] = locatorForChangeSet(project, work, revised)
		project.Works["work"] = work
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("revised Reconcile: %v", err)
	}
	revised := harness.store.snapshot().Projects["project"].Works["work"]
	if revised.State != WorkStateCompleted ||
		revised.IntegrationAttempt != 2 {
		t.Fatalf(
			"revised state/attempt = %s/%d",
			revised.State,
			revised.IntegrationAttempt,
		)
	}
	harness.integrator.mu.Lock()
	defer harness.integrator.mu.Unlock()
	if len(harness.integrator.actualCalls) != 2 ||
		harness.integrator.actualCalls[0].Attempt != 1 ||
		harness.integrator.actualCalls[1].Attempt != 2 {
		t.Fatalf(
			"integration attempts = %#v",
			harness.integrator.actualCalls,
		)
	}
}

func TestRestartReplayUsesStableIdempotencyKeysWithoutDuplicateCalls(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("first Reconcile: %v", err)
	}
	completedWork := harness.store.snapshot().Projects["project"].Works["work"]
	recoveredSubmission := *completedWork.Submission
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		work := project.Works["work"]
		work.State = WorkStateDispatching
		work.Submission = nil
		work.Review = nil
		work.ChangeSet = nil
		work.Integration = nil
		work.IntegrationAttempt = 0
		delete(work.Locators, LocatorChangeSet)
		delete(work.Locators, LocatorCompletion)
		project.Works["work"] = work
		state.Projects["project"] = project
	})
	harness.recovery.observations["work"] = RecoveryObservation{
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: completedWork.AttemptID,
		Execution: RecoveryExecutionSubmitted, Submission: &recoveredSubmission,
	}
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("replay Reconcile: %v", err)
	}
	if harness.invoker.callCount() != 1 ||
		harness.reviewer.callCount() != 1 ||
		harness.integrator.callCount() != 1 {
		t.Fatalf(
			"actual external calls after replay invoke/review/integrate = %d/%d/%d",
			harness.invoker.callCount(), harness.reviewer.callCount(), harness.integrator.callCount(),
		)
	}
	if harness.store.snapshot().Projects["project"].Works["work"].State != WorkStateCompleted {
		t.Fatalf("replayed work did not complete")
	}

	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		work := project.Works["work"]
		work.State = WorkStateIntegrating
		work.Integration = nil
		project.Works["work"] = work
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("integration replay Reconcile: %v", err)
	}
	if harness.integrator.callCount() != 1 {
		t.Fatalf(
			"integration crash-window replay made %d actual calls, want stable key with 1",
			harness.integrator.callCount(),
		)
	}
	if harness.store.snapshot().Projects["project"].Works["work"].State != WorkStateCompleted {
		t.Fatalf("integration replay did not complete")
	}
}

func TestIntegrationEventIdentityDistinguishesChangeSetsAttemptsAndReplays(t *testing.T) {
	base := Event{
		Type:               EventIntegrationResult,
		ProjectID:          "project",
		WorkspaceID:        "workspace",
		WorkUnitID:         "work",
		CommandID:          "cmd-1",
		WorkflowRevision:   "wfrev-1",
		AttemptID:          "att-1",
		Ordinal:            1,
		ChangeSetID:        "cs-1",
		ChangeSetRevision:  "csrev-1",
		IntegrationAttempt: 1,
		State:              WorkStateCompleted,
		Detail:             "integrated",
	}

	emit := func(e Event) string {
		var id string
		m := &Manager{clock: systemClock{}, events: &testSink{onEmit: func(event Event) { id = event.EventID }}}
		m.emit(context.Background(), e)
		return id
	}

	baseID := emit(base)

	diffChangeSetID := base
	diffChangeSetID.ChangeSetID = "cs-2"
	if emit(diffChangeSetID) == baseID {
		t.Fatalf("different ChangeSetID produced identical EventID: %q", baseID)
	}

	diffRevision := base
	diffRevision.ChangeSetRevision = "csrev-2"
	if emit(diffRevision) == baseID {
		t.Fatalf("different ChangeSetRevision produced identical EventID: %q", baseID)
	}

	diffAttempt := base
	diffAttempt.IntegrationAttempt = 2
	if emit(diffAttempt) == baseID {
		t.Fatalf("different IntegrationAttempt produced identical EventID: %q", baseID)
	}

	replayID := emit(base)
	if replayID != baseID {
		t.Fatalf("exact replay produced different EventID: %q vs %q", baseID, replayID)
	}
}