Add remote session loading

2026-05-20 20:55:13 -05:00
parent b29d8eeb50
commit 3aae4bbb12
23 changed files with 587 additions and 101 deletions
--- a/internal/app/config_loader.go
+++ b/internal/app/config_loader.go
@@ -0,0 +1,133 @@
+package app
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+)
+
+func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, sessionFlag string, sessionOpts config.SessionLoadOptions) (*config.Config, error) {
+	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelineFlag)
+	if err != nil {
+		return nil, err
+	}
+	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignFlag)
+	if err != nil {
+		return nil, err
+	}
+
+	if explicitSession := strings.TrimSpace(sessionFlag); explicitSession != "" {
+		return config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, explicitSession, sessionOpts)
+	}
+
+	discoveredSession, err := discoverSessionConfigPathWithCandidates(config.DefaultSessionConfigSearchPaths)
+	if err != nil {
+		return nil, err
+	}
+	if discoveredSession.Path != "" {
+		return config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, discoveredSession.Path, sessionOpts)
+	}
+
+	pipelineCfg, err := config.LoadPipeline(resolvedPipelinePath)
+	if err != nil {
+		return nil, err
+	}
+	campaignCfg, err := config.LoadCampaign(resolvedCampaignPath)
+	if err != nil {
+		return nil, err
+	}
+
+	sessionID := strings.TrimSpace(sessionOpts.SessionID)
+	if sessionID == "" {
+		return nil, missingSessionConfigError(discoveredSession.Searched, "remote session loading requires --session-id")
+	}
+
+	sessionPrefix := artifacts.S3SessionPrefix(pipelineCfg.Storage.S3.RootPrefix, campaignCfg.Campaign, sessionID)
+	remoteKey := artifacts.S3SessionConfigKey(sessionPrefix)
+	partialCfg := &config.Config{
+		Pipeline:     pipelineCfg,
+		Campaign:     campaignCfg,
+		PipelinePath: resolvedPipelinePath,
+		CampaignPath: resolvedCampaignPath,
+	}
+	store, err := newObjectStoreFromConfigFn(ctx, partialCfg)
+	if err != nil {
+		return nil, missingSessionConfigError(discoveredSession.Searched, fmt.Sprintf("remote session %q unavailable: %v", remoteKey, err))
+	}
+
+	sessionInfo, err := findRemoteSessionConfig(ctx, store, sessionPrefix, remoteKey)
+	if err != nil {
+		return nil, missingSessionConfigError(discoveredSession.Searched, err.Error())
+	}
+	sessionTempPath, err := downloadRemoteSessionConfig(ctx, store, remoteKey)
+	if err != nil {
+		return nil, missingSessionConfigError(discoveredSession.Searched, fmt.Sprintf("remote session %q download failed: %v", remoteKey, err))
+	}
+	sessionBytes, err := os.ReadFile(sessionTempPath)
+	if err != nil {
+		return nil, fmt.Errorf("read downloaded remote session %q: %w", sessionTempPath, err)
+	}
+	sessionCfg, err := config.LoadSessionBytesWithOptions("s3://"+s3BucketName(pipelineCfg)+"/"+remoteKey, sessionBytes, sessionOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	return config.Resolve(
+		resolvedPipelinePath,
+		pipelineCfg,
+		resolvedCampaignPath,
+		campaignCfg,
+		sessionTempPath,
+		sessionCfg,
+		config.SessionSource{
+			Source:    "session_config.s3",
+			LocalPath: sessionTempPath,
+			S3Bucket:  s3BucketName(pipelineCfg),
+			S3Key:     remoteKey,
+			S3Size:    sessionInfo.Size,
+			S3ETag:    sessionInfo.ETag,
+			SpoolPath: sessionTempPath,
+		},
+	)
+}
+
+func findRemoteSessionConfig(ctx context.Context, store storage.ObjectStore, sessionPrefix, remoteKey string) (storage.ObjectInfo, error) {
+	objects, err := store.List(ctx, sessionPrefix)
+	if err != nil {
+		return storage.ObjectInfo{}, fmt.Errorf("remote session %q list failed: %w", remoteKey, err)
+	}
+	for _, obj := range objects {
+		if obj.Key == remoteKey {
+			return obj, nil
+		}
+	}
+	return storage.ObjectInfo{}, fmt.Errorf("remote session %q not found", remoteKey)
+}
+
+func downloadRemoteSessionConfig(ctx context.Context, store storage.ObjectStore, remoteKey string) (string, error) {
+	f, err := os.CreateTemp("", "narratio-session-*.yml")
+	if err != nil {
+		return "", fmt.Errorf("create temp file: %w", err)
+	}
+	path := f.Name()
+	if err := f.Close(); err != nil {
+		return "", fmt.Errorf("close temp file %q: %w", path, err)
+	}
+	if err := store.Download(ctx, remoteKey, path); err != nil {
+		return "", err
+	}
+	return filepath.Clean(path), nil
+}
+
+func s3BucketName(cfg *config.PipelineConfig) string {
+	if cfg == nil || cfg.Storage.S3 == nil {
+		return ""
+	}
+	return strings.TrimSpace(cfg.Storage.S3.Bucket)
+}
--- a/internal/app/plan.go
+++ b/internal/app/plan.go
@@ -38,20 +38,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
 	if fs.NArg() != 0 {
 		return fmt.Errorf("plan: unexpected positional arguments")
 	}
