From 622677d0384b7882bcc1dff9f76b61cce9280581 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 17 May 2026 21:15:51 +0000 Subject: [PATCH] Added run manifest scaffolding and helpers --- internal/app/runner.go | 144 ++++++++++++++++++--- internal/app/runner_test.go | 76 ++++++++++++ internal/artifacts/paths.go | 5 + internal/artifacts/paths_model_test.go | 10 ++ internal/manifest/run_manifest.go | 164 ++++++++++++++++++++++++ internal/manifest/run_manifest_test.go | 50 ++++++++ internal/manifest/store.go | 165 +++++++++++++++++++++++++ internal/manifest/store_test.go | 70 +++++++++++ 8 files changed, 665 insertions(+), 19 deletions(-) create mode 100644 internal/manifest/run_manifest.go create mode 100644 internal/manifest/run_manifest_test.go diff --git a/internal/app/runner.go b/internal/app/runner.go index 1963636..a244476 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -26,11 +26,13 @@ type RunOptions struct { } type RunSummary struct { - SessionID string - ManifestPath string - StageNames []string - Executed []string - Skipped []string + SessionID string + RunID string + ManifestPath string + RunManifestPath string + StageNames []string + Executed []string + Skipped []string } func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) { @@ -110,7 +112,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if err != nil { return nil, err } - identityChanged, err := ensureManifestIdentity(cfg, m) + runID, err := artifacts.NewRunID() + if err != nil { + return nil, fmt.Errorf("generate run id: %w", err) + } + identityChanged, err := ensureManifestIdentity(cfg, m, runID) if err != nil { return nil, fmt.Errorf("initialize manifest identity: %w", err) } @@ -119,6 +125,29 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err) } } + runManifestPath := artifacts.SessionRunManifestPathForCampaign( + cfg.Pipeline.Workspace.Root, + cfg.Session.Campaign, + cfg.Session.SessionID, + runID, + ) + runManifestStore := &manifest.LocalStore{} + runManifest, err := runManifestStore.CreateRun( + ctx, + cfg.Session.SessionID, + cfg.Session.Campaign, + runID, + opts.Force, + requestedStageNames(stages), + ) + if err != nil { + return nil, fmt.Errorf("create run manifest: %w", err) + } + runManifest.SessionManifestPath = manifestPath + syncRunManifestIdentityFromSession(m, runManifest) + if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { + return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err) + } stageEnv := env @@ -133,12 +162,23 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if d.Action == stageActionSkip { skipped = append(skipped, s.Name()) + skipAt := nowUTC() + runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt) + runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded") + if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { + return nil, fmt.Errorf("save run manifest after skip %q: %w", s.Name(), err) + } env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force) continue } executed = append(executed, s.Name()) now := nowUTC() + runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now) + runManifest.MarkStageRunning(s.Name(), now) + if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { + return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err) + } m.MarkStageRunning(s.Name(), now) env.Logger.Info("starting stage", "stage", s.Name()) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { @@ -148,35 +188,62 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage result, err := s.Run(ctx, stageEnv, m) if err != nil { - m.MarkStageFailed(s.Name(), nowUTC(), err.Error()) + failedAt := nowUTC() + m.MarkStageFailed(s.Name(), failedAt, err.Error()) if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil { return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr) } + runManifest.MarkStageFailed(s.Name(), failedAt, err.Error()) + syncRunManifestIdentityFromSession(m, runManifest) + if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil { + return nil, fmt.Errorf("stage %q failed (%v) and run-manifest save failed (%v)", s.Name(), err, saveErr) + } env.Logger.Info("stage failed", "stage", s.Name(), "error", err) return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err) } outputs := mapResultOutputs(result) - m.MarkStageSucceeded(s.Name(), nowUTC(), outputs) + succeededAt := nowUTC() + m.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToManifest(m, s.Name(), result) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) } + runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs) + applyStageResultToRunManifest(runManifest, s.Name(), result) + syncRunManifestIdentityFromSession(m, runManifest) + if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { + return nil, fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err) + } env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath) env.Logger.Info("stage succeeded", "stage", s.Name()) } if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil { + failedAt := nowUTC() + runManifest.MarkFailed(failedAt, err.Error()) + syncRunManifestIdentityFromSession(m, runManifest) + if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil { + return nil, fmt.Errorf("post-archive cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr) + } return nil, fmt.Errorf("post-archive cleanup: %w", err) } + completedAt := nowUTC() + runManifest.MarkSucceeded(completedAt) + syncRunManifestIdentityFromSession(m, runManifest) + if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { + return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err) + } return &RunSummary{ - SessionID: cfg.Session.SessionID, - ManifestPath: manifestPath, - StageNames: runNames, - Executed: executed, - Skipped: skipped, + SessionID: cfg.Session.SessionID, + RunID: runID, + ManifestPath: manifestPath, + RunManifestPath: runManifestPath, + StageNames: runNames, + Executed: executed, + Skipped: skipped, }, nil } @@ -357,7 +424,7 @@ func applyStageResultToManifest(m *manifest.Manifest, stageName string, result * } } -func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, error) { +func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID string) (bool, error) { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil || m == nil { return false, nil } @@ -373,11 +440,8 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err m.Campaign = campaign changed = true } - if m.RunID == "" { - runID, err := artifacts.NewRunID() - if err != nil { - return false, err - } + runID = strings.TrimSpace(runID) + if runID != "" && m.RunID != runID { m.RunID = runID changed = true } @@ -409,6 +473,48 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err return changed, nil } +func requestedStageNames(stages []stage.Stage) []string { + out := make([]string, 0, len(stages)) + for _, s := range stages { + if s == nil { + continue + } + out = append(out, s.Name()) + } + return out +} + +func applyStageResultToRunManifest(m *manifest.RunManifest, stageName string, result *stage.StageResult) { + if m == nil || result == nil { + return + } + sr := m.Stages[stageName] + if sr == nil { + return + } + if len(result.Logs) > 0 { + sr.Logs = append([]string(nil), result.Logs...) + } + if len(result.GeneratedConfigs) > 0 { + sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...) + } + if len(result.Metadata) > 0 { + sr.Metadata = result.Metadata + } +} + +func syncRunManifestIdentityFromSession(session *manifest.Manifest, run *manifest.RunManifest) { + if session == nil || run == nil { + return + } + run.Campaign = session.Campaign + run.LocalWorkDir = session.LocalWorkDir + run.LocalSpoolDir = session.LocalSpoolDir + run.S3Bucket = session.S3Bucket + run.S3SessionPrefix = session.S3SessionPrefix + run.S3RunPrefix = session.S3RunPrefix +} + func manifestPathFor(cfg *config.Config) string { return artifacts.SessionManifestPathForCampaign( cfg.Pipeline.Workspace.Root, diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 5e6863f..711c6ca 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -322,6 +322,82 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) { } } +func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) { + cfg := testConfig(t) + run1, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{}) + if err != nil { + t.Fatalf("first executeStages() error = %v", err) + } + run2, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{Force: true}) + if err != nil { + t.Fatalf("second executeStages() error = %v", err) + } + + if run1.RunID == "" || run2.RunID == "" { + t.Fatalf("run ids must be set, got %q and %q", run1.RunID, run2.RunID) + } + if run1.RunID == run2.RunID { + t.Fatalf("expected distinct run ids, got %q", run1.RunID) + } + if run1.RunManifestPath == "" || run2.RunManifestPath == "" { + t.Fatalf("run manifest paths must be set, got %q and %q", run1.RunManifestPath, run2.RunManifestPath) + } + if run1.RunManifestPath == run2.RunManifestPath { + t.Fatalf("expected distinct run manifest paths, got %q", run1.RunManifestPath) + } + for _, path := range []string{run1.RunManifestPath, run2.RunManifestPath} { + if _, statErr := os.Stat(path); statErr != nil { + t.Fatalf("run manifest missing at %q: %v", path, statErr) + } + } + + store := &manifest.LocalStore{} + sessionManifest, err := store.Load(context.Background(), run2.ManifestPath) + if err != nil { + t.Fatalf("Load session manifest error = %v", err) + } + if sessionManifest.RunID != run2.RunID { + t.Fatalf("session manifest run_id = %q, want latest run id %q", sessionManifest.RunID, run2.RunID) + } +} + +func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + manifestPath := manifestPathFor(cfg) + existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) + existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) + if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := store.Save(context.Background(), manifestPath, existing); err != nil { + t.Fatalf("Save manifest error = %v", err) + } + + summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" { + t.Fatalf("summary = %#v, want skipped transcribe", summary) + } + + runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatalf("LoadRun() error = %v", err) + } + sr := runManifest.Stages["transcribe"] + if sr == nil { + t.Fatal("run manifest transcribe stage missing") + } + if sr.Action != manifest.RunStageActionSkip { + t.Fatalf("action = %q, want %q", sr.Action, manifest.RunStageActionSkip) + } + if sr.Status != manifest.StatusSkipped { + t.Fatalf("status = %q, want %q", sr.Status, manifest.StatusSkipped) + } +} + func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { cases := []struct { name string diff --git a/internal/artifacts/paths.go b/internal/artifacts/paths.go index ccd3f5c..95b6e03 100644 --- a/internal/artifacts/paths.go +++ b/internal/artifacts/paths.go @@ -52,6 +52,11 @@ func SessionRunRootForCampaign(rootDir, campaign, sessionID, runID string) strin return filepath.Join(SessionRunsDirForCampaign(rootDir, campaign, sessionID), runID) } +// SessionRunManifestPathForCampaign returns the canonical run manifest path under runs/{run_id}/manifest.json. +func SessionRunManifestPathForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), config.PathManifestFile) +} + // SessionRunStageDirForCampaign returns the canonical stage directory under runs/{run_id}/{stage}. func SessionRunStageDirForCampaign(rootDir, campaign, sessionID, runID, stageName string) string { return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), stageName) diff --git a/internal/artifacts/paths_model_test.go b/internal/artifacts/paths_model_test.go index 4a427b3..480c1e0 100644 --- a/internal/artifacts/paths_model_test.go +++ b/internal/artifacts/paths_model_test.go @@ -48,6 +48,16 @@ func TestSessionRunRootAndStageDirForCampaign(t *testing.T) { } } +func TestSessionRunManifestPathForCampaign(t *testing.T) { + root := "/tmp/workspace" + runID := "20260515T031522Z-a1b2c3d4" + got := SessionRunManifestPathForCampaign(root, "forsaken", "2026-04-19", runID) + want := filepath.Join(root, "work", "forsaken", "2026-04-19", "runs", runID, "manifest.json") + if got != want { + t.Fatalf("SessionRunManifestPathForCampaign() = %q, want %q", got, want) + } +} + func TestSessionSpoolAudioDir(t *testing.T) { root := "/var/spool/narratio" got := SessionSpoolAudioDir(root, "forsaken", "2026-04-19", "20260515T031522Z-a1b2c3d4") diff --git a/internal/manifest/run_manifest.go b/internal/manifest/run_manifest.go new file mode 100644 index 0000000..63f08e3 --- /dev/null +++ b/internal/manifest/run_manifest.go @@ -0,0 +1,164 @@ +package manifest + +import ( + "strings" + "time" +) + +type RunManifestStatus string + +const ( + RunManifestStatusRunning RunManifestStatus = "running" + RunManifestStatusSucceeded RunManifestStatus = "succeeded" + RunManifestStatusFailed RunManifestStatus = "failed" +) + +type RunStageAction string + +const ( + RunStageActionRun RunStageAction = "run" + RunStageActionSkip RunStageAction = "skip" +) + +// RunStageRecord tracks lifecycle and provenance for one stage within a single invocation. +type RunStageRecord struct { + Name string `json:"name"` + Action RunStageAction `json:"action"` + Status StageStatus `json:"status"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + Outputs []ArtifactRecord `json:"outputs,omitempty"` + Logs []string `json:"logs,omitempty"` + GeneratedConfigs []string `json:"generated_configs,omitempty"` + Error *ErrorRecord `json:"error,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +// RunManifest is the invocation-scoped execution record under runs/{run_id}/manifest.json. +type RunManifest struct { + SessionID string `json:"session_id"` + Campaign string `json:"campaign,omitempty"` + RunID string `json:"run_id"` + Force bool `json:"force"` + RequestedStages []string `json:"requested_stages,omitempty"` + SessionManifestPath string `json:"session_manifest_path,omitempty"` + LocalWorkDir string `json:"local_workdir,omitempty"` + LocalSpoolDir string `json:"local_spool_dir,omitempty"` + S3Bucket string `json:"s3_bucket,omitempty"` + S3SessionPrefix string `json:"s3_session_prefix,omitempty"` + S3RunPrefix string `json:"s3_run_prefix,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + Status RunManifestStatus `json:"status"` + LastError *ErrorRecord `json:"last_error,omitempty"` + Stages map[string]*RunStageRecord `json:"stages"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +// NewRun constructs a new run manifest with deterministic timestamps. +func NewRun(sessionID, campaign, runID string, force bool, requestedStages []string, now time.Time) *RunManifest { + return &RunManifest{ + SessionID: strings.TrimSpace(sessionID), + Campaign: strings.TrimSpace(campaign), + RunID: strings.TrimSpace(runID), + Force: force, + RequestedStages: append([]string(nil), requestedStages...), + CreatedAt: now, + UpdatedAt: now, + StartedAt: timePtr(now), + Status: RunManifestStatusRunning, + Stages: map[string]*RunStageRecord{}, + } +} + +func (m *RunManifest) SetStageAction(name string, action RunStageAction, at time.Time) { + s := m.ensureStage(name, at) + s.Action = action + s.UpdatedAt = at + m.UpdatedAt = at +} + +func (m *RunManifest) MarkStageRunning(name string, at time.Time) { + s := m.ensureStage(name, at) + s.Status = StatusRunning + s.StartedAt = timePtr(at) + s.CompletedAt = nil + s.Error = nil + s.UpdatedAt = at + m.UpdatedAt = at +} + +func (m *RunManifest) MarkStageSucceeded(name string, at time.Time, outputs []ArtifactRecord) { + s := m.ensureStage(name, at) + s.Status = StatusSucceeded + s.CompletedAt = timePtr(at) + s.Error = nil + s.Outputs = append([]ArtifactRecord(nil), outputs...) + s.UpdatedAt = at + m.UpdatedAt = at +} + +func (m *RunManifest) MarkStageFailed(name string, at time.Time, message string) { + s := m.ensureStage(name, at) + s.Status = StatusFailed + s.CompletedAt = timePtr(at) + s.Error = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)} + s.UpdatedAt = at + m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)} + m.UpdatedAt = at + m.Status = RunManifestStatusFailed + m.CompletedAt = timePtr(at) +} + +func (m *RunManifest) MarkStageSkipped(name string, at time.Time, reason string) { + s := m.ensureStage(name, at) + s.Status = StatusSkipped + s.CompletedAt = timePtr(at) + s.Error = &ErrorRecord{Message: strings.TrimSpace(reason), Code: "skipped", At: timePtr(at)} + s.UpdatedAt = at + m.UpdatedAt = at +} + +func (m *RunManifest) MarkSucceeded(at time.Time) { + m.Status = RunManifestStatusSucceeded + m.CompletedAt = timePtr(at) + m.UpdatedAt = at +} + +func (m *RunManifest) MarkFailed(at time.Time, message string) { + m.Status = RunManifestStatusFailed + m.CompletedAt = timePtr(at) + m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)} + m.UpdatedAt = at +} + +func (m *RunManifest) ensureStage(name string, at time.Time) *RunStageRecord { + if m.Stages == nil { + m.Stages = map[string]*RunStageRecord{} + } + + stageName := strings.TrimSpace(name) + s, ok := m.Stages[stageName] + if !ok || s == nil { + s = &RunStageRecord{ + Name: stageName, + Action: RunStageActionRun, + Status: StatusPending, + CreatedAt: at, + UpdatedAt: at, + } + m.Stages[stageName] = s + } + if s.Name == "" { + s.Name = stageName + } + if s.CreatedAt.IsZero() { + s.CreatedAt = at + } + + return s +} diff --git a/internal/manifest/run_manifest_test.go b/internal/manifest/run_manifest_test.go new file mode 100644 index 0000000..34f2baf --- /dev/null +++ b/internal/manifest/run_manifest_test.go @@ -0,0 +1,50 @@ +package manifest + +import ( + "testing" + "time" +) + +func TestRunManifestStageMarkHelpers(t *testing.T) { + rm := NewRun("2026-05-03", "forsaken", "20260517T000000Z-abcdef12", false, []string{"prepare"}, time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) + if rm.Status != RunManifestStatusRunning { + t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusRunning) + } + + runningAt := time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC) + rm.SetStageAction("prepare", RunStageActionRun, runningAt) + rm.MarkStageRunning("prepare", runningAt) + rm.MarkStageSucceeded("prepare", runningAt.Add(30*time.Second), []ArtifactRecord{ + {Kind: "input", LocalPath: "inputs/session.yml"}, + }) + + stage := rm.Stages["prepare"] + if stage == nil { + t.Fatal("prepare stage missing") + } + if stage.Action != RunStageActionRun { + t.Fatalf("action = %q, want %q", stage.Action, RunStageActionRun) + } + if stage.Status != StatusSucceeded { + t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded) + } + + skippedAt := runningAt.Add(1 * time.Minute) + rm.SetStageAction("notify", RunStageActionSkip, skippedAt) + rm.MarkStageSkipped("notify", skippedAt, "already_succeeded") + skipped := rm.Stages["notify"] + if skipped == nil { + t.Fatal("notify stage missing") + } + if skipped.Action != RunStageActionSkip { + t.Fatalf("action = %q, want %q", skipped.Action, RunStageActionSkip) + } + if skipped.Status != StatusSkipped { + t.Fatalf("status = %q, want %q", skipped.Status, StatusSkipped) + } + + rm.MarkSucceeded(skippedAt.Add(10 * time.Second)) + if rm.Status != RunManifestStatusSucceeded { + t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusSucceeded) + } +} diff --git a/internal/manifest/store.go b/internal/manifest/store.go index eb6a140..2ac34a0 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -129,6 +129,89 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { return nil } +// CreateRun returns a new in-memory run manifest for one invocation. +func (s *LocalStore) CreateRun( + ctx context.Context, + sessionID, campaign, runID string, + force bool, + requestedStages []string, +) (*RunManifest, error) { + if err := checkContext(ctx); err != nil { + return nil, err + } + if strings.TrimSpace(sessionID) == "" { + return nil, fmt.Errorf("create run manifest: session_id is required") + } + if strings.TrimSpace(runID) == "" { + return nil, fmt.Errorf("create run manifest: run_id is required") + } + + now := time.Now().UTC() + return NewRun(sessionID, campaign, runID, force, requestedStages, now), nil +} + +// LoadRun reads and validates a local JSON run manifest from path. +func (s *LocalStore) LoadRun(ctx context.Context, path string) (*RunManifest, error) { + if err := checkContext(ctx); err != nil { + return nil, err + } + if strings.TrimSpace(path) == "" { + return nil, fmt.Errorf("load run manifest: path is required") + } + + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("load run manifest %q: %w", path, err) + } + + var m RunManifest + if err := json.Unmarshal(data, &m); err != nil { + return nil, fmt.Errorf("decode run manifest %q: %w", path, err) + } + + if err := validateLoadedRunManifest(&m); err != nil { + return nil, fmt.Errorf("run manifest %q invalid: %w", path, err) + } + normalizeRunManifest(&m) + + return &m, nil +} + +// SaveRun writes the run manifest to path atomically via temp file + rename. +func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) error { + if err := checkContext(ctx); err != nil { + return err + } + if strings.TrimSpace(path) == "" { + return fmt.Errorf("save run manifest: path is required") + } + if m == nil { + return fmt.Errorf("save run manifest: manifest is nil") + } + if strings.TrimSpace(m.SessionID) == "" { + return fmt.Errorf("save run manifest: session_id is required") + } + if strings.TrimSpace(m.RunID) == "" { + return fmt.Errorf("save run manifest: run_id is required") + } + if m.CreatedAt.IsZero() { + return fmt.Errorf("save run manifest: created_at is required") + } + + m.UpdatedAt = time.Now().UTC() + if m.Stages == nil { + m.Stages = map[string]*RunStageRecord{} + } + + data, err := json.MarshalIndent(m, "", " ") + if err != nil { + return fmt.Errorf("save run manifest: marshal: %w", err) + } + data = append(data, '\n') + + return writeJSONAtomically(ctx, path, ".run-manifest.json.tmp-*", data) +} + func validateLoadedManifest(m *Manifest) error { if m == nil { return fmt.Errorf("manifest is nil") @@ -161,6 +244,88 @@ func normalizeManifest(m *Manifest) { } } +func validateLoadedRunManifest(m *RunManifest) error { + if m == nil { + return fmt.Errorf("manifest is nil") + } + if strings.TrimSpace(m.SessionID) == "" { + return fmt.Errorf("session_id is required") + } + if strings.TrimSpace(m.RunID) == "" { + return fmt.Errorf("run_id is required") + } + if m.CreatedAt.IsZero() { + return fmt.Errorf("created_at is required") + } + if m.UpdatedAt.IsZero() { + return fmt.Errorf("updated_at is required") + } + + return nil +} + +func normalizeRunManifest(m *RunManifest) { + if m.Stages == nil { + m.Stages = map[string]*RunStageRecord{} + } + for name, stage := range m.Stages { + if stage == nil { + stage = &RunStageRecord{ + Name: name, + Action: RunStageActionRun, + Status: StatusPending, + CreatedAt: m.CreatedAt, + UpdatedAt: m.UpdatedAt, + } + m.Stages[name] = stage + } + if stage.Name == "" { + stage.Name = name + } + } +} + +func writeJSONAtomically(ctx context.Context, path, tempPattern string, data []byte) error { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("create directory %q: %w", dir, err) + } + + tmp, err := os.CreateTemp(dir, tempPattern) + if err != nil { + return fmt.Errorf("create temp file: %w", err) + } + tmpName := tmp.Name() + removeTmp := true + defer func() { + if removeTmp { + _ = os.Remove(tmpName) + } + }() + + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return fmt.Errorf("write temp file: %w", err) + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return fmt.Errorf("sync temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close temp file: %w", err) + } + if err := checkContext(ctx); err != nil { + return err + } + + if err := os.Rename(tmpName, path); err != nil { + return fmt.Errorf("rename temp file: %w", err) + } + removeTmp = false + + return nil +} + func checkContext(ctx context.Context) error { if ctx == nil { return nil diff --git a/internal/manifest/store_test.go b/internal/manifest/store_test.go index e708083..012e9a1 100644 --- a/internal/manifest/store_test.go +++ b/internal/manifest/store_test.go @@ -150,3 +150,73 @@ func TestLoadRejectsInvalidManifest(t *testing.T) { t.Fatalf("error = %q, want session_id validation", err.Error()) } } + +func TestLocalStoreCreateSaveLoadRunManifestRoundTrip(t *testing.T) { + store := &LocalStore{} + ctx := context.Background() + + run, err := store.CreateRun( + ctx, + "2026-05-03", + "forsaken", + "20260517T000000Z-abcdef12", + true, + []string{"prepare", "transcribe"}, + ) + if err != nil { + t.Fatalf("CreateRun() error = %v", err) + } + run.SessionManifestPath = "/var/lib/narratio/work/forsaken/2026-05-03/manifest.json" + run.MarkStageRunning("prepare", time.Date(2026, 5, 3, 12, 1, 0, 0, time.UTC)) + run.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 12, 2, 0, 0, time.UTC), []ArtifactRecord{ + {Kind: "input", LocalPath: "inputs/session.yml"}, + }) + run.MarkSucceeded(time.Date(2026, 5, 3, 12, 3, 0, 0, time.UTC)) + + path := filepath.Join(t.TempDir(), "run-manifest.json") + if err := store.SaveRun(ctx, path, run); err != nil { + t.Fatalf("SaveRun() error = %v", err) + } + + loaded, err := store.LoadRun(ctx, path) + if err != nil { + t.Fatalf("LoadRun() error = %v", err) + } + + if loaded.SessionID != "2026-05-03" { + t.Fatalf("SessionID = %q, want %q", loaded.SessionID, "2026-05-03") + } + if loaded.Campaign != "forsaken" { + t.Fatalf("Campaign = %q, want %q", loaded.Campaign, "forsaken") + } + if loaded.RunID != "20260517T000000Z-abcdef12" { + t.Fatalf("RunID = %q, want %q", loaded.RunID, "20260517T000000Z-abcdef12") + } + if loaded.Status != RunManifestStatusSucceeded { + t.Fatalf("Status = %q, want %q", loaded.Status, RunManifestStatusSucceeded) + } + if loaded.Stages["prepare"] == nil || loaded.Stages["prepare"].Status != StatusSucceeded { + t.Fatalf("prepare stage = %#v, want succeeded", loaded.Stages["prepare"]) + } + if loaded.Stages["prepare"].Action != RunStageActionRun { + t.Fatalf("prepare action = %q, want %q", loaded.Stages["prepare"].Action, RunStageActionRun) + } +} + +func TestLoadRunRejectsInvalidManifest(t *testing.T) { + store := &LocalStore{} + ctx := context.Background() + + path := filepath.Join(t.TempDir(), "run-manifest.json") + if err := os.WriteFile(path, []byte(`{"session_id":"2026-05-03"}`), 0o644); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + + _, err := store.LoadRun(ctx, path) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "run_id is required") { + t.Fatalf("error = %q, want run_id validation", err.Error()) + } +}