package checkpoint import ( "encoding/base64" "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) { 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}}}, } chunks := []contracts.SourceChunk{ { ID: "chunk-1", SourceID: "source-1", Index: 0, StartUnitID: 1, EndUnitID: 1, Content: []byte("chunk content"), MediaType: "text/plain", Units: doc.Units, }, } 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) } if err := recorder.ChunkRunning("generic", doc.Digest); err != nil { t.Fatalf("ChunkRunning: %v", err) } if err := recorder.ChunkSucceeded("generic", doc.Digest, chunks, nil); err != nil { t.Fatalf("ChunkSucceeded: %v", err) } assertManifestStatus(t, filepath.Join(root, "chunk", "manifest.json"), coreworkspace.StatusSucceeded) var chunkPayload struct { Chunks []struct { Content struct { ContentBase64 string `json:"content_base64"` ContentDigest string `json:"content_digest"` } `json:"content"` } `json:"chunks"` } readJSON(t, filepath.Join(root, "chunk", "chunks.json"), &chunkPayload) if len(chunkPayload.Chunks) != 1 { t.Fatalf("checkpoint chunks = %#v, want one", chunkPayload.Chunks) } decoded, err := base64.StdEncoding.DecodeString(chunkPayload.Chunks[0].Content.ContentBase64) if err != nil { t.Fatalf("decode chunk content: %v", err) } if string(decoded) != "chunk content" { t.Fatalf("chunk content = %q, want original content", decoded) } if got, want := chunkPayload.Chunks[0].Content.ContentDigest, contentDigest([]byte("chunk content")); got != want { t.Fatalf("content digest = %q, want %q", got, want) } } func TestWorkspaceLoaderReusesSuccessfulCheckpointFiles(t *testing.T) { root := t.TempDir() recorder := newTestRecorder(t, root) loader := &WorkspaceLoader{root: 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}}}, } chunks := []contracts.SourceChunk{ { ID: "chunk-1", SourceID: "source-1", Index: 0, StartUnitID: 1, EndUnitID: 1, Content: []byte("chunk content"), MediaType: "text/plain", Units: doc.Units, }, } extractOutput := contracts.ExtractOutput{ LaneID: "spells", ExtractorKey: "dnd/spells", SourceID: doc.ID, ChunkID: "chunk-1", Payload: contracts.RawPayload{ Content: []byte(`{"spell":"cure wounds"}`), MediaType: "application/json", }, } mergeOutput := contracts.MergeOutput{ LaneID: "spells", MergerKey: "appendorder", SourceID: doc.ID, Payload: contracts.RawPayload{ Content: []byte(`{"merged":true}`), MediaType: "application/json", }, } normalizeOutput := contracts.NormalizeOutput{ LaneID: "spells", NormalizerKey: "noop", SourceID: doc.ID, Payload: contracts.RawPayload{ Content: []byte(`{"normalized":true}`), MediaType: "application/json", }, } if err := recorder.SourceSucceeded("seriatim", doc); err != nil { t.Fatalf("SourceSucceeded: %v", err) } if err := recorder.ChunkSucceeded("generic", doc.Digest, chunks, nil); err != nil { t.Fatalf("ChunkSucceeded: %v", err) } extractDeps := []pipeline.CheckpointFingerprint{{Name: "chunks", Value: "sha256:chunks"}} if err := recorder.ExtractSucceeded("spells", "dnd/spells", extractDeps, []contracts.ExtractOutput{extractOutput}, nil, nil); err != nil { t.Fatalf("ExtractSucceeded: %v", err) } mergeDeps := rawOutputDigests([]contracts.RawPayload{extractOutput.Payload}) if err := recorder.MergeSucceeded("spells", "appendorder", mergeDeps, mergeOutput, nil); err != nil { t.Fatalf("MergeSucceeded: %v", err) } normalizeDeps := rawOutputDigests([]contracts.RawPayload{mergeOutput.Payload}) if err := recorder.NormalizeSucceeded("spells", "noop", normalizeDeps, normalizeOutput, nil); err != nil { t.Fatalf("NormalizeSucceeded: %v", err) } sourceCheckpoint, decision := loader.Source("seriatim") if !decision.Reused || sourceCheckpoint.Document.ID != "source-1" { t.Fatalf("source decision = %#v checkpoint=%#v, want reused", decision, sourceCheckpoint) } if got, want := sourceCheckpoint.Document.Units[0].Ref, doc.Units[0].Ref; got != want { t.Fatalf("checkpoint source unit ref = %#v, want %#v", got, want) } chunkCheckpoint, decision := loader.Chunk("generic", doc.Digest) if !decision.Reused || len(chunkCheckpoint.Chunks) != 1 || string(chunkCheckpoint.Chunks[0].Content) != "chunk content" { t.Fatalf("chunk decision = %#v checkpoint=%#v, want reused", decision, chunkCheckpoint) } extractCheckpoint, decision := loader.Extract("spells", "dnd/spells", extractDeps) if !decision.Reused || len(extractCheckpoint.Outputs) != 1 || string(extractCheckpoint.Outputs[0].Payload.Content) != `{"spell":"cure wounds"}` { t.Fatalf("extract decision = %#v checkpoint=%#v, want reused", decision, extractCheckpoint) } mergeCheckpoint, decision := loader.Merge("spells", "appendorder", mergeDeps) if !decision.Reused || string(mergeCheckpoint.Output.Payload.Content) != `{"merged":true}` { t.Fatalf("merge decision = %#v checkpoint=%#v, want reused", decision, mergeCheckpoint) } normalizeCheckpoint, decision := loader.Normalize("spells", "noop", normalizeDeps) if !decision.Reused || string(normalizeCheckpoint.Output.Payload.Content) != `{"normalized":true}` { t.Fatalf("normalize decision = %#v checkpoint=%#v, want reused", decision, normalizeCheckpoint) } } 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("dependency mismatch", func(t *testing.T) { root := t.TempDir() recorder := newTestRecorder(t, root) chunks := []contracts.SourceChunk{{ ID: "chunk-1", SourceID: "source-1", Content: []byte("chunk content"), MediaType: "text/plain", }} if err := recorder.ChunkSucceeded("generic", "sha256:source-a", chunks, nil); err != nil { t.Fatalf("ChunkSucceeded: %v", err) } loader := &WorkspaceLoader{root: root} if _, decision := loader.Chunk("generic", "sha256:source-b"); decision.Reused || !strings.Contains(decision.Reason, "dependency") { t.Fatalf("decision = %#v, want dependency invalidation", decision) } }) t.Run("corrupt payload", func(t *testing.T) { root := t.TempDir() recorder := newTestRecorder(t, root) chunks := []contracts.SourceChunk{{ ID: "chunk-1", SourceID: "source-1", Content: []byte("chunk content"), MediaType: "text/plain", }} if err := recorder.ChunkSucceeded("generic", "sha256:source", chunks, nil); err != nil { t.Fatalf("ChunkSucceeded: %v", err) } payloadPath := filepath.Join(root, "chunk", "chunks.json") data := strings.ReplaceAll(string(readFile(t, payloadPath)), contentDigest([]byte("chunk content")), "sha256:bad") if err := os.WriteFile(payloadPath, []byte(data), 0o644); err != nil { t.Fatalf("corrupt chunk payload: %v", err) } loader := &WorkspaceLoader{root: root} if _, decision := loader.Chunk("generic", "sha256:source"); decision.Reused || !strings.Contains(decision.Reason, "invalid") { t.Fatalf("decision = %#v, want corrupt payload invalidation", decision) } }) } 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) output := contracts.NormalizeOutput{ LaneID: "spells", NormalizerKey: "noop", SourceID: "source-1", Payload: contracts.RawPayload{ Content: []byte(`{"ok":true}`), MediaType: "application/json", }, } warnings := []contracts.Warning{{ReasonCode: "note", Message: "warning"}} if err := recorder.NormalizeSucceeded("spells", "noop", nil, output, warnings); err != nil { t.Fatalf("NormalizeSucceeded: %v", err) } var manifest coreworkspace.NormalizeLaneManifest readJSON(t, filepath.Join(root, "normalize", "spells", "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) if err != nil { t.Fatalf("read %q: %v", path, err) } return data } type assertErr string func (e assertErr) Error() string { return string(e) }