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

Implement ordered pipeline step resolution

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-21 21:05:28 +00:00
parent f5618d1f0c
commit f846f252c0
44 changed files with 1335 additions and 398 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
docs/adr/0008-ordered-pipeline-steps.md Normal file
View File

@@ -0,0 +1,50 @@
# ADR-0008: Bounded ordered pipeline steps and explicit artifact references
**Status:** Accepted
**Date:** 2026-07-21
## Context
Notarius currently models one pipeline-wide input, chunking plan, artifact
lanes, and output boundary. Some workflows need a deterministic handoff from
one set of normalized artifacts to a later set of artifacts, such as using
extracted NPC records while grounding later combat events. The workflow needs
an explicit topology without turning the pipeline into a general-purpose
workflow engine.
## Decision
Add an ordered collection of pipeline steps. Each step owns one or more
artifact lanes, and lanes within a step retain the existing independent
execution model. The pipeline continues to have one input, chunk plan, output,
and failure boundary. Steps are barriers: a later step may consume only
normalized artifacts from an earlier step.
Generated references use an explicit step-and-lane selector. Reference slots
declare the generated artifact kinds and media types they accept. The resolver
validates the topology, ordering, lane identity, artifact kind, schema, and
codec compatibility before execution. External references remain supported as
path sources, and the legacy top-level artifact map is interpreted as an
implicit `default` step.
Pipeline-level references may not select generated artifacts. General DAGs,
branches, loops, conditional execution, joins, and inferred dependencies are
not part of this model.
## Alternatives considered
- A general DAG would provide more flexibility but would also require a new
scheduler, lifecycle model, failure semantics, and provenance model.
- Separate pipeline runs connected through filesystem paths would lose the
static topology and typed compatibility checks.
- Inferring dependencies from module or lane names would make ordering and
configuration errors difficult to detect reliably.
## Consequences
The resolved pipeline has a deterministic, inspectable topology and can
include it in its identity digest. Configuration validation can reject invalid
generated bindings before any work begins. Existing single-step profiles keep
their behavior through the implicit `default` step. Execution handoff and
multi-step scheduling require follow-up work in the runner and checkpoint
layers.

8
internal/cli/dnd_combat_contract_test.go
View File

