Compare commits

..

3 Commits

14 changed files with 878 additions and 165 deletions

View File

@@ -181,8 +181,9 @@ payload rules are defined in the
## Production Registration ## Production Registration
The CLI allocates one complete framework registry set and one LLM asset Production composition occurs through family registrars. The CLI allocates one
registry. It invokes `internal/modules/generic/register`, 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 `internal/modules/seriatim/register`, and `internal/modules/dnd/register` in
that order, then exposes the matching catalog for resolution. The generic and that order, then exposes the matching catalog for resolution. The generic and
Seriatim registrars own their production leaf registrations. The D&D registrar 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 point for specializing reusable generic implementations, while the generic
registrar composes only generic children. registrar composes only generic children.
Framework packages must not import production extensions. Tests may compose Core and framework production packages do not import production extensions.
registries and catalogs directly with fakes. 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 ## Adding An Extension

View File

@@ -228,14 +228,26 @@ normally. Dependency fingerprints and debug content digests use the same stable
codec bytes that cross those boundaries. codec bytes that cross those boundaries.
Debug instrumentation wraps run, stage, attempt, validator, and structured LLM Debug instrumentation wraps run, stage, attempt, validator, and structured LLM
boundaries. Every executed module retry has an attempt envelope containing its boundaries. Every executed chunk, extract, merge, and normalize attempt writes
candidate, accepted-attempt warnings, rejection or error, and only the LLM one terminal envelope for acceptance, validator rejection, module or validator
calls made by that module attempt. Validator attempts retain independent scopes error, or applicable candidate or final serialization error. The envelope
under `validate/`. Debug-write failures are framework errors; debug data is 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 never used as a checkpoint source. Typed artifact debug envelopes are
domain-neutral, redact sensitive metadata and bytes through the common debug domain-neutral, redact sensitive metadata and bytes through the common debug
policy, and record codec identity plus schema and content digests. 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 Checkpoint identity, physical layout, reuse behavior, and debug artifact
handling are operator contracts in [Operations](../operations.md). Serialization handling are operator contracts in [Operations](../operations.md). Serialization
and recorder implementation are inventoried in and recorder implementation are inventoried in

View File

@@ -128,15 +128,17 @@ checkpointing does not write debug output.
Debug artifacts include inputs and outputs for source, chunk, extract, merge, Debug artifacts include inputs and outputs for source, chunk, extract, merge,
normalize, and output work, structured LLM request and response data, validator normalize, and output work, structured LLM request and response data, validator
requests and results, timing, and retry attempt metadata. LLM calls made inside requests and results, timing, and retry attempt metadata. LLM calls made inside
a retry or validator attempt a module retry write `prompt-000N.json`, `response-000N.json`, and
write `prompt-000N.json`, `response-000N.json`, and `response-content-000N.*` files under that attempt directory and are linked
`response-content-000N.*` files under that attempt directory and are linked from from its `llm_calls` array. Validator calls use separate attempt scopes under
the attempt `llm_calls` array. Prompt content is written inline in the prompt `validate/` and are not duplicated into the module attempt. Prompt content is
artifact. The response metadata and body use the paired files described above; written inline in the prompt artifact. The response metadata and body use the
the body is pretty-printed JSON when possible and raw text otherwise. Merge and paired files described above; the body is pretty-printed JSON when possible
normalize retries use these stable paths: and raw text otherwise. Retrying stages use these stable module-attempt paths:
```text ```text
chunk/attempt-<NN>.json
extract/<lane-id>/chunk-<NNNNNN>/attempt-<NN>.json
merge/<lane-id>/attempt-<NN>.json merge/<lane-id>/attempt-<NN>.json
merge/<lane-id>/attempt-<NN>/prompt-<NNNN>.json merge/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
merge/<lane-id>/attempt-<NN>/response-<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> normalize/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
``` ```
Checkpoint-reused merge and normalize work retains the stage-level input and Every executed chunk, extract, merge, and normalize attempt has one terminal
output artifacts but has no retry-attempt artifacts because no module attempt envelope recording acceptance, validator rejection, or a module, validator,
executed. Debug artifacts may contain source material, reference material, candidate-serialization, or final-serialization error as applicable. It
prompt inputs, model outputs, and other sensitive data. Typed artifact 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, envelopes include domain-neutral codec identity, redacted metadata and content,
and digests of the stable codec bytes. API keys are not written, and obvious 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 credential-shaped values and sensitive map keys are redacted, but debug

