iop/packages/go/agentworkspace/integrator.go

package agentworkspace

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"sort"
	"strings"

	"iop/packages/go/agenttask"
)

const integrationJournalSchemaVersion uint32 = 1

// IntegrationRecordStatus is the durable apply lifecycle.
type IntegrationRecordStatus string

const (
	IntegrationRecordApplying         IntegrationRecordStatus = "applying"
	IntegrationRecordIntegrated       IntegrationRecordStatus = "integrated"
	IntegrationRecordTerminalDeferred IntegrationRecordStatus = "terminal_deferred"
)

// IntegrationRecordStore is the durable host-store extension used by the
// serial integrator. Each key is independently compare-and-swapped; one
// workspace journal contains both the managed head and all attempt records so
// those values change atomically.
type IntegrationRecordStore interface {
	LoadIntegrationRecord(
		context.Context,
		string,
	) (payload []byte, revision string, found bool, err error)
	CompareAndSwapIntegrationRecord(
		context.Context,
		string,
		string,
		[]byte,
	) (string, error)
}

// ValidationRequest identifies the exact post-apply state presented to the
// host validator. CanonicalRoot remains the immutable workspace identity while
// ValidationRoot is a runtime-owned candidate that already carries the prepared
// three-way result, so validator side effects can never survive a rejected
// attempt.
type ValidationRequest struct {
	CanonicalRoot   string
	ValidationRoot  string
	ChangeSet       ChangeSet
	DispatchOrdinal agenttask.DispatchOrdinal
	Attempt         agenttask.IntegrationAttempt
	BeforeRevision  string
}

// ValidationFunc performs deterministic post-apply validation.
type ValidationFunc func(context.Context, ValidationRequest) error

// IntegratorConfig wires the overlay backend to the durable host store.
type IntegratorConfig struct {
	Backend   *Backend
	Store     IntegrationRecordStore
	Validator ValidationFunc
}

// SerialIntegrator implements the shared runtime Integrator port. The manager
// owns workspace lease and ordinal admission; this backend owns immutable
// content, managed-head validation, apply, validation, and rollback.
type SerialIntegrator struct {
	backend   *Backend
	store     IntegrationRecordStore
	validator ValidationFunc
}

var _ agenttask.Integrator = (*SerialIntegrator)(nil)

// IntegrationRecord is the durable attempt evidence retained in the
// workspace journal.
type IntegrationRecord struct {
	Revision                  string                       `json:"revision"`
	IdempotencyKey            string                       `json:"idempotency_key"`
	ProjectID                 agenttask.ProjectID          `json:"project_id"`
	WorkspaceID               agenttask.WorkspaceID        `json:"workspace_id"`
	WorkUnitID                agenttask.WorkUnitID         `json:"work_unit_id"`
	ChangeSet                 agenttask.ChangeSetIdentity  `json:"change_set"`
	DispatchOrdinal           agenttask.DispatchOrdinal    `json:"dispatch_ordinal"`
	Attempt                   agenttask.IntegrationAttempt `json:"attempt"`
	Status                    IntegrationRecordStatus      `json:"status"`
	ExpectedBeforeFingerprint string                       `json:"expected_before_fingerprint"`
	ObservedBeforeFingerprint string                       `json:"observed_before_fingerprint"`
	AfterFingerprint          string                       `json:"after_fingerprint,omitempty"`
	RollbackRoot              string                       `json:"rollback_root,omitempty"`
	RollbackVerified          bool                         `json:"rollback_verified"`
	Validation                string                       `json:"validation"`
	Blocker                   string                       `json:"blocker,omitempty"`
	Retained                  bool                         `json:"retained"`
	CleanupState              string                       `json:"cleanup_state"`
}

type workspaceIntegrationJournal struct {
	SchemaVersion   uint32                       `json:"schema_version"`
	Revision        string                       `json:"revision"`
	WorkspaceID     agenttask.WorkspaceID        `json:"workspace_id"`
	CanonicalRoot   string                       `json:"canonical_root"`
	HeadFingerprint string                       `json:"head_fingerprint"`
	Records         map[string]IntegrationRecord `json:"records"`
}

type backupManifest struct {
	SchemaVersion uint32        `json:"schema_version"`
	Before        string        `json:"before"`
	Entries       []backupEntry `json:"entries"`
}

type backupEntry struct {
	Path        string       `json:"path"`
	Entry       *ChangeEntry `json:"entry,omitempty"`
	ContentFile string       `json:"content_file,omitempty"`
}

type preparedChange struct {
	Path    string
	Result  *ChangeEntry
	Content []byte
}

type integrationBlocker struct {
	message string
}

func (blocker *integrationBlocker) Error() string { return blocker.message }

// NewIntegrator constructs the strict integration backend.
func NewIntegrator(config IntegratorConfig) (*SerialIntegrator, error) {
	if config.Backend == nil {
		return nil, errors.New("agentworkspace: integration backend is required")
	}
	if config.Store == nil {
		return nil, errors.New("agentworkspace: integration record store is required")
	}
	if config.Validator == nil {
		return nil, errors.New("agentworkspace: post-apply validator is required")
	}
	return &SerialIntegrator{
		backend: config.Backend, store: config.Store, validator: config.Validator,
	}, nil
}

