Implement restore execution for the restore subcommand

internal/app/restore.go

@@ -8,12 +8,14 @@ import (
 	"io"
 
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
 	"gitea.maximumdirect.net/eric/narratio/internal/config"
 )
 
 var newObjectStoreFromConfigFn = storage.NewObjectStoreFromConfig
 var discoverRemoteCurrentStateFn = discoverRemoteCurrentState
 var buildRestorePlanFn = buildRestorePlan
+var executeRestorePlanFn = executeRestorePlan
 
 // Restore validates restore CLI/config inputs and storage preflight for future restore phases.
 func Restore(ctx context.Context, args []string, out io.Writer) error {
@@ -93,16 +95,43 @@ func Restore(ctx context.Context, args []string, out io.Writer) error {
 	if dryRun {
 		return nil
 	}
+
+	artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
+	if _, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID); err != nil {
+		return fmt.Errorf("restore: prepare workdir: %w", err)
+	}
+	lock, err := artifactStore.AcquireSessionLockFor(cfg.Session.Campaign, cfg.Session.SessionID)
+	if err != nil {
+		return fmt.Errorf("restore: acquire session lock: %w", err)
+	}
+	defer func() {
+		_ = artifactStore.ReleaseSessionLock(lock)
+	}()
+
 	if plan.ConflictCount > 0 && !force {
 		return fmt.Errorf(
 			"restore: plan has %d conflicting path(s); rerun with --force or resolve local conflicts",
 			plan.ConflictCount,
 		)
 	}
-	return fmt.Errorf(
+
-		"restore: planned remote restore for %s/%s (run %s); not yet implemented (phase 4: restore execution)",
+	result, err := executeRestorePlanFn(ctx, cfg, current, plan, objectStore)
+	if err != nil {
+		return fmt.Errorf("restore: execute plan: %w", err)
+	}
+
+	_, err = fmt.Fprintf(
+		out,
+		"restore complete: session %s/%s run=%s downloaded=%d skipped_same=%d conflicts=%d\n",
 		current.Campaign,
 		current.SessionID,
 		current.RunID,
+		result.DownloadedCount,
+		plan.SkipSameCount,
+		plan.ConflictCount,
 	)
+	if err != nil {
+		return fmt.Errorf("restore: write summary: %w", err)
+	}
+	return nil
 }

internal/app/restore_execute.go

@@ -0,0 +1,173 @@
+package app
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+// RestoreExecutionResult captures concrete file-install results for one restore execution.
+type RestoreExecutionResult struct {
+	DownloadedCount int
+}
+
+func executeRestorePlan(
+	ctx context.Context,
+	cfg *config.Config,
+	current *RemoteCurrentState,
+	plan *RestorePlan,
+	store storage.ObjectStore,
+) (*RestoreExecutionResult, error) {
+	if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
+		return nil, fmt.Errorf("resolved config with pipeline/session is required")
+	}
+	if current == nil {
+		return nil, fmt.Errorf("remote current state is required")
+	}
+	if plan == nil {
+		return nil, fmt.Errorf("restore plan is required")
+	}
+	if store == nil {
+		return nil, fmt.Errorf("remote object store is required")
+	}
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
+	manifestActions := make([]RestoreAction, 0, 1)
+	actions := make([]RestoreAction, 0, len(plan.Actions))
+	for _, action := range plan.Actions {
+		if action.Kind != RestoreActionDownload {
+			continue
+		}
+		if action.LocalRelativePath == config.PathManifestFile {
+			manifestActions = append(manifestActions, action)
+			continue
+		}
+		actions = append(actions, action)
+	}
+	if len(manifestActions) > 1 {
+		return nil, fmt.Errorf("restore plan includes multiple manifest download actions")
+	}
+	if len(manifestActions) == 1 {
+		actions = append(actions, manifestActions[0])
+	}
+
+	result := &RestoreExecutionResult{}
+	for _, action := range actions {
+		if err := executeRestoreDownloadAction(ctx, cfg, sessionRoot, current, action, store); err != nil {
+			return nil, fmt.Errorf("install %q from %q: %w", action.LocalRelativePath, action.RemoteKey, err)
+		}
+		result.DownloadedCount++
+	}
+
+	return result, nil
+}
+
+func executeRestoreDownloadAction(
+	ctx context.Context,
+	cfg *config.Config,
+	sessionRoot string,
+	current *RemoteCurrentState,
+	action RestoreAction,
+	store storage.ObjectStore,
+) error {
+	safeLocalPath, err := joinWithinSessionRoot(sessionRoot, action.LocalRelativePath)
+	if err != nil {
+		return fmt.Errorf("resolve safe local path: %w", err)
+	}
+	if strings.TrimSpace(action.LocalPath) != "" && filepath.Clean(action.LocalPath) != safeLocalPath {
+		return fmt.Errorf("restore plan local path mismatch for %q", action.LocalRelativePath)
+	}
+
+	tmpPath, err := downloadObjectToSiblingTemp(ctx, store, action.RemoteKey, safeLocalPath)
+	if err != nil {
+		return fmt.Errorf("download to temp file: %w", err)
+	}
+	removeTmp := true
+	defer func() {
+		if removeTmp {
+			_ = os.Remove(tmpPath)
+		}
+	}()
+
+	if action.LocalRelativePath == config.PathManifestFile {
+		if err := validateRestoredManifest(ctx, cfg, current, tmpPath); err != nil {
+			return err
+		}
+	}
+
+	if err := os.Chmod(tmpPath, 0o644); err != nil {
+		return fmt.Errorf("set file permissions: %w", err)
+	}
+	if err := os.Rename(tmpPath, safeLocalPath); err != nil {
+		return fmt.Errorf("install file atomically: %w", err)
+	}
+	removeTmp = false
+
+	return nil
+}
+
+func downloadObjectToSiblingTemp(ctx context.Context, store storage.ObjectStore, remoteKey, destPath string) (string, error) {
+	if strings.TrimSpace(destPath) == "" {
+		return "", fmt.Errorf("destination path is required")
+	}
+	dir := filepath.Dir(destPath)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return "", fmt.Errorf("create destination directory: %w", err)
+	}
+	base := filepath.Base(destPath)
+	tmp, err := os.CreateTemp(dir, "."+base+".restore-*.tmp")
+	if err != nil {
+		return "", fmt.Errorf("create temp file: %w", err)
+	}
+	tmpPath := tmp.Name()
+	if err := tmp.Close(); err != nil {
+		_ = os.Remove(tmpPath)
+		return "", fmt.Errorf("close temp file: %w", err)
+	}
+
+	if err := store.Download(ctx, remoteKey, tmpPath); err != nil {
+		_ = os.Remove(tmpPath)
+		return "", err
+	}
+
+	return tmpPath, nil
+}
+
+func validateRestoredManifest(ctx context.Context, cfg *config.Config, current *RemoteCurrentState, path string) error {
+	manifestStore := &manifest.LocalStore{}
+	m, err := manifestStore.Load(ctx, path)
+	if err != nil {
+		return fmt.Errorf("validate manifest decode: %w", err)
+	}
+
+	requestedSession := strings.TrimSpace(cfg.Session.SessionID)
+	requestedCampaign := strings.TrimSpace(cfg.Session.Campaign)
+	manifestSession := strings.TrimSpace(m.SessionID)
+	manifestCampaign := strings.TrimSpace(m.Campaign)
+	if manifestSession != requestedSession {
+		return fmt.Errorf("manifest session_id %q does not match requested session_id %q", manifestSession, requestedSession)
+	}
+	if manifestCampaign == "" {
+		return fmt.Errorf("manifest campaign is required")
+	}
+	if manifestCampaign != requestedCampaign {
+		return fmt.Errorf("manifest campaign %q does not match requested campaign %q", manifestCampaign, requestedCampaign)
+	}
+	if current != nil {
+		if expected := strings.TrimSpace(current.SessionID); expected != "" && manifestSession != expected {
+			return fmt.Errorf("manifest session_id %q does not match discovered session_id %q", manifestSession, expected)
+		}
+		if expected := strings.TrimSpace(current.Campaign); expected != "" && manifestCampaign != expected {
+			return fmt.Errorf("manifest campaign %q does not match discovered campaign %q", manifestCampaign, expected)
+		}
+	}
+
+	return nil
+}

