From 8375ad83f31b565c56138e1a68d67482fafd59a3 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 21:01:28 +0000 Subject: [PATCH] Serialize restore recovery and rebase manifest paths --- docs/internal/command-restore.md | 14 +- docs/operations.md | 6 + docs/policy/architecture.md | 6 + docs/roadmap/implementation.md | 2 +- internal/app/restore.go | 56 +++++-- internal/app/restore_execute.go | 57 +++---- internal/app/restore_execution_test.go | 28 ++-- internal/app/restore_manifest_paths.go | 216 +++++++++++++++++++++++++ internal/app/restore_recovery.go | 63 ++++++++ internal/app/restore_recovery_test.go | 187 +++++++++++++++++++++ internal/app/restore_workflow_test.go | 2 +- internal/app/runner.go | 10 +- internal/app/runner_test.go | 4 +- internal/artifacts/paths.go | 8 + 14 files changed, 581 insertions(+), 78 deletions(-) create mode 100644 internal/app/restore_manifest_paths.go create mode 100644 internal/app/restore_recovery.go create mode 100644 internal/app/restore_recovery_test.go diff --git a/docs/internal/command-restore.md b/docs/internal/command-restore.md index c1e3d01..7b92ed2 100644 --- a/docs/internal/command-restore.md +++ b/docs/internal/command-restore.md @@ -8,8 +8,8 @@ reporting flow in `internal/app`. User invocation belongs in physical restore scope belong in [Operations](../operations.md#restore-workflow). -Restore is split into explicit phases so remote authority, local conflict -policy, and filesystem mutation can be tested independently. +Restore separates remote authority, local conflict policy, and filesystem +mutation so each remains testable independently. ## Discovery Contract @@ -40,6 +40,9 @@ Planner behavior: - force converts differing eligible regular files from conflicts to downloads; directories and other non-regular targets remain conflicts. +For a non-dry-run restore, planning/classification happens only after acquiring +the session lock. Runner manifest/reuse checks acquire that same lock first. + Previous-cache files are planned separately through `previouscache.BuildPlan` when configured previous-session requirements exist. @@ -54,6 +57,13 @@ Execution order and safety: - 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. +- a durable `.restore-incomplete.json` marker is written before installation. + It blocks runners until a restore retry completes all verified installs and + the local manifest replacement, at which point it is removed. +- restored manifest local references are rebased beneath the selected local + session root. Unsafe relative references and producer-machine absolute paths + outside the manifest's producer session root are rejected; producer-local + spool/cache and cleanup locations are not restored as authority. Audio restore path: diff --git a/docs/operations.md b/docs/operations.md index e811f5b..f963d34 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -283,6 +283,12 @@ Optional: Restore writes an execution report at `reports/restore-latest.json`. +If restore fails after beginning installation, it leaves a durable +`.restore-incomplete.json` marker in the session root. Pipeline runs will stop +until you rerun the same restore command and it completes. Restore intentionally +does not try to roll back files already installed; retrying the selected remote +snapshot is the recovery procedure. + ## Local State Layout Session root: diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index fd1d9da..111a021 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -119,6 +119,12 @@ install the validated session manifest after other restored durable files. The physical workflow and recovery procedures belong in [Operations](../operations.md). +Restore and runner transitions for one session use the same local lock. A +durable incomplete-restore marker blocks runner reuse after a partial restore; +safe retry, rather than rollback of arbitrary local effects, is the recovery +mechanism. Restored manifest-local references must be confined to the selected +local session root, never trusted as producer-machine absolute paths. + 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 diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 184033f..44426f5 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -33,7 +33,7 @@ All stages are pending when this plan is created. | 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 | Completed | -| 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Pending | +| 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Completed | | 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 | | 21 | Tighten configuration parsing, values, and expectations | COR-012–COR-015, TST-011, TST-014 | Pending | diff --git a/internal/app/restore.go b/internal/app/restore.go index 998951c..96c55f1 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -69,23 +69,23 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error if err != nil { return fmt.Errorf("restore: %w", err) } - plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{ - IncludeAudio: includeAudio, - Force: force, - DryRun: dryRun, - }) - if err != nil { - return fmt.Errorf("restore: %w", err) - } - report, err := newRestoreReport(current, plan, RestorePlanOptions{ - IncludeAudio: includeAudio, - Force: force, - DryRun: dryRun, - }) - if err != nil { - return fmt.Errorf("restore: %w", err) - } if dryRun { + plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{ + IncludeAudio: includeAudio, + Force: force, + DryRun: true, + }) + if err != nil { + return fmt.Errorf("restore: %w", err) + } + report, err := newRestoreReport(current, plan, RestorePlanOptions{ + IncludeAudio: includeAudio, + Force: force, + DryRun: true, + }) + if err != nil { + return fmt.Errorf("restore: %w", err) + } if err := writeRestoreDryRunSummary(out, report); err != nil { return fmt.Errorf("restore: write plan output: %w", err) } @@ -110,6 +110,24 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error } }() + // Classification and installation share the same transition lock as a + // runner. This prevents a runner from making reuse decisions against state + // that restore is about to replace. + plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{ + IncludeAudio: includeAudio, + Force: force, + }) + if err != nil { + return fmt.Errorf("restore: %w", err) + } + report, err := newRestoreReport(current, plan, RestorePlanOptions{ + IncludeAudio: includeAudio, + Force: force, + }) + if err != nil { + return fmt.Errorf("restore: %w", err) + } + if plan.ConflictCount > 0 { report.setFailed(fmt.Errorf("conflict: %d conflicting path(s)", plan.ConflictCount)) if _, reportErr := persistRestoreReport(artifactStore, cfg, report); reportErr != nil { @@ -123,6 +141,9 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error plan.ConflictCount, ) } + if err := writeRestoreMarker(artifactStore.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)); err != nil { + return fmt.Errorf("restore: mark incomplete restore: %w", err) + } result, err := executeRestorePlanFn(ctx, cfg, current, plan, report, objectStore) if err != nil { @@ -134,6 +155,9 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error } report.Execution.Downloaded = result.DownloadedCount report.setSucceeded() + if err := clearRestoreMarker(artifactStore.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)); err != nil { + return fmt.Errorf("restore: clear incomplete restore marker: %w", err) + } if _, err := persistRestoreReport(artifactStore, cfg, report); err != nil { return fmt.Errorf("restore: write report: %w", err) } diff --git a/internal/app/restore_execute.go b/internal/app/restore_execute.go index a677e4d..7436828 100644 --- a/internal/app/restore_execute.go +++ b/internal/app/restore_execute.go @@ -114,19 +114,28 @@ func executeRestoreDownloadAction( return err } - if action.LocalRelativePath == config.PathManifestFile { + if isRestoreManifest(action.LocalRelativePath) { file, err := temporary.Open() if err != nil { return fmt.Errorf("open restored manifest: %w", err) } - err = validateRestoredManifest(ctx, cfg, current, file) + destinationRoot := sessionRoot + requireCurrentIdentity := action.LocalRelativePath == config.PathManifestFile + if !requireCurrentIdentity { + destinationRoot = filepath.Dir(safeLocalPath) + } + restored, prepareErr := prepareRestoredManifest(ctx, cfg, current, file, destinationRoot, requireCurrentIdentity) closeErr := file.Close() - if err != nil { - return err + if prepareErr != nil { + return prepareErr } if closeErr != nil { return fmt.Errorf("close restored manifest: %w", closeErr) } + if err := (&manifest.LocalStore{}).Save(ctx, safeLocalPath, restored); err != nil { + return fmt.Errorf("install rebased manifest atomically: %w", err) + } + return nil } if err := temporary.Install(filepath.Base(safeLocalPath), fileops.WorkspaceFileMode); err != nil { @@ -136,6 +145,11 @@ func executeRestoreDownloadAction( return nil } +func isRestoreManifest(relativePath string) bool { + clean := filepath.ToSlash(filepath.Clean(strings.TrimSpace(relativePath))) + return clean == config.PathManifestFile || clean == config.PathPreviousDirSegment+"/"+config.PathManifestFile +} + 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 @@ -217,38 +231,3 @@ func executeRestoreAudioAction( } return nil } - -func validateRestoredManifest(ctx context.Context, cfg *config.Config, current *RemoteCurrentState, source io.Reader) error { - manifestStore := &manifest.LocalStore{} - m, err := manifestStore.LoadReader(ctx, source) - 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) - } - 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_execution_test.go b/internal/app/restore_execution_test.go index 63f8cff..be35400 100644 --- a/internal/app/restore_execution_test.go +++ b/internal/app/restore_execution_test.go @@ -76,11 +76,13 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi remoteManifest := manifest.New(cfg.Session.SessionID, time.Now().UTC()) remoteManifest.Campaign = cfg.Session.Campaign remoteManifest.RunID = "20260519T010203Z-a1b2c3d4" + producerRoot := "/prior/workspace/work/sample-campaign/2026-05-03" + remoteManifest.LocalWorkDir = filepath.Join(producerRoot, "runs", remoteManifest.RunID) remoteManifest.Stages["extract"] = &manifest.StageRecord{ Name: "extract", Status: manifest.StatusSucceeded, Outputs: []manifest.ArtifactRecord{{ Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID("encounters"), - LocalPath: "/prior/workspace/artifacts/notarius/extract-run-1/lanes/encounters.json", + LocalPath: filepath.Join(producerRoot, "artifacts", "encounters.json"), Contract: &artifactmodel.ContractMetadata{ MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1", }, @@ -112,6 +114,9 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi if lane.Contract == nil || lane.Contract.SchemaID != "encounters" || lane.ExternalProvenance == nil || lane.ExternalProvenance.RunID != "notarius-run-1" { t.Fatalf("restored extraction metadata = %#v", lane) } + if want := filepath.Join(sessionRoot, "artifacts", "encounters.json"); lane.LocalPath != want { + t.Fatalf("rebased extraction path = %q, want %q", lane.LocalPath, want) + } } func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) { @@ -200,7 +205,7 @@ func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) { if err != nil { t.Fatalf("read restored previous manifest: %v", err) } - if !strings.Contains(string(previousManifestBytes), `"session_id":"2026-04-26"`) { + if !strings.Contains(string(previousManifestBytes), `"session_id": "2026-04-26"`) { t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes)) } mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n") @@ -293,6 +298,9 @@ func TestExecuteRestoreForceOverwritesDifferingFile(t *testing.T) { if !report.Force { t.Fatalf("report force = %v, want true", report.Force) } + if _, err := os.Stat(artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)); !os.IsNotExist(err) { + t.Fatalf("restore marker should be cleared after success; stat err=%v", err) + } } func TestExecuteRestoreForceOverwritesDifferingPreviousCacheFile(t *testing.T) { @@ -335,14 +343,11 @@ func TestExecuteRestoreLockConflictFailsAndWritesNothing(t *testing.T) { restoreWithStoreAndRealPhases(t, fake) - var stdout bytes.Buffer - var stderr bytes.Buffer - code := Execute([]string{"session", "restore", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--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()) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err = Restore(ctx, []string{"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &bytes.Buffer{}) + if err == nil || !strings.Contains(err.Error(), "acquire session lock") { + t.Fatalf("Restore() error = %v, want lock failure", err) } sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) @@ -406,6 +411,9 @@ func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing. if string(afterData) != string(existingData) { t.Fatalf("local manifest changed after failed restore; before=%q after=%q", string(existingData), string(afterData)) } + if _, err := os.Stat(artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)); err != nil { + t.Fatalf("incomplete restore marker should remain after failed forced restore: %v", err) + } } func TestExecuteRestorePlanPathMismatchFails(t *testing.T) { diff --git a/internal/app/restore_manifest_paths.go b/internal/app/restore_manifest_paths.go new file mode 100644 index 0000000..b05d28f --- /dev/null +++ b/internal/app/restore_manifest_paths.go @@ -0,0 +1,216 @@ +package app + +import ( + "context" + "fmt" + "io" + "path" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func prepareRestoredManifest( + ctx context.Context, + cfg *config.Config, + current *RemoteCurrentState, + source io.Reader, + destinationRoot string, + requireCurrentIdentity bool, +) (*manifest.Manifest, error) { + manifestStore := &manifest.LocalStore{} + m, err := manifestStore.LoadReader(ctx, source) + if err != nil { + return nil, fmt.Errorf("validate manifest decode: %w", err) + } + if requireCurrentIdentity { + if err := validateRestoredManifestIdentity(cfg, current, m); err != nil { + return nil, err + } + } + if err := rebaseRestoredManifestPaths(cfg, m, destinationRoot); err != nil { + return nil, err + } + return m, nil +} + +func validateRestoredManifestIdentity(cfg *config.Config, current *RemoteCurrentState, m *manifest.Manifest) error { + if cfg == nil || cfg.Session == nil { + return fmt.Errorf("resolved session config is required") + } + 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) + } + 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 +} + +func rebaseRestoredManifestPaths(cfg *config.Config, m *manifest.Manifest, destinationRoot string) error { + if cfg == nil || cfg.Pipeline == nil || m == nil { + return fmt.Errorf("resolved config and manifest are required") + } + destinationRoot = filepath.Clean(strings.TrimSpace(destinationRoot)) + if destinationRoot == "" || destinationRoot == "." { + return fmt.Errorf("restored manifest destination root is required") + } + producerRoot, hasProducerRoot := restoredManifestSessionRoot(m) + rebase := func(field, value string) (string, error) { + return rebaseRestoredLocalReference(field, value, producerRoot, hasProducerRoot, destinationRoot) + } + + for i := range m.Inputs { + value, err := rebase("inputs.path", m.Inputs[i].Path) + if err != nil { + return err + } + m.Inputs[i].Path = value + // Spool and cache locations are host-local implementation details. They + // are deliberately not authoritative after a restore. + m.Inputs[i].SpoolPath = "" + m.Inputs[i].CachePath = "" + } + for i := range m.Artifacts { + value, err := rebase("artifacts.local_path", m.Artifacts[i].LocalPath) + if err != nil { + return err + } + m.Artifacts[i].LocalPath = value + } + for stageName, record := range m.Stages { + if record == nil { + continue + } + for i := range record.Outputs { + value, err := rebase("stages."+stageName+".outputs.local_path", record.Outputs[i].LocalPath) + if err != nil { + return err + } + record.Outputs[i].LocalPath = value + } + // Logs and generated configuration files are invocation-local diagnostics, + // not restored artifacts. Dropping them prevents a producer-machine path + // from becoming a usable local reference. + record.Logs = nil + record.GeneratedConfigs = nil + } + + if runID := strings.TrimSpace(m.RunID); runID != "" { + m.LocalWorkDir = filepath.Join(destinationRoot, config.PathRunsDirSegment, runID) + } else { + m.LocalWorkDir = destinationRoot + } + m.LocalSpoolDir = "" + // A post-publish cleanup record is authority to delete producer-local + // directories. It must never cross a restore boundary. + m.PostPublishCleanup = nil + return nil +} + +func restoredManifestSessionRoot(m *manifest.Manifest) (string, bool) { + if m == nil { + return "", false + } + runRoot := portableCleanPath(m.LocalWorkDir) + runID := strings.TrimSpace(m.RunID) + if !portableAbsolutePath(runRoot) || runID == "" || path.Base(runRoot) != runID { + return "", false + } + runsDir := path.Dir(runRoot) + if path.Base(runsDir) != config.PathRunsDirSegment { + return "", false + } + return path.Dir(runsDir), true +} + +func rebaseRestoredLocalReference(field, value, producerRoot string, hasProducerRoot bool, destinationRoot string) (string, error) { + value = strings.TrimSpace(value) + if value == "" { + return "", nil + } + if portableDrivePath(value) && !portableAbsolutePath(value) { + return "", fmt.Errorf("%s has an unsafe relative path %q", field, value) + } + if portableAbsolutePath(value) { + if !hasProducerRoot { + return "", fmt.Errorf("%s has an absolute path without a producer session root", field) + } + relative, ok := portableRelativeWithinRoot(producerRoot, value) + if !ok || relative == "" { + return "", fmt.Errorf("%s absolute path is outside the producer session root", field) + } + resolved, err := joinWithinSessionRoot(destinationRoot, relative) + if err != nil { + return "", fmt.Errorf("%s: %w", field, err) + } + return resolved, nil + } + resolved, err := joinWithinSessionRoot(destinationRoot, strings.ReplaceAll(value, "\\", "/")) + if err != nil { + return "", fmt.Errorf("%s has an unsafe relative path: %w", field, err) + } + return resolved, nil +} + +func portableRelativeWithinRoot(root, candidate string) (string, bool) { + root = strings.TrimSuffix(portableCleanPath(root), "/") + candidate = portableCleanPath(candidate) + if root == "" || candidate == "" { + return "", false + } + compareRoot, compareCandidate := root, candidate + if portableDrivePath(root) || portableDrivePath(candidate) || strings.HasPrefix(root, "//") || strings.HasPrefix(candidate, "//") { + compareRoot = strings.ToLower(compareRoot) + compareCandidate = strings.ToLower(compareCandidate) + } + if compareCandidate == compareRoot { + return "", true + } + if !strings.HasPrefix(compareCandidate, compareRoot+"/") { + return "", false + } + return strings.TrimPrefix(candidate, root+"/"), true +} + +func portableCleanPath(value string) string { + value = strings.ReplaceAll(strings.TrimSpace(value), "\\", "/") + if value == "" { + return "" + } + return path.Clean(value) +} + +func portableAbsolutePath(value string) bool { + value = strings.TrimSpace(value) + return strings.HasPrefix(value, "/") || strings.HasPrefix(value, "\\") || (len(value) >= 3 && isASCIIAlpha(value[0]) && value[1] == ':' && (value[2] == '/' || value[2] == '\\')) +} + +func portableDrivePath(value string) bool { + value = strings.TrimSpace(value) + return len(value) >= 2 && isASCIIAlpha(value[0]) && value[1] == ':' +} + +func isASCIIAlpha(value byte) bool { + return (value >= 'a' && value <= 'z') || (value >= 'A' && value <= 'Z') +} diff --git a/internal/app/restore_recovery.go b/internal/app/restore_recovery.go new file mode 100644 index 0000000..4e48f9f --- /dev/null +++ b/internal/app/restore_recovery.go @@ -0,0 +1,63 @@ +package app + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" +) + +type restoreMarker struct { + StartedAt time.Time `json:"started_at"` +} + +func writeRestoreMarker(paths artifacts.SessionPaths) error { + marker := restoreMarker{StartedAt: nowUTC()} + data, err := json.Marshal(marker) + if err != nil { + return fmt.Errorf("encode recovery marker: %w", err) + } + if err := fileops.WriteFileAtomic(paths.RestoreMarkerPath, data, fileops.WorkspaceFileMode); err != nil { + return fmt.Errorf("write recovery marker: %w", err) + } + return nil +} + +func clearRestoreMarker(paths artifacts.SessionPaths) error { + if strings.TrimSpace(paths.RestoreMarkerPath) == "" { + return fmt.Errorf("recovery marker path is required") + } + if err := os.Remove(paths.RestoreMarkerPath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove recovery marker: %w", err) + } + return nil +} + +func requireCompleteRestore(paths artifacts.SessionPaths) error { + if strings.TrimSpace(paths.RestoreMarkerPath) == "" { + return fmt.Errorf("restore recovery marker path is required") + } + info, err := os.Lstat(paths.RestoreMarkerPath) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return fmt.Errorf("check restore recovery marker: %w", err) + } + if !info.Mode().IsRegular() { + return fmt.Errorf("restore recovery marker is not a regular file; rerun session restore") + } + data, err := os.ReadFile(paths.RestoreMarkerPath) + if err != nil { + return fmt.Errorf("read restore recovery marker: %w", err) + } + var marker restoreMarker + if err := json.Unmarshal(data, &marker); err != nil || marker.StartedAt.IsZero() { + return fmt.Errorf("restore recovery marker is invalid; rerun session restore") + } + return fmt.Errorf("restore is incomplete; rerun session restore before running pipeline stages") +} diff --git a/internal/app/restore_recovery_test.go b/internal/app/restore_recovery_test.go new file mode 100644 index 0000000..6d8b1d7 --- /dev/null +++ b/internal/app/restore_recovery_test.go @@ -0,0 +1,187 @@ +package app + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "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 TestRestoreHoldsTransitionLockBeforePlanning(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + cfg, _, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + + origStoreFn := newObjectStoreFromConfigFn + origDiscoverFn := discoverRemoteCurrentStateFn + origPlanFn := buildRestorePlanFn + t.Cleanup(func() { + newObjectStoreFromConfigFn = origStoreFn + discoverRemoteCurrentStateFn = origDiscoverFn + buildRestorePlanFn = origPlanFn + }) + + current, err := discoverRemoteCurrentState(context.Background(), cfg, fake) + if err != nil { + t.Fatalf("discoverRemoteCurrentState() error = %v", err) + } + planning := make(chan struct{}) + releasePlanning := make(chan struct{}) + newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) { return fake, nil } + discoverRemoteCurrentStateFn = func(context.Context, *config.Config, storage.ObjectStore) (*RemoteCurrentState, error) { + return current, nil + } + buildRestorePlanFn = func(context.Context, *config.Config, *RemoteCurrentState, storage.ObjectStore, RestorePlanOptions) (*RestorePlan, error) { + close(planning) + <-releasePlanning + return &RestorePlan{}, nil + } + + restoreDone := make(chan error, 1) + go func() { + restoreDone <- Restore(context.Background(), []string{ + cfg.Session.SessionID, + "--config", pipelinePath, + "--campaign-file", campaignPath, + "--session", sessionPath, + }, &bytes.Buffer{}) + }() + <-planning + + runnerCtx, cancelRunner := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancelRunner() + if _, err := executeStages(runnerCtx, cfg, nil, RunOptions{}); err == nil || !strings.Contains(err.Error(), "acquire session lock") { + t.Fatalf("runner while restore plans error = %v, want session-lock failure", err) + } + + close(releasePlanning) + if err := <-restoreDone; err != nil { + t.Fatalf("Restore() error = %v", err) + } +} + +func TestExecuteStagesRefusesIncompleteRestore(t *testing.T) { + cfg := testConfig(t) + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + if err := writeRestoreMarker(paths); err != nil { + t.Fatalf("writeRestoreMarker() error = %v", err) + } + + _, err := executeStages(context.Background(), cfg, nil, RunOptions{}) + if err == nil || !strings.Contains(err.Error(), "restore is incomplete") { + t.Fatalf("executeStages() error = %v, want incomplete restore failure", err) + } +} + +func TestForcedRestoreRetryClearsIncompleteMarker(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + artifactKey := sessionPrefix + "artifacts/session_recap.md" + transcriptKey := sessionPrefix + "transcripts/full.json" + seedRestoreObject(fake, artifactKey, []byte("remote recap\n")) + seedRestoreObject(fake, transcriptKey, []byte("remote transcript\n")) + + sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) + mustWriteTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "local recap\n") + mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local transcript\n") + fake.DownloadHook = func(call storage.FakeDownloadCall) error { + if call.Key == transcriptKey { + return context.Canceled + } + return nil + } + restoreWithStoreAndRealPhases(t, fake) + + var stdout bytes.Buffer + var stderr bytes.Buffer + args := []string{"session", "restore", cfg.Session.SessionID, "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"} + if code := Execute(args, &stdout, &stderr); code == 0 { + t.Fatal("forced restore exit code = 0, want partial-install failure") + } + markerPath := artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) + if _, err := os.Stat(markerPath); err != nil { + t.Fatalf("incomplete restore marker missing after partial forced restore: %v", err) + } + mustReadEquals(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "remote recap\n") + + fake.DownloadHook = nil + stdout.Reset() + stderr.Reset() + if code := Execute(args, &stdout, &stderr); code != 0 { + t.Fatalf("restore retry exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if _, err := os.Stat(markerPath); !os.IsNotExist(err) { + t.Fatalf("incomplete restore marker should be cleared after retry; stat err=%v", err) + } + mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "remote transcript\n") +} + +func TestRebaseRestoredManifestPathsRejectsUnsafeReferencesAndRebasesPortablePaths(t *testing.T) { + for _, test := range []struct { + name string + workDir string + output string + wantPath string + wantError string + }{ + { + name: "unix absolute path within producer root", + workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4", + output: "/producer/work/sample-campaign/2026-05-03/artifacts/recap.md", + wantPath: "artifacts/recap.md", + }, + { + name: "windows absolute path within producer root", + workDir: `C:\producer\work\sample-campaign\2026-05-03\runs\20260519T010203Z-a1b2c3d4`, + output: `C:\producer\work\sample-campaign\2026-05-03\artifacts\recap.md`, + wantPath: "artifacts/recap.md", + }, + { + name: "absolute path outside producer root", + workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4", + output: "/another-machine/recap.md", + wantError: "outside the producer session root", + }, + { + name: "relative traversal", + workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4", + output: "../recap.md", + wantError: "unsafe relative path", + }, + } { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + m := manifest.New(cfg.Session.SessionID, nowUTC()) + m.Campaign = cfg.Session.Campaign + m.RunID = "20260519T010203Z-a1b2c3d4" + m.LocalWorkDir = test.workDir + m.Stages["analyze"] = &manifest.StageRecord{Outputs: []manifest.ArtifactRecord{{Kind: "recap", LocalPath: test.output}}} + destination := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + + err := rebaseRestoredManifestPaths(cfg, m, destination) + if test.wantError != "" { + if err == nil || !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("rebaseRestoredManifestPaths() error = %v, want %q", err, test.wantError) + } + return + } + if err != nil { + t.Fatalf("rebaseRestoredManifestPaths() error = %v", err) + } + if got, want := m.Stages["analyze"].Outputs[0].LocalPath, filepath.Join(destination, test.wantPath); got != want { + t.Fatalf("rebased local path = %q, want %q", got, want) + } + }) + } +} diff --git a/internal/app/restore_workflow_test.go b/internal/app/restore_workflow_test.go index b842022..f9cbe5d 100644 --- a/internal/app/restore_workflow_test.go +++ b/internal/app/restore_workflow_test.go @@ -211,7 +211,7 @@ previous_session_id: 2026-04-26 if err != nil { t.Fatalf("read restored previous manifest: %v", err) } - if !strings.Contains(string(previousManifestBytes), `"session_id":"2026-04-26"`) { + if !strings.Contains(string(previousManifestBytes), `"session_id": "2026-04-26"`) { t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes)) } mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n") diff --git a/internal/app/runner.go b/internal/app/runner.go index 8e5f4ea..04e780e 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -67,13 +67,6 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage identity.Campaign, identity.SessionID, ) - if existing, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath); err != nil { - return nil, err - } else if present { - if err := identity.validateSessionManifest(existing); err != nil { - return nil, err - } - } if env.ArtifactStore == nil { env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) } @@ -104,6 +97,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if paths.ManifestPath != manifestPath { return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath) } + if err := requireCompleteRestore(paths); err != nil { + return nil, err + } m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath) if err != nil { return nil, err diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 52bfdd0..770c307 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -1076,8 +1076,8 @@ func TestExecuteStagesRejectsPersistedManifestIdentityMismatch(t *testing.T) { t.Fatal("persisted manifest changed after identity rejection") } paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) - if _, err := os.Stat(paths.LockPath); !os.IsNotExist(err) { - t.Fatalf("lock path stat error = %v, want no lock side effect", err) + if _, err := os.Stat(paths.LockPath); err != nil { + t.Fatalf("lock path stat error = %v, want transition lock", err) } }) } diff --git a/internal/artifacts/paths.go b/internal/artifacts/paths.go index 0da16bc..435c8a8 100644 --- a/internal/artifacts/paths.go +++ b/internal/artifacts/paths.go @@ -32,6 +32,7 @@ type SessionPaths struct { PreviousArtifactsDir string ManifestPath string LockPath string + RestoreMarkerPath string } // SessionWorkDirForCampaign returns the canonical campaign-aware work directory for one session. @@ -44,6 +45,12 @@ func SessionManifestPathForCampaign(rootDir, campaign, sessionID string) string return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathManifestFile) } +// SessionRestoreMarkerPathForCampaign returns the durable recovery marker for +// a restore that has started changing one session's local state. +func SessionRestoreMarkerPathForCampaign(rootDir, campaign, sessionID string) string { + return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), ".restore-incomplete.json") +} + // SessionRunsDirForCampaign returns the canonical runs directory for one session. func SessionRunsDirForCampaign(rootDir, campaign, sessionID string) string { return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathRunsDirSegment) @@ -264,5 +271,6 @@ func buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root string) PreviousArtifactsDir: filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment), ManifestPath: filepath.Join(root, config.PathManifestFile), LockPath: filepath.Join(root, config.PathLockFile), + RestoreMarkerPath: filepath.Join(root, ".restore-incomplete.json"), } }