diff --git a/docs/internal/overview.md b/docs/internal/overview.md index bbb8583..f7bf3ed 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -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 diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index a03ea54..2b93fd3 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -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, diff --git a/internal/cli/catalog.go b/internal/cli/catalog.go index 22535c8..dbaf6c7 100644 --- a/internal/cli/catalog.go +++ b/internal/cli/catalog.go @@ -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 && diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index c1d9909..3b4f3e7 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -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, diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index 0ae855b..eb68d03 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -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, diff --git a/internal/framework/contracts/artifact.go b/internal/framework/contracts/artifact.go new file mode 100644 index 0000000..0ba3873 --- /dev/null +++ b/internal/framework/contracts/artifact.go @@ -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 +} diff --git a/internal/framework/contracts/artifact_test.go b/internal/framework/contracts/artifact_test.go new file mode 100644 index 0000000..657b984 --- /dev/null +++ b/internal/framework/contracts/artifact_test.go @@ -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) + } +} diff --git a/internal/framework/pipeline/artifact_codec_registry.go b/internal/framework/pipeline/artifact_codec_registry.go new file mode 100644 index 0000000..3d3015a --- /dev/null +++ b/internal/framework/pipeline/artifact_codec_registry.go @@ -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 := "" + if expected != nil { + expectedName = expected.String() + } + actualName := "" + if actual != nil { + actualName = actual.String() + } + return &ArtifactCodecTypeError{ + Operation: operation, + Kind: kind, + ExpectedType: expectedName, + ActualType: actualName, + } +} diff --git a/internal/framework/pipeline/artifact_codec_registry_test.go b/internal/framework/pipeline/artifact_codec_registry_test.go new file mode 100644 index 0000000..4bced7b --- /dev/null +++ b/internal/framework/pipeline/artifact_codec_registry_test.go @@ -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 +} diff --git a/internal/framework/pipeline/default_modules_test.go b/internal/framework/pipeline/default_modules_test.go index 6ed9022..dd44856 100644 --- a/internal/framework/pipeline/default_modules_test.go +++ b/internal/framework/pipeline/default_modules_test.go @@ -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, diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index e9c8454..6d68818 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -138,6 +138,7 @@ type ResolvedPipeline struct { type ModuleCatalog struct { Inputs *InputAdapterRegistry Chunkers *ChunkerRegistry + ArtifactCodecs *ArtifactCodecRegistry Extractors *ExtractorRegistry Mergers *MergerRegistry Normalizers *NormalizerRegistry diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 9570b00..b69c391 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -1214,6 +1214,7 @@ func emptyProfileCatalog() ModuleCatalog { return ModuleCatalog{ Inputs: NewInputAdapterRegistry(), Chunkers: NewChunkerRegistry(), + ArtifactCodecs: NewArtifactCodecRegistry(), Extractors: NewExtractorRegistry(), Mergers: NewMergerRegistry(), Normalizers: NewNormalizerRegistry(), diff --git a/internal/framework/pipeline/registry_integration_test.go b/internal/framework/pipeline/registry_integration_test.go index 48114f8..bb2f838 100644 --- a/internal/framework/pipeline/registry_integration_test.go +++ b/internal/framework/pipeline/registry_integration_test.go @@ -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") diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index c7b35e0..f35e55a 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -20,6 +20,7 @@ import ( type Registries struct { Inputs *InputAdapterRegistry Chunkers *ChunkerRegistry + ArtifactCodecs *ArtifactCodecRegistry Extractors *ExtractorRegistry Mergers *MergerRegistry Normalizers *NormalizerRegistry diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 481564a..f3fcd65 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -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(), diff --git a/internal/framework/pipeline/walking_skeleton_test.go b/internal/framework/pipeline/walking_skeleton_test.go index 1cadfdc..87998c5 100644 --- a/internal/framework/pipeline/walking_skeleton_test.go +++ b/internal/framework/pipeline/walking_skeleton_test.go @@ -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, } } diff --git a/internal/modules/dnd/register/register_test.go b/internal/modules/dnd/register/register_test.go index 23e0a3f..79ff5dd 100644 --- a/internal/modules/dnd/register/register_test.go +++ b/internal/modules/dnd/register/register_test.go @@ -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(), diff --git a/internal/modules/generic/register/register_test.go b/internal/modules/generic/register/register_test.go index bf796dd..93035d1 100644 --- a/internal/modules/generic/register/register_test.go +++ b/internal/modules/generic/register/register_test.go @@ -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(), diff --git a/internal/modules/integration/dnd_spells_config_test.go b/internal/modules/integration/dnd_spells_config_test.go index a481702..9af3bcd 100644 --- a/internal/modules/integration/dnd_spells_config_test.go +++ b/internal/modules/integration/dnd_spells_config_test.go @@ -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, diff --git a/internal/modules/integration/dnd_spells_runner_test.go b/internal/modules/integration/dnd_spells_runner_test.go index ba858ee..c802103 100644 --- a/internal/modules/integration/dnd_spells_runner_test.go +++ b/internal/modules/integration/dnd_spells_runner_test.go @@ -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, } } diff --git a/internal/modules/seriatim/input/transcript/config_test.go b/internal/modules/seriatim/input/transcript/config_test.go index 3405fa6..0851acd 100644 --- a/internal/modules/seriatim/input/transcript/config_test.go +++ b/internal/modules/seriatim/input/transcript/config_test.go @@ -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, diff --git a/internal/modules/seriatim/input/transcript/runner_test.go b/internal/modules/seriatim/input/transcript/runner_test.go index 6c2e81e..bdccf22 100644 --- a/internal/modules/seriatim/input/transcript/runner_test.go +++ b/internal/modules/seriatim/input/transcript/runner_test.go @@ -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, } }