From ae9c2e1d5eb2d7af6987182166ff1ec65cdf7796 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 8 Jul 2026 03:02:50 +0000 Subject: [PATCH] Reuse valid workspace checkpoints on request --- docs/cli.md | 14 +- docs/config.md | 4 +- docs/internal/pipeline.md | 6 + docs/operations.md | 17 +- internal/cli/run.go | 32 +- internal/cli/run_test.go | 266 ++++++++- internal/core/diagnostics/artifacts.go | 1 + internal/core/diagnostics/run_dir.go | 5 + internal/framework/checkpoint/loader.go | 345 +++++++++++ internal/framework/checkpoint/recorder.go | 53 +- .../framework/checkpoint/recorder_test.go | 157 ++++- internal/framework/pipeline/checkpoint.go | 66 +++ internal/framework/pipeline/runner.go | 561 ++++++++++-------- internal/framework/pipeline/runner_test.go | 169 ++++++ 14 files changed, 1406 insertions(+), 290 deletions(-) create mode 100644 internal/framework/checkpoint/loader.go diff --git a/docs/cli.md b/docs/cli.md index 906ea00..942cbdc 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -21,7 +21,7 @@ ID in config or with `--llm-profile`. ```text notarius help -notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector] +notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--resume] [--session-id id] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ``` @@ -41,6 +41,8 @@ Flags: `NOTARIUS_CONFIG`, then `/usr/local/etc/notarius/config.yml`. - `--only lane-a,lane-b`: run only the named artifact lanes. Values are comma-separated and must be non-empty. +- `--resume`: reuse valid workspace checkpoints for this invocation. Requires + `workspace.resume.enabled: true`. - `--output-dir path`: output root. The run writes to `//`. Defaults to `./notarius-output`. - `--diagnostics-dir path`: diagnostics work directory override for this @@ -140,6 +142,16 @@ go run ./cmd/notarius run dnd-session \ --session-id campaign-17-session-04 ``` +Use `--resume` to reuse valid checkpoints from a previous compatible +invocation: + +```sh +go run ./cmd/notarius run dnd-session \ + --config examples/dnd-spells.config.yml \ + --input examples/seriatim-minimal-transcript.json \ + --resume +``` + For durable output, diagnostics, retention, and failure inspection, see [Operations](operations.md). diff --git a/docs/config.md b/docs/config.md index 5c4d99f..d5bf37c 100644 --- a/docs/config.md +++ b/docs/config.md @@ -353,8 +353,8 @@ casts still must be present in the source transcript. - `debug.enabled`: boolean debug artifact setting. Default: `false`. When `workspace.resume.enabled` is true, runs write stage-owned checkpoint -artifacts under `/checkpoints/`. Checkpoint reads and -resume execution are not implemented. +artifacts under `/checkpoints/`. `notarius run --resume` +can reuse valid checkpoints from a compatible invocation. Debug artifact writers are not part of the current workflow. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 4c3f339..4e1db7f 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -77,6 +77,12 @@ 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. +For `run --resume`, the CLI also passes a checkpoint loader. The runner consults +the loader in workflow order and reuses only checkpoints whose manifest schema, +status, identity digest, dependency fingerprints, payload files, and payload +digests validate for the current invocation. Missing or invalid checkpoints fall +back to normal execution and are refreshed by the recorder. + ## Registries And Module Specs `pipeline.Registries` holds concrete constructors for execution. A diff --git a/docs/operations.md b/docs/operations.md index b111fb3..8a3c2af 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -74,6 +74,8 @@ Implemented diagnostics artifacts: - `resolved-references.json`: resolved reference provenance, including target stage, lane ID when present, origin, digest, media type, byte size, and binding source, without reference content. +- `checkpoint-events.json`: checkpoint steps that were reused or executed + during an explicit resume invocation. - `run-manifest.json`: the same run manifest written to durable output when it is available, including top-level module metadata when present. - `warnings.json`: warning list. @@ -94,14 +96,21 @@ write checkpoints under: ``` 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. +root-level checkpoint summary. Ordinary `notarius run` invocations execute the +pipeline normally and refresh checkpoints. `notarius run --resume` reuses valid +checkpoints and executes any missing, invalid, or incompatible step normally. 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. +A checkpoint is reused only when its workspace schema version, checkpoint +identity digest, step status, dependency fingerprints, payload files, and +payload digests match the current invocation. Changes to input bytes, resolved +pipeline digest, selected lanes, runtime LLM profile override, or materialized +reference digests invalidate reuse. + ## Retention Diagnostics retention is configured with `workspace.diagnostics.retention`, @@ -156,8 +165,8 @@ directories unless they are part of your own operational policy. ## Operational Limits -Checkpoint writing does not provide resume execution yet. Re-run `notarius run` -after fixing the cause of a failed run. +If `--resume` cannot reuse a checkpoint, Notarius executes that step and writes +a fresh checkpoint when checkpointing is enabled. 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 33674e3..4748966 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -29,7 +29,7 @@ const defaultOutputRoot = "./notarius-output" const usage = `Usage: notarius help - notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector] + notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--resume] [--session-id id] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -99,6 +99,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i outputDir := fs.String("output-dir", "", "output directory") diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory") llmProfile := fs.String("llm-profile", "", "LLM profile override") + resume := fs.Bool("resume", false, "reuse valid workspace checkpoints") sessionID := sessionIDFlag{} referenceFlags := stringListFlag{} withoutReferenceFlags := stringListFlag{} @@ -179,12 +180,16 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i ConfigPath: loadedConfigPath, ConfigSource: configSource(*configPath), OnlyLanes: append([]string(nil), only...), + Resume: *resume, RunID: runID, StartedAt: startedAt, } if err := writeDiagnostics(runDir, func() error { return runDir.WriteInvocationMetadata(invocation) }); err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics invocation metadata: %w", err)) } + if *resume && !workspaceSettings.ResumeEnabled { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("--resume requires workspace.resume.enabled: true")) + } catalog, err := effectiveCatalog(opts) if err != nil { @@ -256,7 +261,7 @@ 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)) + checkpointRecorder, checkpointLoader, err := checkpointHandlersForRun(workspaceSettings, effective.ResolvedPipeline, rawInput, only, llmProfiles, strings.TrimSpace(*llmProfile), strings.TrimSpace(sessionID.value), *resume) if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) } @@ -273,10 +278,12 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i Metadata: runMetadata(*outputDir, *diagnosticsDir), Warnings: referenceWarnings, Checkpoints: checkpointRecorder, + Checkpoint: checkpointLoader, }) if err != nil { if output.Manifest.PipelineID != "" && runDir != nil { _ = runDir.WriteRunManifest(output.Manifest) + _ = runDir.WriteCheckpointEvents(output.CheckpointEvents) } return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("run pipeline %q: %w", pipelineID, err)) } @@ -288,6 +295,9 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err := writeDiagnostics(runDir, func() error { return runDir.WriteWarnings(output.Warnings) }); err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics warnings: %w", err)) } + if err := writeDiagnostics(runDir, func() error { return runDir.WriteCheckpointEvents(output.CheckpointEvents) }); err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics checkpoint events: %w", err)) + } if err := writeDiagnostics(runDir, func() error { return runDir.WriteRunReport(runReport{ RunID: runDir.RunID(), @@ -356,7 +366,7 @@ func writeDiagnostics(runDir *diagnostics.RunDirectory, write func() error) erro return write() } -func checkpointRecorderForRun( +func checkpointHandlersForRun( settings workspace.Settings, resolved pipeline.ResolvedPipeline, rawInput []byte, @@ -364,7 +374,8 @@ func checkpointRecorderForRun( llmProfiles []artifacts.LLMProfileManifest, llmProfileOverride string, sessionID string, -) (pipeline.CheckpointRecorder, error) { + resume bool, +) (pipeline.CheckpointRecorder, pipeline.CheckpointLoader, error) { identity, err := workspace.NewCheckpointIdentity(workspace.CheckpointIdentityInput{ Pipeline: resolved, InputKey: resolved.Input.Module, @@ -375,13 +386,20 @@ func checkpointRecorderForRun( ProvenanceFingerprints: llmProfileFingerprints(llmProfiles), }) if err != nil { - return nil, fmt.Errorf("create checkpoint identity: %w", err) + return nil, 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 nil, nil, fmt.Errorf("create checkpoint recorder: %w", err) } - return recorder, nil + loader := pipeline.NoopCheckpointLoader() + if resume { + loader, err = checkpoint.NewWorkspaceLoader(settings, identity) + if err != nil { + return nil, nil, fmt.Errorf("create checkpoint loader: %w", err) + } + } + return recorder, loader, nil } func rawInputDigest(data []byte) string { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index f2fcf57..11effa6 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2185,16 +2185,20 @@ func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { outputDir := t.TempDir() configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(false) 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), + LLMClientFactory: fakeLLMFactory(client, nil), }) if code != 0 { t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) } + if client.calls != 1 { + t.Fatalf("LLM calls = %d, want ordinary run to execute despite checkpoint writing", client.calls) + } checkpointDir := onlyCheckpointIdentityDir(t, workspaceDir) for _, name := range []string{ "source/manifest.json", @@ -2215,6 +2219,133 @@ func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) } +func TestRunPipelineResumeRequiresWorkspaceResumeEnabled(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceDiagnostics("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, "--resume"}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "--resume requires workspace.resume.enabled: true") { + t.Fatalf("stderr = %q, want resume configuration error", stderr.String()) + } +} + +func TestRunPipelineResumeReusesWorkspaceCheckpoints(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + firstClient := newFakeRunLLMClient(false) + 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(firstClient, nil), + }) + if code != 0 { + t.Fatalf("first RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if firstClient.calls != 1 { + t.Fatalf("first LLM calls = %d, want checkpoint seed run to execute", firstClient.calls) + } + + secondClient := newFakeRunLLMClient(false) + stdout.Reset() + stderr.Reset() + code = RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir, "--resume"}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(secondClient, nil), + }) + if code != 0 { + t.Fatalf("resume RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if secondClient.calls != 0 { + t.Fatalf("resume LLM calls = %d, want checkpoint reuse", secondClient.calls) + } + runDirs := childDirs(t, filepath.Join(workspaceDir, "diagnostics")) + if len(runDirs) != 2 { + t.Fatalf("diagnostics run dirs = %v, want fresh diagnostics for each invocation", runDirs) + } + if !anyDiagnosticsFileContains(t, runDirs, diagnostics.ArtifactCheckpointEvents, `"action": "reused"`) { + t.Fatalf("checkpoint event diagnostics under %v did not record reuse", runDirs) + } +} + +func TestRunPipelineResumeInvalidatesWhenInvocationIdentityChanges(t *testing.T) { + t.Run("input", func(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) + seedInput := writeSeriatimInput(t) + changedInput := writeFile(t, "source-changed.json", `{ + "metadata": {"id": "session-alpha"}, + "segments": [{"id": 1, "start": 0, "end": 1, "speaker": "Aria", "text": "Aria casts Shield."}] +}`) + seedWorkspaceCheckpoint(t, configPath, seedInput, nil) + client := runResumeWithClient(t, configPath, changedInput, nil) + if client.calls == 0 { + t.Fatal("resume LLM calls = 0, want execution after input identity change") + } + }) + + t.Run("pipeline digest", func(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + seedConfig := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) + changedConfig := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeAndChunkOptions("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + seedWorkspaceCheckpoint(t, seedConfig, inputPath, nil) + client := runResumeWithClient(t, changedConfig, inputPath, nil) + if client.calls == 0 { + t.Fatal("resume LLM calls = 0, want execution after pipeline digest change") + } + }) + + t.Run("selected lanes", func(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeLanes("dnd-session", workspaceDir, "always", "spells", "items")) + inputPath := writeSeriatimInput(t) + seedWorkspaceCheckpoint(t, configPath, inputPath, nil) + client := runResumeWithClient(t, configPath, inputPath, []string{"--only", "spells"}) + if client.calls == 0 { + t.Fatal("resume LLM calls = 0, want execution after selected lane change") + } + }) + + t.Run("references", func(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + referencePath := writeFile(t, "players.md", "Aria is a cleric.\n") + seedWorkspaceCheckpoint(t, configPath, inputPath, []string{"--reference", "players=" + referencePath}) + if err := os.WriteFile(referencePath, []byte("Aria is a wizard.\n"), 0o644); err != nil { + t.Fatalf("update reference: %v", err) + } + client := runResumeWithClient(t, configPath, inputPath, []string{"--reference", "players=" + referencePath}) + if client.calls == 0 { + t.Fatal("resume LLM calls = 0, want execution after reference digest change") + } + }) + + t.Run("llm profile override", func(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + profilePath := writeScriptoriumProfileFile(t, "runtime", "http://profile.test/v1", "test-model") + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceResumeAndProfileFile("dnd-session", workspaceDir, "always", profilePath)) + inputPath := writeSeriatimInput(t) + seedWorkspaceCheckpoint(t, configPath, inputPath, nil) + client := runResumeWithClient(t, configPath, inputPath, []string{"--llm-profile", "runtime"}) + if client.calls == 0 { + t.Fatal("resume LLM calls = 0, want execution after LLM profile override change") + } + }) +} + func TestRunPipelineSkipsDiagnosticsWhenWorkspaceDiagnosticsDisabled(t *testing.T) { workspaceDir := filepath.Join(t.TempDir(), "workspace") outputDir := t.TempDir() @@ -2923,6 +3054,88 @@ pipelines: ` } +func mvpConfigYAMLWithWorkspaceResumeAndChunker(pipelineID, workspaceDir, retention, chunker string) string { + return `version: 2 +workspace: + directory: ` + workspaceDir + ` + diagnostics: + enabled: true + retention: ` + retention + ` + resume: + enabled: true +pipelines: + ` + pipelineID + `: + input: seriatim + chunk: ` + chunker + ` + artifacts: + spells: + extract: dnd/spells +` +} + +func mvpConfigYAMLWithWorkspaceResumeAndChunkOptions(pipelineID, workspaceDir, retention string) string { + return `version: 2 +workspace: + directory: ` + workspaceDir + ` + diagnostics: + enabled: true + retention: ` + retention + ` + resume: + enabled: true +pipelines: + ` + pipelineID + `: + input: seriatim + chunk: + module: generic + options: + max_units: 10 + artifacts: + spells: + extract: dnd/spells +` +} + +func mvpConfigYAMLWithWorkspaceResumeAndProfileFile(pipelineID, workspaceDir, retention, profileFile string) string { + return `version: 2 +scriptorium: + profile_file: ` + profileFile + ` +workspace: + directory: ` + workspaceDir + ` + diagnostics: + enabled: true + retention: ` + retention + ` + resume: + enabled: true +pipelines: + ` + pipelineID + `: + input: seriatim + artifacts: + spells: + extract: dnd/spells +` +} + +func mvpConfigYAMLWithWorkspaceResumeLanes(pipelineID, workspaceDir, retention string, laneIDs ...string) string { + var b strings.Builder + b.WriteString("version: 2\n") + b.WriteString("workspace:\n") + b.WriteString(" directory: " + workspaceDir + "\n") + b.WriteString(" diagnostics:\n") + b.WriteString(" enabled: true\n") + b.WriteString(" retention: " + retention + "\n") + b.WriteString(" resume:\n") + b.WriteString(" enabled: true\n") + b.WriteString("pipelines:\n") + b.WriteString(" " + pipelineID + ":\n") + b.WriteString(" input: seriatim\n") + b.WriteString(" artifacts:\n") + for _, laneID := range laneIDs { + b.WriteString(" " + laneID + ":\n") + b.WriteString(" extract: dnd/spells\n") + } + return b.String() +} + func mvpConfigYAMLWithWorkspaceDiagnosticsDisabled(pipelineID, workspaceDir string) string { return `version: 2 workspace: @@ -3293,6 +3506,57 @@ func onlyCheckpointIdentityDir(t *testing.T, workspaceDir string) string { return onlyChildDir(t, inputDir) } +func anyDiagnosticsFileContains(t *testing.T, runDirs []string, name string, want string) bool { + t.Helper() + for _, runDir := range runDirs { + data, err := os.ReadFile(filepath.Join(runDir, name)) + if err != nil { + if os.IsNotExist(err) { + continue + } + t.Fatalf("read diagnostics artifact %q under %q: %v", name, runDir, err) + } + if strings.Contains(string(data), want) { + return true + } + } + return false +} + +func seedWorkspaceCheckpoint(t *testing.T, configPath string, inputPath string, extraArgs []string) { + t.Helper() + client := newFakeRunLLMClient(false) + args := []string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", t.TempDir()} + args = append(args, extraArgs...) + var stdout bytes.Buffer + var stderr bytes.Buffer + code := RunWithOptions(args, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + if code != 0 { + t.Fatalf("seed RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if client.calls == 0 { + t.Fatal("seed LLM calls = 0, want checkpoint seed run to execute") + } +} + +func runResumeWithClient(t *testing.T, configPath string, inputPath string, extraArgs []string) *fakeRunLLMClient { + t.Helper() + client := newFakeRunLLMClient(false) + args := []string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", t.TempDir(), "--resume"} + args = append(args, extraArgs...) + var stdout bytes.Buffer + var stderr bytes.Buffer + code := RunWithOptions(args, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + if code != 0 { + t.Fatalf("resume RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + return client +} + func childDirs(t *testing.T, root string) []string { t.Helper() entries, err := os.ReadDir(root) diff --git a/internal/core/diagnostics/artifacts.go b/internal/core/diagnostics/artifacts.go index 2154c41..8dfadf4 100644 --- a/internal/core/diagnostics/artifacts.go +++ b/internal/core/diagnostics/artifacts.go @@ -5,6 +5,7 @@ const ( ArtifactEffectiveConfig = "effective-config.json" ArtifactResolvedPipeline = "resolved-pipeline.json" ArtifactResolvedReferences = "resolved-references.json" + ArtifactCheckpointEvents = "checkpoint-events.json" ArtifactSourceDocument = "source-document.json" ArtifactRunManifest = "run-manifest.json" ArtifactRunReport = "run-report.json" diff --git a/internal/core/diagnostics/run_dir.go b/internal/core/diagnostics/run_dir.go index 1fb58d4..e7fcc69 100644 --- a/internal/core/diagnostics/run_dir.go +++ b/internal/core/diagnostics/run_dir.go @@ -51,6 +51,7 @@ type InvocationMetadata struct { Operation string `json:"operation"` PipelineID string `json:"pipeline_id,omitempty"` PipelineDigest string `json:"pipeline_digest,omitempty"` + Resume bool `json:"resume,omitempty"` InputPath string `json:"input_path,omitempty"` ConfigPath string `json:"config_path,omitempty"` ConfigSource string `json:"config_source,omitempty"` @@ -153,6 +154,10 @@ func (r *RunDirectory) WriteResolvedReferences(payload any) error { return r.WriteJSONArtifact(ArtifactResolvedReferences, payload) } +func (r *RunDirectory) WriteCheckpointEvents(payload any) error { + return r.WriteJSONArtifact(ArtifactCheckpointEvents, payload) +} + func (r *RunDirectory) WriteSourceDocument(payload any) error { return r.WriteJSONArtifact(ArtifactSourceDocument, payload) } diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go new file mode 100644 index 0000000..026bb49 --- /dev/null +++ b/internal/framework/checkpoint/loader.go @@ -0,0 +1,345 @@ +package checkpoint + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "os" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +type WorkspaceLoader struct { + root string + identityDigest string +} + +func NewWorkspaceLoader(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointLoader, error) { + root, err := settings.CheckpointDirectory(identity) + if err != nil { + return nil, err + } + if strings.TrimSpace(root) == "" { + return pipeline.NoopCheckpointLoader(), nil + } + return &WorkspaceLoader{root: root, identityDigest: identity.Digest}, nil +} + +func (l *WorkspaceLoader) Enabled() bool { + return l != nil && strings.TrimSpace(l.root) != "" +} + +func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.SourceManifest + if decision := l.readJSON("source/manifest.json", &manifest); !decision.Reused { + return pipeline.SourceCheckpoint{}, decision + } + if decision := l.validateManifest(manifest.StageManifest, coreworkspace.StageSource, "", moduleKey, coreworkspace.StatusSucceeded, nil); !decision.Reused { + return pipeline.SourceCheckpoint{}, decision + } + var payload sourceDocumentEnvelope + if decision := l.readJSON("source/source-document.json", &payload); !decision.Reused { + return pipeline.SourceCheckpoint{}, decision + } + doc := cloneSourceDocument(payload.Document) + if err := source.ValidateDocument(&doc); err != nil { + return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint document is invalid: %v", err) + } + if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID { + return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint source id does not match payload") + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) { + return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint output digest does not match payload") + } + return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision() +} + +func (l *WorkspaceLoader) Chunk(moduleKey string, sourceDigest string) (pipeline.ChunkCheckpoint, pipeline.CheckpointDecision) { + expectedDependencies := digestFingerprints("source_document", sourceDigest) + var manifest coreworkspace.ChunkManifest + if decision := l.readJSON("chunk/manifest.json", &manifest); !decision.Reused { + return pipeline.ChunkCheckpoint{}, decision + } + if decision := l.validateManifest(manifest.StageManifest, coreworkspace.StageChunk, "", moduleKey, coreworkspace.StatusSucceeded, expectedDependencies); !decision.Reused { + return pipeline.ChunkCheckpoint{}, decision + } + var payload chunksEnvelope + if decision := l.readJSON("chunk/chunks.json", &payload); !decision.Reused { + return pipeline.ChunkCheckpoint{}, decision + } + chunks, err := sourceChunksFromEnvelope(payload.Chunks) + if err != nil { + return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint payload is invalid: %v", err) + } + if len(chunks) == 0 { + return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint payload has no chunks") + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), chunkOutputDigests(chunks)) { + return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint output digests do not match payload") + } + return pipeline.ChunkCheckpoint{Chunks: chunks, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + +func (l *WorkspaceLoader) Extract(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.ExtractLaneManifest + if decision := l.readJSON(laneManifestPath("extract", laneID), &manifest); !decision.Reused { + return pipeline.ExtractCheckpoint{}, decision + } + if decision := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageExtract, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded, coreworkspace.StatusSucceededWithRejections); !decision.Reused { + return pipeline.ExtractCheckpoint{}, decision + } + var payload extractOutputsEnvelope + if decision := l.readJSON(lanePayloadPath("extract", laneID, "outputs.json"), &payload); !decision.Reused { + return pipeline.ExtractCheckpoint{}, decision + } + outputs, err := extractOutputsFromEnvelope(payload.Outputs) + if err != nil { + return pipeline.ExtractCheckpoint{}, invalidDecision("extract checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), rawOutputDigests(extractPayloads(outputs))) { + return pipeline.ExtractCheckpoint{}, invalidDecision("extract checkpoint output digests do not match payload") + } + return pipeline.ExtractCheckpoint{ + Outputs: outputs, + Rejected: cloneRejectedOutputs(payload.Rejected), + Warnings: cloneWarnings(payload.Warnings), + }, reusedDecision() +} + +func (l *WorkspaceLoader) Merge(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.MergeLaneManifest + if decision := l.readJSON(laneManifestPath("merge", laneID), &manifest); !decision.Reused { + return pipeline.MergeCheckpoint{}, decision + } + if decision := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageMerge, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !decision.Reused { + return pipeline.MergeCheckpoint{}, decision + } + var payload mergeOutputEnvelope + if decision := l.readJSON(lanePayloadPath("merge", laneID, "output.json"), &payload); !decision.Reused { + return pipeline.MergeCheckpoint{}, decision + } + output, err := mergeOutputFromEnvelope(payload.Output) + if err != nil { + return pipeline.MergeCheckpoint{}, invalidDecision("merge checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), rawOutputDigests([]contracts.RawPayload{output.Payload})) { + return pipeline.MergeCheckpoint{}, invalidDecision("merge checkpoint output digest does not match payload") + } + return pipeline.MergeCheckpoint{Output: output, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + +func (l *WorkspaceLoader) Normalize(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.NormalizeLaneManifest + if decision := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !decision.Reused { + return pipeline.NormalizeCheckpoint{}, decision + } + if decision := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageNormalize, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !decision.Reused { + return pipeline.NormalizeCheckpoint{}, decision + } + var payload normalizeOutputEnvelope + if decision := l.readJSON(lanePayloadPath("normalize", laneID, "output.json"), &payload); !decision.Reused { + return pipeline.NormalizeCheckpoint{}, decision + } + output, err := normalizeOutputFromEnvelope(payload.Output) + if err != nil { + return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), rawOutputDigests([]contracts.RawPayload{output.Payload})) { + return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize checkpoint output digest does not match payload") + } + return pipeline.NormalizeCheckpoint{Output: output, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + +func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDecision { + if !l.Enabled() { + return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"} + } + target, err := coreworkspace.SafePath(l.root, name) + if err != nil { + return invalidDecision("checkpoint path is invalid: %v", err) + } + data, err := os.ReadFile(target) + if err != nil { + if os.IsNotExist(err) { + return pipeline.CheckpointDecision{Reason: "checkpoint artifact is missing"} + } + return invalidDecision("read checkpoint artifact: %v", err) + } + if err := json.Unmarshal(data, out); err != nil { + return invalidDecision("decode checkpoint artifact: %v", err) + } + return reusedDecision() +} + +func (l *WorkspaceLoader) validateManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, status coreworkspace.StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision { + return l.validateLaneManifest(manifest, stage, laneID, moduleKey, dependencies, status) +} + +func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...coreworkspace.StageStatus) pipeline.CheckpointDecision { + if manifest.WorkspaceSchemaVersion != coreworkspace.WorkspaceSchemaVersion { + return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion) + } + if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest { + return invalidDecision("checkpoint identity digest does not match current invocation") + } + if manifest.Stage != stage { + return invalidDecision("checkpoint stage %q does not match %q", manifest.Stage, stage) + } + if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID { + return invalidDecision("checkpoint lane %q does not match %q", manifest.LaneID, laneID) + } + if strings.TrimSpace(moduleKey) != "" && manifest.ModuleKey != moduleKey { + return invalidDecision("checkpoint module %q does not match %q", manifest.ModuleKey, moduleKey) + } + statusOK := false + for _, status := range statuses { + if manifest.Status == status { + statusOK = true + break + } + } + if !statusOK { + return invalidDecision("checkpoint status %q cannot be reused", manifest.Status) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) { + return invalidDecision("checkpoint dependency fingerprints do not match") + } + return reusedDecision() +} + +func sourceChunksFromEnvelope(values []chunkEnvelope) ([]contracts.SourceChunk, error) { + if len(values) == 0 { + return nil, nil + } + out := make([]contracts.SourceChunk, 0, len(values)) + for _, value := range values { + content, err := contentFromEnvelope(value.Content) + if err != nil { + return nil, err + } + out = append(out, contracts.SourceChunk{ + ID: value.ID, + SourceID: value.SourceID, + Index: value.Index, + StartUnitID: value.StartUnitID, + EndUnitID: value.EndUnitID, + Content: content, + MediaType: value.Content.MediaType, + Units: cloneSourceUnits(value.Units), + Metadata: cloneMetadata(value.Metadata), + }) + } + return out, nil +} + +func extractOutputsFromEnvelope(values []extractOutputEnvelope) ([]contracts.ExtractOutput, error) { + if len(values) == 0 { + return nil, nil + } + out := make([]contracts.ExtractOutput, 0, len(values)) + for _, value := range values { + payload, err := rawPayloadFromEnvelope(value.Payload) + if err != nil { + return nil, err + } + out = append(out, contracts.ExtractOutput{ + LaneID: value.LaneID, + ExtractorKey: value.ExtractorKey, + SourceID: value.SourceID, + ChunkID: value.ChunkID, + ChunkIndex: value.ChunkIndex, + Schema: value.Schema, + Payload: payload, + }) + } + return out, nil +} + +func mergeOutputFromEnvelope(value mergeOutputPayload) (contracts.MergeOutput, error) { + payload, err := rawPayloadFromEnvelope(value.Payload) + if err != nil { + return contracts.MergeOutput{}, err + } + return contracts.MergeOutput{ + LaneID: value.LaneID, + MergerKey: value.MergerKey, + SourceID: value.SourceID, + Schema: value.Schema, + Payload: payload, + }, nil +} + +func normalizeOutputFromEnvelope(value normalizeOutputPayload) (contracts.NormalizeOutput, error) { + payload, err := rawPayloadFromEnvelope(value.Payload) + if err != nil { + return contracts.NormalizeOutput{}, err + } + return contracts.NormalizeOutput{ + LaneID: value.LaneID, + NormalizerKey: value.NormalizerKey, + SourceID: value.SourceID, + Schema: value.Schema, + Payload: payload, + }, nil +} + +func rawPayloadFromEnvelope(value binaryEnvelope) (contracts.RawPayload, error) { + content, err := contentFromEnvelope(value) + if err != nil { + return contracts.RawPayload{}, err + } + return contracts.RawPayload{ + Content: content, + MediaType: value.MediaType, + Metadata: cloneMetadata(value.Metadata), + Warnings: cloneWarnings(value.Warnings), + }, nil +} + +func contentFromEnvelope(value binaryEnvelope) ([]byte, error) { + content, err := base64.StdEncoding.DecodeString(value.ContentBase64) + if err != nil { + return nil, fmt.Errorf("decode content_base64: %w", err) + } + if digest := strings.TrimSpace(value.ContentDigest); digest != "" && digest != contentDigest(content) { + return nil, fmt.Errorf("content digest mismatch") + } + return content, nil +} + +func coreworkspaceToPipelineFingerprints(values []coreworkspace.Fingerprint) []pipeline.CheckpointFingerprint { + if len(values) == 0 { + return nil + } + out := make([]pipeline.CheckpointFingerprint, 0, len(values)) + for _, value := range values { + out = append(out, pipeline.CheckpointFingerprint{Name: value.Name, Value: value.Value}) + } + return normalizeFingerprints(out) +} + +func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.CheckpointFingerprint) bool { + a = normalizeFingerprints(a) + b = normalizeFingerprints(b) + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func reusedDecision() pipeline.CheckpointDecision { + return pipeline.CheckpointDecision{Reused: true, Reason: "checkpoint is valid"} +} + +func invalidDecision(format string, args ...any) pipeline.CheckpointDecision { + return pipeline.CheckpointDecision{Reason: fmt.Sprintf(format, args...)} +} diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index b328342..681d392 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -17,8 +17,9 @@ import ( ) type WorkspaceRecorder struct { - root string - now func() time.Time + root string + identityDigest string + now func() time.Time } func NewWorkspaceRecorder(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointRecorder, error) { @@ -29,11 +30,11 @@ func NewWorkspaceRecorder(settings coreworkspace.Settings, identity coreworkspac if strings.TrimSpace(root) == "" { return pipeline.NoopCheckpointRecorder(), nil } - return &WorkspaceRecorder{root: root, now: time.Now}, nil + return &WorkspaceRecorder{root: root, identityDigest: identity.Digest, now: time.Now}, nil } func (r *WorkspaceRecorder) SourceRunning(moduleKey string) error { - manifest := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusRunning) + manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusRunning) manifest.ModuleKey = moduleKey manifest.StartedAt = timePtr(r.timestamp()) return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) @@ -46,7 +47,7 @@ func (r *WorkspaceRecorder) SourceSucceeded(moduleKey string, doc *source.Source if err := r.writePayload("source/source-document.json", sourceDocumentEnvelope{Document: cloneSourceDocument(*doc)}); err != nil { return err } - manifest := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusSucceeded) + manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusSucceeded) manifest.ModuleKey = moduleKey manifest.OutputDigests = workspaceFingerprints(digestFingerprints("source_document", doc.Digest)) manifest.CompletedAt = timePtr(r.timestamp()) @@ -57,7 +58,7 @@ func (r *WorkspaceRecorder) SourceSucceeded(moduleKey string, doc *source.Source } func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error { - manifest := coreworkspace.NewStageManifest(coreworkspace.StageSource, coreworkspace.StatusFailed) + manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusFailed) manifest.ModuleKey = moduleKey manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) @@ -65,7 +66,7 @@ func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error { } func (r *WorkspaceRecorder) ChunkRunning(moduleKey string, sourceDigest string) error { - manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusRunning) + manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusRunning) manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) manifest.StartedAt = timePtr(r.timestamp()) @@ -77,7 +78,7 @@ func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string if err := r.writePayload("chunk/chunks.json", payload); err != nil { return err } - manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded) + manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded) manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) manifest.OutputDigests = workspaceFingerprints(chunkOutputDigests(chunks)) @@ -90,7 +91,7 @@ func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string } func (r *WorkspaceRecorder) ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error { - manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceededWithRejections) + manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceededWithRejections) manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) manifest.ValidationStatus = "rejected" @@ -100,7 +101,7 @@ func (r *WorkspaceRecorder) ChunkRejected(moduleKey string, sourceDigest string, } func (r *WorkspaceRecorder) ChunkFailed(moduleKey string, sourceDigest string, err error) error { - manifest := coreworkspace.NewStageManifest(coreworkspace.StageChunk, coreworkspace.StatusFailed) + manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusFailed) manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) manifest.CompletedAt = timePtr(r.timestamp()) @@ -109,7 +110,7 @@ func (r *WorkspaceRecorder) ChunkFailed(moduleKey string, sourceDigest string, e } func (r *WorkspaceRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) + manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest}) } @@ -123,7 +124,7 @@ func (r *WorkspaceRecorder) ExtractSucceeded(laneID string, moduleKey string, de if err := r.writePayload(lanePayloadPath("extract", laneID, "outputs.json"), payload); err != nil { return err } - manifest := laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) + manifest := r.laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) manifest.OutputDigests = workspaceFingerprints(rawOutputDigests(extractPayloads(outputs))) manifest.ValidationStatus = validationStatusString(warnings, rejected) manifest.Rejections = rejectionSummaries(rejected) @@ -136,14 +137,14 @@ func (r *WorkspaceRecorder) ExtractSucceeded(laneID string, moduleKey string, de } func (r *WorkspaceRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := laneManifest(coreworkspace.StageExtract, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest := r.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 := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest}) } @@ -153,7 +154,7 @@ func (r *WorkspaceRecorder) MergeSucceeded(laneID string, moduleKey string, depe if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), payload); err != nil { return err } - manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest := r.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()) @@ -164,7 +165,7 @@ func (r *WorkspaceRecorder) MergeSucceeded(laneID string, moduleKey string, depe } 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 := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) @@ -172,14 +173,14 @@ func (r *WorkspaceRecorder) MergeRejected(laneID string, moduleKey string, depen } func (r *WorkspaceRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := laneManifest(coreworkspace.StageMerge, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest := r.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 := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest}) } @@ -189,7 +190,7 @@ func (r *WorkspaceRecorder) NormalizeSucceeded(laneID string, moduleKey string, if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), payload); err != nil { return err } - manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest := r.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()) @@ -197,7 +198,7 @@ func (r *WorkspaceRecorder) NormalizeSucceeded(laneID string, moduleKey string, } 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 := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) @@ -205,7 +206,7 @@ func (r *WorkspaceRecorder) NormalizeRejected(laneID string, moduleKey string, d } func (r *WorkspaceRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) + manifest := r.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}) @@ -233,8 +234,16 @@ func (r *WorkspaceRecorder) timestamp() time.Time { return r.now().UTC() } -func laneManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) coreworkspace.StageManifest { +func (r *WorkspaceRecorder) newStageManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus) coreworkspace.StageManifest { manifest := coreworkspace.NewStageManifest(stage, status) + if strings.TrimSpace(r.identityDigest) != "" { + manifest.Metadata = map[string]string{"checkpoint_identity_digest": r.identityDigest} + } + return manifest +} + +func (r *WorkspaceRecorder) laneManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) coreworkspace.StageManifest { + manifest := r.newStageManifest(stage, status) manifest.LaneID = laneID manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = workspaceFingerprints(dependencies) diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 9b9f818..5f92e9c 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -80,6 +80,149 @@ func TestWorkspaceRecorderWritesSuccessfulCheckpointFiles(t *testing.T) { } } +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"}}, + } + 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) + } + 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) @@ -181,15 +324,21 @@ func assertManifestStatus(t *testing.T, path string, want coreworkspace.StageSta 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) - } + 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) } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 3859b15..974d69d 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -37,9 +37,58 @@ type CheckpointRecorder interface { NormalizeFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error } +type CheckpointDecision struct { + Reused bool `json:"reused"` + Reason string `json:"reason,omitempty"` +} + +type CheckpointEvent struct { + Stage string `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + Action string `json:"action"` + Reason string `json:"reason,omitempty"` +} + +type SourceCheckpoint struct { + Document *source.SourceDocument +} + +type ChunkCheckpoint struct { + Chunks []contracts.SourceChunk + Warnings []contracts.Warning +} + +type ExtractCheckpoint struct { + Outputs []contracts.ExtractOutput + Rejected []contracts.RejectedOutput + Warnings []contracts.Warning +} + +type MergeCheckpoint struct { + Output contracts.MergeOutput + Warnings []contracts.Warning +} + +type NormalizeCheckpoint struct { + Output contracts.NormalizeOutput + Warnings []contracts.Warning +} + +type CheckpointLoader interface { + Enabled() bool + Source(moduleKey string) (SourceCheckpoint, CheckpointDecision) + Chunk(moduleKey string, sourceDigest string) (ChunkCheckpoint, CheckpointDecision) + Extract(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) + Merge(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) + Normalize(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) +} + type noopCheckpointRecorder struct{} +type noopCheckpointLoader struct{} func NoopCheckpointRecorder() CheckpointRecorder { return noopCheckpointRecorder{} } +func NoopCheckpointLoader() CheckpointLoader { return noopCheckpointLoader{} } func (noopCheckpointRecorder) SourceRunning(string) error { return nil } func (noopCheckpointRecorder) SourceSucceeded(string, *source.SourceDocument) error { return nil } @@ -84,6 +133,23 @@ func (noopCheckpointRecorder) NormalizeFailed(string, string, []CheckpointFinger return nil } +func (noopCheckpointLoader) Enabled() bool { return false } +func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { + return SourceCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} +} +func (noopCheckpointLoader) Chunk(string, string) (ChunkCheckpoint, CheckpointDecision) { + return ChunkCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} +} +func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + return ExtractCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} +} +func (noopCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + return MergeCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} +} +func (noopCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + return NormalizeCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} +} + func rawOutputDigests(payloads []contracts.RawPayload) []CheckpointFingerprint { values := make([]CheckpointFingerprint, 0, len(payloads)) for i, payload := range payloads { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 6c00b17..67f962c 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -49,6 +49,7 @@ type RunInput struct { Metadata map[string]any Warnings []contracts.Warning Checkpoints CheckpointRecorder + Checkpoint CheckpointLoader } type RunOutput struct { @@ -57,6 +58,7 @@ type RunOutput struct { Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` Warnings []contracts.Warning `json:"warnings,omitempty"` OutputFiles []contracts.OutputFile `json:"-"` + CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"` } func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) { @@ -75,6 +77,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if checkpoints == nil { checkpoints = NoopCheckpointRecorder() } + checkpointLoader := input.Checkpoint + if checkpointLoader == nil { + checkpointLoader = NoopCheckpointLoader() + } defer func() { output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.LLMClient)) }() @@ -85,27 +91,32 @@ 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, - Raw: input.RawInput, - LLMProfile: input.Pipeline.Input.LLMProfile, - Options: cloneOptions(input.Pipeline.Input.Options), - 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) + sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) + recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) + doc := sourceCheckpoint.Document + if !sourceDecision.Reused { + 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, + Raw: input.RawInput, + LLMProfile: input.Pipeline.Input.LLMProfile, + Options: cloneOptions(input.Pipeline.Input.Options), + 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) @@ -117,59 +128,69 @@ 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) { - chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{ - Source: doc, - SourceInput: sourceInput.Clone(), - SessionID: sessionID, - References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: input.Pipeline.Chunk.LLMProfile, - Options: cloneOptions(input.Pipeline.Chunk.Options), - Metadata: input.Metadata, + chunkCheckpoint, chunkDecision := checkpointLoader.Chunk(chunker.Key(), doc.Digest) + recordCheckpointEvent(&output, checkpointLoader, string(StageChunk), "", chunker.Key(), chunkDecision) + chunksAccepted := chunkDecision.Reused + var chunkRejection *contracts.RejectedOutput + if chunkDecision.Reused { + canonicalChunks = cloneSourceChunks(chunkCheckpoint.Chunks) + chunkWarnings = cloneWarnings(chunkCheckpoint.Warnings) + output.Warnings = append(output.Warnings, chunkWarnings...) + } else { + if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil { + return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) + } + chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.Pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{ + Source: doc, + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: input.Pipeline.Chunk.LLMProfile, + Options: cloneOptions(input.Pipeline.Chunk.Options), + Metadata: input.Metadata, + }) + if err != nil { + return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) + } + if len(chunkResult.Chunks) == 0 { + return false, nil, fmt.Errorf("chunker %q returned no chunks", chunker.Key()) + } + chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) + if err != nil { + return false, nil, fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) + } + validationWarnings, rejection, err := r.validateChunksRaw(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt) + if err != nil || rejection != nil { + return false, rejection, err + } + canonicalChunks = chunks + chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) + return true, nil, nil }) if err != nil { - return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) + _ = checkpoints.ChunkFailed(chunker.Key(), doc.Digest, err) + return failOutput(output), err } - if len(chunkResult.Chunks) == 0 { - return false, nil, fmt.Errorf("chunker %q returned no chunks", chunker.Key()) - } - chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) - if err != nil { - return false, nil, fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) - } - validationWarnings, rejection, err := r.validateChunksRaw(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt) - if err != nil || rejection != nil { - return false, rejection, err - } - canonicalChunks = chunks - chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) - 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 { + 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, checkpoints, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil { + if err := r.runLane(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil { return failOutput(output), err } } @@ -210,7 +231,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, checkpoints CheckpointRecorder, 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, checkpointLoader CheckpointLoader, 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) @@ -229,76 +250,85 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp 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 - var acceptedWarnings []contracts.Warning - accepted, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - result, err := extractor.Extract(ctx, contracts.ExtractionRequest{ - Source: doc, - Chunk: &chunk, - SourceInput: chunkInputMaterial(sourceInput, chunk), - SessionID: sessionID, - References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: lane.Extract.LLMProfile, - Options: cloneOptions(lane.Extract.Options), - Metadata: input.Metadata, + extractCheckpoint, extractDecision := checkpointLoader.Extract(lane.ID, extractor.Key(), extractDependencies) + recordCheckpointEvent(output, checkpointLoader, string(StageExtract), lane.ID, extractor.Key(), extractDecision) + if extractDecision.Reused { + extractOutputs = cloneExtractOutputs(extractCheckpoint.Outputs) + extractWarnings = cloneWarnings(extractCheckpoint.Warnings) + output.Rejected = append(output.Rejected, cloneRejectedOutputs(extractCheckpoint.Rejected)...) + output.Warnings = append(output.Warnings, extractWarnings...) + } else { + 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 + var acceptedWarnings []contracts.Warning + accepted, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + result, err := extractor.Extract(ctx, contracts.ExtractionRequest{ + Source: doc, + Chunk: &chunk, + SourceInput: chunkInputMaterial(sourceInput, chunk), + SessionID: sessionID, + References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: lane.Extract.LLMProfile, + Options: cloneOptions(lane.Extract.Options), + Metadata: input.Metadata, + }) + if err != nil { + return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err) + } + extractOutput := result.Output + extractOutput.LaneID = lane.ID + extractOutput.ExtractorKey = extractor.Key() + extractOutput.SourceID = doc.ID + extractOutput.ChunkID = chunk.ID + extractOutput.ChunkIndex = chunk.Index + extractOutput.Payload.Warnings = append(extractOutput.Payload.Warnings, result.Warnings...) + validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + stage: StageExtract, + laneID: lane.ID, + moduleKey: extractor.Key(), + source: doc, + sourceID: doc.ID, + chunkID: chunk.ID, + chunkIndex: chunk.Index, + chunk: &chunk, + sourceInput: chunkInputMaterial(sourceInput, chunk), + sessionID: sessionID, + references: lane.ExtractReferences.ReferenceSet, + llmClient: input.LLMClient, + schema: extractOutput.Schema, + payload: extractOutput.Payload, + metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, + attempt: attempt, + }) + if err != nil || rejection != nil { + return false, rejection, err + } + acceptedOutput = cloneExtractOutput(extractOutput) + acceptedWarnings = append(cloneWarnings(result.Warnings), validationWarnings...) + return true, nil, nil }) if err != nil { - return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err) + _ = checkpoints.ExtractFailed(lane.ID, extractor.Key(), extractDependencies, err) + return err } - extractOutput := result.Output - extractOutput.LaneID = lane.ID - extractOutput.ExtractorKey = extractor.Key() - extractOutput.SourceID = doc.ID - extractOutput.ChunkID = chunk.ID - extractOutput.ChunkIndex = chunk.Index - extractOutput.Payload.Warnings = append(extractOutput.Payload.Warnings, result.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ - stage: StageExtract, - laneID: lane.ID, - moduleKey: extractor.Key(), - source: doc, - sourceID: doc.ID, - chunkID: chunk.ID, - chunkIndex: chunk.Index, - chunk: &chunk, - sourceInput: chunkInputMaterial(sourceInput, chunk), - sessionID: sessionID, - references: lane.ExtractReferences.ReferenceSet, - llmClient: input.LLMClient, - schema: extractOutput.Schema, - payload: extractOutput.Payload, - metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, - attempt: attempt, - }) - if err != nil || rejection != nil { - return false, rejection, err + if !accepted { + output.Rejected = append(output.Rejected, *rejection) + continue } - acceptedOutput = cloneExtractOutput(extractOutput) - acceptedWarnings = append(cloneWarnings(result.Warnings), validationWarnings...) - return true, nil, nil - }) - if err != nil { - _ = checkpoints.ExtractFailed(lane.ID, extractor.Key(), extractDependencies, err) - return err + output.Warnings = append(output.Warnings, acceptedWarnings...) + extractWarnings = append(extractWarnings, acceptedWarnings...) + extractOutputs = append(extractOutputs, acceptedOutput) } - if !accepted { - output.Rejected = append(output.Rejected, *rejection) - continue + 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) } - 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 { @@ -308,135 +338,151 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp 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, - LaneID: lane.ID, - ExtractOutputs: cloneExtractOutputs(extractOutputs), - SourceInput: sourceInput.Clone(), - SessionID: sessionID, - References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: lane.Merge.LLMProfile, - Options: cloneOptions(lane.Merge.Options), - Metadata: input.Metadata, - }) - if err != nil { - return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err) - } - mergeOutput := mergeResult.Output - mergeOutput.LaneID = lane.ID - mergeOutput.MergerKey = merger.Key() - mergeOutput.SourceID = doc.ID - mergeOutput.Payload.Warnings = append(mergeOutput.Payload.Warnings, mergeResult.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ - stage: StageMerge, - laneID: lane.ID, - moduleKey: merger.Key(), - source: doc, - sourceID: doc.ID, - sourceInput: sourceInput.Clone(), - sessionID: sessionID, - references: lane.MergeReferences.ReferenceSet, - llmClient: input.LLMClient, - schema: mergeOutput.Schema, - payload: mergeOutput.Payload, - extractOutputs: extractOutputs, - metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, - attempt: attempt, - }) - if err != nil || rejection != nil { - return false, rejection, err - } - acceptedMerge = cloneMergeOutput(mergeOutput) - mergeWarnings = append(cloneWarnings(mergeResult.Warnings), validationWarnings...) - 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 { + mergeCheckpoint, mergeDecision := checkpointLoader.Merge(lane.ID, merger.Key(), mergeDependencies) + recordCheckpointEvent(output, checkpointLoader, string(StageMerge), lane.ID, merger.Key(), mergeDecision) + if mergeDecision.Reused { + acceptedMerge = cloneMergeOutput(mergeCheckpoint.Output) + mergeWarnings = cloneWarnings(mergeCheckpoint.Warnings) + output.Warnings = append(output.Warnings, mergeWarnings...) + } else { + 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, + LaneID: lane.ID, + ExtractOutputs: cloneExtractOutputs(extractOutputs), + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: lane.Merge.LLMProfile, + Options: cloneOptions(lane.Merge.Options), + Metadata: input.Metadata, + }) + if err != nil { + return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err) + } + mergeOutput := mergeResult.Output + mergeOutput.LaneID = lane.ID + mergeOutput.MergerKey = merger.Key() + mergeOutput.SourceID = doc.ID + mergeOutput.Payload.Warnings = append(mergeOutput.Payload.Warnings, mergeResult.Warnings...) + validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + stage: StageMerge, + laneID: lane.ID, + moduleKey: merger.Key(), + source: doc, + sourceID: doc.ID, + sourceInput: sourceInput.Clone(), + sessionID: sessionID, + references: lane.MergeReferences.ReferenceSet, + llmClient: input.LLMClient, + schema: mergeOutput.Schema, + payload: mergeOutput.Payload, + extractOutputs: extractOutputs, + metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, + attempt: attempt, + }) + if err != nil || rejection != nil { + return false, rejection, err + } + acceptedMerge = cloneMergeOutput(mergeOutput) + mergeWarnings = append(cloneWarnings(mergeResult.Warnings), validationWarnings...) + 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) } - 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, - LaneID: lane.ID, - MergeOutput: cloneMergeOutput(acceptedMerge), - SourceInput: sourceInput.Clone(), - SessionID: sessionID, - References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: lane.Normalize.LLMProfile, - Options: cloneOptions(lane.Normalize.Options), - Metadata: input.Metadata, - }) - if err != nil { - return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err) - } - normalizeOutput := normalizeResult.Output - normalizeOutput.LaneID = lane.ID - normalizeOutput.NormalizerKey = normalizer.Key() - normalizeOutput.SourceID = doc.ID - normalizeOutput.Payload.Warnings = append(normalizeOutput.Payload.Warnings, normalizeResult.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ - stage: StageNormalize, - laneID: lane.ID, - moduleKey: normalizer.Key(), - source: doc, - sourceID: doc.ID, - sourceInput: sourceInput.Clone(), - sessionID: sessionID, - references: lane.NormalizeReferences.ReferenceSet, - llmClient: input.LLMClient, - schema: normalizeOutput.Schema, - payload: normalizeOutput.Payload, - mergeOutput: acceptedMerge, - metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, - attempt: attempt, - }) - if err != nil || rejection != nil { - return false, rejection, err - } - acceptedNormalize = cloneNormalizeOutput(normalizeOutput) - normalizeWarnings = append(cloneWarnings(normalizeResult.Warnings), validationWarnings...) - 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 { + normalizeCheckpoint, normalizeDecision := checkpointLoader.Normalize(lane.ID, normalizer.Key(), normalizeDependencies) + recordCheckpointEvent(output, checkpointLoader, string(StageNormalize), lane.ID, normalizer.Key(), normalizeDecision) + if normalizeDecision.Reused { + acceptedNormalize = cloneNormalizeOutput(normalizeCheckpoint.Output) + normalizeWarnings = cloneWarnings(normalizeCheckpoint.Warnings) + output.Warnings = append(output.Warnings, normalizeWarnings...) + } else { + 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, + LaneID: lane.ID, + MergeOutput: cloneMergeOutput(acceptedMerge), + SourceInput: sourceInput.Clone(), + SessionID: sessionID, + References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), + LLMClient: input.LLMClient, + LLMProfile: lane.Normalize.LLMProfile, + Options: cloneOptions(lane.Normalize.Options), + Metadata: input.Metadata, + }) + if err != nil { + return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err) + } + normalizeOutput := normalizeResult.Output + normalizeOutput.LaneID = lane.ID + normalizeOutput.NormalizerKey = normalizer.Key() + normalizeOutput.SourceID = doc.ID + normalizeOutput.Payload.Warnings = append(normalizeOutput.Payload.Warnings, normalizeResult.Warnings...) + validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + stage: StageNormalize, + laneID: lane.ID, + moduleKey: normalizer.Key(), + source: doc, + sourceID: doc.ID, + sourceInput: sourceInput.Clone(), + sessionID: sessionID, + references: lane.NormalizeReferences.ReferenceSet, + llmClient: input.LLMClient, + schema: normalizeOutput.Schema, + payload: normalizeOutput.Payload, + mergeOutput: acceptedMerge, + metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, + attempt: attempt, + }) + if err != nil || rejection != nil { + return false, rejection, err + } + acceptedNormalize = cloneNormalizeOutput(normalizeOutput) + normalizeWarnings = append(cloneWarnings(normalizeResult.Warnings), validationWarnings...) + 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) } - 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 @@ -759,6 +805,23 @@ func failOutput(output RunOutput) RunOutput { return output } +func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage string, laneID string, moduleKey string, decision CheckpointDecision) { + if output == nil || loader == nil || !loader.Enabled() { + return + } + action := "executed" + if decision.Reused { + action = "reused" + } + output.CheckpointEvents = append(output.CheckpointEvents, CheckpointEvent{ + Stage: stage, + LaneID: laneID, + ModuleKey: moduleKey, + Action: action, + Reason: decision.Reason, + }) +} + func populateRawOutputManifest(output *RunOutput) { if output == nil { return diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index f63c81a..d24cda2 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -1023,6 +1023,127 @@ func TestRunDoesNotPassCheckpointPathsToModules(t *testing.T) { } } +func TestRunReusesCheckpointedWorkflowOutputs(t *testing.T) { + modules := defaultRunnerModules() + doc := validSourceDocument() + chunks := []contracts.SourceChunk{sourceChunkWithID("chunk-0", 0)} + extractOutput := contracts.ExtractOutput{ + LaneID: "alpha", + ExtractorKey: "extract-alpha", + SourceID: doc.ID, + ChunkID: "chunk-0", + ChunkIndex: 0, + Payload: contracts.RawPayload{ + Content: []byte(`{"cached_extract":true}`), + MediaType: "application/json", + }, + } + mergeOutput := contracts.MergeOutput{ + LaneID: "alpha", + MergerKey: "merge", + SourceID: doc.ID, + Payload: contracts.RawPayload{ + Content: []byte(`{"cached_merge":true}`), + MediaType: "application/json", + }, + } + normalizeOutput := contracts.NormalizeOutput{ + LaneID: "alpha", + NormalizerKey: "normalize", + SourceID: doc.ID, + Payload: contracts.RawPayload{ + Content: []byte(`{"cached_normalize":true}`), + MediaType: "application/json", + }, + } + loader := &runnerCheckpointLoader{ + source: SourceCheckpoint{Document: doc}, + chunk: ChunkCheckpoint{Chunks: chunks}, + extract: ExtractCheckpoint{Outputs: []contracts.ExtractOutput{extractOutput}}, + merge: MergeCheckpoint{Output: mergeOutput}, + normalize: NormalizeCheckpoint{Output: normalizeOutput}, + reuse: map[string]bool{ + "source": true, + "chunk": true, + "extract": true, + "merge": true, + "normalize": true, + }, + } + + output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + Checkpoint: loader, + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + if len(modules.input.requests) != 0 || len(modules.chunker.requests) != 0 || len(modules.extractors["extract-alpha"].requests) != 0 || len(modules.mergers["merge"].requests) != 0 || len(modules.normalizers["normalize"].requests) != 0 { + t.Fatalf("module requests = input:%d chunk:%d extract:%d merge:%d normalize:%d, want all skipped", len(modules.input.requests), len(modules.chunker.requests), len(modules.extractors["extract-alpha"].requests), len(modules.mergers["merge"].requests), len(modules.normalizers["normalize"].requests)) + } + if len(output.NormalizeOutputs) != 1 || string(output.NormalizeOutputs[0].Payload.Content) != `{"cached_normalize":true}` { + t.Fatalf("NormalizeOutputs = %#v, want cached normalize output", output.NormalizeOutputs) + } + if len(output.CheckpointEvents) != 5 { + t.Fatalf("checkpoint events = %#v, want one per reusable workflow step", output.CheckpointEvents) + } + for _, event := range output.CheckpointEvents { + if event.Action != "reused" { + t.Fatalf("checkpoint event = %#v, want reused", event) + } + } +} + +func TestRunPreservesCheckpointedExtractRejections(t *testing.T) { + modules := defaultRunnerModules() + extractOutput := contracts.ExtractOutput{ + LaneID: "alpha", + ExtractorKey: "extract-alpha", + SourceID: "source-1", + ChunkID: "chunk-1", + ChunkIndex: 1, + Payload: contracts.RawPayload{ + Content: []byte(`{"cached_extract":true}`), + MediaType: "application/json", + }, + } + rejected := contracts.RejectedOutput{ + Stage: string(StageExtract), + LaneID: "alpha", + ModuleKey: "extract-alpha", + ChunkID: "chunk-0", + ReasonCode: "invalid_shape", + Message: "invalid extract", + } + loader := &runnerCheckpointLoader{ + extract: ExtractCheckpoint{ + Outputs: []contracts.ExtractOutput{extractOutput}, + Rejected: []contracts.RejectedOutput{rejected}, + }, + reuse: map[string]bool{"extract": true}, + } + + output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + Checkpoint: loader, + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + if len(modules.extractors["extract-alpha"].requests) != 0 { + t.Fatalf("extract requests = %d, want reused checkpoint", len(modules.extractors["extract-alpha"].requests)) + } + if len(output.Rejected) != 1 || output.Rejected[0].ChunkID != "chunk-0" { + t.Fatalf("rejected outputs = %#v, want checkpointed extract rejection", output.Rejected) + } + mergeRequests := modules.mergers["merge"].requests + if len(mergeRequests) != 1 || len(mergeRequests[0].ExtractOutputs) != 1 || mergeRequests[0].ExtractOutputs[0].ChunkID != "chunk-1" { + t.Fatalf("merge extract outputs = %#v, want only checkpointed accepted extract", mergeRequests) + } +} + func TestRunOmitsRejectedExtractOutputsFromMerge(t *testing.T) { modules := defaultRunnerModules() validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false, true}, reason: "bad_extract", message: "extract rejected"} @@ -2089,6 +2210,54 @@ func (encoder *runnerOutputEncoder) ManifestMetadata() map[string]any { return encoder.manifestMetadata } +type runnerCheckpointLoader struct { + source SourceCheckpoint + chunk ChunkCheckpoint + extract ExtractCheckpoint + merge MergeCheckpoint + normalize NormalizeCheckpoint + reuse map[string]bool +} + +func (loader *runnerCheckpointLoader) Enabled() bool { + return true +} + +func (loader *runnerCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { + if loader.reuse["source"] { + return loader.source, CheckpointDecision{Reused: true, Reason: "test checkpoint"} + } + return SourceCheckpoint{}, CheckpointDecision{Reason: "test checkpoint missing"} +} + +func (loader *runnerCheckpointLoader) Chunk(string, string) (ChunkCheckpoint, CheckpointDecision) { + if loader.reuse["chunk"] { + return loader.chunk, CheckpointDecision{Reused: true, Reason: "test checkpoint"} + } + return ChunkCheckpoint{}, CheckpointDecision{Reason: "test checkpoint missing"} +} + +func (loader *runnerCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + if loader.reuse["extract"] { + return loader.extract, CheckpointDecision{Reused: true, Reason: "test checkpoint"} + } + return ExtractCheckpoint{}, CheckpointDecision{Reason: "test checkpoint missing"} +} + +func (loader *runnerCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + if loader.reuse["merge"] { + return loader.merge, CheckpointDecision{Reused: true, Reason: "test checkpoint"} + } + return MergeCheckpoint{}, CheckpointDecision{Reason: "test checkpoint missing"} +} + +func (loader *runnerCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + if loader.reuse["normalize"] { + return loader.normalize, CheckpointDecision{Reused: true, Reason: "test checkpoint"} + } + return NormalizeCheckpoint{}, CheckpointDecision{Reason: "test checkpoint missing"} +} + type fakeLLMClient struct{} func (client fakeLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {