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