diff --git a/docs/cli.md b/docs/cli.md index 706aaae..3b99f32 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -168,7 +168,8 @@ Read-only preflight checks for config validity, required inputs, audio mode, pre narratio session status [...common config flags] ``` -Prints local manifest state and, when storage is available, remote current-state and published-output status. +Prints local manifest state and, when storage is available, status for the +pointer-selected remote commit and its declared published outputs. ### `session init` @@ -211,7 +212,8 @@ Behavior: - discovers committed remote current state; - plans local restores; - writes an execution report; -- blocks conflicting overwrites unless `--force` is set. +- blocks unresolved conflicts. `--force` permits replacement only of eligible + regular files. See [Operations: Restore Workflow](./operations.md#restore-workflow) for the default restore scope, report location, and conflict-handling workflow. diff --git a/docs/internal/artifacts.md b/docs/internal/artifacts.md index aaf81ac..6eade94 100644 --- a/docs/internal/artifacts.md +++ b/docs/internal/artifacts.md @@ -100,6 +100,11 @@ Validation by content type: Artifacts package owns shared remote current-state loading mechanics used by restore, status and validation checks, and previous-cache planning. +For a new-protocol current state, the pointer-selected immutable commit is the +complete restore authority. Callers receive its declared object identities and +must not supplement them by listing mutable session prefixes. The legacy reader +is intentionally separate and remains migration-only support. + Core helpers: - `LoadCurrentState` diff --git a/docs/internal/command-restore.md b/docs/internal/command-restore.md index a7d5078..c1e3d01 100644 --- a/docs/internal/command-restore.md +++ b/docs/internal/command-restore.md @@ -18,6 +18,7 @@ Discovery delegates current-state pointer and manifest loading to - campaign must match; - session ID must match. +- run ID must match the pointer-selected committed run. Restore treats any missing or invalid remote current state as a command error. @@ -31,10 +32,13 @@ Restore planner action kinds: Planner behavior: -- remote list scope is the resolved session prefix; +- a new-protocol restore uses only the selected commit's declared artifact set; + each action carries that artifact's immutable key, checksum, size, and + generation. Coherent legacy state remains on the isolated compatibility path; - remote-to-local mapping is traversal-safe; - actions are sorted by local relative path and then remote key; -- force converts differing local targets from conflicts to downloads. +- force converts differing eligible regular files from conflicts to downloads; + directories and other non-regular targets remain conflicts. Previous-cache files are planned separately through `previouscache.BuildPlan` when configured previous-session requirements exist. @@ -47,6 +51,8 @@ Execution order and safety: - `manifest.json` installs last; - downloads use sibling temp files plus atomic rename; - manifest replacement is validated before rename; +- each committed object is verified against its declared checksum, size, and + generation before installation; - failed installs do not roll back files already written in the same execution. Audio restore path: @@ -65,6 +71,9 @@ Audio restore path: ## Invariants - restore uses committed remote current state as authority; +- one restore or status inspection observes the single pointer-selected commit + loaded at discovery; later pointer changes cannot add objects or substitute a + different run into its plan; - a verified `current/commit-pointer.json` and its selected immutable commit establish new-protocol remote commitment; coherent legacy `current/run_id.txt` plus `current/manifest.json` remains read-only migration diff --git a/docs/operations.md b/docs/operations.md index 2f816f5..e811f5b 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -271,15 +271,15 @@ narratio session restore 2026-04-04 Default restore scope: -- `manifest.json` -- `transcripts/**` -- `artifacts/**` +- the committed session manifest and the committed transcript/artifact objects + declared by the selected remote commit - `previous/**` when needed by configured previous-session artifact inputs Optional: - `--include-audio` to include `audio/**` -- `--force` to overwrite local conflicts +- `--force` to overwrite eligible conflicting regular files; it never replaces + directories or other non-regular local targets Restore writes an execution report at `reports/restore-latest.json`. diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 9e694fb..fd1d9da 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -119,6 +119,11 @@ install the validated session manifest after other restored durable files. The physical workflow and recovery procedures belong in [Operations](../operations.md). +For the immutable remote-commit protocol, a restore or status operation binds +to one pointer-selected commit and only its declared object identities. A force +flag may replace an eligible regular managed file, but never turns a directory +or other non-regular conflict into a successful restore. + ## Configuration Configuration is strict, explicit, centralized, and operator-oriented. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2ae3426..184033f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -32,7 +32,7 @@ All stages are pending when this plan is created. | 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Completed | | 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Completed | | 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Completed | -| 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Pending | +| 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Completed | | 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Pending | | 19 | Bind audio cache reuse to remote object identity | RSK-007 | Pending | | 20 | Unify previous-source readiness and eliminate duplicate transfers | COR-010, EFF-001 | Pending | diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index 8869b4a..a10f4e0 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -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 { diff --git a/internal/app/operator_artifact_rendering.go b/internal/app/operator_artifact_rendering.go index 687d83b..ca167fe 100644 --- a/internal/app/operator_artifact_rendering.go +++ b/internal/app/operator_artifact_rendering.go @@ -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( diff --git a/internal/app/operator_status.go b/internal/app/operator_status.go index 704137b..c0cc5f0 100644 --- a/internal/app/operator_status.go +++ b/internal/app/operator_status.go @@ -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) diff --git a/internal/app/restore.go b/internal/app/restore.go index 8dc0f78..998951c 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -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, diff --git a/internal/app/restore_commit_test.go b/internal/app/restore_commit_test.go new file mode 100644 index 0000000..edf15de --- /dev/null +++ b/internal/app/restore_commit_test.go @@ -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[:]) +} diff --git a/internal/app/restore_discovery.go b/internal/app/restore_discovery.go index a45774d..d5b2abf 100644 --- a/internal/app/restore_discovery.go +++ b/internal/app/restore_discovery.go @@ -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 } diff --git a/internal/app/restore_discovery_test.go b/internal/app/restore_discovery_test.go index 305b38d..2ec5b5c 100644 --- a/internal/app/restore_discovery_test.go +++ b/internal/app/restore_discovery_test.go @@ -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{}, diff --git a/internal/app/restore_execute.go b/internal/app/restore_execute.go index 28ab871..a677e4d 100644 --- a/internal/app/restore_execute.go +++ b/internal/app/restore_execute.go @@ -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 = © + } + 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 diff --git a/internal/app/restore_plan.go b/internal/app/restore_plan.go index 00ab3ac..992c216 100644 --- a/internal/app/restore_plan.go +++ b/internal/app/restore_plan.go @@ -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 } diff --git a/internal/app/restore_test.go b/internal/app/restore_test.go index 7d95cfa..ecc371b 100644 --- a/internal/app/restore_test.go +++ b/internal/app/restore_test.go @@ -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 diff --git a/internal/previouscache/previouscache.go b/internal/previouscache/previouscache.go index 5251ce3..11321c1 100644 --- a/internal/previouscache/previouscache.go +++ b/internal/previouscache/previouscache.go @@ -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 {