// Integrate applies one exact immutable change set or returns a retained
// terminal-deferred result. Known conflict, drift, and validation failures are
// results rather than Go errors so later independent ordinals may advance.
func (integrator *SerialIntegrator) Integrate(
	ctx context.Context,
	request agenttask.IntegrationRequest,
) (agenttask.IntegrationResult, error) {
	if err := ctx.Err(); err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if strings.TrimSpace(request.IdempotencyKey) == "" ||
		request.Ordinal == 0 ||
		request.Attempt == 0 {
		return agenttask.IntegrationResult{}, errors.New(
			"agentworkspace: integration identity is incomplete",
		)
	}
	if request.Work.Isolation == nil {
		return agenttask.IntegrationResult{}, errors.New(
			"agentworkspace: integration work has no retained isolation",
		)
	}
	changeSet, err := integrator.backend.LoadChangeSet(
		*request.Work.Isolation,
		request.ChangeSet,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if changeSet.ProjectID != request.Project.ProjectID ||
		changeSet.WorkspaceID != request.Project.WorkspaceID ||
		changeSet.WorkUnitID != request.Work.Unit.ID ||
		changeSet.AttemptID != request.Work.AttemptID {
		return agenttask.IntegrationResult{}, errors.New(
			"agentworkspace: integration request identity mismatch",
		)
	}

	journalKey := integrationJournalKey(
		changeSet.WorkspaceID,
		changeSet.CanonicalRoot,
	)
	journal, storeRevision, err := integrator.loadJournal(
		ctx,
		journalKey,
		changeSet,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if existing, ok := journal.Records[request.IdempotencyKey]; ok {
		if err := validateIntegrationRecord(existing, request); err != nil {
			return agenttask.IntegrationResult{}, err
		}
		switch existing.Status {
		case IntegrationRecordIntegrated, IntegrationRecordTerminalDeferred:
			return resultFromIntegrationRecord(request, existing), nil
		case IntegrationRecordApplying:
			return integrator.recoverApplying(
				ctx,
				request,
				journalKey,
				journal,
				storeRevision,
				existing,
				changeSet,
			)
		default:
			return agenttask.IntegrationResult{}, errors.New(
				"agentworkspace: retained integration status is invalid",
			)
		}
	}

	expected := journal.HeadFingerprint
	if expected == "" {
		expected = changeSet.BaseFingerprint
	}
	observed, err := captureWorkspaceSnapshot(
		ctx,
		changeSet.CanonicalRoot,
		"",
		changeSet.ConfigRevision,
		changeSet.GrantRevision,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if observed.Revision != expected {
		record := newIntegrationRecord(
			request,
			expected,
			observed.Revision,
			IntegrationRecordTerminalDeferred,
		)
		record.Validation = "not_run"
		record.Blocker = "unmanaged base drift"
		record.AfterFingerprint = observed.Revision
		record.Retained = true
		record.CleanupState = "not_started"
		return integrator.commitTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			record,
		)
	}

	prepared, err := integrator.prepareChanges(
		ctx,
		changeSet,
		observed,
	)
	if blocker := new(integrationBlocker); errors.As(err, &blocker) {
		record := newIntegrationRecord(
			request,
			expected,
			observed.Revision,
			IntegrationRecordTerminalDeferred,
		)
		record.Validation = "not_run"
		record.Blocker = blocker.Error()
		record.AfterFingerprint = observed.Revision
		record.RollbackVerified = true
		record.Retained = true
		record.CleanupState = "not_started"
		return integrator.commitTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			record,
		)
	} else if err != nil {
		return agenttask.IntegrationResult{}, err
	}

	// Validate the exact three-way candidate outside the canonical workspace so
	// validator side effects can never survive a rejected attempt.
	validationRoot, cleanupCandidate, err := integrator.prepareValidationCandidate(
		ctx,
		changeSet,
		prepared,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	validationErr := integrator.validator(ctx, ValidationRequest{
		CanonicalRoot:   changeSet.CanonicalRoot,
		ValidationRoot:  validationRoot,
		ChangeSet:       changeSet,
		DispatchOrdinal: request.Ordinal,
		Attempt:         request.Attempt,
		BeforeRevision:  observed.Revision,
	})
	cleanupCandidate()
	if validationErr != nil {
		record := newIntegrationRecord(
			request,
			expected,
			observed.Revision,
			IntegrationRecordTerminalDeferred,
		)
		record.Validation = "failed"
		record.Blocker = fmt.Sprintf("post-apply validation failed: %v", validationErr)
		record.AfterFingerprint = observed.Revision
		record.RollbackVerified = true
		record.Retained = true
		record.CleanupState = "not_started"
		return integrator.commitTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			record,
		)
	}

	// Reconfirm the immutable canonical identity before any mutation so a
	// concurrent drift during validation cannot be silently overwritten.
	reconfirmed, err := captureWorkspaceSnapshot(
		ctx,
		changeSet.CanonicalRoot,
		"",
		changeSet.ConfigRevision,
		changeSet.GrantRevision,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if reconfirmed.Revision != observed.Revision {
		record := newIntegrationRecord(
			request,
			expected,
			observed.Revision,
			IntegrationRecordTerminalDeferred,
		)
		record.Validation = "not_run"
		record.Blocker = "canonical workspace drifted during validation"
		record.AfterFingerprint = reconfirmed.Revision
		record.RollbackVerified = true
		record.Retained = true
		record.CleanupState = "not_started"
		return integrator.commitTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			record,
		)
	}

	rollbackRoot, err := integrator.captureRollback(
		ctx,
		request,
		changeSet,
		observed,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	applying := newIntegrationRecord(
		request,
		expected,
		observed.Revision,
		IntegrationRecordApplying,
	)
	applying.RollbackRoot = rollbackRoot
	applying.Validation = "pending"
	applying.Retained = true
	applying.CleanupState = "pending"
	applying = sealIntegrationRecord(applying)
	journal.Records[request.IdempotencyKey] = applying
	storeRevision, err = integrator.saveJournal(
		ctx,
		journalKey,
		storeRevision,
		journal,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}

	// Apply only the frozen prepared changes. Rollback restores the exact
	// observed fingerprint if the apply or its fingerprint capture fails.
	if err := applyPreparedChanges(changeSet.CanonicalRoot, prepared); err != nil {
		return integrator.rollbackTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			applying,
			changeSet,
			fmt.Sprintf("apply failed: %v", err),
		)
	}
	after, err := captureWorkspaceSnapshot(
		ctx,
		changeSet.CanonicalRoot,
		"",
		changeSet.ConfigRevision,
		changeSet.GrantRevision,
	)
	if err != nil {
		return integrator.rollbackTerminal(
			ctx,
			request,
			journalKey,
			journal,
			storeRevision,
			applying,
			changeSet,
			fmt.Sprintf("post-apply fingerprint failed: %v", err),
		)
	}
	integrated := applying
	integrated.Status = IntegrationRecordIntegrated
	integrated.AfterFingerprint = after.Revision
	integrated.Validation = "passed"
	integrated.Retained = false
	integrated.CleanupState = "pending"
	journal.HeadFingerprint = after.Revision
	integrated = sealIntegrationRecord(integrated)
	journal.Records[request.IdempotencyKey] = integrated
	storeRevision, err = integrator.saveJournal(
		ctx,
		journalKey,
		storeRevision,
		journal,
	)
	if err != nil {
		_ = restoreRollback(
			ctx,
			changeSet.CanonicalRoot,
			rollbackRoot,
		)
		return agenttask.IntegrationResult{}, err
	}
	if err := os.RemoveAll(rollbackRoot); err == nil {
		integrated.CleanupState = "complete"
		integrated.RollbackRoot = ""
		integrated = sealIntegrationRecord(integrated)
		journal.Records[request.IdempotencyKey] = integrated
		if _, cleanupErr := integrator.saveJournal(
			context.WithoutCancel(ctx),
			journalKey,
			storeRevision,
			journal,
		); cleanupErr == nil {
			integrated = journal.Records[request.IdempotencyKey]
		}
	}
	return resultFromIntegrationRecord(request, integrated), nil
}

// LocatorRoot returns the canonical workspace bound into this change set.
func (changeSet ChangeSet) LocatorRoot() string {
	return changeSet.CanonicalRoot
}

func (integrator *SerialIntegrator) loadJournal(
	ctx context.Context,
	key string,
	changeSet ChangeSet,
) (workspaceIntegrationJournal, string, error) {
	payload, revision, found, err := integrator.store.LoadIntegrationRecord(ctx, key)
	if err != nil {
		return workspaceIntegrationJournal{}, "", err
	}
	if !found {
		return workspaceIntegrationJournal{
			SchemaVersion: integrationJournalSchemaVersion,
			WorkspaceID:   changeSet.WorkspaceID,
			CanonicalRoot: changeSet.CanonicalRoot,
			Records:       make(map[string]IntegrationRecord),
		}, "", nil
	}
	var journal workspaceIntegrationJournal
	if err := decodeStrictJSON(payload, &journal); err != nil {
		return workspaceIntegrationJournal{}, "", fmt.Errorf(
			"agentworkspace: decode integration journal: %w",
			err,
		)
	}
	if journal.SchemaVersion != integrationJournalSchemaVersion ||
		journal.Revision != integrationJournalRevision(journal) ||
		journal.WorkspaceID != changeSet.WorkspaceID ||
		journal.CanonicalRoot != changeSet.CanonicalRoot ||
		journal.Records == nil {
		return workspaceIntegrationJournal{}, "", errors.New(
			"agentworkspace: integration journal is corrupt or rebound",
		)
	}
	for key, record := range journal.Records {
		if key != record.IdempotencyKey ||
			record.Revision != integrationRecordRevision(record) {
			return workspaceIntegrationJournal{}, "", errors.New(
				"agentworkspace: integration record is corrupt",
			)
		}
	}
	return journal, revision, nil
}

func (integrator *SerialIntegrator) saveJournal(
	ctx context.Context,
	key string,
	expected string,
	journal workspaceIntegrationJournal,
) (string, error) {
	journal.Revision = integrationJournalRevision(journal)
	payload, err := json.Marshal(journal)
	if err != nil {
		return "", err
	}
	return integrator.store.CompareAndSwapIntegrationRecord(
		ctx,
		key,
		expected,
		payload,
	)
}

func (integrator *SerialIntegrator) commitTerminal(
	ctx context.Context,
	request agenttask.IntegrationRequest,
	journalKey string,
	journal workspaceIntegrationJournal,
	storeRevision string,
	record IntegrationRecord,
) (agenttask.IntegrationResult, error) {
	record = sealIntegrationRecord(record)
	journal.Records[request.IdempotencyKey] = record
	if _, err := integrator.saveJournal(
		ctx,
		journalKey,
		storeRevision,
		journal,
	); err != nil {
		return agenttask.IntegrationResult{}, err
	}
	return resultFromIntegrationRecord(request, record), nil
}

func (integrator *SerialIntegrator) rollbackTerminal(
	ctx context.Context,
	request agenttask.IntegrationRequest,
	journalKey string,
	journal workspaceIntegrationJournal,
	storeRevision string,
	applying IntegrationRecord,
	changeSet ChangeSet,
	reason string,
) (agenttask.IntegrationResult, error) {
	expectedRollback := integrator.rollbackRoot(request.IdempotencyKey)
	if applying.RollbackRoot != expectedRollback {
		return agenttask.IntegrationResult{}, errors.New(
			"agentworkspace: retained rollback identity mismatch",
		)
	}
	if err := restoreRollback(
		ctx,
		changeSet.CanonicalRoot,
		applying.RollbackRoot,
	); err != nil {
		return agenttask.IntegrationResult{}, fmt.Errorf(
			"agentworkspace: rollback failed after %s: %w",
			reason,
			err,
		)
	}
	restored, err := captureWorkspaceSnapshot(
		ctx,
		changeSet.CanonicalRoot,
		"",
		changeSet.ConfigRevision,
		changeSet.GrantRevision,
	)
	if err != nil {
		return agenttask.IntegrationResult{}, err
	}
	if restored.Revision != applying.ObservedBeforeFingerprint {
		return agenttask.IntegrationResult{}, fmt.Errorf(
			"agentworkspace: rollback fingerprint %q does not match %q",
			restored.Revision,
			applying.ObservedBeforeFingerprint,
		)
	}
	terminal := applying
	terminal.Status = IntegrationRecordTerminalDeferred
	terminal.AfterFingerprint = restored.Revision
	terminal.RollbackVerified = true
	terminal.Validation = "failed"
	terminal.Blocker = reason
	terminal.Retained = true
	terminal.CleanupState = "retained"
	terminal = sealIntegrationRecord(terminal)
	journal.Records[request.IdempotencyKey] = terminal
	if _, err := integrator.saveJournal(
		ctx,
		journalKey,
		storeRevision,
		journal,
	); err != nil {
		return agenttask.IntegrationResult{}, err
	}
	return resultFromIntegrationRecord(request, terminal), nil
}

func (integrator *SerialIntegrator) recoverApplying(
	ctx context.Context,
	request agenttask.IntegrationRequest,
	journalKey string,
	journal workspaceIntegrationJournal,
	storeRevision string,
	record IntegrationRecord,
	changeSet ChangeSet,
) (agenttask.IntegrationResult, error) {
	return integrator.rollbackTerminal(
		ctx,
		request,
		journalKey,
		journal,
		storeRevision,
		record,
		changeSet,
		"restart recovered an interrupted apply by exact rollback",
	)
}

func (integrator *SerialIntegrator) prepareChanges(
	ctx context.Context,
	changeSet ChangeSet,
	observed WorkspaceSnapshot,
) ([]preparedChange, error) {
	current := changeEntriesFromSnapshot(observed)
	owned := make(map[string]struct{}, len(changeSet.Operations))
	for _, operation := range changeSet.Operations {
		owned[operation.Path] = struct{}{}
	}
	result := make([]preparedChange, 0, len(changeSet.Operations))
	for _, operation := range changeSet.Operations {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		currentEntry, currentExists := current[operation.Path]
		var currentPointer *ChangeEntry
		if currentExists {
			copy := currentEntry
			currentPointer = &copy
		}
		// A managed predecessor may have added independent descendants under a
		// directory this change set deletes or replaces. Removing the parent
		// recursively would destroy work the runtime already integrated, so
		// reconcile the base/current descendant sets before emitting any
		// hierarchy change.
		if currentExists && currentEntry.Kind == SnapshotEntryDirectory {
			resultKeepsDirectory := operation.Result != nil &&
				operation.Result.Kind == SnapshotEntryDirectory
			if !resultKeepsDirectory &&
				hasIndependentDescendants(current, owned, operation.Path) {
				if operation.Result == nil {
					// Preserve the container so predecessor additions survive.
					continue
				}
				return nil, &integrationBlocker{
					message: fmt.Sprintf(
						"directory replacement conflict at %s",
						operation.Path,
					),
				}
			}
		}
		if reflect.DeepEqual(currentPointer, operation.Result) {
			prepared, err := preparedFromResult(changeSet, operation, operation.Result)
			if err != nil {
				return nil, err
			}
			result = append(result, prepared)
			continue
		}
		if reflect.DeepEqual(currentPointer, operation.Base) {
			prepared, err := preparedFromResult(changeSet, operation, operation.Result)
			if err != nil {
				return nil, err
			}
			result = append(result, prepared)
			continue
		}
		merged, err := mergeRegularChange(
			ctx,
			changeSet,
			operation,
			currentPointer,
		)
		if err != nil {
			return nil, err
		}
		result = append(result, merged)
	}
	return result, nil
}

// hasIndependentDescendants reports whether the observed workspace holds any
// path beneath directory that this change set does not own. Such a path was
// introduced by a managed predecessor and must never be removed by a later
// parent-directory operation.
func hasIndependentDescendants(
	current map[string]ChangeEntry,
	owned map[string]struct{},
	directory string,
) bool {
	prefix := directory + "/"
	for path := range current {
		if !strings.HasPrefix(path, prefix) {
			continue
		}
		if _, isOwned := owned[path]; !isOwned {
			return true
		}
	}
	return false
}

// prepareValidationCandidate materializes the exact observed workspace plus the
// prepared three-way result in a runtime-owned directory. The validator inspects
// this candidate instead of the canonical root, so any mutation it makes is
// discarded when the candidate is cleaned up.
func (integrator *SerialIntegrator) prepareValidationCandidate(
	ctx context.Context,
	changeSet ChangeSet,
	prepared []preparedChange,
) (string, func(), error) {
	base := filepath.Join(integrator.backend.root, "integrations")
	if err := os.MkdirAll(base, 0o700); err != nil {
		return "", nil, err
	}
	candidate, err := os.MkdirTemp(base, ".candidate-")
	if err != nil {
		return "", nil, err
	}
	cleanup := func() { _ = os.RemoveAll(candidate) }
	if err := copyWorkspaceTree(ctx, changeSet.CanonicalRoot, candidate); err != nil {
		cleanup()
		return "", nil, err
	}
	if err := applyPreparedChanges(candidate, prepared); err != nil {
		cleanup()
		return "", nil, err
	}
	return candidate, cleanup, nil
}

// copyWorkspaceTree copies every workspace entry except the Git metadata into a
// candidate root, preserving type, content, and symlink identity.
func copyWorkspaceTree(ctx context.Context, sourceRoot, destinationRoot string) error {
	return filepath.WalkDir(sourceRoot, func(path string, entry fs.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if err := ctx.Err(); err != nil {
			return err
		}
		if path == sourceRoot {
			return nil
		}
		relative, err := filepath.Rel(sourceRoot, path)
		if err != nil {
			return err
		}
		if relative == ".git" ||
			strings.HasPrefix(relative, ".git"+string(filepath.Separator)) {
			if entry.IsDir() && relative == ".git" {
				return filepath.SkipDir
			}
			return nil
		}
		destination := filepath.Join(destinationRoot, relative)
		info, err := os.Lstat(path)
		if err != nil {
			return err
		}
		switch {
		case info.IsDir():
			return os.MkdirAll(destination, 0o700)
		case info.Mode().IsRegular():
			if err := os.MkdirAll(filepath.Dir(destination), 0o700); err != nil {
				return err
			}
			if _, err := digestAndCopyRegular(path, destination); err != nil {
				return err
			}
			return os.Chmod(destination, info.Mode().Perm())
		case info.Mode()&os.ModeSymlink != 0:
			target, err := os.Readlink(path)
			if err != nil {
				return err
			}
			if err := os.MkdirAll(filepath.Dir(destination), 0o700); err != nil {
				return err
			}
			return os.Symlink(target, destination)
		default:
			return fmt.Errorf(
				"agentworkspace: unsupported workspace object %q",
				filepath.ToSlash(relative),
			)
		}
	})
}

func preparedFromResult(
	changeSet ChangeSet,
	operation ChangeOperation,
	entry *ChangeEntry,
) (preparedChange, error) {
	prepared := preparedChange{Path: operation.Path}
	if entry == nil {
		return prepared, nil
	}
	copy := *entry
	prepared.Result = &copy
	if entry.Kind == SnapshotEntryRegular {
		content, err := os.ReadFile(
			filepath.Join(changeSet.Locator.Root, filepath.FromSlash(operation.ContentFile)),
		)
		if err != nil {
			return preparedChange{}, err
		}
		if digestBytes(content) != entry.ContentDigest {
			return preparedChange{}, errors.New(
				"agentworkspace: immutable change-set content is corrupt",
			)
		}
		prepared.Content = content
	}
	return prepared, nil
}

func mergeRegularChange(
	ctx context.Context,
	changeSet ChangeSet,
	operation ChangeOperation,
	current *ChangeEntry,
) (preparedChange, error) {
	if operation.Base == nil || operation.Result == nil || current == nil ||
		operation.Base.Kind != SnapshotEntryRegular ||
		operation.Result.Kind != SnapshotEntryRegular ||
		current.Kind != SnapshotEntryRegular {
		return preparedChange{}, &integrationBlocker{
			message: fmt.Sprintf("merge conflict at %s", operation.Path),
		}
	}
	baseMode, modeConflict := mergeMode(
		operation.Base.Mode,
		current.Mode,
		operation.Result.Mode,
	)
	if modeConflict {
		return preparedChange{}, &integrationBlocker{
			message: fmt.Sprintf("mode conflict at %s", operation.Path),
		}
	}
	overlay, err := readOverlayRecord(changeSet.Locator.OverlayRecord)
	if err != nil {
		return preparedChange{}, err
	}
	basePath := filepath.Join(
		overlay.Locator.SnapshotRoot,
		"tree",
		filepath.FromSlash(operation.Path),
	)
	currentPath := filepath.Join(
		overlay.CanonicalRoot,
		filepath.FromSlash(operation.Path),
	)
	desiredPath := filepath.Join(
		changeSet.Locator.Root,
		filepath.FromSlash(operation.ContentFile),
	)
	merged, conflict, err := runMergeFile(
		ctx,
		currentPath,
		basePath,
		desiredPath,
	)
	if err != nil {
		return preparedChange{}, err
	}
	if conflict {
		return preparedChange{}, &integrationBlocker{
			message: fmt.Sprintf("content conflict at %s", operation.Path),
		}
	}
	entry := *operation.Result
	entry.Mode = baseMode
	entry.ContentDigest = digestBytes(merged)
	return preparedChange{Path: operation.Path, Result: &entry, Content: merged}, nil
}

func mergeMode(base, current, desired uint32) (uint32, bool) {
	switch {
	case desired == base:
		return current, false
	case current == base || current == desired:
		return desired, false
	default:
		return 0, true
	}
}

func runMergeFile(
	ctx context.Context,
	current string,
	base string,
	desired string,
) ([]byte, bool, error) {
	command := exec.CommandContext(
		ctx,
		"git",
		"merge-file",
		"-p",
		current,
		base,
		desired,
	)
	command.Env = append(os.Environ(), "LC_ALL=C")
	output, err := command.Output()
	if err == nil {
		return output, false, nil
	}
	var exitError *exec.ExitError
	if errors.As(err, &exitError) && exitError.ExitCode() == 1 {
		return nil, true, nil
	}
	if errors.As(err, &exitError) {
		return nil, false, fmt.Errorf(
			"agentworkspace: git merge-file failed: %s",
			boundedText(exitError.Stderr),
		)
	}
	return nil, false, fmt.Errorf("agentworkspace: git merge-file: %w", err)
}

func (integrator *SerialIntegrator) captureRollback(
	ctx context.Context,
	request agenttask.IntegrationRequest,
	changeSet ChangeSet,
	observed WorkspaceSnapshot,
) (string, error) {
	root := integrator.rollbackRoot(request.IdempotencyKey)
	if _, err := os.Stat(root); err == nil {
		return "", errors.New("agentworkspace: integration rollback identity already exists")
	} else if !os.IsNotExist(err) {
		return "", err
	}
	staging, err := os.MkdirTemp(
		filepath.Join(integrator.backend.root, "integrations"),
		".capturing-",
	)
	if err != nil {
		if os.IsNotExist(err) {
			if mkdirErr := os.MkdirAll(
				filepath.Join(integrator.backend.root, "integrations"),
				0o700,
			); mkdirErr != nil {
				return "", mkdirErr
			}
			staging, err = os.MkdirTemp(
				filepath.Join(integrator.backend.root, "integrations"),
				".capturing-",
			)
		}
		if err != nil {
			return "", err
		}
	}
	cleanup := true
	defer func() {
		if cleanup {
			_ = os.RemoveAll(staging)
		}
	}()
	current := changeEntriesFromSnapshot(observed)
	manifest := backupManifest{
		SchemaVersion: 1,
		Before:        observed.Revision,
		Entries:       make([]backupEntry, 0, len(changeSet.WriteSet)),
	}
	for _, path := range changeSet.WriteSet {
		entry, exists := current[path]
		backup := backupEntry{Path: path}
		if exists {
			copy := entry
			backup.Entry = &copy
			if entry.Kind == SnapshotEntryRegular {
				backup.ContentFile = filepath.ToSlash(filepath.Join("content", path))
				source := filepath.Join(changeSet.LocatorRoot(), filepath.FromSlash(path))
				destination := filepath.Join(staging, filepath.FromSlash(backup.ContentFile))
				digest, copyErr := digestAndCopyRegular(source, destination)
				if copyErr != nil {
					return "", copyErr
				}
				if digest != entry.ContentDigest {
					return "", errors.New(
						"agentworkspace: canonical content changed during rollback capture",
					)
				}
			}
		}
		manifest.Entries = append(manifest.Entries, backup)
	}
	if err := writeJSONFile(
		filepath.Join(staging, "backup.json"),
		manifest,
		0o400,
	); err != nil {
		return "", err
	}
	if err := os.Rename(staging, root); err != nil {
		return "", err
	}
	cleanup = false
	return root, nil
}

func (integrator *SerialIntegrator) rollbackRoot(idempotencyKey string) string {
	return filepath.Join(
		integrator.backend.root,
		"integrations",
		strings.TrimPrefix(digestText(idempotencyKey), "sha256:"),
	)
}

func applyPreparedChanges(root string, changes []preparedChange) error {
	ordered := append([]preparedChange(nil), changes...)
	sort.SliceStable(ordered, func(left, right int) bool {
		leftDelete := ordered[left].Result == nil
		rightDelete := ordered[right].Result == nil
		if leftDelete != rightDelete {
			return leftDelete
		}
		leftDepth := strings.Count(ordered[left].Path, "/")
		rightDepth := strings.Count(ordered[right].Path, "/")
		if leftDelete {
			return leftDepth > rightDepth
		}
		return leftDepth < rightDepth
	})
	for _, change := range ordered {
		target := filepath.Join(root, filepath.FromSlash(change.Path))
		if !pathContains(root, target) {
			return errors.New("agentworkspace: prepared change escapes canonical root")
		}
		if change.Result == nil {
			if err := removePreparedTarget(target); err != nil {
				return err
			}
			continue
		}
		switch change.Result.Kind {
		case SnapshotEntryDirectory:
			if info, err := os.Lstat(target); err == nil && !info.IsDir() {
				if err := os.RemoveAll(target); err != nil {
					return err
				}
			} else if err != nil && !os.IsNotExist(err) {
				return err
			}
			if err := os.MkdirAll(target, os.FileMode(change.Result.Mode).Perm()); err != nil {
				return err
			}
			if err := os.Chmod(target, os.FileMode(change.Result.Mode).Perm()); err != nil {
				return err
			}
		case SnapshotEntryRegular:
			if err := installRegular(
				target,
				change.Content,
				os.FileMode(change.Result.Mode).Perm(),
			); err != nil {
				return err
			}
		case SnapshotEntrySymlink:
			if err := installSymlink(target, change.Result.SymlinkTarget); err != nil {
				return err
			}
		default:
			return fmt.Errorf(
				"agentworkspace: unsupported integration entry %q",
				change.Result.Kind,
			)
		}
	}
	return nil
}

func installRegular(path string, content []byte, mode os.FileMode) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	temporary, err := os.CreateTemp(filepath.Dir(path), ".iop-integrating-*")
	if err != nil {
		return err
	}
	name := temporary.Name()
	defer os.Remove(name)
	if _, err := temporary.Write(content); err != nil {
		_ = temporary.Close()
		return err
	}
	if err := temporary.Chmod(mode); err != nil {
		_ = temporary.Close()
		return err
	}
	if err := temporary.Sync(); err != nil {
		_ = temporary.Close()
		return err
	}
	if err := temporary.Close(); err != nil {
		return err
	}
	if info, err := os.Lstat(path); err == nil && info.IsDir() {
		if err := removeEmptyDirectory(path); err != nil {
			return err
		}
	} else if err != nil && !os.IsNotExist(err) {
		return err
	}
	return os.Rename(name, path)
}

