iop/packages/go/agentworkspace/integrator_test.go

package agentworkspace_test

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"io/fs"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"runtime"
	"sort"
	"strings"
	"sync/atomic"
	"testing"

	"iop/packages/go/agentconfig"
	"iop/packages/go/agentguard"
	"iop/packages/go/agentstate"
	"iop/packages/go/agenttask"
	"iop/packages/go/agentworkspace"
)

func TestFreezeChangeSetCapturesContentModeSymlinkAndDeletion(t *testing.T) {
	if runtime.GOOS == "windows" {
		t.Skip("mode and symlink operations require Unix filesystem semantics")
	}
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "modified.txt"), "base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "deleted.txt"), "delete\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "mode.sh"), "#!/bin/sh\n", 0o644)
	if err := os.Symlink("modified.txt", filepath.Join(fixture.baseRoot, "link")); err != nil {
		t.Fatalf("Symlink: %v", err)
	}
	fixture.commit("base")

	task := fixture.prepare("freeze", 1)
	writeFixtureFile(t, filepath.Join(task.view, "modified.txt"), "reviewed\n", 0o644)
	writeFixtureFile(t, filepath.Join(task.view, "added.txt"), "added\n", 0o644)
	if err := os.Remove(filepath.Join(task.view, "deleted.txt")); err != nil {
		t.Fatalf("remove deleted fixture: %v", err)
	}
	if err := os.Chmod(filepath.Join(task.view, "mode.sh"), 0o755); err != nil {
		t.Fatalf("chmod fixture: %v", err)
	}
	if err := os.Remove(filepath.Join(task.view, "link")); err != nil {
		t.Fatalf("remove link fixture: %v", err)
	}
	if err := os.Symlink("added.txt", filepath.Join(task.view, "link")); err != nil {
		t.Fatalf("replace link fixture: %v", err)
	}

	changeSet := task.freeze()
	operations := make(map[string]agentworkspace.ChangeOperation)
	for _, operation := range changeSet.Operations {
		operations[operation.Path] = operation
	}
	for path, want := range map[string]agentworkspace.ChangeOperationKind{
		"added.txt":    agentworkspace.ChangeOperationAdd,
		"deleted.txt":  agentworkspace.ChangeOperationDelete,
		"link":         agentworkspace.ChangeOperationModify,
		"mode.sh":      agentworkspace.ChangeOperationModify,
		"modified.txt": agentworkspace.ChangeOperationModify,
	} {
		if operation, ok := operations[path]; !ok || operation.Kind != want {
			t.Fatalf("operation %s = %#v, want %s", path, operation, want)
		}
	}
	if operations["link"].Result == nil ||
		operations["link"].Result.Kind != agentworkspace.SnapshotEntrySymlink ||
		operations["link"].Result.SymlinkTarget != "added.txt" {
		t.Fatalf("symlink operation = %#v", operations["link"])
	}
	if operations["mode.sh"].Result == nil ||
		os.FileMode(operations["mode.sh"].Result.Mode).Perm() != 0o755 {
		t.Fatalf("mode operation = %#v", operations["mode.sh"])
	}
	if changeSet.Identity().ArtifactID != task.artifactID ||
		!strings.HasPrefix(changeSet.Revision, "sha256:") {
		t.Fatalf("change-set identity = %#v", changeSet.Identity())
	}
	replayed := task.freeze()
	if !reflect.DeepEqual(replayed, changeSet) {
		t.Fatalf("idempotent freeze changed record\nfirst=%#v\nsecond=%#v", changeSet, replayed)
	}
}

