From 22d4f296706c9a606261c5909369784a324bcc6b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 21 Jul 2026 21:39:23 +0000 Subject: [PATCH] Implement generated artifact handoff and provenance --- internal/core/artifacts/artifacts.go | 32 +- internal/framework/contracts/artifact.go | 1 + internal/framework/contracts/contracts.go | 65 +++- internal/framework/pipeline/debug.go | 3 +- internal/framework/pipeline/handoff.go | 287 ++++++++++++++++++ internal/framework/pipeline/handoff_test.go | 287 ++++++++++++++++++ internal/framework/pipeline/prepare.go | 40 +-- internal/framework/pipeline/references.go | 41 ++- internal/framework/pipeline/runner.go | 18 +- .../pipeline/runner_concurrency_test.go | 29 +- .../framework/pipeline/runner_concurrent.go | 11 +- internal/framework/pipeline/runner_typed.go | 20 +- 12 files changed, 749 insertions(+), 85 deletions(-) create mode 100644 internal/framework/pipeline/handoff.go create mode 100644 internal/framework/pipeline/handoff_test.go diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index cb9691e..a1ef506 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -5,6 +5,7 @@ import ( ) type ArtifactLaneManifest struct { + StepID string `json:"step_id,omitempty"` ID string `json:"id"` Extractor string `json:"extractor"` Merger string `json:"merger"` @@ -31,16 +32,25 @@ type LLMProfileManifest struct { } type ReferenceProvenance struct { - Stage string `json:"stage,omitempty"` - StepID string `json:"step_id,omitempty"` - LaneID string `json:"lane_id,omitempty"` - SlotName string `json:"slot_name"` - OriginType string `json:"origin_type"` - OriginURI string `json:"origin_uri,omitempty"` - Digest string `json:"digest,omitempty"` - MediaType string `json:"media_type,omitempty"` - SizeBytes int64 `json:"size_bytes,omitempty"` - BindingSource string `json:"binding_source,omitempty"` + Stage string `json:"stage,omitempty"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + SlotName string `json:"slot_name"` + OriginType string `json:"origin_type"` + OriginURI string `json:"origin_uri,omitempty"` + Digest string `json:"digest,omitempty"` + MediaType string `json:"media_type,omitempty"` + SizeBytes int64 `json:"size_bytes,omitempty"` + BindingSource string `json:"binding_source,omitempty"` + ArtifactKind string `json:"artifact_kind,omitempty"` + SchemaID string `json:"schema_id,omitempty"` + SchemaName string `json:"schema_name,omitempty"` + SchemaVersion string `json:"schema_version,omitempty"` + SchemaDigest string `json:"schema_digest,omitempty"` + ProducerPipeline string `json:"producer_pipeline_id,omitempty"` + ProducerStep string `json:"producer_step_id,omitempty"` + ProducerLane string `json:"producer_lane_id,omitempty"` + ProducerModule string `json:"producer_module_key,omitempty"` } type OutputSchemaProvenance struct { @@ -50,6 +60,7 @@ type OutputSchemaProvenance struct { } type NormalizedOutputManifest struct { + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id"` ModuleKey string `json:"module_key,omitempty"` SourceID string `json:"source_id,omitempty"` @@ -59,6 +70,7 @@ type NormalizedOutputManifest struct { type RejectedOutputManifest struct { Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id,omitempty"` ModuleKey string `json:"module_key,omitempty"` ChunkID string `json:"chunk_id,omitempty"` diff --git a/internal/framework/contracts/artifact.go b/internal/framework/contracts/artifact.go index 7981874..f296b82 100644 --- a/internal/framework/contracts/artifact.go +++ b/internal/framework/contracts/artifact.go @@ -30,6 +30,7 @@ type SerializedArtifact struct { // SerializedOutput associates a domain-neutral artifact with the pipeline // operation that produced it. Provenance remains outside codec-owned bytes. type SerializedOutput struct { + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id"` NormalizerKey string `json:"normalizer_key"` SourceID string `json:"source_id,omitempty"` diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 5639cf9..763b5eb 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -190,14 +190,64 @@ type ReferenceOrigin struct { URI string `json:"uri,omitempty"` } +// ReferenceProducer identifies the operation that produced a generated +// reference. It contains provenance only; referenced bytes remain in Content. +type ReferenceProducer struct { + PipelineID string `json:"pipeline_id,omitempty"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` +} + type ReferenceItem struct { - SlotName string `json:"slot_name"` - MediaType string `json:"media_type,omitempty"` - Content []byte `json:"-"` - Digest string `json:"digest,omitempty"` - Origin ReferenceOrigin `json:"origin"` - SizeBytes int64 `json:"size_bytes,omitempty"` - BindingSource string `json:"binding_source,omitempty"` + SlotName string `json:"slot_name"` + MediaType string `json:"media_type,omitempty"` + Content []byte `json:"-"` + Digest string `json:"digest,omitempty"` + Origin ReferenceOrigin `json:"origin"` + SizeBytes int64 `json:"size_bytes,omitempty"` + BindingSource string `json:"binding_source,omitempty"` + ArtifactKind ArtifactKind `json:"artifact_kind,omitempty"` + ArtifactSchema ArtifactSchema `json:"artifact_schema,omitempty"` + Producer ReferenceProducer `json:"producer,omitempty"` +} + +func CloneReferenceItem(item ReferenceItem) ReferenceItem { + item.Content = append([]byte(nil), item.Content...) + item.ArtifactSchema = CloneArtifactSchema(item.ArtifactSchema) + return item +} + +func (item ReferenceItem) MarshalJSON() ([]byte, error) { + type referenceItemJSON struct { + SlotName string `json:"slot_name"` + MediaType string `json:"media_type,omitempty"` + Digest string `json:"digest,omitempty"` + Origin ReferenceOrigin `json:"origin"` + SizeBytes int64 `json:"size_bytes,omitempty"` + BindingSource string `json:"binding_source,omitempty"` + ArtifactKind ArtifactKind `json:"artifact_kind,omitempty"` + ArtifactSchema *ArtifactSchema `json:"artifact_schema,omitempty"` + Producer *ReferenceProducer `json:"producer,omitempty"` + } + encoded := referenceItemJSON{ + SlotName: item.SlotName, + MediaType: item.MediaType, + Digest: item.Digest, + Origin: item.Origin, + SizeBytes: item.SizeBytes, + BindingSource: item.BindingSource, + ArtifactKind: item.ArtifactKind, + } + if item.ArtifactSchema.ID != "" || item.ArtifactSchema.Name != "" || item.ArtifactSchema.Version != "" || len(item.ArtifactSchema.JSONSchema) > 0 { + schema := CloneArtifactSchema(item.ArtifactSchema) + encoded.ArtifactSchema = &schema + } + if item.Producer.PipelineID != "" || item.Producer.StepID != "" || item.Producer.LaneID != "" || item.Producer.ModuleKey != "" { + producer := item.Producer + encoded.Producer = &producer + } + return json.Marshal(encoded) } type ResolvedReferenceSlot struct { @@ -261,6 +311,7 @@ type OutputEncoder interface { type RejectedOutput struct { Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id,omitempty"` ModuleKey string `json:"module_key,omitempty"` ChunkID string `json:"chunk_id,omitempty"` diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index eb9bfb9..46a9c2f 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -125,6 +125,7 @@ type debugChunkRange struct { } type debugSerializedOutput struct { + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id"` NormalizerKey string `json:"normalizer_key"` SourceID string `json:"source_id,omitempty"` @@ -557,7 +558,7 @@ func debugSerializedOutputEnvelope(output contracts.SerializedOutput) debugSeria content := debugContentEnvelope(output.Artifact.Content, output.Artifact.MediaType, output.Artifact.Metadata, nil) content.ContentDigest = debugContentDigest(output.Artifact.Content) return debugSerializedOutput{ - LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, + StepID: output.StepID, LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, Kind: output.Artifact.Kind, Schema: schema, SchemaDigest: digest, Content: content, } diff --git a/internal/framework/pipeline/handoff.go b/internal/framework/pipeline/handoff.go new file mode 100644 index 0000000..b457a1f --- /dev/null +++ b/internal/framework/pipeline/handoff.go @@ -0,0 +1,287 @@ +package pipeline + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type referenceTargetKey struct { + StepID string + LaneID string + Stage ModuleStage +} + +func keyForReferenceTarget(target ResolvedReferenceTarget) referenceTargetKey { + return referenceTargetKey{StepID: target.StepID, LaneID: target.LaneID, Stage: target.Stage} +} + +func operationReferenceSet(input RunInput, target ResolvedReferenceTarget) contracts.ReferenceSet { + if input.references != nil { + if set, ok := input.references[keyForReferenceTarget(target)]; ok { + return CloneReferenceSet(set) + } + } + return CloneReferenceSet(target.ReferenceSet) +} + +// buildStepReferenceSets resolves every generated binding for a step before +// any lane in that step is allowed to start. Each returned set is a fresh +// operation-time view; prepared reference sets are never modified. +func buildStepReferenceSets(input RunInput, step PreparedPipelineStep, outputs []contracts.SerializedOutput) (map[referenceTargetKey]contracts.ReferenceSet, []artifacts.ReferenceProvenance, error) { + if len(step.lanes) == 0 { + return nil, nil, nil + } + sets := make(map[referenceTargetKey]contracts.ReferenceSet) + var provenance []artifacts.ReferenceProvenance + canonical := make(map[string]contracts.ReferenceItem) + for _, prepared := range step.lanes { + lane := prepared.resolved + for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { + generated := false + resolved := CloneReferenceSet(target.ReferenceSet) + for _, binding := range target.Bindings { + if binding.Artifact == nil { + continue + } + generated = true + item, err := generatedReferenceItem(input, binding, outputs, canonical) + if err != nil { + return nil, nil, contextualHandoffError(input, target, binding, err) + } + slotName := strings.TrimSpace(binding.SlotName) + slot, ok := resolved.Slots[slotName] + if !ok { + return nil, nil, contextualHandoffError(input, target, binding, fmt.Errorf("reference slot %q has no resolved declaration", slotName)) + } + if len(slot.Items) > 0 { + return nil, nil, contextualHandoffError(input, target, binding, fmt.Errorf("reference slot %q already contains materialized items", slotName)) + } + if len(slot.Slot.AcceptedArtifactKinds) == 0 || !containsArtifactKind(slot.Slot.AcceptedArtifactKinds, item.ArtifactKind) { + return nil, nil, contextualHandoffError(input, target, binding, fmt.Errorf("reference slot %q does not accept artifact kind %q", slotName, item.ArtifactKind)) + } + if !referenceMediaTypeAccepted(item.MediaType, slot.Slot.AcceptedMediaTypes) { + return nil, nil, contextualHandoffError(input, target, binding, fmt.Errorf("reference slot %q does not accept media type %q", slotName, item.MediaType)) + } + if slot.Slot.MaxBytes > 0 && item.SizeBytes > slot.Slot.MaxBytes { + return nil, nil, contextualHandoffError(input, target, binding, fmt.Errorf("reference slot %q is %d bytes, limit %d", slotName, item.SizeBytes, slot.Slot.MaxBytes)) + } + slot.Items = []contracts.ReferenceItem{contracts.CloneReferenceItem(item)} + resolved.Slots[slotName] = slot + } + if !generated { + continue + } + key := keyForReferenceTarget(target) + sets[key] = resolved + for _, item := range generatedItems(resolved) { + provenance = append(provenance, referenceProvenanceForItem(target, item)) + } + } + } + return sets, provenance, nil +} + +func containsArtifactKind(kinds []contracts.ArtifactKind, want contracts.ArtifactKind) bool { + for _, kind := range kinds { + if kind == want { + return true + } + } + return false +} + +func contextualHandoffError(input RunInput, target ResolvedReferenceTarget, binding ReferenceBinding, err error) error { + return fmt.Errorf("pipeline %q step %q lane %q %s generated dependency %q from %q/%q: %w", input.pipeline.ID, target.StepID, target.LaneID, target.Stage, binding.SlotName, binding.Artifact.Step, binding.Artifact.Lane, err) +} + +func generatedReferenceItem(input RunInput, binding ReferenceBinding, outputs []contracts.SerializedOutput, cache map[string]contracts.ReferenceItem) (contracts.ReferenceItem, error) { + selector := binding.Artifact + if selector == nil { + return contracts.ReferenceItem{}, fmt.Errorf("generated reference selector must not be nil") + } + stepID := strings.TrimSpace(selector.Step) + laneID := strings.TrimSpace(selector.Lane) + cacheKey := stepID + "\x00" + laneID + if item, ok := cache[cacheKey]; ok { + item.SlotName = strings.TrimSpace(binding.SlotName) + item.BindingSource = strings.TrimSpace(binding.BindingSource) + return contracts.CloneReferenceItem(item), nil + } + matches := make([]contracts.SerializedOutput, 0, 1) + for _, output := range outputs { + if strings.TrimSpace(output.StepID) == stepID && strings.TrimSpace(output.LaneID) == laneID { + matches = append(matches, output) + } + } + if len(matches) == 0 { + return contracts.ReferenceItem{}, fmt.Errorf("producer has no accepted normalized output") + } + if len(matches) > 1 { + return contracts.ReferenceItem{}, fmt.Errorf("producer has %d accepted normalized outputs; exactly one is required", len(matches)) + } + producer, ok := findResolvedLane(input.pipeline, stepID, laneID) + if !ok { + return contracts.ReferenceItem{}, fmt.Errorf("producer lane is not present in the resolved pipeline") + } + if input.Prepared == nil || input.Prepared.artifactCodecs == nil { + return contracts.ReferenceItem{}, fmt.Errorf("artifact codec registry is unavailable") + } + serialized := contracts.CloneSerializedArtifact(matches[0].Artifact) + if serialized.Kind != producer.ArtifactKind { + return contracts.ReferenceItem{}, fmt.Errorf("producer artifact kind %q does not match resolved kind %q", serialized.Kind, producer.ArtifactKind) + } + value, err := input.Prepared.artifactCodecs.Decode(serialized) + if err != nil { + return contracts.ReferenceItem{}, fmt.Errorf("decode producer artifact: %w", err) + } + canonical, err := input.Prepared.artifactCodecs.Encode(producer.ArtifactKind, value) + if err != nil { + return contracts.ReferenceItem{}, fmt.Errorf("canonicalize producer artifact: %w", err) + } + expected, ok := input.Prepared.artifactCodecs.Spec(producer.ArtifactKind) + if !ok { + return contracts.ReferenceItem{}, fmt.Errorf("producer artifact codec %q is not registered", producer.ArtifactKind) + } + if canonical.Kind != expected.Kind || canonical.Schema.ID != expected.Schema.ID || canonical.Schema.Name != expected.Schema.Name || canonical.Schema.Version != expected.Schema.Version || contracts.DigestArtifactSchema(canonical.Schema) != expected.SchemaDigest || canonical.MediaType != expected.MediaType { + return contracts.ReferenceItem{}, fmt.Errorf("canonical producer artifact does not match registered codec identity") + } + item := contracts.ReferenceItem{ + SlotName: strings.TrimSpace(binding.SlotName), + MediaType: canonical.MediaType, + Content: append([]byte(nil), canonical.Content...), + Digest: referenceDigest(canonical.Content), + Origin: contracts.ReferenceOrigin{Type: "generated"}, + SizeBytes: int64(len(canonical.Content)), + BindingSource: strings.TrimSpace(binding.BindingSource), + ArtifactKind: canonical.Kind, + ArtifactSchema: contracts.CloneArtifactSchema(canonical.Schema), + Producer: contracts.ReferenceProducer{ + PipelineID: input.pipeline.ID, + StepID: stepID, + LaneID: laneID, + ModuleKey: producer.Normalize.Module, + }, + } + cacheItem := contracts.CloneReferenceItem(item) + cacheItem.SlotName = "" + cacheItem.BindingSource = "" + cache[cacheKey] = cacheItem + return item, nil +} + +func findResolvedLane(pipeline ResolvedPipeline, stepID, laneID string) (ResolvedArtifactLane, bool) { + for _, step := range pipeline.Steps { + if step.ID != stepID { + continue + } + for _, lane := range step.ArtifactLanes { + if lane.ID == laneID { + return lane, true + } + } + } + return ResolvedArtifactLane{}, false +} + +func generatedItems(set contracts.ReferenceSet) []contracts.ReferenceItem { + var items []contracts.ReferenceItem + keys := make([]string, 0, len(set.Slots)) + for key := range set.Slots { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + for _, item := range set.Slots[key].Items { + if item.Origin.Type == "generated" { + items = append(items, item) + } + } + } + return items +} + +func referenceProvenanceForItem(target ResolvedReferenceTarget, item contracts.ReferenceItem) artifacts.ReferenceProvenance { + return artifacts.ReferenceProvenance{ + Stage: string(target.Stage), + StepID: target.StepID, + LaneID: target.LaneID, + SlotName: item.SlotName, + OriginType: item.Origin.Type, + OriginURI: item.Origin.URI, + Digest: item.Digest, + MediaType: item.MediaType, + SizeBytes: item.SizeBytes, + BindingSource: item.BindingSource, + ArtifactKind: string(item.ArtifactKind), + SchemaID: item.ArtifactSchema.ID, + SchemaName: item.ArtifactSchema.Name, + SchemaVersion: item.ArtifactSchema.Version, + SchemaDigest: contracts.DigestArtifactSchema(item.ArtifactSchema), + ProducerPipeline: item.Producer.PipelineID, + ProducerStep: item.Producer.StepID, + ProducerLane: item.Producer.LaneID, + ProducerModule: item.Producer.ModuleKey, + } +} + +type generatedReferenceFingerprintIdentity struct { + ProducerPipeline string `json:"producer_pipeline_id"` + ProducerStep string `json:"producer_step_id"` + ProducerLane string `json:"producer_lane_id"` + ProducerModule string `json:"producer_module_key"` + ArtifactKind string `json:"artifact_kind"` + SchemaID string `json:"schema_id"` + SchemaName string `json:"schema_name"` + SchemaVersion string `json:"schema_version"` + SchemaDigest string `json:"schema_digest"` + MediaType string `json:"media_type"` + ContentDigest string `json:"content_digest"` +} + +// generatedReferenceDependencies returns the canonical semantic dependency +// identity that receiving operation checkpoints must include. +func generatedReferenceDependencies(set contracts.ReferenceSet) []CheckpointFingerprint { + var fingerprints []CheckpointFingerprint + keys := make([]string, 0, len(set.Slots)) + for key := range set.Slots { + keys = append(keys, key) + } + sort.Strings(keys) + for _, slotName := range keys { + for index, item := range set.Slots[slotName].Items { + if item.Origin.Type != "generated" { + continue + } + identity := generatedReferenceFingerprintIdentity{ + ProducerPipeline: item.Producer.PipelineID, + ProducerStep: item.Producer.StepID, + ProducerLane: item.Producer.LaneID, + ProducerModule: item.Producer.ModuleKey, + ArtifactKind: string(item.ArtifactKind), + SchemaID: item.ArtifactSchema.ID, + SchemaName: item.ArtifactSchema.Name, + SchemaVersion: item.ArtifactSchema.Version, + SchemaDigest: contracts.DigestArtifactSchema(item.ArtifactSchema), + MediaType: item.MediaType, + ContentDigest: item.Digest, + } + encoded, err := json.Marshal(identity) + if err != nil { + continue + } + sum := sha256.Sum256(encoded) + fingerprints = append(fingerprints, CheckpointFingerprint{ + Name: fmt.Sprintf("generated-reference:%s:%d", slotName, index), + Value: "sha256:" + hex.EncodeToString(sum[:]), + }) + } + } + return normalizeCheckpointFingerprints(fingerprints) +} diff --git a/internal/framework/pipeline/handoff_test.go b/internal/framework/pipeline/handoff_test.go new file mode 100644 index 0000000..f3a152b --- /dev/null +++ b/internal/framework/pipeline/handoff_test.go @@ -0,0 +1,287 @@ +package pipeline + +import ( + "context" + "encoding/json" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func handoffFixture(t *testing.T, value codecNotes) (RunInput, PreparedPipelineStep, contracts.SerializedOutput) { + t.Helper() + prepared := preparedOrderedPipeline(t, 1, + orderedLaneSpec{id: "notes", profile: "notes"}, + orderedLaneSpec{id: "score", profile: "score"}, + ) + consumer := &prepared.Steps[1].lanes[0] + installGeneratedTarget := func(target *ResolvedReferenceTarget, stage ModuleStage, module string) { + *target = ResolvedReferenceTarget{ + Stage: stage, + StepID: consumer.resolved.StepID, + LaneID: consumer.resolved.ID, + Module: module, + Bindings: []ReferenceBinding{{ + Stage: stage, + LaneID: consumer.resolved.ID, + SlotName: "producer-output", + Artifact: &ArtifactReference{Step: "step-1", Lane: "notes"}, + }}, + ReferenceSet: contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + "producer-output": {Slot: contracts.ReferenceSlot{ + Name: "producer-output", + AcceptedArtifactKinds: []contracts.ArtifactKind{"test/notes"}, + AcceptedMediaTypes: []string{"application/json"}, + }}, + }}, + } + } + installGeneratedTarget(&consumer.resolved.ExtractReferences, StageExtract, consumer.resolved.Extract.Module) + installGeneratedTarget(&consumer.resolved.MergeReferences, StageMerge, consumer.resolved.Merge.Module) + installGeneratedTarget(&consumer.resolved.NormalizeReferences, StageNormalize, consumer.resolved.Normalize.Module) + + artifact, err := checkpointArtifact(prepared.Steps[0].lanes[0].typed.codec, "notes", "typed/normalize", "source", value) + if err != nil { + t.Fatalf("checkpointArtifact() error = %v", err) + } + input := RunInput{Prepared: prepared} + input.pipeline = prepared.resolved + return input, prepared.Steps[1], contracts.SerializedOutput{ + StepID: "step-1", + LaneID: "notes", + NormalizerKey: "typed/normalize", + SourceID: "source", + Artifact: contracts.CloneSerializedArtifact(artifact.Artifact), + } +} + +func TestBuildStepReferenceSetsCanonicalizesAndClonesFanout(t *testing.T) { + input, step, producerOutput := handoffFixture(t, codecNotes{Items: []string{"first"}}) + sets, provenance, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{producerOutput}) + if err != nil { + t.Fatalf("buildStepReferenceSets() error = %v", err) + } + if got, want := len(sets), 3; got != want { + t.Fatalf("reference target set count = %d, want %d", got, want) + } + if got, want := len(provenance), 3; got != want { + t.Fatalf("generated provenance count = %d, want %d", got, want) + } + + var first []byte + for key, set := range sets { + item := set.Slots["producer-output"].Items[0] + if item.Origin.Type != "generated" || item.Origin.URI != "" { + t.Fatalf("target %v origin = %#v, want generated origin without URI", key, item.Origin) + } + if item.Producer.StepID != "step-1" || item.Producer.LaneID != "notes" || item.Producer.ModuleKey != "typed/normalize" { + t.Fatalf("target %v producer = %#v, want producer provenance", key, item.Producer) + } + if first == nil { + first = item.Content + continue + } + if &first[0] == &item.Content[0] { + t.Fatalf("target %v shares generated content backing storage", key) + } + } + if got := len(step.lanes[0].resolved.ExtractReferences.ReferenceSet.Slots["producer-output"].Items); got != 0 { + t.Fatalf("prepared extract reference items = %d, want zero", got) + } + + encoded, err := json.Marshal(provenance) + if err != nil { + t.Fatalf("marshal generated provenance: %v", err) + } + text := string(encoded) + for _, forbidden := range []string{"first", "file://", "content_base64"} { + if strings.Contains(text, forbidden) { + t.Fatalf("generated provenance contains forbidden %q: %s", forbidden, text) + } + } + if !strings.Contains(text, "producer_pipeline_id") || !strings.Contains(text, "schema_digest") { + t.Fatalf("generated provenance = %s, want bounded producer and schema identity", text) + } +} + +func TestGeneratedReferenceAcceptsTypedEmptyCollection(t *testing.T) { + input, step, producerOutput := handoffFixture(t, codecNotes{}) + sets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{producerOutput}) + if err != nil { + t.Fatalf("buildStepReferenceSets() error = %v, want nil for accepted empty collection", err) + } + if got := len(sets[keyForReferenceTarget(step.lanes[0].resolved.ExtractReferences)].Slots["producer-output"].Items); got != 1 { + t.Fatalf("generated empty collection item count = %d, want one artifact", got) + } +} + +func TestGeneratedReferenceRejectsInvalidProducerOutputsDeterministically(t *testing.T) { + tests := []struct { + name string + outputs func(contracts.SerializedOutput) []contracts.SerializedOutput + mutate func(*ResolvedReferenceTarget, *contracts.SerializedOutput) + want string + }{ + {name: "missing", outputs: func(contracts.SerializedOutput) []contracts.SerializedOutput { return nil }, want: "no accepted normalized output"}, + {name: "multiple", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + return []contracts.SerializedOutput{output, output} + }, want: "exactly one is required"}, + {name: "kind mismatch", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + return []contracts.SerializedOutput{output} + }, mutate: func(target *ResolvedReferenceTarget, _ *contracts.SerializedOutput) { + target.ReferenceSet.Slots["producer-output"] = contracts.ResolvedReferenceSlot{Slot: contracts.ReferenceSlot{Name: "producer-output", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/score"}}} + }, want: "does not accept artifact kind"}, + {name: "schema mismatch", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + return []contracts.SerializedOutput{output} + }, mutate: func(_ *ResolvedReferenceTarget, output *contracts.SerializedOutput) { + output.Artifact.Schema.ID = "wrong.schema" + }, want: "schema identity"}, + {name: "media mismatch", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + return []contracts.SerializedOutput{output} + }, mutate: func(_ *ResolvedReferenceTarget, output *contracts.SerializedOutput) { + output.Artifact.MediaType = "text/plain" + }, want: "media type"}, + {name: "size mismatch", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + return []contracts.SerializedOutput{output} + }, mutate: func(target *ResolvedReferenceTarget, _ *contracts.SerializedOutput) { + slot := target.ReferenceSet.Slots["producer-output"] + slot.Slot.MaxBytes = 1 + target.ReferenceSet.Slots["producer-output"] = slot + }, want: "limit"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + input, step, output := handoffFixture(t, codecNotes{Items: []string{"first"}}) + output = contracts.CloneSerializedOutput(output) + if test.mutate != nil { + target := &step.lanes[0].resolved.ExtractReferences + test.mutate(target, &output) + } + _, _, err := buildStepReferenceSets(input, step, test.outputs(output)) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("buildStepReferenceSets() error = %v, want %q", err, test.want) + } + }) + } +} + +func TestGeneratedReferenceFingerprintChangesWithCanonicalContent(t *testing.T) { + input, step, first := handoffFixture(t, codecNotes{Items: []string{"first"}}) + firstSets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{first}) + if err != nil { + t.Fatalf("build first reference set: %v", err) + } + _, _, second := handoffFixture(t, codecNotes{Items: []string{"second"}}) + secondSets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{second}) + if err != nil { + t.Fatalf("build second reference set: %v", err) + } + firstDeps := generatedReferenceDependencies(firstSets[keyForReferenceTarget(step.lanes[0].resolved.ExtractReferences)]) + secondDeps := generatedReferenceDependencies(secondSets[keyForReferenceTarget(step.lanes[0].resolved.ExtractReferences)]) + if reflect.DeepEqual(firstDeps, secondDeps) { + t.Fatalf("generated dependencies = %#v, want content-sensitive fingerprint", firstDeps) + } +} + +func TestRunnerHandsOffAcceptedNormalizedOutputBeforeConsumerLanes(t *testing.T) { + input, _, _ := handoffFixture(t, codecNotes{Items: []string{"first"}}) + prepared := input.Prepared + var received contracts.ReferenceSet + prepared.Steps[0].lanes[0].typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"first"}}}, nil + } + prepared.Steps[1].lanes[0].typed.extract = func(_ context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + received = CloneReferenceSet(request.References) + return erasedTypedResult{Value: codecScore{Value: 3}}, nil + } + + output, err := New().Run(context.Background(), input) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if got, want := len(output.NormalizeOutputs), 2; got != want { + t.Fatalf("normalized output count = %d, want %d", got, want) + } + if output.NormalizeOutputs[0].StepID != "step-1" || output.NormalizeOutputs[1].StepID != "step-2" { + t.Fatalf("normalized output step IDs = %#v, want step-1 and step-2", output.NormalizeOutputs) + } + item := received.Slots["producer-output"].Items[0] + if item.Origin.Type != "generated" || item.Producer.StepID != "step-1" || item.Producer.LaneID != "notes" { + t.Fatalf("consumer reference = %#v, want generated producer reference", item) + } + if len(output.Manifest.References) == 0 || output.Manifest.References[len(output.Manifest.References)-1].OriginType != "generated" { + t.Fatalf("manifest references = %#v, want generated provenance", output.Manifest.References) + } +} + +type handoffDependencyLoader struct { + CheckpointLoader + extract [][]CheckpointFingerprint + merge [][]CheckpointFingerprint + normalize [][]CheckpointFingerprint +} + +func (l *handoffDependencyLoader) Enabled() bool { return true } + +func (l *handoffDependencyLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + l.extract = append(l.extract, append([]CheckpointFingerprint(nil), dependencies...)) + return ExtractCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func (l *handoffDependencyLoader) Merge(_ string, _ string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + l.merge = append(l.merge, append([]CheckpointFingerprint(nil), dependencies...)) + return MergeCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func (l *handoffDependencyLoader) Normalize(_ string, _ string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + l.normalize = append(l.normalize, append([]CheckpointFingerprint(nil), dependencies...)) + return NormalizeCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func runHandoffWithProducerValue(t *testing.T, value codecNotes) *handoffDependencyLoader { + t.Helper() + input, _, _ := handoffFixture(t, value) + prepared := input.Prepared + prepared.Steps[0].lanes[0].typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + prepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + prepared.Steps[1].lanes[0].typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecScore{Value: 3}}, nil + } + loader := &handoffDependencyLoader{CheckpointLoader: NoopCheckpointLoader()} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader}); err != nil { + t.Fatalf("Run() error = %v", err) + } + return loader +} + +func TestRunnerAddsGeneratedDependencyToEveryReceivingCheckpoint(t *testing.T) { + first := runHandoffWithProducerValue(t, codecNotes{Items: []string{"first"}}) + second := runHandoffWithProducerValue(t, codecNotes{Items: []string{"second"}}) + if len(first.extract) != 2 || len(first.merge) != 2 || len(first.normalize) != 2 { + t.Fatalf("checkpoint load calls = extract %d merge %d normalize %d, want two each", len(first.extract), len(first.merge), len(first.normalize)) + } + for name, calls := range map[string][][]CheckpointFingerprint{"extract": first.extract, "merge": first.merge, "normalize": first.normalize} { + if got := generatedFingerprintCount(calls[1]); got != 1 { + t.Fatalf("%s consumer dependency count = %d, want one", name, got) + } + } + if reflect.DeepEqual(first.extract[1], second.extract[1]) || reflect.DeepEqual(first.merge[1], second.merge[1]) || reflect.DeepEqual(first.normalize[1], second.normalize[1]) { + t.Fatalf("consumer checkpoint dependencies did not change with producer content") + } +} + +func generatedFingerprintCount(dependencies []CheckpointFingerprint) int { + count := 0 + for _, dependency := range dependencies { + if strings.HasPrefix(dependency.Name, "generated-reference:") { + count++ + } + } + return count +} diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index 9e02c3d..1f74ecf 100644 --- a/internal/framework/pipeline/prepare.go +++ b/internal/framework/pipeline/prepare.go @@ -23,6 +23,7 @@ type PreparedPipeline struct { chunker contracts.Chunker chunkValidators preparedValidatorChain output contracts.OutputEncoder + artifactCodecs *ArtifactCodecRegistry checkpointFingerprints []CheckpointFingerprint } @@ -78,11 +79,12 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend } stable := cloneResolvedPipeline(resolved) prepared := &PreparedPipeline{ - Input: cloneModuleBinding(stable.Input), - Chunk: cloneModuleBinding(stable.Chunk), - Output: cloneModuleBinding(stable.Output), - resolved: stable, - dependencies: deps, + Input: cloneModuleBinding(stable.Input), + Chunk: cloneModuleBinding(stable.Chunk), + Output: cloneModuleBinding(stable.Output), + resolved: stable, + dependencies: deps, + artifactCodecs: registries.ArtifactCodecs, } request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { return BuildRequest{Dependencies: deps, Options: cloneOptions(binding.Options), References: references} @@ -134,31 +136,6 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend return prepared, nil } -func rejectGeneratedBindings(resolved ResolvedPipeline) error { - check := func(stepID, laneID string, target ResolvedReferenceTarget) error { - for _, binding := range target.Bindings { - if binding.Artifact == nil { - continue - } - return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q uses a generated artifact; generated reference execution is not supported yet", resolved.ID, stepID, laneID, target.Stage, binding.SlotName) - } - return nil - } - if err := check("", "", resolved.ChunkReferences); err != nil { - return err - } - for _, step := range resolved.Steps { - for _, lane := range step.ArtifactLanes { - for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { - if err := check(step.ID, lane.ID, target); err != nil { - return err - } - } - } - } - return nil -} - func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { @@ -339,6 +316,9 @@ func validateRegistrySet(resolved ResolvedPipeline, registries Registries) error if registries.Chunkers == nil { return fmt.Errorf("chunker registry must not be nil") } + if registries.ArtifactCodecs == nil { + return fmt.Errorf("artifact codec registry must not be nil") + } if registries.Extractors == nil { return fmt.Errorf("extractor registry must not be nil") } diff --git a/internal/framework/pipeline/references.go b/internal/framework/pipeline/references.go index 289f10f..dfed190 100644 --- a/internal/framework/pipeline/references.go +++ b/internal/framework/pipeline/references.go @@ -114,8 +114,9 @@ func materializeReferenceTarget( return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetContext(pipelineID, target), slotName, target.Stage, target.Module) } if binding.Artifact != nil { - // Generated selectors are resolved at the step handoff. They do not - // have bytes during static reference materialization. + // Generated selectors are resolved at the step handoff. Keep the + // declared slot and its constraints, but do not materialize bytes yet. + set.Slots[slotName] = contracts.ResolvedReferenceSlot{Slot: cloneReferenceSlot(slot)} continue } @@ -312,8 +313,7 @@ func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet { if len(slot.Items) > 0 { items := make([]contracts.ReferenceItem, len(slot.Items)) for i, item := range slot.Items { - item.Content = append([]byte(nil), item.Content...) - items[i] = item + items[i] = contracts.CloneReferenceItem(item) } slot.Items = items } @@ -366,17 +366,30 @@ func referenceTargetProvenance(target ResolvedReferenceTarget) []artifacts.Refer for _, slotName := range slotNames { slot := target.ReferenceSet.Slots[slotName] for _, item := range slot.Items { + schemaDigest := "" + if item.ArtifactSchema.ID != "" || item.ArtifactSchema.Name != "" || item.ArtifactSchema.Version != "" || len(item.ArtifactSchema.JSONSchema) > 0 { + schemaDigest = contracts.DigestArtifactSchema(item.ArtifactSchema) + } provenance = append(provenance, artifacts.ReferenceProvenance{ - Stage: string(target.Stage), - StepID: target.StepID, - LaneID: target.LaneID, - SlotName: item.SlotName, - OriginType: item.Origin.Type, - OriginURI: item.Origin.URI, - Digest: item.Digest, - MediaType: item.MediaType, - SizeBytes: item.SizeBytes, - BindingSource: item.BindingSource, + Stage: string(target.Stage), + StepID: target.StepID, + LaneID: target.LaneID, + SlotName: item.SlotName, + OriginType: item.Origin.Type, + OriginURI: item.Origin.URI, + Digest: item.Digest, + MediaType: item.MediaType, + SizeBytes: item.SizeBytes, + BindingSource: item.BindingSource, + ArtifactKind: string(item.ArtifactKind), + SchemaID: item.ArtifactSchema.ID, + SchemaName: item.ArtifactSchema.Name, + SchemaVersion: item.ArtifactSchema.Version, + SchemaDigest: schemaDigest, + ProducerPipeline: item.Producer.PipelineID, + ProducerStep: item.Producer.StepID, + ProducerLane: item.Producer.LaneID, + ProducerModule: item.Producer.ModuleKey, }) } } diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 93aebc4..28174ee 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -57,9 +57,10 @@ type RunInput struct { // single worker so direct framework callers retain deterministic behavior. ExtractWorkers int - pipeline ResolvedPipeline - llmClient contracts.StructuredLLMClient - stepID string + pipeline ResolvedPipeline + llmClient contracts.StructuredLLMClient + stepID string + references map[referenceTargetKey]contracts.ReferenceSet } type RunOutput struct { @@ -253,6 +254,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err for _, step := range input.Prepared.Steps { stepInput := input stepInput.stepID = step.ID + stepReferences, referenceProvenance, handoffErr := buildStepReferenceSets(input, step, output.NormalizeOutputs) + if handoffErr != nil { + return failOutput(output), fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, handoffErr) + } + stepInput.references = stepReferences + output.Manifest.References = append(output.Manifest.References, referenceProvenance...) laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) if err := mergeLaneOutput(&output, laneOutput); err != nil { return failOutput(output), err @@ -468,7 +475,7 @@ func validateRunInput(input RunInput) error { if err := validateResolvedPipeline(input.Prepared.resolved); err != nil { return err } - return rejectGeneratedBindings(input.Prepared.resolved) + return nil } func validateResolvedPipeline(pipeline ResolvedPipeline) error { @@ -547,6 +554,7 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { } for _, lane := range pipeline.AllArtifactLanes() { laneManifest := artifacts.ArtifactLaneManifest{ + StepID: lane.StepID, ID: lane.ID, Extractor: lane.Extract.Module, Merger: lane.Merge.Module, @@ -621,6 +629,7 @@ func normalizedOutputManifests(outputs []contracts.SerializedOutput) []artifacts manifests := make([]artifacts.NormalizedOutputManifest, 0, len(outputs)) for _, output := range outputs { manifests = append(manifests, artifacts.NormalizedOutputManifest{ + StepID: output.StepID, LaneID: output.LaneID, ModuleKey: output.NormalizerKey, SourceID: output.SourceID, @@ -643,6 +652,7 @@ func rejectedOutputManifests(rejected []contracts.RejectedOutput) []artifacts.Re for _, output := range rejected { manifests = append(manifests, artifacts.RejectedOutputManifest{ Stage: output.Stage, + StepID: output.StepID, LaneID: output.LaneID, ModuleKey: output.ModuleKey, ChunkID: output.ChunkID, diff --git a/internal/framework/pipeline/runner_concurrency_test.go b/internal/framework/pipeline/runner_concurrency_test.go index 769ba89..c3cc8a1 100644 --- a/internal/framework/pipeline/runner_concurrency_test.go +++ b/internal/framework/pipeline/runner_concurrency_test.go @@ -119,23 +119,38 @@ func (a *countingInputAdapter) Parse(ctx context.Context, request contracts.Pars return a.InputAdapter.Parse(ctx, request) } -func TestRunnerRejectsGeneratedBindingsBeforeSourceParsing(t *testing.T) { +func TestRunnerFailsGeneratedHandoffBeforeConsumerExtraction(t *testing.T) { prepared := preparedConcurrentPipeline(t, 1) input := &countingInputAdapter{InputAdapter: prepared.input} prepared.input = input - prepared.resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings = []ReferenceBinding{{ + preparedLane := &prepared.Steps[0].lanes[0] + lane := &preparedLane.resolved + binding := ReferenceBinding{ Stage: StageExtract, - LaneID: prepared.resolved.Steps[0].ArtifactLanes[0].ID, + LaneID: lane.ID, SlotName: "generated", Artifact: &ArtifactReference{Step: "producer", Lane: "source"}, + } + lane.ExtractReferences.Bindings = []ReferenceBinding{binding} + lane.ExtractReferences.ReferenceSet = contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + "generated": {Slot: contracts.ReferenceSlot{Name: "generated", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/notes"}}}, }} + prepared.resolved.Steps[0].ArtifactLanes[0].ExtractReferences = CloneReferenceTarget(lane.ExtractReferences) + var extractCalls atomic.Int32 + preparedLane.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + extractCalls.Add(1) + return erasedTypedResult{Value: codecNotes{}}, nil + } _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) - if err == nil || !strings.Contains(err.Error(), "generated reference execution is not supported yet") { - t.Fatalf("Run() error = %v, want generated binding rejection", err) + if err == nil || !strings.Contains(err.Error(), "generated dependency") { + t.Fatalf("Run() error = %v, want generated dependency failure", err) } - if got := input.calls.Load(); got != 0 { - t.Fatalf("input Parse calls = %d, want zero", got) + if got := input.calls.Load(); got != 1 { + t.Fatalf("input Parse calls = %d, want one before handoff", got) + } + if got := extractCalls.Load(); got != 0 { + t.Fatalf("consumer extract calls = %d, want zero", got) } } diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index b7664c4..8eddc08 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -230,7 +230,9 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou if err != nil { return nil, fmt.Errorf("digest chunks for lane %q: %w", lane.ID, err) } - state := &laneExtractState{index: index, prepared: prepared, deps: digestFingerprints("chunks", digest), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))} + extractReferences := operationReferenceSet(input, lane.ExtractReferences) + deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...) + state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))} cp, decision := loadExtract(loader, lane.ID, lane.Extract.Module, state.deps) if decision.Reused { for _, stored := range cp.Outputs { @@ -285,7 +287,8 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr)) } - extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata}) + extractReferences := operationReferenceSet(input, lane.ExtractReferences) + extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(extractReferences), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr) return false, nil, terminal.record(nil, attemptErr) @@ -299,7 +302,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. return false, nil, terminal.record(payload, attemptErr) } serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: extractReferences, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -440,7 +443,7 @@ func mergeLaneOutput(dst *RunOutput, src RunOutput) error { dst.CheckpointEvents = append(dst.CheckpointEvents, src.CheckpointEvents...) for i := range dst.Manifest.ArtifactLanes { for j := range src.Manifest.ArtifactLanes { - if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && src.Manifest.ArtifactLanes[j].Metadata != nil { + if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && dst.Manifest.ArtifactLanes[i].StepID == src.Manifest.ArtifactLanes[j].StepID && src.Manifest.ArtifactLanes[j].Metadata != nil { metadata, err := cloneMetadata(src.Manifest.ArtifactLanes[j].Metadata) if err != nil { return fmt.Errorf("clone lane %q manifest metadata: %w", dst.Manifest.ArtifactLanes[i].ID, err) diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 5c42c5a..15d8cc1 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -154,7 +154,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi for i, value := range extracts.accepted { mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value} } - mergeDeps := artifactCheckpointDigests(extracts.serialized) + mergeReferences := operationReferenceSet(input, lane.MergeReferences) + mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...) + mergeDeps = normalizeCheckpointFingerprints(mergeDeps) mergeCP, mergeDecision := loadMerge(loader, lane.ID, lane.Merge.Module, mergeDeps) if mergeDecision.Reused { if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { @@ -193,7 +195,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr)) } - result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata}) + result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) return false, nil, terminal.record(nil, attemptErr) @@ -205,7 +207,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr) return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -244,7 +246,9 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi } activeStage = StageNormalize - normalizeDeps := artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}) + normalizeReferences := operationReferenceSet(input, lane.NormalizeReferences) + normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...) + normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps) normalizeCP, normalizeDecision := loadNormalize(loader, lane.ID, lane.Normalize.Module, normalizeDeps) if normalizeDecision.Reused { if _, decodeErr := decodeCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil { @@ -281,7 +285,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr)) } - result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata}) + result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) return false, nil, terminal.record(nil, attemptErr) @@ -292,7 +296,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr) return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: normalizeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -329,7 +333,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { return err } - output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)}) + output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)}) return nil } @@ -431,7 +435,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if message == "" { message = "artifact rejected" } - return warnings, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { + return warnings, &contracts.RejectedOutput{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { if target.chunk != nil { return target.chunk.ID }