Add typed artifact codec foundation

This commit is contained in:
2026-07-17 05:41:20 +00:00
parent 075888c97f
commit fc1b57bde2
22 changed files with 795 additions and 33 deletions

View File

@@ -38,8 +38,8 @@ a sorted set of artifact lanes before the runner constructs any stage module.
| Package | Implemented responsibility |
| --- | --- |
| `internal/framework/contracts` | Stage, validator, reference, output, and structured-completion interfaces and data types. |
| `internal/framework/pipeline` | Registries, profile resolution, capability checks, reference materialization, validator-chain resolution, retries, orchestration, warnings, and manifest population. |
| `internal/framework/contracts` | Source-stage contracts plus artifact identity, schema, serialized representation, codec, validator, reference, output, and structured-completion interfaces and data types. |
| `internal/framework/pipeline` | Module and artifact-codec registries, profile resolution, capability checks, reference materialization, validator-chain resolution, retries, orchestration, warnings, and manifest population. |
| `internal/framework/validate` | Shared validator decision and cardinality helpers. |
| `internal/framework/llm` | Scriptorium-backed structured completions, prompt/schema registration, scheduling, profile recording, and secret redaction. |
| `internal/framework/promptfs` | Builds module prompt filesystems from module-owned and caller-provided shared prompt assets. |
@@ -50,6 +50,12 @@ Framework contracts carry raw stage results between implementations. The
runner owns handoff provenance, validation sequencing, rejection handling,
checkpoint and debug boundaries, and final manifest assembly.
The artifact codec registry is an implemented foundation for heterogeneous
typed artifacts. It validates codec metadata and schema identity and keeps
exact Go-type checks behind framework-owned erased operations. Production
module families do not register codecs yet, so pipeline resolution and
execution continue to use the existing raw artifact contracts.
## Production Extensions
The canonical catalogs of user-selectable

View File

@@ -53,7 +53,15 @@ runtime sensitive-data handling belongs in [Operations](../operations.md).
`pipeline.Registries` holds constructors used during execution.
`pipeline.ModuleCatalog` exposes their specs during configuration validation and
resolution. Separate registries exist for every stage and for validators;
`ValidatorChainRegistry` stores production default-chain mappings.
`ValidatorChainRegistry` stores production default-chain mappings. Both
containers also carry an `ArtifactCodecRegistry`. Generic registration records
one codec per stable artifact kind, validates its schema metadata and JSON
Schema, retains the exact schema digest and Go type, and safely encodes or
decodes framework-erased values with typed errors on incompatibility.
Production composition initializes the artifact codec registry without adding
codec entries. The resolver and runner do not consult it yet, and production
lanes remain on the raw artifact path.
A `ModuleSpec` declares its stage plus required and provided capabilities.
Chunk, extract, merge, and normalize specs may also declare reference slots.
@@ -183,6 +191,8 @@ durable manifest and logical file schemas are defined in the
- `internal/core/config/effective_config_test.go`: config-to-resolution boundary.
- `internal/framework/pipeline/profile_test.go`: selection, defaults,
capabilities, validator chains, and digest behavior.
- `internal/framework/pipeline/artifact_codec_registry_test.go`: typed codec
metadata, registration, erasure safety, strict decoding, and cloning.
- `internal/framework/pipeline/references_test.go`: target resolution and
materialization.
- `internal/framework/pipeline/runner_test.go`: stage transitions, retries,

View File

