From 561d65a505c9c2df0df4407d43fc9686e211007d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 18 Jul 2026 02:10:05 +0000 Subject: [PATCH] Deep-clone source metadata during chunk materialization --- docs/roadmap/implementation.md | 2 +- internal/core/source/chunk_plan.go | 53 ++------ internal/core/source/chunk_plan_test.go | 86 +++++++++++++ internal/core/source/metadata.go | 119 ++++++++++++++++++ internal/framework/pipeline/runner.go | 38 +----- .../pipeline/runner_extract_handoff_test.go | 58 +++++++++ 6 files changed, 279 insertions(+), 77 deletions(-) create mode 100644 internal/core/source/metadata.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e2b6fc5..279d8bb 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -469,7 +469,7 @@ concurrency tests and the race detector pass, and the repository is green. ## Stage 11: Deep-clone source metadata during materialization -**Status:** Pending +**Status:** Complete ### Objective diff --git a/internal/core/source/chunk_plan.go b/internal/core/source/chunk_plan.go index 5150c1e..ac4a6c9 100644 --- a/internal/core/source/chunk_plan.go +++ b/internal/core/source/chunk_plan.go @@ -162,7 +162,10 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error) for index, chunkRange := range plan.Ranges { start, _ := UnitIndex(doc, chunkRange.StartUnitID) end, _ := UnitIndex(doc, chunkRange.EndUnitID) - units := cloneSourceUnits(doc.Units[start : end+1]) + units, err := cloneSourceUnits(doc.Units[start : end+1]) + if err != nil { + return nil, fmt.Errorf("clone chunk plan range[%d] units: %w", index, err) + } content, err := json.Marshal(struct { Units []SourceUnit `json:"units"` }{Units: units}) @@ -189,52 +192,18 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error) return chunks, nil } -func cloneSourceUnits(units []SourceUnit) []SourceUnit { +func cloneSourceUnits(units []SourceUnit) ([]SourceUnit, error) { if len(units) == 0 { - return nil + return nil, nil } cloned := make([]SourceUnit, len(units)) for i, unit := range units { cloned[i] = unit - cloned[i].Metadata = cloneJSONMap(unit.Metadata) - } - return cloned -} - -func cloneJSONMap(values map[string]any) map[string]any { - if len(values) == 0 { - return nil - } - cloned := make(map[string]any, len(values)) - for key, value := range values { - cloned[key] = cloneJSONValue(value) - } - return cloned -} - -func cloneJSONValue(value any) any { - switch typed := value.(type) { - case map[string]any: - return cloneJSONMap(typed) - case []any: - cloned := make([]any, len(typed)) - for i := range typed { - cloned[i] = cloneJSONValue(typed[i]) + metadata, err := CloneMetadata(unit.Metadata) + if err != nil { + return nil, fmt.Errorf("source unit[%d] metadata: %w", i, err) } - return cloned - case json.RawMessage: - return append(json.RawMessage(nil), typed...) - case []byte: - return append([]byte(nil), typed...) - case []string: - return append([]string(nil), typed...) - case map[string]string: - cloned := make(map[string]string, len(typed)) - for key, item := range typed { - cloned[key] = item - } - return cloned - default: - return value + cloned[i].Metadata = metadata } + return cloned, nil } diff --git a/internal/core/source/chunk_plan_test.go b/internal/core/source/chunk_plan_test.go index beaf600..e696239 100644 --- a/internal/core/source/chunk_plan_test.go +++ b/internal/core/source/chunk_plan_test.go @@ -3,11 +3,16 @@ package source import ( "bytes" "encoding/json" + "math" "reflect" "strings" "testing" ) +type typedMetadataMap map[string]any +type typedMetadataSlice []typedMetadataMap +type typedMetadataArray [2]any + func TestCanonicalizeChunkAnnotations(t *testing.T) { original := ChunkAnnotations{ "domain/items": json.RawMessage(` { "z": [3, 2, 1], "a": 1.0 } `), @@ -218,6 +223,87 @@ func TestMaterializeChunkPlanDeepClonesUnitMetadata(t *testing.T) { } } +func TestMaterializeChunkPlanClonesConcreteJSONMetadata(t *testing.T) { + doc := planDocument() + doc.Units[0].Metadata = map[string]any{ + "typed_map": typedMetadataMap{"bytes": []byte("map")}, + "typed_slice": typedMetadataSlice{{"raw": json.RawMessage(`{"slice":true}`)}}, + "typed_array": typedMetadataArray{map[string]any{"bytes": []byte("array")}, []any{json.RawMessage(`{"array":true}`)}}, + "interface": any(typedMetadataMap{"bytes": []byte("interface")}), + "raw": json.RawMessage(`{"raw":true}`), + "bytes": []byte("bytes"), + } + + chunks, err := MaterializeChunkPlan(doc, validChunkPlan(doc)) + if err != nil { + t.Fatal(err) + } + again, err := MaterializeChunkPlan(doc, validChunkPlan(doc)) + if err != nil { + t.Fatal(err) + } + metadata := chunks[0].Units[0].Metadata + metadata["typed_map"].(typedMetadataMap)["bytes"].([]byte)[0] = 'M' + metadata["typed_slice"].(typedMetadataSlice)[0]["raw"].(json.RawMessage)[0] = '[' + metadata["typed_array"].(typedMetadataArray)[0].(map[string]any)["bytes"].([]byte)[0] = 'A' + metadata["typed_array"].(typedMetadataArray)[1].([]any)[0].(json.RawMessage)[0] = '[' + metadata["interface"].(typedMetadataMap)["bytes"].([]byte)[0] = 'I' + metadata["raw"].(json.RawMessage)[0] = '[' + metadata["bytes"].([]byte)[0] = 'B' + + for name, candidate := range map[string]map[string]any{ + "source": doc.Units[0].Metadata, + "again": again[0].Units[0].Metadata, + } { + if got := string(candidate["typed_map"].(typedMetadataMap)["bytes"].([]byte)); got != "map" { + t.Fatalf("%s typed map bytes = %q, want map", name, got) + } + if got := string(candidate["typed_slice"].(typedMetadataSlice)[0]["raw"].(json.RawMessage)); got != `{"slice":true}` { + t.Fatalf("%s typed slice raw = %q", name, got) + } + array := candidate["typed_array"].(typedMetadataArray) + if got := string(array[0].(map[string]any)["bytes"].([]byte)); got != "array" || string(array[1].([]any)[0].(json.RawMessage)) != `{"array":true}` { + t.Fatalf("%s typed array = %#v", name, array) + } + if got := string(candidate["interface"].(typedMetadataMap)["bytes"].([]byte)); got != "interface" { + t.Fatalf("%s interface bytes = %q, want interface", name, got) + } + if got := string(candidate["raw"].(json.RawMessage)); got != `{"raw":true}` { + t.Fatalf("%s raw = %q", name, got) + } + if got := string(candidate["bytes"].([]byte)); got != "bytes" { + t.Fatalf("%s bytes = %q, want bytes", name, got) + } + } +} + +func TestMaterializeChunkPlanRejectsInvalidMetadata(t *testing.T) { + cyclic := make(map[string]any) + cyclic["self"] = cyclic + for _, tc := range []struct { + name string + value any + want string + }{ + {name: "cycle", value: cyclic, want: "metadata.cycle.self contains a cycle"}, + {name: "unsupported", value: func() {}, want: "metadata.unsupported has unsupported type func()"}, + {name: "nonfinite", value: math.NaN(), want: "metadata.nonfinite has a non-finite number"}, + } { + t.Run(tc.name, func(t *testing.T) { + doc := planDocument() + doc.Units[0].Metadata = map[string]any{tc.name: tc.value} + _, first := MaterializeChunkPlan(doc, validChunkPlan(doc)) + _, second := MaterializeChunkPlan(doc, validChunkPlan(doc)) + if first == nil || !strings.Contains(first.Error(), "clone chunk plan range[0] units: source unit[0] metadata: "+tc.want) { + t.Fatalf("first MaterializeChunkPlan() error = %v, want %q", first, tc.want) + } + if second == nil || second.Error() != first.Error() { + t.Fatalf("MaterializeChunkPlan() errors = %v and %v, want deterministic error", first, second) + } + }) + } +} + func planDocument() *SourceDocument { doc := &SourceDocument{ID: "source-plan", Kind: "test", Format: "application/test", Digest: "sha256:source-plan"} for _, id := range []int{10, 20, 30, 40, 50} { diff --git a/internal/core/source/metadata.go b/internal/core/source/metadata.go new file mode 100644 index 0000000..cb01c7f --- /dev/null +++ b/internal/core/source/metadata.go @@ -0,0 +1,119 @@ +package source + +import ( + "fmt" + "math" + "reflect" + "sort" +) + +// CloneMetadata returns an independently owned copy of JSON-shaped metadata. +// It preserves concrete map, slice, and array types while rejecting values that +// cannot be safely represented as JSON-shaped metadata. +func CloneMetadata(metadata map[string]any) (map[string]any, error) { + if len(metadata) == 0 { + return nil, nil + } + cloned, err := cloneMetadataValue(reflect.ValueOf(metadata), "metadata", make(map[metadataVisit]struct{})) + if err != nil { + return nil, err + } + return cloned.Interface().(map[string]any), nil +} + +type metadataVisit struct { + typ reflect.Type + ptr uintptr +} + +func cloneMetadataValue(value reflect.Value, location string, active map[metadataVisit]struct{}) (reflect.Value, error) { + if !value.IsValid() { + return value, nil + } + + switch value.Kind() { + case reflect.Interface: + if value.IsNil() { + return reflect.Zero(value.Type()), nil + } + cloned, err := cloneMetadataValue(value.Elem(), location, active) + if err != nil { + return reflect.Value{}, err + } + result := reflect.New(value.Type()).Elem() + result.Set(cloned) + return result, nil + case reflect.Map: + if value.IsNil() { + return reflect.Zero(value.Type()), nil + } + if value.Type().Key().Kind() != reflect.String { + return reflect.Value{}, fmt.Errorf("%s has unsupported map key type %s", location, value.Type().Key()) + } + leave, err := enterMetadataValue(value, active, location) + if err != nil { + return reflect.Value{}, err + } + defer leave() + + keys := value.MapKeys() + sort.Slice(keys, func(i, j int) bool { return keys[i].String() < keys[j].String() }) + result := reflect.MakeMapWithSize(value.Type(), value.Len()) + for _, key := range keys { + cloned, err := cloneMetadataValue(value.MapIndex(key), location+"."+key.String(), active) + if err != nil { + return reflect.Value{}, err + } + result.SetMapIndex(key, cloned) + } + return result, nil + case reflect.Slice: + if value.IsNil() { + return reflect.Zero(value.Type()), nil + } + leave, err := enterMetadataValue(value, active, location) + if err != nil { + return reflect.Value{}, err + } + defer leave() + + result := reflect.MakeSlice(value.Type(), value.Len(), value.Len()) + for i := 0; i < value.Len(); i++ { + cloned, err := cloneMetadataValue(value.Index(i), fmt.Sprintf("%s[%d]", location, i), active) + if err != nil { + return reflect.Value{}, err + } + result.Index(i).Set(cloned) + } + return result, nil + case reflect.Array: + result := reflect.New(value.Type()).Elem() + for i := 0; i < value.Len(); i++ { + cloned, err := cloneMetadataValue(value.Index(i), fmt.Sprintf("%s[%d]", location, i), active) + if err != nil { + return reflect.Value{}, err + } + result.Index(i).Set(cloned) + } + return result, nil + case reflect.Float32, reflect.Float64: + if math.IsNaN(value.Float()) || math.IsInf(value.Float(), 0) { + return reflect.Value{}, fmt.Errorf("%s has a non-finite number", location) + } + return value, nil + case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, + reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.String: + return value, nil + default: + return reflect.Value{}, fmt.Errorf("%s has unsupported type %s", location, value.Type()) + } +} + +func enterMetadataValue(value reflect.Value, active map[metadataVisit]struct{}, location string) (func(), error) { + visit := metadataVisit{typ: value.Type(), ptr: value.Pointer()} + if _, exists := active[visit]; exists { + return nil, fmt.Errorf("%s contains a cycle", location) + } + active[visit] = struct{}{} + return func() { delete(active, visit) }, nil +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index fd42487..6f541f9 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -679,41 +679,11 @@ func validateOutputFileName(name string) error { } func cloneMetadata(metadata map[string]any) map[string]any { - if len(metadata) == 0 { - return nil - } - out := make(map[string]any, len(metadata)) - for key, value := range metadata { - out[key] = cloneMetadataValue(value) - } - return out -} - -func cloneMetadataValue(value any) any { - switch typed := value.(type) { - case map[string]any: - return cloneMetadata(typed) - case []any: - out := make([]any, len(typed)) - for i := range typed { - out[i] = cloneMetadataValue(typed[i]) - } - return out - case json.RawMessage: - return append(json.RawMessage(nil), typed...) - case []byte: - return append([]byte(nil), typed...) - case []string: - return append([]string(nil), typed...) - case map[string]string: - out := make(map[string]string, len(typed)) - for key, item := range typed { - out[key] = item - } - return out - default: - return value + cloned, err := source.CloneMetadata(metadata) + if err != nil { + return metadata } + return cloned } func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest { diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index 57c7491..b6e0839 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -16,6 +16,24 @@ type extractCaptureRecorder struct { checkpoint ExtractCheckpoint } +type pipelineTypedMetadata map[string]any + +type mutatingMetadataValidator struct { + seen string +} + +func (*mutatingMetadataValidator) Name() string { return "test/mutating-metadata" } +func (*mutatingMetadataValidator) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} +func (v *mutatingMetadataValidator) Validate(_ context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + nested := request.Chunks[0].Units[0].Metadata["typed"].(pipelineTypedMetadata) + v.seen = string(nested["raw"].(json.RawMessage)) + nested["raw"].(json.RawMessage)[0] = '[' + nested["changed"] = true + return contracts.ValidationResult{Approved: true}, nil +} + func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { r.checkpoint = ExtractCheckpoint{ Outputs: cloneCheckpointArtifacts(outputs), @@ -77,6 +95,46 @@ func TestRunnerPassesIndependentAnnotationScopesToExtractors(t *testing.T) { } } +func TestRunnerPassesIndependentConcreteMetadataToValidatorsAndExtractors(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + input := prepared.input.(*typedTestInput) + input.doc.Units[0].Metadata = map[string]any{ + "typed": pipelineTypedMetadata{"raw": json.RawMessage(`{"value":true}`)}, + } + var err error + input.doc.Digest, err = source.DigestDocument(input.doc) + if err != nil { + t.Fatal(err) + } + chunker := prepared.chunker.(*typedTestChunker) + chunker.plan = typedTestPlan(input.doc) + + validator := &mutatingMetadataValidator{} + prepared.chunkValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, + chunk: validator, + }} + var extractorSeen string + installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + nested := request.Chunk.Units[0].Metadata["typed"].(pipelineTypedMetadata) + extractorSeen = string(nested["raw"].(json.RawMessage)) + nested["raw"].(json.RawMessage)[0] = '[' + nested["changed"] = true + return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil + }) + + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil { + t.Fatalf("Run() error = %v", err) + } + if validator.seen != `{"value":true}` || extractorSeen != `{"value":true}` { + t.Fatalf("validator/extractor metadata = %q / %q", validator.seen, extractorSeen) + } + original := input.doc.Units[0].Metadata["typed"].(pipelineTypedMetadata) + if _, changed := original["changed"]; changed || string(original["raw"].(json.RawMessage)) != `{"value":true}` { + t.Fatalf("source metadata changed through runner handoff: %#v", original) + } +} + func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { prepared := preparedAttemptDebugPipeline(t) extractCalls := 0