eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Bind restore to committed remote snapshots

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 20:42:43 +00:00
parent eac7e155a5
commit 4158394dcf
17 changed files with 686 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
internal/adapters/storage/fake.go
View File

@@ -23,11 +23,12 @@ type FakeBackend struct {
Uploads []FakeUploadCall
Downloads []FakeDownloadCall
ListErr error
DownloadErr error
UploadErr error
ExistsErr error
UploadHook func(FakeUploadCall) error
ListErr error
DownloadErr error
UploadErr error
ExistsErr error
UploadHook func(FakeUploadCall) error
DownloadHook func(FakeDownloadCall) error
}
// FakeUploadCall captures one upload invocation in call order.
@@ -139,6 +140,11 @@ func (f *FakeBackend) DownloadTo(ctx context.Context, key string, destination io
if destination == nil {
return fmt.Errorf("download object: destination writer is required")
}
if f.DownloadHook != nil {
if err := f.DownloadHook(FakeDownloadCall{Key: normalizeObjectKey(key)}); err != nil {
return err
}
}
_, source, err := f.Read(ctx, key)
if err != nil {

41
internal/app/operator_artifact_rendering.go
View File

@@ -129,6 +129,47 @@ func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config,
return out
}
func remotePublishedOutputAvailabilityForCurrent(
ctx context.Context,
cfg *config.Config,
store storage.ObjectStore,
catalog *artifacts.ArtifactCatalog,
current *RemoteCurrentState,
) map[string]string {
if current == nil || current.Commit == nil {
return remotePublishedOutputAvailability(ctx, cfg, store, catalog)
}
out := map[string]string{}
runPrefix := artifacts.S3RunPrefix(current.SessionPrefix, current.RunID)
for _, rule := range cfg.Pipeline.Publish.Outputs {
source := strings.TrimSpace(rule.Source)
dest, _, err := helperPublishedOutputDest(rule, catalog)
if err != nil {
out[publishedOutputRemoteStateKey(source, "")] = "remote=error"
continue
}
key := artifacts.S3RunRelativeDestinationKey(runPrefix, dest)
if committedPublishedOutput(current.Commit, key) {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=published"
} else {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing"
}
}
return out
}
func committedPublishedOutput(commit *artifacts.RemoteCommitManifest, key string) bool {
if commit == nil {
return false
}
for _, artifact := range commit.Artifacts {
if artifact.Type == artifacts.RemoteArtifactTypePublishedOutput && artifact.DestinationKey == key {
return true
}
}
return false
}
func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog) (string, bool, error) {
source := strings.TrimSpace(rule.Source)
normalized, err := artifactpolicy.ResolvePublishedDestinationWithExtractions(

4
internal/app/operator_status.go
View File

@@ -53,6 +53,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
}
store, storeErr := objectStoreIfConfigured(ctx, cfg)
var remoteCurrent *RemoteCurrentState
if storeErr != nil {
fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr)
} else if store != nil {
@@ -60,6 +61,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
if current.Err != nil {
fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", current.Err)
} else {
remoteCurrent = current.State
fmt.Fprintf(out, "Remote publish: current run %s\n", current.State.RunID)
fmt.Fprintf(out, "Remote manifest: %s\n", current.State.CurrentManifestKey)
}
@@ -87,7 +89,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
}
publishedRemoteState := map[string]string{}
if store != nil {
publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog)
publishedRemoteState = remotePublishedOutputAvailabilityForCurrent(ctx, cfg, store, catalog, remoteCurrent)
}
fmt.Fprintln(out, "Remote outputs:")
writeArtifactList(out, cfg, catalog, catalogLocks, publishedRemoteState)

4
internal/app/restore.go
View File