internal/app/restore_execution_test.go

@@ -0,0 +1,313 @@
+package app
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+func TestExecuteRestoreNonDryRunRestoresDurableFiles(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	fake := &storage.FakeBackend{}
+	cfg, sessionPrefix, manifestKey, runIDKey := seedRestoreCommittedState(t, fake, pipelinePath, sessionPath)
+	seedRestoreObject(fake, sessionPrefix+"transcripts/full.json", []byte(`{"segments":[1,2,3]}`))
+	seedRestoreObject(fake, sessionPrefix+"artifacts/session_recap.md", []byte("# recap\n"))
+	seedRestoreObject(fake, sessionPrefix+"audio/alice.flac", []byte("remote-audio"))
+	seedRestoreObject(fake, runIDKey, []byte("20260519T010203Z-a1b2c3d4\n"))
+	seedRestoreObject(fake, manifestKey, restoreManifestJSON(t, cfg.Session.SessionID, cfg.Session.Campaign))
+
+	restoreWithStoreAndRealPhases(t, fake)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
+	}
+	if !strings.Contains(stdout.String(), "restore complete:") {
+		t.Fatalf("stdout = %q, want completion summary", stdout.String())
+	}
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), `{"segments":[1,2,3]}`)
+	mustReadEquals(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n")
+	if _, err := os.Stat(filepath.Join(sessionRoot, "audio", "alice.flac")); !os.IsNotExist(err) {
+		t.Fatalf("audio should not be restored by default; stat err=%v", err)
+	}
+}
+
+func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	fake := &storage.FakeBackend{}
+	cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, sessionPath)
+	seedRestoreObject(fake, sessionPrefix+"audio/alice.flac", []byte("remote-audio"))
+
+	restoreWithStoreAndRealPhases(t, fake)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath, "--include-audio"}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	mustReadEquals(t, filepath.Join(sessionRoot, "audio", "alice.flac"), "remote-audio")
+}
+
+func TestExecuteRestoreConflictWithoutForceDoesNotOverwrite(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	fake := &storage.FakeBackend{}
+	cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, sessionPath)
+	seedRestoreObject(fake, sessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local-transcript")
+
+	restoreWithStoreAndRealPhases(t, fake)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "conflicting path") {
+		t.Fatalf("stderr = %q, want conflict failure", stderr.String())
+	}
+	mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local-transcript")
+}
+
+func TestExecuteRestoreForceOverwritesDifferingFile(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	fake := &storage.FakeBackend{}
+	cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, sessionPath)
+	seedRestoreObject(fake, sessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local-transcript")
+
+	restoreWithStoreAndRealPhases(t, fake)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath, "--force"}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+	mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "remote-transcript")
+}
+
+func TestExecuteRestoreLockConflictFailsAndWritesNothing(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	fake := &storage.FakeBackend{}
+	cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, sessionPath)
+	seedRestoreObject(fake, sessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
+
+	store := artifacts.NewLocalStore(workspaceRoot)
+	lock, err := store.AcquireSessionLockFor(cfg.Session.Campaign, cfg.Session.SessionID)
+	if err != nil {
+		t.Fatalf("AcquireSessionLockFor() error = %v", err)
+	}
+	defer func() { _ = store.ReleaseSessionLock(lock) }()
+
+	restoreWithStoreAndRealPhases(t, fake)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "acquire session lock") {
+		t.Fatalf("stderr = %q, want lock failure", stderr.String())
+	}
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	if _, err := os.Stat(filepath.Join(sessionRoot, "transcripts", "full.json")); !os.IsNotExist(err) {
+		t.Fatalf("transcript should not be restored when lock acquisition fails; stat err=%v", err)
+	}
+}
+
+func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+
+	base := &storage.FakeBackend{}
+	cfg, sessionPrefix, manifestKey, _ := seedRestoreCommittedState(t, base, pipelinePath, sessionPath)
+	seedRestoreObject(base, sessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
+
+	toggled := &stagedManifestDownloadStore{
+		delegate:       base,
+		manifestKey:    manifestKey,
+		firstManifest:  restoreManifestJSON(t, cfg.Session.SessionID, cfg.Session.Campaign),
+		secondManifest: []byte("{invalid json"),
+		manifestReads:  0,
+	}
+
+	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	existing := manifest.New(cfg.Session.SessionID, nowUTC())
+	existing.Campaign = cfg.Session.Campaign
+	existingPath := filepath.Join(sessionRoot, "manifest.json")
+	manifestStore := &manifest.LocalStore{}
+	if err := manifestStore.Save(context.Background(), existingPath, existing); err != nil {
+		t.Fatalf("save existing local manifest: %v", err)
+	}
+	existingData, err := os.ReadFile(existingPath)
+	if err != nil {
+		t.Fatalf("read existing local manifest: %v", err)
+	}
+
+	restoreWithStoreAndRealPhases(t, toggled)
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath, "--force"}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "validate manifest decode") {
+		t.Fatalf("stderr = %q, want manifest validation failure", stderr.String())
+	}
+	mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "remote-transcript")
+	afterData, err := os.ReadFile(existingPath)
+	if err != nil {
+		t.Fatalf("read local manifest after failure: %v", err)
+	}
+	if string(afterData) != string(existingData) {
+		t.Fatalf("local manifest changed after failed restore; before=%q after=%q", string(existingData), string(afterData))
+	}
+}
+
+func TestExecuteRestorePlanPathMismatchFails(t *testing.T) {
+	cfg := restorePlanConfig(t)
+	current := restorePlanCurrentState(t, cfg)
+	store := &storage.FakeBackend{}
+	seedRestoreObject(store, current.SessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
+
+	plan := &RestorePlan{Actions: []RestoreAction{{
+		Kind:              RestoreActionDownload,
+		RemoteKey:         current.SessionPrefix + "transcripts/full.json",
+		LocalRelativePath: "transcripts/full.json",
+		LocalPath:         "/tmp/escape.txt",
+	}}}
+
+	_, err := executeRestorePlan(context.Background(), cfg, current, plan, store)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "local path mismatch") {
+		t.Fatalf("error = %v, want local path mismatch", err)
+	}
+}
+
+func restoreWithStoreAndRealPhases(t *testing.T, objectStore storage.ObjectStore) {
+	t.Helper()
+	origStoreFn := newObjectStoreFromConfigFn
+	origDiscoverFn := discoverRemoteCurrentStateFn
+	origPlanFn := buildRestorePlanFn
+	origExecuteFn := executeRestorePlanFn
+	t.Cleanup(func() {
+		newObjectStoreFromConfigFn = origStoreFn
+		discoverRemoteCurrentStateFn = origDiscoverFn
+		buildRestorePlanFn = origPlanFn
+		executeRestorePlanFn = origExecuteFn
+	})
+
+	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
+		return objectStore, nil
+	}
+	discoverRemoteCurrentStateFn = discoverRemoteCurrentState
+	buildRestorePlanFn = buildRestorePlan
+	executeRestorePlanFn = executeRestorePlan
+}
+
+func seedRestoreCommittedState(t *testing.T, fake *storage.FakeBackend, pipelinePath, sessionPath string) (*config.Config, string, string, string) {
+	t.Helper()
+	cfg, err := config.LoadWithSessionOptions(pipelinePath, sessionPath, config.SessionLoadOptions{})
+	if err != nil {
+		t.Fatalf("LoadWithSessionOptions() error = %v", err)
+	}
+	if err := config.Validate(cfg); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+
+	sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID)
+	manifestKey, runIDKey := artifacts.ResolveArchiveCurrentStateKeys(sessionPrefix)
+
+	seedRestoreObject(fake, runIDKey, []byte("20260519T010203Z-a1b2c3d4\n"))
+	seedRestoreObject(fake, manifestKey, restoreManifestJSON(t, cfg.Session.SessionID, cfg.Session.Campaign))
+
+	return cfg, sessionPrefix, manifestKey, runIDKey
+}
+
+func mustReadEquals(t *testing.T, path, want string) {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("ReadFile(%q): %v", path, err)
+	}
+	if string(data) != want {
+		t.Fatalf("file %q = %q, want %q", path, string(data), want)
+	}
+}
+
+type stagedManifestDownloadStore struct {
+	delegate       *storage.FakeBackend
+	manifestKey    string
+	firstManifest  []byte
+	secondManifest []byte
+	manifestReads  int
+}
+
+func (s *stagedManifestDownloadStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) {
+	return s.delegate.List(ctx, prefix)
+}
+
+func (s *stagedManifestDownloadStore) Download(ctx context.Context, key, localPath string) error {
+	if strings.TrimSpace(key) == strings.TrimSpace(s.manifestKey) {
+		s.manifestReads++
+		payload := s.secondManifest
+		if s.manifestReads <= 1 {
+			payload = s.firstManifest
+		}
+		if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil {
+			return fmt.Errorf("download staged manifest: create parent: %w", err)
+		}
+		if err := os.WriteFile(localPath, payload, 0o644); err != nil {
+			return fmt.Errorf("download staged manifest: write local file: %w", err)
+		}
+		return nil
+	}
+	return s.delegate.Download(ctx, key, localPath)
+}
+
+func (s *stagedManifestDownloadStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
+	return s.delegate.Upload(ctx, localPath, key, opts)
+}
+
+func (s *stagedManifestDownloadStore) Exists(ctx context.Context, key string) (bool, error) {
+	return s.delegate.Exists(ctx, key)
+}

