Implement manifest model and local store

This commit is contained in:
2026-05-02 10:51:23 -05:00
parent 4172849f20
commit 6caecb1baa
7 changed files with 652 additions and 28 deletions

View File

@@ -2,14 +2,19 @@ package app
import (
"bytes"
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
func TestExecuteValidCommands(t *testing.T) {
pipelinePath, sessionPath := writeValidConfigFiles(t)
manifestPath := writeManifestPathForExecute(t)
cases := []struct {
name string
@@ -18,7 +23,7 @@ func TestExecuteValidCommands(t *testing.T) {
}{
{name: "run", args: []string{"run", "--config", pipelinePath, "--session", sessionPath}, wantOut: "narratio run: configuration loaded and valid"},
{name: "plan", args: []string{"plan", "--config", pipelinePath, "--session", sessionPath}, wantOut: "narratio plan: configuration loaded and valid"},
{name: "status", args: []string{"status"}, wantOut: "narratio status: not yet implemented"},
{name: "status", args: []string{"status", "--manifest", manifestPath}, wantOut: "session_id: 2026-05-03"},
{name: "resume", args: []string{"resume"}, wantOut: "narratio resume: not yet implemented"},
{name: "run-stage", args: []string{"run-stage", "polish"}, wantOut: "narratio run-stage: not yet implemented"},
}
@@ -42,7 +47,7 @@ func TestExecuteValidCommands(t *testing.T) {
}
}
func TestExecuteMissingConfigFlags(t *testing.T) {
func TestExecuteMissingRequiredFlags(t *testing.T) {
cases := []struct {
name string
args []string
@@ -50,6 +55,7 @@ func TestExecuteMissingConfigFlags(t *testing.T) {
}{
{name: "run missing flags", args: []string{"run"}, want: "run: --config and --session are required"},
{name: "plan missing flags", args: []string{"plan"}, want: "plan: --config and --session are required"},
{name: "status missing flags", args: []string{"status"}, want: "status: --manifest is required"},
}
for _, tc := range cases {
@@ -149,3 +155,18 @@ inputs:
return pipelinePath, sessionPath
}
func writeManifestPathForExecute(t *testing.T) string {
t.Helper()
store := &manifest.LocalStore{}
m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
m.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
path := filepath.Join(t.TempDir(), "manifest.json")
if err := store.Save(context.Background(), path, m); err != nil {
t.Fatalf("save manifest: %v", err)
}
return path
}

View File

@@ -2,10 +2,66 @@ package app
import (
"context"
"flag"
"fmt"
"io"
"sort"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
// Status is a placeholder for future manifest status inspection behavior.
func Status(_ context.Context, _ []string, out io.Writer) error {
return placeholder(out, "status")
// Status reads and prints stage statuses from an existing manifest.
func Status(ctx context.Context, args []string, out io.Writer) error {
fs := flag.NewFlagSet("status", flag.ContinueOnError)
fs.SetOutput(io.Discard)
var manifestPath string
fs.StringVar(&manifestPath, "manifest", "", "path to manifest.json")
if err := fs.Parse(args); err != nil {
return fmt.Errorf("status: invalid flags: %w", err)
}
if fs.NArg() != 0 {
return fmt.Errorf("status: unexpected positional arguments")
}
if manifestPath == "" {
return fmt.Errorf("status: --manifest is required")
}
store := &manifest.LocalStore{}
m, err := store.Load(ctx, manifestPath)
if err != nil {
return fmt.Errorf("status: %w", err)
}
if _, err := fmt.Fprintf(out, "session_id: %s\n", m.SessionID); err != nil {
return err
}
if _, err := fmt.Fprintf(out, "updated_at: %s\n", m.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z07:00")); err != nil {
return err
}
if len(m.Stages) == 0 {
_, err := fmt.Fprintln(out, "stages: no stages recorded")
return err
}
if _, err := fmt.Fprintln(out, "stages:"); err != nil {
return err
}
names := make([]string, 0, len(m.Stages))
for name := range m.Stages {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
status := m.Stages[name].Status
if _, err := fmt.Fprintf(out, "- %s: %s\n", name, status); err != nil {
return err
}
}
return nil
}

View File

@@ -0,0 +1,73 @@
package app
import (
"bytes"
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
func TestStatusCommandReadsManifest(t *testing.T) {
manifestPath := writeManifestForStatus(t)
var out bytes.Buffer
err := Status(context.Background(), []string{"--manifest", manifestPath}, &out)
if err != nil {
t.Fatalf("Status() error = %v", err)
}
s := out.String()
if !strings.Contains(s, "session_id: 2026-05-03") {
t.Fatalf("output = %q, want session_id", s)
}
if !strings.Contains(s, "- merge: succeeded") {
t.Fatalf("output = %q, want stage status", s)
}
}
func TestStatusCommandMissingManifestFlag(t *testing.T) {
var out bytes.Buffer
err := Status(context.Background(), nil, &out)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "--manifest is required") {
t.Fatalf("error = %q, want missing manifest flag", err.Error())
}
}
func TestStatusCommandBadManifest(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "manifest.json")
if err := os.WriteFile(path, []byte("{not-json"), 0o644); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
var out bytes.Buffer
err := Status(context.Background(), []string{"--manifest", path}, &out)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "decode manifest") {
t.Fatalf("error = %q, want decode error", err.Error())
}
}
func writeManifestForStatus(t *testing.T) string {
t.Helper()
store := &manifest.LocalStore{}
m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
m.MarkStageSucceeded("merge", time.Date(2026, 5, 3, 10, 5, 0, 0, time.UTC), nil)
path := filepath.Join(t.TempDir(), "manifest.json")
if err := store.Save(context.Background(), path, m); err != nil {
t.Fatalf("Save() error = %v", err)
}
return path
}

View File

@@ -1,18 +1,140 @@
package manifest
import "time"
import (
"strings"
"time"
)
// StageState tracks status and metadata for one stage execution.
type StageState struct {
Status StageStatus
UpdatedAt time.Time
Error string
// ErrorRecord captures structured error metadata at run or stage scope.
type ErrorRecord struct {
Message string `json:"message"`
Code string `json:"code,omitempty"`
At *time.Time `json:"at,omitempty"`
}
// Manifest is the durable state record for a session run.
// InputRecord captures one resolved input and optional checksum.
type InputRecord struct {
Kind string `json:"kind"`
Path string `json:"path"`
Checksum string `json:"checksum,omitempty"`
}
// ArtifactRecord captures one produced artifact and optional remote metadata.
type ArtifactRecord struct {
Kind string `json:"kind"`
LocalPath string `json:"local_path"`
RemoteKey string `json:"remote_key,omitempty"`
Checksum string `json:"checksum,omitempty"`
}
// StageRecord tracks lifecycle and provenance for one pipeline stage.
type StageRecord struct {
Name string `json:"name"`
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"`
}
// Manifest is the durable run-state record for a session execution.
type Manifest struct {
SessionID string
CreatedAt time.Time
UpdatedAt time.Time
Stages map[string]StageState
SessionID string `json:"session_id"`
PipelineVersion string `json:"pipeline_version,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
LastError *ErrorRecord `json:"last_error,omitempty"`
Inputs []InputRecord `json:"inputs,omitempty"`
Artifacts []ArtifactRecord `json:"artifacts,omitempty"`
Stages map[string]*StageRecord `json:"stages"`
}
// New constructs a new manifest with deterministic timestamps.
func New(sessionID string, now time.Time) *Manifest {
return &Manifest{
SessionID: strings.TrimSpace(sessionID),
CreatedAt: now,
UpdatedAt: now,
Stages: map[string]*StageRecord{},
}
}
// MarkStageRunning marks a stage as running and updates timestamps.
func (m *Manifest) 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
}
// MarkStageSucceeded marks a stage as succeeded, stores outputs, and updates timestamps.
func (m *Manifest) 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
}
// MarkStageFailed marks a stage as failed and records error metadata.
func (m *Manifest) 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
}
// MarkStageSkipped marks a stage as skipped and records the skip reason.
func (m *Manifest) 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 *Manifest) ensureStage(name string, at time.Time) *StageRecord {
if m.Stages == nil {
m.Stages = map[string]*StageRecord{}
}
stageName := strings.TrimSpace(name)
s, ok := m.Stages[stageName]
if !ok || s == nil {
s = &StageRecord{
Name: stageName,
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
}
func timePtr(t time.Time) *time.Time {
v := t
return &v
}

View File

@@ -0,0 +1,66 @@
package manifest
import (
"testing"
"time"
)
func TestStageMarkHelpers(t *testing.T) {
m := New("2026-05-03", time.Date(2026, 5, 3, 12, 0, 0, 0, time.UTC))
runningAt := time.Date(2026, 5, 3, 12, 1, 0, 0, time.UTC)
m.MarkStageRunning("transcribe", runningAt)
stage := m.Stages["transcribe"]
if stage == nil {
t.Fatal("stage not created")
}
if stage.Status != StatusRunning {
t.Fatalf("status = %q, want %q", stage.Status, StatusRunning)
}
if stage.StartedAt == nil || !stage.StartedAt.Equal(runningAt) {
t.Fatalf("started_at = %v, want %v", stage.StartedAt, runningAt)
}
succeededAt := runningAt.Add(2 * time.Minute)
outputs := []ArtifactRecord{{Kind: "processed_transcript", LocalPath: "transcripts/processed.json"}}
m.MarkStageSucceeded("transcribe", succeededAt, outputs)
if stage.Status != StatusSucceeded {
t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded)
}
if stage.CompletedAt == nil || !stage.CompletedAt.Equal(succeededAt) {
t.Fatalf("completed_at = %v, want %v", stage.CompletedAt, succeededAt)
}
if len(stage.Outputs) != 1 {
t.Fatalf("outputs len = %d, want 1", len(stage.Outputs))
}
failedAt := succeededAt.Add(1 * time.Minute)
m.MarkStageFailed("analyze", failedAt, "analyzer crashed")
failed := m.Stages["analyze"]
if failed == nil {
t.Fatal("failed stage missing")
}
if failed.Status != StatusFailed {
t.Fatalf("status = %q, want %q", failed.Status, StatusFailed)
}
if failed.Error == nil || failed.Error.Message != "analyzer crashed" {
t.Fatalf("error = %#v, want message", failed.Error)
}
skippedAt := failedAt.Add(1 * time.Minute)
m.MarkStageSkipped("notify", skippedAt, "notifications disabled")
skipped := m.Stages["notify"]
if skipped == nil {
t.Fatal("skipped stage missing")
}
if skipped.Status != StatusSkipped {
t.Fatalf("status = %q, want %q", skipped.Status, StatusSkipped)
}
if skipped.Error == nil || skipped.Error.Message != "notifications disabled" {
t.Fatalf("error = %#v, want skip reason", skipped.Error)
}
if skipped.Error.Code != "skipped" {
t.Fatalf("error code = %q, want %q", skipped.Error.Code, "skipped")
}
}

View File

@@ -2,26 +2,173 @@ package manifest
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"time"
)
// Store is a placeholder interface for manifest persistence.
// Store persists manifests to and from durable storage.
type Store interface {
Load(ctx context.Context, sessionID string) (*Manifest, error)
Save(ctx context.Context, m *Manifest) error
Create(ctx context.Context, sessionID string) (*Manifest, error)
Load(ctx context.Context, path string) (*Manifest, error)
Save(ctx context.Context, path string, m *Manifest) error
}
// LocalStore is a placeholder local-filesystem manifest store.
type LocalStore struct {
RootDir string
// LocalStore stores manifests as JSON on the local filesystem.
type LocalStore struct{}
// Create returns a new in-memory manifest for a session.
func (s *LocalStore) Create(ctx context.Context, sessionID string) (*Manifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(sessionID) == "" {
return nil, fmt.Errorf("create manifest: sessionID is required")
}
now := time.Now().UTC()
return New(sessionID, now), nil
}
// Load returns a not-yet-implemented error in the scaffold.
func (s *LocalStore) Load(_ context.Context, _ string) (*Manifest, error) {
return nil, fmt.Errorf("manifest local load: not yet implemented")
// Load reads and validates a local JSON manifest from path.
func (s *LocalStore) Load(ctx context.Context, path string) (*Manifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(path) == "" {
return nil, fmt.Errorf("load manifest: path is required")
}
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("load manifest %q: %w", path, err)
}
var m Manifest
if err := json.Unmarshal(data, &m); err != nil {
return nil, fmt.Errorf("decode manifest %q: %w", path, err)
}
if err := validateLoadedManifest(&m); err != nil {
return nil, fmt.Errorf("manifest %q invalid: %w", path, err)
}
normalizeManifest(&m)
return &m, nil
}
// Save returns a not-yet-implemented error in the scaffold.
func (s *LocalStore) Save(_ context.Context, _ *Manifest) error {
return fmt.Errorf("manifest local save: not yet implemented")
// Save writes the manifest to path atomically via temp file + rename.
func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
if err := checkContext(ctx); err != nil {
return err
}
if strings.TrimSpace(path) == "" {
return fmt.Errorf("save manifest: path is required")
}
if m == nil {
return fmt.Errorf("save manifest: manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("save manifest: session_id is required")
}
if m.CreatedAt.IsZero() {
return fmt.Errorf("save manifest: created_at is required")
}
m.UpdatedAt = time.Now().UTC()
if m.Stages == nil {
m.Stages = map[string]*StageRecord{}
}
data, err := json.MarshalIndent(m, "", " ")
if err != nil {
return fmt.Errorf("save manifest: marshal: %w", err)
}
data = append(data, '\n')
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("save manifest: create directory %q: %w", dir, err)
}
tmp, err := os.CreateTemp(dir, ".manifest.json.tmp-*")
if err != nil {
return fmt.Errorf("save manifest: 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("save manifest: write temp file: %w", err)
}
if err := tmp.Sync(); err != nil {
_ = tmp.Close()
return fmt.Errorf("save manifest: sync temp file: %w", err)
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("save manifest: close temp file: %w", err)
}
if err := checkContext(ctx); err != nil {
return err
}
if err := os.Rename(tmpName, path); err != nil {
return fmt.Errorf("save manifest: rename temp file: %w", err)
}
removeTmp = false
return nil
}
func validateLoadedManifest(m *Manifest) error {
if m == nil {
return fmt.Errorf("manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("session_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 normalizeManifest(m *Manifest) {
if m.Stages == nil {
m.Stages = map[string]*StageRecord{}
}
for name, stage := range m.Stages {
if stage == nil {
stage = &StageRecord{Name: name, Status: StatusPending, CreatedAt: m.CreatedAt, UpdatedAt: m.UpdatedAt}
m.Stages[name] = stage
}
if stage.Name == "" {
stage.Name = name
}
}
}
func checkContext(ctx context.Context) error {
if ctx == nil {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
default:
return nil
}
}

View File

@@ -0,0 +1,139 @@
package manifest
import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func TestLocalStoreCreateSaveLoadRoundTrip(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
m, err := store.Create(ctx, "2026-05-03")
if err != nil {
t.Fatalf("Create() error = %v", err)
}
now := time.Date(2026, 5, 3, 12, 0, 0, 0, time.UTC)
m.MarkStageRunning("prepare", now)
m.MarkStageSucceeded("prepare", now.Add(2*time.Second), []ArtifactRecord{{Kind: "transcript", LocalPath: "transcripts/merged.json"}})
path := filepath.Join(t.TempDir(), "manifest.json")
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("Save() error = %v", err)
}
loaded, err := store.Load(ctx, path)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if loaded.SessionID != "2026-05-03" {
t.Fatalf("SessionID = %q, want %q", loaded.SessionID, "2026-05-03")
}
stage, ok := loaded.Stages["prepare"]
if !ok {
t.Fatalf("stage prepare not found")
}
if stage.Status != StatusSucceeded {
t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded)
}
if len(stage.Outputs) != 1 {
t.Fatalf("outputs len = %d, want 1", len(stage.Outputs))
}
if loaded.UpdatedAt.IsZero() {
t.Fatal("UpdatedAt is zero")
}
}
func TestLocalStoreSaveUpdatesTimestamp(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
createdAt := time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)
m := New("2026-05-03", createdAt)
m.UpdatedAt = createdAt
path := filepath.Join(t.TempDir(), "manifest.json")
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("first Save() error = %v", err)
}
firstUpdated := m.UpdatedAt
time.Sleep(5 * time.Millisecond)
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("second Save() error = %v", err)
}
if !m.UpdatedAt.After(firstUpdated) {
t.Fatalf("UpdatedAt = %v, want after %v", m.UpdatedAt, firstUpdated)
}
}
func TestLocalStoreSaveAtomicPractical(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
dir := t.TempDir()
path := filepath.Join(dir, "manifest.json")
m := New("session-a", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("Save() error = %v", err)
}
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
var decoded Manifest
if err := json.Unmarshal(data, &decoded); err != nil {
t.Fatalf("saved manifest is not valid JSON: %v", err)
}
entries, err := os.ReadDir(dir)
if err != nil {
t.Fatalf("ReadDir() error = %v", err)
}
for _, e := range entries {
if strings.HasPrefix(e.Name(), ".manifest.json.tmp-") {
t.Fatalf("found unexpected temp file after save: %s", e.Name())
}
}
m.SessionID = "session-b"
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("second Save() error = %v", err)
}
loaded, err := store.Load(ctx, path)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if loaded.SessionID != "session-b" {
t.Fatalf("SessionID = %q, want %q", loaded.SessionID, "session-b")
}
}
func TestLoadRejectsInvalidManifest(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
path := filepath.Join(t.TempDir(), "manifest.json")
if err := os.WriteFile(path, []byte(`{"updated_at":"2026-05-03T10:00:00Z"}`), 0o644); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
_, err := store.Load(ctx, path)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "session_id is required") {
t.Fatalf("error = %q, want session_id validation", err.Error())
}
}