@@ -23,6 +23,7 @@ func newProductionComponents() (productionComponents, error) {
registries := pipeline.Registries{
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),
@@ -89,6 +90,7 @@ func catalogFromRegistries(registries pipeline.Registries) pipeline.ModuleCatalo
return pipeline.ModuleCatalog{
Inputs: registries.Inputs,
Chunkers: registries.Chunkers,
ArtifactCodecs: registries.ArtifactCodecs,
Extractors: registries.Extractors,
Mergers: registries.Mergers,
Normalizers: registries.Normalizers,
@@ -102,6 +104,7 @@ func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries {
return pipeline.Registries{
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
ArtifactCodecs: catalog.ArtifactCodecs,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
@@ -114,6 +117,7 @@ func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries {
func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool {
return catalog.Inputs == nil &&
catalog.Chunkers == nil &&
catalog.ArtifactCodecs == nil &&
catalog.Extractors == nil &&
catalog.Mergers == nil &&
catalog.Normalizers == nil &&
@@ -125,6 +129,7 @@ func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool {
func isEmptyRegistries(registries pipeline.Registries) bool {
return registries.Inputs == nil &&
registries.Chunkers == nil &&
registries.ArtifactCodecs == nil &&
registries.Extractors == nil &&
registries.Mergers == nil &&
registries.Normalizers == nil &&

View File

@@ -135,6 +135,12 @@ func TestProductionCatalogIncludesProductionModulesValidatorsAndDefaults(t *test
if err != nil {
t.Fatalf("productionCatalog() error = %v, want nil", err)
}
if catalog.ArtifactCodecs == nil {
t.Fatal("production artifact codec registry = nil, want initialized empty registry")
}
if got := catalog.ArtifactCodecs.RegisteredKinds(); len(got) != 0 {
t.Fatalf("production artifact codec kinds = %#v, want legacy raw production path", got)
}
moduleTests := []struct {
name string
@@ -227,6 +233,23 @@ func TestProductionCatalogIncludesProductionModulesValidatorsAndDefaults(t *test
}
}
func TestCatalogConversionsPreserveArtifactCodecRegistry(t *testing.T) {
codecs := pipeline.NewArtifactCodecRegistry()
registries := pipeline.Registries{ArtifactCodecs: codecs}
if isEmptyRegistries(registries) {
t.Fatal("registries with artifact codecs reported empty")
}
catalog := catalogFromRegistries(registries)
if catalog.ArtifactCodecs != codecs || isEmptyCatalog(catalog) {
t.Fatalf("catalog artifact codecs = %p empty=%t, want %p and non-empty", catalog.ArtifactCodecs, isEmptyCatalog(catalog), codecs)
}
converted := registriesFromCatalog(catalog)
if converted.ArtifactCodecs != codecs {
t.Fatalf("converted artifact codecs = %p, want %p", converted.ArtifactCodecs, codecs)
}
}
func TestProductionPromptAssetsRegisterAndPrepareDndPrompts(t *testing.T) {
registry, err := productionPromptAssets()
if err != nil {
@@ -3572,12 +3595,13 @@ func fakeExecutionRegistries(t *testing.T) pipeline.Registries {
}
return pipeline.Registries{
Inputs: inputs,
Chunkers: chunkers,
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Outputs: outputs,
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Outputs: outputs,
}
}
@@ -3953,6 +3977,7 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -518,6 +518,7 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -0,0 +1,67 @@
package contracts
import (
"crypto/sha256"
"encoding/hex"
)
// ArtifactKind is the stable logical identity of a domain artifact.
type ArtifactKind string
// ArtifactSchema describes the durable representation owned by an artifact
// codec. JSONSchema is cloned whenever framework ownership changes.
type ArtifactSchema struct {
ID string `json:"id"`
Name string `json:"name"`
Version string `json:"version"`
JSONSchema []byte `json:"-"`
}
// SerializedArtifact is the domain-neutral representation of a typed
// artifact at an explicit serialization boundary.
type SerializedArtifact struct {
Kind ArtifactKind `json:"kind"`
Schema ArtifactSchema `json:"schema"`
MediaType string `json:"media_type"`
Content []byte `json:"-"`
Metadata map[string]any `json:"metadata,omitempty"`
}
// ArtifactCodec owns the stable encoding for one concrete artifact type.
type ArtifactCodec[T any] interface {
Kind() ArtifactKind
Schema() ArtifactSchema
MediaType() string
Encode(T) ([]byte, error)
Decode([]byte) (T, error)
}
// DigestArtifactSchema returns the SHA-256 digest of the exact JSON Schema
// bytes. Schema formatting is therefore part of the registered identity.
func DigestArtifactSchema(schema ArtifactSchema) string {
sum := sha256.Sum256(schema.JSONSchema)
return "sha256:" + hex.EncodeToString(sum[:])
}
func CloneArtifactSchema(schema ArtifactSchema) ArtifactSchema {
schema.JSONSchema = append([]byte(nil), schema.JSONSchema...)
return schema
}
func CloneSerializedArtifact(artifact SerializedArtifact) SerializedArtifact {
artifact.Schema = CloneArtifactSchema(artifact.Schema)
artifact.Content = append([]byte(nil), artifact.Content...)
artifact.Metadata = cloneArtifactMetadata(artifact.Metadata)
return artifact
}
func cloneArtifactMetadata(metadata map[string]any) map[string]any {
if len(metadata) == 0 {
return nil
}
out := make(map[string]any, len(metadata))
for key, value := range metadata {
out[key] = value
}
return out
}

View File

@@ -0,0 +1,45 @@
package contracts
import "testing"
func TestDigestArtifactSchemaUsesExactBytes(t *testing.T) {
first := ArtifactSchema{JSONSchema: []byte(`{"type":"object"}`)}
second := ArtifactSchema{JSONSchema: []byte("{\n \"type\": \"object\"\n}")}
if got := DigestArtifactSchema(first); got != "sha256:a2c799262a3ce3c19ef5cdd983bf3d12b43ab3c426227091b909dcb7054738c0" {
t.Fatalf("DigestArtifactSchema() = %q, want stable SHA-256", got)
}
if DigestArtifactSchema(first) == DigestArtifactSchema(second) {
t.Fatal("schema digests match for different exact bytes")
}
}
func TestArtifactCloneHelpersOwnSlicesAndMaps(t *testing.T) {
schema := ArtifactSchema{ID: "notes.v1", Name: "notes", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)}
artifact := SerializedArtifact{
Kind: "test/notes",
Schema: schema,
MediaType: "application/json",
Content: []byte(`{"items":["one"]}`),
Metadata: map[string]any{"origin": "test"},
}
clonedSchema := CloneArtifactSchema(schema)
cloned := CloneSerializedArtifact(artifact)
schema.JSONSchema[0] = '['
artifact.Content[0] = '['
artifact.Metadata["origin"] = "changed"
if string(clonedSchema.JSONSchema) != `{"type":"object"}` {
t.Fatalf("cloned schema = %q, want original bytes", clonedSchema.JSONSchema)
}
if string(cloned.Schema.JSONSchema) != `{"type":"object"}` {
t.Fatalf("serialized artifact schema = %q, want original bytes", cloned.Schema.JSONSchema)
}
if string(cloned.Content) != `{"items":["one"]}` {
t.Fatalf("cloned content = %q, want original bytes", cloned.Content)
}
if cloned.Metadata["origin"] != "test" {
t.Fatalf("cloned metadata = %#v, want independent map", cloned.Metadata)
}
}

View File

@@ -0,0 +1,282 @@
package pipeline
import (
"bytes"
"encoding/json"
"fmt"
"reflect"
"sort"
"strings"
"github.com/santhosh-tekuri/jsonschema/v6"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type ArtifactCodecSpec struct {
Kind contracts.ArtifactKind
Schema contracts.ArtifactSchema
SchemaDigest string
MediaType string
}
// ArtifactCodecTypeError reports a value that does not have the exact Go type
// registered for an artifact kind.
type ArtifactCodecTypeError struct {
Operation string
Kind contracts.ArtifactKind
ExpectedType string
ActualType string
}
func (e *ArtifactCodecTypeError) Error() string {
return fmt.Sprintf("%s artifact %q: expected exact Go type %s, got %s", e.Operation, e.Kind, e.ExpectedType, e.ActualType)
}
// ArtifactCodecCompatibilityError reports serialized metadata that does not
// identify the registered representation for an artifact kind.
type ArtifactCodecCompatibilityError struct {
Kind contracts.ArtifactKind
Reason string
}
func (e *ArtifactCodecCompatibilityError) Error() string {
return fmt.Sprintf("decode artifact %q: %s", e.Kind, e.Reason)
}
// ArtifactCodecOperationError preserves an encode or decode failure from the
// registered domain codec.
type ArtifactCodecOperationError struct {
Operation string
Kind contracts.ArtifactKind
Err error
}
func (e *ArtifactCodecOperationError) Error() string {
return fmt.Sprintf("%s artifact %q: %v", e.Operation, e.Kind, e.Err)
}
func (e *ArtifactCodecOperationError) Unwrap() error { return e.Err }
type ArtifactCodecRegistry struct {
entries map[contracts.ArtifactKind]artifactCodecEntry
}
type artifactCodecEntry struct {
spec ArtifactCodecSpec
valueType reflect.Type
encode func(any) ([]byte, error)
decode func([]byte) (any, error)
}
func NewArtifactCodecRegistry() *ArtifactCodecRegistry {
return &ArtifactCodecRegistry{entries: make(map[contracts.ArtifactKind]artifactCodecEntry)}
}
// RegisterArtifactCodec registers one codec for T. The concrete type is kept
// private and checked at every erased encode boundary.
func RegisterArtifactCodec[T any](registry *ArtifactCodecRegistry, codec contracts.ArtifactCodec[T]) error {
if registry == nil {
return fmt.Errorf("artifact codec registry must not be nil")
}
if nilInterface(codec) {
return fmt.Errorf("artifact codec must not be nil")
}
spec, err := artifactCodecSpec(codec.Kind(), codec.Schema(), codec.MediaType())
if err != nil {
return err
}
if _, ok := registry.entries[spec.Kind]; ok {
return fmt.Errorf("artifact codec %q is already registered", spec.Kind)
}
valueType := reflect.TypeFor[T]()
entry := artifactCodecEntry{
spec: cloneArtifactCodecSpec(spec),
valueType: valueType,
encode: func(value any) ([]byte, error) {
actualType := reflect.TypeOf(value)
if actualType != valueType {
return nil, newArtifactCodecTypeError("encode", spec.Kind, valueType, actualType)
}
typed, ok := value.(T)
if !ok {
return nil, newArtifactCodecTypeError("encode", spec.Kind, valueType, actualType)
}
encoded, err := codec.Encode(typed)
if err != nil {
return nil, &ArtifactCodecOperationError{Operation: "encode", Kind: spec.Kind, Err: err}
}
return append([]byte(nil), encoded...), nil
},
decode: func(content []byte) (any, error) {
decoded, err := codec.Decode(append([]byte(nil), content...))
if err != nil {
return nil, &ArtifactCodecOperationError{Operation: "decode", Kind: spec.Kind, Err: err}
}
return decoded, nil
},
}
if registry.entries == nil {
registry.entries = make(map[contracts.ArtifactKind]artifactCodecEntry)
}
registry.entries[spec.Kind] = entry
return nil
}
func (r *ArtifactCodecRegistry) Spec(kind contracts.ArtifactKind) (ArtifactCodecSpec, bool) {
if r == nil {
return ArtifactCodecSpec{}, false
}
entry, ok := r.entries[normalizeArtifactKind(kind)]
if !ok {
return ArtifactCodecSpec{}, false
}
return cloneArtifactCodecSpec(entry.spec), true
}
func (r *ArtifactCodecRegistry) RegisteredKinds() []contracts.ArtifactKind {
if r == nil || len(r.entries) == 0 {
return nil
}
kinds := make([]contracts.ArtifactKind, 0, len(r.entries))
for kind := range r.entries {
kinds = append(kinds, kind)
}
sort.Slice(kinds, func(i, j int) bool { return kinds[i] < kinds[j] })
return kinds
}
// Encode serializes an erased framework value after proving its exact
// registered Go type.
func (r *ArtifactCodecRegistry) Encode(kind contracts.ArtifactKind, value any) (contracts.SerializedArtifact, error) {
entry, normalizedKind, err := r.entry(kind)
if err != nil {
return contracts.SerializedArtifact{}, err
}
content, err := entry.encode(value)
if err != nil {
return contracts.SerializedArtifact{}, err
}
return contracts.SerializedArtifact{
Kind: normalizedKind,
Schema: contracts.CloneArtifactSchema(entry.spec.Schema),
MediaType: entry.spec.MediaType,
Content: content,
}, nil
}
// Decode verifies serialized identity before invoking the registered codec.
func (r *ArtifactCodecRegistry) Decode(artifact contracts.SerializedArtifact) (any, error) {
entry, normalizedKind, err := r.entry(artifact.Kind)
if err != nil {
return nil, err
}
if artifact.Schema.ID != entry.spec.Schema.ID || artifact.Schema.Name != entry.spec.Schema.Name || artifact.Schema.Version != entry.spec.Schema.Version {
return nil, &ArtifactCodecCompatibilityError{Kind: normalizedKind, Reason: "schema identity does not match registered codec"}
}
if contracts.DigestArtifactSchema(artifact.Schema) != entry.spec.SchemaDigest {
return nil, &ArtifactCodecCompatibilityError{Kind: normalizedKind, Reason: "schema digest does not match registered codec"}
}
if strings.TrimSpace(artifact.MediaType) != entry.spec.MediaType {
return nil, &ArtifactCodecCompatibilityError{Kind: normalizedKind, Reason: "media type does not match registered codec"}
}
return entry.decode(artifact.Content)
}
func (r *ArtifactCodecRegistry) entry(kind contracts.ArtifactKind) (artifactCodecEntry, contracts.ArtifactKind, error) {
if r == nil {
return artifactCodecEntry{}, "", fmt.Errorf("artifact codec registry must not be nil")
}
normalizedKind := normalizeArtifactKind(kind)
if normalizedKind == "" {
return artifactCodecEntry{}, "", fmt.Errorf("artifact kind must not be empty")
}
entry, ok := r.entries[normalizedKind]
if !ok {
return artifactCodecEntry{}, normalizedKind, fmt.Errorf("artifact codec %q is not registered", normalizedKind)
}
return entry, normalizedKind, nil
}
func artifactCodecSpec(kind contracts.ArtifactKind, schema contracts.ArtifactSchema, mediaType string) (ArtifactCodecSpec, error) {
kind = normalizeArtifactKind(kind)
schema.ID = strings.TrimSpace(schema.ID)
schema.Name = strings.TrimSpace(schema.Name)
schema.Version = strings.TrimSpace(schema.Version)
mediaType = strings.TrimSpace(mediaType)
switch {
case kind == "":
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec kind must not be empty")
case schema.ID == "":
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q schema id must not be empty", kind)
case schema.Name == "":
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q schema name must not be empty", kind)
case schema.Version == "":
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q schema version must not be empty", kind)
case mediaType == "":
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q media type must not be empty", kind)
case len(bytes.TrimSpace(schema.JSONSchema)) == 0:
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q JSON Schema must not be empty", kind)
case !json.Valid(schema.JSONSchema):
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q JSON Schema must be valid JSON", kind)
}
schemaDocument, err := jsonschema.UnmarshalJSON(bytes.NewReader(schema.JSONSchema))
if err != nil {
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q JSON Schema is invalid: %w", kind, err)
}
compiler := jsonschema.NewCompiler()
if err := compiler.AddResource("artifact-schema.json", schemaDocument); err != nil {
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q JSON Schema is invalid: %w", kind, err)
}
if _, err := compiler.Compile("artifact-schema.json"); err != nil {
return ArtifactCodecSpec{}, fmt.Errorf("artifact codec %q JSON Schema is invalid: %w", kind, err)
}
schema = contracts.CloneArtifactSchema(schema)
return ArtifactCodecSpec{
Kind: kind,
Schema: schema,
SchemaDigest: contracts.DigestArtifactSchema(schema),
MediaType: mediaType,
}, nil
}
func normalizeArtifactKind(kind contracts.ArtifactKind) contracts.ArtifactKind {
return contracts.ArtifactKind(strings.TrimSpace(string(kind)))
}
func cloneArtifactCodecSpec(spec ArtifactCodecSpec) ArtifactCodecSpec {
spec.Schema = contracts.CloneArtifactSchema(spec.Schema)
return spec
}
func nilInterface(value any) bool {
if value == nil {
return true
}
reflected := reflect.ValueOf(value)
switch reflected.Kind() {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
return reflected.IsNil()
default:
return false
}
}
func newArtifactCodecTypeError(operation string, kind contracts.ArtifactKind, expected reflect.Type, actual reflect.Type) error {
expectedName := "<nil>"
if expected != nil {
expectedName = expected.String()
}
actualName := "<nil>"
if actual != nil {
actualName = actual.String()
}
return &ArtifactCodecTypeError{
Operation: operation,
Kind: kind,
ExpectedType: expectedName,
ActualType: actualName,
}
}

View File

@@ -0,0 +1,307 @@
package pipeline
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type codecNotes struct {
Items []string `json:"items"`
}
type codecScore struct {
Value int `json:"value"`
}
type codecNotesAlias codecNotes
type testArtifactCodec[T any] struct {
kind contracts.ArtifactKind
schema contracts.ArtifactSchema
mediaType string
encodeFunc func(T) ([]byte, error)
decodeFunc func([]byte) (T, error)
}
func (c testArtifactCodec[T]) Kind() contracts.ArtifactKind { return c.kind }
func (c testArtifactCodec[T]) Schema() contracts.ArtifactSchema { return c.schema }
func (c testArtifactCodec[T]) MediaType() string { return c.mediaType }
func (c testArtifactCodec[T]) Encode(value T) ([]byte, error) { return c.encodeFunc(value) }
func (c testArtifactCodec[T]) Decode(content []byte) (T, error) { return c.decodeFunc(content) }
func TestArtifactCodecRegistryStoresHeterogeneousExactTypes(t *testing.T) {
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, notesCodec()); err != nil {
t.Fatalf("RegisterArtifactCodec(notes) error = %v, want nil", err)
}
if err := RegisterArtifactCodec(registry, scoreCodec()); err != nil {
t.Fatalf("RegisterArtifactCodec(score) error = %v, want nil", err)
}
if got, want := registry.RegisteredKinds(), []contracts.ArtifactKind{"test/notes", "test/score"}; !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredKinds() = %#v, want %#v", got, want)
}
notes := codecNotes{Items: []string{"second", "first"}}
first, err := registry.Encode("test/notes", notes)
if err != nil {
t.Fatalf("Encode(notes) error = %v, want nil", err)
}
second, err := registry.Encode(" test/notes ", notes)
if err != nil {
t.Fatalf("Encode(notes again) error = %v, want nil", err)
}
if !bytes.Equal(first.Content, second.Content) {
t.Fatalf("equal values encoded as %q and %q, want deterministic bytes", first.Content, second.Content)
}
decodedNotes, err := registry.Decode(first)
if err != nil {
t.Fatalf("Decode(notes) error = %v, want nil", err)
}
if !reflect.DeepEqual(decodedNotes, notes) {
t.Fatalf("Decode(notes) = %#v, want %#v", decodedNotes, notes)
}
score := codecScore{Value: 17}
serializedScore, err := registry.Encode("test/score", score)
if err != nil {
t.Fatalf("Encode(score) error = %v, want nil", err)
}
decodedScore, err := registry.Decode(serializedScore)
if err != nil {
t.Fatalf("Decode(score) error = %v, want nil", err)
}
if decodedScore != score {
t.Fatalf("Decode(score) = %#v, want %#v", decodedScore, score)
}
_, err = registry.Encode("test/notes", codecNotesAlias(notes))
var typeErr *ArtifactCodecTypeError
if !errors.As(err, &typeErr) {
t.Fatalf("Encode(alias) error = %T %v, want ArtifactCodecTypeError", err, err)
}
if typeErr.ExpectedType == typeErr.ActualType {
t.Fatalf("type error = %#v, want distinct exact types", typeErr)
}
}
func TestArtifactCodecRegistryStoresValidatedSchemaMetadata(t *testing.T) {
registry := NewArtifactCodecRegistry()
codec := notesCodec()
codec.kind = " test/notes "
codec.schema.ID = " notes.v1 "
codec.schema.Name = " notes "
codec.schema.Version = " v1 "
codec.mediaType = " application/json "
if err := RegisterArtifactCodec(registry, codec); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
codec.schema.JSONSchema[0] = '['
spec, ok := registry.Spec("test/notes")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
if spec.Kind != "test/notes" || spec.Schema.ID != "notes.v1" || spec.Schema.Name != "notes" || spec.Schema.Version != "v1" || spec.MediaType != "application/json" {
t.Fatalf("Spec() = %#v, want normalized metadata", spec)
}
if spec.SchemaDigest != contracts.DigestArtifactSchema(spec.Schema) {
t.Fatalf("schema digest = %q, want %q", spec.SchemaDigest, contracts.DigestArtifactSchema(spec.Schema))
}
spec.Schema.JSONSchema[0] = '['
again, _ := registry.Spec("test/notes")
if string(again.Schema.JSONSchema) != `{"additionalProperties":false,"properties":{"items":{"items":{"type":"string"},"type":"array"}},"required":["items"],"type":"object"}` {
t.Fatalf("stored JSON Schema changed through Spec result: %q", again.Schema.JSONSchema)
}
}
func TestArtifactCodecRegistryRejectsInvalidRegistration(t *testing.T) {
tests := []struct {
name string
mutate func(*testArtifactCodec[codecNotes])
want string
}{
{name: "kind", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.kind = " " }, want: "kind"},
{name: "schema id", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.ID = "" }, want: "schema id"},
{name: "schema name", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.Name = "" }, want: "schema name"},
{name: "schema version", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.Version = "" }, want: "schema version"},
{name: "media type", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.mediaType = "" }, want: "media type"},
{name: "empty JSON Schema", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.JSONSchema = nil }, want: "JSON Schema"},
{name: "invalid JSON Schema", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.JSONSchema = []byte(`{"type":`) }, want: "valid JSON"},
{name: "non-schema JSON", mutate: func(codec *testArtifactCodec[codecNotes]) { codec.schema.JSONSchema = []byte(`[]`) }, want: "JSON Schema is invalid"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registry := NewArtifactCodecRegistry()
codec := notesCodec()
test.mutate(&codec)
if err := RegisterArtifactCodec(registry, codec); err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("RegisterArtifactCodec() error = %v, want %q", err, test.want)
}
})
}
}
func TestArtifactCodecRegistryRejectsDuplicateKind(t *testing.T) {
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, notesCodec()); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
duplicate := notesCodec()
duplicate.kind = " test/notes "
if err := RegisterArtifactCodec(registry, duplicate); err == nil || !strings.Contains(err.Error(), "already registered") {
t.Fatalf("duplicate registration error = %v, want duplicate kind error", err)
}
}
func TestArtifactCodecRegistryRejectsNilRegistryAndCodec(t *testing.T) {
codec := notesCodec()
if err := RegisterArtifactCodec[codecNotes](nil, codec); err == nil || !strings.Contains(err.Error(), "registry") {
t.Fatalf("nil registry error = %v, want registry error", err)
}
var nilCodec *testArtifactCodec[codecNotes]
if err := RegisterArtifactCodec(NewArtifactCodecRegistry(), nilCodec); err == nil || !strings.Contains(err.Error(), "must not be nil") {
t.Fatalf("nil codec error = %v, want codec error", err)
}
}
func TestArtifactCodecRegistryStrictDecodeAndTypedFailures(t *testing.T) {
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, notesCodec()); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
valid, err := registry.Encode("test/notes", codecNotes{Items: []string{"one"}})
if err != nil {
t.Fatalf("Encode() error = %v, want nil", err)
}
for _, content := range [][]byte{
[]byte(`{"items":["one"],"unknown":true}`),
[]byte(`{"items":["one"]} {}`),
} {
candidate := contracts.CloneSerializedArtifact(valid)
candidate.Content = content
_, err := registry.Decode(candidate)
var operationErr *ArtifactCodecOperationError
if !errors.As(err, &operationErr) || operationErr.Operation != "decode" {
t.Fatalf("Decode(%q) error = %T %v, want typed decode error", content, err, err)
}
}
wrongSchema := contracts.CloneSerializedArtifact(valid)
wrongSchema.Schema.Version = "v2"
_, err = registry.Decode(wrongSchema)
var compatibilityErr *ArtifactCodecCompatibilityError
if !errors.As(err, &compatibilityErr) {
t.Fatalf("Decode(wrong schema) error = %T %v, want ArtifactCodecCompatibilityError", err, err)
}
}
func TestArtifactCodecRegistryWrapsEncodeFailure(t *testing.T) {
codec := notesCodec()
cause := errors.New("cannot encode notes")
codec.encodeFunc = func(codecNotes) ([]byte, error) { return nil, cause }
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, codec); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
_, err := registry.Encode("test/notes", codecNotes{})
var operationErr *ArtifactCodecOperationError
if !errors.As(err, &operationErr) || operationErr.Operation != "encode" || !errors.Is(err, cause) {
t.Fatalf("Encode() error = %T %v, want typed wrapping encode error", err, err)
}
}
func TestArtifactCodecRegistryClonesCodecBytes(t *testing.T) {
shared := []byte(`{"items":["one"]}`)
codec := notesCodec()
codec.encodeFunc = func(codecNotes) ([]byte, error) { return shared, nil }
codec.decodeFunc = func(content []byte) (codecNotes, error) {
content[0] = '['
return codecNotes{Items: []string{"one"}}, nil
}
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, codec); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
artifact, err := registry.Encode("test/notes", codecNotes{})
if err != nil {
t.Fatalf("Encode() error = %v, want nil", err)
}
shared[0] = '['
if string(artifact.Content) != `{"items":["one"]}` {
t.Fatalf("encoded content = %q after codec buffer mutation, want owned bytes", artifact.Content)
}
before := append([]byte(nil), artifact.Content...)
if _, err := registry.Decode(artifact); err != nil {
t.Fatalf("Decode() error = %v, want nil", err)
}
if !bytes.Equal(artifact.Content, before) {
t.Fatalf("serialized content changed during decode: %q", artifact.Content)
}
}
func notesCodec() testArtifactCodec[codecNotes] {
return testArtifactCodec[codecNotes]{
kind: "test/notes",
schema: contracts.ArtifactSchema{
ID: "notes.v1",
Name: "notes",
Version: "v1",
JSONSchema: []byte(`{"additionalProperties":false,"properties":{"items":{"items":{"type":"string"},"type":"array"}},"required":["items"],"type":"object"}`),
},
mediaType: "application/json",
encodeFunc: func(value codecNotes) ([]byte, error) {
return json.Marshal(value)
},
decodeFunc: func(content []byte) (codecNotes, error) {
var value codecNotes
return value, decodeStrictJSON(content, &value)
},
}
}
func scoreCodec() testArtifactCodec[codecScore] {
return testArtifactCodec[codecScore]{
kind: "test/score",
schema: contracts.ArtifactSchema{
ID: "score.v1",
Name: "score",
Version: "v1",
JSONSchema: []byte(`{"additionalProperties":false,"properties":{"value":{"type":"integer"}},"required":["value"],"type":"object"}`),
},
mediaType: "application/json",
encodeFunc: func(value codecScore) ([]byte, error) {
return json.Marshal(value)
},
decodeFunc: func(content []byte) (codecScore, error) {
var value codecScore
return value, decodeStrictJSON(content, &value)
},
}
}
func decodeStrictJSON(content []byte, out any) error {
decoder := json.NewDecoder(bytes.NewReader(content))
decoder.DisallowUnknownFields()
if err := decoder.Decode(out); err != nil {
return err
}
if err := decoder.Decode(&struct{}{}); err != io.EOF {
if err == nil {
return fmt.Errorf("unexpected trailing JSON value")
}
return fmt.Errorf("decode trailing JSON: %w", err)
}
return nil
}