-	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
-	if err != nil {
-		return fmt.Errorf("plan: %w", err)
-	}
-	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath)
-	if err != nil {
-		return fmt.Errorf("plan: %w", err)
-	}
-	resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
-	if err != nil {
-		return fmt.Errorf("plan: %w", err)
-	}
-
-	cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{
+	cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{
 		SessionID:         sessionID,
 		PreviousSessionID: previousSessionID,
 	})
--- a/internal/app/remote_session_test.go
+++ b/internal/app/remote_session_test.go
@@ -0,0 +1,204 @@
+package app
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+)
+
+func TestExecuteRemoteSessionFallbackLoadsFromObjectStore(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
+	fake := &storage.FakeBackend{}
+	remoteKey := seedRemoteSessionConfig(t, fake, "2026-05-03", `session_id: "{{ session_id }}"
+inputs:
+  audio_s3:
+    prefix: audio/
+`)
+	var storeInitCalls int
+	restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+	if storeInitCalls != 1 {
+		t.Fatalf("object store init calls = %d, want 1", storeInitCalls)
+	}
+	if !strings.Contains(stdout.String(), "narratio plan: workdir prepared") {
+		t.Fatalf("stdout = %q, want plan output", stdout.String())
+	}
+	if _, ok := fake.Objects[remoteKey]; !ok {
+		t.Fatalf("remote session key %q was not seeded", remoteKey)
+	}
+}
+
+func TestExecuteExplicitLocalSessionPrecedenceSkipsRemote(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+	fake := &storage.FakeBackend{}
+	var storeInitCalls int
+	restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+	if storeInitCalls != 0 {
+		t.Fatalf("object store init calls = %d, want 0", storeInitCalls)
+	}
+}
+
+func TestExecuteLocalSessionDiscoveryPrecedenceSkipsRemote(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
+	fake := &storage.FakeBackend{}
+	var storeInitCalls int
+	restoreAppConfigTestGlobals(t, fake, &storeInitCalls, config.DefaultSessionConfigSearchPaths)
+
+	originalWD, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("Getwd(): %v", err)
+	}
+	if err := os.Chdir(filepath.Dir(sessionPath)); err != nil {
+		t.Fatalf("Chdir(%q): %v", filepath.Dir(sessionPath), err)
+	}
+	t.Cleanup(func() { _ = os.Chdir(originalWD) })
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code != 0 {
+		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
+	}
+	if storeInitCalls != 0 {
+		t.Fatalf("object store init calls = %d, want 0", storeInitCalls)
+	}
+}
+
+func TestExecuteRemoteSessionMissingObjectFailsClearly(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
+	fake := &storage.FakeBackend{}
+	var storeInitCalls int
+	missingSessionPath := filepath.Join(t.TempDir(), "session.yml")
+	restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{missingSessionPath})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "remote session") || !strings.Contains(stderr.String(), "session.yml") || !strings.Contains(stderr.String(), "not found") {
+		t.Fatalf("stderr = %q, want remote session not found context", stderr.String())
+	}
+	if !strings.Contains(stderr.String(), missingSessionPath) {
+		t.Fatalf("stderr = %q, want local searched path", stderr.String())
+	}
+}
+
+func TestExecuteRemoteSessionRequiresSessionID(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
+	var storeInitCalls int
+	restoreAppConfigTestGlobals(t, &storage.FakeBackend{}, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "remote session loading requires --session-id") {
+		t.Fatalf("stderr = %q, want session-id guidance", stderr.String())
+	}
+	if storeInitCalls != 0 {
+		t.Fatalf("object store init calls = %d, want 0", storeInitCalls)
+	}
+}
+
+func TestExecuteRemoteSessionStorageInitErrorFailsClearly(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
+	origStoreFn := newObjectStoreFromConfigFn
+	origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...)
+	config.DefaultSessionConfigSearchPaths = []string{filepath.Join(t.TempDir(), "session.yml")}
+	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
+		return nil, errors.New("storage unavailable")
+	}
+	t.Cleanup(func() {
+		newObjectStoreFromConfigFn = origStoreFn
+		config.DefaultSessionConfigSearchPaths = origSessionDefaults
+	})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "storage unavailable") || !strings.Contains(stderr.String(), "remote session") {
+		t.Fatalf("stderr = %q, want remote storage context", stderr.String())
+	}
+}
+
+func TestExecuteRemoteSessionMalformedYAMLFailsStrictDecode(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
+	fake := &storage.FakeBackend{}
+	seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-03\nunknown: true\n")
+	var storeInitCalls int
+	restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")})
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
+	if code == 0 {
+		t.Fatal("exit code = 0, want non-zero")
+	}
+	if !strings.Contains(stderr.String(), "strict decode failed") {
+		t.Fatalf("stderr = %q, want strict decode context", stderr.String())
+	}
+}
+
+func restoreAppConfigTestGlobals(t *testing.T, fake *storage.FakeBackend, storeInitCalls *int, sessionDefaults []string) {
+	t.Helper()
+	origStoreFn := newObjectStoreFromConfigFn
+	origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...)
+	config.DefaultSessionConfigSearchPaths = append([]string(nil), sessionDefaults...)
+	newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
+		if storeInitCalls != nil {
+			(*storeInitCalls)++
+		}
+		return fake, nil
+	}
+	t.Cleanup(func() {
+		newObjectStoreFromConfigFn = origStoreFn
+		config.DefaultSessionConfigSearchPaths = origSessionDefaults
+	})
+}
+
+func seedRemoteSessionConfig(t *testing.T, fake *storage.FakeBackend, sessionID, content string) string {
+	t.Helper()
+	sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", sessionID)
+	remoteKey := artifacts.S3SessionConfigKey(sessionPrefix)
+	fake.SeedObject(storage.FakeObject{
+		Key:  remoteKey,
+		Data: []byte(content),
+		ETag: "remote-session-etag",
+	})
+	return remoteKey
+}
--- a/internal/app/restore.go
+++ b/internal/app/restore.go
@@ -57,20 +57,7 @@ func Restore(ctx context.Context, args []string, out io.Writer) error {
 	if fs.NArg() != 0 {
 		return fmt.Errorf("restore: unexpected positional arguments")
 	}
-	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
-	if err != nil {
-		return fmt.Errorf("restore: %w", err)
-	}
-	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath)
-	if err != nil {
-		return fmt.Errorf("restore: %w", err)
-	}
-	resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
-	if err != nil {
-		return fmt.Errorf("restore: %w", err)
-	}
-
-	cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{
+	cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{
 		SessionID:         sessionID,
 		PreviousSessionID: previousSessionID,
 	})