internal/app/restore_test.go

@@ -38,10 +38,12 @@ func TestExecuteRestoreRecognizedAndReturnsNYI(t *testing.T) {
 	origStoreFn := newObjectStoreFromConfigFn
 	origDiscoverFn := discoverRemoteCurrentStateFn
 	origPlanFn := buildRestorePlanFn
+	origExecuteFn := executeRestorePlanFn
 	t.Cleanup(func() {
 		newObjectStoreFromConfigFn = origStoreFn
 		discoverRemoteCurrentStateFn = origDiscoverFn
 		buildRestorePlanFn = origPlanFn
+		executeRestorePlanFn = origExecuteFn
 	})
 	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
 		return &storage.FakeBackend{}, nil
@@ -174,10 +176,12 @@ func TestExecuteRestoreNonDryRunConflictFailsBeforeNYI(t *testing.T) {
 	origStoreFn := newObjectStoreFromConfigFn
 	origDiscoverFn := discoverRemoteCurrentStateFn
 	origPlanFn := buildRestorePlanFn
+	origExecuteFn := executeRestorePlanFn
 	t.Cleanup(func() {
 		newObjectStoreFromConfigFn = origStoreFn
 		discoverRemoteCurrentStateFn = origDiscoverFn
 		buildRestorePlanFn = origPlanFn
+		executeRestorePlanFn = origExecuteFn
 	})
 	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
 		return &storage.FakeBackend{}, nil
@@ -217,14 +221,16 @@ func TestExecuteRestoreNonDryRunConflictFailsBeforeNYI(t *testing.T) {
 	}
 }
 
-func TestExecuteRestoreNonDryRunForceStillStopsAtNYI(t *testing.T) {
+func TestExecuteRestoreNonDryRunForceExecutesPlan(t *testing.T) {
 	origStoreFn := newObjectStoreFromConfigFn
 	origDiscoverFn := discoverRemoteCurrentStateFn
 	origPlanFn := buildRestorePlanFn
+	origExecuteFn := executeRestorePlanFn
 	t.Cleanup(func() {
 		newObjectStoreFromConfigFn = origStoreFn
 		discoverRemoteCurrentStateFn = origDiscoverFn
 		buildRestorePlanFn = origPlanFn
+		executeRestorePlanFn = origExecuteFn
 	})
 	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
 		return &storage.FakeBackend{}, nil
@@ -244,20 +250,26 @@ func TestExecuteRestoreNonDryRunForceStillStopsAtNYI(t *testing.T) {
 			DownloadCount: 1,
 		}, nil
 	}
+	executeRestorePlanFn = func(context.Context, *config.Config, *RemoteCurrentState, *RestorePlan, storage.ObjectStore) (*RestoreExecutionResult, error) {
+		return &RestoreExecutionResult{DownloadedCount: 1}, nil
+	}
 
 	workspaceRoot := t.TempDir()
 	pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
 	code := Execute([]string{"restore", "--config", pipelinePath, "--session", sessionPath, "--force"}, &stdout, &stderr)
-	if code == 0 {
+	if code != 0 {
-		t.Fatal("exit code = 0, want non-zero at phase-4 execution boundary")
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
 	}
 	if !strings.Contains(stdout.String(), "restore plan: session sample-campaign/2026-05-03 run=20260519T010203Z-a1b2c3d4") {
 		t.Fatalf("stdout = %q, want plan output", stdout.String())
 	}
-	if !strings.Contains(stderr.String(), "not yet implemented (phase 4: restore execution)") {
+	if !strings.Contains(stdout.String(), "restore complete: session sample-campaign/2026-05-03 run=20260519T010203Z-a1b2c3d4 downloaded=1 skipped_same=0 conflicts=0") {
-		t.Fatalf("stderr = %q, want phase-4 NYI marker", stderr.String())
+		t.Fatalf("stdout = %q, want completion summary", stdout.String())
+	}
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
 	}
 }