func TestSerialIntegratorAppliesDisjointSetsAndReplaysAfterRestart(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "a.txt"), "a base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "b.txt"), "b base\n", 0o644)
	fixture.commit("base")

	first := fixture.prepare("first", 1)
	second := fixture.prepare("second", 2)
	writeFixtureFile(t, filepath.Join(first.view, "a.txt"), "a integrated\n", 0o644)
	writeFixtureFile(t, filepath.Join(second.view, "b.txt"), "b integrated\n", 0o644)
	firstChangeSet := first.freeze()
	secondChangeSet := second.freeze()

	var validations atomic.Int32
	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		validations.Add(1)
		return nil
	})
	firstResult := integrateTask(t, integrator, first, firstChangeSet, 1)
	secondResult := integrateTask(t, integrator, second, secondChangeSet, 1)
	if firstResult.Outcome != agenttask.IntegrationOutcomeIntegrated ||
		secondResult.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("integration outcomes = %s/%s", firstResult.Outcome, secondResult.Outcome)
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "a.txt"), "a integrated\n")
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "b.txt"), "b integrated\n")

	managerState, managerRevision, err := fixture.store.Load(context.Background())
	if err != nil {
		t.Fatalf("Load manager state: %v", err)
	}
	managerState.NextOrdinal = 99
	if _, err := fixture.store.CompareAndSwap(
		context.Background(),
		managerRevision,
		managerState,
	); err != nil {
		t.Fatalf("manager CAS after integration: %v", err)
	}
	reopenedStore, err := agentstate.NewStore(fixture.statePath)
	if err != nil {
		t.Fatalf("NewStore(reopen): %v", err)
	}
	restarted, err := agentworkspace.NewIntegrator(agentworkspace.IntegratorConfig{
		Backend: fixture.backend,
		Store:   reopenedStore,
		Validator: func(
			_ context.Context,
			_ agentworkspace.ValidationRequest,
		) error {
			validations.Add(1)
			return nil
		},
	})
	if err != nil {
		t.Fatalf("NewIntegrator(restart): %v", err)
	}
	replayed := integrateTask(t, restarted, second, secondChangeSet, 1)
	if replayed.Outcome != agenttask.IntegrationOutcomeIntegrated ||
		validations.Load() != 2 {
		t.Fatalf(
			"restart replay outcome/validations = %s/%d",
			replayed.Outcome,
			validations.Load(),
		)
	}
}

func TestSerialIntegratorThreeWayMergesManagedPredecessor(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(
		t,
		filepath.Join(fixture.baseRoot, "shared.txt"),
		"first base\nmiddle\nlast base\n",
		0o644,
	)
	fixture.commit("base")

	first := fixture.prepare("first", 1)
	second := fixture.prepare("second", 2)
	writeFixtureFile(
		t,
		filepath.Join(first.view, "shared.txt"),
		"first integrated\nmiddle\nlast base\n",
		0o644,
	)
	writeFixtureFile(
		t,
		filepath.Join(second.view, "shared.txt"),
		"first base\nmiddle\nlast integrated\n",
		0o644,
	)
	firstChangeSet := first.freeze()
	secondChangeSet := second.freeze()
	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		return nil
	})
	if result := integrateTask(t, integrator, first, firstChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("first outcome = %s", result.Outcome)
	}
	if result := integrateTask(t, integrator, second, secondChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("second outcome = %#v", result)
	}
	assertFixtureContent(
		t,
		filepath.Join(fixture.baseRoot, "shared.txt"),
		"first integrated\nmiddle\nlast integrated\n",
	)
}

func TestSerialIntegratorRetainsConflictAndAdvancesIndependentChangeSet(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "shared.txt"), "base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "other.txt"), "other base\n", 0o644)
	fixture.commit("base")

	first := fixture.prepare("first", 1)
	conflicting := fixture.prepare("conflicting", 2)
	independent := fixture.prepare("independent", 3)
	writeFixtureFile(t, filepath.Join(first.view, "shared.txt"), "first\n", 0o644)
	writeFixtureFile(t, filepath.Join(conflicting.view, "shared.txt"), "second\n", 0o644)
	writeFixtureFile(t, filepath.Join(independent.view, "other.txt"), "other integrated\n", 0o644)
	firstChangeSet := first.freeze()
	conflictingChangeSet := conflicting.freeze()
	independentChangeSet := independent.freeze()

	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		return nil
	})
	if result := integrateTask(t, integrator, first, firstChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("first outcome = %s", result.Outcome)
	}
	beforeConflict := workspaceDigest(t, fixture.baseRoot)
	conflictResult := integrateTask(t, integrator, conflicting, conflictingChangeSet, 1)
	if conflictResult.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		!conflictResult.Retained ||
		conflictResult.Blocker == nil ||
		!strings.Contains(conflictResult.Blocker.Message, "conflict") {
		t.Fatalf("conflict result = %#v", conflictResult)
	}
	if afterConflict := workspaceDigest(t, fixture.baseRoot); afterConflict != beforeConflict {
		t.Fatalf("conflict changed canonical digest: %s != %s", afterConflict, beforeConflict)
	}
	if _, err := os.Stat(conflictingChangeSet.Locator.Record); err != nil {
		t.Fatalf("retained change set: %v", err)
	}
	independentResult := integrateTask(
		t,
		integrator,
		independent,
		independentChangeSet,
		1,
	)
	if independentResult.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("independent outcome = %s", independentResult.Outcome)
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "shared.txt"), "first\n")
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "other.txt"), "other integrated\n")
}

