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

Final cleanup after checkpoint 3 and remove the completed implementation plan

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-03 11:26:49 -05:00
parent 25fbd791c8
commit d6dadb6c70
5 changed files with 235 additions and 924 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
internal/framework/contracts/contracts.go
View File

@@ -35,10 +35,12 @@ type StructuredLLMClient interface {
}
type ParseRequest struct {
SourceID string `json:"source_id,omitempty"`
Path string `json:"path,omitempty"`
Raw []byte `json:"-"`
Metadata map[string]any `json:"metadata,omitempty"`
SourceID string `json:"source_id,omitempty"`
Path string `json:"path,omitempty"`
Raw []byte `json:"-"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type InputAdapter interface {
@@ -55,8 +57,10 @@ type SourceChunk struct {
}
type ChunkRequest struct {
Source *source.SourceDocument `json:"-"`
Metadata map[string]any `json:"metadata,omitempty"`
Source *source.SourceDocument `json:"-"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type ChunkResult struct {
@@ -74,6 +78,8 @@ type ExtractionRequest struct {
Chunk *SourceChunk `json:"chunk,omitempty"`
AmbientContext map[string]any `json:"ambient_context,omitempty"`
LLMClient StructuredLLMClient `json:"-"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
@@ -99,6 +105,8 @@ type MergeRequest struct {
Source *source.SourceDocument `json:"-"`
LaneID string `json:"lane_id"`
ChunkArtifacts []ChunkArtifacts `json:"chunk_artifacts"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
@@ -116,6 +124,8 @@ type NormalizeRequest struct {
Source *source.SourceDocument `json:"-"`
LaneID string `json:"lane_id"`
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
@@ -132,6 +142,8 @@ type Normalizer interface {
type ValidationRequest struct {
Source *source.SourceDocument `json:"-"`
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
@@ -161,11 +173,13 @@ type Warning struct {
}
type OutputRequest struct {
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []Warning `json:"warnings,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []Warning `json:"warnings,omitempty"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type OutputResult struct {

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

@@ -65,10 +65,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err)
}
doc, err := adapter.Parse(ctx, contracts.ParseRequest{
SourceID: input.SourceID,
Path: input.Path,
Raw: input.RawInput,
Metadata: input.Metadata,
SourceID: input.SourceID,
Path: input.Path,
Raw: input.RawInput,
LLMProfile: input.Pipeline.Input.LLMProfile,
Options: cloneOptions(input.Pipeline.Input.Options),
Metadata: input.Metadata,
})
if err != nil {
return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err)
@@ -83,8 +85,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err)
}
chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{
Source: doc,
Metadata: input.Metadata,
Source: doc,
LLMProfile: input.Pipeline.Chunk.LLMProfile,
Options: cloneOptions(input.Pipeline.Chunk.Options),
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, chunkResult.Warnings...)
if err != nil {
@@ -112,11 +116,13 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err)
}
encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
Manifest: output.Manifest,
Approved: output.Approved,
Rejected: output.Rejected,
Warnings: output.Warnings,
Metadata: input.Metadata,
Manifest: output.Manifest,
Approved: output.Approved,
Rejected: output.Rejected,
Warnings: output.Warnings,
LLMProfile: input.Pipeline.Output.LLMProfile,
Options: cloneOptions(input.Pipeline.Output.Options),
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, encoded.Warnings...)
if err != nil {
@@ -142,24 +148,28 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err)
}
var validators []contracts.Validator
var validators []validatorExecution
if len(lane.Validators) > 0 {
validators, err = r.buildConfiguredValidators(lane)
if err != nil {
return err
}
} else {
validators = extractor.Validators()
for _, validator := range extractor.Validators() {
validators = append(validators, validatorExecution{validator: validator})
}
}
chunkArtifacts := make([]contracts.ChunkArtifacts, 0, len(chunks))
for index := range chunks {
chunk := chunks[index]
result, err := extractor.Extract(ctx, contracts.ExtractionRequest{
Source: doc,
Chunk: &chunk,
LLMClient: input.LLMClient,
Metadata: input.Metadata,
Source: doc,
Chunk: &chunk,
LLMClient: input.LLMClient,
LLMProfile: lane.Extract.LLMProfile,
Options: cloneOptions(lane.Extract.Options),
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, result.Warnings...)
if err != nil {
@@ -180,6 +190,8 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
Source: doc,
LaneID: lane.ID,
ChunkArtifacts: chunkArtifacts,
LLMProfile: lane.Merge.LLMProfile,
Options: cloneOptions(lane.Merge.Options),
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, mergeResult.Warnings...)
@@ -191,6 +203,8 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
Source: doc,
LaneID: lane.ID,
Candidates: mergeResult.Candidates,
LLMProfile: lane.Normalize.LLMProfile,
Options: cloneOptions(lane.Normalize.Options),
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, normalizeResult.Warnings...)
@@ -198,6 +212,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
return fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err)
}
if err := validateCandidateEnvelope(extractor, normalizeResult.Candidates); err != nil {
return fmt.Errorf("validate normalized candidates for lane %q: %w", lane.ID, err)
}
approved, rejected, warnings, err := runValidators(ctx, extractor.Key(), validators, doc, normalizeResult.Candidates, input.Metadata)
output.Warnings = append(output.Warnings, warnings...)
output.Rejected = append(output.Rejected, rejected...)
@@ -211,14 +229,22 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
return nil
}
func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]contracts.Validator, error) {
validators := make([]contracts.Validator, 0, len(lane.Validators))
type validatorExecution struct {
validator contracts.Validator
binding ModuleBinding
}
func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]validatorExecution, error) {
validators := make([]validatorExecution, 0, len(lane.Validators))
for _, binding := range lane.Validators {
validator, err := r.registries.Validators.Build(binding.Module)
if err != nil {
return nil, fmt.Errorf("build validator %q for lane %q: %w", binding.Module, lane.ID, err)
}
validators = append(validators, validator)
validators = append(validators, validatorExecution{
validator: validator,
binding: binding,
})
}
return validators, nil
}
@@ -359,18 +385,51 @@ func normalizeCandidates(extractor contracts.Extractor, candidates []artifacts.A
return normalized, nil
}
func runValidators(ctx context.Context, extractorKey string, validators []contracts.Validator, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) {
func validateCandidateEnvelope(extractor contracts.Extractor, candidates []artifacts.ArtifactCandidate) error {
seen := make(map[int]struct{}, len(candidates))
for _, candidate := range candidates {
if _, ok := seen[candidate.Index]; ok {
return fmt.Errorf("candidate index %d is duplicated", candidate.Index)
}
seen[candidate.Index] = struct{}{}
if candidate.ExtractorKey == "" {
return fmt.Errorf("candidate index %d extractor_key must not be empty", candidate.Index)
}
if candidate.ExtractorKey != extractor.Key() {
return fmt.Errorf("candidate index %d extractor_key %q does not match extractor %q", candidate.Index, candidate.ExtractorKey, extractor.Key())
}
if candidate.ArtifactType == "" {
return fmt.Errorf("candidate index %d artifact_type must not be empty", candidate.Index)
}
if candidate.ArtifactType != extractor.ArtifactType() {
return fmt.Errorf("candidate index %d artifact_type %q does not match extractor %q artifact type %q", candidate.Index, candidate.ArtifactType, extractor.Key(), extractor.ArtifactType())
}
if candidate.SchemaVersion == "" {
return fmt.Errorf("candidate index %d schema_version must not be empty", candidate.Index)
}
if candidate.SchemaVersion != extractor.SchemaVersion() {
return fmt.Errorf("candidate index %d schema_version %q does not match extractor %q schema version %q", candidate.Index, candidate.SchemaVersion, extractor.Key(), extractor.SchemaVersion())
}
}
return nil
}
func runValidators(ctx context.Context, extractorKey string, validators []validatorExecution, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) {
eligible := candidates
var rejected []artifacts.RejectedArtifact
var warnings []contracts.Warning
for validatorIndex, validator := range validators {
for validatorIndex, execution := range validators {
validator := execution.validator
if validator == nil {
return nil, rejected, warnings, fmt.Errorf("extractor %q validator[%d] must not be nil", extractorKey, validatorIndex)
}
result, err := validator.Validate(ctx, contracts.ValidationRequest{
Source: doc,
Candidates: eligible,
LLMProfile: execution.binding.LLMProfile,
Options: cloneOptions(execution.binding.Options),
Metadata: metadata,
})
warnings = append(warnings, result.Warnings...)

122
internal/framework/pipeline/runner_test.go
View File

@@ -311,6 +311,66 @@ func TestRunPassesInputRequestFields(t *testing.T) {
}
}
func TestRunPassesModuleBindingConfigToStageRequests(t *testing.T) {
modules := defaultRunnerModules()
pipeline := resolvedPipelineWithValidators("configured")
pipeline.Input = ModuleBinding{Module: "input", LLMProfile: "input-profile", Options: map[string]any{"input_option": "input-value"}}
pipeline.Chunk = ModuleBinding{Module: "chunk", LLMProfile: "chunk-profile", Options: map[string]any{"chunk_option": "chunk-value"}}
pipeline.Output = ModuleBinding{Module: "output", LLMProfile: "output-profile", Options: map[string]any{"output_option": "output-value"}}
pipeline.ArtifactLanes[0].Extract = ModuleBinding{Module: "extract-alpha", LLMProfile: "extract-profile", Options: map[string]any{"extract_option": "extract-value"}}
pipeline.ArtifactLanes[0].Merge = ModuleBinding{Module: "merge", LLMProfile: "merge-profile", Options: map[string]any{"merge_option": "merge-value"}}
pipeline.ArtifactLanes[0].Normalize = ModuleBinding{Module: "normalize", LLMProfile: "normalize-profile", Options: map[string]any{"normalize_option": "normalize-value"}}
pipeline.ArtifactLanes[0].Validators[0] = ModuleBinding{Module: "configured", LLMProfile: "validator-profile", Options: map[string]any{"validator_option": "validator-value"}}
_, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if got := modules.input.requests[0].LLMProfile; got != "input-profile" {
t.Fatalf("input LLMProfile = %q, want input-profile", got)
}
if got := modules.input.requests[0].Options["input_option"]; got != "input-value" {
t.Fatalf("input Options = %#v, want input option", modules.input.requests[0].Options)
}
if got := modules.chunker.requests[0].LLMProfile; got != "chunk-profile" {
t.Fatalf("chunk LLMProfile = %q, want chunk-profile", got)
}
if got := modules.chunker.requests[0].Options["chunk_option"]; got != "chunk-value" {
t.Fatalf("chunk Options = %#v, want chunk option", modules.chunker.requests[0].Options)
}
if got := modules.extractors["extract-alpha"].requests[0].LLMProfile; got != "extract-profile" {
t.Fatalf("extract LLMProfile = %q, want extract-profile", got)
}
if got := modules.extractors["extract-alpha"].requests[0].Options["extract_option"]; got != "extract-value" {
t.Fatalf("extract Options = %#v, want extract option", modules.extractors["extract-alpha"].requests[0].Options)
}
if got := modules.mergers["merge"].requests[0].LLMProfile; got != "merge-profile" {
t.Fatalf("merge LLMProfile = %q, want merge-profile", got)
}
if got := modules.mergers["merge"].requests[0].Options["merge_option"]; got != "merge-value" {
t.Fatalf("merge Options = %#v, want merge option", modules.mergers["merge"].requests[0].Options)
}
if got := modules.normalizers["normalize"].requests[0].LLMProfile; got != "normalize-profile" {
t.Fatalf("normalize LLMProfile = %q, want normalize-profile", got)
}
if got := modules.normalizers["normalize"].requests[0].Options["normalize_option"]; got != "normalize-value" {
t.Fatalf("normalize Options = %#v, want normalize option", modules.normalizers["normalize"].requests[0].Options)
}
if got := modules.validators["configured"].requests[0].LLMProfile; got != "validator-profile" {
t.Fatalf("validator LLMProfile = %q, want validator-profile", got)
}
if got := modules.validators["configured"].requests[0].Options["validator_option"]; got != "validator-value" {
t.Fatalf("validator Options = %#v, want validator option", modules.validators["configured"].requests[0].Options)
}
if got := modules.output.requests[0].LLMProfile; got != "output-profile" {
t.Fatalf("output LLMProfile = %q, want output-profile", got)
}
if got := modules.output.requests[0].Options["output_option"]; got != "output-value" {
t.Fatalf("output Options = %#v, want output option", modules.output.requests[0].Options)
}
}
func TestRunPassesPerChunkCandidatesToMergeAndNormalize(t *testing.T) {
modules := defaultRunnerModules()
@@ -346,6 +406,54 @@ func TestRunPassesPerChunkCandidatesToMergeAndNormalize(t *testing.T) {
}
}
func TestRunRejectsInvalidPostNormalizeCandidateEnvelope(t *testing.T) {
tests := []struct {
name string
candidates []artifacts.ArtifactCandidate
want string
}{
{
name: "duplicate index",
candidates: []artifacts.ArtifactCandidate{
runnerCandidate(0),
runnerCandidate(0),
},
want: "duplicated",
},
{
name: "missing extractor key",
candidates: []artifacts.ArtifactCandidate{
{Index: 0, ArtifactType: "artifact", SchemaVersion: "v1", Payload: []byte(`{"value":true}`)},
},
want: "extractor_key",
},
{
name: "mismatched schema version",
candidates: []artifacts.ArtifactCandidate{
{Index: 0, ExtractorKey: "extract-alpha", ArtifactType: "artifact", SchemaVersion: "other", Payload: []byte(`{"value":true}`)},
},
want: "schema_version",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
modules := defaultRunnerModules()
modules.normalizers["normalize"].result = test.candidates
output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()})
assertRunError(t, err, test.want)
if output.Manifest.ValidationStatus != "failed" {
t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus)
}
if len(output.Approved) != 0 {
t.Fatalf("len(Approved) = %d, want no approved artifacts", len(output.Approved))
}
})
}
}
func TestRunValidatorApprovalAndRejection(t *testing.T) {
modules := defaultRunnerModules()
rejectFirst := &runnerValidator{
@@ -784,6 +892,7 @@ type runnerExtractor struct {
validators []contracts.Validator
warnings []contracts.Warning
err error
requests []contracts.ExtractionRequest
seenChunkIDs []string
seenLLMClients []contracts.StructuredLLMClient
seenMetadata []map[string]any
@@ -806,6 +915,7 @@ func (extractor *runnerExtractor) Validators() []contracts.Validator {
}
func (extractor *runnerExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
extractor.requests = append(extractor.requests, req)
if req.Chunk != nil {
extractor.seenChunkIDs = append(extractor.seenChunkIDs, req.Chunk.ID)
}
@@ -880,6 +990,7 @@ type runnerValidator struct {
err error
order *[]string
calls int
requests []contracts.ValidationRequest
}
func (validator *runnerValidator) Name() string {
@@ -888,6 +999,7 @@ func (validator *runnerValidator) Name() string {
func (validator *runnerValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
validator.calls++
validator.requests = append(validator.requests, req)
if validator.order != nil {
*validator.order = append(*validator.order, validator.name)
}
@@ -981,6 +1093,16 @@ func candidateIndices(candidates []artifacts.ArtifactCandidate) []int {
return indices
}
func runnerCandidate(index int) artifacts.ArtifactCandidate {
return artifacts.ArtifactCandidate{
Index: index,
ExtractorKey: "extract-alpha",
ArtifactType: "artifact",
SchemaVersion: "v1",
Payload: []byte(`{"value":true}`),
}
}
func assertRunError(t *testing.T, err error, want string) {
t.Helper()