Compare commits
3 Commits
3772b308e9
...
68481804a7
| Author | SHA1 | Date | |
|---|---|---|---|
| 68481804a7 | |||
| 236ccc62ad | |||
| 35bffdf336 |
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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{
|
||||||
|
|||||||
@@ -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...)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
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()
|
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
|
||||||
|
|||||||
@@ -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())
|
||||||
|
|||||||
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"
|
"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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -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) {
|
||||||
|
|||||||
@@ -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] = '['
|
||||||
|
|||||||
@@ -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/")
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user