func TestSerialIntegratorRejectsUnmanagedDriftWithoutMutation(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "target.txt"), "base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "unrelated.txt"), "stable\n", 0o644)
	fixture.commit("base")

	task := fixture.prepare("drift", 1)
	writeFixtureFile(t, filepath.Join(task.view, "target.txt"), "integrated\n", 0o644)
	changeSet := task.freeze()
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "unrelated.txt"), "external drift\n", 0o644)
	before := workspaceDigest(t, fixture.baseRoot)
	var validations atomic.Int32
	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		validations.Add(1)
		return nil
	})
	result := integrateTask(t, integrator, task, changeSet, 1)
	if result.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		result.Blocker == nil ||
		!strings.Contains(result.Blocker.Message, "unmanaged base drift") {
		t.Fatalf("drift result = %#v", result)
	}
	if after := workspaceDigest(t, fixture.baseRoot); after != before {
		t.Fatalf("drift rejection changed canonical digest: %s != %s", after, before)
	}
	if validations.Load() != 0 {
		t.Fatalf("validator calls = %d, want 0", validations.Load())
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "target.txt"), "base\n")
}

func TestSerialIntegratorRollsBackFailedValidationAndReplaysTerminalResult(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "modified.txt"), "base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "deleted.txt"), "keep\n", 0o644)
	fixture.commit("base")

	task := fixture.prepare("validation", 1)
	writeFixtureFile(t, filepath.Join(task.view, "modified.txt"), "candidate\n", 0o755)
	writeFixtureFile(t, filepath.Join(task.view, "added.txt"), "candidate\n", 0o644)
	if err := os.Remove(filepath.Join(task.view, "deleted.txt")); err != nil {
		t.Fatalf("remove task file: %v", err)
	}
	changeSet := task.freeze()
	before := workspaceDigest(t, fixture.baseRoot)
	var validations atomic.Int32
	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		validations.Add(1)
		return errorsForTest("fixture validation failure")
	})
	result := integrateTask(t, integrator, task, changeSet, 1)
	if result.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		result.BeforeRevision == "" ||
		result.AfterRevision != result.BeforeRevision ||
		result.Blocker == nil ||
		!strings.Contains(result.Blocker.Message, "validation failed") {
		t.Fatalf("validation result = %#v", result)
	}
	if after := workspaceDigest(t, fixture.baseRoot); after != before {
		t.Fatalf("validation rollback digest = %s, want %s", after, before)
	}
	replayed := integrateTask(t, integrator, task, changeSet, 1)
	if replayed.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		validations.Load() != 1 {
		t.Fatalf(
			"terminal replay outcome/validations = %s/%d",
			replayed.Outcome,
			validations.Load(),
		)
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "modified.txt"), "base\n")
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "deleted.txt"), "keep\n")
	if _, err := os.Stat(filepath.Join(fixture.baseRoot, "added.txt")); !os.IsNotExist(err) {
		t.Fatalf("rolled-back addition exists: %v", err)
	}
}

func TestSerialIntegratorRestartRecoversInterruptedApplyByRollback(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "target.txt"), "base\n", 0o644)
	fixture.commit("base")
	task := fixture.prepare("restart", 1)
	writeFixtureFile(t, filepath.Join(task.view, "target.txt"), "candidate\n", 0o644)
	changeSet := task.freeze()
	before := workspaceDigest(t, fixture.baseRoot)

	faultStore := &failNthIntegrationStore{delegate: fixture.store, failAt: 2}
	var validations atomic.Int32
	integrator, err := agentworkspace.NewIntegrator(agentworkspace.IntegratorConfig{
		Backend: fixture.backend,
		Store:   faultStore,
		Validator: func(
			_ context.Context,
			_ agentworkspace.ValidationRequest,
		) error {
			validations.Add(1)
			return nil
		},
	})
	if err != nil {
		t.Fatalf("NewIntegrator: %v", err)
	}
	request := integrationRequest(task, changeSet, 1)
	if _, err := integrator.Integrate(context.Background(), request); !errorsIsRevisionConflict(err) {
		t.Fatalf("interrupted Integrate error = %v", err)
	}
	if after := workspaceDigest(t, fixture.baseRoot); after != before {
		t.Fatalf("interrupted apply rollback digest = %s, want %s", after, before)
	}

	restarted := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		validations.Add(1)
		return nil
	})
	result, err := restarted.Integrate(context.Background(), request)
	if err != nil {
		t.Fatalf("restart Integrate: %v", err)
	}
	if result.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		result.BeforeRevision != result.AfterRevision ||
		result.Blocker == nil ||
		!strings.Contains(result.Blocker.Message, "restart recovered") {
		t.Fatalf("restart recovery result = %#v", result)
	}
	if validations.Load() != 1 {
		t.Fatalf("validator calls = %d, want 1", validations.Load())
	}
}

