Add prepare helper to hydrate previous-session artifacts from archive
This commit is contained in:
475
internal/stage/prepare_previous.go
Normal file
475
internal/stage/prepare_previous.go
Normal file
@@ -0,0 +1,475 @@
|
|||||||
|
package stage
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
preparePreviousInputKindManifest = "previous_manifest"
|
||||||
|
preparePreviousInputKindArtifact = "previous_artifact"
|
||||||
|
preparePreviousInputSource = "previous_session_archive.current"
|
||||||
|
)
|
||||||
|
|
||||||
|
// previousSessionHydrationResult captures prepare-time previous-session cache materialization.
|
||||||
|
type previousSessionHydrationResult struct {
|
||||||
|
Inputs []manifest.InputRecord
|
||||||
|
Hydrated []string
|
||||||
|
SkippedMissing []string
|
||||||
|
PreviousRunID string
|
||||||
|
}
|
||||||
|
|
||||||
|
func hydratePreviousSessionArtifacts(
|
||||||
|
ctx context.Context,
|
||||||
|
env *Env,
|
||||||
|
paths artifacts.SessionPaths,
|
||||||
|
requirements []artifacts.PreviousArtifactRequirement,
|
||||||
|
) (*previousSessionHydrationResult, error) {
|
||||||
|
if len(requirements) == 0 {
|
||||||
|
return &previousSessionHydrationResult{}, nil
|
||||||
|
}
|
||||||
|
if env == nil || env.Config == nil || env.Config.Session == nil || env.Config.Pipeline == nil {
|
||||||
|
return nil, fmt.Errorf("resolved config with session/pipeline is required")
|
||||||
|
}
|
||||||
|
if env.ArtifactStore == nil {
|
||||||
|
return nil, fmt.Errorf("artifact store 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(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, ", "),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("checksum previous-session artifact %q: %w", requirement.Name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
result.Inputs = append(result.Inputs, manifest.InputRecord{
|
||||||
|
Kind: preparePreviousInputKindArtifact,
|
||||||
|
Path: localPath,
|
||||||
|
Checksum: checksum,
|
||||||
|
Source: preparePreviousInputSource,
|
||||||
|
S3Bucket: bucket,
|
||||||
|
S3Key: selectedKey,
|
||||||
|
})
|
||||||
|
result.Hydrated = append(result.Hydrated, requirement.Name)
|
||||||
|
}
|
||||||
|
|
||||||
|
sort.Strings(result.Hydrated)
|
||||||
|
sort.Strings(result.SkippedMissing)
|
||||||
|
sort.Slice(result.Inputs, func(i, j int) bool {
|
||||||
|
if result.Inputs[i].Kind != result.Inputs[j].Kind {
|
||||||
|
return result.Inputs[i].Kind < result.Inputs[j].Kind
|
||||||
|
}
|
||||||
|
return result.Inputs[i].Path < result.Inputs[j].Path
|
||||||
|
})
|
||||||
|
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
|
||||||
|
}
|
||||||
415
internal/stage/prepare_previous_test.go
Normal file
415
internal/stage/prepare_previous_test.go
Normal file
@@ -0,0 +1,415 @@
|
|||||||
|
package stage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsDownloadsManifestAndRequiredArtifact(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: true,
|
||||||
|
includeArtifactObject: true,
|
||||||
|
artifactBody: "# previous recap\n",
|
||||||
|
})
|
||||||
|
|
||||||
|
result, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("hydratePreviousSessionArtifacts() error = %v", err)
|
||||||
|
}
|
||||||
|
if result == nil {
|
||||||
|
t.Fatal("result is nil")
|
||||||
|
}
|
||||||
|
if len(result.Inputs) != 2 {
|
||||||
|
t.Fatalf("inputs len = %d, want 2", len(result.Inputs))
|
||||||
|
}
|
||||||
|
if !containsString(result.Hydrated, "session_recap") {
|
||||||
|
t.Fatalf("hydrated = %#v, want session_recap", result.Hydrated)
|
||||||
|
}
|
||||||
|
if len(result.SkippedMissing) != 0 {
|
||||||
|
t.Fatalf("skipped missing = %#v, want none", result.SkippedMissing)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := os.Stat(sessionPaths.PreviousManifestPath); err != nil {
|
||||||
|
t.Fatalf("previous manifest missing: %v", err)
|
||||||
|
}
|
||||||
|
recapPath := artifacts.SessionPreviousArtifactPath(sessionPaths, "artifacts/session_recap.md")
|
||||||
|
if _, err := os.Stat(recapPath); err != nil {
|
||||||
|
t.Fatalf("previous artifact missing: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
manifestInput := findInputByKind(result.Inputs, preparePreviousInputKindManifest)
|
||||||
|
if manifestInput == nil {
|
||||||
|
t.Fatalf("missing input kind %q", preparePreviousInputKindManifest)
|
||||||
|
}
|
||||||
|
if manifestInput.Source != preparePreviousInputSource {
|
||||||
|
t.Fatalf("manifest input source = %q, want %q", manifestInput.Source, preparePreviousInputSource)
|
||||||
|
}
|
||||||
|
|
||||||
|
artifactInput := findInputByKind(result.Inputs, preparePreviousInputKindArtifact)
|
||||||
|
if artifactInput == nil {
|
||||||
|
t.Fatalf("missing input kind %q", preparePreviousInputKindArtifact)
|
||||||
|
}
|
||||||
|
if artifactInput.Source != preparePreviousInputSource {
|
||||||
|
t.Fatalf("artifact input source = %q, want %q", artifactInput.Source, preparePreviousInputSource)
|
||||||
|
}
|
||||||
|
if artifactInput.Path != recapPath {
|
||||||
|
t.Fatalf("artifact input path = %q, want %q", artifactInput.Path, recapPath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsSkipsMissingOptionalArtifact(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: false},
|
||||||
|
}
|
||||||
|
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: true,
|
||||||
|
includeManifestObject: true,
|
||||||
|
includeArtifactObject: false,
|
||||||
|
})
|
||||||
|
|
||||||
|
result, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("hydratePreviousSessionArtifacts() error = %v", err)
|
||||||
|
}
|
||||||
|
if result == nil {
|
||||||
|
t.Fatal("result is nil")
|
||||||
|
}
|
||||||
|
if !containsString(result.SkippedMissing, "session_recap") {
|
||||||
|
t.Fatalf("skipped missing = %#v, want session_recap", result.SkippedMissing)
|
||||||
|
}
|
||||||
|
if len(result.Hydrated) != 0 {
|
||||||
|
t.Fatalf("hydrated = %#v, want none", result.Hydrated)
|
||||||
|
}
|
||||||
|
manifestInput := findInputByKind(result.Inputs, preparePreviousInputKindManifest)
|
||||||
|
if manifestInput == nil {
|
||||||
|
t.Fatalf("missing input kind %q", preparePreviousInputKindManifest)
|
||||||
|
}
|
||||||
|
if findInputByKind(result.Inputs, preparePreviousInputKindArtifact) != nil {
|
||||||
|
t.Fatalf("unexpected %q input for missing optional artifact", preparePreviousInputKindArtifact)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsFailsMissingRequiredArtifact(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: true,
|
||||||
|
includeArtifactObject: false,
|
||||||
|
})
|
||||||
|
|
||||||
|
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "required previous-session artifact") {
|
||||||
|
t.Fatalf("error = %v, want required artifact failure", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsFailsRequiredWhenPreviousSessionIDUnset(t *testing.T) {
|
||||||
|
env, sessionPaths, _ := previousHydrationFixture(t)
|
||||||
|
env.Config.Session.PreviousSessionID = ""
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "previous_session_id is required") {
|
||||||
|
t.Fatalf("error = %v, want previous_session_id required failure", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsOptionalWithNoPreviousSessionID(t *testing.T) {
|
||||||
|
env, sessionPaths, _ := previousHydrationFixture(t)
|
||||||
|
env.Config.Session.PreviousSessionID = ""
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: false},
|
||||||
|
}
|
||||||
|
|
||||||
|
result, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("hydratePreviousSessionArtifacts() error = %v", err)
|
||||||
|
}
|
||||||
|
if result == nil {
|
||||||
|
t.Fatal("result is nil")
|
||||||
|
}
|
||||||
|
if len(result.Inputs) != 0 {
|
||||||
|
t.Fatalf("inputs len = %d, want 0", len(result.Inputs))
|
||||||
|
}
|
||||||
|
if !containsString(result.SkippedMissing, "session_recap") {
|
||||||
|
t.Fatalf("skipped missing = %#v, want session_recap", result.SkippedMissing)
|
||||||
|
}
|
||||||
|
if _, err := os.Stat(sessionPaths.PreviousManifestPath); !os.IsNotExist(err) {
|
||||||
|
t.Fatalf("previous manifest should not be created, stat err = %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsRespectsCurrentCommitMarker(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: false,
|
||||||
|
includeArtifactObject: true,
|
||||||
|
artifactBody: "# previous recap\n",
|
||||||
|
})
|
||||||
|
|
||||||
|
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "remote current run pointer missing") {
|
||||||
|
t.Fatalf("error = %v, want current run pointer missing failure", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsDoesNotUseLocalPreviousWorkspaceState(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
seed := seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: true,
|
||||||
|
includeManifestObject: true,
|
||||||
|
includeArtifactObject: false,
|
||||||
|
})
|
||||||
|
// Write a local previous-session workspace file that should be ignored.
|
||||||
|
writeFile(t, filepath.Join(seed.PreviousSessionRoot, "artifacts", "session_recap.md"), "# local stale recap\n")
|
||||||
|
|
||||||
|
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "object missing from archive") {
|
||||||
|
t.Fatalf("error = %v, want remote-object-missing failure", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHydratePreviousSessionArtifactsUsesExplicitStorageKeys(t *testing.T) {
|
||||||
|
env, sessionPaths, fake := previousHydrationFixture(t)
|
||||||
|
requirements := []artifacts.PreviousArtifactRequirement{
|
||||||
|
{Name: "session_recap", Required: true},
|
||||||
|
}
|
||||||
|
seed := seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
|
||||||
|
includeRunPointerObject: true,
|
||||||
|
includeArtifactObject: true,
|
||||||
|
artifactBody: "# previous recap\n",
|
||||||
|
})
|
||||||
|
|
||||||
|
capture := &preparePreviousCaptureStore{delegate: fake}
|
||||||
|
env.ObjectStore = capture
|
||||||
|
|
||||||
|
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("hydratePreviousSessionArtifacts() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !containsString(capture.existsKeys, seed.RunPointerKey) {
|
||||||
|
t.Fatalf("exists keys = %#v, want %q", capture.existsKeys, seed.RunPointerKey)
|
||||||
|
}
|
||||||
|
if !containsString(capture.existsKeys, seed.ManifestKey) {
|
||||||
|
t.Fatalf("exists keys = %#v, want %q", capture.existsKeys, seed.ManifestKey)
|
||||||
|
}
|
||||||
|
if !containsString(capture.existsKeys, seed.ArtifactKey) {
|
||||||
|
t.Fatalf("exists keys = %#v, want %q", capture.existsKeys, seed.ArtifactKey)
|
||||||
|
}
|
||||||
|
if !containsString(capture.downloadKeys, seed.RunPointerKey) {
|
||||||
|
t.Fatalf("download keys = %#v, want %q", capture.downloadKeys, seed.RunPointerKey)
|
||||||
|
}
|
||||||
|
if !containsString(capture.downloadKeys, seed.ManifestKey) {
|
||||||
|
t.Fatalf("download keys = %#v, want %q", capture.downloadKeys, seed.ManifestKey)
|
||||||
|
}
|
||||||
|
if !containsString(capture.downloadKeys, seed.ArtifactKey) {
|
||||||
|
t.Fatalf("download keys = %#v, want %q", capture.downloadKeys, seed.ArtifactKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type previousStateSeedResult struct {
|
||||||
|
PreviousSessionPrefix string
|
||||||
|
PreviousSessionRoot string
|
||||||
|
ManifestKey string
|
||||||
|
RunPointerKey string
|
||||||
|
ArtifactKey string
|
||||||
|
}
|
||||||
|
|
||||||
|
type previousStateSeedOptions struct {
|
||||||
|
includeRunPointerObject bool
|
||||||
|
includeManifestObject bool
|
||||||
|
includeArtifactObject bool
|
||||||
|
artifactBody string
|
||||||
|
}
|
||||||
|
|
||||||
|
func seedPreviousCurrentState(
|
||||||
|
t *testing.T,
|
||||||
|
env *Env,
|
||||||
|
fake *storage.FakeBackend,
|
||||||
|
options previousStateSeedOptions,
|
||||||
|
) previousStateSeedResult {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
if !options.includeRunPointerObject && !options.includeManifestObject && !options.includeArtifactObject {
|
||||||
|
// Keep default behavior deterministic when caller omits explicit flags.
|
||||||
|
options.includeRunPointerObject = true
|
||||||
|
options.includeManifestObject = true
|
||||||
|
}
|
||||||
|
if options.includeManifestObject == false && options.includeArtifactObject {
|
||||||
|
options.includeManifestObject = true
|
||||||
|
}
|
||||||
|
|
||||||
|
previousSessionID := strings.TrimSpace(env.Config.Session.PreviousSessionID)
|
||||||
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
||||||
|
rootPrefix := strings.TrimSpace(env.Config.Pipeline.Storage.S3.RootPrefix)
|
||||||
|
previousSessionPrefix := artifacts.S3SessionPrefix(rootPrefix, campaign, previousSessionID)
|
||||||
|
manifestKey, runPointerKey := artifacts.ResolveArchiveCurrentStateKeys(previousSessionPrefix)
|
||||||
|
|
||||||
|
previousRunID := "20260510T010203Z-a1b2c3d4"
|
||||||
|
previousSessionRoot := filepath.Join(t.TempDir(), "work", campaign, previousSessionID)
|
||||||
|
artifactLocalPath := filepath.Join(previousSessionRoot, "artifacts", "session_recap.md")
|
||||||
|
previousManifest := buildPreviousManifestForSeed(
|
||||||
|
t,
|
||||||
|
previousSessionID,
|
||||||
|
campaign,
|
||||||
|
previousRunID,
|
||||||
|
filepath.Join(previousSessionRoot, "runs", previousRunID),
|
||||||
|
artifactLocalPath,
|
||||||
|
)
|
||||||
|
|
||||||
|
if options.includeRunPointerObject || (!options.includeRunPointerObject && !options.includeManifestObject && !options.includeArtifactObject) {
|
||||||
|
fake.SeedObject(storage.FakeObject{
|
||||||
|
Key: runPointerKey,
|
||||||
|
Data: []byte(previousRunID + "\n"),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
if options.includeManifestObject || options.includeArtifactObject {
|
||||||
|
fake.SeedObject(storage.FakeObject{
|
||||||
|
Key: manifestKey,
|
||||||
|
Data: previousManifest,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
artifactKey := artifacts.S3PromotedArtifactKey(previousSessionPrefix, "artifacts/session_recap.md")
|
||||||
|
if options.includeArtifactObject {
|
||||||
|
body := options.artifactBody
|
||||||
|
if body == "" {
|
||||||
|
body = "# previous recap\n"
|
||||||
|
}
|
||||||
|
fake.SeedObject(storage.FakeObject{
|
||||||
|
Key: artifactKey,
|
||||||
|
Data: []byte(body),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
return previousStateSeedResult{
|
||||||
|
PreviousSessionPrefix: previousSessionPrefix,
|
||||||
|
PreviousSessionRoot: previousSessionRoot,
|
||||||
|
ManifestKey: manifestKey,
|
||||||
|
RunPointerKey: runPointerKey,
|
||||||
|
ArtifactKey: artifactKey,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func previousHydrationFixture(t *testing.T) (*Env, artifacts.SessionPaths, *storage.FakeBackend) {
|
||||||
|
t.Helper()
|
||||||
|
env, m := setupPrepareEnv(t)
|
||||||
|
env.Config.Session.Campaign = "forsaken"
|
||||||
|
env.Config.Session.PreviousSessionID = "2026-05-10"
|
||||||
|
env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{
|
||||||
|
Bucket: "my-dnd-archive",
|
||||||
|
RootPrefix: "dnd",
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Scriptorium = &config.ScriptoriumConfig{
|
||||||
|
Artifacts: map[string]config.ScriptoriumArtifactConfig{
|
||||||
|
"session_recap": {
|
||||||
|
Enabled: true,
|
||||||
|
OutputPath: "artifacts/session_recap.md",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
fake := &storage.FakeBackend{}
|
||||||
|
env.ObjectStore = fake
|
||||||
|
|
||||||
|
paths, err := ensureLayoutForEnv(env, m.SessionID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ensureLayoutForEnv() error = %v", err)
|
||||||
|
}
|
||||||
|
return env, paths, fake
|
||||||
|
}
|
||||||
|
|
||||||
|
func buildPreviousManifestForSeed(
|
||||||
|
t *testing.T,
|
||||||
|
sessionID, campaign, runID, runRoot, artifactPath string,
|
||||||
|
) []byte {
|
||||||
|
t.Helper()
|
||||||
|
now := time.Date(2026, 5, 19, 22, 0, 0, 0, time.UTC)
|
||||||
|
m := manifest.New(sessionID, now)
|
||||||
|
m.Campaign = campaign
|
||||||
|
m.RunID = runID
|
||||||
|
m.LocalWorkDir = runRoot
|
||||||
|
m.MarkStageSucceeded("analyze", now, []manifest.ArtifactRecord{
|
||||||
|
{
|
||||||
|
Kind: "scriptorium_artifact",
|
||||||
|
SourceID: artifacts.ConfiguredArtifactSourceID("session_recap"),
|
||||||
|
LocalPath: artifactPath,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
m.MarkStageSucceeded("archive", now, nil)
|
||||||
|
m.Stages["archive"].Metadata = map[string]any{
|
||||||
|
"promoted_paths": []string{"artifacts/session_recap.md"},
|
||||||
|
}
|
||||||
|
data, err := json.MarshalIndent(m, "", " ")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal manifest: %v", err)
|
||||||
|
}
|
||||||
|
return append(data, '\n')
|
||||||
|
}
|
||||||
|
|
||||||
|
type preparePreviousCaptureStore struct {
|
||||||
|
delegate storage.ObjectStore
|
||||||
|
existsKeys []string
|
||||||
|
downloadKeys []string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *preparePreviousCaptureStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) {
|
||||||
|
return s.delegate.List(ctx, prefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *preparePreviousCaptureStore) Download(ctx context.Context, key, localPath string) error {
|
||||||
|
s.downloadKeys = append(s.downloadKeys, key)
|
||||||
|
return s.delegate.Download(ctx, key, localPath)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *preparePreviousCaptureStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
|
||||||
|
return s.delegate.Upload(ctx, localPath, key, opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *preparePreviousCaptureStore) Exists(ctx context.Context, key string) (bool, error) {
|
||||||
|
s.existsKeys = append(s.existsKeys, key)
|
||||||
|
return s.delegate.Exists(ctx, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
func findInputByKind(inputs []manifest.InputRecord, kind string) *manifest.InputRecord {
|
||||||
|
for i := range inputs {
|
||||||
|
if inputs[i].Kind == kind {
|
||||||
|
return &inputs[i]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func containsString(values []string, target string) bool {
|
||||||
|
for _, value := range values {
|
||||||
|
if value == target {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user