diff --git a/docs/config.md b/docs/config.md index cb6ac35..5c4d99f 100644 --- a/docs/config.md +++ b/docs/config.md @@ -352,8 +352,11 @@ casts still must be present in the source transcript. - `resume.enabled`: boolean resume checkpointing setting. Default: `false`. - `debug.enabled`: boolean debug artifact setting. Default: `false`. -The current run workflow uses workspace settings for diagnostics configuration. -Checkpoint and debug artifact writers are not part of the current workflow. +When `workspace.resume.enabled` is true, runs write stage-owned checkpoint +artifacts under `/checkpoints/`. Checkpoint reads and +resume execution are not implemented. + +Debug artifact writers are not part of the current workflow. ## Diagnostics diff --git a/docs/internal/overview.md b/docs/internal/overview.md index c56196e..bf11c54 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -42,9 +42,11 @@ production modules. - `internal/framework/contracts`: interfaces and request/result structs for input adapters, chunkers, extractors, mergers, normalizers, validators, output encoders, and structured LLM clients. +- `internal/framework/checkpoint`: workspace-backed checkpoint recorder and + checkpoint payload envelope serialization. - `internal/framework/pipeline`: module registries, module specs, profile - resolution, capability checks, run orchestration, warnings, validation, and - manifest population. + resolution, capability checks, run orchestration, checkpoint recorder + boundaries, warnings, validation, and manifest population. - `internal/framework/llm`: Scriptorium-backed structured-output client, prompt/schema asset registry, scheduler, schema registry, and secret redaction. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index a9aa414..4c3f339 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -71,6 +71,12 @@ to chunk, extract, merge, and normalize requests; LLM-backed modules forward it through their structured completion requests so Scriptorium can include it in prompt execution metadata. +When workspace resume checkpointing is enabled, the CLI constructs a checkpoint +recorder after pipeline resolution and reference materialization and passes it +through `pipeline.RunInput`. The runner records source, chunk, extract, merge, +and normalize outcomes through that interface. Concrete modules do not receive +workspace paths and do not write checkpoint files directly. + ## Registries And Module Specs `pipeline.Registries` holds concrete constructors for execution. A diff --git a/docs/operations.md b/docs/operations.md index 89326bd..b111fb3 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -84,6 +84,24 @@ Implemented diagnostics artifacts: `source-document.json` is supported by the diagnostics writer but is not written by the current CLI run workflow. +## Checkpoints + +When `workspace.resume.enabled: true` and `workspace.directory` is set, runs +write checkpoints under: + +```text +/checkpoints//-// +``` + +Each workflow step owns its own manifest and payload files. There is no +root-level checkpoint summary. Current runs write checkpoints for inspection and +future recovery support only; the CLI does not read checkpoints or skip work. + +Checkpoint payloads preserve byte content with base64 envelopes, media type, +metadata, warnings, and content digests where applicable. Checkpoints do not +include raw prompts, raw reference contents, raw LLM request payloads, or debug +traces. + ## Retention Diagnostics retention is configured with `workspace.diagnostics.retention`, @@ -138,8 +156,8 @@ directories unless they are part of your own operational policy. ## Operational Limits -There is no command to resume a failed run. Re-run `notarius run` after fixing -the cause. +Checkpoint writing does not provide resume execution yet. Re-run `notarius run` +after fixing the cause of a failed run. Provider retries and timeouts are handled by Scriptorium according to the selected execution profile. Pipeline module retries are controlled by module diff --git a/internal/cli/run.go b/internal/cli/run.go index 4f3a690..33674e3 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -2,6 +2,8 @@ package cli import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "flag" "fmt" @@ -17,6 +19,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/core/config" "gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics" "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" + "gitea.maximumdirect.net/eric/notarius/internal/framework/checkpoint" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) @@ -253,6 +256,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create LLM client for profile %q: %w", factoryProfileID, err)) } + checkpointRecorder, err := checkpointRecorderForRun(workspaceSettings, effective.ResolvedPipeline, rawInput, only, llmProfiles, strings.TrimSpace(*llmProfile), strings.TrimSpace(sessionID.value)) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) + } output, err := pipeline.New(registries).Run(ctx, pipeline.RunInput{ Pipeline: effective.ResolvedPipeline, @@ -265,6 +272,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i LLMProfiles: llmProfiles, Metadata: runMetadata(*outputDir, *diagnosticsDir), Warnings: referenceWarnings, + Checkpoints: checkpointRecorder, }) if err != nil { if output.Manifest.PipelineID != "" && runDir != nil { @@ -348,6 +356,68 @@ func writeDiagnostics(runDir *diagnostics.RunDirectory, write func() error) erro return write() } +func checkpointRecorderForRun( + settings workspace.Settings, + resolved pipeline.ResolvedPipeline, + rawInput []byte, + only []string, + llmProfiles []artifacts.LLMProfileManifest, + llmProfileOverride string, + sessionID string, +) (pipeline.CheckpointRecorder, error) { + identity, err := workspace.NewCheckpointIdentity(workspace.CheckpointIdentityInput{ + Pipeline: resolved, + InputKey: resolved.Input.Module, + RawInputDigest: rawInputDigest(rawInput), + SelectedLanes: only, + RuntimeOverrides: runtimeOverrideFingerprints(llmProfileOverride, sessionID), + References: pipeline.ReferenceProvenance(resolved), + ProvenanceFingerprints: llmProfileFingerprints(llmProfiles), + }) + if err != nil { + return nil, fmt.Errorf("create checkpoint identity: %w", err) + } + recorder, err := checkpoint.NewWorkspaceRecorder(settings, identity) + if err != nil { + return nil, fmt.Errorf("create checkpoint recorder: %w", err) + } + return recorder, nil +} + +func rawInputDigest(data []byte) string { + sum := sha256.Sum256(data) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func runtimeOverrideFingerprints(llmProfileOverride string, sessionID string) []workspace.Fingerprint { + var values []workspace.Fingerprint + if strings.TrimSpace(llmProfileOverride) != "" { + values = append(values, workspace.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)}) + } + return values +} + +func llmProfileFingerprints(profiles []artifacts.LLMProfileManifest) []workspace.Fingerprint { + if len(profiles) == 0 { + return nil + } + values := make([]workspace.Fingerprint, 0, len(profiles)) + for _, profile := range profiles { + id := strings.TrimSpace(profile.ID) + if id == "" { + continue + } + values = append(values, workspace.Fingerprint{ + Name: "llm_profile:" + id, + Value: strings.TrimSpace(profile.Provider) + ":" + strings.TrimSpace(profile.Model), + }) + } + return values +} + func configSource(configPath string) string { if strings.TrimSpace(configPath) != "" { return "flag" diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index afd6aee..f2fcf57 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2180,6 +2180,41 @@ func TestRunPipelineWritesWorkspaceDiagnosticsArtifactsOnSuccess(t *testing.T) { assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) } +func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + checkpointDir := onlyCheckpointIdentityDir(t, workspaceDir) + for _, name := range []string{ + "source/manifest.json", + "source/source-document.json", + "chunk/manifest.json", + "chunk/chunks.json", + "extract/spells/manifest.json", + "extract/spells/outputs.json", + "merge/spells/manifest.json", + "merge/spells/output.json", + "normalize/spells/manifest.json", + "normalize/spells/output.json", + } { + if _, err := os.Stat(filepath.Join(checkpointDir, name)); err != nil { + t.Fatalf("expected checkpoint artifact %q: %v", name, err) + } + } + assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) +} + func TestRunPipelineSkipsDiagnosticsWhenWorkspaceDiagnosticsDisabled(t *testing.T) { workspaceDir := filepath.Join(t.TempDir(), "workspace") outputDir := t.TempDir() @@ -2335,7 +2370,9 @@ func TestRunPipelineDiagnosticsDirFlagOverridesWorkspaceDiagnosticsOnly(t *testi t.Fatalf("override diagnostics dir entries = %v, want one run dir", entries) } assertPathNotExist(t, filepath.Join(workspaceDir, "diagnostics")) - assertPathNotExist(t, filepath.Join(workspaceDir, "checkpoints")) + if entries := childDirs(t, filepath.Join(workspaceDir, "checkpoints")); len(entries) != 1 { + t.Fatalf("workspace checkpoint pipeline dirs = %v, want one", entries) + } assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) } @@ -2868,6 +2905,24 @@ pipelines: ` } +func mvpConfigYAMLWithWorkspaceResumeEnabled(pipelineID, workspaceDir, retention string) string { + return `version: 2 +workspace: + directory: ` + workspaceDir + ` + diagnostics: + enabled: true + retention: ` + retention + ` + resume: + enabled: true +pipelines: + ` + pipelineID + `: + input: seriatim + artifacts: + spells: + extract: dnd/spells +` +} + func mvpConfigYAMLWithWorkspaceDiagnosticsDisabled(pipelineID, workspaceDir string) string { return `version: 2 workspace: @@ -3231,6 +3286,13 @@ func onlyChildDir(t *testing.T, root string) string { return children[0] } +func onlyCheckpointIdentityDir(t *testing.T, workspaceDir string) string { + t.Helper() + pipelineDir := onlyChildDir(t, filepath.Join(workspaceDir, "checkpoints")) + inputDir := onlyChildDir(t, pipelineDir) + return onlyChildDir(t, inputDir) +} + func childDirs(t *testing.T, root string) []string { t.Helper() entries, err := os.ReadDir(root) diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go new file mode 100644 index 0000000..b328342 --- /dev/null +++ b/internal/framework/checkpoint/recorder.go @@ -0,0 +1,625 @@ +package checkpoint + +import ( + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "fmt" + "path" + "sort" + "strings" + "time" + + "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 { + root string + now func() time.Time +} + +func NewWorkspaceRecorder(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointRecorder, error) { + root, err := settings.CheckpointDirectory(identity) + if err != nil { + return nil, err + } + if strings.TrimSpace(root) == "" { + return pipeline.NoopCheckpointRecorder(), nil + } + return &WorkspaceRecorder{root: root, now: time.Now}, nil +} + +func (r *WorkspaceRecorder) SourceRunning(moduleKey string) error { + manifest := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusRunning) + manifest.ModuleKey = moduleKey + manifest.StartedAt = timePtr(r.timestamp()) + return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) 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 := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusSucceeded) + manifest.ModuleKey = moduleKey + manifest.OutputDigests = workspaceFingerprints(digestFingerprints("source_document", doc.Digest)) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{ + StageManifest: manifest, + SourceID: doc.ID, + }) +} + +func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error { + manifest := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusFailed) + manifest.ModuleKey = moduleKey + manifest.CompletedAt = timePtr(r.timestamp()) + manifest.Metadata = errorMetadata(err) + return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) ChunkRunning(moduleKey string, sourceDigest string) error { + manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusRunning) + manifest.ModuleKey = moduleKey + manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) + manifest.StartedAt = timePtr(r.timestamp()) + return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string, chunks []contracts.SourceChunk, warnings []contracts.Warning) error { + payload := chunksEnvelope{Chunks: chunkEnvelopes(chunks), Warnings: cloneWarnings(warnings)} + if err := r.writePayload("chunk/chunks.json", payload); err != nil { + return err + } + manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded) + manifest.ModuleKey = moduleKey + manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) + manifest.OutputDigests = workspaceFingerprints(chunkOutputDigests(chunks)) + manifest.ValidationStatus = validationStatusString(warnings, nil) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{ + StageManifest: manifest, + ChunkCount: len(chunks), + }) +} + +func (r *WorkspaceRecorder) ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error { + manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceededWithRejections) + manifest.ModuleKey = moduleKey + manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) + manifest.ValidationStatus = "rejected" + manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) ChunkFailed(moduleKey string, sourceDigest string, err error) error { + manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusFailed) + manifest.ModuleKey = moduleKey + manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) + manifest.CompletedAt = timePtr(r.timestamp()) + manifest.Metadata = errorMetadata(err) + return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) + manifest.StartedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) ExtractSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []contracts.ExtractOutput, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + payload := extractOutputsEnvelope{ + Outputs: extractOutputEnvelopes(outputs), + Rejected: cloneRejectedOutputs(rejected), + Warnings: cloneWarnings(warnings), + } + if err := r.writePayload(lanePayloadPath("extract", laneID, "outputs.json"), payload); err != nil { + return err + } + manifest := laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(rawOutputDigests(extractPayloads(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), + }) +} + +func (r *WorkspaceRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := laneManifest(coreworkspace.StageExtract, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest.CompletedAt = timePtr(r.timestamp()) + manifest.Metadata = errorMetadata(err) + return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) MergeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) + manifest.StartedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) MergeSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output contracts.MergeOutput, warnings []contracts.Warning) error { + payload := mergeOutputEnvelope{Output: mergeOutputEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)} + if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), payload); err != nil { + return err + } + manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(rawOutputDigests([]contracts.RawPayload{output.Payload})) + manifest.ValidationStatus = validationStatusString(warnings, nil) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{ + StageManifest: manifest, + InputCount: len(dependencies), + }) +} + +func (r *WorkspaceRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.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)}) +} + +func (r *WorkspaceRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest.CompletedAt = timePtr(r.timestamp()) + manifest.Metadata = errorMetadata(err) + return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) NormalizeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) + manifest.StartedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) NormalizeSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output contracts.NormalizeOutput, warnings []contracts.Warning) error { + payload := normalizeOutputEnvelope{Output: normalizeOutputEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)} + if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), payload); err != nil { + return err + } + manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(rawOutputDigests([]contracts.RawPayload{output.Payload})) + manifest.ValidationStatus = validationStatusString(warnings, nil) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) +} + +func (r *WorkspaceRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.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)}) +} + +func (r *WorkspaceRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest.CompletedAt = timePtr(r.timestamp()) + manifest.Metadata = errorMetadata(err) + return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest}) +} + +func (r *WorkspaceRecorder) writeManifest(name string, payload any) error { + return r.writeJSON(name, payload) +} + +func (r *WorkspaceRecorder) writePayload(name string, payload any) error { + return r.writeJSON(name, payload) +} + +func (r *WorkspaceRecorder) writeJSON(name string, payload any) error { + if r == nil || strings.TrimSpace(r.root) == "" { + return nil + } + return coreworkspace.WriteJSON(r.root, name, payload) +} + +func (r *WorkspaceRecorder) timestamp() time.Time { + if r == nil || r.now == nil { + return time.Now().UTC() + } + return r.now().UTC() +} + +func laneManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) coreworkspace.StageManifest { + manifest := coreworkspace.NewStageManifest(stage, status) + manifest.LaneID = laneID + manifest.ModuleKey = moduleKey + manifest.DependencyFingerprints = workspaceFingerprints(dependencies) + return manifest +} + +type sourceDocumentEnvelope struct { + Document source.SourceDocument `json:"document"` +} + +type chunksEnvelope struct { + Chunks []chunkEnvelope `json:"chunks"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +type chunkEnvelope struct { + ID string `json:"id"` + SourceID string `json:"source_id"` + Index int `json:"index"` + StartUnitID int `json:"start_unit_id"` + EndUnitID int `json:"end_unit_id"` + Content binaryEnvelope `json:"content"` + Units []source.SourceUnit `json:"units,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type extractOutputsEnvelope struct { + Outputs []extractOutputEnvelope `json:"outputs"` + Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +type extractOutputEnvelope struct { + LaneID string `json:"lane_id"` + ExtractorKey string `json:"extractor_key"` + SourceID string `json:"source_id"` + ChunkID string `json:"chunk_id"` + ChunkIndex int `json:"chunk_index"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload binaryEnvelope `json:"payload"` +} + +type mergeOutputEnvelope struct { + Output mergeOutputPayload `json:"output"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +type mergeOutputPayload struct { + LaneID string `json:"lane_id"` + MergerKey string `json:"merger_key"` + SourceID string `json:"source_id,omitempty"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload binaryEnvelope `json:"payload"` +} + +type normalizeOutputEnvelope struct { + Output normalizeOutputPayload `json:"output"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +type normalizeOutputPayload struct { + LaneID string `json:"lane_id"` + NormalizerKey string `json:"normalizer_key"` + SourceID string `json:"source_id,omitempty"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload binaryEnvelope `json:"payload"` +} + +type binaryEnvelope struct { + ContentBase64 string `json:"content_base64,omitempty"` + ContentDigest string `json:"content_digest,omitempty"` + MediaType string `json:"media_type,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +func chunkEnvelopes(chunks []contracts.SourceChunk) []chunkEnvelope { + if len(chunks) == 0 { + return nil + } + out := make([]chunkEnvelope, 0, len(chunks)) + for _, chunk := range chunks { + out = append(out, chunkEnvelope{ + ID: chunk.ID, + SourceID: chunk.SourceID, + Index: chunk.Index, + StartUnitID: chunk.StartUnitID, + EndUnitID: chunk.EndUnitID, + Content: binaryEnvelopeFromContent(chunk.Content, chunk.MediaType, chunk.Metadata, nil), + Units: cloneSourceUnits(chunk.Units), + Metadata: cloneMetadata(chunk.Metadata), + }) + } + return out +} + +func extractOutputEnvelopes(outputs []contracts.ExtractOutput) []extractOutputEnvelope { + if len(outputs) == 0 { + return nil + } + out := make([]extractOutputEnvelope, 0, len(outputs)) + for _, output := range outputs { + out = append(out, extractOutputEnvelope{ + LaneID: output.LaneID, + ExtractorKey: output.ExtractorKey, + SourceID: output.SourceID, + ChunkID: output.ChunkID, + ChunkIndex: output.ChunkIndex, + Schema: schemaEnvelope(output.Schema), + Payload: binaryEnvelopeFromPayload(output.Payload), + }) + } + return out +} + +func mergeOutputEnvelopeFromOutput(output contracts.MergeOutput) mergeOutputPayload { + return mergeOutputPayload{ + LaneID: output.LaneID, + MergerKey: output.MergerKey, + SourceID: output.SourceID, + Schema: schemaEnvelope(output.Schema), + Payload: binaryEnvelopeFromPayload(output.Payload), + } +} + +func normalizeOutputEnvelopeFromOutput(output contracts.NormalizeOutput) normalizeOutputPayload { + return normalizeOutputPayload{ + LaneID: output.LaneID, + NormalizerKey: output.NormalizerKey, + SourceID: output.SourceID, + Schema: schemaEnvelope(output.Schema), + Payload: binaryEnvelopeFromPayload(output.Payload), + } +} + +func schemaEnvelope(schema contracts.ResponseSchema) contracts.ResponseSchema { + schema.JSONSchema = nil + return schema +} + +func binaryEnvelopeFromPayload(payload contracts.RawPayload) binaryEnvelope { + return binaryEnvelopeFromContent(payload.Content, payload.MediaType, payload.Metadata, payload.Warnings) +} + +func binaryEnvelopeFromContent(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) binaryEnvelope { + return binaryEnvelope{ + ContentBase64: base64.StdEncoding.EncodeToString(content), + ContentDigest: contentDigest(content), + MediaType: mediaType, + Metadata: cloneMetadata(metadata), + Warnings: cloneWarnings(warnings), + } +} + +func cloneSourceDocument(doc source.SourceDocument) source.SourceDocument { + doc.Units = cloneSourceUnits(doc.Units) + doc.Metadata = cloneMetadata(doc.Metadata) + return doc +} + +func cloneSourceUnits(units []source.SourceUnit) []source.SourceUnit { + if len(units) == 0 { + return nil + } + out := make([]source.SourceUnit, 0, len(units)) + for _, unit := range units { + out = append(out, source.SourceUnit{ + ID: unit.ID, + Kind: unit.Kind, + Text: unit.Text, + Metadata: cloneMetadata(unit.Metadata), + }) + } + return out +} + +func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { + if len(warnings) == 0 { + return nil + } + return append([]contracts.Warning(nil), warnings...) +} + +func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput { + if len(rejected) == 0 { + return nil + } + return append([]contracts.RejectedOutput(nil), rejected...) +} + +func cloneMetadata(metadata map[string]any) map[string]any { + if len(metadata) == 0 { + return nil + } + out := make(map[string]any, len(metadata)) + for key, value := range metadata { + out[key] = value + } + return out +} + +func rawOutputDigests(payloads []contracts.RawPayload) []pipeline.CheckpointFingerprint { + values := make([]pipeline.CheckpointFingerprint, 0, len(payloads)) + for i, payload := range payloads { + values = append(values, pipeline.CheckpointFingerprint{ + Name: fmt.Sprintf("payload[%d]", i), + Value: contentDigest(payload.Content), + }) + } + return normalizeFingerprints(values) +} + +func extractPayloads(outputs []contracts.ExtractOutput) []contracts.RawPayload { + if len(outputs) == 0 { + return nil + } + payloads := make([]contracts.RawPayload, 0, len(outputs)) + for _, output := range outputs { + payloads = append(payloads, output.Payload) + } + return payloads +} + +func chunkOutputDigests(chunks []contracts.SourceChunk) []pipeline.CheckpointFingerprint { + values := make([]pipeline.CheckpointFingerprint, 0, len(chunks)) + for _, chunk := range chunks { + values = append(values, pipeline.CheckpointFingerprint{ + Name: chunk.ID, + Value: contentDigest(chunk.Content), + }) + } + return normalizeFingerprints(values) +} + +func digestFingerprints(name string, digest string) []pipeline.CheckpointFingerprint { + digest = strings.TrimSpace(digest) + if digest == "" { + return nil + } + return []pipeline.CheckpointFingerprint{{Name: name, Value: digest}} +} + +func workspaceFingerprints(values []pipeline.CheckpointFingerprint) []coreworkspace.Fingerprint { + normalized := normalizeFingerprints(values) + if len(normalized) == 0 { + return nil + } + out := make([]coreworkspace.Fingerprint, 0, len(normalized)) + for _, value := range normalized { + out = append(out, coreworkspace.Fingerprint{Name: value.Name, Value: value.Value}) + } + return out +} + +func normalizeFingerprints(values []pipeline.CheckpointFingerprint) []pipeline.CheckpointFingerprint { + 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([]pipeline.CheckpointFingerprint, 0, len(names)) + for _, name := range names { + out = append(out, pipeline.CheckpointFingerprint{Name: name, Value: byName[name]}) + } + return out +} + +func rejectionSummaries(rejected []contracts.RejectedOutput) []coreworkspace.RejectionSummary { + if len(rejected) == 0 { + return nil + } + type key struct { + validatorName string + reasonCode string + message string + } + counts := make(map[key]int, len(rejected)) + for _, item := range rejected { + k := key{validatorName: item.ValidatorName, reasonCode: item.ReasonCode, message: item.Message} + counts[k]++ + } + keys := make([]key, 0, len(counts)) + for k := range counts { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].validatorName != keys[j].validatorName { + return keys[i].validatorName < keys[j].validatorName + } + if keys[i].reasonCode != keys[j].reasonCode { + return keys[i].reasonCode < keys[j].reasonCode + } + return keys[i].message < keys[j].message + }) + out := make([]coreworkspace.RejectionSummary, 0, len(keys)) + for _, k := range keys { + out = append(out, coreworkspace.RejectionSummary{ + ValidatorName: k.validatorName, + ReasonCode: k.reasonCode, + Message: k.message, + Count: counts[k], + }) + } + return out +} + +func statusForRejected(rejected []contracts.RejectedOutput) coreworkspace.StageStatus { + if len(rejected) > 0 { + return coreworkspace.StatusSucceededWithRejections + } + return coreworkspace.StatusSucceeded +} + +func validationStatusString(warnings []contracts.Warning, rejected []contracts.RejectedOutput) string { + if len(rejected) > 0 { + return "rejected" + } + if len(warnings) > 0 { + return "approved_with_warnings" + } + return "approved" +} + +func errorMetadata(err error) map[string]string { + if err == nil { + return nil + } + return map[string]string{"error": err.Error()} +} + +func laneManifestPath(stage string, laneID string) string { + return lanePayloadPath(stage, laneID, "manifest.json") +} + +func lanePayloadPath(stage string, laneID string, file string) string { + return path.Join(stage, checkpointPathComponent(laneID), file) +} + +func checkpointPathComponent(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "_" + } + 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)) + } + } + out := b.String() + if out == "." || out == ".." || strings.Contains(out, "..") { + return "_" + } + return out +} + +func contentDigest(content []byte) string { + sum := sha256.Sum256(content) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func timePtr(t time.Time) *time.Time { + return &t +} diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go new file mode 100644 index 0000000..9b9f818 --- /dev/null +++ b/internal/framework/checkpoint/recorder_test.go @@ -0,0 +1,195 @@ +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"}}, + } + 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 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, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %q: %v", path, err) + } + if err := json.Unmarshal(data, out); err != nil { + t.Fatalf("decode %q: %v", path, err) + } +} + +type assertErr string + +func (e assertErr) Error() string { return string(e) } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go new file mode 100644 index 0000000..3859b15 --- /dev/null +++ b/internal/framework/pipeline/checkpoint.go @@ -0,0 +1,161 @@ +package pipeline + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type CheckpointFingerprint struct { + Name string `json:"name"` + Value string `json:"value"` +} + +type CheckpointRecorder interface { + SourceRunning(moduleKey string) error + SourceSucceeded(moduleKey string, doc *source.SourceDocument) error + SourceFailed(moduleKey string, err error) error + ChunkRunning(moduleKey string, sourceDigest string) error + ChunkSucceeded(moduleKey string, sourceDigest string, chunks []contracts.SourceChunk, warnings []contracts.Warning) error + ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error + ChunkFailed(moduleKey string, sourceDigest string, err error) error + ExtractRunning(laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + ExtractSucceeded(laneID string, moduleKey string, dependencies []CheckpointFingerprint, outputs []contracts.ExtractOutput, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error + ExtractFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error + MergeRunning(laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + MergeSucceeded(laneID string, moduleKey string, dependencies []CheckpointFingerprint, output contracts.MergeOutput, warnings []contracts.Warning) error + MergeRejected(laneID string, moduleKey string, dependencies []CheckpointFingerprint, rejected contracts.RejectedOutput) error + MergeFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error + NormalizeRunning(laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + NormalizeSucceeded(laneID string, moduleKey string, dependencies []CheckpointFingerprint, output contracts.NormalizeOutput, warnings []contracts.Warning) error + NormalizeRejected(laneID string, moduleKey string, dependencies []CheckpointFingerprint, rejected contracts.RejectedOutput) error + NormalizeFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error +} + +type noopCheckpointRecorder struct{} + +func NoopCheckpointRecorder() CheckpointRecorder { return noopCheckpointRecorder{} } + +func (noopCheckpointRecorder) SourceRunning(string) error { return nil } +func (noopCheckpointRecorder) SourceSucceeded(string, *source.SourceDocument) error { return nil } +func (noopCheckpointRecorder) SourceFailed(string, error) error { return nil } +func (noopCheckpointRecorder) ChunkRunning(string, string) error { return nil } +func (noopCheckpointRecorder) ChunkSucceeded(string, string, []contracts.SourceChunk, []contracts.Warning) error { + return nil +} +func (noopCheckpointRecorder) ChunkRejected(string, string, contracts.RejectedOutput) error { + return nil +} +func (noopCheckpointRecorder) ChunkFailed(string, string, error) error { return nil } +func (noopCheckpointRecorder) ExtractRunning(string, string, []CheckpointFingerprint) error { + return nil +} +func (noopCheckpointRecorder) ExtractSucceeded(string, string, []CheckpointFingerprint, []contracts.ExtractOutput, []contracts.RejectedOutput, []contracts.Warning) error { + return nil +} +func (noopCheckpointRecorder) ExtractFailed(string, string, []CheckpointFingerprint, error) error { + return nil +} +func (noopCheckpointRecorder) MergeRunning(string, string, []CheckpointFingerprint) error { return nil } +func (noopCheckpointRecorder) MergeSucceeded(string, string, []CheckpointFingerprint, contracts.MergeOutput, []contracts.Warning) error { + return nil +} +func (noopCheckpointRecorder) MergeRejected(string, string, []CheckpointFingerprint, contracts.RejectedOutput) error { + return nil +} +func (noopCheckpointRecorder) MergeFailed(string, string, []CheckpointFingerprint, error) error { + return nil +} +func (noopCheckpointRecorder) NormalizeRunning(string, string, []CheckpointFingerprint) error { + return nil +} +func (noopCheckpointRecorder) NormalizeSucceeded(string, string, []CheckpointFingerprint, contracts.NormalizeOutput, []contracts.Warning) error { + return nil +} +func (noopCheckpointRecorder) NormalizeRejected(string, string, []CheckpointFingerprint, contracts.RejectedOutput) error { + return nil +} +func (noopCheckpointRecorder) NormalizeFailed(string, string, []CheckpointFingerprint, error) error { + return nil +} + +func rawOutputDigests(payloads []contracts.RawPayload) []CheckpointFingerprint { + values := make([]CheckpointFingerprint, 0, len(payloads)) + for i, payload := range payloads { + values = append(values, CheckpointFingerprint{ + Name: fmt.Sprintf("payload[%d]", i), + Value: checkpointContentDigest(payload.Content), + }) + } + return normalizeCheckpointFingerprints(values) +} + +func extractPayloads(outputs []contracts.ExtractOutput) []contracts.RawPayload { + if len(outputs) == 0 { + return nil + } + payloads := make([]contracts.RawPayload, 0, len(outputs)) + for _, output := range outputs { + payloads = append(payloads, output.Payload) + } + return payloads +} + +func digestFingerprints(name string, digest string) []CheckpointFingerprint { + digest = strings.TrimSpace(digest) + if digest == "" { + return nil + } + return []CheckpointFingerprint{{Name: name, Value: digest}} +} + +func joinedChunkDigest(chunks []contracts.SourceChunk) string { + if len(chunks) == 0 { + return "" + } + values := make([]string, 0, len(chunks)) + for _, chunk := range chunks { + values = append(values, chunk.ID+"="+checkpointContentDigest(chunk.Content)) + } + sort.Strings(values) + sum := sha256.Sum256([]byte(strings.Join(values, "\n"))) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func normalizeCheckpointFingerprints(values []CheckpointFingerprint) []CheckpointFingerprint { + 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([]CheckpointFingerprint, 0, len(names)) + for _, name := range names { + out = append(out, CheckpointFingerprint{Name: name, Value: byName[name]}) + } + return out +} + +func checkpointContentDigest(content []byte) string { + sum := sha256.Sum256(content) + return "sha256:" + hex.EncodeToString(sum[:]) +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 2bc83f8..6c00b17 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -48,6 +48,7 @@ type RunInput struct { LLMProfiles []artifacts.LLMProfileManifest Metadata map[string]any Warnings []contracts.Warning + Checkpoints CheckpointRecorder } type RunOutput struct { @@ -70,6 +71,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } output.Manifest = manifestFromPipeline(input) + checkpoints := input.Checkpoints + if checkpoints == nil { + checkpoints = NoopCheckpointRecorder() + } defer func() { output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.LLMClient)) }() @@ -80,6 +85,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err) } attachModuleManifestMetadata(&output, "input", adapter) + if err := checkpoints.SourceRunning(adapter.Key()); err != nil { + return failOutput(output), fmt.Errorf("write source checkpoint: %w", err) + } doc, err := adapter.Parse(ctx, contracts.ParseRequest{ SourceID: input.SourceID, Path: input.Path, @@ -89,11 +97,16 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Metadata: input.Metadata, }) if err != nil { + _ = checkpoints.SourceFailed(adapter.Key(), err) return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err) } if err := source.ValidateDocument(doc); err != nil { + _ = checkpoints.SourceFailed(adapter.Key(), err) return failOutput(output), fmt.Errorf("validate source document: %w", err) } + if err := checkpoints.SourceSucceeded(adapter.Key(), doc); err != nil { + return failOutput(output), fmt.Errorf("write source checkpoint: %w", err) + } sourceInput := sourceInputMaterial(input.Path, input.RawInput) sessionID := resolvedSessionID(input.SessionID, doc.ID) output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID) @@ -104,6 +117,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err) } attachModuleManifestMetadata(&output, "chunker", chunker) + if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil { + return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) + } var canonicalChunks []contracts.SourceChunk var chunkWarnings []contracts.Warning chunksAccepted, chunkRejection, err := runWithRetry(ctx, input.Pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { @@ -136,17 +152,24 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return true, nil, nil }) if err != nil { + _ = checkpoints.ChunkFailed(chunker.Key(), doc.Digest, err) return failOutput(output), err } if !chunksAccepted { output.Rejected = append(output.Rejected, *chunkRejection) + if err := checkpoints.ChunkRejected(chunker.Key(), doc.Digest, *chunkRejection); err != nil { + return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) + } } else { output.Warnings = append(output.Warnings, chunkWarnings...) + if err := checkpoints.ChunkSucceeded(chunker.Key(), doc.Digest, canonicalChunks, chunkWarnings); err != nil { + return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) + } } if chunksAccepted { for _, lane := range input.Pipeline.ArtifactLanes { - if err := r.runLane(ctx, input, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil { + if err := r.runLane(ctx, input, checkpoints, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil { return failOutput(output), err } } @@ -187,7 +210,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return output, nil } -func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput) error { +func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput) error { extractor, err := r.registries.Extractors.Build(lane.Extract.Module) if err != nil { return fmt.Errorf("build extractor %q for lane %q: %w", lane.Extract.Module, lane.ID, err) @@ -203,6 +226,12 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source setLaneManifestMetadata(output, lane.ID, extractor, merger, normalizer) extractOutputs := make([]contracts.ExtractOutput, 0, len(chunks)) + extractWarnings := []contracts.Warning{} + extractRejectedStart := len(output.Rejected) + extractDependencies := digestFingerprints("chunks", joinedChunkDigest(chunks)) + if err := checkpoints.ExtractRunning(lane.ID, extractor.Key(), extractDependencies); err != nil { + return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) + } for index := range chunks { chunk := chunks[index] var acceptedOutput contracts.ExtractOutput @@ -256,6 +285,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return true, nil, nil }) if err != nil { + _ = checkpoints.ExtractFailed(lane.ID, extractor.Key(), extractDependencies, err) return err } if !accepted { @@ -263,8 +293,13 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source continue } output.Warnings = append(output.Warnings, acceptedWarnings...) + extractWarnings = append(extractWarnings, acceptedWarnings...) extractOutputs = append(extractOutputs, acceptedOutput) } + extractRejected := cloneRejectedOutputs(output.Rejected[extractRejectedStart:]) + if err := checkpoints.ExtractSucceeded(lane.ID, extractor.Key(), extractDependencies, extractOutputs, extractRejected, extractWarnings); err != nil { + return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) + } if len(extractOutputs) == 0 { return nil @@ -272,6 +307,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source var acceptedMerge contracts.MergeOutput var mergeWarnings []contracts.Warning + mergeDependencies := rawOutputDigests(extractPayloads(extractOutputs)) + if err := checkpoints.MergeRunning(lane.ID, merger.Key(), mergeDependencies); err != nil { + return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) + } mergeAccepted, mergeRejection, err := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { mergeResult, err := merger.Merge(ctx, contracts.MergeRequest{ Source: doc, @@ -318,16 +357,27 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return true, nil, nil }) if err != nil { + _ = checkpoints.MergeFailed(lane.ID, merger.Key(), mergeDependencies, err) return err } if !mergeAccepted { output.Rejected = append(output.Rejected, *mergeRejection) + if err := checkpoints.MergeRejected(lane.ID, merger.Key(), mergeDependencies, *mergeRejection); err != nil { + return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) + } return nil } output.Warnings = append(output.Warnings, mergeWarnings...) + if err := checkpoints.MergeSucceeded(lane.ID, merger.Key(), mergeDependencies, acceptedMerge, mergeWarnings); err != nil { + return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) + } var acceptedNormalize contracts.NormalizeOutput var normalizeWarnings []contracts.Warning + normalizeDependencies := rawOutputDigests([]contracts.RawPayload{acceptedMerge.Payload}) + if err := checkpoints.NormalizeRunning(lane.ID, normalizer.Key(), normalizeDependencies); err != nil { + return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) + } normalizeAccepted, normalizeRejection, err := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { normalizeResult, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{ Source: doc, @@ -374,13 +424,20 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source return true, nil, nil }) if err != nil { + _ = checkpoints.NormalizeFailed(lane.ID, normalizer.Key(), normalizeDependencies, err) return err } if !normalizeAccepted { output.Rejected = append(output.Rejected, *normalizeRejection) + if err := checkpoints.NormalizeRejected(lane.ID, normalizer.Key(), normalizeDependencies, *normalizeRejection); err != nil { + return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) + } return nil } output.Warnings = append(output.Warnings, normalizeWarnings...) + if err := checkpoints.NormalizeSucceeded(lane.ID, normalizer.Key(), normalizeDependencies, acceptedNormalize, normalizeWarnings); err != nil { + return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) + } output.NormalizeOutputs = append(output.NormalizeOutputs, acceptedNormalize) return nil } diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index a9ce9e3..f63c81a 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -992,6 +992,37 @@ func TestRunPassesChunkContentAndMediaTypeToExtractors(t *testing.T) { } } +func TestRunDoesNotPassCheckpointPathsToModules(t *testing.T) { + modules := defaultRunnerModules() + + _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + Checkpoints: NoopCheckpointRecorder(), + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + for _, req := range modules.input.requests { + assertNoCheckpointMetadata(t, req.Metadata) + } + for _, req := range modules.chunker.requests { + assertNoCheckpointMetadata(t, req.Metadata) + } + for _, req := range modules.extractors["extract-alpha"].requests { + assertNoCheckpointMetadata(t, req.Metadata) + } + for _, req := range modules.mergers["merge"].requests { + assertNoCheckpointMetadata(t, req.Metadata) + } + for _, req := range modules.normalizers["normalize"].requests { + assertNoCheckpointMetadata(t, req.Metadata) + } + for _, req := range modules.output.requests { + assertNoCheckpointMetadata(t, req.Metadata) + } +} + func TestRunOmitsRejectedExtractOutputsFromMerge(t *testing.T) { modules := defaultRunnerModules() validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false, true}, reason: "bad_extract", message: "extract rejected"} @@ -2181,6 +2212,25 @@ func warningReasons(warnings []contracts.Warning) []string { return reasons } +func assertNoCheckpointMetadata(t *testing.T, metadata map[string]any) { + t.Helper() + + for key, value := range metadata { + lowerKey := strings.ToLower(key) + if strings.Contains(lowerKey, "checkpoint") || strings.Contains(lowerKey, "workspace") { + t.Fatalf("metadata key %q exposes checkpoint/workspace state", key) + } + text, ok := value.(string) + if !ok { + continue + } + lowerValue := strings.ToLower(text) + if strings.Contains(lowerValue, "checkpoint") || strings.Contains(lowerValue, "workspace") { + t.Fatalf("metadata value for %q exposes checkpoint/workspace state: %q", key, text) + } + } +} + func assertRunError(t *testing.T, err error, want string) { t.Helper()