func TestSerialIntegratorPreservesManagedDescendantAcrossDirectoryDeletion(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "tree", "keep.txt"), "keep\n", 0o644)
	fixture.commit("base")

	first := fixture.prepare("first", 1)
	second := fixture.prepare("second", 2)
	// The managed predecessor adds an independent descendant under tree/.
	writeFixtureFile(t, filepath.Join(first.view, "tree", "new.txt"), "new\n", 0o644)
	// The later change set deletes the whole tree/ directory it saw in its base.
	if err := os.RemoveAll(filepath.Join(second.view, "tree")); err != nil {
		t.Fatalf("remove tree fixture: %v", err)
	}
	firstChangeSet := first.freeze()
	secondChangeSet := second.freeze()

	integrator := fixture.integrator(func(
		_ context.Context,
		_ agentworkspace.ValidationRequest,
	) error {
		return nil
	})
	if result := integrateTask(t, integrator, first, firstChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("first outcome = %s", result.Outcome)
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "tree", "new.txt"), "new\n")

	if result := integrateTask(t, integrator, second, secondChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("second outcome = %#v", result)
	}
	// The predecessor's independent descendant must survive the directory delete.
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "tree", "new.txt"), "new\n")
	// The change set's own descendant is deleted as requested.
	if _, err := os.Stat(filepath.Join(fixture.baseRoot, "tree", "keep.txt")); !os.IsNotExist(err) {
		t.Fatalf("owned descendant was not deleted: %v", err)
	}
	// The container is preserved because it still holds independent content.
	if info, err := os.Stat(filepath.Join(fixture.baseRoot, "tree")); err != nil || !info.IsDir() {
		t.Fatalf("preserved container = %#v, err=%v", info, err)
	}
}

func TestSerialIntegratorReplacesNonEmptyDirectoryByType(t *testing.T) {
	if runtime.GOOS == "windows" {
		t.Skip("symlink replacement requires Unix filesystem semantics")
	}
	testCases := []struct {
		name    string
		prepare func(*testing.T, string)
		assert  func(*testing.T, string)
	}{
		{
			name: "regular",
			prepare: func(t *testing.T, path string) {
				writeFixtureFile(t, path, "flat\n", 0o644)
			},
			assert: func(t *testing.T, path string) {
				info, err := os.Lstat(path)
				if err != nil || !info.Mode().IsRegular() {
					t.Fatalf("regular replacement = %#v, err=%v", info, err)
				}
				assertFixtureContent(t, path, "flat\n")
			},
		},
		{
			name: "symlink",
			prepare: func(t *testing.T, path string) {
				if err := os.Symlink("replacement.txt", path); err != nil {
					t.Fatalf("create replacement symlink: %v", err)
				}
			},
			assert: func(t *testing.T, path string) {
				info, err := os.Lstat(path)
				if err != nil || info.Mode()&os.ModeSymlink == 0 {
					t.Fatalf("symlink replacement = %#v, err=%v", info, err)
				}
				target, err := os.Readlink(path)
				if err != nil || target != "replacement.txt" {
					t.Fatalf("symlink target = %q, err=%v", target, err)
				}
			},
		},
	}
	for _, testCase := range testCases {
		t.Run(testCase.name, func(t *testing.T) {
			fixture := newIntegrationFixture(t)
			writeFixtureFile(
				t,
				filepath.Join(fixture.baseRoot, "tree", "child.txt"),
				"base\n",
				0o644,
			)
			writeFixtureFile(
				t,
				filepath.Join(fixture.baseRoot, "replacement.txt"),
				"replacement target\n",
				0o644,
			)
			fixture.commit("base")

			task := fixture.prepare("replace-tree-"+testCase.name, 1)
			if err := os.RemoveAll(filepath.Join(task.view, "tree")); err != nil {
				t.Fatalf("remove task tree: %v", err)
			}
			testCase.prepare(t, filepath.Join(task.view, "tree"))
			changeSet := task.freeze()

			var validations atomic.Int32
			integrator := fixture.integrator(func(
				_ context.Context,
				request agentworkspace.ValidationRequest,
			) error {
				validations.Add(1)
				testCase.assert(t, filepath.Join(request.ValidationRoot, "tree"))
				return nil
			})
			result := integrateTask(t, integrator, task, changeSet, 1)
			if result.Outcome != agenttask.IntegrationOutcomeIntegrated {
				t.Fatalf("integration result = %#v", result)
			}
			if validations.Load() != 1 {
				t.Fatalf("validator calls = %d, want 1", validations.Load())
			}
			testCase.assert(t, filepath.Join(fixture.baseRoot, "tree"))
			if _, err := os.Lstat(filepath.Join(fixture.baseRoot, "tree", "child.txt")); err == nil {
				t.Fatal("owned descendant was not removed")
			}
		})
	}
}