@@ -110,13 +110,13 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error
}
}()
if plan.ConflictCount > 0 && !force {
if plan.ConflictCount > 0 {
report.setFailed(fmt.Errorf("conflict: %d conflicting path(s)", plan.ConflictCount))
if _, reportErr := persistRestoreReport(artifactStore, cfg, report); reportErr != nil {
return fmt.Errorf("restore: report failure: %w", reportErr)
}
return fmt.Errorf(
"restore conflict: %d conflicting path(s); rerun with --force to overwrite (download=%d skip_same=%d conflicts=%d)",
"restore conflict: %d conflicting path(s); --force can replace eligible regular files but not unresolved conflicts (download=%d skip_same=%d conflicts=%d)",
plan.ConflictCount,
plan.DownloadCount,
plan.SkipSameCount,

258
internal/app/restore_commit_test.go Normal file
View File

@@ -0,0 +1,258 @@
package app
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"os"
"path/filepath"
"sort"
"testing"
"time"
"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 TestCommittedRestorePlanUsesOnlyDeclaredObjects(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte(`{"segments":[1]}`),
})
fake.SeedObject(storage.FakeObject{Key: current.SessionPrefix + "artifacts/stale.md", Data: []byte("stale")})
fake.SeedObject(storage.FakeObject{Key: artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, "20260519T010204Z-e5f6a7b8"), "transcripts/other.json"), Data: []byte("other")})
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
got := actionRelPaths(plan.Actions)
want := []string{"manifest.json", "transcripts/full.json"}
if len(got) != len(want) {
t.Fatalf("action paths = %#v, want %#v", got, want)
}
for index := range want {
if got[index] != want[index] {
t.Fatalf("action paths = %#v, want %#v", got, want)
}
}
}
func TestCommittedStatusReportsOnlyDeclaredPublishedOutputs(t *testing.T) {
cfg := restorePlanConfig(t)
cfg.Pipeline.Publish = &config.PublishConfig{Outputs: []config.PublishOutputRule{{
Source: "narratio.transcript.final_trimmed",
Dest: "transcripts/full.json",
}}}
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("declared\n"),
})
fake.SeedObject(storage.FakeObject{Key: current.SessionPrefix + "transcripts/full.json", Data: []byte("mutable stale copy\n")})
availability := remotePublishedOutputAvailabilityForCurrent(context.Background(), cfg, fake, nil, current)
key := publishedOutputRemoteStateKey("narratio.transcript.final_trimmed", "transcripts/full.json")
if availability[key] != "remote=published" {
t.Fatalf("availability = %#v, want committed published output", availability)
}
}
func TestCommittedRestoreKeepsSelectedSnapshotWhenPointerChanges(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
first := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("from first commit\n"),
})
plan, err := buildRestorePlan(context.Background(), cfg, first, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
_ = seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010204Z-e5f6a7b8", map[string][]byte{
"transcripts/full.json": []byte("from second commit\n"),
})
if _, err := executeRestorePlan(context.Background(), cfg, first, plan, nil, fake); err != nil {
t.Fatalf("executeRestorePlan() error = %v", err)
}
root := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
mustReadEquals(t, filepath.Join(root, "transcripts", "full.json"), "from first commit\n")
restored, err := (&manifest.LocalStore{}).Load(context.Background(), filepath.Join(root, "manifest.json"))
if err != nil {
t.Fatalf("load restored manifest: %v", err)
}
if restored.RunID != first.RunID {
t.Fatalf("restored run id = %q, want %q", restored.RunID, first.RunID)
}
}
func TestCommittedRestoreRejectsChangedDeclaredObjectBeforeManifestInstall(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("committed bytes\n"),
})
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("changed bytes\n")})
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
t.Fatal("executeRestorePlan() error = nil, want committed-object verification failure")
}
root := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
if _, err := os.Stat(filepath.Join(root, "manifest.json")); !os.IsNotExist(err) {
t.Fatalf("manifest should not be installed after failed restore; stat err=%v", err)
}
}
func TestCommittedRestoreRejectsChangedDeclaredObjectGeneration(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("committed bytes\n"),
})
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("committed bytes\n"), ETag: "replacement-generation"})
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
t.Fatal("executeRestorePlan() error = nil, want generation verification failure")
}
root := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
if _, err := os.Stat(filepath.Join(root, "manifest.json")); !os.IsNotExist(err) {
t.Fatalf("manifest should not be installed after failed restore; stat err=%v", err)
}
}
func TestCommittedRestoreRejectsMissingDeclaredObjectBeforeManifestInstall(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("committed bytes\n"),
})
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
missingKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
if call.Key == missingKey {
return os.ErrNotExist
}
return nil
}
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
t.Fatal("executeRestorePlan() error = nil, want missing-object failure")
}
root := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
if _, err := os.Stat(filepath.Join(root, "manifest.json")); !os.IsNotExist(err) {
t.Fatalf("manifest should not be installed after failed restore; stat err=%v", err)
}
}
func TestCommittedRestoreForceRetainsDirectoryConflict(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("committed bytes\n"),
})
root := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
if err := os.MkdirAll(filepath.Join(root, "transcripts", "full.json"), 0o755); err != nil {
t.Fatalf("MkdirAll() error = %v", err)
}
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{Force: true})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
if plan.ConflictCount != 1 {
t.Fatalf("ConflictCount = %d, want 1", plan.ConflictCount)
}
for _, action := range plan.Actions {
if action.LocalRelativePath == "transcripts/full.json" && action.ConflictKind != RestoreConflictDirectory {
t.Fatalf("ConflictKind = %q, want %q", action.ConflictKind, RestoreConflictDirectory)
}
}
}
func seedCommittedRestoreSnapshot(t *testing.T, cfg *config.Config, fake *storage.FakeBackend, runID string, outputs map[string][]byte) *RemoteCurrentState {
t.Helper()
sessionPrefix := artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.SessionID)
remoteManifest := manifest.New(cfg.Session.SessionID, time.Now().UTC())
remoteManifest.Campaign = cfg.Session.Campaign
remoteManifest.RunID = runID
manifestData, err := json.Marshal(remoteManifest)
if err != nil {
t.Fatalf("marshal manifest: %v", err)
}
manifestData = append(manifestData, '\n')
manifestKey := artifacts.S3RunSessionManifestKey(sessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: manifestKey, Data: manifestData})
artifactsByKey := []artifacts.RemoteArtifact{remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypeSessionManifest, "session.manifest", manifestKey)}
paths := make([]string, 0, len(outputs))
for relative := range outputs {
paths = append(paths, relative)
}
sort.Strings(paths)
for _, relative := range paths {
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), relative)
fake.SeedObject(storage.FakeObject{Key: key, Data: outputs[relative]})
artifactsByKey = append(artifactsByKey, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, "narratio.test", key))
}
commit := artifacts.RemoteCommitManifest{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: cfg.Session.Campaign,
SessionID: cfg.Session.SessionID,
RunID: runID,
Artifacts: artifactsByKey,
}
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
if err != nil {
t.Fatalf("encode remote commit: %v", err)
}
commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: commitKey, Data: commitData})
commitObject := fake.Objects[commitKey]
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: cfg.Session.Campaign,
SessionID: cfg.Session.SessionID,
RunID: runID,
CommitKey: commitKey,
CommitSHA256: restoreCommitSHA256(commitData),
CommitSize: int64(len(commitData)),
CommitGeneration: commitObject.ETag,
})
if err != nil {
t.Fatalf("encode current pointer: %v", err)
}
fake.SeedObject(storage.FakeObject{Key: artifacts.S3CurrentCommitPointerKey(sessionPrefix), Data: pointerData})
current, err := discoverRemoteCurrentState(context.Background(), cfg, fake)
if err != nil {
t.Fatalf("discoverRemoteCurrentState() error = %v", err)
}
return current
}
func remoteRestoreArtifact(fake *storage.FakeBackend, artifactType artifacts.RemoteArtifactType, source, key string) artifacts.RemoteArtifact {
object := fake.Objects[key]
return artifacts.RemoteArtifact{
Type: artifactType, Source: source, DestinationKey: key, SHA256: restoreCommitSHA256(object.Data), Size: int64(len(object.Data)), Generation: object.ETag,
}
}
func restoreCommitSHA256(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}

