diff --git a/docs/adr/0006-separate-output-cache-and-debug-state.md b/docs/adr/0006-separate-output-cache-and-debug-state.md index a49afef..50fafa7 100644 --- a/docs/adr/0006-separate-output-cache-and-debug-state.md +++ b/docs/adr/0006-separate-output-cache-and-debug-state.md @@ -1,6 +1,6 @@ # ADR-0006: Separate output, cache, and debug state -**Status:** Proposed +**Status:** Accepted **Date:** 2026-07-17 ## Context diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index eebf665..7d99aa8 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -232,7 +232,7 @@ updating imports. ## Stage 1: Accept ADR-0006 and decouple cache storage internals -**Status:** Not started +**Status:** Complete ### Objective diff --git a/internal/cli/run.go b/internal/cli/run.go index cadadf8..f32ddce 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -432,7 +432,7 @@ func checkpointHandlersForRun( sessionID string, resume bool, ) (pipeline.CheckpointRecorder, pipeline.CheckpointLoader, error) { - identity, err := workspace.NewCheckpointIdentity(workspace.CheckpointIdentityInput{ + identity, err := checkpoint.NewIdentity(checkpoint.IdentityInput{ Pipeline: resolved, InputKey: resolved.Input.Module, RawInputDigest: rawInputDigest(rawInput), @@ -444,13 +444,17 @@ func checkpointHandlersForRun( if err != nil { return nil, nil, fmt.Errorf("create checkpoint identity: %w", err) } - recorder, err := checkpoint.NewWorkspaceRecorder(settings, identity) + checkpointRoot := "" + if settings.ResumeEnabled { + checkpointRoot = settings.CheckpointsRoot + } + recorder, err := checkpoint.NewFilesystemRecorder(checkpointRoot, identity) if err != nil { return nil, nil, fmt.Errorf("create checkpoint recorder: %w", err) } loader := pipeline.NoopCheckpointLoader() if resume { - loader, err = checkpoint.NewWorkspaceLoader(settings, identity) + loader, err = checkpoint.NewFilesystemLoader(checkpointRoot, identity) if err != nil { return nil, nil, fmt.Errorf("create checkpoint loader: %w", err) } @@ -463,28 +467,28 @@ func rawInputDigest(data []byte) string { return "sha256:" + hex.EncodeToString(sum[:]) } -func runtimeOverrideFingerprints(llmProfileOverride string, sessionID string) []workspace.Fingerprint { - var values []workspace.Fingerprint +func runtimeOverrideFingerprints(llmProfileOverride string, sessionID string) []checkpoint.Fingerprint { + var values []checkpoint.Fingerprint if strings.TrimSpace(llmProfileOverride) != "" { - values = append(values, workspace.Fingerprint{Name: "llm_profile_override", Value: strings.TrimSpace(llmProfileOverride)}) + values = append(values, checkpoint.Fingerprint{Name: "llm_profile_override", Value: strings.TrimSpace(llmProfileOverride)}) } if strings.TrimSpace(sessionID) != "" { - values = append(values, workspace.Fingerprint{Name: "session_id", Value: strings.TrimSpace(sessionID)}) + values = append(values, checkpoint.Fingerprint{Name: "session_id", Value: strings.TrimSpace(sessionID)}) } return values } -func llmProfileFingerprints(profiles []artifacts.LLMProfileManifest) []workspace.Fingerprint { +func llmProfileFingerprints(profiles []artifacts.LLMProfileManifest) []checkpoint.Fingerprint { if len(profiles) == 0 { return nil } - values := make([]workspace.Fingerprint, 0, len(profiles)) + values := make([]checkpoint.Fingerprint, 0, len(profiles)) for _, profile := range profiles { id := strings.TrimSpace(profile.ID) if id == "" { continue } - values = append(values, workspace.Fingerprint{ + values = append(values, checkpoint.Fingerprint{ Name: "llm_profile:" + id, Value: strings.TrimSpace(profile.Provider) + ":" + strings.TrimSpace(profile.Model), }) @@ -670,7 +674,7 @@ func chunkPlanStoreForRun(cfg config.WorkspaceChunkCacheConfig, opts Options) (p root := strings.TrimSpace(cfg.Directory) if root == "" { var err error - root, err = workspace.DefaultChunkPlanRoot(opts.UserCacheDir) + root, err = config.DefaultChunkPlanRoot(opts.UserCacheDir) if err != nil { return nil, fmt.Errorf("resolve chunk plan root: %w", err) } diff --git a/internal/core/workspace/chunk_plan.go b/internal/core/config/cache.go similarity index 85% rename from internal/core/workspace/chunk_plan.go rename to internal/core/config/cache.go index 4fc5342..efb8e0f 100644 --- a/internal/core/workspace/chunk_plan.go +++ b/internal/core/config/cache.go @@ -1,4 +1,4 @@ -package workspace +package config import ( "fmt" @@ -6,6 +6,7 @@ import ( "strings" ) +// DefaultChunkPlanRoot resolves the existing per-user chunk-plan cache root. func DefaultChunkPlanRoot(userCacheDir func() (string, error)) (string, error) { if userCacheDir == nil { return "", fmt.Errorf("user cache directory resolver must not be nil") diff --git a/internal/core/fileio/fileio.go b/internal/core/fileio/fileio.go new file mode 100644 index 0000000..f02b029 --- /dev/null +++ b/internal/core/fileio/fileio.go @@ -0,0 +1,102 @@ +// Package fileio provides confined, atomic artifact writes. +package fileio + +import ( + "encoding/json" + "fmt" + "os" + "path" + "path/filepath" + "strings" +) + +func SafePath(root, name string) (string, error) { + root = strings.TrimSpace(root) + if root == "" { + return "", fmt.Errorf("file root must not be empty") + } + name = strings.TrimSpace(name) + if name == "" { + return "", fmt.Errorf("artifact name must not be empty") + } + if strings.Contains(name, `\\`) { + return "", fmt.Errorf("artifact name %q must use slash-separated relative paths", name) + } + if path.IsAbs(name) || filepath.IsAbs(name) { + return "", fmt.Errorf("artifact name %q must be relative", name) + } + if name == "." || strings.Contains(name, "..") { + return "", fmt.Errorf("artifact name %q must not contain ..", name) + } + if path.Clean(name) != name { + return "", fmt.Errorf("artifact name %q must be clean", name) + } + absRoot, err := filepath.Abs(root) + if err != nil { + return "", fmt.Errorf("resolve file root %q: %w", root, err) + } + target, err := filepath.Abs(filepath.Join(absRoot, filepath.FromSlash(name))) + if err != nil { + return "", fmt.Errorf("resolve artifact %q: %w", name, err) + } + rel, err := filepath.Rel(absRoot, target) + if err != nil { + return "", fmt.Errorf("resolve artifact %q: %w", name, err) + } + if rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("artifact name %q resolves outside file root", name) + } + return target, nil +} + +func WriteJSON(root, name string, payload any, dirMode, fileMode os.FileMode) error { + data, err := json.MarshalIndent(payload, "", " ") + if err != nil { + return fmt.Errorf("marshal artifact %q: %w", name, err) + } + return WriteBytes(root, name, append(data, '\n'), dirMode, fileMode) +} + +func WriteBytes(root, name string, data []byte, dirMode, fileMode os.FileMode) error { + target, err := SafePath(root, name) + if err != nil { + return err + } + if err := writeAtomic(target, data, dirMode, fileMode); err != nil { + return fmt.Errorf("write artifact %q: %w", name, err) + } + return nil +} + +func writeAtomic(target string, data []byte, dirMode, fileMode os.FileMode) error { + if err := os.MkdirAll(filepath.Dir(target), dirMode); err != nil { + return err + } + temp, err := os.CreateTemp(filepath.Dir(target), "."+filepath.Base(target)+".tmp-*") + if err != nil { + return err + } + tempPath := temp.Name() + keep := true + defer func() { + if keep { + _ = os.Remove(tempPath) + } + }() + if _, err := temp.Write(data); err != nil { + _ = temp.Close() + return err + } + if err := temp.Chmod(fileMode); err != nil { + _ = temp.Close() + return err + } + if err := temp.Close(); err != nil { + return err + } + if err := os.Rename(tempPath, target); err != nil { + return err + } + keep = false + return nil +} diff --git a/internal/core/fileio/fileio_test.go b/internal/core/fileio/fileio_test.go new file mode 100644 index 0000000..7f2ba00 --- /dev/null +++ b/internal/core/fileio/fileio_test.go @@ -0,0 +1,41 @@ +package fileio + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestSafePathRejectsUnsafeNames(t *testing.T) { + for _, name := range []string{"/tmp/x", "a/../x", "a//x", `a\\x`} { + if _, err := SafePath(t.TempDir(), name); err == nil { + t.Fatalf("SafePath(%q) accepted unsafe path", name) + } + } +} + +func TestWriteBytesIsAtomicAndUsesRequestedModes(t *testing.T) { + root := t.TempDir() + if err := WriteBytes(root, "nested/value", []byte("value"), 0o700, 0o600); err != nil { + t.Fatal(err) + } + for path, want := range map[string]os.FileMode{filepath.Join(root, "nested"): 0o700, filepath.Join(root, "nested", "value"): 0o600} { + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != want { + t.Fatalf("%s mode=%#o want %#o", path, info.Mode().Perm(), want) + } + } + entries, err := os.ReadDir(filepath.Join(root, "nested")) + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + if strings.Contains(entry.Name(), ".tmp-") { + t.Fatalf("temporary file remains: %s", entry.Name()) + } + } +} diff --git a/internal/core/workspace/chunk_plan_test.go b/internal/core/workspace/chunk_plan_test.go deleted file mode 100644 index b806848..0000000 --- a/internal/core/workspace/chunk_plan_test.go +++ /dev/null @@ -1,57 +0,0 @@ -package workspace - -import ( - "errors" - "path/filepath" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/core/config" -) - -func TestDefaultChunkPlanRoot(t *testing.T) { - got, err := DefaultChunkPlanRoot(func() (string, error) { return "/cache/user", nil }) - if err != nil { - t.Fatalf("DefaultChunkPlanRoot() error = %v", err) - } - if want := filepath.Join("/cache/user", "notarius", "chunk-plans"); got != want { - t.Fatalf("root = %q, want %q", got, want) - } -} - -func TestDefaultChunkPlanRootRejectsResolverFailures(t *testing.T) { - boom := errors.New("resolver failed") - if _, err := DefaultChunkPlanRoot(func() (string, error) { return "", boom }); !errors.Is(err, boom) { - t.Fatalf("resolver error = %v", err) - } - if _, err := DefaultChunkPlanRoot(func() (string, error) { return " \t ", nil }); err == nil || !strings.Contains(err.Error(), "empty") { - t.Fatalf("empty result error = %v", err) - } - if _, err := DefaultChunkPlanRoot(nil); err == nil { - t.Fatal("nil resolver error = nil") - } -} - -func TestChunkPlanDirectoryIsIndependentFromWorkspaceSettings(t *testing.T) { - base := config.Default() - base.Workspace.Directory = "/workspace/one" - base.Workspace.ChunkCache.Directory = "/cache/plans" - first := FromConfig(base) - - changedWorkspace := base - changedWorkspace.Workspace.Directory = "/workspace/two" - second := FromConfig(changedWorkspace) - if base.Workspace.ChunkCache.Directory != changedWorkspace.Workspace.ChunkCache.Directory { - t.Fatal("workspace directory changed chunk plan directory") - } - if first.RootDir == second.RootDir || first.CheckpointsRoot == second.CheckpointsRoot || first.DebugRoot == second.DebugRoot { - t.Fatalf("workspace settings did not follow workspace directory: %#v %#v", first, second) - } - - changedCache := base - changedCache.Workspace.ChunkCache.Directory = "/cache/other" - third := FromConfig(changedCache) - if first != third { - t.Fatalf("chunk plan directory changed workspace settings: %#v %#v", first, third) - } -} diff --git a/internal/core/workspace/identity.go b/internal/core/workspace/identity.go deleted file mode 100644 index 361dbb1..0000000 --- a/internal/core/workspace/identity.go +++ /dev/null @@ -1,276 +0,0 @@ -package workspace - -import ( - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "path/filepath" - "sort" - "strings" - - "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -const digestPrefixLength = 16 - -type Fingerprint struct { - Name string `json:"name"` - Value string `json:"value"` -} - -type CheckpointIdentityInput struct { - Pipeline pipeline.ResolvedPipeline - InputKey string - RawInputDigest string - SourceDigest string - SelectedLanes []string - RuntimeOverrides []Fingerprint - References []artifacts.ReferenceProvenance - ProvenanceFingerprints []Fingerprint -} - -type CheckpointIdentity struct { - Digest string `json:"digest"` - PipelineID string `json:"pipeline_id"` - PipelineDigest string `json:"pipeline_digest"` - InputKey string `json:"input_key"` - RawInputDigest string `json:"raw_input_digest,omitempty"` - SourceDigest string `json:"source_digest,omitempty"` - SelectedLanes []string `json:"selected_lanes,omitempty"` - RuntimeOverrides []Fingerprint `json:"runtime_overrides,omitempty"` - ReferenceDigests []Fingerprint `json:"reference_digests,omitempty"` - ProvenanceFingerprints []Fingerprint `json:"provenance_fingerprints,omitempty"` -} - -func NewCheckpointIdentity(input CheckpointIdentityInput) (CheckpointIdentity, error) { - pipelineID := strings.TrimSpace(input.Pipeline.ID) - if pipelineID == "" { - return CheckpointIdentity{}, fmt.Errorf("checkpoint identity pipeline id must not be empty") - } - pipelineDigest := strings.TrimSpace(input.Pipeline.Digest) - if pipelineDigest == "" { - return CheckpointIdentity{}, fmt.Errorf("checkpoint identity pipeline digest must not be empty") - } - inputKey := strings.TrimSpace(input.InputKey) - if inputKey == "" { - inputKey = strings.TrimSpace(input.Pipeline.Input.Module) - } - if inputKey == "" { - return CheckpointIdentity{}, fmt.Errorf("checkpoint identity input key must not be empty") - } - - rawInputDigest := strings.TrimSpace(input.RawInputDigest) - sourceDigest := strings.TrimSpace(input.SourceDigest) - if rawInputDigest == "" && sourceDigest == "" { - return CheckpointIdentity{}, fmt.Errorf("checkpoint identity raw input digest or source digest must be set") - } - - identity := CheckpointIdentity{ - PipelineID: pipelineID, - PipelineDigest: pipelineDigest, - InputKey: inputKey, - RawInputDigest: rawInputDigest, - SourceDigest: sourceDigest, - SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.ArtifactLanes), - RuntimeOverrides: normalizeFingerprints(input.RuntimeOverrides), - ReferenceDigests: referenceFingerprints(input.References), - ProvenanceFingerprints: normalizeFingerprints(input.ProvenanceFingerprints), - } - digest, err := identityDigest(identity) - if err != nil { - return CheckpointIdentity{}, err - } - identity.Digest = digest - return identity, nil -} - -func (s Settings) CheckpointDirectory(identity CheckpointIdentity) (string, error) { - if !s.ResumeEnabled || strings.TrimSpace(s.CheckpointsRoot) == "" { - return "", nil - } - relative, err := identity.RelativePath() - if err != nil { - return "", err - } - return SafePath(s.CheckpointsRoot, relative) -} - -func (i CheckpointIdentity) RelativePath() (string, error) { - pipelineID, err := safePathComponent(i.PipelineID) - if err != nil { - return "", fmt.Errorf("checkpoint identity pipeline id: %w", err) - } - inputKey, err := safePathComponent(i.InputKey) - if err != nil { - return "", fmt.Errorf("checkpoint identity input key: %w", err) - } - sourceDigest := digestPrefix(i.SourceDigest) - if sourceDigest == "" { - sourceDigest = digestPrefix(i.RawInputDigest) - } - if sourceDigest == "" { - return "", fmt.Errorf("checkpoint identity source digest prefix must not be empty") - } - pipelineDigest := digestPrefix(i.PipelineDigest) - if pipelineDigest == "" { - return "", fmt.Errorf("checkpoint identity pipeline digest prefix must not be empty") - } - sourceComponent, err := safePathComponent(sourceDigest) - if err != nil { - return "", fmt.Errorf("checkpoint identity source digest: %w", err) - } - pipelineComponent, err := safePathComponent(pipelineDigest) - if err != nil { - return "", fmt.Errorf("checkpoint identity pipeline digest: %w", err) - } - identityDigest := digestPrefix(i.Digest) - if identityDigest == "" { - return "", fmt.Errorf("checkpoint identity digest prefix must not be empty") - } - identityComponent, err := safePathComponent(identityDigest) - if err != nil { - return "", fmt.Errorf("checkpoint identity digest: %w", err) - } - return filepath.ToSlash(filepath.Join(pipelineID, inputKey+"-"+sourceComponent, pipelineComponent, identityComponent)), nil -} - -func identityDigest(identity CheckpointIdentity) (string, error) { - payload := identity - payload.Digest = "" - data, err := json.Marshal(payload) - if err != nil { - return "", fmt.Errorf("marshal checkpoint identity: %w", err) - } - sum := sha256.Sum256(data) - return "sha256:" + hex.EncodeToString(sum[:]), nil -} - -func normalizedLanes(selected []string, resolved []pipeline.ResolvedArtifactLane) []string { - if len(selected) > 0 { - return normalizeStrings(selected) - } - lanes := make([]string, 0, len(resolved)) - for _, lane := range resolved { - lanes = append(lanes, lane.ID) - } - return normalizeStrings(lanes) -} - -func normalizeFingerprints(values []Fingerprint) []Fingerprint { - if len(values) == 0 { - return nil - } - byName := make(map[string]string, len(values)) - for _, value := range values { - name := strings.TrimSpace(value.Name) - fingerprint := strings.TrimSpace(value.Value) - if name == "" || fingerprint == "" { - continue - } - byName[name] = fingerprint - } - if len(byName) == 0 { - return nil - } - names := make([]string, 0, len(byName)) - for name := range byName { - names = append(names, name) - } - sort.Strings(names) - out := make([]Fingerprint, 0, len(names)) - for _, name := range names { - out = append(out, Fingerprint{Name: name, Value: byName[name]}) - } - return out -} - -func referenceFingerprints(references []artifacts.ReferenceProvenance) []Fingerprint { - if len(references) == 0 { - return nil - } - values := make([]Fingerprint, 0, len(references)) - for _, reference := range references { - digest := strings.TrimSpace(reference.Digest) - if digest == "" { - continue - } - parts := []string{ - strings.TrimSpace(reference.Stage), - strings.TrimSpace(reference.LaneID), - strings.TrimSpace(reference.SlotName), - strings.TrimSpace(reference.OriginURI), - } - values = append(values, Fingerprint{ - Name: strings.Join(parts, ":"), - Value: digest, - }) - } - return normalizeFingerprints(values) -} - -func normalizeStrings(values []string) []string { - if len(values) == 0 { - return nil - } - seen := make(map[string]struct{}, len(values)) - for _, value := range values { - value = strings.TrimSpace(value) - if value == "" { - continue - } - seen[value] = struct{}{} - } - if len(seen) == 0 { - return nil - } - out := make([]string, 0, len(seen)) - for value := range seen { - out = append(out, value) - } - sort.Strings(out) - return out -} - -func digestPrefix(digest string) string { - digest = strings.TrimSpace(digest) - if digest == "" { - return "" - } - if idx := strings.Index(digest, ":"); idx >= 0 { - digest = digest[idx+1:] - } - digest = strings.TrimSpace(digest) - if len(digest) > digestPrefixLength { - return digest[:digestPrefixLength] - } - return digest -} - -func safePathComponent(value string) (string, error) { - value = strings.TrimSpace(value) - if value == "" { - return "", fmt.Errorf("must not be empty") - } - var b strings.Builder - for _, r := range value { - switch { - case r >= 'a' && r <= 'z': - b.WriteRune(r) - case r >= 'A' && r <= 'Z': - b.WriteRune(r) - case r >= '0' && r <= '9': - b.WriteRune(r) - case r == '-' || r == '_' || r == '.': - b.WriteRune(r) - default: - b.WriteString(fmt.Sprintf("~%x", r)) - } - } - encoded := b.String() - if encoded == "." || encoded == ".." || strings.Contains(encoded, "..") || strings.ContainsAny(encoded, `/\`) { - return "", fmt.Errorf("%q is not filesystem safe", value) - } - return encoded, nil -} diff --git a/internal/core/workspace/identity_test.go b/internal/core/workspace/identity_test.go deleted file mode 100644 index d9f5b1f..0000000 --- a/internal/core/workspace/identity_test.go +++ /dev/null @@ -1,266 +0,0 @@ -package workspace - -import ( - "path/filepath" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -func TestCheckpointIdentityIsDeterministic(t *testing.T) { - first := mustIdentity(t, identityInput()) - second := mustIdentity(t, identityInput()) - - if first.Digest != second.Digest { - t.Fatalf("digest changed for same input: %q != %q", first.Digest, second.Digest) - } - if !strings.HasPrefix(first.Digest, "sha256:") { - t.Fatalf("digest = %q, want sha256 prefix", first.Digest) - } -} - -func TestCheckpointIdentityChangesWhenInputsChange(t *testing.T) { - base := mustIdentity(t, identityInput()) - tests := []struct { - name string - mutate func(CheckpointIdentityInput) CheckpointIdentityInput - }{ - { - name: "pipeline digest", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.Pipeline.Digest = "sha256:pipeline-b" - return input - }, - }, - { - name: "raw input digest", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.RawInputDigest = "sha256:raw-b" - return input - }, - }, - { - name: "selected lanes", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.SelectedLanes = []string{"items"} - return input - }, - }, - { - name: "reference digest", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.References[0].Digest = "sha256:reference-b" - return input - }, - }, - { - name: "runtime override", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.RuntimeOverrides = []Fingerprint{{Name: "llm_profile", Value: "careful"}} - return input - }, - }, - { - name: "provenance fingerprint", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.ProvenanceFingerprints = []Fingerprint{{Name: "prompt:dnd.spells", Value: "sha256:prompt-b"}} - return input - }, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - changed := mustIdentity(t, tc.mutate(identityInput())) - if changed.Digest == base.Digest { - t.Fatalf("digest did not change after %s mutation: %q", tc.name, changed.Digest) - } - }) - } -} - -func TestCheckpointIdentityNormalizesOrder(t *testing.T) { - input := identityInput() - input.SelectedLanes = []string{"spells", "items", "spells"} - input.RuntimeOverrides = []Fingerprint{ - {Name: "z", Value: "2"}, - {Name: "a", Value: "1"}, - } - input.ProvenanceFingerprints = []Fingerprint{ - {Name: "schema", Value: "sha256:schema"}, - {Name: "prompt", Value: "sha256:prompt"}, - } - - identity := mustIdentity(t, input) - - if got := strings.Join(identity.SelectedLanes, ","); got != "items,spells" { - t.Fatalf("selected lanes = %q, want sorted unique values", got) - } - if identity.RuntimeOverrides[0].Name != "a" || identity.ProvenanceFingerprints[0].Name != "prompt" { - t.Fatalf("fingerprints not sorted: runtime=%+v provenance=%+v", identity.RuntimeOverrides, identity.ProvenanceFingerprints) - } -} - -func TestCheckpointIdentityPathIsFilesystemSafe(t *testing.T) { - input := identityInput() - input.Pipeline.ID = "campaign/main" - input.InputKey = "seriatim/input" - input.SourceDigest = "sha256:abcdef0123456789ffffffff" - input.Pipeline.Digest = "sha256:1234567890abcdefeeeeeeee" - identity := mustIdentity(t, input) - - relative, err := identity.RelativePath() - if err != nil { - t.Fatalf("RelativePath: %v", err) - } - if strings.Contains(relative, `\`) || strings.Contains(relative, "..") { - t.Fatalf("relative path is not filesystem safe: %q", relative) - } - identityDigest := digestPrefix(identity.Digest) - wantRelative := "campaign~2fmain/seriatim~2finput-abcdef0123456789/1234567890abcdef/" + identityDigest - if relative != wantRelative { - t.Fatalf("relative path = %q", relative) - } - - root := t.TempDir() - settings := Settings{ - CheckpointsRoot: filepath.Join(root, "checkpoints"), - ResumeEnabled: true, - } - got, err := settings.CheckpointDirectory(identity) - if err != nil { - t.Fatalf("CheckpointDirectory: %v", err) - } - want := filepath.Join(root, "checkpoints", "campaign~2fmain", "seriatim~2finput-abcdef0123456789", "1234567890abcdef", identityDigest) - if got != want { - t.Fatalf("checkpoint directory = %q, want %q", got, want) - } -} - -func TestCheckpointIdentityPathIncludesInvocationIdentity(t *testing.T) { - base := mustIdentity(t, identityInput()) - changedInput := identityInput() - changedInput.References[0].Digest = "sha256:reference-b" - changed := mustIdentity(t, changedInput) - - if base.Digest == changed.Digest { - t.Fatalf("test setup produced same identity digest: %q", base.Digest) - } - basePath, err := base.RelativePath() - if err != nil { - t.Fatalf("base RelativePath: %v", err) - } - changedPath, err := changed.RelativePath() - if err != nil { - t.Fatalf("changed RelativePath: %v", err) - } - if basePath == changedPath { - t.Fatalf("relative path did not change with invocation identity: %q", basePath) - } -} - -func TestCheckpointDirectoryDisabledReturnsEmptyPath(t *testing.T) { - settings := Settings{CheckpointsRoot: filepath.Join(t.TempDir(), "checkpoints")} - got, err := settings.CheckpointDirectory(mustIdentity(t, identityInput())) - if err != nil { - t.Fatalf("CheckpointDirectory: %v", err) - } - if got != "" { - t.Fatalf("CheckpointDirectory = %q, want empty path", got) - } -} - -func TestNewCheckpointIdentityRequiresCoreInputs(t *testing.T) { - tests := []struct { - name string - mutate func(CheckpointIdentityInput) CheckpointIdentityInput - want string - }{ - { - name: "pipeline id", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.Pipeline.ID = "" - return input - }, - want: "pipeline id", - }, - { - name: "pipeline digest", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.Pipeline.Digest = "" - return input - }, - want: "pipeline digest", - }, - { - name: "input key", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.InputKey = "" - input.Pipeline.Input.Module = "" - return input - }, - want: "input key", - }, - { - name: "input digest", - mutate: func(input CheckpointIdentityInput) CheckpointIdentityInput { - input.RawInputDigest = "" - input.SourceDigest = "" - return input - }, - want: "raw input digest or source digest", - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - _, err := NewCheckpointIdentity(tc.mutate(identityInput())) - if err == nil || !strings.Contains(err.Error(), tc.want) { - t.Fatalf("expected error containing %q, got %v", tc.want, err) - } - }) - } -} - -func identityInput() CheckpointIdentityInput { - return CheckpointIdentityInput{ - Pipeline: pipeline.ResolvedPipeline{ - ID: "dnd-session", - Digest: "sha256:pipeline-a", - Input: pipeline.Binding("seriatim"), - ArtifactLanes: []pipeline.ResolvedArtifactLane{ - {ID: "spells"}, - {ID: "items"}, - }, - }, - InputKey: "seriatim", - RawInputDigest: "sha256:raw-a", - SelectedLanes: []string{"spells"}, - RuntimeOverrides: []Fingerprint{ - {Name: "llm_profile", Value: "fast"}, - }, - References: []artifacts.ReferenceProvenance{ - { - Stage: "extract", - LaneID: "spells", - SlotName: "party", - OriginURI: "file:///party.yml", - Digest: "sha256:reference-a", - }, - }, - ProvenanceFingerprints: []Fingerprint{ - {Name: "prompt:dnd.spells", Value: "sha256:prompt-a"}, - }, - } -} - -func mustIdentity(t *testing.T, input CheckpointIdentityInput) CheckpointIdentity { - t.Helper() - identity, err := NewCheckpointIdentity(input) - if err != nil { - t.Fatalf("NewCheckpointIdentity: %v", err) - } - return identity -} diff --git a/internal/core/workspace/manifest_test.go b/internal/core/workspace/manifest_test.go deleted file mode 100644 index fd509d0..0000000 --- a/internal/core/workspace/manifest_test.go +++ /dev/null @@ -1,137 +0,0 @@ -package workspace - -import ( - "encoding/json" - "testing" - "time" -) - -func TestStageManifestDefaults(t *testing.T) { - if WorkspaceSchemaVersion != "notarius.workspace.v2" { - t.Fatalf("current schema version = %q, want notarius.workspace.v2", WorkspaceSchemaVersion) - } - if WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { - t.Fatalf("legacy schema version = %q, want notarius.workspace.v1", WorkspaceSchemaVersionV1) - } - manifest := NewStageManifest(StageExtract, StatusRunning) - - if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { - t.Fatalf("schema version = %q, want %q", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion) - } - if manifest.Stage != StageExtract { - t.Fatalf("stage = %q, want extract", manifest.Stage) - } - if manifest.Status != StatusRunning { - t.Fatalf("status = %q, want running", manifest.Status) - } -} - -func TestManifestJSONRoundTrips(t *testing.T) { - started := time.Unix(100, 0).UTC() - completed := time.Unix(200, 0).UTC() - - t.Run("source", func(t *testing.T) { - manifest := SourceManifest{ - StageManifest: populatedManifest(StageSource, "", "seriatim", started, completed), - SourceID: "source-1", - } - var got SourceManifest - roundTripManifest(t, manifest, &got) - if got.SourceID != manifest.SourceID || got.Stage != StageSource { - t.Fatalf("round trip source manifest = %+v", got) - } - }) - - t.Run("extract", func(t *testing.T) { - manifest := ExtractLaneManifest{ - StageManifest: populatedManifest(StageExtract, "spells", "dnd/spells", started, completed), - ChunkCount: 3, - OutputCount: 2, - } - var got ExtractLaneManifest - roundTripManifest(t, manifest, &got) - if got.LaneID != "spells" || got.OutputCount != manifest.OutputCount || got.Stage != StageExtract { - t.Fatalf("round trip extract manifest = %+v", got) - } - }) - - t.Run("merge", func(t *testing.T) { - manifest := MergeLaneManifest{ - StageManifest: populatedManifest(StageMerge, "spells", "appendorder", started, completed), - InputCount: 2, - } - var got MergeLaneManifest - roundTripManifest(t, manifest, &got) - if got.InputCount != manifest.InputCount || got.Stage != StageMerge { - t.Fatalf("round trip merge manifest = %+v", got) - } - }) - - t.Run("normalize", func(t *testing.T) { - manifest := NormalizeLaneManifest{ - StageManifest: populatedManifest(StageNormalize, "spells", "noop", started, completed), - InputCount: 1, - } - var got NormalizeLaneManifest - roundTripManifest(t, manifest, &got) - if got.InputCount != manifest.InputCount || got.Stage != StageNormalize { - t.Fatalf("round trip normalize manifest = %+v", got) - } - }) -} - -func TestStatusValues(t *testing.T) { - values := []StageStatus{ - StatusPending, - StatusRunning, - StatusSucceeded, - StatusSucceededWithRejections, - StatusFailed, - StatusInvalidated, - } - want := []string{ - "pending", - "running", - "succeeded", - "succeeded_with_rejections", - "failed", - "invalidated", - } - for i, value := range values { - if string(value) != want[i] { - t.Fatalf("status[%d] = %q, want %q", i, value, want[i]) - } - } -} - -func populatedManifest(stage StageName, laneID string, moduleKey string, started time.Time, completed time.Time) StageManifest { - manifest := NewStageManifest(stage, StatusSucceededWithRejections) - manifest.LaneID = laneID - manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = []Fingerprint{{Name: "source", Value: "sha256:source"}} - manifest.OutputDigests = []Fingerprint{{Name: "output", Value: "sha256:output"}} - manifest.ValidationStatus = "approved_with_warnings" - manifest.Rejections = []RejectionSummary{ - { - ValidatorName: "shape", - ReasonCode: "invalid_shape", - Message: "invalid output shape", - Count: 1, - }, - } - manifest.StartedAt = &started - manifest.CompletedAt = &completed - manifest.Metadata = map[string]string{"attempt": "1"} - return manifest -} - -func roundTripManifest(t *testing.T, in any, out any) { - t.Helper() - data, err := json.Marshal(in) - if err != nil { - t.Fatalf("marshal manifest: %v", err) - } - if err := json.Unmarshal(data, out); err != nil { - t.Fatalf("unmarshal manifest: %v", err) - } -} diff --git a/internal/core/workspace/settings.go b/internal/core/workspace/settings.go index 421170e..c23c43a 100644 --- a/internal/core/workspace/settings.go +++ b/internal/core/workspace/settings.go @@ -45,13 +45,6 @@ func (s Settings) DiagnosticsRunDirectory(runID string) (string, error) { return safeSingleDirectory(s.DiagnosticsRoot, runID, "diagnostics run ID") } -func (s Settings) CheckpointIdentityDirectory(identity string) (string, error) { - if !s.ResumeEnabled || strings.TrimSpace(s.CheckpointsRoot) == "" { - return "", nil - } - return SafePath(s.CheckpointsRoot, identity) -} - func (s Settings) DebugRunDirectory(runID string) (string, error) { if !s.DebugEnabled || strings.TrimSpace(s.DebugRoot) == "" { return "", nil diff --git a/internal/core/workspace/settings_test.go b/internal/core/workspace/settings_test.go index 3d7c727..3bac3e7 100644 --- a/internal/core/workspace/settings_test.go +++ b/internal/core/workspace/settings_test.go @@ -72,14 +72,6 @@ func TestPathConstructors(t *testing.T) { t.Fatalf("diagnostics dir = %q", diagnosticsDir) } - checkpointDir, err := settings.CheckpointIdentityDirectory("pipeline/input-digest/pipeline-digest") - if err != nil { - t.Fatalf("CheckpointIdentityDirectory: %v", err) - } - if checkpointDir != filepath.Join(root, "checkpoints", "pipeline", "input-digest", "pipeline-digest") { - t.Fatalf("checkpoint dir = %q", checkpointDir) - } - debugDir, err := settings.DebugRunDirectory("run-456") if err != nil { t.Fatalf("DebugRunDirectory: %v", err) @@ -94,7 +86,6 @@ func TestDisabledPathConstructorsReturnEmptyPaths(t *testing.T) { for name, call := range map[string]func() (string, error){ "diagnostics": func() (string, error) { return settings.DiagnosticsRunDirectory("run-1") }, - "checkpoint": func() (string, error) { return settings.CheckpointIdentityDirectory("identity") }, "debug": func() (string, error) { return settings.DebugRunDirectory("run-1") }, } { t.Run(name, func(t *testing.T) { diff --git a/internal/framework/checkpoint/identity.go b/internal/framework/checkpoint/identity.go new file mode 100644 index 0000000..782886e --- /dev/null +++ b/internal/framework/checkpoint/identity.go @@ -0,0 +1,184 @@ +package checkpoint + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const digestPrefixLength = 16 + +type Fingerprint struct { + Name string `json:"name"` + Value string `json:"value"` +} +type IdentityInput struct { + Pipeline pipeline.ResolvedPipeline + InputKey, RawInputDigest, SourceDigest string + SelectedLanes []string + RuntimeOverrides []Fingerprint + References []artifacts.ReferenceProvenance + ProvenanceFingerprints []Fingerprint +} +type Identity struct { + Digest string `json:"digest"` + PipelineID string `json:"pipeline_id"` + PipelineDigest string `json:"pipeline_digest"` + InputKey string `json:"input_key"` + RawInputDigest string `json:"raw_input_digest,omitempty"` + SourceDigest string `json:"source_digest,omitempty"` + SelectedLanes []string `json:"selected_lanes,omitempty"` + RuntimeOverrides []Fingerprint `json:"runtime_overrides,omitempty"` + ReferenceDigests []Fingerprint `json:"reference_digests,omitempty"` + ProvenanceFingerprints []Fingerprint `json:"provenance_fingerprints,omitempty"` +} + +func NewIdentity(input IdentityInput) (Identity, error) { + pipelineID, pipelineDigest, inputKey := strings.TrimSpace(input.Pipeline.ID), strings.TrimSpace(input.Pipeline.Digest), strings.TrimSpace(input.InputKey) + if pipelineID == "" { + return Identity{}, fmt.Errorf("checkpoint identity pipeline id must not be empty") + } + if pipelineDigest == "" { + return Identity{}, fmt.Errorf("checkpoint identity pipeline digest must not be empty") + } + if inputKey == "" { + inputKey = strings.TrimSpace(input.Pipeline.Input.Module) + } + if inputKey == "" { + return Identity{}, fmt.Errorf("checkpoint identity input key must not be empty") + } + if strings.TrimSpace(input.RawInputDigest) == "" && strings.TrimSpace(input.SourceDigest) == "" { + return Identity{}, fmt.Errorf("checkpoint identity raw input digest or source digest must be set") + } + v := Identity{PipelineID: pipelineID, PipelineDigest: pipelineDigest, InputKey: inputKey, RawInputDigest: strings.TrimSpace(input.RawInputDigest), SourceDigest: strings.TrimSpace(input.SourceDigest), SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.ArtifactLanes), RuntimeOverrides: normalizeIdentityFingerprints(input.RuntimeOverrides), ReferenceDigests: referenceFingerprints(input.References), ProvenanceFingerprints: normalizeIdentityFingerprints(input.ProvenanceFingerprints)} + data, err := json.Marshal(Identity{PipelineID: v.PipelineID, PipelineDigest: v.PipelineDigest, InputKey: v.InputKey, RawInputDigest: v.RawInputDigest, SourceDigest: v.SourceDigest, SelectedLanes: v.SelectedLanes, RuntimeOverrides: v.RuntimeOverrides, ReferenceDigests: v.ReferenceDigests, ProvenanceFingerprints: v.ProvenanceFingerprints}) + if err != nil { + return Identity{}, fmt.Errorf("marshal checkpoint identity: %w", err) + } + sum := sha256.Sum256(data) + v.Digest = "sha256:" + hex.EncodeToString(sum[:]) + return v, nil +} +func (i Identity) RelativePath() (string, error) { + p, err := safeComponent(i.PipelineID) + if err != nil { + return "", fmt.Errorf("checkpoint identity pipeline id: %w", err) + } + k, err := safeComponent(i.InputKey) + if err != nil { + return "", fmt.Errorf("checkpoint identity input key: %w", err) + } + s := digestPrefix(i.SourceDigest) + if s == "" { + s = digestPrefix(i.RawInputDigest) + } + d := digestPrefix(i.PipelineDigest) + x := digestPrefix(i.Digest) + if s == "" || d == "" || x == "" { + return "", fmt.Errorf("checkpoint identity digest prefix must not be empty") + } + s, err = safeComponent(s) + if err != nil { + return "", fmt.Errorf("checkpoint identity source digest: %w", err) + } + d, err = safeComponent(d) + if err != nil { + return "", fmt.Errorf("checkpoint identity pipeline digest: %w", err) + } + x, err = safeComponent(x) + if err != nil { + return "", fmt.Errorf("checkpoint identity digest: %w", err) + } + return filepath.ToSlash(filepath.Join(p, k+"-"+s, d, x)), nil +} +func normalizedLanes(selected []string, resolved []pipeline.ResolvedArtifactLane) []string { + if len(selected) == 0 { + for _, lane := range resolved { + selected = append(selected, lane.ID) + } + } + return normalizeStrings(selected) +} +func normalizeIdentityFingerprints(values []Fingerprint) []Fingerprint { + by := map[string]string{} + for _, v := range values { + if n, x := strings.TrimSpace(v.Name), strings.TrimSpace(v.Value); n != "" && x != "" { + by[n] = x + } + } + names := make([]string, 0, len(by)) + for n := range by { + names = append(names, n) + } + sort.Strings(names) + out := make([]Fingerprint, 0, len(names)) + for _, n := range names { + out = append(out, Fingerprint{Name: n, Value: by[n]}) + } + if len(out) == 0 { + return nil + } + return out +} +func referenceFingerprints(refs []artifacts.ReferenceProvenance) []Fingerprint { + var values []Fingerprint + for _, r := range refs { + if d := strings.TrimSpace(r.Digest); d != "" { + values = append(values, Fingerprint{Name: strings.Join([]string{strings.TrimSpace(r.Stage), strings.TrimSpace(r.LaneID), strings.TrimSpace(r.SlotName), strings.TrimSpace(r.OriginURI)}, ":"), Value: d}) + } + } + return normalizeIdentityFingerprints(values) +} +func normalizeStrings(values []string) []string { + seen := map[string]struct{}{} + for _, v := range values { + if v = strings.TrimSpace(v); v != "" { + seen[v] = struct{}{} + } + } + out := make([]string, 0, len(seen)) + for v := range seen { + out = append(out, v) + } + sort.Strings(out) + if len(out) == 0 { + return nil + } + return out +} +func digestPrefix(v string) string { + v = strings.TrimSpace(v) + if n := strings.Index(v, ":"); n >= 0 { + v = v[n+1:] + } + if len(v) > digestPrefixLength { + return v[:digestPrefixLength] + } + return v +} +func safeComponent(v string) (string, error) { + v = strings.TrimSpace(v) + if v == "" { + return "", fmt.Errorf("must not be empty") + } + var b strings.Builder + for _, r := range v { + if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' { + b.WriteRune(r) + } else { + b.WriteString(fmt.Sprintf("~%x", r)) + } + } + out := b.String() + if out == "." || out == ".." || strings.Contains(out, "..") || strings.ContainsAny(out, `/\\`) { + return "", fmt.Errorf("%q is not filesystem safe", v) + } + return out, nil +} diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index 427131c..f6a3005 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -7,38 +7,43 @@ import ( "os" "strings" + "gitea.maximumdirect.net/eric/notarius/internal/core/fileio" "gitea.maximumdirect.net/eric/notarius/internal/core/source" - coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -type WorkspaceLoader struct { +type FilesystemLoader struct { root string identityDigest string } -func NewWorkspaceLoader(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointLoader, error) { - root, err := settings.CheckpointDirectory(identity) +func NewFilesystemLoader(root string, identity Identity) (pipeline.CheckpointLoader, error) { + root = strings.TrimSpace(root) + if root == "" { + return pipeline.NoopCheckpointLoader(), nil + } + relative, err := identity.RelativePath() if err != nil { return nil, err } - if strings.TrimSpace(root) == "" { - return pipeline.NoopCheckpointLoader(), nil + target, err := fileio.SafePath(root, relative) + if err != nil { + return nil, err } - return &WorkspaceLoader{root: root, identityDigest: identity.Digest}, nil + return &FilesystemLoader{root: target, identityDigest: identity.Digest}, nil } -func (l *WorkspaceLoader) Enabled() bool { +func (l *FilesystemLoader) Enabled() bool { return l != nil && strings.TrimSpace(l.root) != "" } -func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) { - var manifest coreworkspace.SourceManifest +func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) { + var manifest SourceManifest if decision := l.readJSON("source/manifest.json", &manifest); !decision.Reused { return pipeline.SourceCheckpoint{}, decision } - if decision := l.validateManifest(manifest.StageManifest, coreworkspace.StageSource, "", moduleKey, coreworkspace.StatusSucceeded, nil); !decision.Reused { + if decision := l.validateManifest(manifest.StageManifest, StageSource, "", moduleKey, StatusSucceeded, nil); !decision.Reused { return pipeline.SourceCheckpoint{}, decision } var payload sourceDocumentEnvelope @@ -52,18 +57,18 @@ func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, p if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID { return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint source id does not match payload") } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) { + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) { return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint output digest does not match payload") } return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision() } -func (l *WorkspaceLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { - var manifest coreworkspace.ExtractLaneManifest +func (l *FilesystemLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { + var manifest ExtractLaneManifest if d := l.readJSON(laneManifestPath("extract", laneID), &manifest); !d.Reused { return pipeline.ExtractCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageExtract, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded, coreworkspace.StatusSucceededWithRejections); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused { return pipeline.ExtractCheckpoint{}, d } var payload artifactExtractEnvelope @@ -74,18 +79,18 @@ func (l *WorkspaceLoader) Extract(laneID, moduleKey string, dependencies []pipel if err != nil { return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint payload is invalid: %v", err) } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) { + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) { return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint output digests do not match payload") } return pipeline.ExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } -func (l *WorkspaceLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { - var manifest coreworkspace.MergeLaneManifest +func (l *FilesystemLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { + var manifest MergeLaneManifest if d := l.readJSON(laneManifestPath("merge", laneID), &manifest); !d.Reused { return pipeline.MergeCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageMerge, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { return pipeline.MergeCheckpoint{}, d } var payload artifactSingleEnvelope @@ -96,18 +101,18 @@ func (l *WorkspaceLoader) Merge(laneID, moduleKey string, dependencies []pipelin if err != nil { return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint payload is invalid: %v", err) } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint output digest does not match payload") } return pipeline.MergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } -func (l *WorkspaceLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { - var manifest coreworkspace.NormalizeLaneManifest +func (l *FilesystemLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { + var manifest NormalizeLaneManifest if d := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !d.Reused { return pipeline.NormalizeCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageNormalize, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { return pipeline.NormalizeCheckpoint{}, d } var payload artifactSingleEnvelope @@ -118,7 +123,7 @@ func (l *WorkspaceLoader) Normalize(laneID, moduleKey string, dependencies []pip if err != nil { return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint payload is invalid: %v", err) } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint output digest does not match payload") } return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() @@ -142,11 +147,11 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. return out, nil } -func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDecision { +func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"} } - target, err := coreworkspace.SafePath(l.root, name) + target, err := fileio.SafePath(l.root, name) if err != nil { return invalidDecision("checkpoint path is invalid: %v", err) } @@ -163,15 +168,15 @@ func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDeci return reusedDecision() } -func (l *WorkspaceLoader) validateManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, status coreworkspace.StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision { +func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, status StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision { return l.validateLaneManifest(manifest, stage, laneID, moduleKey, dependencies, status) } -func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...coreworkspace.StageStatus) pipeline.CheckpointDecision { - if manifest.WorkspaceSchemaVersion == coreworkspace.WorkspaceSchemaVersionV1 { - return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, coreworkspace.WorkspaceSchemaVersion) +func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision { + if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV1 { + return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion) } - if manifest.WorkspaceSchemaVersion != coreworkspace.WorkspaceSchemaVersion { + if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion) } if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest { @@ -196,7 +201,7 @@ func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManif if !statusOK { return invalidDecision("checkpoint status %q cannot be reused", manifest.Status) } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) { + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) { return invalidDecision("checkpoint dependency fingerprints do not match") } return reusedDecision() @@ -213,7 +218,7 @@ func contentFromEnvelope(value binaryEnvelope) ([]byte, error) { return content, nil } -func coreworkspaceToPipelineFingerprints(values []coreworkspace.Fingerprint) []pipeline.CheckpointFingerprint { +func checkpointToPipelineFingerprints(values []Fingerprint) []pipeline.CheckpointFingerprint { if len(values) == 0 { return nil } diff --git a/internal/core/workspace/manifest.go b/internal/framework/checkpoint/manifest.go similarity index 93% rename from internal/core/workspace/manifest.go rename to internal/framework/checkpoint/manifest.go index c1a9e1e..20b0325 100644 --- a/internal/core/workspace/manifest.go +++ b/internal/framework/checkpoint/manifest.go @@ -1,4 +1,4 @@ -package workspace +package checkpoint import "time" @@ -41,39 +41,30 @@ type StageManifest struct { CompletedAt *time.Time `json:"completed_at,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` } - type RejectionSummary struct { ValidatorName string `json:"validator_name,omitempty"` ReasonCode string `json:"reason_code,omitempty"` Message string `json:"message,omitempty"` Count int `json:"count,omitempty"` } - type SourceManifest struct { StageManifest SourceID string `json:"source_id,omitempty"` } - type ExtractLaneManifest struct { StageManifest ChunkCount int `json:"chunk_count,omitempty"` OutputCount int `json:"output_count,omitempty"` } - type MergeLaneManifest struct { StageManifest InputCount int `json:"input_count,omitempty"` } - type NormalizeLaneManifest struct { StageManifest InputCount int `json:"input_count,omitempty"` } func NewStageManifest(stage StageName, status StageStatus) StageManifest { - return StageManifest{ - WorkspaceSchemaVersion: WorkspaceSchemaVersion, - Stage: stage, - Status: status, - } + return StageManifest{WorkspaceSchemaVersion: WorkspaceSchemaVersion, Stage: stage, Status: status} } diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index 8ee7d8d..0fe7c7f 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -10,186 +10,191 @@ import ( "strings" "time" + "gitea.maximumdirect.net/eric/notarius/internal/core/fileio" "gitea.maximumdirect.net/eric/notarius/internal/core/source" - coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -type WorkspaceRecorder struct { +type FilesystemRecorder struct { root string identityDigest string now func() time.Time } -func NewWorkspaceRecorder(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointRecorder, error) { - root, err := settings.CheckpointDirectory(identity) +func NewFilesystemRecorder(root string, identity Identity) (pipeline.CheckpointRecorder, error) { + root = strings.TrimSpace(root) + if root == "" { + return pipeline.NoopCheckpointRecorder(), nil + } + relative, err := identity.RelativePath() if err != nil { return nil, err } - if strings.TrimSpace(root) == "" { - return pipeline.NoopCheckpointRecorder(), nil + target, err := fileio.SafePath(root, relative) + if err != nil { + return nil, err } - return &WorkspaceRecorder{root: root, identityDigest: identity.Digest, now: time.Now}, nil + return &FilesystemRecorder{root: target, identityDigest: identity.Digest, now: time.Now}, nil } -func (r *WorkspaceRecorder) SourceRunning(moduleKey string) error { - manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusRunning) +func (r *FilesystemRecorder) SourceRunning(moduleKey string) error { + manifest := r.newStageManifest(StageSource, StatusRunning) manifest.ModuleKey = moduleKey manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) + return r.writeManifest("source/manifest.json", SourceManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) SourceSucceeded(moduleKey string, doc *source.SourceDocument) error { +func (r *FilesystemRecorder) SourceSucceeded(moduleKey string, doc *source.SourceDocument) error { if doc == nil { return fmt.Errorf("checkpoint source document must not be nil") } if err := r.writePayload("source/source-document.json", sourceDocumentEnvelope{Document: cloneSourceDocument(*doc)}); err != nil { return err } - manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusSucceeded) + manifest := r.newStageManifest(StageSource, StatusSucceeded) manifest.ModuleKey = moduleKey - manifest.OutputDigests = workspaceFingerprints(digestFingerprints("source_document", doc.Digest)) + manifest.OutputDigests = checkpointFingerprints(digestFingerprints("source_document", doc.Digest)) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{ + return r.writeManifest("source/manifest.json", SourceManifest{ StageManifest: manifest, SourceID: doc.ID, }) } -func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error { - manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusFailed) +func (r *FilesystemRecorder) SourceFailed(moduleKey string, err error) error { + manifest := r.newStageManifest(StageSource, StatusFailed) manifest.ModuleKey = moduleKey manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) + return r.writeManifest("source/manifest.json", SourceManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageExtract, StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) ExtractSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []pipeline.CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { +func (r *FilesystemRecorder) ExtractSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []pipeline.CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { payload := artifactExtractEnvelope{Outputs: artifactCheckpointEnvelopes(outputs), Rejected: cloneRejectedOutputs(rejected), Warnings: cloneWarnings(warnings)} if err := r.writePayload(lanePayloadPath("extract", laneID, "outputs.json"), payload); err != nil { return err } - manifest := r.laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) - manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests(outputs)) + manifest := r.laneManifest(StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) + manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests(outputs)) manifest.ValidationStatus = validationStatusString(warnings, rejected) manifest.Rejections = rejectionSummaries(rejected) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest, ChunkCount: len(outputs) + len(rejected), OutputCount: len(outputs)}) + return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest, ChunkCount: len(outputs) + len(rejected), OutputCount: len(outputs)}) } -func (r *WorkspaceRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageExtract, StatusFailed, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) MergeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) MergeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageMerge, StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) MergeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { +func (r *FilesystemRecorder) MergeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { return err } - manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) - manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) + manifest := r.laneManifest(StageMerge, StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) manifest.ValidationStatus = validationStatusString(warnings, nil) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } -func (r *WorkspaceRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { - manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := r.laneManifest(StageMerge, StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } -func (r *WorkspaceRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageMerge, StatusFailed, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) NormalizeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) NormalizeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageNormalize, StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) NormalizeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { +func (r *FilesystemRecorder) NormalizeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { return err } - manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) - manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) + manifest := r.laneManifest(StageNormalize, StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) manifest.ValidationStatus = validationStatusString(warnings, nil) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } -func (r *WorkspaceRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { - manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := r.laneManifest(StageNormalize, StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } -func (r *WorkspaceRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) +func (r *FilesystemRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageNormalize, StatusFailed, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) writeManifest(name string, payload any) error { +func (r *FilesystemRecorder) writeManifest(name string, payload any) error { return r.writeJSON(name, payload) } -func (r *WorkspaceRecorder) writePayload(name string, payload any) error { +func (r *FilesystemRecorder) writePayload(name string, payload any) error { return r.writeJSON(name, payload) } -func (r *WorkspaceRecorder) writeJSON(name string, payload any) error { +func (r *FilesystemRecorder) writeJSON(name string, payload any) error { if r == nil || strings.TrimSpace(r.root) == "" { return nil } - return coreworkspace.WriteJSON(r.root, name, payload) + return fileio.WriteJSON(r.root, name, payload, 0o700, 0o600) } -func (r *WorkspaceRecorder) timestamp() time.Time { +func (r *FilesystemRecorder) timestamp() time.Time { if r == nil || r.now == nil { return time.Now().UTC() } return r.now().UTC() } -func (r *WorkspaceRecorder) newStageManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus) coreworkspace.StageManifest { - manifest := coreworkspace.NewStageManifest(stage, status) +func (r *FilesystemRecorder) newStageManifest(stage StageName, status StageStatus) StageManifest { + manifest := NewStageManifest(stage, status) if strings.TrimSpace(r.identityDigest) != "" { manifest.Metadata = map[string]string{"checkpoint_identity_digest": r.identityDigest} } return manifest } -func (r *WorkspaceRecorder) laneManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) coreworkspace.StageManifest { +func (r *FilesystemRecorder) laneManifest(stage StageName, status StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) StageManifest { manifest := r.newStageManifest(stage, status) manifest.LaneID = laneID manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = workspaceFingerprints(dependencies) + manifest.DependencyFingerprints = checkpointFingerprints(dependencies) return manifest } @@ -316,14 +321,14 @@ func digestFingerprints(name string, digest string) []pipeline.CheckpointFingerp return []pipeline.CheckpointFingerprint{{Name: name, Value: digest}} } -func workspaceFingerprints(values []pipeline.CheckpointFingerprint) []coreworkspace.Fingerprint { +func checkpointFingerprints(values []pipeline.CheckpointFingerprint) []Fingerprint { normalized := normalizeFingerprints(values) if len(normalized) == 0 { return nil } - out := make([]coreworkspace.Fingerprint, 0, len(normalized)) + out := make([]Fingerprint, 0, len(normalized)) for _, value := range normalized { - out = append(out, coreworkspace.Fingerprint{Name: value.Name, Value: value.Value}) + out = append(out, Fingerprint{Name: value.Name, Value: value.Value}) } return out } @@ -356,7 +361,7 @@ func normalizeFingerprints(values []pipeline.CheckpointFingerprint) []pipeline.C return out } -func rejectionSummaries(rejected []contracts.RejectedOutput) []coreworkspace.RejectionSummary { +func rejectionSummaries(rejected []contracts.RejectedOutput) []RejectionSummary { if len(rejected) == 0 { return nil } @@ -383,9 +388,9 @@ func rejectionSummaries(rejected []contracts.RejectedOutput) []coreworkspace.Rej } return keys[i].message < keys[j].message }) - out := make([]coreworkspace.RejectionSummary, 0, len(keys)) + out := make([]RejectionSummary, 0, len(keys)) for _, k := range keys { - out = append(out, coreworkspace.RejectionSummary{ + out = append(out, RejectionSummary{ ValidatorName: k.validatorName, ReasonCode: k.reasonCode, Message: k.message, @@ -395,11 +400,11 @@ func rejectionSummaries(rejected []contracts.RejectedOutput) []coreworkspace.Rej return out } -func statusForRejected(rejected []contracts.RejectedOutput) coreworkspace.StageStatus { +func statusForRejected(rejected []contracts.RejectedOutput) StageStatus { if len(rejected) > 0 { - return coreworkspace.StatusSucceededWithRejections + return StatusSucceededWithRejections } - return coreworkspace.StatusSucceeded + return StatusSucceeded } func validationStatusString(warnings []contracts.Warning, rejected []contracts.RejectedOutput) string { diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 304b7e1..5297e7e 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -1,270 +1,51 @@ package checkpoint import ( - "encoding/json" "os" "path/filepath" - "strings" "testing" - "gitea.maximumdirect.net/eric/notarius/internal/core/source" - coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" - "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -func TestWorkspaceRecorderWritesSuccessfulCheckpointFiles(t *testing.T) { +func TestRootBasedRecorderOutputIsReusable(t *testing.T) { root := t.TempDir() - recorder := newTestRecorder(t, root) - doc := &source.SourceDocument{ - ID: "source-1", - Kind: "document", - Format: "text/plain", - Digest: "sha256:source", - Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}}, - } - if err := recorder.SourceRunning("seriatim"); err != nil { - t.Fatalf("SourceRunning: %v", err) - } - assertManifestStatus(t, filepath.Join(root, "source", "manifest.json"), coreworkspace.StatusRunning) - if err := recorder.SourceSucceeded("seriatim", doc); err != nil { - t.Fatalf("SourceSucceeded: %v", err) - } - assertManifestStatus(t, filepath.Join(root, "source", "manifest.json"), coreworkspace.StatusSucceeded) - if _, err := os.Stat(filepath.Join(root, "source", "source-document.json")); err != nil { - t.Fatalf("expected source checkpoint payload: %v", err) - } -} - -func TestWorkspaceArtifactCheckpointsRoundTripCodecIdentityAndBytes(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - loader := &WorkspaceLoader{root: root} - schema := contracts.ArtifactSchema{ID: "dnd.spell_response", Name: "spell response", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)} - artifact := contracts.SerializedArtifact{Kind: "dnd.spells", Schema: schema, MediaType: "application/json", Content: []byte(`{"spell_casts":[]}`), Metadata: map[string]any{"spell_cast_count": float64(0)}} - stored := pipeline.CheckpointArtifact{LaneID: "spells", ModuleKey: "dnd/spells", SourceID: "source-1", ChunkID: "chunk-1", ChunkIndex: 2, ChunkRef: source.SourceRef{SourceID: "source-1", StartUnitID: 4, EndUnitID: 8}, Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(schema)} - - extractDeps := []pipeline.CheckpointFingerprint{{Name: "chunks", Value: "sha256:chunks"}} - if err := recorder.ExtractSucceeded("spells", "dnd/spells", extractDeps, []pipeline.CheckpointArtifact{stored}, nil, nil); err != nil { - t.Fatalf("ExtractSucceeded: %v", err) - } - extracted, decision := loader.Extract("spells", "dnd/spells", extractDeps) - if !decision.Reused || len(extracted.Outputs) != 1 { - t.Fatalf("extract decision=%#v checkpoint=%#v, want reused", decision, extracted) - } - got := extracted.Outputs[0] - if got.Artifact.Kind != artifact.Kind || got.Artifact.Schema.ID != schema.ID || got.Artifact.Schema.Version != schema.Version || got.SchemaDigest != stored.SchemaDigest || string(got.Artifact.Content) != string(artifact.Content) || got.ChunkRef != stored.ChunkRef { - t.Fatalf("artifact checkpoint = %#v, want codec identity, bytes, and provenance", got) - } - - mergeDeps := artifactOutputDigests([]pipeline.CheckpointArtifact{stored}) - if err := recorder.MergeSucceeded("spells", "merge", mergeDeps, stored, nil); err != nil { - t.Fatalf("MergeSucceeded: %v", err) - } - merged, decision := loader.Merge("spells", "merge", mergeDeps) - if !decision.Reused || string(merged.Output.Artifact.Content) != string(artifact.Content) { - t.Fatalf("merge decision=%#v checkpoint=%#v, want reused", decision, merged) - } - - normalizeDeps := artifactOutputDigests([]pipeline.CheckpointArtifact{stored}) - if err := recorder.NormalizeSucceeded("spells", "normalize", normalizeDeps, stored, nil); err != nil { - t.Fatalf("NormalizeSucceeded: %v", err) - } - normalized, decision := loader.Normalize("spells", "normalize", normalizeDeps) - if !decision.Reused || normalized.Output.SchemaDigest != stored.SchemaDigest { - t.Fatalf("normalize decision=%#v checkpoint=%#v, want reused", decision, normalized) - } -} - -func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *testing.T) { - t.Run("missing", func(t *testing.T) { - loader := &WorkspaceLoader{root: t.TempDir()} - if _, decision := loader.Source("seriatim"); decision.Reused || !strings.Contains(decision.Reason, "missing") { - t.Fatalf("decision = %#v, want missing invalidation", decision) - } - }) - - t.Run("incompatible workspace schema remains untouched", func(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - doc := &source.SourceDocument{ - ID: "source-1", Kind: "document", Format: "text/plain", Digest: "sha256:source", - Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}}, - } - if err := recorder.SourceSucceeded("seriatim", doc); err != nil { - t.Fatalf("SourceSucceeded: %v", err) - } - manifestPath := filepath.Join(root, "source", "manifest.json") - manifest := strings.Replace(string(readFile(t, manifestPath)), coreworkspace.WorkspaceSchemaVersion, coreworkspace.WorkspaceSchemaVersionV1, 1) - if err := os.WriteFile(manifestPath, []byte(manifest), 0o644); err != nil { - t.Fatalf("write legacy manifest: %v", err) - } - beforeManifest := readFile(t, manifestPath) - payloadPath := filepath.Join(root, "source", "source-document.json") - beforePayload := readFile(t, payloadPath) - - loader := &WorkspaceLoader{root: root} - if _, decision := loader.Source("seriatim"); decision.Reused || !strings.Contains(decision.Reason, "incompatible") || !strings.Contains(decision.Reason, coreworkspace.WorkspaceSchemaVersionV1) { - t.Fatalf("decision = %#v, want incompatible legacy schema invalidation", decision) - } - if got := readFile(t, manifestPath); string(got) != string(beforeManifest) { - t.Fatal("legacy manifest changed during reuse decision") - } - if got := readFile(t, payloadPath); string(got) != string(beforePayload) { - t.Fatal("legacy payload changed during reuse decision") - } - }) - -} - -func TestWorkspaceCheckpointsIgnoreLegacyChunkFiles(t *testing.T) { - root := t.TempDir() - legacyManifest := []byte(`{"legacy":"manifest"}`) - legacyPayload := []byte(`{"legacy":"chunks"}`) - legacyDir := filepath.Join(root, "chunk") - if err := os.MkdirAll(legacyDir, 0o700); err != nil { - t.Fatal(err) - } - manifestPath := filepath.Join(legacyDir, "manifest.json") - payloadPath := filepath.Join(legacyDir, "chunks.json") - if err := os.WriteFile(manifestPath, legacyManifest, 0o600); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(payloadPath, legacyPayload, 0o600); err != nil { - t.Fatal(err) - } - - doc := &source.SourceDocument{ID: "source-1", Kind: "document", Format: "text/plain", Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}}} - doc.Digest, _ = source.DigestDocument(doc) - recorder := newTestRecorder(t, root) - if err := recorder.SourceSucceeded("seriatim", doc); err != nil { - t.Fatal(err) - } - loaded, decision := (&WorkspaceLoader{root: root}).Source("seriatim") - if !decision.Reused || loaded.Document == nil || loaded.Document.Digest != doc.Digest { - t.Fatalf("source checkpoint = %#v decision = %#v", loaded, decision) - } - if got := readFile(t, manifestPath); string(got) != string(legacyManifest) { - t.Fatalf("legacy manifest changed: %s", got) - } - if got := readFile(t, payloadPath); string(got) != string(legacyPayload) { - t.Fatalf("legacy payload changed: %s", got) - } -} - -func TestWorkspaceRecorderRecordsRejectedExtractOutputs(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - rejected := []contracts.RejectedOutput{ - { - Stage: string(pipeline.StageExtract), - LaneID: "spells", - ModuleKey: "dnd/spells", - ChunkID: "chunk-1", - ValidatorName: "shape", - ReasonCode: "invalid_shape", - Message: "bad shape", - }, - } - - if err := recorder.ExtractRunning("spells", "dnd/spells", []pipeline.CheckpointFingerprint{{Name: "chunks", Value: "sha256:chunks"}}); err != nil { - t.Fatalf("ExtractRunning: %v", err) - } - if err := recorder.ExtractSucceeded("spells", "dnd/spells", nil, nil, rejected, nil); err != nil { - t.Fatalf("ExtractSucceeded: %v", err) - } - - var manifest coreworkspace.ExtractLaneManifest - readJSON(t, filepath.Join(root, "extract", "spells", "manifest.json"), &manifest) - if manifest.Status != coreworkspace.StatusSucceededWithRejections || manifest.ValidationStatus != "rejected" { - t.Fatalf("extract manifest status = %q validation=%q", manifest.Status, manifest.ValidationStatus) - } - if len(manifest.Rejections) != 1 || manifest.Rejections[0].Count != 1 || manifest.Rejections[0].ReasonCode != "invalid_shape" { - t.Fatalf("rejections = %#v", manifest.Rejections) - } - var payload struct { - Rejected []contracts.RejectedOutput `json:"rejected"` - } - readJSON(t, filepath.Join(root, "extract", "spells", "outputs.json"), &payload) - if len(payload.Rejected) != 1 || payload.Rejected[0].ChunkID != "chunk-1" { - t.Fatalf("checkpoint rejected payload = %#v", payload.Rejected) - } -} - -func TestWorkspaceRecorderRecordsFailedStages(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - - if err := recorder.MergeRunning("spells", "appendorder", nil); err != nil { - t.Fatalf("MergeRunning: %v", err) - } - if err := recorder.MergeFailed("spells", "appendorder", nil, assertErr("merge failed")); err != nil { - t.Fatalf("MergeFailed: %v", err) - } - - var manifest coreworkspace.MergeLaneManifest - readJSON(t, filepath.Join(root, "merge", "spells", "manifest.json"), &manifest) - if manifest.Status != coreworkspace.StatusFailed { - t.Fatalf("status = %q, want failed", manifest.Status) - } - if !strings.Contains(manifest.Metadata["error"], "merge failed") { - t.Fatalf("metadata = %#v, want error", manifest.Metadata) - } -} - -func TestWorkspaceRecorderRecordsWarningOnlyValidation(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - schema := contracts.ArtifactSchema{ID: "test.artifact", Name: "test_artifact", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)} - output := pipeline.CheckpointArtifact{ - LaneID: "events", ModuleKey: "noop", SourceID: "source-1", - Artifact: contracts.SerializedArtifact{Kind: "test/artifact", Schema: schema, MediaType: "application/json", Content: []byte(`{"ok":true}`)}, - SchemaDigest: contracts.DigestArtifactSchema(schema), - } - warnings := []contracts.Warning{{ReasonCode: "note", Message: "warning"}} - - if err := recorder.NormalizeSucceeded("events", "noop", nil, output, warnings); err != nil { - t.Fatalf("NormalizeSucceeded: %v", err) - } - - var manifest coreworkspace.NormalizeLaneManifest - readJSON(t, filepath.Join(root, "normalize", "events", "manifest.json"), &manifest) - if manifest.Status != coreworkspace.StatusSucceeded || manifest.ValidationStatus != "approved_with_warnings" { - t.Fatalf("normalize manifest status = %q validation=%q", manifest.Status, manifest.ValidationStatus) - } -} - -func newTestRecorder(t *testing.T, root string) *WorkspaceRecorder { - t.Helper() - return &WorkspaceRecorder{root: root} -} - -func assertManifestStatus(t *testing.T, path string, want coreworkspace.StageStatus) { - t.Helper() - var manifest coreworkspace.StageManifest - readJSON(t, path, &manifest) - if manifest.Status != want { - t.Fatalf("%s status = %q, want %q", path, manifest.Status, want) - } -} - -func readJSON(t *testing.T, path string, out any) { - t.Helper() - data := readFile(t, path) - if err := json.Unmarshal(data, out); err != nil { - t.Fatalf("decode %q: %v", path, err) - } -} - -func readFile(t *testing.T, path string) []byte { - t.Helper() - data, err := os.ReadFile(path) + identity := testIdentity(t) + recorder, err := NewFilesystemRecorder(root, identity) if err != nil { - t.Fatalf("read %q: %v", path, err) + t.Fatal(err) + } + if err := recorder.ExtractSucceeded("lane", "module", nil, nil, nil, nil); err != nil { + t.Fatal(err) + } + loader, err := NewFilesystemLoader(root, identity) + if err != nil { + t.Fatal(err) + } + result, decision := loader.Extract("lane", "module", nil) + if !decision.Reused || len(result.Outputs) != 0 { + t.Fatalf("load result=%#v decision=%#v", result, decision) + } + relative, err := identity.RelativePath() + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(root, relative, "extract", "lane", "manifest.json")); err != nil { + t.Fatal(err) } - return data } -type assertErr string +func TestCheckpointSchemaCompatibilityIsUnchanged(t *testing.T) { + if WorkspaceSchemaVersion != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { + t.Fatal("checkpoint schema identifiers changed") + } +} -func (e assertErr) Error() string { return string(e) } +func testIdentity(t *testing.T) Identity { + t.Helper() + identity, err := NewIdentity(IdentityInput{Pipeline: pipeline.ResolvedPipeline{ID: "pipeline", Digest: "sha256:aaaaaaaaaaaaaaaa", Input: pipeline.Binding("input")}, RawInputDigest: "sha256:bbbbbbbbbbbbbbbb"}) + if err != nil { + t.Fatal(err) + } + return identity +}