func TestSerialIntegratorValidationMutationDoesNotEscapeCandidateView(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "target.txt"), "base\n", 0o644)
	writeFixtureFile(t, filepath.Join(fixture.baseRoot, "unrelated.txt"), "stable\n", 0o644)
	fixture.commit("base")

	task := fixture.prepare("mutate", 1)
	writeFixtureFile(t, filepath.Join(task.view, "target.txt"), "integrated\n", 0o644)
	changeSet := task.freeze()
	before := workspaceDigest(t, fixture.baseRoot)

	var seenValidationRoot string
	integrator := fixture.integrator(func(
		_ context.Context,
		request agentworkspace.ValidationRequest,
	) error {
		seenValidationRoot = request.ValidationRoot
		// A misbehaving validator writes outside the change-set write set and
		// then fails; the mutation must never reach the canonical workspace.
		for name, content := range map[string]string{
			"escape.txt":    "leak\n",
			"unrelated.txt": "mutated\n",
		} {
			if err := os.WriteFile(
				filepath.Join(request.ValidationRoot, name),
				[]byte(content),
				0o644,
			); err != nil {
				t.Fatalf("validator side-effect write %s: %v", name, err)
			}
		}
		return errorsForTest("validation failure after side effects")
	})
	result := integrateTask(t, integrator, task, changeSet, 1)
	if result.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		result.BeforeRevision == "" ||
		result.AfterRevision != result.BeforeRevision ||
		result.Blocker == nil ||
		!strings.Contains(result.Blocker.Message, "validation failed") {
		t.Fatalf("validation result = %#v", result)
	}
	if seenValidationRoot == "" || seenValidationRoot == fixture.baseRoot {
		t.Fatalf(
			"validation root = %q, want a candidate outside %q",
			seenValidationRoot,
			fixture.baseRoot,
		)
	}
	if after := workspaceDigest(t, fixture.baseRoot); after != before {
		t.Fatalf("validator side effect changed canonical digest: %s != %s", after, before)
	}
	if _, err := os.Stat(filepath.Join(fixture.baseRoot, "escape.txt")); !os.IsNotExist(err) {
		t.Fatalf("validator escape survived in canonical: %v", err)
	}
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "unrelated.txt"), "stable\n")
	assertFixtureContent(t, filepath.Join(fixture.baseRoot, "target.txt"), "base\n")
}