3
internal/app/restore_discovery.go
View File

@@ -21,6 +21,7 @@ type RemoteCurrentState struct {
SessionID string
Campaign string
Manifest *manifest.Manifest
Commit *artifacts.RemoteCommitManifest
}
func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*RemoteCurrentState, error) {
@@ -44,6 +45,7 @@ func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store s
current, err := artifacts.LoadCurrentState(ctx, store, sessionPrefix, artifacts.CurrentStateValidation{
ExpectedSessionID: requestedSession,
ExpectedCampaign: requestedCampaign,
ValidateRunID: true,
})
if err != nil {
return nil, fmt.Errorf("remote %w", err)
@@ -58,5 +60,6 @@ func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store s
SessionID: strings.TrimSpace(current.Manifest.SessionID),
Campaign: strings.TrimSpace(current.Manifest.Campaign),
Manifest: current.Manifest,
Commit: current.Commit,
}, nil
}

1
internal/app/restore_discovery_test.go
View File

@@ -228,6 +228,7 @@ func restoreManifestJSON(t *testing.T, sessionID, campaign string) []byte {
payload := map[string]any{
"session_id": sessionID,
"campaign": campaign,
"run_id": "20260519T010203Z-a1b2c3d4",
"created_at": now,
"updated_at": now,
"stages": map[string]any{},

58
internal/app/restore_execute.go
View File

@@ -2,6 +2,8 @@ package app
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"path/filepath"
@@ -108,6 +110,9 @@ func executeRestoreDownloadAction(
return fmt.Errorf("download to temp file: %w", err)
}
defer func() { _ = temporary.Cleanup() }()
if err := verifyRestoredObject(ctx, store, action, temporary); err != nil {
return err
}
if action.LocalRelativePath == config.PathManifestFile {
file, err := temporary.Open()
@@ -131,6 +136,56 @@ func executeRestoreDownloadAction(
return nil
}
func verifyRestoredObject(ctx context.Context, store storage.ObjectStore, action RestoreAction, temporary *fileops.DownloadedTempFile) error {
if strings.TrimSpace(action.SHA256) == "" && strings.TrimSpace(action.Generation) == "" {
return nil
}
if strings.TrimSpace(action.SHA256) == "" || strings.TrimSpace(action.Generation) == "" {
return fmt.Errorf("committed object identity for %q is incomplete", action.RemoteKey)
}
file, err := temporary.Open()
if err != nil {
return fmt.Errorf("open downloaded object for verification: %w", err)
}
digest := sha256.New()
count, copyErr := io.Copy(digest, file)
closeErr := file.Close()
if copyErr != nil {
return fmt.Errorf("checksum downloaded object: %w", copyErr)
}
if closeErr != nil {
return fmt.Errorf("close downloaded object: %w", closeErr)
}
if count != action.Size {
return fmt.Errorf("committed object size mismatch for %q: got %d, want %d", action.RemoteKey, count, action.Size)
}
if got := hex.EncodeToString(digest.Sum(nil)); got != action.SHA256 {
return fmt.Errorf("committed object checksum mismatch for %q: got %s, want %s", action.RemoteKey, got, action.SHA256)
}
objects, err := store.List(ctx, action.RemoteKey)
if err != nil {
return fmt.Errorf("read committed object identity for %q: %w", action.RemoteKey, err)
}
var found *storage.ObjectInfo
for _, object := range objects {
if normalizeRemoteKey(object.Key) != normalizeRemoteKey(action.RemoteKey) {
continue
}
if found != nil {
return fmt.Errorf("committed object %q is ambiguous", action.RemoteKey)
}
copy := object
found = &copy
}
if found == nil {
return fmt.Errorf("committed object %q is missing", action.RemoteKey)
}
if found.Size != action.Size || strings.TrimSpace(found.ETag) != action.Generation {
return fmt.Errorf("committed object generation mismatch for %q", action.RemoteKey)
}
return nil
}
func executeRestoreAudioAction(
ctx context.Context,
cfg *config.Config,
@@ -190,6 +245,9 @@ func validateRestoredManifest(ctx context.Context, cfg *config.Config, current *
if expected := strings.TrimSpace(current.Campaign); expected != "" && manifestCampaign != expected {
return fmt.Errorf("manifest campaign %q does not match discovered campaign %q", manifestCampaign, expected)
}
if expected := strings.TrimSpace(current.RunID); expected != "" && strings.TrimSpace(m.RunID) != expected {
return fmt.Errorf("manifest run_id %q does not match discovered run_id %q", strings.TrimSpace(m.RunID), expected)
}
}
return nil

236
internal/app/restore_plan.go
View File

@@ -27,17 +27,29 @@ const (
RestoreActionConflict RestoreActionKind = "conflict"
)
// RestoreConflictKind identifies why a local target cannot be restored safely.
type RestoreConflictKind string
const (
RestoreConflictContentMismatch RestoreConflictKind = "content_mismatch"
RestoreConflictDirectory RestoreConflictKind = "directory"
RestoreConflictNonRegular RestoreConflictKind = "non_regular"
)
// RestoreAction is one deterministic planner action.
type RestoreAction struct {
Kind RestoreActionKind
RemoteKey string
LocalRelativePath string
LocalPath string
SHA256 string
Generation string
Size int64
ETag string
ExistsLocal bool
SameLocal bool
Conflict bool
ConflictKind RestoreConflictKind
Reason string
}
@@ -66,55 +78,10 @@ func buildRestorePlan(ctx context.Context, cfg *config.Config, current *RemoteCu
if store == nil {
return nil, fmt.Errorf("remote object store is required")
}
prefix := normalizeRemoteKey(current.SessionPrefix)
if strings.TrimSpace(prefix) == "" {
return nil, fmt.Errorf("remote session prefix is required")
}
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
}
sessionPaths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
objects, err := store.List(ctx, prefix)
actions, err := buildCurrentRestoreActions(ctx, current, store, sessionPaths, opts)
if err != nil {
return nil, fmt.Errorf("list remote session objects under %q: %w", prefix, err)
}
candidates := make(map[string]storage.ObjectInfo, len(objects)+1)
for _, obj := range objects {
key := normalizeRemoteKey(obj.Key)
if key == "" {
continue
}
obj.Key = key
candidates[key] = obj
}
if strings.TrimSpace(current.CurrentManifestKey) != "" {
key := normalizeRemoteKey(current.CurrentManifestKey)
if _, ok := candidates[key]; !ok {
candidates[key] = storage.ObjectInfo{Key: key}
}
}
actions := make([]RestoreAction, 0, len(candidates))
for key, obj := range candidates {
rel, include, err := restoreLocalRelativePathForKey(prefix, normalizeRemoteKey(current.CurrentManifestKey), key, opts.IncludeAudio)
if err != nil {
return nil, fmt.Errorf("map remote key %q: %w", key, err)
}
if !include {
continue
}
localPath, err := joinWithinSessionRoot(sessionPaths.Root, rel)
if err != nil {
return nil, fmt.Errorf("map remote key %q: %w", key, err)
}
action, err := classifyRestoreAction(ctx, store, obj, rel, localPath, opts.Force)
if err != nil {
return nil, fmt.Errorf("classify remote key %q: %w", key, err)
}
actions = append(actions, action)
return nil, err
}
previousActions, err := buildPreviousCacheRestoreActions(ctx, cfg, sessionPaths, store, opts.Force)
@@ -146,6 +113,138 @@ func buildRestorePlan(ctx context.Context, cfg *config.Config, current *RemoteCu
return plan, nil
}
func buildCurrentRestoreActions(
ctx context.Context,
current *RemoteCurrentState,
store storage.ObjectStore,
sessionPaths artifacts.SessionPaths,
opts RestorePlanOptions,
) ([]RestoreAction, error) {
if current != nil && current.Commit != nil {
return buildCommittedRestoreActions(ctx, current, store, sessionPaths, opts)
}
return buildLegacyRestoreActions(ctx, current, store, sessionPaths, opts)
}
func buildCommittedRestoreActions(
ctx context.Context,
current *RemoteCurrentState,
store storage.ObjectStore,
sessionPaths artifacts.SessionPaths,
opts RestorePlanOptions,
) ([]RestoreAction, error) {
if current == nil || current.Commit == nil {
return nil, fmt.Errorf("committed remote current state is required")
}
runPrefix := artifacts.S3RunPrefix(current.SessionPrefix, current.RunID)
if runPrefix == "" {
return nil, fmt.Errorf("committed run prefix is required")
}
actions := make([]RestoreAction, 0, len(current.Commit.Artifacts))
for _, artifact := range current.Commit.Artifacts {
rel, include, err := restoreLocalRelativePathForCommittedArtifact(runPrefix, artifact, opts.IncludeAudio)
if err != nil {
return nil, fmt.Errorf("map committed object %q: %w", artifact.DestinationKey, err)
}
if !include {
continue
}
localPath, err := joinWithinSessionRoot(sessionPaths.Root, rel)
if err != nil {
return nil, fmt.Errorf("map committed object %q: %w", artifact.DestinationKey, err)
}
action, err := classifyRestoreAction(ctx, store, storage.ObjectInfo{
Key: artifact.DestinationKey, Size: artifact.Size, ETag: artifact.Generation,
}, artifact.SHA256, rel, localPath, opts.Force)
if err != nil {
return nil, fmt.Errorf("classify committed object %q: %w", artifact.DestinationKey, err)
}
actions = append(actions, action)
}
return actions, nil
}
func buildLegacyRestoreActions(
ctx context.Context,
current *RemoteCurrentState,
store storage.ObjectStore,
sessionPaths artifacts.SessionPaths,
opts RestorePlanOptions,
) ([]RestoreAction, error) {
prefix := normalizeRemoteKey(current.SessionPrefix)
if strings.TrimSpace(prefix) == "" {
return nil, fmt.Errorf("remote session prefix is required")
}
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
}
objects, err := store.List(ctx, prefix)
if err != nil {
return nil, fmt.Errorf("list remote session objects under %q: %w", prefix, err)
}
candidates := make(map[string]storage.ObjectInfo, len(objects)+1)
for _, obj := range objects {
key := normalizeRemoteKey(obj.Key)
if key != "" {
obj.Key = key
candidates[key] = obj
}
}
if key := normalizeRemoteKey(current.CurrentManifestKey); key != "" {
if _, ok := candidates[key]; !ok {
candidates[key] = storage.ObjectInfo{Key: key}
}
}
actions := make([]RestoreAction, 0, len(candidates))
for key, object := range candidates {
rel, include, err := restoreLocalRelativePathForKey(prefix, normalizeRemoteKey(current.CurrentManifestKey), key, opts.IncludeAudio)
if err != nil {
return nil, fmt.Errorf("map remote key %q: %w", key, err)
}
if !include {
continue
}
localPath, err := joinWithinSessionRoot(sessionPaths.Root, rel)
if err != nil {
return nil, fmt.Errorf("map remote key %q: %w", key, err)
}
action, err := classifyRestoreAction(ctx, store, object, "", rel, localPath, opts.Force)
if err != nil {
return nil, fmt.Errorf("classify remote key %q: %w", key, err)
}
actions = append(actions, action)
}
return actions, nil
}
func restoreLocalRelativePathForCommittedArtifact(runPrefix string, artifact artifacts.RemoteArtifact, includeAudio bool) (string, bool, error) {
if artifact.Type == artifacts.RemoteArtifactTypeSessionManifest {
return config.PathManifestFile, true, nil
}
if artifact.Type != artifacts.RemoteArtifactTypePublishedOutput {
return "", false, nil
}
key := normalizeRemoteKey(artifact.DestinationKey)
if !strings.HasPrefix(key, runPrefix) {
return "", false, fmt.Errorf("object is outside committed run prefix %q", runPrefix)
}
rel := strings.TrimPrefix(key, runPrefix)
cleanRel, err := pathsafe.NormalizeRelativeDestination(rel)
if err != nil {
return "", false, fmt.Errorf("committed destination is unsafe: %w", err)
}
if cleanRel == config.PathManifestFile {
return "", false, fmt.Errorf("published output conflicts with the session manifest path")
}
if strings.HasPrefix(cleanRel, config.PathTranscriptsSegment+"/") || strings.HasPrefix(cleanRel, config.PathArtifactsDirSegment+"/") {
return cleanRel, true, nil
}
if includeAudio && strings.HasPrefix(cleanRel, config.PathAudioDirSegment+"/") {
return cleanRel, true, nil
}
return "", false, nil
}
func normalizeRemoteKey(v string) string {
return strings.Trim(strings.ReplaceAll(strings.TrimSpace(v), "\\", "/"), "/")
}
@@ -250,7 +349,9 @@ func buildPreviousCacheRestoreActions(
}
actions := make([]RestoreAction, 0, len(plan.Records))
for _, record := range plan.Records {
action, err := classifyRestoreAction(ctx, store, storage.ObjectInfo{Key: record.RemoteKey}, record.LocalRelativePath, record.LocalPath, force)
action, err := classifyRestoreAction(ctx, store, storage.ObjectInfo{
Key: record.RemoteKey, Size: record.Size, ETag: record.Generation,
}, record.SHA256, record.LocalRelativePath, record.LocalPath, force)
if err != nil {
return nil, fmt.Errorf("classify previous-session cache object %q: %w", record.RemoteKey, err)
}
@@ -263,6 +364,7 @@ func classifyRestoreAction(
ctx context.Context,
store storage.ObjectStore,
object storage.ObjectInfo,
expectedSHA256 string,
localRelPath string,
localPath string,
force bool,
@@ -271,9 +373,13 @@ func classifyRestoreAction(
RemoteKey: normalizeRemoteKey(object.Key),
LocalRelativePath: localRelPath,
LocalPath: localPath,
SHA256: strings.TrimSpace(expectedSHA256),
Size: object.Size,
ETag: object.ETag,
}
if action.SHA256 != "" {
action.Generation = strings.TrimSpace(object.ETag)
}
info, err := os.Stat(localPath)
if err != nil {
@@ -289,9 +395,39 @@ func classifyRestoreAction(
if info.IsDir() {
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictDirectory
action.Reason = "local path is a directory"
return action, nil
}
if !info.Mode().IsRegular() {
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictNonRegular
action.Reason = "local path is not a regular file"
return action, nil
}
if action.SHA256 != "" {
localDigest, err := artifacts.SHA256File(localPath)
if err != nil {
return RestoreAction{}, fmt.Errorf("checksum local file: %w", err)
}
if localDigest == action.SHA256 {
action.Kind = RestoreActionSkipSame
action.SameLocal = true
action.Reason = "local file matches committed content"
return action, nil
}
if force {
action.Kind = RestoreActionDownload
action.Reason = "local file differs from committed content; overwrite with --force"
return action, nil
}
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictContentMismatch
action.Reason = "local file differs from committed content"
return action, nil
}
if restoreRelativePathIsAudio(localRelPath) {
if object.Size > 0 {
@@ -308,6 +444,7 @@ func classifyRestoreAction(
}
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictContentMismatch
action.Reason = "local audio differs (size mismatch)"
return action, nil
}
@@ -318,6 +455,7 @@ func classifyRestoreAction(
}
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictContentMismatch
action.Reason = "local audio exists; remote size unavailable"
return action, nil
}
@@ -330,6 +468,7 @@ func classifyRestoreAction(
}
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictContentMismatch
action.Reason = "local file differs (size mismatch)"
return action, nil
}
@@ -364,6 +503,7 @@ func classifyRestoreAction(
action.Kind = RestoreActionConflict
action.Conflict = true
action.ConflictKind = RestoreConflictContentMismatch
action.Reason = "local file differs"
return action, nil
}

41
internal/app/restore_test.go
View File

@@ -320,7 +320,7 @@ func TestExecuteRestoreNonDryRunConflictFailsBeforeNYI(t *testing.T) {
if stdout.Len() != 0 {
t.Fatalf("stdout = %q, want empty on conflict failure", stdout.String())
}
if !strings.Contains(stderr.String(), "restore conflict: 1 conflicting path(s); rerun with --force to overwrite") {
if !strings.Contains(stderr.String(), "restore conflict: 1 conflicting path(s)") {
t.Fatalf("stderr = %q, want conflict failure", stderr.String())
}
if strings.Contains(stderr.String(), "phase 4: restore execution") {
@@ -328,6 +328,45 @@ func TestExecuteRestoreNonDryRunConflictFailsBeforeNYI(t *testing.T) {
}
}
func TestExecuteRestoreForceStillBlocksUnresolvedConflict(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
}
discoverRemoteCurrentStateFn = func(context.Context, *config.Config, storage.ObjectStore) (*RemoteCurrentState, error) {
return &RemoteCurrentState{SessionID: "2026-05-03", Campaign: "sample-campaign", RunID: "20260519T010203Z-a1b2c3d4"}, nil
}
buildRestorePlanFn = func(context.Context, *config.Config, *RemoteCurrentState, storage.ObjectStore, RestorePlanOptions) (*RestorePlan, error) {
return &RestorePlan{Actions: []RestoreAction{{Kind: RestoreActionConflict, LocalRelativePath: "artifacts/session_recap.md", Reason: "local path is a directory"}}, ConflictCount: 1}, nil
}
executed := false
executeRestorePlanFn = func(context.Context, *config.Config, *RemoteCurrentState, *RestorePlan, *RestoreReport, storage.ObjectStore) (*RestoreExecutionResult, error) {
executed = true
return &RestoreExecutionResult{}, nil
}
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"session", "restore", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"}, &stdout, &stderr)
if code == 0 {
t.Fatal("exit code = 0, want conflict failure")
}
if executed {
t.Fatal("restore execution ran despite unresolved conflict")
}
}
func TestExecuteRestoreNonDryRunForceExecutesPlan(t *testing.T) {
origStoreFn := newObjectStoreFromConfigFn
origDiscoverFn := discoverRemoteCurrentStateFn

60
internal/previouscache/previouscache.go
View File

@@ -36,6 +36,9 @@ type Record struct {
LocalRelativePath string
LocalPath string
RemoteKey string
SHA256 string
Size int64
Generation string
S3Bucket string
}
@@ -119,13 +122,23 @@ func BuildPlan(
if err != nil {
return nil, err
}
result.Records = append(result.Records, Record{
manifestRecord := Record{
Kind: InputKindManifest,
LocalRelativePath: manifestRel,
LocalPath: paths.PreviousManifestPath,
RemoteKey: currentManifestKey,
S3Bucket: bucket,
})
}
if current.Commit != nil {
committedManifest, ok := current.Commit.Artifact(artifacts.RemoteArtifactTypeSessionManifest)
if !ok || committedManifest.DestinationKey != currentManifestKey {
return nil, fmt.Errorf("previous-session remote commit does not declare its session manifest")
}
manifestRecord.SHA256 = committedManifest.SHA256
manifestRecord.Size = committedManifest.Size
manifestRecord.Generation = committedManifest.Generation
}
result.Records = append(result.Records, manifestRecord)
for _, requirement := range orderedRequirements {
candidates := artifactRelativePathCandidates(requirement.Name, previousManifest, cfg)
@@ -142,18 +155,28 @@ func BuildPlan(
selectedRel := ""
selectedKey := ""
var selectedArtifact *artifacts.RemoteArtifact
for _, candidate := range candidates {
if current.Commit != nil {
artifact, ok := committedPublishedArtifact(current.Commit, previousSessionPrefix, candidate)
if !ok {
continue
}
selectedRel = candidate
selectedKey = artifact.DestinationKey
selectedArtifact = &artifact
break
}
remoteKey := artifacts.S3PublishedOutputKey(previousSessionPrefix, candidate)
exists, err := store.Exists(ctx, remoteKey)
if err != nil {
return nil, fmt.Errorf("check previous-session artifact object %q: %w", remoteKey, err)
}
if !exists {
continue
if exists {
selectedRel = candidate
selectedKey = remoteKey
break
}
selectedRel = candidate
selectedKey = remoteKey
break
}
if selectedRel == "" {
if requirement.Required {
@@ -174,7 +197,7 @@ func BuildPlan(
if err != nil {
return nil, err
}
result.Records = append(result.Records, Record{
record := Record{
Kind: InputKindArtifact,
RequirementName: requirement.Name,
Required: requirement.Required,
@@ -182,7 +205,13 @@ func BuildPlan(
LocalPath: localPath,
RemoteKey: selectedKey,
S3Bucket: bucket,
})
}
if selectedArtifact != nil {
record.SHA256 = selectedArtifact.SHA256
record.Size = selectedArtifact.Size
record.Generation = selectedArtifact.Generation
}
result.Records = append(result.Records, record)
}
sort.Strings(result.SkippedMissing)
@@ -195,6 +224,19 @@ func BuildPlan(
return result, nil
}
func committedPublishedArtifact(commit *artifacts.RemoteCommitManifest, sessionPrefix, relativePath string) (artifacts.RemoteArtifact, bool) {
if commit == nil {
return artifacts.RemoteArtifact{}, false
}
want := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, commit.RunID), relativePath)
for _, artifact := range commit.Artifacts {
if artifact.Type == artifacts.RemoteArtifactTypePublishedOutput && artifact.DestinationKey == want {
return artifact, true
}
}
return artifacts.RemoteArtifact{}, false
}
func requiredPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string {
names := make([]string, 0, len(requirements))
for _, requirement := range requirements {