// removeEmptyDirectory removes a directory that a type replacement must
// overwrite. The three-way preparation proves no independent descendant
// remains, so a non-empty directory here is a hard error rather than a silent
// recursive deletion of integrated predecessor work.
func removeEmptyDirectory(path string) error {
	entries, err := os.ReadDir(path)
	if err != nil {
		return err
	}
	if len(entries) != 0 {
		return fmt.Errorf(
			"agentworkspace: refusing to replace non-empty directory %q",
			path,
		)
	}
	return os.Remove(path)
}

// removePreparedTarget deletes a change-set-owned path without recursively
// destroying a managed parent. A directory is removed only when it holds no
// remaining descendant; otherwise the container is preserved so independent
// predecessor additions survive.
func removePreparedTarget(target string) error {
	info, err := os.Lstat(target)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return err
	}
	if info.IsDir() {
		entries, err := os.ReadDir(target)
		if err != nil {
			return err
		}
		if len(entries) != 0 {
			return nil
		}
		return os.Remove(target)
	}
	return os.Remove(target)
}

func installSymlink(path, target string) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	temporary, err := os.MkdirTemp(filepath.Dir(path), ".iop-link-*")
	if err != nil {
		return err
	}
	defer os.RemoveAll(temporary)
	link := filepath.Join(temporary, "link")
	if err := os.Symlink(target, link); err != nil {
		return err
	}
	if info, err := os.Lstat(path); err == nil {
		if info.IsDir() {
			if err := removeEmptyDirectory(path); err != nil {
				return err
			}
		} else if err := os.Remove(path); err != nil {
			return err
		}
	} else if !os.IsNotExist(err) {
		return err
	}
	return os.Rename(link, path)
}