View File

@@ -401,6 +401,31 @@ func writeDebugAttempt(recorder DebugRecorder, attemptPath string, envelope debu
return writeDebugTimed(recorder, attemptPath+".json", debugEnvelopeWithLLMCalls(envelope, scope)) 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 { func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope {
content = redactSecretBytes(content) content = redactSecretBytes(content)
return debugBinaryEnvelope{ return debugBinaryEnvelope{

View File

@@ -5,6 +5,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"mime" "mime"
"path" "path"
@@ -207,6 +208,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
attemptStarted := time.Now().UTC() attemptStarted := time.Now().UTC()
attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) 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{ chunkResult, err := chunker.Chunk(attemptCtx, contracts.ChunkRequest{
Source: doc, Source: doc,
SourceInput: sourceInput.Clone(), SourceInput: sourceInput.Clone(),
@@ -216,68 +218,32 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
Metadata: input.Metadata, Metadata: input.Metadata,
}) })
if err != nil { if err != nil {
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ attemptErr := fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
Stage: string(StageChunk), return false, nil, terminal.record(nil, attemptErr)
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)
} }
if len(chunkResult.Chunks) == 0 { if len(chunkResult.Chunks) == 0 {
err := fmt.Errorf("chunker %q returned no chunks", chunker.Key()) attemptErr := fmt.Errorf("chunker %q returned no chunks", chunker.Key())
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ return false, nil, terminal.record(nil, attemptErr)
Stage: string(StageChunk),
ModuleKey: chunker.Key(),
Attempt: attempt,
StartedAt: attemptStarted,
Error: err.Error(),
}, llmScope)
return false, nil, err
} }
chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks)
if err != nil { if err != nil {
err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) attemptErr := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ payload := map[string]any{"warnings": debugWarningEnvelopes(chunkResult.Warnings)}
Stage: string(StageChunk), return false, nil, terminal.record(payload, attemptErr)
ModuleKey: chunker.Key(),
Attempt: attempt,
StartedAt: attemptStarted,
Payload: map[string]any{
"warnings": cloneWarnings(chunkResult.Warnings),
},
Error: err.Error(),
}, llmScope)
return false, nil, err
} }
validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) 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 { if err != nil || rejection != nil {
_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ return false, rejection, terminal.record(payload, err)
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
} }
canonicalChunks = chunks canonicalChunks = chunks
chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) chunkWarnings = attemptWarnings
if err := writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ if err := terminal.record(payload, nil); err != nil {
Stage: string(StageChunk),
ModuleKey: chunker.Key(),
Attempt: attempt,
StartedAt: attemptStarted,
Payload: map[string]any{
"chunks": debugSourceChunkEnvelopes(chunks),
"warnings": chunkWarnings,
},
}, llmScope); err != nil {
return false, nil, err return false, nil, err
} }
return true, nil, nil return true, nil, nil
@@ -446,11 +412,15 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
if err != nil { if err != nil {
debugCall.Error = err.Error() 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 { 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 { if !result.Approved {
reason := result.ReasonCode reason := result.ReasonCode
@@ -461,7 +431,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
if message == "" { if message == "" {
message = "output rejected" 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...) warnings = append(warnings, result.Warnings...)
} }

View File

@@ -230,6 +230,7 @@ func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *te
laneID := lane.resolved.ID laneID := lane.resolved.ID
firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID) firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID)
secondPath := fmt.Sprintf("%s/%s/attempt-02.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) first := debug.envelope(t, firstPath)
second := debug.envelope(t, secondPath) second := debug.envelope(t, secondPath)
if len(first.LLMCalls) != 1 || len(second.LLMCalls) != 1 || first.LLMCalls[0].CallID == second.LLMCalls[0].CallID { if len(first.LLMCalls) != 1 || len(second.LLMCalls) != 1 || first.LLMCalls[0].CallID == second.LLMCalls[0].CallID {

View 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)
}
})
}
}

