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

Finish the domain pipeline cleanup

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 11:29:52 -05:00
parent 68481804a7
commit 60b86dc40c
16 changed files with 214 additions and 1130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/framework/contracts/artifact.go
View File

@@ -41,6 +41,10 @@ type ArtifactCodec[T any] interface {
Kind() ArtifactKind
Schema() ArtifactSchema
MediaType() string
// EncodeCandidate serializes a stage result before semantic validation. It
// must not apply validity checks owned by typed validators; Encode remains
// the strict final-artifact boundary used after validation succeeds.
EncodeCandidate(T) ([]byte, error)
Encode(T) ([]byte, error)
Decode([]byte) (T, error)
}

21
internal/framework/pipeline/artifact_codec_registry.go
View File

@@ -120,19 +120,16 @@ func RegisterArtifactCodec[T any](registry *ArtifactCodecRegistry, codec contrac
return decoded, nil
},
}
entry.encodeCandidate = entry.encode
if candidate, ok := any(codec).(interface{ EncodeCandidate(T) ([]byte, error) }); ok {
entry.encodeCandidate = func(value any) ([]byte, error) {
typed, err := exactTypedValue[T]("encode candidate artifact", value)
if err != nil {
return nil, err
}
content, err := candidate.EncodeCandidate(typed)
if err != nil {
return nil, &ArtifactCodecOperationError{Operation: "encode", Kind: spec.Kind, Err: err}
}
return append([]byte(nil), content...), nil
entry.encodeCandidate = func(value any) ([]byte, error) {
typed, err := exactTypedValue[T]("encode candidate artifact", value)
if err != nil {
return nil, err
}
content, err := codec.EncodeCandidate(typed)
if err != nil {
return nil, &ArtifactCodecOperationError{Operation: "encode candidate", Kind: spec.Kind, Err: err}
}
return append([]byte(nil), content...), nil
}
if provider, ok := any(codec).(interface{ Metadata(T) map[string]any }); ok {
entry.metadata = func(value any) map[string]any {

56
internal/framework/pipeline/artifact_codec_registry_test.go
View File

@@ -24,19 +24,28 @@ type codecScore struct {
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)
kind contracts.ArtifactKind
schema contracts.ArtifactSchema
mediaType string
encodeFunc func(T) ([]byte, error)
candidateFunc 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]) EncodeCandidate(value T) ([]byte, error) {
if c.candidateFunc != nil {
return c.candidateFunc(value)
}
return c.encodeFunc(value)
}
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) }
var _ contracts.ArtifactCodec[codecNotes] = testArtifactCodec[codecNotes]{}
func TestArtifactCodecRegistryStoresHeterogeneousExactTypes(t *testing.T) {
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, notesCodec()); err != nil {
@@ -92,6 +101,43 @@ func TestArtifactCodecRegistryStoresHeterogeneousExactTypes(t *testing.T) {
}
}
func TestArtifactCodecRegistryKeepsCandidateAndFinalEncodingDistinct(t *testing.T) {
candidateCalls, finalCalls := 0, 0
codec := notesCodec()
codec.candidateFunc = func(codecNotes) ([]byte, error) {
candidateCalls++
return []byte(`{"items":["candidate"]}`), nil
}
codec.encodeFunc = func(codecNotes) ([]byte, error) {
finalCalls++
return []byte(`{"items":["final"]}`), nil
}
registry := NewArtifactCodecRegistry()
if err := RegisterArtifactCodec(registry, codec); err != nil {
t.Fatalf("RegisterArtifactCodec() error = %v, want nil", err)
}
entry, _, err := registry.entry(codec.kind)
if err != nil {
t.Fatalf("entry() error = %v, want nil", err)
}
candidate, err := serializeArtifact(entry, codecNotes{}, true)
if err != nil {
t.Fatalf("serialize candidate error = %v, want nil", err)
}
if string(candidate.Content) != `{"items":["candidate"]}` || candidateCalls != 1 || finalCalls != 0 {
t.Fatalf("candidate content = %s, calls = candidate %d, final %d", candidate.Content, candidateCalls, finalCalls)
}
final, err := serializeArtifact(entry, codecNotes{}, false)
if err != nil {
t.Fatalf("serialize final error = %v, want nil", err)
}
if string(final.Content) != `{"items":["final"]}` || candidateCalls != 1 || finalCalls != 1 {
t.Fatalf("final content = %s, calls = candidate %d, final %d", final.Content, candidateCalls, finalCalls)
}
}
func TestArtifactCodecRegistryStoresValidatedSchemaMetadata(t *testing.T) {
registry := NewArtifactCodecRegistry()
codec := notesCodec()

13
internal/framework/pipeline/debug.go
View File

@@ -409,6 +409,17 @@ type attemptTerminalRecorder struct {
envelope debugTimedEnvelope
}
type attemptDebugPersistenceError struct {
label string
err error
}
func (e *attemptDebugPersistenceError) Error() string {
return fmt.Sprintf("write %s attempt debug artifact: %v", e.label, e.err)
}
func (e *attemptDebugPersistenceError) Unwrap() error { return e.err }
func newAttemptTerminalRecorder(recorder DebugRecorder, attemptPath, label string, scope *debugLLMScope, envelope debugTimedEnvelope) attemptTerminalRecorder {
return attemptTerminalRecorder{recorder: recorder, path: attemptPath, label: label, scope: scope, envelope: envelope}
}
@@ -420,7 +431,7 @@ func (r attemptTerminalRecorder) record(payload any, terminalErr error) error {
envelope.Error = terminalErr.Error()
}
if err := writeDebugAttempt(r.recorder, r.path, envelope, r.scope); err != nil {
debugErr := fmt.Errorf("write %s attempt debug artifact: %w", r.label, err)
debugErr := &attemptDebugPersistenceError{label: r.label, err: err}
return errors.Join(terminalErr, debugErr)
}
return terminalErr

5
internal/framework/pipeline/default_modules_test.go
View File

@@ -143,7 +143,10 @@ func (defaultArtifactCodec) Kind() contracts.ArtifactKind { return defaultArtifa
func (defaultArtifactCodec) Schema() contracts.ArtifactSchema {
return contracts.ArtifactSchema{ID: "urn:notarius:test:default", Name: "default", Version: "1", JSONSchema: []byte(`{"type":"object"}`)}
}
func (defaultArtifactCodec) MediaType() string { return "application/json" }
func (defaultArtifactCodec) MediaType() string { return "application/json" }
func (defaultArtifactCodec) EncodeCandidate(value defaultArtifact) ([]byte, error) {
return json.Marshal(value)
}
func (defaultArtifactCodec) Encode(value defaultArtifact) ([]byte, error) { return json.Marshal(value) }
func (defaultArtifactCodec) Decode(content []byte) (defaultArtifact, error) {
var value defaultArtifact

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

@@ -357,6 +357,10 @@ func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool,
}
accepted, rejection, err := run(attempt)
if err != nil {
var debugErr *attemptDebugPersistenceError
if errors.As(err, &debugErr) {
return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
}
if attempt == attempts {
return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
}

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

@@ -18,6 +18,7 @@ type terminalChunker struct {
chunks []source.Chunk
warnings []contracts.Warning
err error
calls *int
}
func (c terminalChunker) Key() string { return c.key }
@@ -25,6 +26,9 @@ 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) {
if c.calls != nil {
(*c.calls)++
}
return contracts.ChunkResult{Chunks: cloneSourceChunks(c.chunks), Warnings: cloneWarnings(c.warnings)}, c.err
}
@@ -211,6 +215,30 @@ func TestRunnerJoinsPrimaryAndAttemptWriteErrors(t *testing.T) {
})
}
func TestRunnerDoesNotRetryAfterTerminalAttemptWriteFailure(t *testing.T) {
prepared, chunks := preparedTerminalDebugPipeline(t)
prepared.resolved.Chunk.Retries = 1
calls := 0
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, calls: &calls}
prepared.chunkValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("terminal/chunk-validator"), Target: ValidatorTargetChunk},
chunk: terminalChunkValidator{result: contracts.ValidationResult{Approved: true}},
}}
debug := newCapturedDebugRecorder()
debug.failPath = "chunk/attempt-01.json"
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
if err == nil || !strings.Contains(err.Error(), "write chunk attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") {
t.Fatalf("Run() error = %v, want terminal attempt debug failure", err)
}
if calls != 1 {
t.Fatalf("chunk calls = %d, want one attempt without retry", calls)
}
if debug.has("chunk/attempt-02.json") {
t.Fatal("second chunk attempt envelope exists after non-retryable debug persistence failure")
}
}
func TestRunnerKeepsExtractModuleAndValidatorLLMCallsIsolated(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
debug := newCapturedDebugRecorder()