@@ -24,10 +24,10 @@ func TestProductionCombatConfigurationResolvesTypedLane(t *testing.T) {
if effective.ResolvedPipeline.Chunk.Module != pipeline.DefaultChunkModule {
t.Fatalf("chunk module = %q, want %q", effective.ResolvedPipeline.Chunk.Module, pipeline.DefaultChunkModule)
}
if len(effective.ResolvedPipeline.ArtifactLanes) != 1 {
t.Fatalf("artifact lanes = %#v, want one combat lane", effective.ResolvedPipeline.ArtifactLanes)
if len(effective.ResolvedPipeline.Steps[0].ArtifactLanes) != 1 {
t.Fatalf("artifact lanes = %#v, want one combat lane", effective.ResolvedPipeline.Steps[0].ArtifactLanes)
}
lane := effective.ResolvedPipeline.ArtifactLanes[0]
lane := effective.ResolvedPipeline.Steps[0].ArtifactLanes[0]
if lane.ID != "combat" || lane.ArtifactKind != dnd.CombatTurnListKind || lane.Extract.Module != combatextract.Key || lane.Extract.Retries != 2 || lane.Merge.Module != pipeline.DefaultMergeModule || lane.Normalize.Module != combatnormalize.Key {
t.Fatalf("resolved combat lane = %#v, want typed production composition", lane)
}
@@ -89,7 +89,7 @@ func TestProductionCombatConfigurationResolvesTypedLane(t *testing.T) {
if err != nil {
t.Fatalf("Resolve(bound references) error = %v, want nil", err)
}
boundLane := bound.ResolvedPipeline.ArtifactLanes[0]
boundLane := bound.ResolvedPipeline.Steps[0].ArtifactLanes[0]
if len(boundLane.ExtractReferences.Bindings) != 1 || len(boundLane.NormalizeReferences.Bindings) != 1 || boundLane.ExtractReferences.Bindings[0].SlotName != "npcs" || boundLane.NormalizeReferences.Bindings[0].SlotName != "npcs" {
t.Fatalf("bound combat references = %#v / %#v, want one independent NPC binding per stage", boundLane.ExtractReferences, boundLane.NormalizeReferences)
}

10
internal/cli/dnd_npc_contract_test.go
View File

@@ -24,10 +24,10 @@ func TestProductionNPCConfigurationResolvesTypedLane(t *testing.T) {
if effective.ResolvedPipeline.Chunk.Module != pipeline.DefaultChunkModule {
t.Fatalf("chunk module = %q, want %q", effective.ResolvedPipeline.Chunk.Module, pipeline.DefaultChunkModule)
}
if len(effective.ResolvedPipeline.ArtifactLanes) != 1 {
t.Fatalf("artifact lanes = %#v, want one NPC lane", effective.ResolvedPipeline.ArtifactLanes)
if len(effective.ResolvedPipeline.Steps[0].ArtifactLanes) != 1 {
t.Fatalf("artifact lanes = %#v, want one NPC lane", effective.ResolvedPipeline.Steps[0].ArtifactLanes)
}
lane := effective.ResolvedPipeline.ArtifactLanes[0]
lane := effective.ResolvedPipeline.Steps[0].ArtifactLanes[0]
if lane.ID != "npcs" || lane.ArtifactKind != dnd.NPCListKind || lane.Extract.Module != npcextract.Key || lane.Extract.Retries != 2 || lane.Merge.Module != pipeline.DefaultMergeModule || lane.Normalize.Module != npcnormalize.Key {
t.Fatalf("resolved NPC lane = %#v, want typed production composition", lane)
}
@@ -101,11 +101,11 @@ func TestProductionNPCConfigurationValidatesOptionsReferencesAndPlacement(t *tes
t.Fatalf("unknown normalizer option error = %v, want strict option rejection", err)
}
if err := resolve(func(profile *pipeline.PipelineProfile) {
profile.References = map[string]string{
profile.References = pipeline.ExternalReferenceMap(map[string]string{
"players": "players.txt",
"party": "party.txt",
"glossary": "glossary.txt",
}
})
}); err != nil {
t.Fatalf("optional NPC references error = %v, want resolution success", err)
}

8
internal/cli/example_contract_test.go
View File

@@ -33,10 +33,10 @@ func TestMaintainedExamplesLoadResolveAndList(t *testing.T) {
t.Fatalf("materialize maintained example references for %q: %v", pipelineID, err)
}
if example.name == "production" {
if len(materialized.ArtifactLanes) != 1 ||
len(materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["spell_catalog"].Items) != 1 ||
len(materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"].Items) != 1 {
t.Fatalf("production spell catalog reference was not materialized: %#v", materialized.ArtifactLanes)
if len(materialized.Steps[0].ArtifactLanes) != 1 ||
len(materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["spell_catalog"].Items) != 1 ||
len(materialized.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"].Items) != 1 {
t.Fatalf("production spell catalog reference was not materialized: %#v", materialized.Steps[0].ArtifactLanes)
}
}
}

14
internal/cli/production_contract_test.go
View File

@@ -187,8 +187,8 @@ func TestProductionSpellValidatorsPrepareFromMaterializedCatalog(t *testing.T) {
if err != nil {
t.Fatalf("materialize production spell references: %v", err)
}
extractItems := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["spell_catalog"].Items
normalizeItems := materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"].Items
extractItems := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["spell_catalog"].Items
normalizeItems := materialized.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"].Items
if len(extractItems) != 1 || extractItems[0].MediaType != "application/json" || len(extractItems[0].Content) == 0 {
t.Fatalf("materialized extract spell catalog items = %#v, want one JSON item", extractItems)
}
@@ -243,9 +243,9 @@ func TestProductionSpellNormalizerRejectsInvalidCatalogReferencesBeforeExecution
if err != nil {
t.Fatal(err)
}
slot := materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"]
slot := materialized.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"]
slot.Items = append(slot.Items, slot.Items[0])
materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"] = slot
materialized.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["spell_catalog"] = slot
_, err = pipeline.Prepare(materialized, components.registries, pipeline.ModuleDependencies{LLM: &productionFakeLLMClient{}})
for _, fragment := range []string{"normalize", `module "dnd/spells"`, "zero or one item"} {
if err == nil || !strings.Contains(err.Error(), fragment) {
@@ -307,10 +307,10 @@ func TestProductionSpellNormalizerRejectsInvalidCatalogReferencesBeforeExecution
func setNormalizeSpellCatalogSource(t *testing.T, resolved *pipeline.ResolvedPipeline, sourcePath string) {
t.Helper()
if resolved == nil || len(resolved.ArtifactLanes) != 1 {
if resolved == nil || len(resolved.Steps[0].ArtifactLanes) != 1 {
t.Fatalf("resolved pipeline = %#v, want one artifact lane", resolved)
}
bindings := resolved.ArtifactLanes[0].NormalizeReferences.Bindings
bindings := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings
matches := 0
for index := range bindings {
if bindings[index].SlotName == "spell_catalog" {
@@ -321,7 +321,7 @@ func setNormalizeSpellCatalogSource(t *testing.T, resolved *pipeline.ResolvedPip
if matches != 1 {
t.Fatalf("normalize reference bindings = %#v, want exactly one spell_catalog binding", bindings)
}
resolved.ArtifactLanes[0].NormalizeReferences.Bindings = bindings
resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings = bindings
}
func TestProductionLLMClientFactoriesBuildOfflineRuntime(t *testing.T) {

20
internal/cli/reference_contract_test.go
View File

@@ -319,28 +319,28 @@ func referenceContractConfig() config.Config {
Extract: pipeline.Binding("reference/extract-alpha"),
Merge: pipeline.Binding("reference/shared-merge"),
Normalize: pipeline.Binding("reference/shared-normalize"),
References: map[string]string{"required-extract": "required.txt"},
References: pipeline.ExternalReferenceMap(map[string]string{"required-extract": "required.txt"}),
},
"beta": {
Extract: pipeline.Binding("reference/extract-beta"),
Merge: pipeline.Binding("reference/shared-merge"),
Normalize: pipeline.Binding("reference/shared-normalize"),
References: map[string]string{"required-extract": "required.txt"},
References: pipeline.ExternalReferenceMap(map[string]string{"required-extract": "required.txt"}),
},
},
},
}
profile := cfg.Pipelines["demo"]
profile.Chunk.References = map[string]string{"required-chunk": "required.txt"}
profile.Chunk.References = pipeline.ExternalReferenceMap(map[string]string{"required-chunk": "required.txt"})
alpha := profile.Artifacts["alpha"]
alpha.Extract.References = map[string]string{"required-extract": "required.txt", "alpha-slot": "optional.txt"}
alpha.Merge.References = map[string]string{"required-merge": "required.txt"}
alpha.Normalize.References = map[string]string{"required-normalize": "required.txt"}
alpha.Extract.References = pipeline.ExternalReferenceMap(map[string]string{"required-extract": "required.txt", "alpha-slot": "optional.txt"})
alpha.Merge.References = pipeline.ExternalReferenceMap(map[string]string{"required-merge": "required.txt"})
alpha.Normalize.References = pipeline.ExternalReferenceMap(map[string]string{"required-normalize": "required.txt"})
profile.Artifacts["alpha"] = alpha
beta := profile.Artifacts["beta"]
beta.Extract.References = map[string]string{"required-extract": "required.txt"}
beta.Merge.References = map[string]string{"required-merge": "required.txt"}
beta.Normalize.References = map[string]string{"required-normalize": "required.txt"}
beta.Extract.References = pipeline.ExternalReferenceMap(map[string]string{"required-extract": "required.txt"})
beta.Merge.References = pipeline.ExternalReferenceMap(map[string]string{"required-merge": "required.txt"})
beta.Normalize.References = pipeline.ExternalReferenceMap(map[string]string{"required-normalize": "required.txt"})
profile.Artifacts["beta"] = beta
cfg.Pipelines["demo"] = profile
return cfg
@@ -418,7 +418,7 @@ func (referenceContractCodecB) Decode([]byte) (stateTestArtifact, error) {
func referenceContractLane(t *testing.T, resolved pipeline.ResolvedPipeline, id string) pipeline.ResolvedArtifactLane {
t.Helper()
for _, lane := range resolved.ArtifactLanes {
for _, lane := range resolved.Steps[0].ArtifactLanes {
if lane.ID == id {
return lane
}

33
internal/cli/run.go
View File

@@ -749,7 +749,7 @@ func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string {
}
}
add(resolved.Chunk)
for _, lane := range resolved.ArtifactLanes {
for _, lane := range resolved.AllArtifactLanes() {
add(lane.Extract)
add(lane.Merge)
add(lane.Normalize)
@@ -1178,17 +1178,34 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
if !ok {
return nil, fmt.Errorf("pipeline %q is not configured", strings.TrimSpace(pipelineID))
}
if profile.Steps != nil && len(only) > 0 {
return nil, fmt.Errorf("pipeline %q --only is not supported for explicit ordered steps", strings.TrimSpace(pipelineID))
}
lanesByID := make(map[string]pipeline.ArtifactLaneProfile, len(profile.Artifacts))
for rawLaneID, lane := range profile.Artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return nil, fmt.Errorf("pipeline %q artifact lane id must not be empty", strings.TrimSpace(pipelineID))
addLanes := func(values map[string]pipeline.ArtifactLaneProfile) error {
for rawLaneID, lane := range values {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return fmt.Errorf("pipeline %q artifact lane id must not be empty", strings.TrimSpace(pipelineID))
}
if _, ok := lanesByID[laneID]; ok {
return fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", strings.TrimSpace(pipelineID), laneID)
}
lanesByID[laneID] = lane
}
if _, ok := lanesByID[laneID]; ok {
return nil, fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", strings.TrimSpace(pipelineID), laneID)
return nil
}
if profile.Steps == nil {
if err := addLanes(profile.Artifacts); err != nil {
return nil, err
}
} else {
for _, step := range profile.Steps {
if err := addLanes(step.Artifacts); err != nil {
return nil, err
}
}
lanesByID[laneID] = lane
}
selectedIDs := make([]string, 0, len(lanesByID))

9
internal/cli/run_contract_test.go
View File

@@ -268,13 +268,14 @@ func TestEffectiveLLMProfileIDsAreSortedDeduplicatedAndLLMOnly(t *testing.T) {
resolved := pipeline.ResolvedPipeline{
Input: pipeline.ModuleBinding{LLMProfile: "input-profile"},
Chunk: pipeline.ModuleBinding{LLMProfile: " zeta "},
ArtifactLanes: []pipeline.ResolvedArtifactLane{
{
Steps: []pipeline.ResolvedPipelineStep{{
ID: "default",
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
Extract: pipeline.ModuleBinding{LLMProfile: "alpha"},
Merge: pipeline.ModuleBinding{LLMProfile: "zeta"},
Normalize: pipeline.ModuleBinding{LLMProfile: " gamma "},
},
},
}},
}},
ValidatorChains: []pipeline.ResolvedValidatorChain{{Validators: []pipeline.ResolvedValidator{
{Binding: pipeline.ModuleBinding{LLMProfile: "deterministic-profile"}, ExecutionClass: contracts.ExecutionClassDeterministic},
{Binding: pipeline.ModuleBinding{LLMProfile: "beta"}, ExecutionClass: contracts.ExecutionClassLLMBacked},

12
internal/cli/spell_catalog_identity_contract_test.go
View File

@@ -31,7 +31,7 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin
}
overlayPath := filepath.Join(t.TempDir(), "catalog.json")
resolved := effective.ResolvedPipeline
bindings := resolved.ArtifactLanes[0].ExtractReferences.Bindings
bindings := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings
catalogBindingIndex := -1
for index, binding := range bindings {
if binding.SlotName == spellcatalog.SpellCatalogReferenceSlot {
@@ -42,8 +42,8 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin
if catalogBindingIndex < 0 {
t.Fatalf("spell catalog bindings = %#v, want catalog binding", bindings)
}
resolved.ArtifactLanes[0].ExtractReferences.Bindings[catalogBindingIndex].Source = overlayPath
normalizeBindings := resolved.ArtifactLanes[0].NormalizeReferences.Bindings
resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings[catalogBindingIndex].Source = overlayPath
normalizeBindings := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings
normalizeCatalogBindingIndex := -1
for index, binding := range normalizeBindings {
if binding.SlotName == spellcatalog.SpellCatalogReferenceSlot {
@@ -54,7 +54,7 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin
if normalizeCatalogBindingIndex < 0 {
t.Fatalf("normalize spell catalog bindings = %#v, want catalog binding", normalizeBindings)
}
resolved.ArtifactLanes[0].NormalizeReferences.Bindings[normalizeCatalogBindingIndex].Source = overlayPath
resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings[normalizeCatalogBindingIndex].Source = overlayPath
if err := os.WriteFile(overlayPath, []byte(reorderedOverlayA), 0o600); err != nil {
t.Fatal(err)
@@ -415,7 +415,7 @@ func catalogCheckpointIdentity(t *testing.T, resolved pipeline.ResolvedPipeline)
func catalogExtractorMetadata(t *testing.T, resolved pipeline.ResolvedPipeline) map[string]any {
t.Helper()
lane := resolved.ArtifactLanes[0]
lane := resolved.Steps[0].ArtifactLanes[0]
extractor, err := spells.New(&productionFakeLLMClient{}, spells.Options{}, lane.ExtractReferences.ReferenceSet)
if err != nil {
t.Fatalf("construct extractor: %v", err)
@@ -425,7 +425,7 @@ func catalogExtractorMetadata(t *testing.T, resolved pipeline.ResolvedPipeline)
func catalogNormalizerMetadata(t *testing.T, resolved pipeline.ResolvedPipeline) map[string]any {
t.Helper()
lane := resolved.ArtifactLanes[0]
lane := resolved.Steps[0].ArtifactLanes[0]
normalizer, err := spellnormalize.New(spellnormalize.Options{}, lane.NormalizeReferences.ReferenceSet)
if err != nil {
t.Fatalf("construct normalizer: %v", err)

2
internal/cli/spell_catalog_retry_contract_test.go
View File

@@ -63,7 +63,7 @@ func TestProductionSpellCatalogValidationRetries(t *testing.T) {
if err != nil {
t.Fatalf("materialize production references: %v", err)
}
materialized.ArtifactLanes[0].Extract.Retries = retries
materialized.Steps[0].ArtifactLanes[0].Extract.Retries = retries
llmClient := &catalogRetryLLMClient{responses: tt.responses}
prepared, err := pipeline.Prepare(materialized, components.registries, pipeline.ModuleDependencies{LLM: llmClient})

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

@@ -32,6 +32,7 @@ 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"`

45
internal/core/config/config.go
View File

@@ -107,7 +107,26 @@ func clonePipelineProfile(in pipeline.PipelineProfile) pipeline.PipelineProfile
out.Input = cloneModuleBinding(in.Input)
out.Chunk = cloneModuleBinding(in.Chunk)
out.Output = cloneModuleBinding(in.Output)
out.References = cloneStringMap(in.References)
out.References = cloneReferenceSourceMap(in.References)
if len(in.Artifacts) > 0 {
out.Artifacts = make(map[string]pipeline.ArtifactLaneProfile, len(in.Artifacts))
for key, lane := range in.Artifacts {
out.Artifacts[key] = cloneArtifactLaneProfile(lane)
}
}
if in.Steps != nil {
out.Steps = make([]pipeline.PipelineStepProfile, len(in.Steps))
for i, step := range in.Steps {
out.Steps[i] = clonePipelineStepProfile(step)
}
}
return out
}
func clonePipelineStepProfile(in pipeline.PipelineStepProfile) pipeline.PipelineStepProfile {
out := in
out.ID = in.ID
out.References = cloneReferenceSourceMap(in.References)
if len(in.Artifacts) > 0 {
out.Artifacts = make(map[string]pipeline.ArtifactLaneProfile, len(in.Artifacts))
for key, lane := range in.Artifacts {
@@ -122,7 +141,7 @@ func cloneArtifactLaneProfile(in pipeline.ArtifactLaneProfile) pipeline.Artifact
out.Extract = cloneModuleBinding(in.Extract)
out.Merge = cloneModuleBinding(in.Merge)
out.Normalize = cloneModuleBinding(in.Normalize)
out.References = cloneStringMap(in.References)
out.References = cloneReferenceSourceMap(in.References)
if len(in.Validators) > 0 {
out.Validators = make([]pipeline.ModuleBinding, len(in.Validators))
for i, binding := range in.Validators {
@@ -143,12 +162,32 @@ func cloneStringMap(in map[string]string) map[string]string {
return out
}
func cloneReferenceSourceMap(in map[string]pipeline.ReferenceSource) map[string]pipeline.ReferenceSource {
if len(in) == 0 {
return nil
}
out := make(map[string]pipeline.ReferenceSource, len(in))
for key, source := range in {
out[key] = cloneReferenceSource(source)
}
return out
}
func cloneReferenceSource(in pipeline.ReferenceSource) pipeline.ReferenceSource {
out := in
if in.Artifact != nil {
artifact := *in.Artifact
out.Artifact = &artifact
}
return out
}
func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding {
out := in
if len(in.Options) > 0 {
out.Options = cloneOptions(in.Options)
}
out.References = cloneStringMap(in.References)
out.References = cloneReferenceSourceMap(in.References)
out.Validators = cloneValidatorOverride(in.Validators)
return out
}

16
internal/core/config/effective_config.go
View File

@@ -67,11 +67,17 @@ func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) {
func applyLLMProfileOverride(profile *pipeline.PipelineProfile, profileID string) {
profile.Chunk.LLMProfile = profileID
for laneID, lane := range profile.Artifacts {
lane.Extract.LLMProfile = profileID
lane.Merge.LLMProfile = profileID
lane.Normalize.LLMProfile = profileID
profile.Artifacts[laneID] = lane
apply := func(artifacts map[string]pipeline.ArtifactLaneProfile) {
for laneID, lane := range artifacts {
lane.Extract.LLMProfile = profileID
lane.Merge.LLMProfile = profileID
lane.Normalize.LLMProfile = profileID
artifacts[laneID] = lane
}
}
apply(profile.Artifacts)
for index := range profile.Steps {
apply(profile.Steps[index].Artifacts)
}
}

16
internal/core/config/effective_config_contract_test.go
View File

@@ -53,8 +53,8 @@ func TestEffectiveConfigOnlySelectsRequestedLanesWithoutMutatingSource(t *testin
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
if len(effective.ResolvedPipeline.ArtifactLanes) != 1 || effective.ResolvedPipeline.ArtifactLanes[0].ID != "other" {
t.Fatalf("resolved lanes = %#v", effective.ResolvedPipeline.ArtifactLanes)
if len(effective.ResolvedPipeline.Steps[0].ArtifactLanes) != 1 || effective.ResolvedPipeline.Steps[0].ArtifactLanes[0].ID != "other" {
t.Fatalf("resolved lanes = %#v", effective.ResolvedPipeline.Steps[0].ArtifactLanes)
}
if len(cfg.Pipelines["main"].Artifacts) != 2 {
t.Fatalf("source lanes were mutated: %#v", cfg.Pipelines["main"].Artifacts)
@@ -79,8 +79,8 @@ func TestEffectiveConfigMaterializesDefaultBindingsThroughCatalog(t *testing.T)
if resolved.Chunk.Module != pipeline.DefaultChunkModule || resolved.Output.Module != pipeline.DefaultOutputModule {
t.Fatalf("default pipeline bindings = %#v, %#v", resolved.Chunk, resolved.Output)
}
if len(resolved.ArtifactLanes) != 1 || resolved.ArtifactLanes[0].Merge.Module != pipeline.DefaultMergeModule || resolved.ArtifactLanes[0].Normalize.Module != pipeline.DefaultNormalizeModule {
t.Fatalf("default lane bindings = %#v", resolved.ArtifactLanes)
if len(resolved.Steps[0].ArtifactLanes) != 1 || resolved.Steps[0].ArtifactLanes[0].Merge.Module != pipeline.DefaultMergeModule || resolved.Steps[0].ArtifactLanes[0].Normalize.Module != pipeline.DefaultNormalizeModule {
t.Fatalf("default lane bindings = %#v", resolved.Steps[0].ArtifactLanes)
}
}
@@ -182,8 +182,8 @@ func TestEffectiveConfigLLMProfileOverrideChangesDigestWithoutOverridingValidato
t.Fatal("LLM profile override did not change the pipeline digest")
}
resolved := overridden.ResolvedPipeline
if resolved.Chunk.LLMProfile != "override-profile" || resolved.ArtifactLanes[0].Extract.LLMProfile != "override-profile" ||
resolved.ArtifactLanes[0].Merge.LLMProfile != "override-profile" || resolved.ArtifactLanes[0].Normalize.LLMProfile != "override-profile" {
if resolved.Chunk.LLMProfile != "override-profile" || resolved.Steps[0].ArtifactLanes[0].Extract.LLMProfile != "override-profile" ||
resolved.Steps[0].ArtifactLanes[0].Merge.LLMProfile != "override-profile" || resolved.Steps[0].ArtifactLanes[0].Normalize.LLMProfile != "override-profile" {
t.Fatalf("pipeline profile override was not applied: %#v", resolved)
}
validators := findEffectiveValidatorChain(resolved, pipeline.StageExtract, "lane")
@@ -249,7 +249,7 @@ func TestEffectiveConfigValidatorOverridesRemainDistinctAndOrdered(t *testing.T)
func TestEffectiveConfigAndResolutionInputsDoNotAliasSource(t *testing.T) {
profile := effectiveProfile()
profile.Chunk.Options = map[string]any{"nested": map[string]any{"safe": "source"}}
profile.Chunk.References = map[string]string{"chunk-ref": "chunk.txt"}
profile.Chunk.References = pipeline.ExternalReferenceMap(map[string]string{"chunk-ref": "chunk.txt"})
lane := profile.Artifacts["lane"]
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
@@ -282,7 +282,7 @@ func TestEffectiveConfigAndResolutionInputsDoNotAliasSource(t *testing.T) {
if got := cfg.Pipelines["main"].Chunk.Options["nested"].(map[string]any)["safe"]; got != "source" {
t.Fatalf("source config option was aliased: %v", got)
}
if got := cfg.Pipelines["main"].Chunk.References["chunk-ref"]; got != "chunk.txt" {
if got := cfg.Pipelines["main"].Chunk.References["chunk-ref"].Path; got != "chunk.txt" {
t.Fatalf("source config references were aliased: %v", got)
}
if only[0] != "lane" || overrides[0].Source != "source.txt" {

337
internal/core/config/file_config.go
View File

@@ -28,19 +28,88 @@ type FileScriptoriumConfig struct {
}
type FilePipelineProfile struct {
Input fileModuleBinding `yaml:"input"`
Chunk *fileModuleBinding `yaml:"chunk,omitempty"`
Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"`
Output *fileModuleBinding `yaml:"output,omitempty"`
References map[string]string `yaml:"references,omitempty"`
Input fileModuleBinding `yaml:"input"`
Chunk *fileModuleBinding `yaml:"chunk,omitempty"`
Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"`
Steps []FilePipelineStepProfile `yaml:"steps,omitempty"`
Output *fileModuleBinding `yaml:"output,omitempty"`
References map[string]fileReferenceSource `yaml:"references,omitempty"`
artifactsSet bool `yaml:"-"`
stepsSet bool `yaml:"-"`
}
func (p *FilePipelineProfile) UnmarshalYAML(node *yaml.Node) error {
type plainFilePipelineProfile FilePipelineProfile
var decoded plainFilePipelineProfile
seen, err := decodeKnownMapping(node, &decoded, map[string]struct{}{
"input": {}, "chunk": {}, "artifacts": {}, "steps": {}, "output": {}, "references": {},
}, "pipeline profile")
if err != nil {
return err
}
*p = FilePipelineProfile(decoded)
_, p.artifactsSet = seen["artifacts"]
_, p.stepsSet = seen["steps"]
return nil
}
func (s *FilePipelineStepProfile) UnmarshalYAML(node *yaml.Node) error {
type plainFilePipelineStepProfile FilePipelineStepProfile
var decoded plainFilePipelineStepProfile
if _, err := decodeKnownMapping(node, &decoded, map[string]struct{}{
"id": {}, "artifacts": {}, "references": {},
}, "pipeline step"); err != nil {
return err
}
*s = FilePipelineStepProfile(decoded)
return nil
}
func (l *FileArtifactLaneProfile) UnmarshalYAML(node *yaml.Node) error {
type plainFileArtifactLaneProfile FileArtifactLaneProfile
var decoded plainFileArtifactLaneProfile
if _, err := decodeKnownMapping(node, &decoded, map[string]struct{}{
"extract": {}, "merge": {}, "normalize": {}, "validators": {}, "references": {},
}, "artifact lane"); err != nil {
return err
}
*l = FileArtifactLaneProfile(decoded)
return nil
}
func decodeKnownMapping(node *yaml.Node, target any, allowed map[string]struct{}, context string) (map[string]struct{}, error) {
if node.Kind != yaml.MappingNode {
return nil, fmt.Errorf("%s must be an object", context)
}
if err := node.Decode(target); err != nil {
return nil, err
}
seen := make(map[string]struct{}, len(node.Content)/2)
for i := 0; i < len(node.Content); i += 2 {
key := node.Content[i].Value
if _, exists := seen[key]; exists {
return nil, fmt.Errorf("%s field %q is duplicated", context, key)
}
if _, ok := allowed[key]; !ok {
return nil, fmt.Errorf("field %s not found in %s", key, context)
}
seen[key] = struct{}{}
}
return seen, nil
}
type FilePipelineStepProfile struct {
ID string `yaml:"id"`
Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts"`
References map[string]fileReferenceSource `yaml:"references,omitempty"`
}
type FileArtifactLaneProfile struct {
Extract fileModuleBinding `yaml:"extract"`
Merge *fileModuleBinding `yaml:"merge,omitempty"`
Normalize *fileModuleBinding `yaml:"normalize,omitempty"`
Validators []fileModuleBinding `yaml:"validators,omitempty"`
References map[string]string `yaml:"references,omitempty"`
Extract fileModuleBinding `yaml:"extract"`
Merge *fileModuleBinding `yaml:"merge,omitempty"`
Normalize *fileModuleBinding `yaml:"normalize,omitempty"`
Validators []fileModuleBinding `yaml:"validators,omitempty"`
References map[string]fileReferenceSource `yaml:"references,omitempty"`
}
type FileConcurrencyConfig struct {
@@ -72,10 +141,90 @@ type fileModuleBinding struct {
LLMProfile string
Retries int
Options map[string]any
References map[string]string
References map[string]fileReferenceSource
Validators pipeline.ValidatorOverride
}
type fileReferenceSource struct {
path string
artifact *pipeline.ArtifactReference
}
func (source *fileReferenceSource) UnmarshalYAML(node *yaml.Node) error {
if source == nil {
return fmt.Errorf("reference source must not be nil")
}
switch node.Kind {
case yaml.ScalarNode:
if node.Tag != "!!str" {
return fmt.Errorf("external reference path must be a string")
}
path := strings.TrimSpace(node.Value)
if path == "" {
return fmt.Errorf("external reference path must not be empty")
}
source.path = path
source.artifact = nil
return nil
case yaml.MappingNode:
if len(node.Content) != 2 || node.Content[0].Value != "artifact" {
return fmt.Errorf("reference source mapping must contain only artifact")
}
artifactNode := node.Content[1]
if artifactNode.Kind != yaml.MappingNode {
return fmt.Errorf("artifact reference must be an object")
}
var step, lane string
seen := map[string]bool{}
for i := 0; i < len(artifactNode.Content); i += 2 {
key := artifactNode.Content[i].Value
value := artifactNode.Content[i+1]
if seen[key] {
return fmt.Errorf("artifact reference field %q is duplicated", key)
}
seen[key] = true
if value.Tag != "!!str" {
return fmt.Errorf("artifact reference field %q must be a string", key)
}
switch key {
case "step":
step = strings.TrimSpace(value.Value)
case "lane":
lane = strings.TrimSpace(value.Value)
default:
return fmt.Errorf("field %s not found in artifact reference", key)
}
}
if step == "" || lane == "" {
return fmt.Errorf("artifact reference step and lane must not be empty")
}
source.path = ""
source.artifact = &pipeline.ArtifactReference{Step: step, Lane: lane}
return nil
default:
return fmt.Errorf("reference source must be a string or object")
}
}
func (source fileReferenceSource) toPipelineSource() pipeline.ReferenceSource {
if source.artifact != nil {
artifact := *source.artifact
return pipeline.ReferenceSource{Artifact: &artifact}
}
return pipeline.ExternalReference(source.path)
}
func fileReferenceSourcesToPipeline(values map[string]fileReferenceSource) map[string]pipeline.ReferenceSource {
if len(values) == 0 {
return nil
}
out := make(map[string]pipeline.ReferenceSource, len(values))
for key, value := range values {
out[strings.TrimSpace(key)] = value.toPipelineSource()
}
return out
}
func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
switch node.Kind {
case yaml.ScalarNode:
@@ -115,7 +264,7 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
}
b.Options = normalizeOptions(options)
case "references":
var references map[string]string
var references map[string]fileReferenceSource
if err := valueNode.Decode(&references); err != nil {
return err
}
@@ -146,7 +295,7 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
LLMProfile: strings.TrimSpace(b.LLMProfile),
Retries: b.Retries,
Options: cloneOptions(b.Options),
References: normalizedStringMap(b.References),
References: fileReferenceSourcesToPipeline(b.References),
Validators: b.Validators,
}
}
@@ -214,10 +363,49 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
}
for _, pipelineID := range pipelineIDs {
filePipeline := fileCfg.Pipelines[rawPipelineIDs[pipelineID]]
hasArtifacts := filePipeline.artifactsSet || filePipeline.Artifacts != nil
hasSteps := filePipeline.stepsSet || filePipeline.Steps != nil
if hasArtifacts && hasSteps {
return fmt.Errorf("pipeline %q must not declare both artifacts and steps", pipelineID)
}
if hasSteps && len(filePipeline.Steps) == 0 {
return fmt.Errorf("pipeline %q must declare at least one ordered step", pipelineID)
}
if hasSteps {
seenSteps := make(map[string]struct{}, len(filePipeline.Steps))
seenLanes := make(map[string]struct{})
for index, step := range filePipeline.Steps {
stepID := strings.TrimSpace(step.ID)
if stepID == "" {
return fmt.Errorf("pipeline %q step[%d] id must not be empty", pipelineID, index)
}
if _, ok := seenSteps[stepID]; ok {
return fmt.Errorf("pipeline %q step id %q is duplicated after trimming", pipelineID, stepID)
}
seenSteps[stepID] = struct{}{}
laneIDs, rawLaneIDs, err := normalizedMapKeys(step.Artifacts, fmt.Sprintf("pipeline %q step %q artifact lane id", pipelineID, stepID))
if err != nil {
return err
}
for _, laneID := range laneIDs {
if _, ok := seenLanes[laneID]; ok {
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps", pipelineID, laneID)
}
seenLanes[laneID] = struct{}{}
fileLane := step.Artifacts[rawLaneIDs[laneID]]
if err := validateFileLaneReferences(pipelineID, stepID, laneID, fileLane); err != nil {
return err
}
}
if err := validateFileReferenceSources(step.References, fmt.Sprintf("pipeline %q step %q reference slot", pipelineID, stepID)); err != nil {
return err
}
}
}
if _, _, err := normalizedMapKeys(filePipeline.Artifacts, fmt.Sprintf("pipeline %q artifact lane id", pipelineID)); err != nil {
return err
}
if _, _, err := normalizedMapKeys(filePipeline.References, fmt.Sprintf("pipeline %q reference slot", pipelineID)); err != nil {
if err := validateFileReferenceSources(filePipeline.References, fmt.Sprintf("pipeline %q reference slot", pipelineID)); err != nil {
return err
}
if filePipeline.Chunk != nil {
@@ -281,6 +469,7 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
for _, pipelineID := range pipelineIDs {
filePipeline := fileCfg.Pipelines[rawPipelineIDs[pipelineID]]
hasSteps := filePipeline.stepsSet || filePipeline.Steps != nil
laneIDs, rawLaneIDs, err := normalizedMapKeys(filePipeline.Artifacts, fmt.Sprintf("pipeline %q artifact lane id", pipelineID))
if err != nil {
return err
@@ -289,7 +478,7 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
ID: pipelineID,
Input: filePipeline.Input.toPipelineBinding(),
Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(filePipeline.Artifacts)),
References: normalizedStringMap(filePipeline.References),
References: fileReferenceSourcesToPipeline(filePipeline.References),
}
if filePipeline.Chunk != nil {
profile.Chunk = filePipeline.Chunk.toPipelineBinding()
@@ -300,10 +489,10 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
for _, laneID := range laneIDs {
fileLane := filePipeline.Artifacts[rawLaneIDs[laneID]]
extract := fileLane.Extract.toPipelineBinding()
extract.References = mergeStringMaps(normalizedStringMap(fileLane.References), extract.References)
extract.References = mergeReferenceSources(fileReferenceSourcesToPipeline(fileLane.References), extract.References)
lane := pipeline.ArtifactLaneProfile{
Extract: extract,
References: normalizedStringMap(fileLane.References),
References: fileReferenceSourcesToPipeline(fileLane.References),
}
if fileLane.Merge != nil {
lane.Merge = fileLane.Merge.toPipelineBinding()
@@ -319,6 +508,42 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
}
profile.Artifacts[laneID] = lane
}
if hasSteps {
profile.Artifacts = nil
profile.Steps = make([]pipeline.PipelineStepProfile, len(filePipeline.Steps))
for i, fileStep := range filePipeline.Steps {
stepID := strings.TrimSpace(fileStep.ID)
step := pipeline.PipelineStepProfile{
ID: stepID,
Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(fileStep.Artifacts)),
References: fileReferenceSourcesToPipeline(fileStep.References),
}
stepLaneIDs, stepRawLaneIDs, err := normalizedMapKeys(fileStep.Artifacts, fmt.Sprintf("pipeline %q step %q artifact lane id", pipelineID, stepID))
if err != nil {
return err
}
for _, laneID := range stepLaneIDs {
fileLane := fileStep.Artifacts[stepRawLaneIDs[laneID]]
extract := fileLane.Extract.toPipelineBinding()
extract.References = mergeReferenceSources(fileReferenceSourcesToPipeline(fileLane.References), extract.References)
lane := pipeline.ArtifactLaneProfile{Extract: extract, References: fileReferenceSourcesToPipeline(fileLane.References)}
if fileLane.Merge != nil {
lane.Merge = fileLane.Merge.toPipelineBinding()
}
if fileLane.Normalize != nil {
lane.Normalize = fileLane.Normalize.toPipelineBinding()
}
if len(fileLane.Validators) > 0 {
lane.Validators = make([]pipeline.ModuleBinding, len(fileLane.Validators))
for index, validator := range fileLane.Validators {
lane.Validators[index] = validator.toPipelineBinding()
}
}
step.Artifacts[laneID] = lane
}
profile.Steps[i] = step
}
}
c.Pipelines[pipelineID] = profile
}
@@ -430,30 +655,72 @@ func normalizedMapKeys[T any](values map[string]T, keyName string) ([]string, ma
return keys, rawByNormalized, nil
}
func normalizedStringMap(values map[string]string) map[string]string {
if len(values) == 0 {
return nil
func validateFileReferenceSources(values map[string]fileReferenceSource, context string) error {
seen := make(map[string]struct{}, len(values))
for rawSlot, source := range values {
slot := strings.TrimSpace(rawSlot)
if slot == "" {
return fmt.Errorf("%s must not be empty", context)
}
if _, ok := seen[slot]; ok {
return fmt.Errorf("%s %q is duplicated after trimming", context, slot)
}
seen[slot] = struct{}{}
if source.artifact != nil {
if strings.TrimSpace(source.artifact.Step) == "" || strings.TrimSpace(source.artifact.Lane) == "" {
return fmt.Errorf("%s %q artifact selector step and lane must not be empty", context, slot)
}
if strings.TrimSpace(source.path) != "" {
return fmt.Errorf("%s %q must contain either an external path or artifact selector", context, slot)
}
continue
}
if strings.TrimSpace(source.path) == "" {
return fmt.Errorf("%s %q source must not be empty", context, slot)
}
}
out := make(map[string]string, len(values))
keys := make([]string, 0, len(values))
rawByNormalized := make(map[string]string, len(values))
for rawKey := range values {
key := strings.TrimSpace(rawKey)
rawByNormalized[key] = rawKey
keys = append(keys, key)
}
sort.Strings(keys)
for _, key := range keys {
out[key] = strings.TrimSpace(values[rawByNormalized[key]])
}
return out
return nil
}
func mergeStringMaps(base map[string]string, override map[string]string) map[string]string {
func validateFileLaneReferences(pipelineID, stepID, laneID string, lane FileArtifactLaneProfile) error {
prefix := fmt.Sprintf("pipeline %q step %q lane %q", pipelineID, stepID, laneID)
references := []struct {
label string
values map[string]fileReferenceSource
}{
{label: "reference slot", values: lane.References},
{label: "extract reference slot", values: lane.Extract.References},
}
if lane.Merge != nil {
references = append(references, struct {
label string
values map[string]fileReferenceSource
}{label: "merge reference slot", values: lane.Merge.References})
}
if lane.Normalize != nil {
references = append(references, struct {
label string
values map[string]fileReferenceSource
}{label: "normalize reference slot", values: lane.Normalize.References})
}
for _, item := range references {
if err := validateFileReferenceSources(item.values, prefix+" "+item.label); err != nil {
return err
}
}
for index, validator := range lane.Validators {
if err := validateFileReferenceSources(validator.References, fmt.Sprintf("%s validator[%d] reference slot", prefix, index)); err != nil {
return err
}
}
return nil
}
func mergeReferenceSources(base map[string]pipeline.ReferenceSource, override map[string]pipeline.ReferenceSource) map[string]pipeline.ReferenceSource {
if len(base) == 0 && len(override) == 0 {
return nil
}
out := make(map[string]string, len(base)+len(override))
out := make(map[string]pipeline.ReferenceSource, len(base)+len(override))
for key, value := range base {
out[key] = value
}

96
internal/core/config/file_config_contract_test.go
View File

@@ -142,7 +142,7 @@ pipelines:
}
if profile.Chunk.Module != "generic" || profile.Chunk.LLMProfile != "chunk-profile" || profile.Chunk.Retries != 2 ||
!reflect.DeepEqual(profile.Chunk.Options, map[string]any{"max_units": 25}) ||
!reflect.DeepEqual(profile.Chunk.References, map[string]string{"glossary": "./glossary.md"}) {
!reflect.DeepEqual(profile.Chunk.References, pipeline.ExternalReferenceMap(map[string]string{"glossary": "./glossary.md"})) {
t.Fatalf("object binding = %#v", profile.Chunk)
}
if !profile.Chunk.Validators.Set || len(profile.Chunk.Validators.Validators) != 0 {
@@ -192,33 +192,33 @@ pipelines:
normalize-only: ./normalize.txt
`)
profile := cfg.Pipelines["main"]
if !reflect.DeepEqual(profile.References, map[string]string{
if !reflect.DeepEqual(profile.References, pipeline.ExternalReferenceMap(map[string]string{
"pipeline-only": "./pipeline.txt",
"shared": "./pipeline-shared.txt",
}) {
})) {
t.Fatalf("pipeline references = %#v", profile.References)
}
if !reflect.DeepEqual(profile.Chunk.References, map[string]string{"chunk-only": "./chunk.txt"}) {
if !reflect.DeepEqual(profile.Chunk.References, pipeline.ExternalReferenceMap(map[string]string{"chunk-only": "./chunk.txt"})) {
t.Fatalf("chunk references = %#v", profile.Chunk.References)
}
lane := profile.Artifacts["spells"]
if !reflect.DeepEqual(lane.References, map[string]string{
if !reflect.DeepEqual(lane.References, pipeline.ExternalReferenceMap(map[string]string{
"lane-only": "./lane.txt",
"shared": "./lane-shared.txt",
"overridden": "./lane.txt",
}) {
})) {
t.Fatalf("lane compatibility references = %#v", lane.References)
}
if !reflect.DeepEqual(lane.Extract.References, map[string]string{
if !reflect.DeepEqual(lane.Extract.References, pipeline.ExternalReferenceMap(map[string]string{
"lane-only": "./lane.txt",
"shared": "./lane-shared.txt",
"overridden": "./extract-overridden.txt",
"extract-only": "./extract.txt",
}) {
})) {
t.Fatalf("extract references = %#v", lane.Extract.References)
}
if !reflect.DeepEqual(lane.Merge.References, map[string]string{"merge-only": "./merge.txt"}) ||
!reflect.DeepEqual(lane.Normalize.References, map[string]string{"normalize-only": "./normalize.txt"}) {
if !reflect.DeepEqual(lane.Merge.References, pipeline.ExternalReferenceMap(map[string]string{"merge-only": "./merge.txt"})) ||
!reflect.DeepEqual(lane.Normalize.References, pipeline.ExternalReferenceMap(map[string]string{"normalize-only": "./normalize.txt"})) {
t.Fatalf("merge/normalize references = %#v, %#v", lane.Merge.References, lane.Normalize.References)
}
}
@@ -357,6 +357,82 @@ func TestFileConfigRejectsTrimmedKeyCollisions(t *testing.T) {
}
}
func TestFileConfigParsesOrderedStepsAndReferenceSources(t *testing.T) {
file := parseFileConfig(t, `version: 3
pipelines:
session:
input: seriatim
steps:
- id: identify-npcs
artifacts:
npcs:
extract: dnd/npcs
- id: grounded-events
references:
npcs:
artifact:
step: identify-npcs
lane: npcs
artifacts:
spells:
extract: dnd/spells
`)
cfg := Default()
if err := cfg.ApplyFileConfig(file); err != nil {
t.Fatal(err)
}
profile := cfg.Pipelines["session"]
if len(profile.Steps) != 2 || profile.Steps[0].ID != "identify-npcs" || profile.Steps[1].ID != "grounded-events" {
t.Fatalf("steps = %#v", profile.Steps)
}
source := profile.Steps[1].References["npcs"]
if source.Artifact == nil || source.Artifact.Step != "identify-npcs" || source.Artifact.Lane != "npcs" {
t.Fatalf("generated source = %#v", source)
}
}
func TestFileConfigRejectsAmbiguousReferenceSourceForms(t *testing.T) {
for _, source := range []string{
"artifact: {step: a, lane: b, extra: c}",
"artifact: {step: 1, lane: b}",
"1",
} {
_, err := ParseFileConfigYAML([]byte("version: 3\npipelines:\n p:\n input: text\n references:\n slot: " + source + "\n"))
if err == nil {
t.Fatalf("ParseFileConfigYAML(%q) error = nil", source)
}
}
}
func TestFileConfigRejectsEmptyAndAmbiguousPipelineShapes(t *testing.T) {
tests := []struct {
name string
yaml string
want string
}{
{
name: "empty steps",
yaml: "version: 3\npipelines:\n p:\n input: text\n steps: []\n",
want: "at least one ordered step",
},
{
name: "both forms",
yaml: "version: 3\npipelines:\n p:\n input: text\n artifacts: {}\n steps: []\n",
want: "both artifacts and steps",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
file := parseFileConfig(t, tt.yaml)
cfg := Default()
err := cfg.ApplyFileConfig(file)
if err == nil || !strings.Contains(err.Error(), tt.want) {
t.Fatalf("ApplyFileConfig() error = %v, want context %q", err, tt.want)
}
})
}
}
func TestLoadFileConfigReportsPathAndOperationContext(t *testing.T) {
dir := t.TempDir()
missing := filepath.Join(dir, "missing.yml")

34
internal/core/config/redaction.go
View File

@@ -42,10 +42,16 @@ func cloneResolvedPipeline(in pipeline.ResolvedPipeline) pipeline.ResolvedPipeli
out.ValidatorChains[i] = cloneResolvedValidatorChain(chain)
}
}
if len(in.ArtifactLanes) > 0 {
out.ArtifactLanes = make([]pipeline.ResolvedArtifactLane, len(in.ArtifactLanes))
for i, lane := range in.ArtifactLanes {
out.ArtifactLanes[i] = cloneResolvedArtifactLane(lane)
if len(in.Steps) > 0 {
out.Steps = make([]pipeline.ResolvedPipelineStep, len(in.Steps))
for i, step := range in.Steps {
out.Steps[i] = pipeline.ResolvedPipelineStep{ID: step.ID}
if len(step.ArtifactLanes) > 0 {
out.Steps[i].ArtifactLanes = make([]pipeline.ResolvedArtifactLane, len(step.ArtifactLanes))
for j, lane := range step.ArtifactLanes {
out.Steps[i].ArtifactLanes[j] = cloneResolvedArtifactLane(lane)
}
}
}
}
return out
@@ -89,14 +95,20 @@ func redactConfig(cfg Config) Config {
profile.Input = redactBinding(profile.Input)
profile.Chunk = redactBinding(profile.Chunk)
profile.Output = redactBinding(profile.Output)
for laneID, lane := range profile.Artifacts {
lane.Extract = redactBinding(lane.Extract)
lane.Merge = redactBinding(lane.Merge)
lane.Normalize = redactBinding(lane.Normalize)
for i := range lane.Validators {
lane.Validators[i] = redactBinding(lane.Validators[i])
redactLanes := func(lanes map[string]pipeline.ArtifactLaneProfile) {
for laneID, lane := range lanes {
lane.Extract = redactBinding(lane.Extract)
lane.Merge = redactBinding(lane.Merge)
lane.Normalize = redactBinding(lane.Normalize)
for i := range lane.Validators {
lane.Validators[i] = redactBinding(lane.Validators[i])
}
lanes[laneID] = lane
}
profile.Artifacts[laneID] = lane
}
redactLanes(profile.Artifacts)
for i := range profile.Steps {
redactLanes(profile.Steps[i].Artifacts)
}
cfg.Pipelines[id] = profile
}

38
internal/core/config/redaction_test.go
View File

@@ -24,16 +24,19 @@ func TestRedactedResolvedPipelinePayloadRedactsEveryBinding(t *testing.T) {
Input: bindings["input"],
Chunk: bindings["chunk"],
ChunkReferences: redactionTestReferenceTarget(pipeline.StageChunk, "", "chunk-reference-content"),
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
ID: "safe-lane",
ArtifactKind: "safe/artifact",
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
ExtractReferences: redactionTestReferenceTarget(pipeline.StageExtract, "safe-lane", "extract-reference-content"),
MergeReferences: redactionTestReferenceTarget(pipeline.StageMerge, "safe-lane", "merge-reference-content"),
NormalizeReferences: redactionTestReferenceTarget(pipeline.StageNormalize, "safe-lane", "normalize-reference-content"),
Steps: []pipeline.ResolvedPipelineStep{{
ID: "default",
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
ID: "safe-lane",
ArtifactKind: "safe/artifact",
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
ExtractReferences: redactionTestReferenceTarget(pipeline.StageExtract, "safe-lane", "extract-reference-content"),
MergeReferences: redactionTestReferenceTarget(pipeline.StageMerge, "safe-lane", "merge-reference-content"),
NormalizeReferences: redactionTestReferenceTarget(pipeline.StageNormalize, "safe-lane", "normalize-reference-content"),
}},
}},
ValidatorChains: []pipeline.ResolvedValidatorChain{{
Stage: pipeline.StageExtract,
@@ -142,12 +145,15 @@ func TestRedactedSummaryPayloadsCoverEveryEffectiveConfigBinding(t *testing.T) {
Input: bindings["input"],
Chunk: bindings["chunk"],
Output: bindings["output"],
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
ID: "safe-lane",
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
Steps: []pipeline.ResolvedPipelineStep{{
ID: "default",
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
ID: "safe-lane",
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
}},
}},
},
}

96
internal/core/config/validation.go
View File

@@ -106,30 +106,60 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro
if err := validateReferenceMap(id, "", profile.References); err != nil {
return err
}
seenLanes := make(map[string]struct{}, len(profile.Artifacts))
for rawLaneID, lane := range profile.Artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return fmt.Errorf("pipeline %q artifact lane id must not be empty", id)
explicitSteps := profile.Steps != nil
if len(profile.Artifacts) > 0 && explicitSteps {
return fmt.Errorf("pipeline %q must not declare both artifacts and steps", id)
}
steps := profile.Steps
if !explicitSteps {
steps = []pipeline.PipelineStepProfile{{ID: "default", Artifacts: profile.Artifacts}}
}
if explicitSteps && len(steps) == 0 {
return fmt.Errorf("pipeline %q must declare at least one ordered step", id)
}
seenSteps := make(map[string]struct{}, len(steps))
seenLanes := make(map[string]struct{})
for index, step := range steps {
stepID := strings.TrimSpace(step.ID)
if stepID == "" {
return fmt.Errorf("pipeline %q step[%d] id must not be empty", id, index)
}
if _, ok := seenLanes[laneID]; ok {
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated after trimming", id, laneID)
if _, ok := seenSteps[stepID]; ok {
return fmt.Errorf("pipeline %q step id %q is duplicated after trimming", id, stepID)
}
seenLanes[laneID] = struct{}{}
if err := validateReferenceMap(id, laneID, lane.References); err != nil {
return err
seenSteps[stepID] = struct{}{}
if explicitSteps {
if err := validateReferenceMapForContext(id, "", "step "+stepID, step.References, true); err != nil {
return err
}
}
if err := validateBinding(id, laneID, "extract", lane.Extract, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "merge", lane.Merge, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "normalize", lane.Normalize, true); err != nil {
return err
}
if len(lane.Validators) > 0 {
return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", id, laneID)
for rawLaneID, lane := range step.Artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return fmt.Errorf("pipeline %q artifact lane id must not be empty", id)
}
if _, ok := seenLanes[laneID]; ok {
if !explicitSteps {
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated after trimming", id, laneID)
}
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps", id, laneID)
}
seenLanes[laneID] = struct{}{}
if err := validateReferenceMapForContext(id, laneID, "", lane.References, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "extract", lane.Extract, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "merge", lane.Merge, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "normalize", lane.Normalize, true); err != nil {
return err
}
if len(lane.Validators) > 0 {
return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", id, laneID)
}
}
}
}
@@ -164,7 +194,7 @@ func validateBinding(
}
return fmt.Errorf("pipeline %q %s references are not supported", pipelineID, slot)
}
return validateReferenceMapForContext(pipelineID, laneID, slot, binding.References)
return validateReferenceMapForContext(pipelineID, laneID, slot, binding.References, true)
}
func validateValidatorOverride(pipelineID string, laneID string, slot string, override pipeline.ValidatorOverride) error {
@@ -197,13 +227,13 @@ func validateValidatorOverride(pipelineID string, laneID string, slot string, ov
return nil
}
func validateReferenceMap(pipelineID string, laneID string, references map[string]string) error {
return validateReferenceMapForContext(pipelineID, laneID, "", references)
func validateReferenceMap(pipelineID string, laneID string, references map[string]pipeline.ReferenceSource) error {
return validateReferenceMapForContext(pipelineID, laneID, "", references, false)
}
func validateReferenceMapForContext(pipelineID string, laneID string, slot string, references map[string]string) error {
func validateReferenceMapForContext(pipelineID string, laneID string, slot string, references map[string]pipeline.ReferenceSource, generatedAllowed bool) error {
seen := make(map[string]struct{}, len(references))
for rawSlotName, rawSource := range references {
for rawSlotName, source := range references {
slotName := strings.TrimSpace(rawSlotName)
if slotName == "" {
return fmt.Errorf("%s reference slot name must not be empty", referenceContext(pipelineID, laneID, slot))
@@ -212,7 +242,19 @@ func validateReferenceMapForContext(pipelineID string, laneID string, slot strin
return fmt.Errorf("%s reference slot %q is duplicated after trimming", referenceContext(pipelineID, laneID, slot), slotName)
}
seen[slotName] = struct{}{}
if strings.TrimSpace(rawSource) == "" {
if source.Artifact != nil {
if !generatedAllowed {
return fmt.Errorf("%s reference slot %q must use an external path", referenceContext(pipelineID, laneID, slot), slotName)
}
if strings.TrimSpace(source.Artifact.Step) == "" || strings.TrimSpace(source.Artifact.Lane) == "" {
return fmt.Errorf("%s reference slot %q artifact selector step and lane must not be empty", referenceContext(pipelineID, laneID, slot), slotName)
}
if strings.TrimSpace(source.Path) != "" {
return fmt.Errorf("%s reference slot %q must contain exactly one source form", referenceContext(pipelineID, laneID, slot), slotName)
}
continue
}
if strings.TrimSpace(source.Path) == "" {
return fmt.Errorf("%s reference slot %q source must not be empty", referenceContext(pipelineID, laneID, slot), slotName)
}
}

10
internal/core/config/validation_contract_test.go
View File

@@ -200,7 +200,7 @@ func TestValidateIdentifiersAfterTrimming(t *testing.T) {
name: "empty reference slot",
setup: func(cfg *Config) {
profile := validationProfile()
profile.References = map[string]string{" ": "source.txt"}
profile.References = pipeline.ExternalReferenceMap(map[string]string{" ": "source.txt"})
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "reference slot name must not be empty",
@@ -209,7 +209,7 @@ func TestValidateIdentifiersAfterTrimming(t *testing.T) {
name: "duplicate reference slots",
setup: func(cfg *Config) {
profile := validationProfile()
profile.References = map[string]string{"slot": "one.txt", " slot ": "two.txt"}
profile.References = pipeline.ExternalReferenceMap(map[string]string{"slot": "one.txt", " slot ": "two.txt"})
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "reference slot \"slot\" is duplicated after trimming",
@@ -267,14 +267,14 @@ func TestValidateReferencesAreUnsupportedOnInputAndOutput(t *testing.T) {
{
name: "input references",
set: func(profile *pipeline.PipelineProfile) {
profile.Input.References = map[string]string{"slot": "source.txt"}
profile.Input.References = pipeline.ExternalReferenceMap(map[string]string{"slot": "source.txt"})
},
want: "input references are not supported",
},
{
name: "output references",
set: func(profile *pipeline.PipelineProfile) {
profile.Output.References = map[string]string{"slot": "source.txt"}
profile.Output.References = pipeline.ExternalReferenceMap(map[string]string{"slot": "source.txt"})
},
want: "output references are not supported",
},
@@ -326,7 +326,7 @@ func TestValidateValidatorBindingRules(t *testing.T) {
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "validator",
References: map[string]string{"slot": "source.txt"},
References: pipeline.ExternalReferenceMap(map[string]string{"slot": "source.txt"}),
}},
}
},

2
internal/framework/checkpoint/identity.go
View File

@@ -57,7 +57,7 @@ func NewIdentity(input IdentityInput) (Identity, error) {
if strings.TrimSpace(input.RawInputDigest) == "" && strings.TrimSpace(input.SourceDigest) == "" {
return Identity{}, fmt.Errorf("checkpoint identity raw input digest or source digest must be set")
}
v := Identity{PipelineID: pipelineID, PipelineDigest: pipelineDigest, InputKey: inputKey, RawInputDigest: strings.TrimSpace(input.RawInputDigest), SourceDigest: strings.TrimSpace(input.SourceDigest), SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.ArtifactLanes), RuntimeOverrides: normalizeIdentityFingerprints(input.RuntimeOverrides), ReferenceDigests: referenceFingerprints(input.References), ProvenanceFingerprints: normalizeIdentityFingerprints(input.ProvenanceFingerprints)}
v := Identity{PipelineID: pipelineID, PipelineDigest: pipelineDigest, InputKey: inputKey, RawInputDigest: strings.TrimSpace(input.RawInputDigest), SourceDigest: strings.TrimSpace(input.SourceDigest), SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.AllArtifactLanes()), RuntimeOverrides: normalizeIdentityFingerprints(input.RuntimeOverrides), ReferenceDigests: referenceFingerprints(input.References), ProvenanceFingerprints: normalizeIdentityFingerprints(input.ProvenanceFingerprints)}
data, err := json.Marshal(Identity{PipelineID: v.PipelineID, PipelineDigest: v.PipelineDigest, InputKey: v.InputKey, RawInputDigest: v.RawInputDigest, SourceDigest: v.SourceDigest, SelectedLanes: v.SelectedLanes, RuntimeOverrides: v.RuntimeOverrides, ReferenceDigests: v.ReferenceDigests, ProvenanceFingerprints: v.ProvenanceFingerprints})
if err != nil {
return Identity{}, fmt.Errorf("marshal checkpoint identity: %w", err)

15
internal/framework/checkpoint/identity_test.go
View File

@@ -32,7 +32,7 @@ func TestNewIdentityNormalizesOrderAndEmptyValues(t *testing.T) {
v.ProvenanceFingerprints = []Fingerprint{{Name: "source", Value: "v2"}, {Name: "runner", Value: "v1"}}
}},
{"resolved lanes", func(v *IdentityInput) {
v.Pipeline.ArtifactLanes = []pipeline.ResolvedArtifactLane{v.Pipeline.ArtifactLanes[1], v.Pipeline.ArtifactLanes[0]}
v.Pipeline.Steps[0].ArtifactLanes = []pipeline.ResolvedArtifactLane{v.Pipeline.Steps[0].ArtifactLanes[1], v.Pipeline.Steps[0].ArtifactLanes[0]}
}},
} {
t.Run(tt.name, func(t *testing.T) {
@@ -174,10 +174,13 @@ func representativeIdentityInput() IdentityInput {
ID: "pipeline",
Digest: "sha256:pipeline-digest-000000000000",
Input: pipeline.Binding("input"),
ArtifactLanes: []pipeline.ResolvedArtifactLane{
{ID: "lane-b"},
{ID: "lane-a"},
},
Steps: []pipeline.ResolvedPipelineStep{{
ID: "default",
ArtifactLanes: []pipeline.ResolvedArtifactLane{
{ID: "lane-b"},
{ID: "lane-a"},
},
}},
},
InputKey: "input",
RawInputDigest: "sha256:raw-input-000000000000",
@@ -197,6 +200,6 @@ func cloneIdentityInput(input IdentityInput) IdentityInput {
input.RuntimeOverrides = append([]Fingerprint(nil), input.RuntimeOverrides...)
input.References = append([]artifacts.ReferenceProvenance(nil), input.References...)
input.ProvenanceFingerprints = append([]Fingerprint(nil), input.ProvenanceFingerprints...)
input.Pipeline.ArtifactLanes = append([]pipeline.ResolvedArtifactLane(nil), input.Pipeline.ArtifactLanes...)
input.Pipeline.Steps[0].ArtifactLanes = append([]pipeline.ResolvedArtifactLane(nil), input.Pipeline.Steps[0].ArtifactLanes...)
return input
}

14
internal/framework/contracts/contracts.go
View File

@@ -163,12 +163,13 @@ const (
)
type ReferenceSlot struct {
Name string `json:"name"`
Description string `json:"description,omitempty"`
Required bool `json:"required,omitempty"`
AcceptedMediaTypes []string `json:"accepted_media_types,omitempty"`
Multiple bool `json:"multiple,omitempty"`
MaxBytes int64 `json:"max_bytes,omitempty"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Required bool `json:"required,omitempty"`
AcceptedMediaTypes []string `json:"accepted_media_types,omitempty"`
AcceptedArtifactKinds []ArtifactKind `json:"accepted_artifact_kinds,omitempty"`
Multiple bool `json:"multiple,omitempty"`
MaxBytes int64 `json:"max_bytes,omitempty"`
}
func CloneReferenceSlots(slots []ReferenceSlot) []ReferenceSlot {
@@ -178,6 +179,7 @@ func CloneReferenceSlots(slots []ReferenceSlot) []ReferenceSlot {
out := make([]ReferenceSlot, len(slots))
for i, slot := range slots {
slot.AcceptedMediaTypes = append([]string(nil), slot.AcceptedMediaTypes...)
slot.AcceptedArtifactKinds = append([]ArtifactKind(nil), slot.AcceptedArtifactKinds...)
out[i] = slot
}
return out

9
internal/framework/contracts/contracts_test.go
View File

@@ -70,6 +70,15 @@ func TestCloneReferenceSlotsCopiesAcceptedMediaTypes(t *testing.T) {
}
}
func TestCloneReferenceSlotsCopiesAcceptedArtifactKinds(t *testing.T) {
slots := []ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []ArtifactKind{"dnd/npc-list"}}}
clone := CloneReferenceSlots(slots)
clone[0].AcceptedArtifactKinds[0] = "changed"
if slots[0].AcceptedArtifactKinds[0] != "dnd/npc-list" {
t.Fatalf("source AcceptedArtifactKinds aliased clone: %#v", slots[0].AcceptedArtifactKinds)
}
}
func TestReferenceItemJSONOmitsContent(t *testing.T) {
item := ReferenceItem{
SlotName: "roster",

34
internal/framework/pipeline/module.go
View File

@@ -133,6 +133,7 @@ func normalizeReferenceSlots(slots []contracts.ReferenceSlot) []contracts.Refere
slot.Name = strings.TrimSpace(slot.Name)
slot.Description = strings.TrimSpace(slot.Description)
slot.AcceptedMediaTypes = normalizeStringSet(slot.AcceptedMediaTypes)
slot.AcceptedArtifactKinds = normalizeArtifactKinds(slot.AcceptedArtifactKinds)
normalized = append(normalized, slot)
}
sort.SliceStable(normalized, func(i, j int) bool {
@@ -141,6 +142,34 @@ func normalizeReferenceSlots(slots []contracts.ReferenceSlot) []contracts.Refere
return normalized
}
func normalizeArtifactKinds(values []contracts.ArtifactKind) []contracts.ArtifactKind {
if len(values) == 0 {
return nil
}
seen := make(map[contracts.ArtifactKind]struct{}, len(values))
containsEmpty := false
for _, value := range values {
value = normalizeArtifactKind(value)
if value != "" {
seen[value] = struct{}{}
} else {
containsEmpty = true
}
}
if len(seen) == 0 && !containsEmpty {
return nil
}
result := make([]contracts.ArtifactKind, 0, len(seen))
for value := range seen {
result = append(result, value)
}
if containsEmpty {
result = append(result, "")
}
sort.Slice(result, func(i, j int) bool { return result[i] < result[j] })
return result
}
func normalizeStringSet(values []string) []string {
if len(values) == 0 {
return nil
@@ -178,6 +207,11 @@ func validateReferenceSlots(slots []contracts.ReferenceSlot) error {
if slot.MaxBytes < 0 {
return fmt.Errorf("slot %q max_bytes must not be negative", slot.Name)
}
for _, kind := range slot.AcceptedArtifactKinds {
if normalizeArtifactKind(kind) == "" {
return fmt.Errorf("slot %q accepted artifact kind must not be empty", slot.Name)
}
}
}
return nil
}

11
internal/framework/pipeline/options.go
View File

@@ -13,7 +13,7 @@ func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) e
return err
}
for _, lane := range resolved.ArtifactLanes {
validateLane := func(lane ResolvedArtifactLane) error {
if err := catalog.Extractors.validateOptions(lane.Extract.Module, lane.Extract.Options); err != nil {
return moduleOptionsError(resolved.ID, lane.ID, StageExtract, lane.Extract.Module, err)
}
@@ -29,8 +29,13 @@ func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) e
if err := catalog.Normalizers.validateOptions(lane.Normalize.Module, lane.ArtifactKind, lane.Normalize.Options); err != nil {
return moduleOptionsError(resolved.ID, lane.ID, StageNormalize, lane.Normalize.Module, err)
}
if err := validateChainOptions(resolved, catalog, StageNormalize, lane.ID, lane.Normalize.Module); err != nil {
return err
return validateChainOptions(resolved, catalog, StageNormalize, lane.ID, lane.Normalize.Module)
}
for _, step := range resolved.Steps {
for _, lane := range step.ArtifactLanes {
if err := validateLane(lane); err != nil {
return err
}
}
}

10
internal/framework/pipeline/preparation_test.go
View File

@@ -254,9 +254,9 @@ func TestPrepareDeliversTargetReferencesAsIndependentBuildInputs(t *testing.T) {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
resolved.ChunkReferences.ReferenceSet = constructionReferenceSet("chunk", "chunk reference")
resolved.ArtifactLanes[0].ExtractReferences.ReferenceSet = constructionReferenceSet("extract", "extract reference")
resolved.ArtifactLanes[0].MergeReferences.ReferenceSet = constructionReferenceSet("merge", "merge reference")
resolved.ArtifactLanes[0].NormalizeReferences.ReferenceSet = constructionReferenceSet("normalize", "normalize reference")
resolved.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet = constructionReferenceSet("extract", "extract reference")
resolved.Steps[0].ArtifactLanes[0].MergeReferences.ReferenceSet = constructionReferenceSet("merge", "merge reference")
resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet = constructionReferenceSet("normalize", "normalize reference")
prepared, err := Prepare(resolved, registries, ModuleDependencies{})
if err != nil {
@@ -275,12 +275,12 @@ func TestPrepareDeliversTargetReferencesAsIndependentBuildInputs(t *testing.T) {
t.Errorf("build request %d (%s) reference content = %q, want %q", i, observations[i].Name, got, want)
}
}
if got := constructionReferenceContent(resolved.ArtifactLanes[0].ExtractReferences.ReferenceSet); got != "extract reference" {
if got := constructionReferenceContent(resolved.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet); got != "extract reference" {
t.Fatalf("resolved extract references = %q, want original content", got)
}
_, err = prepared.lanes[0].typed.extract(context.Background(), prepared.lanes[0].typed.extractor, contracts.TypedExtractionRequest{
References: CloneReferenceSet(resolved.ArtifactLanes[0].ExtractReferences.ReferenceSet),
References: CloneReferenceSet(resolved.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet),
})
if err != nil {
t.Fatalf("prepared extractor operation error = %v, want nil", err)

24
internal/framework/pipeline/prepare.go
View File

@@ -68,6 +68,9 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
if err := validateResolvedPipeline(resolved); err != nil {
return nil, err
}
if len(resolved.Steps) > 1 {
return nil, fmt.Errorf("pipeline %q contains multiple ordered steps; execution support is not available yet", resolved.ID)
}
if err := validateRegistrySet(resolved, registries); err != nil {
return nil, err
}
@@ -99,9 +102,10 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
return nil, err
}
prepared.ArtifactLanes = make([]PreparedArtifactLane, 0, len(stable.ArtifactLanes))
prepared.lanes = make([]preparedLaneExecutor, 0, len(stable.ArtifactLanes))
for _, lane := range stable.ArtifactLanes {
lanes := stable.AllArtifactLanes()
prepared.ArtifactLanes = make([]PreparedArtifactLane, 0, len(lanes))
prepared.lanes = make([]preparedLaneExecutor, 0, len(lanes))
for _, lane := range lanes {
executor, err := prepareLane(stable, lane, registries, deps)
if err != nil {
return nil, err
@@ -331,10 +335,16 @@ func cloneResolvedPipeline(in ResolvedPipeline) ResolvedPipeline {
out.Output = cloneModuleBinding(in.Output)
out.ChunkReferences = CloneReferenceTarget(in.ChunkReferences)
out.ValidatorChains = cloneResolvedValidatorChains(in.ValidatorChains)
if len(in.ArtifactLanes) > 0 {
out.ArtifactLanes = make([]ResolvedArtifactLane, len(in.ArtifactLanes))
for i, lane := range in.ArtifactLanes {
out.ArtifactLanes[i] = cloneResolvedArtifactLane(lane)
if len(in.Steps) > 0 {
out.Steps = make([]ResolvedPipelineStep, len(in.Steps))
for i, step := range in.Steps {
out.Steps[i] = ResolvedPipelineStep{ID: step.ID}
if len(step.ArtifactLanes) > 0 {
out.Steps[i].ArtifactLanes = make([]ResolvedArtifactLane, len(step.ArtifactLanes))
for j, lane := range step.ArtifactLanes {
out.Steps[i].ArtifactLanes[j] = cloneResolvedArtifactLane(lane)
}
}
}
}
return out

381
internal/framework/pipeline/profile.go
View File

@@ -21,12 +21,56 @@ const (
)
type ModuleBinding struct {
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Retries int `json:"retries,omitempty"`
Options map[string]any `json:"options,omitempty"`
References map[string]string `json:"references,omitempty"`
Validators ValidatorOverride `json:"validators,omitempty"`
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Retries int `json:"retries,omitempty"`
Options map[string]any `json:"options,omitempty"`
References map[string]ReferenceSource `json:"references,omitempty"`
Validators ValidatorOverride `json:"validators,omitempty"`
}
// ArtifactReference identifies a normalized artifact produced by an earlier
// ordered step. It is a selector only; it does not contain artifact bytes.
type ArtifactReference struct {
Step string `json:"step"`
Lane string `json:"lane"`
}
// ReferenceSource is the discriminated source form used by configured
// reference maps. Exactly one of Path and Artifact is set after resolution.
type ReferenceSource struct {
Path string `json:"path,omitempty"`
Artifact *ArtifactReference `json:"artifact,omitempty"`
}
func ExternalReference(path string) ReferenceSource {
return ReferenceSource{Path: strings.TrimSpace(path)}
}
func ExternalReferenceMap(values map[string]string) map[string]ReferenceSource {
out := make(map[string]ReferenceSource, len(values))
for key, value := range values {
out[key] = ExternalReference(value)
}
return out
}
func GeneratedReference(step, lane string) ReferenceSource {
return ReferenceSource{Artifact: &ArtifactReference{Step: strings.TrimSpace(step), Lane: strings.TrimSpace(lane)}}
}
func (source ReferenceSource) IsGenerated() bool { return source.Artifact != nil }
func (source ReferenceSource) MarshalJSON() ([]byte, error) {
if source.Path != "" && source.Artifact != nil {
return nil, fmt.Errorf("reference source must not contain both an external path and an artifact selector")
}
if source.Artifact != nil {
return json.Marshal(struct {
Artifact *ArtifactReference `json:"artifact"`
}{Artifact: source.Artifact})
}
return json.Marshal(source.Path)
}
type ValidatorOverride struct {
@@ -36,12 +80,12 @@ type ValidatorOverride struct {
func (binding ModuleBinding) MarshalJSON() ([]byte, error) {
type moduleBindingJSON struct {
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Retries int `json:"retries,omitempty"`
Options map[string]any `json:"options,omitempty"`
References map[string]string `json:"references,omitempty"`
Validators *[]ModuleBinding `json:"validators,omitempty"`
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Retries int `json:"retries,omitempty"`
Options map[string]any `json:"options,omitempty"`
References map[string]ReferenceSource `json:"references,omitempty"`
Validators *[]ModuleBinding `json:"validators,omitempty"`
}
out := moduleBindingJSON{
Module: binding.Module,
@@ -58,11 +102,17 @@ func (binding ModuleBinding) MarshalJSON() ([]byte, error) {
}
type ArtifactLaneProfile struct {
Extract ModuleBinding `json:"extract"`
Merge ModuleBinding `json:"merge,omitempty"`
Normalize ModuleBinding `json:"normalize,omitempty"`
Validators []ModuleBinding `json:"validators,omitempty"`
References map[string]string `json:"references,omitempty"`
Extract ModuleBinding `json:"extract"`
Merge ModuleBinding `json:"merge,omitempty"`
Normalize ModuleBinding `json:"normalize,omitempty"`
Validators []ModuleBinding `json:"validators,omitempty"`
References map[string]ReferenceSource `json:"references,omitempty"`
}
type PipelineStepProfile struct {
ID string `json:"id"`
Artifacts map[string]ArtifactLaneProfile `json:"artifacts"`
References map[string]ReferenceSource `json:"references,omitempty"`
}
type PipelineProfile struct {
@@ -70,8 +120,9 @@ type PipelineProfile struct {
Input ModuleBinding `json:"input"`
Chunk ModuleBinding `json:"chunk,omitempty"`
Artifacts map[string]ArtifactLaneProfile `json:"artifacts"`
Steps []PipelineStepProfile `json:"steps,omitempty"`
Output ModuleBinding `json:"output,omitempty"`
References map[string]string `json:"references,omitempty"`
References map[string]ReferenceSource `json:"references,omitempty"`
}
type ResolveOptions struct {
@@ -81,11 +132,12 @@ type ResolveOptions struct {
}
type ReferenceBinding struct {
Stage ModuleStage `json:"stage,omitempty"`
LaneID string `json:"lane_id,omitempty"`
SlotName string `json:"slot_name"`
Source string `json:"source"`
BindingSource string `json:"binding_source,omitempty"`
Stage ModuleStage `json:"stage,omitempty"`
LaneID string `json:"lane_id,omitempty"`
SlotName string `json:"slot_name"`
Source string `json:"source,omitempty"`
Artifact *ArtifactReference `json:"artifact,omitempty"`
BindingSource string `json:"binding_source,omitempty"`
}
type ReferenceUnbind struct {
@@ -96,6 +148,7 @@ type ReferenceUnbind struct {
type ResolvedReferenceTarget struct {
Stage ModuleStage `json:"stage"`
StepID string `json:"step_id,omitempty"`
LaneID string `json:"lane_id,omitempty"`
Module string `json:"module"`
Bindings []ReferenceBinding `json:"bindings,omitempty"`
@@ -103,6 +156,7 @@ type ResolvedReferenceTarget struct {
}
type ResolvedArtifactLane struct {
StepID string
ID string
ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"`
ArtifactSchemaID string `json:"artifact_schema_id,omitempty"`
@@ -118,6 +172,11 @@ type ResolvedArtifactLane struct {
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
}
type ResolvedPipelineStep struct {
ID string `json:"id"`
ArtifactLanes []ResolvedArtifactLane `json:"artifact_lanes"`
}
type ResolvedValidatorChain struct {
Stage ModuleStage `json:"stage"`
LaneID string `json:"lane_id,omitempty"`
@@ -138,11 +197,21 @@ type ResolvedPipeline struct {
Input ModuleBinding
Chunk ModuleBinding
ChunkReferences ResolvedReferenceTarget `json:"chunk_references"`
ArtifactLanes []ResolvedArtifactLane
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain `json:"validator_chains"`
Output ModuleBinding
}
// AllArtifactLanes returns lanes in deterministic step order for read-only
// discovery and identity construction.
func (resolved ResolvedPipeline) AllArtifactLanes() []ResolvedArtifactLane {
var lanes []ResolvedArtifactLane
for _, step := range resolved.Steps {
lanes = append(lanes, step.ArtifactLanes...)
}
return lanes
}
type ModuleCatalog struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
@@ -165,7 +234,21 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
return ResolvedPipeline{}, fmt.Errorf("pipeline id must not be empty")
}
if len(profile.Artifacts) == 0 {
explicitSteps := profile.Steps != nil
if len(profile.Artifacts) > 0 && explicitSteps {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must not declare both artifacts and steps", pipelineID)
}
if explicitSteps && len(options.Only) > 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q --only is not supported for explicit ordered steps", pipelineID)
}
steps := profile.Steps
if !explicitSteps {
steps = []PipelineStepProfile{{ID: "default", Artifacts: profile.Artifacts}}
}
if explicitSteps && len(steps) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one ordered step", pipelineID)
}
if len(steps) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one artifact lane", pipelineID)
}
@@ -194,15 +277,25 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
}
capabilities.add(chunkSpec.Provides...)
lanesByID, selectedLaneIDs, err := selectedArtifactLanes(pipelineID, profile.Artifacts, options)
if err != nil {
return ResolvedPipeline{}, err
if !explicitSteps {
if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, profile.Artifacts, catalog); err != nil {
return ResolvedPipeline{}, err
}
} else {
allLanes := make(map[string]ArtifactLaneProfile)
for _, step := range profile.Steps {
for laneID, lane := range step.Artifacts {
allLanes[strings.TrimSpace(laneID)] = lane
}
}
if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, allLanes, catalog); err != nil {
return ResolvedPipeline{}, err
}
}
if len(selectedLaneIDs) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must select at least one artifact lane", pipelineID)
}
if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, lanesByID, catalog); err != nil {
return ResolvedPipeline{}, err
for _, source := range profile.References {
if source.Artifact != nil {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q pipeline-level references may not use generated artifact selectors", pipelineID)
}
}
chunkReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{
@@ -230,16 +323,55 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
}
resolved.ValidatorChains = append(resolved.ValidatorChains, chunkValidatorChain)
outputCapabilities := capabilities.clone()
seenResolvedLanes := make(map[string]string)
seenResolvedSteps := make(map[string]struct{}, len(steps))
for _, laneID := range selectedLaneIDs {
laneProfile := lanesByID[laneID]
lane, validatorChains, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, profile.References, options, capabilities, catalog)
for _, step := range steps {
stepID := strings.TrimSpace(step.ID)
if stepID == "" {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q step id must not be empty", pipelineID)
}
if _, exists := seenResolvedSteps[stepID]; exists {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q step id %q is duplicated after trimming", pipelineID, stepID)
}
seenResolvedSteps[stepID] = struct{}{}
laneOptions := ResolveOptions{}
if !explicitSteps {
laneOptions = options
}
lanesByID, selectedLaneIDs, err := selectedArtifactLanes(pipelineID, step.Artifacts, laneOptions)
if err != nil {
return ResolvedPipeline{}, err
}
resolved.ArtifactLanes = append(resolved.ArtifactLanes, lane)
resolved.ValidatorChains = append(resolved.ValidatorChains, validatorChains...)
outputCapabilities.addSet(laneCapabilities)
if len(selectedLaneIDs) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q step %q must declare at least one artifact lane", pipelineID, stepID)
}
if referenceSourceMapsConflict(profile.References, step.References) {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q step %q has a generated and external reference collision", pipelineID, stepID)
}
if err := validatePipelineReferenceDefaults(pipelineID, step.References, chunkSpec, lanesByID, catalog); err != nil {
return ResolvedPipeline{}, err
}
stepReferences := mergeReferenceMaps(profile.References, step.References)
resolvedStep := ResolvedPipelineStep{ID: stepID}
for _, laneID := range selectedLaneIDs {
if previousStep, exists := seenResolvedLanes[laneID]; exists {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps %q and %q", pipelineID, laneID, previousStep, stepID)
}
seenResolvedLanes[laneID] = stepID
laneProfile := lanesByID[laneID]
lane, validatorChains, laneCapabilities, err := resolveArtifactLane(pipelineID, stepID, laneID, laneProfile, stepReferences, options, capabilities, catalog)
if err != nil {
return ResolvedPipeline{}, err
}
resolvedStep.ArtifactLanes = append(resolvedStep.ArtifactLanes, lane)
resolved.ValidatorChains = append(resolved.ValidatorChains, validatorChains...)
outputCapabilities.addSet(laneCapabilities)
}
resolved.Steps = append(resolved.Steps, resolvedStep)
}
if err := validateGeneratedBindings(pipelineID, resolved, catalog); err != nil {
return ResolvedPipeline{}, err
}
outputSpec, err := outputSpec(catalog, resolved.Output.Module)
@@ -263,14 +395,16 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
func resolveArtifactLane(
pipelineID string,
stepID string,
laneID string,
profile ArtifactLaneProfile,
pipelineReferences map[string]string,
pipelineReferences map[string]ReferenceSource,
options ResolveOptions,
inherited capabilitySet,
catalog ModuleCatalog,
) (ResolvedArtifactLane, []ResolvedValidatorChain, capabilitySet, error) {
lane := ResolvedArtifactLane{
StepID: strings.TrimSpace(stepID),
ID: laneID,
Extract: resolveBinding(profile.Extract, ""),
Merge: resolveBinding(profile.Merge, DefaultMergeModule),
@@ -294,6 +428,9 @@ func resolveArtifactLane(
if err != nil {
return ResolvedArtifactLane{}, nil, nil, err
}
if referenceSourceMapsConflict(profile.References, lane.Extract.References) {
return ResolvedArtifactLane{}, nil, nil, fmt.Errorf("pipeline %q step %q lane %q extract has a generated and external reference collision", pipelineID, stepID, laneID)
}
extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References)
references, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
@@ -309,6 +446,7 @@ func resolveArtifactLane(
return ResolvedArtifactLane{}, nil, nil, err
}
lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references)
lane.ExtractReferences.StepID = strings.TrimSpace(stepID)
capabilities.add(extractSpec.Provides...)
mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType)
@@ -332,6 +470,7 @@ func resolveArtifactLane(
return ResolvedArtifactLane{}, nil, nil, err
}
lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences)
lane.MergeReferences.StepID = strings.TrimSpace(stepID)
capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType)
@@ -355,6 +494,7 @@ func resolveArtifactLane(
return ResolvedArtifactLane{}, nil, nil, err
}
lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences)
lane.NormalizeReferences.StepID = strings.TrimSpace(stepID)
capabilities.add(normalizeSpec.Provides...)
if len(lane.Validators) > 0 {
@@ -382,6 +522,83 @@ func configuredValidatorsError(pipelineID string, laneID string) error {
return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", pipelineID, laneID)
}
func validateGeneratedBindings(pipelineID string, resolved ResolvedPipeline, catalog ModuleCatalog) error {
stepIndex := make(map[string]int, len(resolved.Steps))
laneIndex := make(map[string]ResolvedArtifactLane)
for index, step := range resolved.Steps {
if _, exists := stepIndex[step.ID]; exists {
return fmt.Errorf("pipeline %q step id %q is ambiguous", pipelineID, step.ID)
}
stepIndex[step.ID] = index
for _, lane := range step.ArtifactLanes {
laneIndex[step.ID+"\x00"+lane.ID] = lane
}
}
for consumerIndex, step := range resolved.Steps {
for _, lane := range step.ArtifactLanes {
for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} {
slots, err := resolvedReferenceSlots(target, lane.ArtifactKind, catalog)
if err != nil {
return fmt.Errorf("pipeline %q step %q lane %q: %w", pipelineID, step.ID, lane.ID, err)
}
slotByName := make(map[string]contracts.ReferenceSlot, len(slots))
for _, slot := range slots {
slotByName[slot.Name] = slot
}
for _, binding := range target.Bindings {
if binding.Artifact == nil {
continue
}
producerStep, ok := stepIndex[strings.TrimSpace(binding.Artifact.Step)]
if !ok {
return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q producer step %q is not declared", pipelineID, step.ID, lane.ID, target.Stage, binding.SlotName, binding.Artifact.Step)
}
if producerStep >= consumerIndex {
return fmt.Errorf("pipeline %q step %q lane %q reference slot %q producer %q must be an earlier step", pipelineID, step.ID, lane.ID, binding.SlotName, binding.Artifact.Step)
}
producer, ok := laneIndex[strings.TrimSpace(binding.Artifact.Step)+"\x00"+strings.TrimSpace(binding.Artifact.Lane)]
if !ok {
return fmt.Errorf("pipeline %q step %q lane %q reference slot %q producer lane %q is not selected", pipelineID, step.ID, lane.ID, binding.SlotName, binding.Artifact.Lane)
}
slot, ok := slotByName[strings.TrimSpace(binding.SlotName)]
if !ok {
return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q is not declared", pipelineID, step.ID, lane.ID, target.Stage, binding.SlotName)
}
if len(slot.AcceptedArtifactKinds) == 0 {
return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept generated artifacts", pipelineID, step.ID, lane.ID, binding.SlotName)
}
accepted := false
for _, kind := range slot.AcceptedArtifactKinds {
if kind == producer.ArtifactKind {
accepted = true
break
}
}
if !accepted {
return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept artifact kind %q", pipelineID, step.ID, lane.ID, binding.SlotName, producer.ArtifactKind)
}
codecSpec, ok := catalog.ArtifactCodecs.Spec(producer.ArtifactKind)
if !ok {
return fmt.Errorf("pipeline %q step %q lane %q producer %q artifact codec %q is not registered", pipelineID, step.ID, lane.ID, producer.ID, producer.ArtifactKind)
}
if !referenceMediaTypeAccepted(codecSpec.MediaType, slot.AcceptedMediaTypes) {
return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept producer media type %q", pipelineID, step.ID, lane.ID, binding.SlotName, codecSpec.MediaType)
}
}
}
}
}
return nil
}
func resolvedReferenceSlots(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) ([]contracts.ReferenceSlot, error) {
spec, err := referenceTargetSpec(target, artifactKind, catalog)
if err != nil {
return nil, err
}
return contracts.CloneReferenceSlots(spec.ReferenceSlots), nil
}
func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLane, extractSpec ModuleSpec, catalog ModuleCatalog) (reflect.Type, error) {
if extractSpec.ArtifactKind == "" {
return nil, fmt.Errorf("pipeline %q lane %q extract module %q does not declare an artifact kind", pipelineID, laneID, lane.Extract.Module)
@@ -593,23 +810,35 @@ func referenceTarget(stage ModuleStage, laneID string, module string, bindings [
}
}
func mergeReferenceMaps(base map[string]string, override map[string]string) map[string]string {
func mergeReferenceMaps(base map[string]ReferenceSource, override map[string]ReferenceSource) map[string]ReferenceSource {
if len(base) == 0 && len(override) == 0 {
return nil
}
out := make(map[string]string, len(base)+len(override))
out := make(map[string]ReferenceSource, len(base)+len(override))
for key, value := range base {
out[key] = value
out[key] = cloneReferenceSource(value)
}
for key, value := range override {
out[key] = value
out[key] = cloneReferenceSource(value)
}
return out
}
func referenceSourceMapsConflict(base, override map[string]ReferenceSource) bool {
for rawKey, left := range base {
key := strings.TrimSpace(rawKey)
for rawOverrideKey, right := range override {
if key == strings.TrimSpace(rawOverrideKey) && left.IsGenerated() != right.IsGenerated() {
return true
}
}
}
return false
}
func validatePipelineReferenceDefaults(
pipelineID string,
pipelineReferences map[string]string,
pipelineReferences map[string]ReferenceSource,
chunkSpec ModuleSpec,
lanesByID map[string]ArtifactLaneProfile,
catalog ModuleCatalog,
@@ -680,8 +909,8 @@ type referenceResolutionTarget struct {
Stage ModuleStage
Module string
Slots []contracts.ReferenceSlot
PipelineReferences map[string]string
LocalReferences map[string]string
PipelineReferences map[string]ReferenceSource
LocalReferences map[string]ReferenceSource
Options ResolveOptions
}
@@ -692,24 +921,46 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen
}
bindings := make(map[string]ReferenceBinding)
addBinding := func(slotName, source, bindingSource string) error {
addBinding := func(slotName string, source ReferenceSource, bindingSource string) error {
slotName = strings.TrimSpace(slotName)
source = strings.TrimSpace(source)
if slotName == "" {
return fmt.Errorf("%s reference slot name must not be empty", referenceTargetErrorContext(target))
}
if source == "" {
if source.Artifact == nil {
source.Path = strings.TrimSpace(source.Path)
}
if source.Artifact == nil && source.Path == "" {
return fmt.Errorf("%s reference slot %q source must not be empty", referenceTargetErrorContext(target), slotName)
}
if source.Artifact != nil {
if strings.TrimSpace(source.Path) != "" {
return fmt.Errorf("%s reference slot %q must contain exactly one source form", referenceTargetErrorContext(target), slotName)
}
artifact := *source.Artifact
artifact.Step = strings.TrimSpace(artifact.Step)
artifact.Lane = strings.TrimSpace(artifact.Lane)
if artifact.Step == "" || artifact.Lane == "" {
return fmt.Errorf("%s reference slot %q artifact selector step and lane must not be empty", referenceTargetErrorContext(target), slotName)
}
source.Artifact = &artifact
}
if _, ok := slotByName[slotName]; !ok {
return fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetErrorContext(target), slotName, target.Stage, target.Module)
}
bindings[slotName] = ReferenceBinding{
if previous, exists := bindings[slotName]; exists && (previous.Artifact != nil || source.Artifact != nil) {
return fmt.Errorf("%s reference slot %q has conflicting generated and external bindings", referenceTargetErrorContext(target), slotName)
}
binding := ReferenceBinding{
LaneID: target.LaneID,
SlotName: slotName,
Source: source,
BindingSource: bindingSource,
}
if source.Artifact != nil {
binding.Artifact = source.Artifact
} else {
binding.Source = source.Path
}
bindings[slotName] = binding
return nil
}
@@ -748,7 +999,7 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen
if strings.TrimSpace(source) == "" {
source = contracts.ReferenceBindingSourceCLI
}
if err := addBinding(override.SlotName, override.Source, strings.TrimSpace(source)); err != nil {
if err := addBinding(override.SlotName, ExternalReference(override.Source), strings.TrimSpace(source)); err != nil {
return nil, err
}
}
@@ -768,7 +1019,9 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen
if _, ok := slotByName[slotName]; !ok {
return nil, fmt.Errorf("%s reference slot %q is not declared", referenceTargetErrorContext(target), slotName)
}
delete(bindings, slotName)
if binding, ok := bindings[slotName]; ok && binding.Artifact == nil {
delete(bindings, slotName)
}
}
for _, slot := range target.Slots {
@@ -843,11 +1096,11 @@ func referenceTargetSlotLabel(target referenceResolutionTarget) string {
return fmt.Sprintf("pipeline %q %s reference slot", target.PipelineID, target.Stage)
}
func normalizedReferenceMap(values map[string]string, keyName string) (map[string]string, error) {
func normalizedReferenceMap(values map[string]ReferenceSource, keyName string) (map[string]ReferenceSource, error) {
if len(values) == 0 {
return nil, nil
}
out := make(map[string]string, len(values))
out := make(map[string]ReferenceSource, len(values))
for rawSlotName, rawSource := range values {
slotName := strings.TrimSpace(rawSlotName)
if slotName == "" {
@@ -856,8 +1109,8 @@ func normalizedReferenceMap(values map[string]string, keyName string) (map[strin
if _, ok := out[slotName]; ok {
return nil, fmt.Errorf("%s %q is duplicated after trimming", keyName, slotName)
}
source := strings.TrimSpace(rawSource)
if source == "" {
source := cloneReferenceSource(rawSource)
if source.Artifact == nil && strings.TrimSpace(source.Path) == "" {
return nil, fmt.Errorf("%s %q source must not be empty", keyName, slotName)
}
out[slotName] = source
@@ -877,7 +1130,7 @@ func sortedArtifactLaneProfileKeys(values map[string]ArtifactLaneProfile) []stri
return keys
}
func sortedStringMapKeys(values map[string]string) []string {
func sortedStringMapKeys(values map[string]ReferenceSource) []string {
if len(values) == 0 {
return nil
}
@@ -942,11 +1195,11 @@ func cloneOptions(options map[string]any) map[string]any {
return copied
}
func normalizeReferenceMap(values map[string]string) map[string]string {
func normalizeReferenceMap(values map[string]ReferenceSource) map[string]ReferenceSource {
if len(values) == 0 {
return nil
}
out := make(map[string]string, len(values))
out := make(map[string]ReferenceSource, len(values))
keys := make([]string, 0, len(values))
rawByNormalized := make(map[string]string, len(values))
for rawKey := range values {
@@ -956,7 +1209,7 @@ func normalizeReferenceMap(values map[string]string) map[string]string {
}
sort.Strings(keys)
for _, key := range keys {
out[key] = strings.TrimSpace(values[rawByNormalized[key]])
out[key] = cloneReferenceSource(values[rawByNormalized[key]])
}
return out
}
@@ -1009,7 +1262,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
Input ModuleBinding
Chunk ModuleBinding
ChunkReferences ResolvedReferenceTarget
ArtifactLanes []ResolvedArtifactLane
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain
Output ModuleBinding
}{
@@ -1017,7 +1270,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
Input: resolved.Input,
Chunk: resolved.Chunk,
ChunkReferences: resolved.ChunkReferences,
ArtifactLanes: resolved.ArtifactLanes,
Steps: resolved.Steps,
ValidatorChains: resolved.ValidatorChains,
Output: resolved.Output,
}

168
internal/framework/pipeline/profile_test.go
View File

@@ -53,10 +53,10 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
if resolved.Chunk.Options["size"] != 10 {
t.Fatalf("Chunk.Options = %#v, want size option", resolved.Chunk.Options)
}
if len(resolved.ArtifactLanes) != 1 {
t.Fatalf("len(ArtifactLanes) = %d, want 1", len(resolved.ArtifactLanes))
if len(resolved.Steps) != 1 || resolved.Steps[0].ID != "default" || len(resolved.Steps[0].ArtifactLanes) != 1 {
t.Fatalf("resolved steps = %#v, want one default step with one lane", resolved.Steps)
}
lane := resolved.ArtifactLanes[0]
lane := resolved.Steps[0].ArtifactLanes[0]
if lane.ID != "records" {
t.Fatalf("lane.ID = %q, want records", lane.ID)
}
@@ -98,7 +98,7 @@ func TestResolvePipelineAppliesDefaults(t *testing.T) {
if !reflect.DeepEqual(resolved.Output, ModuleBinding{Module: DefaultOutputModule}) {
t.Fatalf("Output = %#v, want default output binding", resolved.Output)
}
lane := resolved.ArtifactLanes[0]
lane := resolved.Steps[0].ArtifactLanes[0]
if !reflect.DeepEqual(lane.Merge, ModuleBinding{Module: DefaultMergeModule}) {
t.Fatalf("Merge = %#v, want default merge binding", lane.Merge)
}
@@ -326,23 +326,77 @@ func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
got := laneIDs(resolved.ArtifactLanes)
got := laneIDs(resolved.Steps[0].ArtifactLanes)
want := []string{"events", "summaries"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("lane IDs = %#v, want %#v", got, want)
}
}
func TestResolvePipelinePreservesOrderedStepsAndExpandsGeneratedBindings(t *testing.T) {
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{
Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes",
Requires: []string{"chunk"}, Provides: []string{"candidate"},
ReferenceSlots: []contracts.ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/notes"}, AcceptedMediaTypes: []string{"application/json"}}},
})
resolved, err := ResolvePipeline(PipelineProfile{
ID: "ordered",
Input: Binding("text"),
Steps: []PipelineStepProfile{
{ID: "produce", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}},
{ID: "consume", References: map[string]ReferenceSource{"npcs": GeneratedReference("produce", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}},
},
}, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if got := []string{resolved.Steps[0].ID, resolved.Steps[1].ID}; !reflect.DeepEqual(got, []string{"produce", "consume"}) {
t.Fatalf("step order = %#v", got)
}
binding := resolved.Steps[1].ArtifactLanes[0].ExtractReferences.Bindings[0]
if binding.Artifact == nil || binding.Artifact.Step != "produce" || binding.Artifact.Lane != "npcs" {
t.Fatalf("generated binding = %#v", binding)
}
if len(resolved.AllArtifactLanes()) != 2 || resolved.Steps[1].ArtifactLanes[0].StepID != "consume" {
t.Fatalf("resolved lane topology = %#v", resolved.Steps)
}
}
func TestResolvePipelineRejectsGeneratedBindingOrderingAndKind(t *testing.T) {
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/other"}}}},
)
_, err := ResolvePipeline(PipelineProfile{
ID: "invalid-order", Input: Binding("text"), Steps: []PipelineStepProfile{
{ID: "first", References: map[string]ReferenceSource{"npcs": GeneratedReference("later", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}},
{ID: "later", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}},
},
}, ResolveOptions{}, catalog)
if err == nil || !strings.Contains(err.Error(), "earlier step") {
t.Fatalf("ResolvePipeline() error = %v, want ordering failure", err)
}
_, err = ResolvePipeline(PipelineProfile{
ID: "invalid-kind", Input: Binding("text"), Steps: []PipelineStepProfile{
{ID: "produce", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}},
{ID: "consume", References: map[string]ReferenceSource{"npcs": GeneratedReference("produce", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}},
},
}, ResolveOptions{}, catalog)
if err == nil || !strings.Contains(err.Error(), "does not accept artifact kind") {
t.Fatalf("ResolvePipeline() error = %v, want generated kind failure", err)
}
}
func TestResolvePipelineAppliesReferenceBindings(t *testing.T) {
profile := multiLaneProfile()
profile.References = map[string]string{
profile.References = ExternalReferenceMap(map[string]string{
" roster ": " ./shared-roster.yml ",
}
})
lane := profile.Artifacts["events"]
lane.References = map[string]string{
lane.References = ExternalReferenceMap(map[string]string{
"roster": "./lane-roster.yml",
" lore ": " ./lore.md ",
}
})
profile.Artifacts["events"] = lane
catalog := newProfileCatalogWithOverride(t, ModuleSpec{
@@ -360,7 +414,7 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
events := resolvedLane(t, resolved.ArtifactLanes, "events")
events := resolvedLane(t, resolved.Steps[0].ArtifactLanes, "events")
if events.ExtractReferences.Stage != StageExtract || events.ExtractReferences.LaneID != "events" || events.ExtractReferences.Module != "event-extractor" {
t.Fatalf("extract reference target = %#v, want event extractor target", events.ExtractReferences)
}
@@ -380,7 +434,7 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) {
if !reflect.DeepEqual(events.ExtractReferences.Bindings, want) {
t.Fatalf("events references = %#v, want %#v", events.ExtractReferences.Bindings, want)
}
summaries := resolvedLane(t, resolved.ArtifactLanes, "summaries")
summaries := resolvedLane(t, resolved.Steps[0].ArtifactLanes, "summaries")
if len(summaries.ExtractReferences.Bindings) != 0 {
t.Fatalf("summaries references = %#v, want none", summaries.ExtractReferences.Bindings)
}
@@ -388,7 +442,7 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) {
func TestResolvePipelineAppliesPipelineReferenceDefaultToChunkTarget(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"scene_guide": "./scenes.md"}
profile.References = ExternalReferenceMap(map[string]string{"scene_guide": "./scenes.md"})
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "generic",
Stage: StageChunk,
@@ -406,17 +460,17 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToChunkTarget(t *testing.
if !reflect.DeepEqual(resolved.ChunkReferences.Bindings, want) {
t.Fatalf("chunk references = %#v, want %#v", resolved.ChunkReferences.Bindings, want)
}
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("extract references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("normalize references = %#v, want none", refs)
}
}
func TestResolvePipelineAppliesPipelineReferenceDefaultToExtractorTarget(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"roster": "./roster.yml"}
profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"})
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "event-extractor",
Stage: StageExtract,
@@ -431,20 +485,20 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToExtractorTarget(t *test
}
want := []ReferenceBinding{{LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceConfig}}
if !reflect.DeepEqual(resolved.ArtifactLanes[0].ExtractReferences.Bindings, want) {
t.Fatalf("extract references = %#v, want %#v", resolved.ArtifactLanes[0].ExtractReferences.Bindings, want)
if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, want) {
t.Fatalf("extract references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, want)
}
if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 {
t.Fatalf("chunk references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("normalize references = %#v, want none", refs)
}
}
func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"normalization_notes": "./normalize.md"}
profile.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"})
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "noop",
Stage: StageNormalize,
@@ -459,20 +513,20 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *tes
}
want := []ReferenceBinding{{LaneID: "events", SlotName: "normalization_notes", Source: "./normalize.md", BindingSource: contracts.ReferenceBindingSourceConfig}}
if !reflect.DeepEqual(resolved.ArtifactLanes[0].NormalizeReferences.Bindings, want) {
t.Fatalf("normalize references = %#v, want %#v", resolved.ArtifactLanes[0].NormalizeReferences.Bindings, want)
if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, want) {
t.Fatalf("normalize references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, want)
}
if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 {
t.Fatalf("chunk references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("extract references = %#v, want none", refs)
}
}
func TestResolvePipelineAppliesPipelineReferenceDefaultToMergeTarget(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"merge_notes": "./merge.md"}
profile.References = ExternalReferenceMap(map[string]string{"merge_notes": "./merge.md"})
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "appendorder",
Stage: StageMerge,
@@ -487,23 +541,23 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToMergeTarget(t *testing.
}
want := []ReferenceBinding{{LaneID: "events", SlotName: "merge_notes", Source: "./merge.md", BindingSource: contracts.ReferenceBindingSourceConfig}}
if !reflect.DeepEqual(resolved.ArtifactLanes[0].MergeReferences.Bindings, want) {
t.Fatalf("merge references = %#v, want %#v", resolved.ArtifactLanes[0].MergeReferences.Bindings, want)
if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, want) {
t.Fatalf("merge references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, want)
}
if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 {
t.Fatalf("chunk references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("extract references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("normalize references = %#v, want none", refs)
}
}
func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"context": "./context.md"}
profile.References = ExternalReferenceMap(map[string]string{"context": "./context.md"})
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}},
ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}},
@@ -517,14 +571,14 @@ func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *t
}
assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].ExtractReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].MergeReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].NormalizeReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, "context", "./context.md")
}
func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *testing.T) {
profile := multiLaneProfile()
profile.References = map[string]string{"notes_context": "./notes.md"}
profile.References = ExternalReferenceMap(map[string]string{"notes_context": "./notes.md"})
catalog := newProfileCatalogWithOverride(t, ModuleSpec{
Key: "note-extractor",
Stage: StageExtract,
@@ -539,14 +593,14 @@ func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *t
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("selected lane references = %#v, want none", refs)
}
}
func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedNormalizer(t *testing.T) {
profile := multiLaneProfile()
profile.References = map[string]string{"notes_context": "./notes.md"}
profile.References = ExternalReferenceMap(map[string]string{"notes_context": "./notes.md"})
lane := profile.Artifacts["notes"]
lane.Normalize = Binding("note-normalizer")
profile.Artifacts["notes"] = lane
@@ -564,17 +618,17 @@ func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedNormalize
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("selected extract references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("selected normalize references = %#v, want none", refs)
}
}
func TestResolvePipelineRejectsPipelineReferenceNotDeclaredByAnyLane(t *testing.T) {
profile := multiLaneProfile()
profile.References = map[string]string{"missing": "./missing.md"}
profile.References = ExternalReferenceMap(map[string]string{"missing": "./missing.md"})
_, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, newProfileCatalog(t))
if err == nil {
@@ -586,7 +640,7 @@ func TestResolvePipelineRejectsPipelineReferenceNotDeclaredByAnyLane(t *testing.
func TestResolvePipelineRejectsUndeclaredReferenceSlot(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.References = map[string]string{"missing": "./missing.yml"}
lane.References = ExternalReferenceMap(map[string]string{"missing": "./missing.yml"})
profile.Artifacts["events"] = lane
_, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t))
@@ -599,7 +653,7 @@ func TestResolvePipelineRejectsUndeclaredReferenceSlot(t *testing.T) {
func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{"normalization_notes": "./normalize.md"}
lane.Extract.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"})
profile.Artifacts["events"] = lane
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "noop",
@@ -619,7 +673,7 @@ func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t *
func TestResolvePipelineRejectsMergeLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Merge.References = map[string]string{"normalization_notes": "./normalize.md"}
lane.Merge.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"})
profile.Artifacts["events"] = lane
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "noop",
@@ -641,7 +695,7 @@ func TestResolvePipelineRejectsMergeLocalReferenceDeclaredOnlyByNormalizer(t *te
func TestResolvePipelineRejectsNormalizeLocalReferenceDeclaredOnlyByExtractor(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Normalize.References = map[string]string{"roster": "./roster.yml"}
lane.Normalize.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"})
profile.Artifacts["events"] = lane
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "event-extractor",
@@ -692,15 +746,15 @@ func TestResolvePipelineRequiresBoundNormalizeReference(t *testing.T) {
func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTargets(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{
profile.References = ExternalReferenceMap(map[string]string{
"context": "./shared-context.md",
"roster": "./shared-roster.yml",
"normalization_notes": "./shared-normalize.md",
}
profile.Chunk.References = map[string]string{"context": "./chunk-context.md"}
})
profile.Chunk.References = ExternalReferenceMap(map[string]string{"context": "./chunk-context.md"})
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{"roster": "./extract-roster.yml"}
lane.Normalize.References = map[string]string{"normalization_notes": "./local-normalize.md"}
lane.Extract.References = ExternalReferenceMap(map[string]string{"roster": "./extract-roster.yml"})
lane.Normalize.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./local-normalize.md"})
profile.Artifacts["events"] = lane
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}},
@@ -714,8 +768,8 @@ func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTarget
}
assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./chunk-context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].ExtractReferences.Bindings, "roster", "./extract-roster.yml")
assertBindingSource(t, resolved.ArtifactLanes[0].NormalizeReferences.Bindings, "normalization_notes", "./local-normalize.md")
assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, "roster", "./extract-roster.yml")
assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, "normalization_notes", "./local-normalize.md")
}
func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T) {
@@ -742,7 +796,7 @@ func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T
func TestResolvePipelineReferenceUnbindCanLeaveRequiredSlotMissing(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"roster": "./roster.yml"}
profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"})
catalog := newProfileCatalogWithOverride(t, ModuleSpec{
Key: "event-extractor",
Stage: StageExtract,
@@ -764,7 +818,7 @@ func TestResolvePipelineReferenceUnbindCanLeaveRequiredSlotMissing(t *testing.T)
func TestResolvePipelineUsesReferenceSlotsFromSpecWithoutConstructingExtractor(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"roster": "./roster.yml"}
profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"})
catalog := emptyProfileCatalog()
mustRegisterArtifactCodec(t, catalog.ArtifactCodecs, notesCodec())
for _, spec := range defaultProfileSpecs() {
@@ -791,7 +845,7 @@ func TestResolvePipelineUsesReferenceSlotsFromSpecWithoutConstructingExtractor(t
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if got := resolved.ArtifactLanes[0].ExtractReferences.Bindings[0].Source; got != "./roster.yml" {
if got := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings[0].Source; got != "./roster.yml" {
t.Fatalf("reference source = %q, want ./roster.yml", got)
}
}
@@ -998,7 +1052,7 @@ func TestResolvePipelineOrdersLanesDeterministically(t *testing.T) {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
got := laneIDs(resolved.ArtifactLanes)
got := laneIDs(resolved.Steps[0].ArtifactLanes)
want := []string{"events", "notes", "summaries"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("lane IDs = %#v, want %#v", got, want)
@@ -1162,7 +1216,7 @@ func TestResolvedPipelineDigestIncludesCompleteValidatorPolicy(t *testing.T) {
value.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.75
}},
{name: "references", mutate: func(value *ResolvedPipeline) {
value.ValidatorChains[0].Validators[0].Binding.References["rules"] = "other.md"
value.ValidatorChains[0].Validators[0].Binding.References["rules"] = ExternalReference("other.md")
}},
{name: "execution class", mutate: func(value *ResolvedPipeline) {
value.ValidatorChains[0].Validators[0].ExecutionClass = contracts.ExecutionClassDeterministic
@@ -1194,9 +1248,9 @@ func TestResolvedPipelineDigestCanonicalizesValidatorBindingMaps(t *testing.T) {
right.ValidatorChains[0].Validators[0].Binding.Options = map[string]any{}
right.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.5
right.ValidatorChains[0].Validators[0].Binding.Options["mode"] = "strict"
right.ValidatorChains[0].Validators[0].Binding.References = map[string]string{}
right.ValidatorChains[0].Validators[0].Binding.References["examples"] = "examples.md"
right.ValidatorChains[0].Validators[0].Binding.References["rules"] = "rules.md"
right.ValidatorChains[0].Validators[0].Binding.References = ExternalReferenceMap(map[string]string{})
right.ValidatorChains[0].Validators[0].Binding.References["examples"] = ExternalReference("examples.md")
right.ValidatorChains[0].Validators[0].Binding.References["rules"] = ExternalReference("rules.md")
leftDigest, err := resolvedPipelineDigest(left)
if err != nil {
@@ -1225,7 +1279,7 @@ func validatorDigestFixture() ResolvedPipeline {
LLMProfile: "careful",
Retries: 2,
Options: map[string]any{"mode": "strict", "threshold": 0.5},
References: map[string]string{"rules": "rules.md", "examples": "examples.md"},
References: ExternalReferenceMap(map[string]string{"rules": "rules.md", "examples": "examples.md"}),
},
ExecutionClass: contracts.ExecutionClassLLMBacked,
Target: ValidatorTargetTyped,

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

@@ -36,37 +36,51 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
}
out.ChunkReferences.ReferenceSet = chunkReferenceSet
warnings := append([]contracts.Warning(nil), chunkWarnings...)
if len(resolved.ArtifactLanes) == 0 {
if len(resolved.Steps) == 0 {
return out, warnings, nil
}
out.ArtifactLanes = make([]ResolvedArtifactLane, len(resolved.ArtifactLanes))
for i, lane := range resolved.ArtifactLanes {
materializeLane := func(lane ResolvedArtifactLane) (ResolvedArtifactLane, []contracts.Warning, error) {
materializedLane := lane
var allWarnings []contracts.Warning
materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
return ResolvedArtifactLane{}, nil, err
}
materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
warnings = append(warnings, laneWarnings...)
allWarnings = append(allWarnings, laneWarnings...)
mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
return ResolvedArtifactLane{}, nil, err
}
materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
warnings = append(warnings, laneWarnings...)
allWarnings = append(allWarnings, laneWarnings...)
normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
return ResolvedArtifactLane{}, nil, err
}
materializedLane.NormalizeReferences.ReferenceSet = normalizeReferenceSet
warnings = append(warnings, laneWarnings...)
out.ArtifactLanes[i] = materializedLane
allWarnings = append(allWarnings, laneWarnings...)
return materializedLane, allWarnings, nil
}
if len(resolved.Steps) > 0 {
out.Steps = make([]ResolvedPipelineStep, len(resolved.Steps))
for i, step := range resolved.Steps {
out.Steps[i].ID = step.ID
out.Steps[i].ArtifactLanes = make([]ResolvedArtifactLane, len(step.ArtifactLanes))
for j, lane := range step.ArtifactLanes {
materializedLane, laneWarnings, err := materializeLane(lane)
if err != nil {
return ResolvedPipeline{}, nil, err
}
out.Steps[i].ArtifactLanes[j] = materializedLane
warnings = append(warnings, laneWarnings...)
}
}
}
return out, warnings, nil
}
@@ -99,6 +113,11 @@ func materializeReferenceTarget(
if !ok {
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.
continue
}
path, err := referencePath(binding, options)
if err != nil {
@@ -273,6 +292,7 @@ func fileURI(path string) string {
func cloneReferenceSlot(slot contracts.ReferenceSlot) contracts.ReferenceSlot {
slot.AcceptedMediaTypes = append([]string(nil), slot.AcceptedMediaTypes...)
slot.AcceptedArtifactKinds = append([]contracts.ArtifactKind(nil), slot.AcceptedArtifactKinds...)
return slot
}
@@ -304,7 +324,16 @@ func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet {
func CloneReferenceTarget(in ResolvedReferenceTarget) ResolvedReferenceTarget {
out := in
out.Bindings = append([]ReferenceBinding(nil), in.Bindings...)
if len(in.Bindings) > 0 {
out.Bindings = make([]ReferenceBinding, len(in.Bindings))
for i, binding := range in.Bindings {
out.Bindings[i] = binding
if binding.Artifact != nil {
artifact := *binding.Artifact
out.Bindings[i].Artifact = &artifact
}
}
}
out.ReferenceSet = CloneReferenceSet(in.ReferenceSet)
return out
}
@@ -312,10 +341,14 @@ func CloneReferenceTarget(in ResolvedReferenceTarget) ResolvedReferenceTarget {
func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvenance {
provenance := []artifacts.ReferenceProvenance{}
provenance = append(provenance, referenceTargetProvenance(resolved.ChunkReferences)...)
for _, lane := range resolved.ArtifactLanes {
provenance = append(provenance, referenceTargetProvenance(lane.ExtractReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.MergeReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.NormalizeReferences)...)
if len(resolved.Steps) > 0 {
for _, step := range resolved.Steps {
for _, lane := range step.ArtifactLanes {
provenance = append(provenance, referenceTargetProvenance(lane.ExtractReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.MergeReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.NormalizeReferences)...)
}
}
}
return provenance
}
@@ -335,6 +368,7 @@ func referenceTargetProvenance(target ResolvedReferenceTarget) []artifacts.Refer
for _, item := range slot.Items {
provenance = append(provenance, artifacts.ReferenceProvenance{
Stage: string(target.Stage),
StepID: target.StepID,
LaneID: target.LaneID,
SlotName: item.SlotName,
OriginType: item.Origin.Type,

71
internal/framework/pipeline/references_test.go
View File

@@ -21,9 +21,9 @@ func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
writeReferenceFile(t, cliReference, []byte("cli text"))
pipeline := baselineProfile()
pipeline.References = map[string]string{"roster": "config-reference.txt"}
pipeline.References = ExternalReferenceMap(map[string]string{"roster": "config-reference.txt"})
lane := pipeline.Artifacts["events"]
lane.References = map[string]string{"glossary": "cli-reference.txt"}
lane.References = ExternalReferenceMap(map[string]string{"glossary": "cli-reference.txt"})
pipeline.Artifacts["events"] = lane
catalog := referenceCatalog(t, []contracts.ReferenceSlot{
{Name: "roster"},
@@ -56,12 +56,12 @@ func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
t.Fatalf("MaterializeReferences(second) error = %v, want nil", err)
}
referenceSet := first.ArtifactLanes[0].ExtractReferences.ReferenceSet
referenceSet := first.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet
roster := referenceSet.Slots["roster"].Items[0]
if string(roster.Content) != "config text" {
t.Fatalf("roster content = %q, want config text", roster.Content)
}
if roster.Digest != referenceDigest([]byte("config text")) || roster.Digest != second.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Digest {
if roster.Digest != referenceDigest([]byte("config text")) || roster.Digest != second.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Digest {
t.Fatalf("roster digest = %q, want stable digest", roster.Digest)
}
if roster.BindingSource != contracts.ReferenceBindingSourceConfig {
@@ -113,12 +113,12 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize text"))
profile := baselineProfile()
profile.References = map[string]string{
profile.References = ExternalReferenceMap(map[string]string{
"scene_guide": "chunk.txt",
"roster": "extract.txt",
"merge_notes": "merge.txt",
"normalization_notes": "normalize.txt",
}
})
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
@@ -144,15 +144,15 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
if string(chunkItem.Content) != "chunk text" {
t.Fatalf("chunk content = %q, want chunk text", chunkItem.Content)
}
extractItem := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
extractItem := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if string(extractItem.Content) != "extract text" {
t.Fatalf("extract content = %q, want extract text", extractItem.Content)
}
mergeItem := materialized.ArtifactLanes[0].MergeReferences.ReferenceSet.Slots["merge_notes"].Items[0]
mergeItem := materialized.Steps[0].ArtifactLanes[0].MergeReferences.ReferenceSet.Slots["merge_notes"].Items[0]
if string(mergeItem.Content) != "merge text" {
t.Fatalf("merge content = %q, want merge text", mergeItem.Content)
}
normalizeItem := materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["normalization_notes"].Items[0]
normalizeItem := materialized.Steps[0].ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["normalization_notes"].Items[0]
if string(normalizeItem.Content) != "normalize text" {
t.Fatalf("normalize content = %q, want normalize text", normalizeItem.Content)
}
@@ -202,22 +202,25 @@ func TestMaterializeReferencesUsesLaneArtifactVariant(t *testing.T) {
}
resolved := ResolvedPipeline{
ID: "variants",
ArtifactLanes: []ResolvedArtifactLane{{
ID: "alpha",
ArtifactKind: "test/alpha",
MergeReferences: referenceTarget(StageMerge, "alpha", "shared/merge", []ReferenceBinding{{
Stage: StageMerge, LaneID: "alpha", SlotName: "alpha_merge", Source: "merge.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
NormalizeReferences: referenceTarget(StageNormalize, "alpha", "shared/normalize", []ReferenceBinding{{
Stage: StageNormalize, LaneID: "alpha", SlotName: "alpha_normalize", Source: "normalize.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
Steps: []ResolvedPipelineStep{{
ID: "default",
ArtifactLanes: []ResolvedArtifactLane{{
ID: "alpha",
ArtifactKind: "test/alpha",
MergeReferences: referenceTarget(StageMerge, "alpha", "shared/merge", []ReferenceBinding{{
Stage: StageMerge, LaneID: "alpha", SlotName: "alpha_merge", Source: "merge.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
NormalizeReferences: referenceTarget(StageNormalize, "alpha", "shared/normalize", []ReferenceBinding{{
Stage: StageNormalize, LaneID: "alpha", SlotName: "alpha_normalize", Source: "normalize.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
}},
}},
}
materialized, _, err := MaterializeReferences(resolved, ModuleCatalog{Mergers: mergers, Normalizers: normalizers}, ReferenceMaterializationOptions{ConfigPath: filepath.Join(configDir, "notarius.yml")})
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
lane := materialized.ArtifactLanes[0]
lane := materialized.Steps[0].ArtifactLanes[0]
if got := string(lane.MergeReferences.ReferenceSet.Slots["alpha_merge"].Items[0].Content); got != "merge alpha" {
t.Fatalf("merge reference = %q", got)
}
@@ -266,7 +269,7 @@ func TestMaterializeReferencesAllowsAnyMediaTypeWhenSlotDoesNotRestrictIt(t *tes
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if item.MediaType != unknownMediaType {
t.Fatalf("MediaType = %q, want %q", item.MediaType, unknownMediaType)
}
@@ -285,7 +288,7 @@ func TestMaterializeReferencesAcceptsDeclaredMarkdownMediaType(t *testing.T) {
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["glossary"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["glossary"].Items[0]
if item.MediaType != "text/markdown" {
t.Fatalf("MediaType = %q, want text/markdown", item.MediaType)
}
@@ -304,7 +307,7 @@ func TestMaterializeReferencesAcceptsDeclaredJSONMediaType(t *testing.T) {
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if item.MediaType != "application/json" {
t.Fatalf("MediaType = %q, want application/json", item.MediaType)
}
@@ -323,7 +326,7 @@ func TestMaterializeReferencesAcceptsDeclaredYAMLMediaType(t *testing.T) {
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if item.MediaType != "application/yaml" {
t.Fatalf("MediaType = %q, want application/yaml", item.MediaType)
}
@@ -357,7 +360,7 @@ func TestMaterializeReferencesMatchesAcceptedMediaTypesIgnoringParameters(t *tes
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if item.MediaType != referenceMediaType {
t.Fatalf("MediaType = %q, want %q", item.MediaType, referenceMediaType)
}
@@ -393,7 +396,7 @@ func TestMaterializeReferencesWarnsForEmptyFiles(t *testing.T) {
if len(warnings) != 1 || warnings[0].ReasonCode != "empty_reference" {
t.Fatalf("warnings = %#v, want empty reference warning", warnings)
}
item := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if item.SizeBytes != 0 || item.Digest != referenceDigest(nil) {
t.Fatalf("empty item = %#v, want zero size and empty digest", item)
}
@@ -406,11 +409,11 @@ func TestMaterializeReferencesWarningScopesIncludeTargetContext(t *testing.T) {
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), nil)
profile := baselineProfile()
profile.References = map[string]string{
profile.References = ExternalReferenceMap(map[string]string{
"scene_guide": "chunk.txt",
"roster": "extract.txt",
"normalization_notes": "normalize.txt",
}
})
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
@@ -464,18 +467,18 @@ func resolvedPipelineWithTargetReference(t *testing.T, stage ModuleStage, laneID
profile := baselineProfile()
switch stage {
case StageChunk:
profile.Chunk.References = map[string]string{slotName: source}
profile.Chunk.References = ExternalReferenceMap(map[string]string{slotName: source})
case StageExtract:
lane := profile.Artifacts[laneID]
lane.References = map[string]string{slotName: source}
lane.References = ExternalReferenceMap(map[string]string{slotName: source})
profile.Artifacts[laneID] = lane
case StageMerge:
lane := profile.Artifacts[laneID]
lane.Merge.References = map[string]string{slotName: source}
lane.Merge.References = ExternalReferenceMap(map[string]string{slotName: source})
profile.Artifacts[laneID] = lane
case StageNormalize:
lane := profile.Artifacts[laneID]
lane.Normalize.References = map[string]string{slotName: source}
lane.Normalize.References = ExternalReferenceMap(map[string]string{slotName: source})
profile.Artifacts[laneID] = lane
default:
t.Fatalf("unsupported reference target stage %q", stage)
@@ -490,11 +493,11 @@ func resolvedPipelineWithTargetReference(t *testing.T, stage ModuleStage, laneID
case StageChunk:
resolved.ChunkReferences.Bindings[0].BindingSource = bindingSource
case StageExtract:
resolved.ArtifactLanes[0].ExtractReferences.Bindings[0].BindingSource = bindingSource
resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings[0].BindingSource = bindingSource
case StageMerge:
resolved.ArtifactLanes[0].MergeReferences.Bindings[0].BindingSource = bindingSource
resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings[0].BindingSource = bindingSource
case StageNormalize:
resolved.ArtifactLanes[0].NormalizeReferences.Bindings[0].BindingSource = bindingSource
resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings[0].BindingSource = bindingSource
}
}
return resolved

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

@@ -479,10 +479,10 @@ func validateResolvedPipeline(pipeline ResolvedPipeline) error {
if pipeline.Output.Module == "" {
return fmt.Errorf("resolved pipeline output module must not be empty")
}
if len(pipeline.ArtifactLanes) == 0 {
if len(pipeline.Steps) == 0 {
return fmt.Errorf("resolved pipeline artifact lanes must not be empty")
}
for _, lane := range pipeline.ArtifactLanes {
for _, lane := range pipeline.AllArtifactLanes() {
if lane.ID == "" {
return fmt.Errorf("resolved pipeline artifact lane id must not be empty")
}
@@ -530,14 +530,14 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest {
RequestedModule: pipeline.Chunk.Module,
},
OutputEncoder: pipeline.Output.Module,
ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)),
ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.AllArtifactLanes())),
ValidatorChains: validatorChainManifests(pipeline.ValidatorChains),
RunID: runID,
StartedAt: timePtr(startedAt),
References: ReferenceProvenance(pipeline),
LLMProfiles: cloneLLMProfiles(input.LLMProfiles),
}
for _, lane := range pipeline.ArtifactLanes {
for _, lane := range pipeline.AllArtifactLanes() {
laneManifest := artifacts.ArtifactLaneManifest{
ID: lane.ID,
Extractor: lane.Extract.Module,

2
internal/framework/pipeline/runner_attempt_debug_test.go
View File

@@ -110,7 +110,7 @@ func preparedAttemptDebugPipeline(t *testing.T) *PreparedPipeline {
t.Helper()
prepared := preparedConcurrentPipeline(t, 1)
prepared.lanes = prepared.lanes[:1]
prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1]
prepared.resolved.Steps[0].ArtifactLanes = prepared.resolved.Steps[0].ArtifactLanes[:1]
prepared.ArtifactLanes = prepared.ArtifactLanes[:1]
prepared.lanes[0].mergeValidators = preparedValidatorChain{}
prepared.lanes[0].normalizeValidators = preparedValidatorChain{}

2
internal/framework/pipeline/runner_chunk_plan_test.go
View File

@@ -470,7 +470,7 @@ func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) {
doc := typedTestDocumentWithUnits(2)
prepared := preparedConcurrentPipeline(t, 1)
prepared.lanes = prepared.lanes[:1]
prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1]
prepared.resolved.Steps[0].ArtifactLanes = prepared.resolved.Steps[0].ArtifactLanes[:1]
prepared.ArtifactLanes = prepared.ArtifactLanes[:1]
prepared.input = &typedTestInput{key: prepared.resolved.Input.Module, doc: doc}

2
internal/framework/pipeline/runner_concurrency_test.go
View File

@@ -216,7 +216,7 @@ func TestRunnerBoundsConcurrentLaneContinuations(t *testing.T) {
publicLanes[i] = PreparedArtifactLane{Resolved: lane.resolved}
}
prepared.lanes = lanes
prepared.resolved.ArtifactLanes = resolvedLanes
prepared.resolved.Steps[0].ArtifactLanes = resolvedLanes
prepared.ArtifactLanes = publicLanes
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2})

6
internal/framework/pipeline/typed_resolution_test.go
View File

@@ -147,11 +147,11 @@ func TestResolveTypedHeterogeneousLanes(t *testing.T) {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if got, want := resolvedLaneIDs(resolved.ArtifactLanes), []string{"notes", "score"}; !reflect.DeepEqual(got, want) {
if got, want := resolvedLaneIDs(resolved.Steps[0].ArtifactLanes), []string{"notes", "score"}; !reflect.DeepEqual(got, want) {
t.Fatalf("lane order = %#v, want %#v", got, want)
}
assertResolvedArtifactIdentity(t, resolved.ArtifactLanes[0], "test/notes", "notes.v1")
assertResolvedArtifactIdentity(t, resolved.ArtifactLanes[1], "test/score", "score.v1")
assertResolvedArtifactIdentity(t, resolved.Steps[0].ArtifactLanes[0], "test/notes", "notes.v1")
assertResolvedArtifactIdentity(t, resolved.Steps[0].ArtifactLanes[1], "test/score", "score.v1")
chunkChain := resolved.ValidatorChains[0]
if got := resolvedValidatorTargets(chunkChain.Validators); !reflect.DeepEqual(got, []ValidatorTarget{ValidatorTargetChunk, ValidatorTargetSerialized}) {

13
internal/framework/pipeline/validator_chain_registry.go
View File

@@ -92,9 +92,9 @@ func cloneModuleBinding(binding ModuleBinding) ModuleBinding {
binding.LLMProfile = strings.TrimSpace(binding.LLMProfile)
binding.Options = cloneOptions(binding.Options)
if len(binding.References) > 0 {
references := make(map[string]string, len(binding.References))
references := make(map[string]ReferenceSource, len(binding.References))
for key, value := range binding.References {
references[key] = value
references[key] = cloneReferenceSource(value)
}
binding.References = references
}
@@ -102,6 +102,15 @@ func cloneModuleBinding(binding ModuleBinding) ModuleBinding {
return binding
}
func cloneReferenceSource(source ReferenceSource) ReferenceSource {
out := source
if source.Artifact != nil {
artifact := *source.Artifact
out.Artifact = &artifact
}
return out
}
func cloneValidatorOverride(override ValidatorOverride) ValidatorOverride {
return ValidatorOverride{
Set: override.Set,

9
internal/modules/dnd/extract/combatturns/extractor.go
View File

@@ -42,10 +42,11 @@ var referenceSlotDescriptions = shared.ReferenceSlotDescriptions{
func referenceSlots() []contracts.ReferenceSlot {
slots := shared.ReferenceSlots(referenceSlotDescriptions)
slots = append(slots, contracts.ReferenceSlot{
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical actor and target grounding.",
AcceptedMediaTypes: []string{"application/json"},
MaxBytes: NPCRegistryMaxBytes,
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical actor and target grounding.",
AcceptedMediaTypes: []string{"application/json"},
AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.NPCListKind},
MaxBytes: NPCRegistryMaxBytes,
})
sort.Slice(slots, func(i, j int) bool { return slots[i].Name < slots[j].Name })
return slots

9
internal/modules/dnd/extract/spells/extractor.go
View File

@@ -46,10 +46,11 @@ func referenceSlots() []contracts.ReferenceSlot {
AcceptedMediaTypes: []string{"application/json"},
MaxBytes: 1048576,
}, contracts.ReferenceSlot{
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical caster-name grounding.",
AcceptedMediaTypes: []string{"application/json"},
MaxBytes: NPCRegistryMaxBytes,
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical caster-name grounding.",
AcceptedMediaTypes: []string{"application/json"},
AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.NPCListKind},
MaxBytes: NPCRegistryMaxBytes,
})
sort.Slice(slots, func(i, j int) bool { return slots[i].Name < slots[j].Name })
return slots

9
internal/modules/dnd/extract/spells/registry_test.go
View File

@@ -40,10 +40,11 @@ func TestModuleSpec(t *testing.T) {
AcceptedMediaTypes: []string{"application/json", "application/x-yaml", "application/yaml", "text/markdown", "text/plain"},
},
{
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical caster-name grounding.",
AcceptedMediaTypes: []string{"application/json"},
MaxBytes: NPCRegistryMaxBytes,
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical caster-name grounding.",
AcceptedMediaTypes: []string{"application/json"},
AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.NPCListKind},
MaxBytes: NPCRegistryMaxBytes,
},
{
Name: "party",

9
internal/modules/dnd/normalize/combatturns/normalizer.go
View File

@@ -494,10 +494,11 @@ func turnScope(index int) string { return fmt.Sprintf("combat_turns[%d]", index)
func referenceSlots() []contracts.ReferenceSlot {
return []contracts.ReferenceSlot{{
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical actor and target grounding.",
AcceptedMediaTypes: []string{"application/json"},
MaxBytes: NPCRegistryMaxBytes,
Name: NPCRegistryReferenceSlot,
Description: "Optional normalized NPC registry used for canonical actor and target grounding.",
AcceptedMediaTypes: []string{"application/json"},
AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.NPCListKind},
MaxBytes: NPCRegistryMaxBytes,
}}
}

2
internal/modules/integration/dnd_npc_spell_sequential_test.go
View File

@@ -56,7 +56,7 @@ func TestSequentialNPCOutputCanGroundIndependentSpellRun(t *testing.T) {
if err != nil {
t.Fatalf("resolve spell pipeline: %v", err)
}
spellEffective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings = []pipeline.ReferenceBinding{{
spellEffective.ResolvedPipeline.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings = []pipeline.ReferenceBinding{{
Stage: pipeline.StageExtract,
LaneID: "spells",
SlotName: spells.NPCRegistryReferenceSlot,

2
internal/modules/integration/dnd_spells_runner_test.go
View File

@@ -118,7 +118,7 @@ func TestRunnerPassesPartyAndGlossaryReferencesToDNDSpellsPrompt(t *testing.T) {
raw := readDNDSpellsFixture(t)
expectedDoc := parseDNDSpellsFixture(t, raw)
resolved := resolveDNDSpellsPipeline(t)
resolved.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet = dndSpellsReferenceSet(
resolved.ResolvedPipeline.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet = dndSpellsReferenceSet(
"Aria: party cleric\nBorin: fighter",
"Fire Bolt: evocation cantrip",
)