Compare commits
3 Commits
3772b308e9
...
68481804a7
| Author | SHA1 | Date | |
|---|---|---|---|
| 68481804a7 | |||
| 236ccc62ad | |||
| 35bffdf336 |
@@ -181,8 +181,9 @@ payload rules are defined in the
|
||||
|
||||
## Production Registration
|
||||
|
||||
The CLI allocates one complete framework registry set and one LLM asset
|
||||
registry. It invokes `internal/modules/generic/register`,
|
||||
Production composition occurs through family registrars. The CLI allocates one
|
||||
complete framework registry set and one LLM asset registry. It invokes
|
||||
`internal/modules/generic/register`,
|
||||
`internal/modules/seriatim/register`, and `internal/modules/dnd/register` in
|
||||
that order, then exposes the matching catalog for resolution. The generic and
|
||||
Seriatim registrars own their production leaf registrations. The D&D registrar
|
||||
@@ -194,8 +195,13 @@ packages directly. A concrete family's `register` package is its composition
|
||||
point for specializing reusable generic implementations, while the generic
|
||||
registrar composes only generic children.
|
||||
|
||||
Framework packages must not import production extensions. Tests may compose
|
||||
registries and catalogs directly with fakes.
|
||||
Core and framework production packages do not import production extensions.
|
||||
CLI production code imports only exact family registrar packages. Compatibility
|
||||
tests in the CLI, core, and framework trees may import roots and implementation
|
||||
leaves directly. White-box tests within module families retain the production
|
||||
family boundaries. `internal/modules/integration` is test infrastructure: its
|
||||
black-box tests may compose multiple families, but it is not a production
|
||||
module family or production dependency target.
|
||||
|
||||
## Adding An Extension
|
||||
|
||||
|
||||
@@ -228,14 +228,26 @@ normally. Dependency fingerprints and debug content digests use the same stable
|
||||
codec bytes that cross those boundaries.
|
||||
|
||||
Debug instrumentation wraps run, stage, attempt, validator, and structured LLM
|
||||
boundaries. Every executed module retry has an attempt envelope containing its
|
||||
candidate, accepted-attempt warnings, rejection or error, and only the LLM
|
||||
calls made by that module attempt. Validator attempts retain independent scopes
|
||||
under `validate/`. Debug-write failures are framework errors; debug data is
|
||||
boundaries. Every executed chunk, extract, merge, and normalize attempt writes
|
||||
one terminal envelope for acceptance, validator rejection, module or validator
|
||||
error, or applicable candidate or final serialization error. The envelope
|
||||
contains its attempt-local warnings, any available candidate and rejection,
|
||||
and terminal error text; failures before a candidate exists omit that payload.
|
||||
Only LLM calls made by the module operation belong to the module attempt.
|
||||
Validator calls retain independent scopes under `validate/` and are not
|
||||
duplicated into the module envelope. A failed terminal-envelope write is a
|
||||
framework error and is joined with any primary attempt error. Debug data is
|
||||
never used as a checkpoint source. Typed artifact debug envelopes are
|
||||
domain-neutral, redact sensitive metadata and bytes through the common debug
|
||||
policy, and record codec identity plus schema and content digests.
|
||||
|
||||
Merge and normalize attempts serialize their in-memory candidate with the
|
||||
codec's candidate encoder before typed validation. Serialized validators and
|
||||
attempt debug use that candidate representation, which carries the codec media
|
||||
type and schema identity but is never checkpointed or passed downstream. Only
|
||||
a validator-approved value is encoded through the strict final codec and made
|
||||
eligible for a checkpoint or stage output.
|
||||
|
||||
Checkpoint identity, physical layout, reuse behavior, and debug artifact
|
||||
handling are operator contracts in [Operations](../operations.md). Serialization
|
||||
and recorder implementation are inventoried in
|
||||
|
||||
@@ -128,15 +128,17 @@ checkpointing does not write debug output.
|
||||
Debug artifacts include inputs and outputs for source, chunk, extract, merge,
|
||||
normalize, and output work, structured LLM request and response data, validator
|
||||
requests and results, timing, and retry attempt metadata. LLM calls made inside
|
||||
a retry or validator attempt
|
||||
write `prompt-000N.json`, `response-000N.json`, and
|
||||
`response-content-000N.*` files under that attempt directory and are linked from
|
||||
the attempt `llm_calls` array. Prompt content is written inline in the prompt
|
||||
artifact. The response metadata and body use the paired files described above;
|
||||
the body is pretty-printed JSON when possible and raw text otherwise. Merge and
|
||||
normalize retries use these stable paths:
|
||||
a module retry write `prompt-000N.json`, `response-000N.json`, and
|
||||
`response-content-000N.*` files under that attempt directory and are linked
|
||||
from its `llm_calls` array. Validator calls use separate attempt scopes under
|
||||
`validate/` and are not duplicated into the module attempt. Prompt content is
|
||||
written inline in the prompt artifact. The response metadata and body use the
|
||||
paired files described above; the body is pretty-printed JSON when possible
|
||||
and raw text otherwise. Retrying stages use these stable module-attempt paths:
|
||||
|
||||
```text
|
||||
chunk/attempt-<NN>.json
|
||||
extract/<lane-id>/chunk-<NNNNNN>/attempt-<NN>.json
|
||||
merge/<lane-id>/attempt-<NN>.json
|
||||
merge/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
|
||||
merge/<lane-id>/attempt-<NN>/response-<NNNN>.json
|
||||
@@ -148,10 +150,19 @@ normalize/<lane-id>/attempt-<NN>/response-<NNNN>.json
|
||||
normalize/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
|
||||
```
|
||||
|
||||
Checkpoint-reused merge and normalize work retains the stage-level input and
|
||||
output artifacts but has no retry-attempt artifacts because no module attempt
|
||||
executed. Debug artifacts may contain source material, reference material,
|
||||
prompt inputs, model outputs, and other sensitive data. Typed artifact
|
||||
Every executed chunk, extract, merge, and normalize attempt has one terminal
|
||||
envelope recording acceptance, validator rejection, or a module, validator,
|
||||
candidate-serialization, or final-serialization error as applicable. It
|
||||
includes attempt-local warnings and any available candidate or rejection. A
|
||||
failure before a candidate exists has no candidate payload. If the envelope
|
||||
cannot be persisted, the run reports that debug failure together with any
|
||||
primary attempt error.
|
||||
|
||||
Checkpoint-reused chunk, extract, merge, and normalize work retains the
|
||||
stage-level input and output artifacts but has no retry-attempt artifacts
|
||||
because no module attempt executed. Debug artifacts may contain source
|
||||
material, reference material, prompt inputs, model outputs, and other sensitive
|
||||
data. Typed artifact
|
||||
envelopes include domain-neutral codec identity, redacted metadata and content,
|
||||
and digests of the stable codec bytes. API keys are not written, and obvious
|
||||
credential-shaped values and sensitive map keys are redacted, but debug
|
||||
|
||||
@@ -401,6 +401,31 @@ func writeDebugAttempt(recorder DebugRecorder, attemptPath string, envelope debu
|
||||
return writeDebugTimed(recorder, attemptPath+".json", debugEnvelopeWithLLMCalls(envelope, scope))
|
||||
}
|
||||
|
||||
type attemptTerminalRecorder struct {
|
||||
recorder DebugRecorder
|
||||
path string
|
||||
label string
|
||||
scope *debugLLMScope
|
||||
envelope debugTimedEnvelope
|
||||
}
|
||||
|
||||
func newAttemptTerminalRecorder(recorder DebugRecorder, attemptPath, label string, scope *debugLLMScope, envelope debugTimedEnvelope) attemptTerminalRecorder {
|
||||
return attemptTerminalRecorder{recorder: recorder, path: attemptPath, label: label, scope: scope, envelope: envelope}
|
||||
}
|
||||
|
||||
func (r attemptTerminalRecorder) record(payload any, terminalErr error) error {
|
||||
envelope := r.envelope
|
||||
envelope.Payload = payload
|
||||
if terminalErr != nil {
|
||||
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)
|
||||
return errors.Join(terminalErr, debugErr)
|
||||
}
|
||||
return terminalErr
|
||||
}
|
||||
|
||||
func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope {
|
||||
content = redactSecretBytes(content)
|
||||
return debugBinaryEnvelope{
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"mime"
|
||||
"path"
|
||||
@@ -207,6 +208,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
|
||||
attemptStarted := time.Now().UTC()
|
||||
attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
|
||||
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
|
||||
terminal := newAttemptTerminalRecorder(debugRecorder, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
|
||||
chunkResult, err := chunker.Chunk(attemptCtx, contracts.ChunkRequest{
|
||||
Source: doc,
|
||||
SourceInput: sourceInput.Clone(),
|
||||
@@ -216,68 +218,32 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
|
||||
Metadata: input.Metadata,
|
||||
})
|
||||
if err != nil {
|
||||
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
|
||||
Stage: string(StageChunk),
|
||||
ModuleKey: chunker.Key(),
|
||||
Attempt: attempt,
|
||||
StartedAt: attemptStarted,
|
||||
Error: err.Error(),
|
||||
}, llmScope)
|
||||
return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
|
||||
attemptErr := fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
|
||||
return false, nil, terminal.record(nil, attemptErr)
|
||||
}
|
||||
if len(chunkResult.Chunks) == 0 {
|
||||
err := fmt.Errorf("chunker %q returned no chunks", chunker.Key())
|
||||
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
|
||||
Stage: string(StageChunk),
|
||||
ModuleKey: chunker.Key(),
|
||||
Attempt: attempt,
|
||||
StartedAt: attemptStarted,
|
||||
Error: err.Error(),
|
||||
}, llmScope)
|
||||
return false, nil, err
|
||||
attemptErr := fmt.Errorf("chunker %q returned no chunks", chunker.Key())
|
||||
return false, nil, terminal.record(nil, attemptErr)
|
||||
}
|
||||
chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks)
|
||||
if err != nil {
|
||||
err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
|
||||
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
|
||||
Stage: string(StageChunk),
|
||||
ModuleKey: chunker.Key(),
|
||||
Attempt: attempt,
|
||||
StartedAt: attemptStarted,
|
||||
Payload: map[string]any{
|
||||
"warnings": cloneWarnings(chunkResult.Warnings),
|
||||
},
|
||||
Error: err.Error(),
|
||||
}, llmScope)
|
||||
return false, nil, err
|
||||
attemptErr := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
|
||||
payload := map[string]any{"warnings": debugWarningEnvelopes(chunkResult.Warnings)}
|
||||
return false, nil, terminal.record(payload, attemptErr)
|
||||
}
|
||||
validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
|
||||
attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
|
||||
payload := map[string]any{
|
||||
"chunks": debugSourceChunkEnvelopes(chunks),
|
||||
"warnings": debugWarningEnvelopes(attemptWarnings),
|
||||
"rejection": debugRejectedOutputPtr(rejection),
|
||||
}
|
||||
if err != nil || rejection != nil {
|
||||
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
|
||||
Stage: string(StageChunk),
|
||||
ModuleKey: chunker.Key(),
|
||||
Attempt: attempt,
|
||||
StartedAt: attemptStarted,
|
||||
Payload: map[string]any{
|
||||
"chunks": debugSourceChunkEnvelopes(chunks),
|
||||
"warnings": append(cloneWarnings(chunkResult.Warnings), validationWarnings...),
|
||||
"rejection": debugRejectedOutputPtr(rejection),
|
||||
},
|
||||
}, llmScope)
|
||||
return false, rejection, err
|
||||
return false, rejection, terminal.record(payload, err)
|
||||
}
|
||||
canonicalChunks = chunks
|
||||
chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
|
||||
if err := writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
|
||||
Stage: string(StageChunk),
|
||||
ModuleKey: chunker.Key(),
|
||||
Attempt: attempt,
|
||||
StartedAt: attemptStarted,
|
||||
Payload: map[string]any{
|
||||
"chunks": debugSourceChunkEnvelopes(chunks),
|
||||
"warnings": chunkWarnings,
|
||||
},
|
||||
}, llmScope); err != nil {
|
||||
chunkWarnings = attemptWarnings
|
||||
if err := terminal.record(payload, nil); err != nil {
|
||||
return false, nil, err
|
||||
}
|
||||
return true, nil, nil
|
||||
@@ -446,11 +412,15 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
|
||||
if err != nil {
|
||||
debugCall.Error = err.Error()
|
||||
}
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
|
||||
return nil, nil, debugErr
|
||||
}
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err)
|
||||
validationErr := fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err)
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
|
||||
return warnings, nil, errors.Join(validationErr, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr))
|
||||
}
|
||||
return warnings, nil, validationErr
|
||||
}
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil {
|
||||
return warnings, nil, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
if !result.Approved {
|
||||
reason := result.ReasonCode
|
||||
@@ -461,7 +431,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
|
||||
if message == "" {
|
||||
message = "output rejected"
|
||||
}
|
||||
return nil, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil
|
||||
return warnings, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil
|
||||
}
|
||||
warnings = append(warnings, result.Warnings...)
|
||||
}
|
||||
|
||||
@@ -230,6 +230,7 @@ func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *te
|
||||
laneID := lane.resolved.ID
|
||||
firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID)
|
||||
secondPath := fmt.Sprintf("%s/%s/attempt-02.json", stage, laneID)
|
||||
assertAttemptEnvelopeSequence(t, debug, fmt.Sprintf("%s/%s", stage, laneID), 1, 2)
|
||||
first := debug.envelope(t, firstPath)
|
||||
second := debug.envelope(t, secondPath)
|
||||
if len(first.LLMCalls) != 1 || len(second.LLMCalls) != 1 || first.LLMCalls[0].CallID == second.LLMCalls[0].CallID {
|
||||
|
||||
268
internal/framework/pipeline/runner_candidate_encoding_test.go
Normal file
268
internal/framework/pipeline/runner_candidate_encoding_test.go
Normal file
@@ -0,0 +1,268 @@
|
||||
package pipeline
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
)
|
||||
|
||||
type observedNotesCodec struct {
|
||||
candidateValues []codecNotes
|
||||
finalValues []codecNotes
|
||||
candidateError string
|
||||
finalError string
|
||||
}
|
||||
|
||||
func (*observedNotesCodec) Kind() contracts.ArtifactKind { return "test/notes" }
|
||||
|
||||
func (*observedNotesCodec) Schema() contracts.ArtifactSchema { return notesCodec().schema }
|
||||
|
||||
func (*observedNotesCodec) MediaType() string { return "application/json" }
|
||||
|
||||
func (c *observedNotesCodec) EncodeCandidate(value codecNotes) ([]byte, error) {
|
||||
c.candidateValues = append(c.candidateValues, value)
|
||||
if c.candidateError != "" && firstNote(value) == c.candidateError {
|
||||
return nil, errors.New("candidate encoding failed")
|
||||
}
|
||||
return json.Marshal(value)
|
||||
}
|
||||
|
||||
func (c *observedNotesCodec) Encode(value codecNotes) ([]byte, error) {
|
||||
c.finalValues = append(c.finalValues, value)
|
||||
if c.finalError != "" && firstNote(value) == c.finalError {
|
||||
return nil, errors.New("final encoding failed")
|
||||
}
|
||||
if firstNote(value) == "invalid" {
|
||||
return nil, errors.New("invalid note is not a final artifact")
|
||||
}
|
||||
return json.Marshal(value)
|
||||
}
|
||||
|
||||
func (*observedNotesCodec) Decode(content []byte) (codecNotes, error) {
|
||||
var value codecNotes
|
||||
return value, json.Unmarshal(content, &value)
|
||||
}
|
||||
|
||||
func firstNote(value codecNotes) string {
|
||||
if len(value.Items) == 0 {
|
||||
return ""
|
||||
}
|
||||
return value.Items[0]
|
||||
}
|
||||
|
||||
func (c *observedNotesCodec) candidateCalls(value string) int {
|
||||
return matchingNotes(c.candidateValues, value)
|
||||
}
|
||||
|
||||
func (c *observedNotesCodec) finalCalls(value string) int {
|
||||
return matchingNotes(c.finalValues, value)
|
||||
}
|
||||
|
||||
func matchingNotes(values []codecNotes, value string) int {
|
||||
count := 0
|
||||
for _, candidate := range values {
|
||||
if firstNote(candidate) == value {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
type candidateCheckpointRecorder struct {
|
||||
CheckpointRecorder
|
||||
mergeSucceeded int
|
||||
normalizeSucceeded int
|
||||
mergeOutput CheckpointArtifact
|
||||
normalizeOutput CheckpointArtifact
|
||||
}
|
||||
|
||||
func (r *candidateCheckpointRecorder) MergeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
|
||||
r.mergeSucceeded++
|
||||
r.mergeOutput = cloneCheckpointArtifact(output)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *candidateCheckpointRecorder) NormalizeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
|
||||
r.normalizeSucceeded++
|
||||
r.normalizeOutput = cloneCheckpointArtifact(output)
|
||||
return nil
|
||||
}
|
||||
|
||||
func installObservedNotesCodec(t *testing.T, prepared *PreparedPipeline, codec *observedNotesCodec) {
|
||||
t.Helper()
|
||||
registry := NewArtifactCodecRegistry()
|
||||
if err := RegisterArtifactCodec(registry, codec); err != nil {
|
||||
t.Fatalf("RegisterArtifactCodec() error = %v", err)
|
||||
}
|
||||
entry, _, err := registry.entry("test/notes")
|
||||
if err != nil {
|
||||
t.Fatalf("codec entry error = %v", err)
|
||||
}
|
||||
prepared.lanes[0].typed.codec = entry
|
||||
}
|
||||
|
||||
func configureCandidateOperation(prepared *PreparedPipeline, target ModuleStage, value codecNotes) {
|
||||
lane := &prepared.lanes[0]
|
||||
switch target {
|
||||
case StageMerge:
|
||||
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
|
||||
return erasedTypedResult{Value: value}, nil
|
||||
}
|
||||
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
|
||||
return erasedTypedResult{Value: codecNotes{Items: []string{"normalized-other"}}}, nil
|
||||
}
|
||||
case StageNormalize:
|
||||
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
|
||||
return erasedTypedResult{Value: codecNotes{Items: []string{"merged-other"}}}, nil
|
||||
}
|
||||
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
|
||||
return erasedTypedResult{Value: value}, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, approved bool) {
|
||||
validator := preparedValidator{
|
||||
resolved: ResolvedValidator{Binding: Binding("candidate-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
|
||||
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
|
||||
return contracts.ValidationResult{Approved: approved, ReasonCode: "candidate_rejected", Message: "candidate rejected by validator"}, nil
|
||||
},
|
||||
}
|
||||
switch target {
|
||||
case StageMerge:
|
||||
prepared.lanes[0].mergeValidators.validators = []preparedValidator{validator}
|
||||
case StageNormalize:
|
||||
prepared.lanes[0].normalizeValidators.validators = []preparedValidator{validator}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerRejectsCandidatesBeforeFinalEncoding(t *testing.T) {
|
||||
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
|
||||
t.Run(string(target), func(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
codec := &observedNotesCodec{}
|
||||
installObservedNotesCodec(t, prepared, codec)
|
||||
configureCandidateOperation(prepared, target, codecNotes{Items: []string{"invalid"}})
|
||||
setCandidateValidator(prepared, target, false)
|
||||
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
||||
|
||||
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder})
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want validator rejection", err)
|
||||
}
|
||||
if len(output.Rejected) != 1 || output.Rejected[0].Stage != string(target) || output.Rejected[0].ReasonCode != "candidate_rejected" {
|
||||
t.Fatalf("rejected outputs = %#v, want %s validator rejection", output.Rejected, target)
|
||||
}
|
||||
if codec.candidateCalls("invalid") != 1 || codec.finalCalls("invalid") != 0 {
|
||||
t.Fatalf("invalid candidate calls = candidate %d, final %d; want 1, 0", codec.candidateCalls("invalid"), codec.finalCalls("invalid"))
|
||||
}
|
||||
if target == StageMerge && recorder.mergeSucceeded != 0 {
|
||||
t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
|
||||
}
|
||||
if target == StageNormalize && recorder.normalizeSucceeded != 0 {
|
||||
t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerFinalEncodesAcceptedCandidatesOnce(t *testing.T) {
|
||||
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
|
||||
t.Run(string(target), func(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
codec := &observedNotesCodec{}
|
||||
installObservedNotesCodec(t, prepared, codec)
|
||||
value := "accepted-" + string(target)
|
||||
configureCandidateOperation(prepared, target, codecNotes{Items: []string{value}})
|
||||
setCandidateValidator(prepared, target, true)
|
||||
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
||||
loader := &extractResultLoader{CheckpointLoader: NoopCheckpointLoader(), decision: CheckpointDecision{Reason: "not found"}}
|
||||
|
||||
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Checkpoint: loader})
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil", err)
|
||||
}
|
||||
if codec.candidateCalls(value) != 1 || codec.finalCalls(value) != 1 {
|
||||
t.Fatalf("accepted candidate calls = candidate %d, final %d; want 1, 1", codec.candidateCalls(value), codec.finalCalls(value))
|
||||
}
|
||||
if target == StageMerge && recorder.mergeSucceeded != 1 {
|
||||
t.Fatalf("merge checkpoints = %d, want one", recorder.mergeSucceeded)
|
||||
}
|
||||
if target == StageNormalize && recorder.normalizeSucceeded != 1 {
|
||||
t.Fatalf("normalize checkpoints = %d, want one", recorder.normalizeSucceeded)
|
||||
}
|
||||
checkpoint := recorder.mergeOutput
|
||||
if target == StageNormalize {
|
||||
checkpoint = recorder.normalizeOutput
|
||||
}
|
||||
var stored codecNotes
|
||||
if err := json.Unmarshal(checkpoint.Artifact.Content, &stored); err != nil || !reflect.DeepEqual(stored, codecNotes{Items: []string{value}}) {
|
||||
t.Fatalf("checkpoint content = %s, %v; want accepted value %q", checkpoint.Artifact.Content, err, value)
|
||||
}
|
||||
gotStages := make([]string, len(output.CheckpointEvents))
|
||||
for i, event := range output.CheckpointEvents {
|
||||
gotStages[i] = event.Stage
|
||||
}
|
||||
wantStages := []string{"source", string(StageChunk), string(StageExtract), string(StageMerge), string(StageNormalize)}
|
||||
if !reflect.DeepEqual(gotStages, wantStages) {
|
||||
t.Fatalf("checkpoint event stages = %#v, want %#v", gotStages, wantStages)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerRecordsCandidateAndFinalEncodingFailures(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
target ModuleStage
|
||||
value string
|
||||
candidate bool
|
||||
wantError string
|
||||
}{
|
||||
{name: "merge candidate", target: StageMerge, value: "merge-candidate-failure", candidate: true, wantError: "serialize merge candidate"},
|
||||
{name: "normalize candidate", target: StageNormalize, value: "normalize-candidate-failure", candidate: true, wantError: "serialize normalize candidate"},
|
||||
{name: "merge final", target: StageMerge, value: "merge-final-failure", wantError: "serialize accepted merge output"},
|
||||
{name: "normalize final", target: StageNormalize, value: "normalize-final-failure", wantError: "serialize accepted normalize output"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
codec := &observedNotesCodec{}
|
||||
if tc.candidate {
|
||||
codec.candidateError = tc.value
|
||||
} else {
|
||||
codec.finalError = tc.value
|
||||
}
|
||||
installObservedNotesCodec(t, prepared, codec)
|
||||
configureCandidateOperation(prepared, tc.target, codecNotes{Items: []string{tc.value}})
|
||||
setCandidateValidator(prepared, tc.target, true)
|
||||
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
||||
debug := newCapturedDebugRecorder()
|
||||
|
||||
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Debug: debug})
|
||||
if err == nil || !strings.Contains(err.Error(), tc.wantError) {
|
||||
t.Fatalf("Run() error = %v, want %q", err, tc.wantError)
|
||||
}
|
||||
attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", tc.target)
|
||||
envelope := debug.envelope(t, attemptPath)
|
||||
if !strings.Contains(envelope.Error, tc.wantError) {
|
||||
t.Fatalf("attempt error = %q, want %q", envelope.Error, tc.wantError)
|
||||
}
|
||||
if tc.candidate && codec.finalCalls(tc.value) != 0 {
|
||||
t.Fatalf("final encode calls = %d, want none after candidate failure", codec.finalCalls(tc.value))
|
||||
}
|
||||
if tc.target == StageMerge && recorder.mergeSucceeded != 0 {
|
||||
t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
|
||||
}
|
||||
if tc.target == StageNormalize && recorder.normalizeSucceeded != 0 {
|
||||
t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -268,24 +268,36 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
|
||||
started := time.Now().UTC()
|
||||
attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
|
||||
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
|
||||
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started})
|
||||
extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
|
||||
if callErr != nil {
|
||||
_ = writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope)
|
||||
return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
|
||||
attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
|
||||
return false, nil, terminal.record(nil, attemptErr)
|
||||
}
|
||||
artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value}
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value}, state.prepared.extractValidators, attempt, input.Debug)
|
||||
attemptWarnings := cloneWarnings(extracted.Warnings)
|
||||
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
|
||||
if encodeErr != nil {
|
||||
attemptErr := fmt.Errorf("serialize extract candidate for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr)
|
||||
payload := map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}
|
||||
return false, nil, terminal.record(payload, attemptErr)
|
||||
}
|
||||
serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug)
|
||||
attemptWarnings = append(attemptWarnings, warnings...)
|
||||
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
|
||||
if validateErr != nil || rejected != nil {
|
||||
return false, rejected, validateErr
|
||||
return false, rejected, terminal.record(payload, validateErr)
|
||||
}
|
||||
stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
|
||||
if encodeErr != nil {
|
||||
return false, nil, encodeErr
|
||||
attemptErr := fmt.Errorf("serialize accepted extract output for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr)
|
||||
return false, nil, terminal.record(payload, attemptErr)
|
||||
}
|
||||
stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
|
||||
accepted, serialized = artifact, stored
|
||||
acceptedWarnings = append(cloneWarnings(extracted.Warnings), warnings...)
|
||||
if debugErr := writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope); debugErr != nil {
|
||||
acceptedWarnings = attemptWarnings
|
||||
if debugErr := terminal.record(payload, nil); debugErr != nil {
|
||||
return false, nil, debugErr
|
||||
}
|
||||
return true, nil, nil
|
||||
|
||||
@@ -142,6 +142,11 @@ func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) {
|
||||
if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" {
|
||||
t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings)
|
||||
}
|
||||
assertAttemptEnvelopeSequence(t, debug, "extract/notes/chunk-000001", 1, 2)
|
||||
first := debug.envelope(t, "extract/notes/chunk-000001/attempt-01.json")
|
||||
if first.Error != "" || !strings.Contains(string(debug.json["extract/notes/chunk-000001/attempt-01.json"]), "rejection") {
|
||||
t.Fatalf("first extract attempt = %#v, want rejection without error", first)
|
||||
}
|
||||
name := "extract/notes/chunk-000001/attempt-02.json"
|
||||
if !debug.has(name) {
|
||||
t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())
|
||||
|
||||
248
internal/framework/pipeline/runner_terminal_debug_test.go
Normal file
248
internal/framework/pipeline/runner_terminal_debug_test.go
Normal file
@@ -0,0 +1,248 @@
|
||||
package pipeline
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
)
|
||||
|
||||
type terminalChunker struct {
|
||||
key string
|
||||
chunks []source.Chunk
|
||||
warnings []contracts.Warning
|
||||
err error
|
||||
}
|
||||
|
||||
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) {
|
||||
return contracts.ChunkResult{Chunks: cloneSourceChunks(c.chunks), Warnings: cloneWarnings(c.warnings)}, c.err
|
||||
}
|
||||
|
||||
type terminalChunkValidator struct {
|
||||
result contracts.ValidationResult
|
||||
err error
|
||||
}
|
||||
|
||||
func (terminalChunkValidator) Name() string { return "terminal/chunk-validator" }
|
||||
|
||||
func (terminalChunkValidator) ExecutionClass() contracts.ExecutionClass {
|
||||
return contracts.ExecutionClassDeterministic
|
||||
}
|
||||
|
||||
func (v terminalChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
|
||||
return v.result, v.err
|
||||
}
|
||||
|
||||
func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, prefix string, attempts ...int) {
|
||||
t.Helper()
|
||||
marker := strings.TrimSuffix(prefix, "/") + "/attempt-"
|
||||
var got []int
|
||||
for _, name := range debug.names() {
|
||||
if !strings.HasPrefix(name, marker) || !strings.HasSuffix(name, ".json") {
|
||||
continue
|
||||
}
|
||||
remainder := strings.TrimSuffix(strings.TrimPrefix(name, marker), ".json")
|
||||
if strings.Contains(remainder, "/") {
|
||||
continue
|
||||
}
|
||||
value, err := strconv.Atoi(remainder)
|
||||
if err != nil {
|
||||
t.Fatalf("parse attempt index from %q: %v", name, err)
|
||||
}
|
||||
got = append(got, value)
|
||||
}
|
||||
sort.Ints(got)
|
||||
if !reflect.DeepEqual(got, attempts) {
|
||||
t.Fatalf("attempt envelopes under %q = %#v, want %#v; names = %#v", prefix, got, attempts, debug.names())
|
||||
}
|
||||
}
|
||||
|
||||
func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, []source.Chunk) {
|
||||
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)
|
||||
}
|
||||
|
||||
func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
moduleError error
|
||||
validator terminalChunkValidator
|
||||
wantError string
|
||||
wantRejection bool
|
||||
}{
|
||||
{name: "accepted", validator: terminalChunkValidator{result: contracts.ValidationResult{Approved: true}}},
|
||||
{name: "module error", moduleError: errors.New("chunk module failed"), wantError: "chunk module failed"},
|
||||
{name: "validator rejection", validator: terminalChunkValidator{result: contracts.ValidationResult{Approved: false, ReasonCode: "chunk_rejected", Message: "chunk rejected"}}, wantRejection: true},
|
||||
{name: "validator error", validator: terminalChunkValidator{err: errors.New("chunk validator failed")}, wantError: "chunk validator failed"},
|
||||
}
|
||||
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.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding("terminal/chunk-validator"), Target: ValidatorTargetChunk}, chunk: tc.validator}}
|
||||
debug := newCapturedDebugRecorder()
|
||||
|
||||
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
|
||||
assertAttemptEnvelopeSequence(t, debug, "chunk", 1)
|
||||
envelope := debug.envelope(t, "chunk/attempt-01.json")
|
||||
if tc.wantError != "" {
|
||||
if err == nil || !strings.Contains(err.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) {
|
||||
t.Fatalf("Run() error = %v, attempt error = %q; want %q", err, envelope.Error, tc.wantError)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil", err)
|
||||
}
|
||||
if tc.wantRejection {
|
||||
if envelope.Error != "" || len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "chunk_rejected" {
|
||||
t.Fatalf("chunk rejection = envelope %#v, outputs %#v", envelope, output.Rejected)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerRecordsExtractTerminalOutcomes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
moduleError error
|
||||
validatorErr error
|
||||
reject bool
|
||||
candidateFail bool
|
||||
finalFail bool
|
||||
wantError string
|
||||
}{
|
||||
{name: "terminal rejection", reject: true},
|
||||
{name: "module error", moduleError: errors.New("extract module failed"), wantError: "extract module failed"},
|
||||
{name: "validator error", validatorErr: errors.New("extract validator failed"), wantError: "extract validator failed"},
|
||||
{name: "candidate codec error", candidateFail: true, wantError: "serialize extract candidate"},
|
||||
{name: "final codec error", finalFail: true, wantError: "serialize accepted extract output"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
value := "extract-terminal"
|
||||
codec := &observedNotesCodec{}
|
||||
if tc.candidateFail {
|
||||
codec.candidateError = value
|
||||
}
|
||||
if tc.finalFail {
|
||||
codec.finalError = value
|
||||
}
|
||||
installObservedNotesCodec(t, prepared, codec)
|
||||
installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
|
||||
if tc.moduleError != nil {
|
||||
return erasedTypedResult{}, tc.moduleError
|
||||
}
|
||||
return erasedTypedResult{Value: codecNotes{Items: []string{value}}, Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "extract warning"}}}, nil
|
||||
})
|
||||
prepared.lanes[0].extractValidators.validators = []preparedValidator{{
|
||||
resolved: ResolvedValidator{Binding: Binding("terminal/extract-validator"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
|
||||
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
|
||||
return contracts.ValidationResult{Approved: !tc.reject, ReasonCode: "extract_rejected", Message: "extract rejected"}, tc.validatorErr
|
||||
},
|
||||
}}
|
||||
debug := newCapturedDebugRecorder()
|
||||
|
||||
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
|
||||
assertAttemptEnvelopeSequence(t, debug, "extract/notes/chunk-000001", 1)
|
||||
attemptPath := "extract/notes/chunk-000001/attempt-01.json"
|
||||
envelope := debug.envelope(t, attemptPath)
|
||||
if tc.wantError != "" {
|
||||
if err == nil || !strings.Contains(err.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) {
|
||||
t.Fatalf("Run() error = %v, attempt error = %q; want %q", err, envelope.Error, tc.wantError)
|
||||
}
|
||||
} else {
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil rejection", err)
|
||||
}
|
||||
if envelope.Error != "" || len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "extract_rejected" {
|
||||
t.Fatalf("extract rejection = envelope %#v, outputs %#v", envelope, output.Rejected)
|
||||
}
|
||||
}
|
||||
if tc.moduleError == nil && !strings.Contains(string(debug.json[attemptPath]), "extract warning") {
|
||||
t.Fatalf("attempt envelope = %s, want attempt warning", debug.json[attemptPath])
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerJoinsPrimaryAndAttemptWriteErrors(t *testing.T) {
|
||||
t.Run("chunk", func(t *testing.T) {
|
||||
prepared, chunks := preparedTerminalDebugPipeline(t)
|
||||
primary := errors.New("chunk operation failed")
|
||||
prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, err: primary}
|
||||
debug := newCapturedDebugRecorder()
|
||||
debug.failPath = "chunk/attempt-01.json"
|
||||
|
||||
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
|
||||
if err == nil || !errors.Is(err, primary) || !strings.Contains(err.Error(), "write chunk attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") {
|
||||
t.Fatalf("Run() error = %v, want joined operation and debug errors", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("extract", func(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
primary := errors.New("extract operation failed")
|
||||
installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
|
||||
return erasedTypedResult{}, primary
|
||||
})
|
||||
debug := newCapturedDebugRecorder()
|
||||
debug.failPath = "extract/notes/chunk-000001/attempt-01.json"
|
||||
|
||||
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
|
||||
if err == nil || !errors.Is(err, primary) || !strings.Contains(err.Error(), "write extract attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") {
|
||||
t.Fatalf("Run() error = %v, want joined operation and debug errors", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestRunnerKeepsExtractModuleAndValidatorLLMCallsIsolated(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
debug := newCapturedDebugRecorder()
|
||||
client := WithDebugLLMRecording(attemptDebugLLM{}, debug)
|
||||
installExtractOperation(prepared, 0, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
|
||||
if err := callAttemptDebugLLM(ctx, client, "extract-module"); err != nil {
|
||||
return erasedTypedResult{}, err
|
||||
}
|
||||
return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil
|
||||
})
|
||||
prepared.lanes[0].extractValidators.validators = []preparedValidator{{
|
||||
resolved: ResolvedValidator{Binding: Binding("llm-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
|
||||
typedValidate: func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) {
|
||||
if err := callAttemptDebugLLM(ctx, client, "extract-validator"); err != nil {
|
||||
return contracts.ValidationResult{}, err
|
||||
}
|
||||
return contracts.ValidationResult{Approved: true}, nil
|
||||
},
|
||||
}}
|
||||
|
||||
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}); err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil", err)
|
||||
}
|
||||
module := debug.envelope(t, "extract/notes/chunk-000001/attempt-01.json")
|
||||
validator := debug.envelope(t, "validate/extract/notes/typed~2fextract-notes/01-llm-check-attempt-01.json")
|
||||
if len(module.LLMCalls) != 1 || !strings.Contains(module.LLMCalls[0].ResponsePath, "extract/notes/chunk-000001/attempt-01/") {
|
||||
t.Fatalf("module LLM calls = %#v, want extract module call only", module.LLMCalls)
|
||||
}
|
||||
if len(validator.LLMCalls) != 1 || !strings.Contains(validator.LLMCalls[0].ResponsePath, "validate/extract/notes/") {
|
||||
t.Fatalf("validator LLM calls = %#v, want validator call only", validator.LLMCalls)
|
||||
}
|
||||
if module.LLMCalls[0].CallID == validator.LLMCalls[0].CallID {
|
||||
t.Fatalf("module and validator attempts share LLM call %#v", module.LLMCalls)
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"time"
|
||||
@@ -111,6 +112,14 @@ func checkpointArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID st
|
||||
return CheckpointArtifact{LaneID: laneID, ModuleKey: moduleKey, SourceID: sourceID, Artifact: serialized, SchemaDigest: contracts.DigestArtifactSchema(serialized.Schema)}, nil
|
||||
}
|
||||
|
||||
func serializeCandidateArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID string, value any) (CheckpointArtifact, error) {
|
||||
serialized, err := serializeArtifact(codec, value, true)
|
||||
if err != nil {
|
||||
return CheckpointArtifact{}, err
|
||||
}
|
||||
return CheckpointArtifact{LaneID: laneID, ModuleKey: moduleKey, SourceID: sourceID, Artifact: serialized, SchemaDigest: contracts.DigestArtifactSchema(serialized.Schema)}, nil
|
||||
}
|
||||
|
||||
type laneRunError struct {
|
||||
stage ModuleStage
|
||||
err error
|
||||
@@ -167,42 +176,32 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
|
||||
started := time.Now().UTC()
|
||||
attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
|
||||
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
|
||||
attemptEnvelope := func(payload map[string]any, attemptErr error) error {
|
||||
envelope := debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started, Payload: payload}
|
||||
if attemptErr != nil {
|
||||
envelope.Error = attemptErr.Error()
|
||||
}
|
||||
return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope)
|
||||
}
|
||||
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
|
||||
result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
|
||||
if callErr != nil {
|
||||
attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
|
||||
if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, nil, attemptErr
|
||||
return false, nil, terminal.record(nil, attemptErr)
|
||||
}
|
||||
candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
|
||||
stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
|
||||
attemptWarnings := cloneWarnings(result.Warnings)
|
||||
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
|
||||
if encodeErr != nil {
|
||||
attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)
|
||||
if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, nil, attemptErr
|
||||
return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
|
||||
}
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.mergeValidators, attempt, input.Debug)
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug)
|
||||
attemptWarnings = append(attemptWarnings, warnings...)
|
||||
payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
|
||||
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
|
||||
if validateErr != nil || rejected != nil {
|
||||
if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, rejected, validateErr
|
||||
return false, rejected, terminal.record(payload, validateErr)
|
||||
}
|
||||
if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
|
||||
stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
|
||||
if encodeErr != nil {
|
||||
attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
|
||||
return false, nil, terminal.record(payload, attemptErr)
|
||||
}
|
||||
if debugErr := terminal.record(payload, nil); debugErr != nil {
|
||||
return false, nil, debugErr
|
||||
}
|
||||
merged, serializedMerge = candidate, stored
|
||||
mergeWarnings = attemptWarnings
|
||||
@@ -257,41 +256,31 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
|
||||
started := time.Now().UTC()
|
||||
attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
|
||||
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
|
||||
attemptEnvelope := func(payload map[string]any, attemptErr error) error {
|
||||
envelope := debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started, Payload: payload}
|
||||
if attemptErr != nil {
|
||||
envelope.Error = attemptErr.Error()
|
||||
}
|
||||
return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope)
|
||||
}
|
||||
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
|
||||
result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
|
||||
if callErr != nil {
|
||||
attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
|
||||
if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, nil, attemptErr
|
||||
return false, nil, terminal.record(nil, attemptErr)
|
||||
}
|
||||
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
|
||||
attemptWarnings := cloneWarnings(result.Warnings)
|
||||
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
|
||||
if encodeErr != nil {
|
||||
attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
|
||||
if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, nil, attemptErr
|
||||
return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
|
||||
}
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.normalizeValidators, attempt, input.Debug)
|
||||
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug)
|
||||
attemptWarnings = append(attemptWarnings, warnings...)
|
||||
payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
|
||||
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
|
||||
if validateErr != nil || rejected != nil {
|
||||
if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
return false, rejected, validateErr
|
||||
return false, rejected, terminal.record(payload, validateErr)
|
||||
}
|
||||
if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
|
||||
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
|
||||
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
|
||||
if encodeErr != nil {
|
||||
attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
|
||||
return false, nil, terminal.record(payload, attemptErr)
|
||||
}
|
||||
if debugErr := terminal.record(payload, nil); debugErr != nil {
|
||||
return false, nil, debugErr
|
||||
}
|
||||
serializedNormalize = stored
|
||||
normalizeWarnings = attemptWarnings
|
||||
@@ -346,6 +335,13 @@ func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, m
|
||||
|
||||
func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {
|
||||
var warnings []contracts.Warning
|
||||
if len(chain.validators) > 0 && target.candidate == nil {
|
||||
candidate, err := validationCandidateArtifact(codec, target)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("serialize %s candidate for validation: %w", target.stage, err)
|
||||
}
|
||||
target.candidate = &candidate
|
||||
}
|
||||
for index, item := range chain.validators {
|
||||
binding := item.resolved.Binding
|
||||
var result contracts.ValidationResult
|
||||
@@ -358,25 +354,29 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
|
||||
target.llmProfile = binding.LLMProfile
|
||||
result, err = item.typedValidate(validatorCtx, item.typed, target)
|
||||
case ValidatorTargetSerialized:
|
||||
artifact, encodeErr := serializeArtifact(codec, target.value, true)
|
||||
artifact, encodeErr := validationCandidateArtifact(codec, target)
|
||||
if encodeErr != nil {
|
||||
err = encodeErr
|
||||
break
|
||||
}
|
||||
result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.CloneArtifactSchema(artifact.Schema), MediaType: artifact.MediaType, Content: append([]byte(nil), artifact.Content...)})
|
||||
result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)})
|
||||
default:
|
||||
return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)
|
||||
}
|
||||
artifact, _ := serializeArtifact(codec, target.value, true)
|
||||
debugCall := debugValidationCall{ValidatorName: binding.Module, Request: map[string]any{"stage": string(target.stage), "lane_id": target.laneID, "module_key": target.moduleKey, "source_id": target.sourceID, "artifact": debugCheckpointArtifact(CheckpointArtifact{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)}), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)}
|
||||
artifact, _ := validationCandidateArtifact(codec, target)
|
||||
debugCall := debugValidationCall{ValidatorName: binding.Module, Request: map[string]any{"stage": string(target.stage), "lane_id": target.laneID, "module_key": target.moduleKey, "source_id": target.sourceID, "artifact": debugCheckpointArtifact(artifact), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)}
|
||||
if err != nil {
|
||||
debugCall.Error = err.Error()
|
||||
}
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
|
||||
return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err)
|
||||
validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err)
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
|
||||
return warnings, nil, errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr))
|
||||
}
|
||||
return warnings, nil, validationErr
|
||||
}
|
||||
if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil {
|
||||
return warnings, nil, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)
|
||||
}
|
||||
if !result.Approved {
|
||||
reason := result.ReasonCode
|
||||
@@ -387,7 +387,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
|
||||
if message == "" {
|
||||
message = "artifact rejected"
|
||||
}
|
||||
return nil, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string {
|
||||
return warnings, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string {
|
||||
if target.chunk != nil {
|
||||
return target.chunk.ID
|
||||
}
|
||||
@@ -403,3 +403,10 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
|
||||
}
|
||||
return warnings, nil, nil
|
||||
}
|
||||
|
||||
func validationCandidateArtifact(codec artifactCodecEntry, target typedValidationTarget) (CheckpointArtifact, error) {
|
||||
if target.candidate != nil {
|
||||
return cloneCheckpointArtifact(*target.candidate), nil
|
||||
}
|
||||
return serializeCandidateArtifact(codec, target.laneID, target.moduleKey, target.sourceID, target.value)
|
||||
}
|
||||
|
||||
@@ -41,6 +41,7 @@ type typedValidationTarget struct {
|
||||
chunks []source.Chunk
|
||||
ref source.SourceRef
|
||||
value any
|
||||
candidate *CheckpointArtifact
|
||||
}
|
||||
|
||||
func exactTypedValue[T any](operation string, value any) (T, error) {
|
||||
|
||||
@@ -2,6 +2,7 @@ package spells
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"reflect"
|
||||
@@ -12,6 +13,7 @@ import (
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
|
||||
spellshape "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/spells/shape"
|
||||
)
|
||||
|
||||
func TestCodecMatchesMaintainedDurableFixture(t *testing.T) {
|
||||
@@ -101,6 +103,25 @@ func TestCodecRejectsInvalidCanonicalValues(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCodecEncodesIncompleteCandidateWithoutWeakeningFinalEncoding(t *testing.T) {
|
||||
codec := New()
|
||||
candidate := dnd.SpellList{}
|
||||
content, err := codec.EncodeCandidate(candidate)
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeCandidate() error = %v, want nil", err)
|
||||
}
|
||||
if !json.Valid(content) {
|
||||
t.Fatalf("EncodeCandidate() = %q, want JSON", content)
|
||||
}
|
||||
result, err := spellshape.New(spellshape.Options{}).Validate(context.Background(), contracts.TypedValidationRequest[dnd.SpellList]{Value: candidate})
|
||||
if err != nil || result.Approved || result.ReasonCode != spellshape.ReasonCode {
|
||||
t.Fatalf("shape validation = %#v, %v; want candidate rejection", result, err)
|
||||
}
|
||||
if _, err := codec.Encode(candidate); err == nil || !strings.Contains(err.Error(), "spell_casts must be present") {
|
||||
t.Fatalf("Encode() error = %v, want strict final shape error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCodecSchemaIsMutationSafe(t *testing.T) {
|
||||
first := New().Schema()
|
||||
first.JSONSchema[0] = '['
|
||||
|
||||
@@ -155,6 +155,100 @@ func TestImportBoundaryRules(t *testing.T) {
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "almanac/register",
|
||||
},
|
||||
{
|
||||
name: "production CLI cannot import family root",
|
||||
filename: "internal/cli/catalog.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "almanac",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "production CLI cannot import concrete leaf",
|
||||
filename: "internal/cli/catalog.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "almanac/extract/events",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "production CLI cannot import generic leaf",
|
||||
filename: "internal/cli/catalog.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "generic/output/json",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "production CLI cannot import non-registrar package",
|
||||
filename: "internal/cli/catalog.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "almanac/register/helpers",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "CLI test may import concrete leaf",
|
||||
filename: "internal/cli/compatibility_test.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "almanac/extract/events",
|
||||
},
|
||||
{
|
||||
name: "framework production cannot import concrete module",
|
||||
filename: "internal/framework/pipeline/runner.go",
|
||||
sourcePackage: "pipeline",
|
||||
importPath: moduleImportPrefix + "almanac/extract/events",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "framework production cannot import generic module",
|
||||
filename: "internal/framework/pipeline/runner.go",
|
||||
sourcePackage: "pipeline",
|
||||
importPath: moduleImportPrefix + "generic/normalize/noop",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "core production cannot import concrete module",
|
||||
filename: "internal/core/source/source.go",
|
||||
sourcePackage: "source",
|
||||
importPath: moduleImportPrefix + "almanac",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "core production cannot import generic module",
|
||||
filename: "internal/core/source/source.go",
|
||||
sourcePackage: "source",
|
||||
importPath: moduleImportPrefix + "generic/chunk/units",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "framework test may import module implementation",
|
||||
filename: "internal/framework/pipeline/compatibility_test.go",
|
||||
sourcePackage: "pipeline",
|
||||
importPath: moduleImportPrefix + "generic/chunk/units",
|
||||
},
|
||||
{
|
||||
name: "core test may import module implementation",
|
||||
filename: "internal/core/source/compatibility_test.go",
|
||||
sourcePackage: "source",
|
||||
importPath: moduleImportPrefix + "almanac",
|
||||
},
|
||||
{
|
||||
name: "module production cannot import integration infrastructure",
|
||||
filename: "internal/modules/almanac/register/register.go",
|
||||
sourcePackage: "register",
|
||||
importPath: moduleImportPrefix + "integration",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "non-module production cannot import integration infrastructure",
|
||||
filename: "cmd/notarius/main.go",
|
||||
sourcePackage: "main",
|
||||
importPath: moduleImportPrefix + "integration/helpers",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "test may import integration infrastructure",
|
||||
filename: "internal/cli/compatibility_test.go",
|
||||
sourcePackage: "cli",
|
||||
importPath: moduleImportPrefix + "integration",
|
||||
},
|
||||
{
|
||||
name: "black-box integration test may compose families",
|
||||
filename: "internal/modules/integration/example_test.go",
|
||||
@@ -183,6 +277,9 @@ func TestImportBoundaryRules(t *testing.T) {
|
||||
if tt.wantError && err == nil {
|
||||
t.Fatal("validateImport() error = nil, want boundary violation")
|
||||
}
|
||||
if tt.wantError && (!strings.Contains(err.Error(), tt.filename) || !strings.Contains(err.Error(), tt.importPath)) {
|
||||
t.Fatalf("validateImport() error = %q, want importing file and import target", err)
|
||||
}
|
||||
if !tt.wantError && err != nil {
|
||||
t.Fatalf("validateImport() error = %v, want nil", err)
|
||||
}
|
||||
@@ -206,43 +303,84 @@ func checkImportBoundaries(repositoryRoot string, filename string) error {
|
||||
return fmt.Errorf("parse import in %s: %w", relative, err)
|
||||
}
|
||||
if err := validateImport(relative, parsed.Name.Name, importPath); err != nil {
|
||||
return fmt.Errorf("%s imports %s: %w", relative, importPath, err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateImport(filename string, sourcePackage string, importPath string) error {
|
||||
if isBlackBoxIntegrationTest(filename, sourcePackage) {
|
||||
target, ok := moduleTargetForImport(importPath)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
targetFamily, targetChild := moduleFamilyForImport(importPath)
|
||||
if targetFamily == "" {
|
||||
return nil
|
||||
isTest := strings.HasSuffix(filename, "_test.go")
|
||||
if target.integration && !isTest {
|
||||
return importBoundaryViolation(filename, importPath, "module integration infrastructure is not a production dependency target")
|
||||
}
|
||||
if isIntegrationFile(filename) {
|
||||
return fmt.Errorf("module integration composition is allowed only in black-box tests")
|
||||
if isBlackBoxIntegrationTest(filename, sourcePackage) {
|
||||
return nil
|
||||
}
|
||||
return importBoundaryViolation(filename, importPath, "module integration composition is allowed only in black-box tests")
|
||||
}
|
||||
if !isTest && (strings.HasPrefix(filename, "internal/framework/") || strings.HasPrefix(filename, "internal/core/")) {
|
||||
return importBoundaryViolation(filename, importPath, "core and framework production code must not import module implementations")
|
||||
}
|
||||
if !isTest && strings.HasPrefix(filename, "internal/cli/") {
|
||||
if target.registrar {
|
||||
return nil
|
||||
}
|
||||
return importBoundaryViolation(filename, importPath, "CLI production code may import only exact module family registrar packages")
|
||||
}
|
||||
sourceFamily, sourceRoot, sourceRegistrar := moduleFamilyForFile(filename)
|
||||
if sourceFamily == "" {
|
||||
return nil
|
||||
}
|
||||
if sourceRoot && sourceFamily == targetFamily && targetChild {
|
||||
return fmt.Errorf("family root must not import child packages")
|
||||
if sourceRoot && sourceFamily == target.family && target.child {
|
||||
return importBoundaryViolation(filename, importPath, "family root must not import child packages")
|
||||
}
|
||||
if sourceFamily == targetFamily {
|
||||
if sourceFamily == target.family {
|
||||
return nil
|
||||
}
|
||||
if sourceFamily == "generic" {
|
||||
return fmt.Errorf("generic family must not import concrete family %q", targetFamily)
|
||||
return importBoundaryViolation(filename, importPath, fmt.Sprintf("generic family must not import concrete family %q", target.family))
|
||||
}
|
||||
if targetFamily == "generic" {
|
||||
if target.family == "generic" {
|
||||
if sourceRegistrar {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("concrete family %q may import generic implementations only from its registrar", sourceFamily)
|
||||
return importBoundaryViolation(filename, importPath, fmt.Sprintf("concrete family %q may import generic implementations only from its registrar", sourceFamily))
|
||||
}
|
||||
return fmt.Errorf("concrete family %q must not import concrete family %q", sourceFamily, targetFamily)
|
||||
return importBoundaryViolation(filename, importPath, fmt.Sprintf("concrete family %q must not import concrete family %q", sourceFamily, target.family))
|
||||
}
|
||||
|
||||
type moduleImportTarget struct {
|
||||
family string
|
||||
child bool
|
||||
registrar bool
|
||||
integration bool
|
||||
}
|
||||
|
||||
func moduleTargetForImport(importPath string) (moduleImportTarget, bool) {
|
||||
if !strings.HasPrefix(importPath, moduleImportPrefix) {
|
||||
return moduleImportTarget{}, false
|
||||
}
|
||||
remainder := strings.TrimPrefix(importPath, moduleImportPrefix)
|
||||
parts := strings.Split(remainder, "/")
|
||||
if len(parts) == 0 || parts[0] == "" {
|
||||
return moduleImportTarget{}, false
|
||||
}
|
||||
return moduleImportTarget{
|
||||
family: parts[0],
|
||||
child: len(parts) > 1,
|
||||
registrar: len(parts) == 2 && parts[1] == "register",
|
||||
integration: parts[0] == "integration",
|
||||
}, true
|
||||
}
|
||||
|
||||
func importBoundaryViolation(filename string, importPath string, rule string) error {
|
||||
return fmt.Errorf("import boundary violation: %s imports %s: %s", filename, importPath, rule)
|
||||
}
|
||||
|
||||
func moduleFamilyForFile(filename string) (family string, root bool, registrar bool) {
|
||||
@@ -258,18 +396,6 @@ func moduleFamilyForFile(filename string) (family string, root bool, registrar b
|
||||
return parts[0], len(parts) == 2, len(parts) > 2 && parts[1] == "register"
|
||||
}
|
||||
|
||||
func moduleFamilyForImport(importPath string) (family string, child bool) {
|
||||
if !strings.HasPrefix(importPath, moduleImportPrefix) {
|
||||
return "", false
|
||||
}
|
||||
remainder := strings.TrimPrefix(importPath, moduleImportPrefix)
|
||||
parts := strings.Split(remainder, "/")
|
||||
if len(parts) == 0 || parts[0] == "" || parts[0] == "integration" {
|
||||
return "", false
|
||||
}
|
||||
return parts[0], len(parts) > 1
|
||||
}
|
||||
|
||||
func isIntegrationFile(filename string) bool {
|
||||
return strings.HasPrefix(filename, "internal/modules/integration/")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user