--- a/internal/app/resume.go
+++ b/internal/app/resume.go
@@ -37,20 +37,7 @@ func Resume(ctx context.Context, args []string, out io.Writer) error {
 	if fs.NArg() != 0 {
 		return fmt.Errorf("resume: unexpected positional arguments")
 	}
-	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
-	if err != nil {
-		return fmt.Errorf("resume: %w", err)
-	}
-	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath)
-	if err != nil {
-		return fmt.Errorf("resume: %w", err)
-	}
-	resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
-	if err != nil {
-		return fmt.Errorf("resume: %w", err)
-	}
-
-	cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{
+	cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{
 		SessionID:         sessionID,
 		PreviousSessionID: previousSessionID,
 	})
--- a/internal/app/run.go
+++ b/internal/app/run.go
@@ -35,20 +35,7 @@ func Run(ctx context.Context, args []string, out io.Writer) error {
 	if fs.NArg() != 0 {
 		return fmt.Errorf("run: unexpected positional arguments")
 	}
-	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
-	if err != nil {
-		return fmt.Errorf("run: %w", err)
-	}
-	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath)
-	if err != nil {
-		return fmt.Errorf("run: %w", err)
-	}
-	resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
-	if err != nil {
-		return fmt.Errorf("run: %w", err)
-	}
-
-	cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{
+	cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{
 		SessionID:         sessionID,
 		PreviousSessionID: previousSessionID,
 	})
