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

Deep-clone source metadata during chunk materialization

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 02:10:05 +00:00
parent 205e2a9908
commit 561d65a505
6 changed files with 279 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -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

53
internal/core/source/chunk_plan.go
View File

@@ -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
}

86
internal/core/source/chunk_plan_test.go
View File

@@ -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} {

119
internal/core/source/metadata.go Normal file
View File

@@ -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
}

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

@@ -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 {

58
internal/framework/pipeline/runner_extract_handoff_test.go
View File

@@ -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