diff --git a/docs/cli.md b/docs/cli.md index 9c1fac2..2f04251 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -390,7 +390,10 @@ Common failure cases: ### `restore` Purpose: -- Restore durable session state (`manifest.json`, `transcripts/**`, `artifacts/**`, `previous/**`, and optional `audio/**`) from the committed remote archive current state. +- Restore durable session state from the committed remote archive current state. +- Default restore installs `manifest.json`, `transcripts/**`, and `artifacts/**` from the current session archive. +- When configured previous-session inputs require it, restore reconstructs `previous/**` from the previous session's committed current archive, matching what `prepare` would hydrate. +- `audio/**` is restored only with `--include-audio`. Syntax: @@ -417,9 +420,12 @@ Common failure cases: - remote `current/run_id.txt` missing/empty. - remote `current/manifest.json` missing or invalid. - remote manifest session/campaign mismatch. +- required previous-session current state or artifact missing. - local conflicts without `--force`. - session lock conflict. +Dry-run output may include planned previous-cache downloads. Existing differing files under `previous/**` follow the normal restore conflict policy and require `--force` to overwrite. + When `--include-audio` is set, S3 audio files are restored through the shared audio cache. Cache hits avoid re-downloading large audio objects. ### `clean` diff --git a/docs/internal/workspace.md b/docs/internal/workspace.md index 10154de..512f37e 100644 --- a/docs/internal/workspace.md +++ b/docs/internal/workspace.md @@ -18,7 +18,7 @@ Outputs: ## Boundaries Owns: - Session-level path layout (`inputs/`, `audio/`, `transcripts/`, `artifacts/`, `reports/`, `logs/`, `config/`, `current/`, `runs/`, `previous/`) - - `previous/manifest.json` and `previous/artifacts/**` are reserved for prepared previous-session state + - `previous/manifest.json` and `previous/artifacts/**` are reserved for previous-session cache state materialized by `prepare` or `restore` - Run-local stage sandbox layout under `runs/{run_id}/{stage}/` - Session lock acquisition/release (`.lock`) @@ -46,6 +46,7 @@ None directly in this subsystem. Stages may use object storage adapters and then - During each run, stage outputs are often written run-local first (`runs/{run_id}/{stage}/outputs/...`) and promoted to canonical session paths after stage success. - `manifest.Artifacts` entries record `ProducerRunID` for durable outputs. - For S3 audio sessions, `prepare` records work/cache paths, S3 provenance, and spool path when the invocation downloaded the object. +- `previous/**` is reconstructed from configured previous-session requirements; restore uses the previous session's committed current archive rather than treating current-session archived `previous/**` as authoritative. - Durable cache state under `pipeline.cache.root` is not workspace state and is preserved by default by `narratio clean`. - `narratio clean --session-id ` removes the session work root and session spool root. - `narratio clean --all` removes all local session work under `workspace.root/work` and spool children under `spool.root`. diff --git a/docs/operations.md b/docs/operations.md index 4373534..960732f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -62,9 +62,11 @@ narratio analyze --session-id 2026-04-04 Restore source-of-truth: - remote commit marker: `current/run_id.txt` - remote current manifest: `current/manifest.json` +- configured previous-session requirements are reconstructed from the previous session's remote `current/` state, not from archived `previous/**` objects in the current session. Restore default scope: -- includes `manifest.json`, `transcripts/**`, `artifacts/**`, `previous/**` +- includes `manifest.json`, `transcripts/**`, and `artifacts/**` from the current session archive. +- includes `previous/**` only when configured previous-session artifact inputs require it; restore hydrates those files the same way `prepare` would. - includes `audio/**` only with `--include-audio` - excludes `runs/**`, `logs/**`, `reports/**`, `config/**`, `inputs/**`, and `current/**` (except remote `current/manifest.json` as source) @@ -130,9 +132,10 @@ Configured artifact source reuse: Canonical previous-session input behavior: - canonical sources use `narratio.previous_session.artifact.`. -- these inputs are hydrated by `prepare`, not `analyze`. +- these inputs are hydrated by `prepare` and by `restore`; `analyze` expects the local previous cache to already exist. - if analyze fails due to missing canonical previous cache, rerun: - `narratio run-stage --session-id --force prepare` + - or `narratio restore --session-id ` when remote archive current state is authoritative. ## Remote archive layout and publish contract diff --git a/internal/app/restore_execution_test.go b/internal/app/restore_execution_test.go index 72b94da..1fbac70 100644 --- a/internal/app/restore_execution_test.go +++ b/internal/app/restore_execution_test.go @@ -128,11 +128,11 @@ func TestExecuteRestoreIncludeAudioUsesCacheAfterWorkspaceDeletion(t *testing.T) func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + appendRestoreWorkflowPreviousInputConfig(t, pipelinePath, sessionPath) fake := &storage.FakeBackend{} - cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) - seedRestoreObject(fake, sessionPrefix+"previous/manifest.json", []byte(`{"session_id":"2026-04-26"}`)) - seedRestoreObject(fake, sessionPrefix+"previous/artifacts/session_recap.md", []byte("# previous recap\n")) + cfg, _, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + seedRestorePreviousCurrent(t, fake, cfg, "# previous recap\n") restoreWithStoreAndRealPhases(t, fake) @@ -144,7 +144,13 @@ func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) { } sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) - mustReadEquals(t, filepath.Join(sessionRoot, "previous", "manifest.json"), `{"session_id":"2026-04-26"}`) + previousManifestBytes, err := os.ReadFile(filepath.Join(sessionRoot, "previous", "manifest.json")) + if err != nil { + t.Fatalf("read restored previous manifest: %v", err) + } + 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") report := mustReadRestoreReport(t, filepath.Join(sessionRoot, "reports", "restore-latest.json")) if report.Execution.Downloaded != 3 { @@ -152,6 +158,33 @@ func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) { } } +func TestExecuteRestoreDryRunReportsPreviousCacheWithoutWriting(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + appendRestoreWorkflowPreviousInputConfig(t, pipelinePath, sessionPath) + + fake := &storage.FakeBackend{} + cfg, _, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + seedRestorePreviousCurrent(t, fake, cfg, "# previous recap\n") + + restoreWithStoreAndRealPhases(t, fake) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"restore", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--dry-run"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if !strings.Contains(stdout.String(), "previous/artifacts/session_recap.md") { + t.Fatalf("stdout = %q, want planned previous-cache artifact", stdout.String()) + } + + sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) + if _, err := os.Stat(filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md")); !os.IsNotExist(err) { + t.Fatalf("previous artifact should not be written during dry-run; stat err=%v", err) + } +} + func TestExecuteRestoreConflictWithoutForceDoesNotOverwrite(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) @@ -213,10 +246,11 @@ func TestExecuteRestoreForceOverwritesDifferingFile(t *testing.T) { func TestExecuteRestoreForceOverwritesDifferingPreviousCacheFile(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + appendRestoreWorkflowPreviousInputConfig(t, pipelinePath, sessionPath) fake := &storage.FakeBackend{} - cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) - seedRestoreObject(fake, sessionPrefix+"previous/artifacts/session_recap.md", []byte("# remote previous recap\n")) + cfg, _, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + seedRestorePreviousCurrent(t, fake, cfg, "# remote previous recap\n") sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) mustWriteTestFile(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# local previous recap\n") @@ -401,6 +435,50 @@ func seedRestoreCommittedState(t *testing.T, fake *storage.FakeBackend, pipeline return cfg, sessionPrefix, manifestKey, runIDKey } +func appendRestoreWorkflowPreviousInputConfig(t *testing.T, pipelinePath, sessionPath string) { + t.Helper() + appendRestoreWorkflowScriptoriumConfig(t, pipelinePath, ` +scriptorium: + binary: scriptorium + artifacts: + session_recap: + enabled: true + prompt_id: dnd.session_recap + output_path: artifacts/session_recap.md + inputs: + previous_recap: + source: narratio.previous_session.artifact.session_recap + required: true +`) + appendRestoreWorkflowScriptoriumConfig(t, sessionPath, ` +previous_session_id: 2026-04-26 +`) +} + +func seedRestorePreviousCurrent(t *testing.T, fake *storage.FakeBackend, cfg *config.Config, artifactBody string) { + t.Helper() + seedRestorePreviousCurrentManifestOnly(t, fake, cfg) + previousPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) + seedRestoreObject(fake, previousPrefix+"artifacts/session_recap.md", []byte(artifactBody)) +} + +func seedRestorePreviousCurrentManifestOnly(t *testing.T, fake *storage.FakeBackend, cfg *config.Config) { + t.Helper() + previousPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) + manifestKey, runIDKey := artifacts.ResolveArchiveCurrentStateKeys(previousPrefix) + previousRunID := "20260426T010203Z-a1b2c3d4" + seedRestoreObject(fake, runIDKey, []byte(previousRunID+"\n")) + + m := manifest.New(cfg.Session.PreviousSessionID, nowUTC()) + m.Campaign = cfg.Session.Campaign + m.RunID = previousRunID + data, err := json.Marshal(m) + if err != nil { + t.Fatalf("marshal previous restore manifest: %v", err) + } + seedRestoreObject(fake, manifestKey, append(data, '\n')) +} + func mustReadEquals(t *testing.T, path, want string) { t.Helper() data, err := os.ReadFile(path) diff --git a/internal/app/restore_plan.go b/internal/app/restore_plan.go index 41bb7d2..a0b7d2b 100644 --- a/internal/app/restore_plan.go +++ b/internal/app/restore_plan.go @@ -13,6 +13,7 @@ import ( "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/previouscache" ) // RestoreActionKind identifies one restore planner action. @@ -114,6 +115,12 @@ func buildRestorePlan(ctx context.Context, cfg *config.Config, current *RemoteCu actions = append(actions, action) } + previousActions, err := buildPreviousCacheRestoreActions(ctx, cfg, sessionPaths, store, opts.Force) + if err != nil { + return nil, err + } + actions = append(actions, previousActions...) + sort.Slice(actions, func(i, j int) bool { if actions[i].LocalRelativePath == actions[j].LocalRelativePath { return actions[i].RemoteKey < actions[j].RemoteKey @@ -195,7 +202,7 @@ func restoreLocalRelativePathForKey(sessionPrefix, currentManifestKey, key strin return cleanRel, true, nil } if cleanRel == config.PathPreviousDirSegment || strings.HasPrefix(cleanRel, config.PathPreviousDirSegment+"/") { - return cleanRel, true, nil + return "", false, nil } if includeAudio && (cleanRel == config.PathAudioDirSegment || strings.HasPrefix(cleanRel, config.PathAudioDirSegment+"/")) { return cleanRel, true, nil @@ -223,6 +230,35 @@ func joinWithinSessionRoot(sessionRoot, relative string) (string, error) { return abs, nil } +func buildPreviousCacheRestoreActions( + ctx context.Context, + cfg *config.Config, + sessionPaths artifacts.SessionPaths, + store storage.ObjectStore, + force bool, +) ([]RestoreAction, error) { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Scriptorium == nil { + return nil, nil + } + requirements := artifacts.CollectPreviousArtifactRequirements(cfg.Pipeline.Scriptorium.Artifacts) + if len(requirements) == 0 { + return nil, nil + } + plan, err := previouscache.BuildPlan(ctx, cfg, sessionPaths, requirements, store) + if err != nil { + return nil, fmt.Errorf("plan previous-session cache restore: %w", err) + } + 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) + if err != nil { + return nil, fmt.Errorf("classify previous-session cache object %q: %w", record.RemoteKey, err) + } + actions = append(actions, action) + } + return actions, nil +} + func classifyRestoreAction( ctx context.Context, store storage.ObjectStore, diff --git a/internal/app/restore_plan_test.go b/internal/app/restore_plan_test.go index 029513a..1737c86 100644 --- a/internal/app/restore_plan_test.go +++ b/internal/app/restore_plan_test.go @@ -86,6 +86,27 @@ func TestRestorePlanExistingAudioUsesSizeWithoutRemoteChecksumDownload(t *testin } func TestRestorePlanIncludesPreviousCacheByDefault(t *testing.T) { + cfg := restorePlanConfig(t) + configureRestorePlanPreviousRequirement(cfg, true) + current := restorePlanCurrentState(t, cfg) + store := &storage.FakeBackend{} + + seedRestoreObject(store, current.CurrentManifestKey, []byte(`{"session_id":"2026-05-03"}`)) + seedRestorePreviousCurrent(t, store, cfg, "# previous recap\n") + + plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{}) + if err != nil { + t.Fatalf("buildRestorePlan() error = %v", err) + } + + got := actionRelPaths(plan.Actions) + want := []string{"manifest.json", "previous/artifacts/session_recap.md", "previous/manifest.json"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("action local paths = %#v, want %#v", got, want) + } +} + +func TestRestorePlanIgnoresCurrentSessionArchivedPreviousCache(t *testing.T) { cfg := restorePlanConfig(t) current := restorePlanCurrentState(t, cfg) store := &storage.FakeBackend{} @@ -100,12 +121,76 @@ func TestRestorePlanIncludesPreviousCacheByDefault(t *testing.T) { } got := actionRelPaths(plan.Actions) - want := []string{"manifest.json", "previous/artifacts/session_recap.md", "previous/manifest.json"} + want := []string{"manifest.json"} if !reflect.DeepEqual(got, want) { t.Fatalf("action local paths = %#v, want %#v", got, want) } } +func TestRestorePlanMissingOptionalPreviousCacheSkipsArtifact(t *testing.T) { + cfg := restorePlanConfig(t) + configureRestorePlanPreviousRequirement(cfg, false) + current := restorePlanCurrentState(t, cfg) + store := &storage.FakeBackend{} + + seedRestoreObject(store, current.CurrentManifestKey, []byte(`{"session_id":"2026-05-03"}`)) + seedRestorePreviousCurrentManifestOnly(t, store, cfg) + + plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{}) + if err != nil { + t.Fatalf("buildRestorePlan() error = %v", err) + } + + got := actionRelPaths(plan.Actions) + want := []string{"manifest.json", "previous/manifest.json"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("action local paths = %#v, want %#v", got, want) + } +} + +func TestRestorePlanMissingRequiredPreviousCacheFails(t *testing.T) { + cfg := restorePlanConfig(t) + configureRestorePlanPreviousRequirement(cfg, true) + current := restorePlanCurrentState(t, cfg) + store := &storage.FakeBackend{} + + seedRestoreObject(store, current.CurrentManifestKey, []byte(`{"session_id":"2026-05-03"}`)) + seedRestorePreviousCurrentManifestOnly(t, store, cfg) + + _, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{}) + if err == nil || !strings.Contains(err.Error(), "required previous-session artifact") { + t.Fatalf("buildRestorePlan() error = %v, want required previous artifact failure", err) + } +} + +func TestRestorePlanPreviousCacheConflictRequiresForce(t *testing.T) { + cfg := restorePlanConfig(t) + configureRestorePlanPreviousRequirement(cfg, true) + current := restorePlanCurrentState(t, cfg) + store := &storage.FakeBackend{} + + seedRestoreObject(store, current.CurrentManifestKey, []byte(`{"session_id":"2026-05-03"}`)) + seedRestorePreviousCurrent(t, store, cfg, "# remote previous recap\n") + sessionRoot := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + mustWriteTestFile(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# local previous recap\n") + + plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{}) + if err != nil { + t.Fatalf("buildRestorePlan() error = %v", err) + } + if plan.ConflictCount != 1 { + t.Fatalf("ConflictCount = %d, want 1", plan.ConflictCount) + } + + plan, err = buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{Force: true}) + if err != nil { + t.Fatalf("buildRestorePlan(force) error = %v", err) + } + if plan.ConflictCount != 0 { + t.Fatalf("force ConflictCount = %d, want 0", plan.ConflictCount) + } +} + func TestRestorePlanClassifiesSameAndConflict(t *testing.T) { cfg := restorePlanConfig(t) current := restorePlanCurrentState(t, cfg) @@ -207,6 +292,10 @@ func restorePlanConfig(t *testing.T) *config.Config { return &config.Config{ Pipeline: &config.PipelineConfig{ Workspace: config.WorkspaceConfig{Root: workspaceRoot}, + Storage: config.StorageConfig{S3: &config.StorageS3Config{ + Bucket: "test-bucket", + RootPrefix: "dnd", + }}, }, Session: &config.SessionConfig{ SessionID: "2026-05-03", @@ -215,6 +304,24 @@ func restorePlanConfig(t *testing.T) *config.Config { } } +func configureRestorePlanPreviousRequirement(cfg *config.Config, required bool) { + cfg.Session.PreviousSessionID = "2026-04-26" + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "session_recap": { + Enabled: true, + OutputPath: "artifacts/session_recap.md", + Inputs: map[string]config.ScriptoriumInputConfig{ + "previous_recap": { + Source: "narratio.previous_session.artifact.session_recap", + Required: required, + }, + }, + }, + }, + } +} + func restorePlanCurrentState(t *testing.T, cfg *config.Config) *RemoteCurrentState { t.Helper() sessionPrefix := artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.SessionID) diff --git a/internal/app/restore_workflow_test.go b/internal/app/restore_workflow_test.go index 500999e..6749b8e 100644 --- a/internal/app/restore_workflow_test.go +++ b/internal/app/restore_workflow_test.go @@ -175,6 +175,9 @@ scriptorium: previous_recap: source: narratio.previous_session.artifact.session_recap required: true +`) + appendRestoreWorkflowScriptoriumConfig(t, sessionPath, ` +previous_session_id: 2026-04-26 `) fakeStore := &storage.FakeBackend{} @@ -182,8 +185,7 @@ scriptorium: seedRestoreObject(fakeStore, runIDKey, []byte("20260519T010203Z-a1b2c3d4\n")) seedRestoreObject(fakeStore, manifestKey, restoreWorkflowManifestJSON(t, cfg.Session.SessionID, cfg.Session.Campaign)) seedRestoreObject(fakeStore, sessionPrefix+"transcripts/trimmed.json", []byte(`{"segments":[]}`+"\n")) - seedRestoreObject(fakeStore, sessionPrefix+"previous/manifest.json", []byte(`{"session_id":"2026-04-26"}`)) - seedRestoreObject(fakeStore, sessionPrefix+"previous/artifacts/session_recap.md", []byte("# previous recap\n")) + seedRestorePreviousCurrent(t, fakeStore, cfg, "# previous recap\n") restoreWithStoreAndRealPhases(t, fakeStore) @@ -209,7 +211,13 @@ scriptorium: sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), `{"segments":[]}`+"\n") - mustReadEquals(t, filepath.Join(sessionRoot, "previous", "manifest.json"), `{"session_id":"2026-04-26"}`) + previousManifestBytes, err := os.ReadFile(filepath.Join(sessionRoot, "previous", "manifest.json")) + if err != nil { + t.Fatalf("read restored previous manifest: %v", err) + } + 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") scriptoriumFake := &scriptorium.FakeRunner{} diff --git a/internal/previouscache/previouscache.go b/internal/previouscache/previouscache.go new file mode 100644 index 0000000..6d0504b --- /dev/null +++ b/internal/previouscache/previouscache.go @@ -0,0 +1,495 @@ +package previouscache + +import ( + "context" + "fmt" + "os" + "path" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +const ( + InputKindManifest = "previous_manifest" + InputKindArtifact = "previous_artifact" + InputSource = "previous_session_archive.current" +) + +type Plan struct { + Records []Record + SkippedMissing []string + PreviousRunID string +} + +type Record struct { + Kind string + RequirementName string + Required bool + LocalRelativePath string + LocalPath string + RemoteKey string + S3Bucket string +} + +func BuildPlan( + ctx context.Context, + cfg *config.Config, + paths artifacts.SessionPaths, + requirements []artifacts.PreviousArtifactRequirement, + store storage.ObjectStore, +) (*Plan, error) { + if len(requirements) == 0 { + return &Plan{}, nil + } + if cfg == nil || cfg.Session == nil || cfg.Pipeline == nil { + return nil, fmt.Errorf("resolved config with session/pipeline is required") + } + + orderedRequirements := append([]artifacts.PreviousArtifactRequirement(nil), requirements...) + sort.Slice(orderedRequirements, func(i, j int) bool { + return orderedRequirements[i].Name < orderedRequirements[j].Name + }) + + requiredNames := requiredPreviousArtifactNames(orderedRequirements) + optionalNames := optionalPreviousArtifactNames(orderedRequirements) + previousSessionID := strings.TrimSpace(cfg.Session.PreviousSessionID) + if previousSessionID == "" { + if len(requiredNames) > 0 { + return nil, fmt.Errorf( + "previous_session_id is required for required previous-session artifacts: %s", + strings.Join(requiredNames, ", "), + ) + } + return &Plan{SkippedMissing: optionalNames}, nil + } + + if store == nil { + return nil, fmt.Errorf("previous-session artifact hydration requires object store backend") + } + if cfg.Pipeline.Storage.S3 == nil { + return nil, fmt.Errorf("pipeline.storage.s3 configuration is required for previous-session artifact hydration") + } + + campaign := strings.TrimSpace(cfg.Session.Campaign) + if campaign == "" { + return nil, fmt.Errorf("session campaign is required for previous-session artifact hydration") + } + bucket := strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) + if bucket == "" { + return nil, fmt.Errorf("pipeline.storage.s3.bucket is required for previous-session artifact hydration") + } + + previousSessionPrefix := artifacts.S3SessionPrefix( + cfg.Pipeline.Storage.S3.RootPrefix, + campaign, + previousSessionID, + ) + currentManifestKey, currentRunIDKey := artifacts.ResolveArchiveCurrentStateKeys(previousSessionPrefix) + + result := &Plan{} + + runPointerExists, err := store.Exists(ctx, currentRunIDKey) + if err != nil { + return nil, fmt.Errorf("check previous-session current run pointer %q: %w", currentRunIDKey, err) + } + if !runPointerExists { + if len(requiredNames) > 0 { + return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current run pointer missing: %q", currentRunIDKey) + } + result.SkippedMissing = optionalNames + return result, nil + } + + runIDTemp, err := downloadObjectToTemp(ctx, store, currentRunIDKey, "narratio-previous-run-id-*.txt") + if err != nil { + return nil, fmt.Errorf("download previous-session current run pointer %q: %w", currentRunIDKey, err) + } + defer func() { _ = os.Remove(runIDTemp) }() + + runIDBytes, err := os.ReadFile(runIDTemp) + if err != nil { + return nil, fmt.Errorf("read previous-session current run pointer %q: %w", currentRunIDKey, err) + } + previousRunID := strings.TrimSpace(string(runIDBytes)) + if previousRunID == "" { + return nil, fmt.Errorf("previous-session current run pointer %q is empty", currentRunIDKey) + } + result.PreviousRunID = previousRunID + + manifestExists, err := store.Exists(ctx, currentManifestKey) + if err != nil { + return nil, fmt.Errorf("check previous-session current manifest %q: %w", currentManifestKey, err) + } + if !manifestExists { + if len(requiredNames) > 0 { + return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current manifest missing: %q", currentManifestKey) + } + result.SkippedMissing = optionalNames + return result, nil + } + + manifestTemp, err := downloadObjectToTemp(ctx, store, currentManifestKey, "narratio-previous-manifest-*.json") + if err != nil { + return nil, fmt.Errorf("download previous-session current manifest %q: %w", currentManifestKey, err) + } + defer func() { _ = os.Remove(manifestTemp) }() + + manifestStore := &manifest.LocalStore{} + previousManifest, err := manifestStore.Load(ctx, manifestTemp) + if err != nil { + return nil, fmt.Errorf("decode downloaded previous-session manifest %q: %w", currentManifestKey, err) + } + if strings.TrimSpace(previousManifest.SessionID) != previousSessionID { + return nil, fmt.Errorf( + "previous-session manifest session_id %q does not match configured previous_session_id %q", + strings.TrimSpace(previousManifest.SessionID), + previousSessionID, + ) + } + if strings.TrimSpace(previousManifest.Campaign) != campaign { + return nil, fmt.Errorf( + "previous-session manifest campaign %q does not match current campaign %q", + strings.TrimSpace(previousManifest.Campaign), + campaign, + ) + } + if strings.TrimSpace(previousManifest.RunID) == "" { + return nil, fmt.Errorf("previous-session manifest run_id is required") + } + if strings.TrimSpace(previousManifest.RunID) != previousRunID { + return nil, fmt.Errorf( + "previous-session current run pointer %q references run %q but current manifest run_id is %q", + currentRunIDKey, + previousRunID, + strings.TrimSpace(previousManifest.RunID), + ) + } + + manifestRel, err := relativeToSession(paths, paths.PreviousManifestPath) + if err != nil { + return nil, err + } + result.Records = append(result.Records, Record{ + Kind: InputKindManifest, + LocalRelativePath: manifestRel, + LocalPath: paths.PreviousManifestPath, + RemoteKey: currentManifestKey, + S3Bucket: bucket, + }) + + for _, requirement := range orderedRequirements { + candidates := artifactRelativePathCandidates(requirement.Name, previousManifest, cfg) + if len(candidates) == 0 { + if requirement.Required { + return nil, fmt.Errorf( + "required previous-session artifact %q is unavailable in previous-session manifest/archive", + requirement.Name, + ) + } + result.SkippedMissing = append(result.SkippedMissing, requirement.Name) + continue + } + + selectedRel := "" + selectedKey := "" + for _, candidate := range candidates { + remoteKey := artifacts.S3PromotedArtifactKey(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 + } + selectedRel = candidate + selectedKey = remoteKey + break + } + if selectedRel == "" { + if requirement.Required { + return nil, fmt.Errorf( + "required previous-session artifact %q object missing from archive candidate keys", + requirement.Name, + ) + } + result.SkippedMissing = append(result.SkippedMissing, requirement.Name) + continue + } + + localPath := artifacts.SessionPreviousArtifactPath(paths, selectedRel) + localRel, err := relativeToSession(paths, localPath) + if err != nil { + return nil, err + } + result.Records = append(result.Records, Record{ + Kind: InputKindArtifact, + RequirementName: requirement.Name, + Required: requirement.Required, + LocalRelativePath: localRel, + LocalPath: localPath, + RemoteKey: selectedKey, + S3Bucket: bucket, + }) + } + + sort.Strings(result.SkippedMissing) + sort.Slice(result.Records, func(i, j int) bool { + if result.Records[i].LocalRelativePath != result.Records[j].LocalRelativePath { + return result.Records[i].LocalRelativePath < result.Records[j].LocalRelativePath + } + return result.Records[i].RemoteKey < result.Records[j].RemoteKey + }) + return result, nil +} + +func requiredPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { + names := make([]string, 0, len(requirements)) + for _, requirement := range requirements { + if requirement.Required { + names = append(names, strings.TrimSpace(requirement.Name)) + } + } + sort.Strings(names) + return names +} + +func optionalPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { + names := make([]string, 0, len(requirements)) + for _, requirement := range requirements { + if requirement.Required { + continue + } + names = append(names, strings.TrimSpace(requirement.Name)) + } + sort.Strings(names) + return names +} + +func artifactRelativePathCandidates( + artifactName string, + previousManifest *manifest.Manifest, + cfg *config.Config, +) []string { + candidates := []string{} + appendCandidate := func(v string) { + normalized, err := normalizeArchiveRelativePath(v) + if err != nil { + return + } + candidates = append(candidates, normalized) + } + + sourceID := artifacts.ConfiguredArtifactSourceID(artifactName) + if rel, ok := manifestArtifactRelativePathBySourceID(previousManifest, sourceID); ok { + appendCandidate(rel) + base := path.Base(rel) + for _, promoted := range manifestPromotedPaths(previousManifest) { + if path.Base(promoted) == base { + appendCandidate(promoted) + } + } + } + + if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { + if artifactCfg, ok := cfg.Pipeline.Scriptorium.Artifacts[artifactName]; ok { + appendCandidate(artifactCfg.OutputPath) + } + } + + return dedupeOrderedStrings(candidates) +} + +func manifestArtifactRelativePathBySourceID(previousManifest *manifest.Manifest, sourceID string) (string, bool) { + if previousManifest == nil || len(previousManifest.Stages) == 0 { + return "", false + } + sourceID = strings.TrimSpace(sourceID) + if sourceID == "" { + return "", false + } + + stageNames := make([]string, 0, len(previousManifest.Stages)) + if _, ok := previousManifest.Stages["analyze"]; ok { + stageNames = append(stageNames, "analyze") + } + for stageName := range previousManifest.Stages { + if stageName == "analyze" { + continue + } + stageNames = append(stageNames, stageName) + } + start := 0 + if len(stageNames) > 0 && stageNames[0] == "analyze" { + start = 1 + } + sort.Strings(stageNames[start:]) + + for _, stageName := range stageNames { + sr := previousManifest.Stages[stageName] + if sr == nil { + continue + } + for _, out := range sr.Outputs { + if strings.TrimSpace(out.SourceID) != sourceID { + continue + } + rel, ok := deriveManifestRelativePath(previousManifest, out.LocalPath) + if ok { + return rel, true + } + } + } + return "", false +} + +func deriveManifestRelativePath(previousManifest *manifest.Manifest, localPath string) (string, bool) { + trimmed := strings.TrimSpace(localPath) + if trimmed == "" { + return "", false + } + if !filepath.IsAbs(trimmed) { + normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(trimmed)) + if err != nil { + return "", false + } + return normalized, true + } + + sessionRoot, ok := manifestSessionRoot(previousManifest) + if !ok { + return "", false + } + rel, err := filepath.Rel(sessionRoot, trimmed) + if err != nil { + return "", false + } + normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(rel)) + if err != nil { + return "", false + } + return normalized, true +} + +func manifestSessionRoot(previousManifest *manifest.Manifest) (string, bool) { + if previousManifest == nil { + return "", false + } + runRoot := filepath.Clean(strings.TrimSpace(previousManifest.LocalWorkDir)) + runID := strings.TrimSpace(previousManifest.RunID) + if runRoot == "" || runID == "" { + return "", false + } + if filepath.Base(runRoot) != runID { + return "", false + } + runsDir := filepath.Dir(runRoot) + if filepath.Base(runsDir) != config.PathRunsDirSegment { + return "", false + } + return filepath.Dir(runsDir), true +} + +func manifestPromotedPaths(previousManifest *manifest.Manifest) []string { + if previousManifest == nil || len(previousManifest.Stages) == 0 { + return nil + } + sr := previousManifest.Stages["archive"] + if sr == nil || sr.Metadata == nil { + return nil + } + raw, ok := sr.Metadata["promoted_paths"] + if !ok { + return nil + } + values, ok := raw.([]any) + if !ok { + return nil + } + out := make([]string, 0, len(values)) + for _, value := range values { + asString, ok := value.(string) + if !ok { + continue + } + normalized, err := normalizeArchiveRelativePath(asString) + if err != nil { + continue + } + out = append(out, normalized) + } + return dedupeOrderedStrings(out) +} + +func normalizeArchiveRelativePath(rel string) (string, error) { + trimmed := strings.TrimSpace(rel) + if trimmed == "" { + return "", fmt.Errorf("relative path is required") + } + cleaned := filepath.ToSlash(filepath.Clean(filepath.FromSlash(trimmed))) + if cleaned == "." || cleaned == "" { + return "", fmt.Errorf("relative path is required") + } + if filepath.IsAbs(trimmed) || strings.HasPrefix(cleaned, "/") || cleaned == ".." || strings.HasPrefix(cleaned, "../") { + return "", fmt.Errorf("path must be a clean relative path") + } + return cleaned, nil +} + +func relativeToSession(paths artifacts.SessionPaths, localPath string) (string, error) { + root := filepath.Clean(paths.Root) + if strings.TrimSpace(root) == "" { + return "", fmt.Errorf("session root is required") + } + rel, err := filepath.Rel(root, filepath.Clean(localPath)) + if err != nil { + return "", fmt.Errorf("resolve previous-cache relative path: %w", err) + } + normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(rel)) + if err != nil { + return "", fmt.Errorf("resolve previous-cache relative path: %w", err) + } + return normalized, nil +} + +func dedupeOrderedStrings(values []string) []string { + if len(values) == 0 { + return nil + } + seen := map[string]struct{}{} + out := make([]string, 0, len(values)) + for _, value := range values { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + continue + } + if _, ok := seen[trimmed]; ok { + continue + } + seen[trimmed] = struct{}{} + out = append(out, trimmed) + } + return out +} + +func downloadObjectToTemp(ctx context.Context, store storage.ObjectStore, key, pattern string) (string, error) { + tmp, err := os.CreateTemp("", pattern) + if err != nil { + return "", fmt.Errorf("create temp file: %w", err) + } + path := tmp.Name() + if err := tmp.Close(); err != nil { + _ = os.Remove(path) + return "", fmt.Errorf("close temp file: %w", err) + } + if err := store.Download(ctx, key, path); err != nil { + _ = os.Remove(path) + return "", err + } + return path, nil +} diff --git a/internal/previouscache/previouscache_test.go b/internal/previouscache/previouscache_test.go new file mode 100644 index 0000000..60569c3 --- /dev/null +++ b/internal/previouscache/previouscache_test.go @@ -0,0 +1,180 @@ +package previouscache + +import ( + "context" + "encoding/json" + "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 TestBuildPlanResolvesPromotedArtifactFromPreviousManifest(t *testing.T) { + cfg, paths := previousCacheTestConfig(t) + store := &storage.FakeBackend{} + seedPreviousCurrent(t, store, cfg, previousManifestWithOutput(t, cfg, "artifacts/session_recap.md", []string{"artifacts/session_recap.md"})) + previousPrefix := artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.PreviousSessionID) + store.SeedObject(storage.FakeObject{Key: previousPrefix + "artifacts/session_recap.md", Data: []byte("# recap\n")}) + + plan, err := BuildPlan(context.Background(), cfg, paths, []artifacts.PreviousArtifactRequirement{ + {Name: "session_recap", Required: true}, + }, store) + if err != nil { + t.Fatalf("BuildPlan() error = %v", err) + } + if plan.PreviousRunID != "previous-run" { + t.Fatalf("PreviousRunID = %q, want previous-run", plan.PreviousRunID) + } + got := recordRelPaths(plan.Records) + want := []string{"previous/artifacts/session_recap.md", "previous/manifest.json"} + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Fatalf("record rel paths = %#v, want %#v", got, want) + } +} + +func TestBuildPlanFallsBackToConfiguredOutputPath(t *testing.T) { + cfg, paths := previousCacheTestConfig(t) + store := &storage.FakeBackend{} + seedPreviousCurrent(t, store, cfg, previousManifestWithOutput(t, cfg, "", nil)) + previousPrefix := artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.PreviousSessionID) + store.SeedObject(storage.FakeObject{Key: previousPrefix + "artifacts/session_recap.md", Data: []byte("# recap\n")}) + + plan, err := BuildPlan(context.Background(), cfg, paths, []artifacts.PreviousArtifactRequirement{ + {Name: "session_recap", Required: true}, + }, store) + if err != nil { + t.Fatalf("BuildPlan() error = %v", err) + } + if len(plan.Records) != 2 { + t.Fatalf("records len = %d, want 2", len(plan.Records)) + } + if plan.Records[0].LocalRelativePath != "previous/artifacts/session_recap.md" { + t.Fatalf("artifact local relative path = %q", plan.Records[0].LocalRelativePath) + } +} + +func TestBuildPlanSkipsMissingOptionalPreviousArtifact(t *testing.T) { + cfg, paths := previousCacheTestConfig(t) + store := &storage.FakeBackend{} + seedPreviousCurrent(t, store, cfg, previousManifestWithOutput(t, cfg, "", nil)) + + plan, err := BuildPlan(context.Background(), cfg, paths, []artifacts.PreviousArtifactRequirement{ + {Name: "session_recap", Required: false}, + }, store) + if err != nil { + t.Fatalf("BuildPlan() error = %v", err) + } + if strings.Join(plan.SkippedMissing, ",") != "session_recap" { + t.Fatalf("SkippedMissing = %#v, want session_recap", plan.SkippedMissing) + } + if len(plan.Records) != 1 || plan.Records[0].Kind != InputKindManifest { + t.Fatalf("records = %#v, want manifest only", plan.Records) + } +} + +func TestBuildPlanMissingRequiredPreviousArtifactFails(t *testing.T) { + cfg, paths := previousCacheTestConfig(t) + store := &storage.FakeBackend{} + seedPreviousCurrent(t, store, cfg, previousManifestWithOutput(t, cfg, "", nil)) + + _, err := BuildPlan(context.Background(), cfg, paths, []artifacts.PreviousArtifactRequirement{ + {Name: "session_recap", Required: true}, + }, store) + if err == nil || !strings.Contains(err.Error(), `required previous-session artifact "session_recap" object missing`) { + t.Fatalf("BuildPlan() error = %v, want required missing error", err) + } +} + +func TestBuildPlanValidatesPreviousManifestIdentity(t *testing.T) { + cfg, paths := previousCacheTestConfig(t) + store := &storage.FakeBackend{} + manifest := previousManifestWithOutput(t, cfg, "artifacts/session_recap.md", nil) + manifest.SessionID = "wrong-session" + seedPreviousCurrent(t, store, cfg, manifest) + + _, err := BuildPlan(context.Background(), cfg, paths, []artifacts.PreviousArtifactRequirement{ + {Name: "session_recap", Required: true}, + }, store) + if err == nil || !strings.Contains(err.Error(), "does not match configured previous_session_id") { + t.Fatalf("BuildPlan() error = %v, want identity validation error", err) + } +} + +func previousCacheTestConfig(t *testing.T) (*config.Config, artifacts.SessionPaths) { + t.Helper() + workspaceRoot := t.TempDir() + cfg := &config.Config{ + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: workspaceRoot}, + Storage: config.StorageConfig{S3: &config.StorageS3Config{ + Bucket: "test-bucket", + RootPrefix: "dnd", + }}, + Scriptorium: &config.ScriptoriumConfig{Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "session_recap": { + Enabled: true, + OutputPath: "artifacts/session_recap.md", + }, + }}, + }, + Session: &config.SessionConfig{ + Campaign: "sample-campaign", + SessionID: "2026-05-03", + PreviousSessionID: "2026-04-26", + }, + } + return cfg, artifacts.NewLocalStore(workspaceRoot).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) +} + +func seedPreviousCurrent(t *testing.T, store *storage.FakeBackend, cfg *config.Config, m *manifest.Manifest) { + t.Helper() + previousPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) + manifestKey, runIDKey := artifacts.ResolveArchiveCurrentStateKeys(previousPrefix) + store.SeedObject(storage.FakeObject{Key: runIDKey, Data: []byte("previous-run\n")}) + data, err := marshalManifestForPreviousCacheTest(m) + if err != nil { + t.Fatalf("marshal manifest: %v", err) + } + store.SeedObject(storage.FakeObject{Key: manifestKey, Data: data}) +} + +func previousManifestWithOutput(t *testing.T, cfg *config.Config, rel string, promoted []string) *manifest.Manifest { + t.Helper() + m := manifest.New(cfg.Session.PreviousSessionID, time.Date(2026, 4, 26, 10, 0, 0, 0, time.UTC)) + m.Campaign = cfg.Session.Campaign + m.RunID = "previous-run" + m.LocalWorkDir = filepath.Join("/var/lib/narratio/work", cfg.Session.Campaign, cfg.Session.PreviousSessionID, "runs", m.RunID) + if strings.TrimSpace(rel) != "" { + m.MarkStageSucceeded("analyze", time.Date(2026, 4, 26, 10, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{ + {SourceID: "narratio.artifact.session_recap", LocalPath: filepath.Join(filepath.Dir(filepath.Dir(m.LocalWorkDir)), filepath.FromSlash(rel))}, + }) + } + if promoted != nil { + if m.Stages["archive"] == nil { + m.MarkStageSucceeded("archive", time.Date(2026, 4, 26, 10, 2, 0, 0, time.UTC), nil) + } + m.Stages["archive"].Metadata = map[string]any{"promoted_paths": promoted} + } + return m +} + +func marshalManifestForPreviousCacheTest(m *manifest.Manifest) ([]byte, error) { + data, err := json.Marshal(m) + if err != nil { + return nil, err + } + return append(data, '\n'), nil +} + +func recordRelPaths(records []Record) []string { + out := make([]string, 0, len(records)) + for _, record := range records { + out = append(out, record.LocalRelativePath) + } + return out +} diff --git a/internal/stage/prepare_previous.go b/internal/stage/prepare_previous.go index d136f4f..e6ed6b9 100644 --- a/internal/stage/prepare_previous.go +++ b/internal/stage/prepare_previous.go @@ -4,20 +4,18 @@ import ( "context" "fmt" "os" - "path" "path/filepath" "sort" - "strings" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" - "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/previouscache" ) const ( - preparePreviousInputKindManifest = "previous_manifest" - preparePreviousInputKindArtifact = "previous_artifact" - preparePreviousInputSource = "previous_session_archive.current" + preparePreviousInputKindManifest = previouscache.InputKindManifest + preparePreviousInputKindArtifact = previouscache.InputKindArtifact + preparePreviousInputSource = previouscache.InputSource ) // previousSessionHydrationResult captures prepare-time previous-session cache materialization. @@ -43,203 +41,43 @@ func hydratePreviousSessionArtifacts( if env.ArtifactStore == nil { return nil, fmt.Errorf("artifact store is required") } + plan, err := previouscache.BuildPlan(ctx, env.Config, paths, requirements, env.ObjectStore) + if err != nil { + return nil, err + } + result := &previousSessionHydrationResult{ + SkippedMissing: append([]string(nil), plan.SkippedMissing...), + PreviousRunID: plan.PreviousRunID, + } - orderedRequirements := append([]artifacts.PreviousArtifactRequirement(nil), requirements...) - sort.Slice(orderedRequirements, func(i, j int) bool { - return orderedRequirements[i].Name < orderedRequirements[j].Name - }) - - requiredNames := requiredPreviousArtifactNames(orderedRequirements) - optionalNames := optionalPreviousArtifactNames(orderedRequirements) - previousSessionID := strings.TrimSpace(env.Config.Session.PreviousSessionID) - if previousSessionID == "" { - if len(requiredNames) > 0 { - return nil, fmt.Errorf( - "previous_session_id is required for required previous-session artifacts: %s", - strings.Join(requiredNames, ", "), - ) + for _, record := range plan.Records { + if err := os.MkdirAll(filepath.Dir(record.LocalPath), 0o755); err != nil { + return nil, fmt.Errorf("create previous-session path directory for %q: %w", record.LocalPath, err) } - return &previousSessionHydrationResult{SkippedMissing: optionalNames}, nil - } - - if env.ObjectStore == nil { - return nil, fmt.Errorf("previous-session artifact hydration requires object store backend") - } - if env.Config.Pipeline.Storage.S3 == nil { - return nil, fmt.Errorf("pipeline.storage.s3 configuration is required for previous-session artifact hydration") - } - campaign := strings.TrimSpace(env.Config.Session.Campaign) - if campaign == "" { - return nil, fmt.Errorf("session campaign is required for previous-session artifact hydration") - } - bucket := strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket) - if bucket == "" { - return nil, fmt.Errorf("pipeline.storage.s3.bucket is required for previous-session artifact hydration") - } - - previousSessionPrefix := artifacts.S3SessionPrefix( - env.Config.Pipeline.Storage.S3.RootPrefix, - campaign, - previousSessionID, - ) - currentManifestKey, currentRunIDKey := artifacts.ResolveArchiveCurrentStateKeys(previousSessionPrefix) - - result := &previousSessionHydrationResult{} - - runPointerExists, err := env.ObjectStore.Exists(ctx, currentRunIDKey) - if err != nil { - return nil, fmt.Errorf("check previous-session current run pointer %q: %w", currentRunIDKey, err) - } - if !runPointerExists { - if len(requiredNames) > 0 { - return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current run pointer missing: %q", currentRunIDKey) + if err := env.ObjectStore.Download(ctx, record.RemoteKey, record.LocalPath); err != nil { + return nil, fmt.Errorf("download previous-session object %q to %q: %w", record.RemoteKey, record.LocalPath, err) } - result.SkippedMissing = optionalNames - return result, nil - } - - runIDTemp, err := downloadObjectToTempStage(ctx, env.ObjectStore, currentRunIDKey, "narratio-prepare-previous-run-id-*.txt") - if err != nil { - return nil, fmt.Errorf("download previous-session current run pointer %q: %w", currentRunIDKey, err) - } - defer func() { _ = os.Remove(runIDTemp) }() - - runIDBytes, err := os.ReadFile(runIDTemp) - if err != nil { - return nil, fmt.Errorf("read previous-session current run pointer %q: %w", currentRunIDKey, err) - } - previousRunID := strings.TrimSpace(string(runIDBytes)) - if previousRunID == "" { - return nil, fmt.Errorf("previous-session current run pointer %q is empty", currentRunIDKey) - } - result.PreviousRunID = previousRunID - - manifestExists, err := env.ObjectStore.Exists(ctx, currentManifestKey) - if err != nil { - return nil, fmt.Errorf("check previous-session current manifest %q: %w", currentManifestKey, err) - } - if !manifestExists { - if len(requiredNames) > 0 { - return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current manifest missing: %q", currentManifestKey) - } - result.SkippedMissing = optionalNames - return result, nil - } - - if err := os.MkdirAll(filepath.Dir(paths.PreviousManifestPath), 0o755); err != nil { - return nil, fmt.Errorf("create previous manifest directory: %w", err) - } - if err := env.ObjectStore.Download(ctx, currentManifestKey, paths.PreviousManifestPath); err != nil { - return nil, fmt.Errorf("download previous-session current manifest %q: %w", currentManifestKey, err) - } - - manifestStore := &manifest.LocalStore{} - previousManifest, err := manifestStore.Load(ctx, paths.PreviousManifestPath) - if err != nil { - return nil, fmt.Errorf("decode downloaded previous-session manifest %q: %w", currentManifestKey, err) - } - if strings.TrimSpace(previousManifest.SessionID) != previousSessionID { - return nil, fmt.Errorf( - "previous-session manifest session_id %q does not match configured previous_session_id %q", - strings.TrimSpace(previousManifest.SessionID), - previousSessionID, - ) - } - if strings.TrimSpace(previousManifest.Campaign) != campaign { - return nil, fmt.Errorf( - "previous-session manifest campaign %q does not match current campaign %q", - strings.TrimSpace(previousManifest.Campaign), - campaign, - ) - } - if strings.TrimSpace(previousManifest.RunID) == "" { - return nil, fmt.Errorf("previous-session manifest run_id is required") - } - if strings.TrimSpace(previousManifest.RunID) != previousRunID { - return nil, fmt.Errorf( - "previous-session current run pointer %q references run %q but current manifest run_id is %q", - currentRunIDKey, - previousRunID, - strings.TrimSpace(previousManifest.RunID), - ) - } - - manifestChecksum, err := env.ArtifactStore.Checksum(paths.PreviousManifestPath) - if err != nil { - return nil, fmt.Errorf("checksum downloaded previous-session manifest: %w", err) - } - result.Inputs = append(result.Inputs, manifest.InputRecord{ - Kind: preparePreviousInputKindManifest, - Path: paths.PreviousManifestPath, - Checksum: manifestChecksum, - Source: preparePreviousInputSource, - S3Bucket: bucket, - S3Key: currentManifestKey, - }) - - for _, requirement := range orderedRequirements { - candidates := previousArtifactRelativePathCandidates(requirement.Name, previousManifest, env.Config) - if len(candidates) == 0 { - if requirement.Required { - return nil, fmt.Errorf( - "required previous-session artifact %q is unavailable in previous-session manifest/archive", - requirement.Name, - ) + if record.Kind == preparePreviousInputKindArtifact { + if err := requireNonEmptyFile(record.LocalPath, "previous-session artifact "+record.RequirementName); err != nil { + return nil, err } - result.SkippedMissing = append(result.SkippedMissing, requirement.Name) - continue } - - selectedRel := "" - selectedKey := "" - for _, candidate := range candidates { - remoteKey := artifacts.S3PromotedArtifactKey(previousSessionPrefix, candidate) - exists, err := env.ObjectStore.Exists(ctx, remoteKey) - if err != nil { - return nil, fmt.Errorf("check previous-session artifact object %q: %w", remoteKey, err) - } - if !exists { - continue - } - selectedRel = candidate - selectedKey = remoteKey - break - } - if selectedRel == "" { - if requirement.Required { - return nil, fmt.Errorf( - "required previous-session artifact %q object missing from archive candidate keys", - requirement.Name, - ) - } - result.SkippedMissing = append(result.SkippedMissing, requirement.Name) - continue - } - - localPath := artifacts.SessionPreviousArtifactPath(paths, selectedRel) - if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil { - return nil, fmt.Errorf("create previous-session artifact directory for %q: %w", localPath, err) - } - if err := env.ObjectStore.Download(ctx, selectedKey, localPath); err != nil { - return nil, fmt.Errorf("download previous-session artifact %q from %q: %w", requirement.Name, selectedKey, err) - } - if err := requireNonEmptyFile(localPath, "previous-session artifact "+requirement.Name); err != nil { - return nil, err - } - checksum, err := env.ArtifactStore.Checksum(localPath) + checksum, err := env.ArtifactStore.Checksum(record.LocalPath) if err != nil { - return nil, fmt.Errorf("checksum previous-session artifact %q: %w", requirement.Name, err) + return nil, fmt.Errorf("checksum previous-session object %q: %w", record.RemoteKey, err) } result.Inputs = append(result.Inputs, manifest.InputRecord{ - Kind: preparePreviousInputKindArtifact, - Path: localPath, + Kind: record.Kind, + Path: record.LocalPath, Checksum: checksum, Source: preparePreviousInputSource, - S3Bucket: bucket, - S3Key: selectedKey, + S3Bucket: record.S3Bucket, + S3Key: record.RemoteKey, }) - result.Hydrated = append(result.Hydrated, requirement.Name) + if record.Kind == preparePreviousInputKindArtifact { + result.Hydrated = append(result.Hydrated, record.RequirementName) + } } sort.Strings(result.Hydrated) @@ -252,224 +90,3 @@ func hydratePreviousSessionArtifacts( }) return result, nil } - -func requiredPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { - names := make([]string, 0, len(requirements)) - for _, requirement := range requirements { - if requirement.Required { - names = append(names, strings.TrimSpace(requirement.Name)) - } - } - sort.Strings(names) - return names -} - -func optionalPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { - names := make([]string, 0, len(requirements)) - for _, requirement := range requirements { - if requirement.Required { - continue - } - names = append(names, strings.TrimSpace(requirement.Name)) - } - sort.Strings(names) - return names -} - -func previousArtifactRelativePathCandidates( - artifactName string, - previousManifest *manifest.Manifest, - cfg *config.Config, -) []string { - candidates := []string{} - appendCandidate := func(v string) { - normalized, err := normalizeArchiveRelativePath(v) - if err != nil { - return - } - candidates = append(candidates, normalized) - } - - sourceID := artifacts.ConfiguredArtifactSourceID(artifactName) - if rel, ok := previousManifestArtifactRelativePathBySourceID(previousManifest, sourceID); ok { - appendCandidate(rel) - base := path.Base(rel) - for _, promoted := range previousManifestPromotedPaths(previousManifest) { - if path.Base(promoted) == base { - appendCandidate(promoted) - } - } - } - - if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { - if artifactCfg, ok := cfg.Pipeline.Scriptorium.Artifacts[artifactName]; ok { - appendCandidate(artifactCfg.OutputPath) - } - } - - return dedupeOrderedStrings(candidates) -} - -func previousManifestArtifactRelativePathBySourceID(previousManifest *manifest.Manifest, sourceID string) (string, bool) { - if previousManifest == nil || len(previousManifest.Stages) == 0 { - return "", false - } - sourceID = strings.TrimSpace(sourceID) - if sourceID == "" { - return "", false - } - - stageNames := make([]string, 0, len(previousManifest.Stages)) - if _, ok := previousManifest.Stages["analyze"]; ok { - stageNames = append(stageNames, "analyze") - } - for stageName := range previousManifest.Stages { - if stageName == "analyze" { - continue - } - stageNames = append(stageNames, stageName) - } - start := 0 - if len(stageNames) > 0 && stageNames[0] == "analyze" { - start = 1 - } - sort.Strings(stageNames[start:]) - - for _, stageName := range stageNames { - sr := previousManifest.Stages[stageName] - if sr == nil { - continue - } - for _, out := range sr.Outputs { - if strings.TrimSpace(out.SourceID) != sourceID { - continue - } - rel, ok := derivePreviousManifestRelativePath(previousManifest, out.LocalPath) - if ok { - return rel, true - } - } - } - return "", false -} - -func derivePreviousManifestRelativePath(previousManifest *manifest.Manifest, localPath string) (string, bool) { - trimmed := strings.TrimSpace(localPath) - if trimmed == "" { - return "", false - } - if !filepath.IsAbs(trimmed) { - normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(trimmed)) - if err != nil { - return "", false - } - return normalized, true - } - - sessionRoot, ok := previousManifestSessionRoot(previousManifest) - if !ok { - return "", false - } - rel, err := filepath.Rel(sessionRoot, trimmed) - if err != nil { - return "", false - } - normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(rel)) - if err != nil { - return "", false - } - return normalized, true -} - -func previousManifestSessionRoot(previousManifest *manifest.Manifest) (string, bool) { - if previousManifest == nil { - return "", false - } - runRoot := filepath.Clean(strings.TrimSpace(previousManifest.LocalWorkDir)) - runID := strings.TrimSpace(previousManifest.RunID) - if runRoot == "" || runID == "" { - return "", false - } - if filepath.Base(runRoot) != runID { - return "", false - } - runsDir := filepath.Dir(runRoot) - if filepath.Base(runsDir) != config.PathRunsDirSegment { - return "", false - } - return filepath.Dir(runsDir), true -} - -func previousManifestPromotedPaths(previousManifest *manifest.Manifest) []string { - if previousManifest == nil || len(previousManifest.Stages) == 0 { - return nil - } - sr := previousManifest.Stages["archive"] - if sr == nil || sr.Metadata == nil { - return nil - } - raw, ok := sr.Metadata["promoted_paths"] - if !ok { - return nil - } - values, ok := raw.([]any) - if !ok { - return nil - } - out := make([]string, 0, len(values)) - for _, value := range values { - asString, ok := value.(string) - if !ok { - continue - } - normalized, err := normalizeArchiveRelativePath(asString) - if err != nil { - continue - } - out = append(out, normalized) - } - return dedupeOrderedStrings(out) -} - -func dedupeOrderedStrings(values []string) []string { - if len(values) == 0 { - return nil - } - seen := map[string]struct{}{} - out := make([]string, 0, len(values)) - for _, value := range values { - trimmed := strings.TrimSpace(value) - if trimmed == "" { - continue - } - if _, ok := seen[trimmed]; ok { - continue - } - seen[trimmed] = struct{}{} - out = append(out, trimmed) - } - return out -} - -func downloadObjectToTempStage( - ctx context.Context, - store interface { - Download(context.Context, string, string) error - }, - key, pattern string, -) (string, error) { - tmp, err := os.CreateTemp("", pattern) - if err != nil { - return "", fmt.Errorf("create temp file: %w", err) - } - path := tmp.Name() - if err := tmp.Close(); err != nil { - _ = os.Remove(path) - return "", fmt.Errorf("close temp file: %w", err) - } - if err := store.Download(ctx, key, path); err != nil { - _ = os.Remove(path) - return "", err - } - return path, nil -}