--- a/internal/app/run_stage.go
+++ b/internal/app/run_stage.go
@@ -48,20 +48,7 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
 		return fmt.Errorf("run-stage: %w", err)
 	}

-	resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
-	if err != nil {
-		return fmt.Errorf("run-stage: %w", err)
-	}
-	resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath)
-	if err != nil {
-		return fmt.Errorf("run-stage: %w", err)
-	}
-	resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
-	if err != nil {
-		return fmt.Errorf("run-stage: %w", err)
-	}
-
-	cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{
+	cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{
 		SessionID:         sessionID,
 		PreviousSessionID: previousSessionID,
 	})
--- a/internal/app/session_config_path.go
+++ b/internal/app/session_config_path.go
@@ -19,6 +19,22 @@ func resolveSessionConfigPathWithCandidates(flagValue string, candidates []strin
 		return explicit, nil
 	}

+	resolved, err := discoverSessionConfigPathWithCandidates(candidates)
+	if err != nil {
+		return "", err
+	}
+	if resolved.Path != "" {
+		return resolved.Path, nil
+	}
+	return "", missingSessionConfigError(resolved.Searched, "")
+}
+
+type sessionConfigDiscovery struct {
+	Path     string
+	Searched []string
+}
+
+func discoverSessionConfigPathWithCandidates(candidates []string) (sessionConfigDiscovery, error) {
 	ordered := make([]string, 0, len(candidates))
 	for _, raw := range candidates {
 		path := strings.TrimSpace(raw)
@@ -31,19 +47,32 @@ func resolveSessionConfigPathWithCandidates(flagValue string, candidates []strin
 			if info.IsDir() {
 				continue
 			}
-			return filepath.Clean(path), nil
+			return sessionConfigDiscovery{Path: filepath.Clean(path), Searched: ordered}, nil
 		}
 		if errors.Is(err, os.ErrNotExist) {
 			continue
 		}
-		return "", fmt.Errorf("check default session config %q: %w", path, err)
+		return sessionConfigDiscovery{}, fmt.Errorf("check default session config %q: %w", path, err)
 	}

+	return sessionConfigDiscovery{Searched: ordered}, nil
+}
+
+func missingSessionConfigError(searched []string, remoteDetail string) error {
+	ordered := append([]string(nil), searched...)
 	if len(ordered) == 0 {
-		return "", fmt.Errorf("no session config path provided and no default locations configured")
+		if strings.TrimSpace(remoteDetail) != "" {
+			return fmt.Errorf("no session config path provided and no default locations configured; %s", remoteDetail)
+		}
+		return fmt.Errorf("no session config path provided and no default locations configured")
 	}
-	return "", fmt.Errorf(
-		"no session config path provided and no default session config found; searched: %s; pass --session to use an explicit path",
+	msg := fmt.Sprintf(
+		"no session config path provided and no default session config found; searched: %s",
 		strings.Join(ordered, ", "),
 	)
+	if strings.TrimSpace(remoteDetail) != "" {
+		msg += "; " + strings.TrimSpace(remoteDetail)
+	}
+	msg += "; pass --session to use an explicit path"
+	return fmt.Errorf("%s", msg)
 }