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 = © } // 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 = © 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 = © 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 = © 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 }