View File

@@ -94,6 +94,7 @@ func defaultModuleCatalog(t *testing.T) pipeline.ModuleCatalog {
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -138,6 +138,7 @@ type ResolvedPipeline struct {
type ModuleCatalog struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
ArtifactCodecs *ArtifactCodecRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry

View File

@@ -1214,6 +1214,7 @@ func emptyProfileCatalog() ModuleCatalog {
return ModuleCatalog{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
ArtifactCodecs: NewArtifactCodecRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),

View File

@@ -42,12 +42,13 @@ func integrationRegistries(t *testing.T, built, executed *[]string) Registries {
t.Helper()
registries := Registries{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
Outputs: NewOutputEncoderRegistry(),
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
ArtifactCodecs: NewArtifactCodecRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
Outputs: NewOutputEncoderRegistry(),
}
if err := registries.Inputs.Register("input", func() (contracts.InputAdapter, error) {
*built = append(*built, "input")

View File

@@ -20,6 +20,7 @@ import (
type Registries struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
ArtifactCodecs *ArtifactCodecRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry

View File

@@ -1989,6 +1989,7 @@ func newRunnerRegistries(t *testing.T, modules *runnerModules) Registries {
registries := Registries{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
ArtifactCodecs: NewArtifactCodecRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),

View File

@@ -99,6 +99,7 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog {
catalog := ModuleCatalog{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
ArtifactCodecs: NewArtifactCodecRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
@@ -167,12 +168,13 @@ func walkingSkeletonRegistries(t *testing.T) Registries {
catalog := walkingSkeletonCatalog(t)
return Registries{
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
Outputs: catalog.Outputs,
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
ArtifactCodecs: catalog.ArtifactCodecs,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
Outputs: catalog.Outputs,
}
}

View File

@@ -100,6 +100,7 @@ func completeRegistries() pipeline.Registries {
return pipeline.Registries{
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),

View File

@@ -73,6 +73,7 @@ func completeRegistries() pipeline.Registries {
return pipeline.Registries{
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),

View File

@@ -215,6 +215,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -322,12 +322,13 @@ func dndSpellsRunnerRegistries(t *testing.T) pipeline.Registries {
catalog := dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{})
return pipeline.Registries{
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
Outputs: catalog.Outputs,
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
ArtifactCodecs: catalog.ArtifactCodecs,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
Outputs: catalog.Outputs,
}
}

View File

@@ -153,6 +153,7 @@ func seriatimTestCatalog(t *testing.T, inputSpec pipeline.ModuleSpec) pipeline.M
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,

View File

@@ -148,12 +148,13 @@ func seriatimRunnerRegistries(t *testing.T, extractor contracts.Extractor) pipel
}
return pipeline.Registries{
Inputs: inputs,
Chunkers: chunkers,
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Outputs: outputs,
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Outputs: outputs,
}
}