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 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)
	}
	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
		project.Works["work"] = work
		state.Projects["project"] = project
	})
	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)
	}
}