func TestSerialIntegratorRevisedChangeSetAfterConflictAndLaterAdvance(t *testing.T) {
	fixture := newIntegrationFixture(t)
	writeFixtureFile(
		t,
		filepath.Join(fixture.baseRoot, "shared.txt"),
		"line1\nline2\nline3\n",
		0o644,
	)
	fixture.commit("base")

	predecessor := fixture.prepare("predecessor", 1)
	work := fixture.prepare("rework", 2)
	writeFixtureFile(
		t,
		filepath.Join(predecessor.view, "shared.txt"),
		"LINE1\nline2\nline3\n",
		0o644,
	)
	predecessorChangeSet := predecessor.freeze()

	integrator := fixture.integrator(func(
		_ context.Context,
		request agentworkspace.ValidationRequest,
	) error {
		if request.ValidationRoot == "" || request.ValidationRoot == request.CanonicalRoot {
			t.Errorf(
				"validation root %q must be a candidate outside %q",
				request.ValidationRoot,
				request.CanonicalRoot,
			)
		}
		return nil
	})
	if result := integrateTask(t, integrator, predecessor, predecessorChangeSet, 1); result.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("predecessor outcome = %s", result.Outcome)
	}

	// The first attempt conflicts with the managed head on line 1 and is retained.
	writeFixtureFile(
		t,
		filepath.Join(work.view, "shared.txt"),
		"XXXX\nline2\nline3\n",
		0o644,
	)
	firstChangeSet := work.freeze()
	conflict := integrateTask(t, integrator, work, firstChangeSet, 1)
	if conflict.Outcome != agenttask.IntegrationOutcomeTerminalDeferred ||
		!conflict.Retained ||
		conflict.Blocker == nil ||
		!strings.Contains(conflict.Blocker.Message, "conflict") {
		t.Fatalf("conflict result = %#v", conflict)
	}

	// The reworked immutable change set edits an independent line and merges
	// cleanly at the next integration attempt without replaying the first.
	writeFixtureFile(
		t,
		filepath.Join(work.view, "shared.txt"),
		"line1\nline2\nLINE3\n",
		0o644,
	)
	revisedChangeSet := work.freeze()
	if revisedChangeSet.Identity() == firstChangeSet.Identity() {
		t.Fatalf("revised change set kept the retained identity")
	}
	revised := integrateTask(t, integrator, work, revisedChangeSet, 2)
	if revised.Outcome != agenttask.IntegrationOutcomeIntegrated {
		t.Fatalf("revised outcome = %#v", revised)
	}

	assertFixtureContent(
		t,
		filepath.Join(fixture.baseRoot, "shared.txt"),
		"LINE1\nline2\nLINE3\n",
	)
	if _, err := os.Stat(firstChangeSet.Locator.Record); err != nil {
		t.Fatalf("retained first attempt change set is missing: %v", err)
	}
}

type integrationFixture struct {
	t         *testing.T
	baseRoot  string
	localRoot string
	statePath string
	backend   *agentworkspace.Backend
	store     *agentstate.Store
}

type preparedTask struct {
	fixture    *integrationFixture
	request    agenttask.IsolationRequest
	prepared   agenttask.PreparedIsolation
	view       string
	artifactID agenttask.ArtifactID
	ordinal    agenttask.DispatchOrdinal
}

type failNthIntegrationStore struct {
	delegate agentworkspace.IntegrationRecordStore
	failAt   int
	calls    int
}

func (store *failNthIntegrationStore) LoadIntegrationRecord(
	ctx context.Context,
	key string,
) ([]byte, string, bool, error) {
	return store.delegate.LoadIntegrationRecord(ctx, key)
}

func (store *failNthIntegrationStore) CompareAndSwapIntegrationRecord(
	ctx context.Context,
	key string,
	expected string,
	payload []byte,
) (string, error) {
	store.calls++
	if store.calls == store.failAt {
		return "", agenttask.ErrRevisionConflict
	}
	return store.delegate.CompareAndSwapIntegrationRecord(
		ctx,
		key,
		expected,
		payload,
	)
}

