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

Generate and materialize canonical chunk plans

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 23:57:59 +00:00
parent 3bfac14397
commit 7844c0a93f
23 changed files with 396 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
internal/framework/pipeline/runner_terminal_debug_test.go
View File

@@ -15,7 +15,7 @@ import (
type terminalChunker struct {
key string
chunks []source.Chunk
plan source.ChunkPlan
warnings []contracts.Warning
err error
calls *int
@@ -25,11 +25,11 @@ func (c terminalChunker) Key() string { return c.key }
func (terminalChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (c terminalChunker) Chunk(context.Context, contracts.ChunkRequest) (contracts.ChunkResult, error) {
func (c terminalChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
if c.calls != nil {
(*c.calls)++
}
return contracts.ChunkResult{Chunks: cloneSourceChunks(c.chunks), Warnings: cloneWarnings(c.warnings)}, c.err
return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), Warnings: cloneWarnings(c.warnings)}, c.err
}
type terminalChunkValidator struct {
@@ -37,6 +37,19 @@ type terminalChunkValidator struct {
err error
}
type observingChunkValidator struct {
request contracts.ChunkValidationRequest
}
func (*observingChunkValidator) Name() string { return "observing/chunk-validator" }
func (*observingChunkValidator) ExecutionClass() contracts.ExecutionClass {
return contracts.ExecutionClassDeterministic
}
func (v *observingChunkValidator) Validate(_ context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
v.request = request
return contracts.ValidationResult{Approved: true}, nil
}
func (terminalChunkValidator) Name() string { return "terminal/chunk-validator" }
func (terminalChunkValidator) ExecutionClass() contracts.ExecutionClass {
@@ -71,14 +84,14 @@ func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, p
}
}
func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, []source.Chunk) {
func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, source.ChunkPlan) {
t.Helper()
prepared := preparedAttemptDebugPipeline(t)
chunker, ok := prepared.chunker.(*typedTestChunker)
if !ok {
t.Fatalf("prepared chunker = %T, want *typedTestChunker", prepared.chunker)
}
return prepared, cloneSourceChunks(chunker.chunks)
return prepared, source.CloneChunkPlan(chunker.plan)
}
func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) {
@@ -96,8 +109,8 @@ func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) {
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
prepared, chunks := preparedTerminalDebugPipeline(t)
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, warnings: []contracts.Warning{{Scope: "chunk", ReasonCode: "observed", Message: "chunk warning"}}, err: tc.moduleError}
prepared, plan := preparedTerminalDebugPipeline(t)
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, warnings: []contracts.Warning{{Scope: "chunk", ReasonCode: "observed", Message: "chunk warning"}}, err: tc.moduleError}
prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding("terminal/chunk-validator"), Target: ValidatorTargetChunk}, chunk: tc.validator}}
debug := newCapturedDebugRecorder()
@@ -116,10 +129,63 @@ func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) {
t.Fatalf("chunk rejection = envelope %#v, outputs %#v", envelope, output.Rejected)
}
}
if tc.name == "accepted" {
encoded := string(debug.json["chunk/attempt-01.json"])
if !strings.Contains(encoded, `"plan":`) || !strings.Contains(encoded, `"materialized_chunks":`) || strings.Contains(encoded, `"chunks":`) {
t.Fatalf("chunk attempt debug = %s, want separate plan and materialized_chunks", encoded)
}
}
})
}
}
func TestRunnerMaterializesAnnotatedPlanBeforeChunkValidation(t *testing.T) {
prepared, plan := preparedTerminalDebugPipeline(t)
plan.Annotations = source.ChunkAnnotations{"same": []byte(`{"plan":1}`)}
plan.Ranges[0].Annotations = source.ChunkAnnotations{"same": []byte(`{"range":2}`)}
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan}
validator := &observingChunkValidator{}
prepared.chunkValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk},
chunk: validator,
}}
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if len(validator.request.Chunks) != 1 {
t.Fatalf("validator chunks = %#v, want one", validator.request.Chunks)
}
chunk := validator.request.Chunks[0]
if chunk.ID != "chunk-000001" || chunk.Index != 0 || chunk.MediaType != "application/json" {
t.Fatalf("materialized chunk identity = %#v", chunk)
}
if string(chunk.Annotations["same"]) != `{"range":2}` || string(chunk.PlanAnnotations["same"]) != `{"plan":1}` {
t.Fatalf("materialized annotations = %s / %s", chunk.Annotations["same"], chunk.PlanAnnotations["same"])
}
if chunk.Metadata["start_unit_id"] != 1 || chunk.Metadata["end_unit_id"] != 1 || chunk.Metadata["unit_count"] != 1 {
t.Fatalf("materialized metadata = %#v", chunk.Metadata)
}
}
func TestRunnerRetriesMalformedPlanWithDebugEnabled(t *testing.T) {
prepared, plan := preparedTerminalDebugPipeline(t)
plan.Ranges[0].Annotations = source.ChunkAnnotations{"broken": []byte(`{"value":`)}
prepared.resolved.Chunk.Retries = 1
calls := 0
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}
debug := newCapturedDebugRecorder()
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
if err == nil || !strings.Contains(err.Error(), "valid JSON") {
t.Fatalf("Run() error = %v, want malformed annotation error", err)
}
if calls != 2 {
t.Fatalf("plan calls = %d, want two attempts", calls)
}
assertAttemptEnvelopeSequence(t, debug, "chunk", 1, 2)
}
func TestRunnerRecordsExtractTerminalOutcomes(t *testing.T) {
tests := []struct {
name string
@@ -187,9 +253,9 @@ func TestRunnerRecordsExtractTerminalOutcomes(t *testing.T) {
func TestRunnerJoinsPrimaryAndAttemptWriteErrors(t *testing.T) {
t.Run("chunk", func(t *testing.T) {
prepared, chunks := preparedTerminalDebugPipeline(t)
prepared, plan := preparedTerminalDebugPipeline(t)
primary := errors.New("chunk operation failed")
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, err: primary}
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, err: primary}
debug := newCapturedDebugRecorder()
debug.failPath = "chunk/attempt-01.json"
@@ -216,10 +282,10 @@ func TestRunnerJoinsPrimaryAndAttemptWriteErrors(t *testing.T) {
}
func TestRunnerDoesNotRetryAfterTerminalAttemptWriteFailure(t *testing.T) {
prepared, chunks := preparedTerminalDebugPipeline(t)
prepared, plan := preparedTerminalDebugPipeline(t)
prepared.resolved.Chunk.Retries = 1
calls := 0
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, calls: &calls}
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}
prepared.chunkValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("terminal/chunk-validator"), Target: ValidatorTargetChunk},
chunk: terminalChunkValidator{result: contracts.ValidationResult{Approved: true}},