View File

@@ -268,24 +268,36 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
started := time.Now().UTC() started := time.Now().UTC()
attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) 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)}) 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 { 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) attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
return false, nil, 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} 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 { 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) stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
if encodeErr != nil { 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 stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
accepted, serialized = artifact, stored accepted, serialized = artifact, stored
acceptedWarnings = append(cloneWarnings(extracted.Warnings), warnings...) acceptedWarnings = attemptWarnings
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 { if debugErr := terminal.record(payload, nil); debugErr != nil {
return false, nil, debugErr return false, nil, debugErr
} }
return true, nil, nil return true, nil, nil

View File

@@ -142,6 +142,11 @@ func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) {
if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" { if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" {
t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings) 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" name := "extract/notes/chunk-000001/attempt-02.json"
if !debug.has(name) { if !debug.has(name) {
t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names()) t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())

View 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)
}
}

View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors"
"fmt" "fmt"
"path" "path"
"time" "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 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 { type laneRunError struct {
stage ModuleStage stage ModuleStage
err error err error
@@ -167,42 +176,32 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
started := time.Now().UTC() started := time.Now().UTC()
attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
attemptEnvelope := func(payload map[string]any, attemptErr error) error { terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
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)
}
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)}) 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 { if callErr != nil {
attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) 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, terminal.record(nil, attemptErr)
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
}
return false, nil, attemptErr
} }
candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value} 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) attemptWarnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
if encodeErr != nil { if encodeErr != nil {
attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr) 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, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
}
return false, nil, 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...) 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 validateErr != nil || rejected != nil {
if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil { return false, rejected, terminal.record(payload, validateErr)
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
}
return false, rejected, validateErr
} }
if debugErr := attemptEnvelope(payload, nil); debugErr != nil { stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) 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 merged, serializedMerge = candidate, stored
mergeWarnings = attemptWarnings mergeWarnings = attemptWarnings
@@ -257,41 +256,31 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
started := time.Now().UTC() started := time.Now().UTC()
attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
attemptEnvelope := func(payload map[string]any, attemptErr error) error { terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
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)
}
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)}) 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 { if callErr != nil {
attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) 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, terminal.record(nil, attemptErr)
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
}
return false, nil, attemptErr
} }
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
attemptWarnings := cloneWarnings(result.Warnings) attemptWarnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
if encodeErr != nil { if encodeErr != nil {
attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr) 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, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
}
return false, nil, 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...) 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 validateErr != nil || rejected != nil {
if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil { return false, rejected, terminal.record(payload, validateErr)
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
}
return false, rejected, validateErr
} }
if debugErr := attemptEnvelope(payload, nil); debugErr != nil { stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) 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 serializedNormalize = stored
normalizeWarnings = attemptWarnings 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) { 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 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 { for index, item := range chain.validators {
binding := item.resolved.Binding binding := item.resolved.Binding
var result contracts.ValidationResult var result contracts.ValidationResult
@@ -358,25 +354,29 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
target.llmProfile = binding.LLMProfile target.llmProfile = binding.LLMProfile
result, err = item.typedValidate(validatorCtx, item.typed, target) result, err = item.typedValidate(validatorCtx, item.typed, target)
case ValidatorTargetSerialized: case ValidatorTargetSerialized:
artifact, encodeErr := serializeArtifact(codec, target.value, true) artifact, encodeErr := validationCandidateArtifact(codec, target)
if encodeErr != nil { if encodeErr != nil {
err = encodeErr err = encodeErr
break 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: default:
return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module) return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)
} }
artifact, _ := serializeArtifact(codec, target.value, true) 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(CheckpointArtifact{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)}), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)} 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 { if err != nil {
debugCall.Error = err.Error() 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 { 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 { if !result.Approved {
reason := result.ReasonCode reason := result.ReasonCode
@@ -387,7 +387,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
if message == "" { if message == "" {
message = "artifact rejected" 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 { if target.chunk != nil {
return target.chunk.ID return target.chunk.ID
} }
@@ -403,3 +403,10 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
} }
return warnings, nil, nil 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)
}

View File

@@ -41,6 +41,7 @@ type typedValidationTarget struct {
chunks []source.Chunk chunks []source.Chunk
ref source.SourceRef ref source.SourceRef
value any value any
candidate *CheckpointArtifact
} }
func exactTypedValue[T any](operation string, value any) (T, error) { func exactTypedValue[T any](operation string, value any) (T, error) {

View File

@@ -2,6 +2,7 @@ package spells
import ( import (
"bytes" "bytes"
"context"
"encoding/json" "encoding/json"
"os" "os"
"reflect" "reflect"
@@ -12,6 +13,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd" "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) { 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) { func TestCodecSchemaIsMutationSafe(t *testing.T) {
first := New().Schema() first := New().Schema()
first.JSONSchema[0] = '[' first.JSONSchema[0] = '['

View File

@@ -155,6 +155,100 @@ func TestImportBoundaryRules(t *testing.T) {
sourcePackage: "cli", sourcePackage: "cli",
importPath: moduleImportPrefix + "almanac/register", 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", name: "black-box integration test may compose families",
filename: "internal/modules/integration/example_test.go", filename: "internal/modules/integration/example_test.go",
@@ -183,6 +277,9 @@ func TestImportBoundaryRules(t *testing.T) {
if tt.wantError && err == nil { if tt.wantError && err == nil {
t.Fatal("validateImport() error = nil, want boundary violation") 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 { if !tt.wantError && err != nil {
t.Fatalf("validateImport() error = %v, want nil", err) 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) return fmt.Errorf("parse import in %s: %w", relative, err)
} }
if err := validateImport(relative, parsed.Name.Name, importPath); err != nil { if err := validateImport(relative, parsed.Name.Name, importPath); err != nil {
return fmt.Errorf("%s imports %s: %w", relative, importPath, err) return err
} }
} }
return nil return nil
} }
func validateImport(filename string, sourcePackage string, importPath string) error { func validateImport(filename string, sourcePackage string, importPath string) error {
if isBlackBoxIntegrationTest(filename, sourcePackage) { target, ok := moduleTargetForImport(importPath)
if !ok {
return nil return nil
} }
targetFamily, targetChild := moduleFamilyForImport(importPath) isTest := strings.HasSuffix(filename, "_test.go")
if targetFamily == "" { if target.integration && !isTest {
return nil return importBoundaryViolation(filename, importPath, "module integration infrastructure is not a production dependency target")
} }
if isIntegrationFile(filename) { 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) sourceFamily, sourceRoot, sourceRegistrar := moduleFamilyForFile(filename)
if sourceFamily == "" { if sourceFamily == "" {
return nil return nil
} }
if sourceRoot && sourceFamily == targetFamily && targetChild { if sourceRoot && sourceFamily == target.family && target.child {
return fmt.Errorf("family root must not import child packages") return importBoundaryViolation(filename, importPath, "family root must not import child packages")
} }
if sourceFamily == targetFamily { if sourceFamily == target.family {
return nil return nil
} }
if sourceFamily == "generic" { 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 { if sourceRegistrar {
return nil 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) { 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" 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 { func isIntegrationFile(filename string) bool {
return strings.HasPrefix(filename, "internal/modules/integration/") return strings.HasPrefix(filename, "internal/modules/integration/")
} }