func newIntegrationFixture(t *testing.T) *integrationFixture {
	t.Helper()
	if _, err := exec.LookPath("git"); err != nil {
		t.Skip("git is required for change-set integration tests")
	}
	root, err := filepath.EvalSymlinks(t.TempDir())
	if err != nil {
		t.Fatalf("EvalSymlinks: %v", err)
	}
	baseRoot := filepath.Join(root, "workspace")
	localRoot := filepath.Join(root, "runtime")
	for _, directory := range []string{baseRoot, localRoot} {
		if err := os.MkdirAll(directory, 0o700); err != nil {
			t.Fatalf("MkdirAll %s: %v", directory, err)
		}
	}
	gitFixtureRun(t, baseRoot, "init", "-q")
	gitFixtureRun(t, baseRoot, "config", "user.email", "integration@example.invalid")
	gitFixtureRun(t, baseRoot, "config", "user.name", "Integration Fixture")

	resolver := agentworkspace.InputResolverFunc(func(
		_ context.Context,
		request agenttask.IsolationRequest,
	) (agentworkspace.ResolvedInputs, error) {
		return agentworkspace.ResolvedInputs{
			Grant: agentguard.WorkspaceGrant{
				ProjectID:   string(request.Project.ProjectID),
				WorkspaceID: string(request.Project.WorkspaceID),
				Root:        baseRoot,
				Revision:    string(request.Project.Intent.GrantRevision),
			},
			Profile: agentguard.ProviderProfile{
				ProviderID:              request.Target.ProviderID,
				ModelID:                 request.Target.ModelID,
				ProfileID:               request.Target.ProfileID,
				Revision:                request.Target.ProfileRevision,
				Unattended:              true,
				ApprovalBypass:          true,
				WritableRootConfinement: true,
			},
		}, nil
	})
	backend, err := agentworkspace.NewBackend(agentworkspace.BackendConfig{
		LocalRoot: localRoot,
		Retention: agentconfig.RetentionPolicy{CompletedDays: 7, BlockedDays: 14},
	}, resolver)
	if err != nil {
		t.Fatalf("NewBackend: %v", err)
	}
	statePath := filepath.Join(localRoot, "state", "manager.json")
	store, err := agentstate.NewStore(statePath)
	if err != nil {
		t.Fatalf("NewStore: %v", err)
	}
	return &integrationFixture{
		t: t, baseRoot: baseRoot, localRoot: localRoot,
		statePath: statePath, backend: backend, store: store,
	}
}

func (fixture *integrationFixture) commit(message string) {
	fixture.t.Helper()
	gitFixtureRun(fixture.t, fixture.baseRoot, "add", "-A")
	gitFixtureRun(fixture.t, fixture.baseRoot, "commit", "-q", "-m", message)
}

func (fixture *integrationFixture) prepare(
	workID string,
	ordinal agenttask.DispatchOrdinal,
) preparedTask {
	fixture.t.Helper()
	projectID := agenttask.ProjectID("project")
	workspaceID := agenttask.WorkspaceID("workspace")
	request := agenttask.IsolationRequest{
		Project: agenttask.ProjectRecord{
			ProjectID: projectID, WorkspaceID: workspaceID,
			Intent: &agenttask.StartIntent{
				ProjectID: projectID, WorkspaceID: workspaceID,
				ConfigRevision: "config-r1", GrantRevision: "grant-r1",
			},
		},
		Work: agenttask.WorkRecord{
			Unit: agenttask.WorkUnit{
				ID:            agenttask.WorkUnitID(workID),
				IsolationMode: agentguard.IsolationModeOverlay,
			},
			AttemptID:       agenttask.AttemptID(workID + "#1"),
			DispatchOrdinal: ordinal,
		},
		Target: agenttask.ExecutionTarget{
			ProviderID: "provider", ModelID: "model", ProfileID: "profile",
			ProfileRevision: "profile-r1", ConfigRevision: "config-r1", Capacity: 3,
		},
		IdempotencyKey: "dispatch/" + workID + "/1/isolation",
	}
	prepared, err := fixture.backend.Prepare(context.Background(), request)
	if err != nil {
		fixture.t.Fatalf("Prepare(%s): %v", workID, err)
	}
	return preparedTask{
		fixture: fixture, request: request, prepared: prepared,
		view:       prepared.Descriptor.WorkingDir,
		artifactID: agenttask.ArtifactID("artifact-" + workID),
		ordinal:    ordinal,
	}
}

func (task preparedTask) freeze() agentworkspace.ChangeSet {
	task.fixture.t.Helper()
	changeSet, err := task.fixture.backend.Freeze(
		context.Background(),
		agentworkspace.FreezeRequest{
			Descriptor: *task.prepared.Descriptor,
			ArtifactID: task.artifactID,
			ValidationEvidence: []agentworkspace.ValidationEvidence{{
				Name: "review", Result: "pass", Digest: "sha256:fixture",
			}},
		},
	)
	if err != nil {
		task.fixture.t.Fatalf("Freeze(%s): %v", task.request.Work.Unit.ID, err)
	}
	return changeSet
}

func (fixture *integrationFixture) integrator(
	validator agentworkspace.ValidationFunc,
) *agentworkspace.SerialIntegrator {
	fixture.t.Helper()
	integrator, err := agentworkspace.NewIntegrator(agentworkspace.IntegratorConfig{
		Backend: fixture.backend, Store: fixture.store, Validator: validator,
	})
	if err != nil {
		fixture.t.Fatalf("NewIntegrator: %v", err)
	}
	return integrator
}

