eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reuse valid workspace checkpoints on request

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-08 03:02:50 +00:00
parent 1d3a444df8
commit ae9c2e1d5e
14 changed files with 1406 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
docs/cli.md
View File

@@ -21,7 +21,7 @@ ID in config or with `--llm-profile`.
```text
notarius help
notarius run <pipeline-id> --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 <pipeline-id> --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 `<path>/<run-id>/`.
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).

4
docs/config.md
View File

@@ -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 `<workspace.directory>/checkpoints/`. Checkpoint reads and
resume execution are not implemented.
artifacts under `<workspace.directory>/checkpoints/`. `notarius run --resume`
can reuse valid checkpoints from a compatible invocation.
Debug artifact writers are not part of the current workflow.

6
docs/internal/pipeline.md
View File

@@ -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

17
docs/operations.md
View File

@@ -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

32
internal/cli/run.go
View File

@@ -29,7 +29,7 @@ const defaultOutputRoot = "./notarius-output"
const usage = `Usage:
notarius help
notarius run <pipeline-id> --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 <pipeline-id> --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 {

266
internal/cli/run_test.go
View File

@@ -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)

1
internal/core/diagnostics/artifacts.go
View File

@@ -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"

5
internal/core/diagnostics/run_dir.go
View File

@@ -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)
}

345
internal/framework/checkpoint/loader.go Normal file
View File

@@ -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...)}
}

53
internal/framework/checkpoint/recorder.go
View File

@@ -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)

157
internal/framework/checkpoint/recorder_test.go
View File

@@ -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) }

66
internal/framework/pipeline/checkpoint.go
View File

@@ -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 {

561
internal/framework/pipeline/runner.go
View File

@@ -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

169
internal/framework/pipeline/runner_test.go
View File

@@ -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) {