func restoreRollback(ctx context.Context, root, rollbackRoot string) error {
	var manifest backupManifest
	if err := readJSONFile(
		filepath.Join(rollbackRoot, "backup.json"),
		&manifest,
	); err != nil {
		return err
	}
	if manifest.SchemaVersion != 1 {
		return errors.New("agentworkspace: unsupported rollback manifest")
	}
	changes := make([]preparedChange, 0, len(manifest.Entries))
	for _, backup := range manifest.Entries {
		if err := ctx.Err(); err != nil {
			return err
		}
		if filepath.ToSlash(filepath.Clean(backup.Path)) != backup.Path ||
			backup.Path == "." ||
			strings.HasPrefix(backup.Path, "../") {
			return errors.New("agentworkspace: rollback path is invalid")
		}
		prepared := preparedChange{Path: backup.Path}
		if backup.Entry != nil {
			copy := *backup.Entry
			prepared.Result = &copy
			if copy.Kind == SnapshotEntryRegular {
				if backup.ContentFile != filepath.ToSlash(
					filepath.Join("content", backup.Path),
				) {
					return errors.New(
						"agentworkspace: rollback content locator is invalid",
					)
				}
				content, err := os.ReadFile(
					filepath.Join(rollbackRoot, filepath.FromSlash(backup.ContentFile)),
				)
				if err != nil {
					return err
				}
				if digestBytes(content) != copy.ContentDigest {
					return errors.New("agentworkspace: rollback content is corrupt")
				}
				prepared.Content = content
			} else if backup.ContentFile != "" {
				return errors.New(
					"agentworkspace: non-regular rollback entry contains content",
				)
			}
		} else if backup.ContentFile != "" {
			return errors.New(
				"agentworkspace: absent rollback entry contains content",
			)
		}
		changes = append(changes, prepared)
	}
	return applyPreparedChanges(root, changes)
}

func newIntegrationRecord(
	request agenttask.IntegrationRequest,
	expected string,
	observed string,
	status IntegrationRecordStatus,
) IntegrationRecord {
	return IntegrationRecord{
		IdempotencyKey:            request.IdempotencyKey,
		ProjectID:                 request.Project.ProjectID,
		WorkspaceID:               request.Project.WorkspaceID,
		WorkUnitID:                request.Work.Unit.ID,
		ChangeSet:                 request.ChangeSet,
		DispatchOrdinal:           request.Ordinal,
		Attempt:                   request.Attempt,
		Status:                    status,
		ExpectedBeforeFingerprint: expected,
		ObservedBeforeFingerprint: observed,
	}
}

func sealIntegrationRecord(record IntegrationRecord) IntegrationRecord {
	record.Revision = integrationRecordRevision(record)
	return record
}

func validateIntegrationRecord(
	record IntegrationRecord,
	request agenttask.IntegrationRequest,
) error {
	if record.IdempotencyKey != request.IdempotencyKey ||
		record.ProjectID != request.Project.ProjectID ||
		record.WorkspaceID != request.Project.WorkspaceID ||
		record.WorkUnitID != request.Work.Unit.ID ||
		record.ChangeSet != request.ChangeSet ||
		record.DispatchOrdinal != request.Ordinal ||
		record.Attempt != request.Attempt ||
		record.Revision != integrationRecordRevision(record) {
		return errors.New("agentworkspace: integration replay identity mismatch")
	}
	if record.ExpectedBeforeFingerprint == "" ||
		record.ObservedBeforeFingerprint == "" {
		return errors.New("agentworkspace: integration fingerprint evidence is missing")
	}
	switch record.Status {
	case IntegrationRecordApplying:
		if record.RollbackRoot == "" ||
			record.Validation != "pending" ||
			!record.Retained {
			return errors.New("agentworkspace: applying integration record is incomplete")
		}
	case IntegrationRecordIntegrated:
		if record.AfterFingerprint == "" ||
			record.Validation != "passed" ||
			record.Retained {
			return errors.New("agentworkspace: integrated record is incomplete")
		}
	case IntegrationRecordTerminalDeferred:
		if record.AfterFingerprint == "" ||
			record.Blocker == "" ||
			!record.Retained {
			return errors.New("agentworkspace: terminal integration record is incomplete")
		}
	default:
		return errors.New("agentworkspace: integration record status is invalid")
	}
	return nil
}