func integrateTask(
	t *testing.T,
	integrator *agentworkspace.SerialIntegrator,
	task preparedTask,
	changeSet agentworkspace.ChangeSet,
	attempt agenttask.IntegrationAttempt,
) agenttask.IntegrationResult {
	t.Helper()
	result, err := integrator.Integrate(
		context.Background(),
		integrationRequest(task, changeSet, attempt),
	)
	if err != nil {
		t.Fatalf("Integrate(%s): %v", task.request.Work.Unit.ID, err)
	}
	return result
}

func integrationRequest(
	task preparedTask,
	changeSet agentworkspace.ChangeSet,
	attempt agenttask.IntegrationAttempt,
) agenttask.IntegrationRequest {
	work := task.request.Work
	work.ChangeSet = pointer(changeSet.Identity())
	work.Isolation = &agenttask.IsolationIdentity{
		ID:                 task.prepared.Descriptor.ID,
		Revision:           task.prepared.Descriptor.Revision,
		Mode:               task.prepared.Descriptor.Mode,
		PinnedBaseRevision: task.prepared.Descriptor.PinnedBaseRevision,
		TaskRoot:           task.prepared.Descriptor.TaskRoot,
	}
	return agenttask.IntegrationRequest{
		Project: task.request.Project,
		Work:    work, ChangeSet: changeSet.Identity(),
		Ordinal: task.ordinal, Attempt: attempt,
		IdempotencyKey: fmt.Sprintf(
			"integrate/%s/%s/%d",
			task.request.Work.Unit.ID,
			changeSet.Revision,
			attempt,
		),
	}
}

func pointer[T any](value T) *T { return &value }

type fixtureError string

func (err fixtureError) Error() string { return string(err) }

func errorsForTest(message string) error { return fixtureError(message) }

func errorsIsRevisionConflict(err error) bool {
	return err == agenttask.ErrRevisionConflict
}

func workspaceDigest(t *testing.T, root string) string {
	t.Helper()
	hash := sha256.New()
	var paths []string
	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if path == root {
			return nil
		}
		relative, err := filepath.Rel(root, path)
		if err != nil {
			return err
		}
		if relative == ".git" {
			return filepath.SkipDir
		}
		paths = append(paths, path)
		if entry.IsDir() {
			return nil
		}
		return nil
	})
	if err != nil {
		t.Fatalf("WalkDir: %v", err)
	}
	sort.Strings(paths)
	for _, path := range paths {
		relative, _ := filepath.Rel(root, path)
		info, err := os.Lstat(path)
		if err != nil {
			t.Fatalf("Lstat %s: %v", path, err)
		}
		_, _ = io.WriteString(hash, filepath.ToSlash(relative))
		_, _ = io.WriteString(hash, fmt.Sprintf("\x00%d\x00", uint32(info.Mode())))
		switch {
		case info.Mode().IsRegular():
			content, err := os.ReadFile(path)
			if err != nil {
				t.Fatalf("ReadFile %s: %v", path, err)
			}
			_, _ = hash.Write(content)
		case info.Mode()&os.ModeSymlink != 0:
			target, err := os.Readlink(path)
			if err != nil {
				t.Fatalf("Readlink %s: %v", path, err)
			}
			_, _ = io.WriteString(hash, target)
		}
	}
	return hex.EncodeToString(hash.Sum(nil))
}

func writeFixtureFile(t *testing.T, path, content string, mode fs.FileMode) {
	t.Helper()
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		t.Fatalf("MkdirAll %s: %v", filepath.Dir(path), err)
	}
	if err := os.WriteFile(path, []byte(content), mode); err != nil {
		t.Fatalf("WriteFile %s: %v", path, err)
	}
	if err := os.Chmod(path, mode); err != nil {
		t.Fatalf("Chmod %s: %v", path, err)
	}
}

func assertFixtureContent(t *testing.T, path, want string) {
	t.Helper()
	content, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("ReadFile %s: %v", path, err)
	}
	if string(content) != want {
		t.Fatalf("content %s = %q, want %q", path, content, want)
	}
}

func gitFixtureRun(t *testing.T, root string, args ...string) {
	t.Helper()
	command := exec.Command("git", append([]string{"-C", root}, args...)...)
	command.Env = append(os.Environ(), "LC_ALL=C")
	if output, err := command.CombinedOutput(); err != nil {
		t.Fatalf("git %v: %v\n%s", args, err, output)
	}
}