diff --git a/docs/internal/overview.md b/docs/internal/overview.md index a58104f..c56196e 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -30,8 +30,9 @@ belongs in modules, not in command handlers. - `internal/core/source`: source documents, source units, source references, and validation. - `internal/core/workspace`: effective workspace roots, enabled-state helpers, - safe workspace-relative path construction, and atomic workspace artifact - writes. + safe workspace-relative path construction, atomic workspace artifact writes, + checkpoint identities, checkpoint path construction, and checkpoint manifest + types. Core packages should remain deterministic and concrete. They should not import production modules. diff --git a/internal/core/workspace/identity.go b/internal/core/workspace/identity.go new file mode 100644 index 0000000..bf17c7d --- /dev/null +++ b/internal/core/workspace/identity.go @@ -0,0 +1,268 @@ +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) + } + return filepath.ToSlash(filepath.Join(pipelineID, inputKey+"-"+sourceComponent, pipelineComponent)), 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 new file mode 100644 index 0000000..cf3c760 --- /dev/null +++ b/internal/core/workspace/identity_test.go @@ -0,0 +1,242 @@ +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) + } + if relative != "campaign~2fmain/seriatim~2finput-abcdef0123456789/1234567890abcdef" { + 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") + if got != want { + t.Fatalf("checkpoint directory = %q, want %q", got, want) + } +} + +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.go b/internal/core/workspace/manifest.go new file mode 100644 index 0000000..84cdc12 --- /dev/null +++ b/internal/core/workspace/manifest.go @@ -0,0 +1,82 @@ +package workspace + +import "time" + +const WorkspaceSchemaVersion = "notarius.workspace.v1" + +type StageName string + +const ( + StageSource StageName = "source" + StageChunk StageName = "chunk" + StageExtract StageName = "extract" + StageMerge StageName = "merge" + StageNormalize StageName = "normalize" +) + +type StageStatus string + +const ( + StatusPending StageStatus = "pending" + StatusRunning StageStatus = "running" + StatusSucceeded StageStatus = "succeeded" + StatusSucceededWithRejections StageStatus = "succeeded_with_rejections" + StatusFailed StageStatus = "failed" + StatusInvalidated StageStatus = "invalidated" +) + +type StageManifest struct { + WorkspaceSchemaVersion string `json:"workspace_schema_version"` + Stage StageName `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + DependencyFingerprints []Fingerprint `json:"dependency_fingerprints,omitempty"` + Status StageStatus `json:"status"` + OutputDigests []Fingerprint `json:"output_digests,omitempty"` + ValidationStatus string `json:"validation_status,omitempty"` + Rejections []RejectionSummary `json:"rejections,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + 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 ChunkManifest struct { + StageManifest + ChunkCount int `json:"chunk_count,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, + } +} diff --git a/internal/core/workspace/manifest_test.go b/internal/core/workspace/manifest_test.go new file mode 100644 index 0000000..2990b7f --- /dev/null +++ b/internal/core/workspace/manifest_test.go @@ -0,0 +1,143 @@ +package workspace + +import ( + "encoding/json" + "testing" + "time" +) + +func TestStageManifestDefaults(t *testing.T) { + 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("chunk", func(t *testing.T) { + manifest := ChunkManifest{ + StageManifest: populatedManifest(StageChunk, "", "generic", started, completed), + ChunkCount: 3, + } + var got ChunkManifest + roundTripManifest(t, manifest, &got) + if got.ChunkCount != manifest.ChunkCount || got.Stage != StageChunk { + t.Fatalf("round trip chunk 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) + } +}