func resultFromIntegrationRecord(
	request agenttask.IntegrationRequest,
	record IntegrationRecord,
) agenttask.IntegrationResult {
	result := agenttask.IntegrationResult{
		ProjectID:      request.Project.ProjectID,
		WorkUnitID:     request.Work.Unit.ID,
		ChangeSet:      request.ChangeSet,
		Ordinal:        request.Ordinal,
		Attempt:        request.Attempt,
		BeforeRevision: record.ObservedBeforeFingerprint,
		AfterRevision:  record.AfterFingerprint,
	}
	switch record.Status {
	case IntegrationRecordIntegrated:
		result.Outcome = agenttask.IntegrationOutcomeIntegrated
		result.CompletionLocator = &agenttask.LocatorRecord{
			Kind:        agenttask.LocatorCompletion,
			Opaque:      "integration:" + strings.TrimPrefix(digestText(request.IdempotencyKey), "sha256:"),
			Revision:    record.Revision,
			ProjectID:   request.Project.ProjectID,
			WorkspaceID: request.Project.WorkspaceID,
			WorkUnitID:  request.Work.Unit.ID,
			AttemptID:   request.Work.AttemptID,
		}
	case IntegrationRecordTerminalDeferred:
		result.Outcome = agenttask.IntegrationOutcomeTerminalDeferred
		result.Retained = true
		result.Blocker = &agenttask.Blocker{
			Code:      agenttask.BlockerIntegrationFailed,
			Message:   record.Blocker,
			Retryable: true,
		}
	}
	return result
}

func integrationRecordRevision(record IntegrationRecord) string {
	copy := record
	copy.Revision = ""
	encoded, _ := json.Marshal(copy)
	return digestBytes(encoded)
}

func integrationJournalRevision(journal workspaceIntegrationJournal) string {
	copy := journal
	copy.Revision = ""
	encoded, _ := json.Marshal(copy)
	return digestBytes(encoded)
}

func integrationJournalKey(
	workspace agenttask.WorkspaceID,
	canonicalRoot string,
) string {
	// Bind the journal to the physical workspace without exposing its raw path
	// as the store key.
	return "workspace-integration:" + strings.TrimPrefix(
		digestText(
			string(workspace)+"\x00"+canonicalRoot,
		),
		"sha256:",
	)
}

func decodeStrictJSON(payload []byte, destination any) error {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		return err
	}
	var trailing any
	if err := decoder.Decode(&trailing); err == nil {
		return errors.New("multiple JSON values")
	} else if !errors.Is(err, io.EOF) {
		return err
	}
	return nil
}