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

Add run manifest and logical output file contracts

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-04 00:45:21 +00:00
parent ac14667797
commit bc4203a264
10 changed files with 455 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3,6 +3,9 @@ package pipeline
import (
"context"
"fmt"
"path"
"strings"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
@@ -29,21 +32,27 @@ func New(registries Registries) *Runner {
}
type RunInput struct {
Pipeline ResolvedPipeline
SourceID string
Path string
RawInput []byte
LLMClient contracts.StructuredLLMClient
Metadata map[string]any
Pipeline ResolvedPipeline
SourceID string
Path string
RawInput []byte
LLMClient contracts.StructuredLLMClient
RunID string
StartedAt time.Time
LLMProfiles []artifacts.LLMProfileManifest
Metadata map[string]any
}
type RunOutput struct {
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []contracts.Warning `json:"warnings,omitempty"`
EncodedOutput []byte `json:"-"`
ContentType string `json:"content_type,omitempty"`
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []contracts.Warning `json:"warnings,omitempty"`
OutputFiles []contracts.OutputFile `json:"-"`
// EncodedOutput is the legacy single-output payload. New callers should use OutputFiles.
EncodedOutput []byte `json:"-"`
// ContentType is the legacy single-output content type. New callers should use OutputFiles.
ContentType string `json:"content_type,omitempty"`
}
func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
@@ -58,7 +67,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
return output, err
}
output.Manifest = manifestFromPipeline(input.Pipeline)
output.Manifest = manifestFromPipeline(input)
adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module)
if err != nil {
@@ -110,6 +119,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
} else {
output.Manifest.ValidationStatus = "approved"
}
output.Manifest.CompletedAt = timePtr(time.Now().UTC())
encoder, err := r.registries.Outputs.Build(input.Pipeline.Output.Module)
if err != nil {
@@ -128,8 +138,15 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
if err != nil {
return failOutput(output), fmt.Errorf("encode output with encoder %q: %w", encoder.Key(), err)
}
output.EncodedOutput = encoded.Bytes
output.ContentType = encoded.ContentType
files, err := outputFilesFromResult(encoded)
if err != nil {
return failOutput(output), fmt.Errorf("validate output files from encoder %q: %w", encoder.Key(), err)
}
output.OutputFiles = files
if len(encoded.Files) == 0 {
output.EncodedOutput = append([]byte(nil), encoded.Bytes...)
output.ContentType = encoded.ContentType
}
return output, nil
}
@@ -147,6 +164,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
if err != nil {
return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err)
}
setLaneManifestMetadata(output, lane.ID, extractor, merger, normalizer)
var validators []validatorExecution
if len(lane.Validators) > 0 {
@@ -315,7 +333,17 @@ func validateRunInput(input RunInput) error {
return nil
}
func manifestFromPipeline(pipeline ResolvedPipeline) artifacts.RunManifest {
func manifestFromPipeline(input RunInput) artifacts.RunManifest {
startedAt := input.StartedAt
if startedAt.IsZero() {
startedAt = time.Now().UTC()
}
runID := strings.TrimSpace(input.RunID)
if runID == "" {
runID = fmt.Sprintf("run-%d", startedAt.UnixNano())
}
pipeline := input.Pipeline
manifest := artifacts.RunManifest{
PipelineID: pipeline.ID,
PipelineDigest: pipeline.Digest,
@@ -323,6 +351,9 @@ func manifestFromPipeline(pipeline ResolvedPipeline) artifacts.RunManifest {
Chunker: pipeline.Chunk.Module,
OutputEncoder: pipeline.Output.Module,
ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)),
RunID: runID,
StartedAt: timePtr(startedAt),
LLMProfiles: cloneLLMProfiles(input.LLMProfiles),
}
for _, lane := range pipeline.ArtifactLanes {
@@ -343,10 +374,124 @@ func manifestFromPipeline(pipeline ResolvedPipeline) artifacts.RunManifest {
func failOutput(output RunOutput) RunOutput {
if output.Manifest.PipelineID != "" {
output.Manifest.ValidationStatus = "failed"
output.Manifest.CompletedAt = timePtr(time.Now().UTC())
}
return output
}
func setLaneManifestMetadata(output *RunOutput, laneID string, modules ...any) {
if output == nil {
return
}
for i := range output.Manifest.ArtifactLanes {
if output.Manifest.ArtifactLanes[i].ID != laneID {
continue
}
metadata := make(map[string]any)
for _, module := range modules {
provider, ok := module.(contracts.ManifestMetadataProvider)
if !ok {
continue
}
moduleMetadata := cloneMetadata(provider.ManifestMetadata())
if len(moduleMetadata) == 0 {
continue
}
key := manifestMetadataKey(module)
if key == "" {
continue
}
metadata[key] = moduleMetadata
}
if len(metadata) > 0 {
output.Manifest.ArtifactLanes[i].Metadata = metadata
}
return
}
}
func manifestMetadataKey(module any) string {
switch module.(type) {
case contracts.Extractor:
return "extractor"
case contracts.Merger:
return "merger"
case contracts.Normalizer:
return "normalizer"
default:
return ""
}
}
func outputFilesFromResult(result contracts.OutputResult) ([]contracts.OutputFile, error) {
files := result.Files
if len(files) == 0 && len(result.Bytes) > 0 {
files = []contracts.OutputFile{
{
Name: "output",
ContentType: result.ContentType,
Bytes: result.Bytes,
},
}
}
out := make([]contracts.OutputFile, 0, len(files))
for _, file := range files {
if err := validateOutputFileName(file.Name); err != nil {
return nil, err
}
out = append(out, contracts.OutputFile{
Name: file.Name,
ContentType: file.ContentType,
Bytes: append([]byte(nil), file.Bytes...),
})
}
return out, nil
}
func validateOutputFileName(name string) error {
if strings.TrimSpace(name) == "" {
return fmt.Errorf("output file name must not be empty")
}
if strings.Contains(name, "\\") {
return fmt.Errorf("output file name %q must use slash-separated relative paths", name)
}
if path.IsAbs(name) {
return fmt.Errorf("output file name %q must be relative", name)
}
if strings.Contains(name, "..") {
return fmt.Errorf("output file name %q must not contain ..", name)
}
cleaned := path.Clean(name)
if cleaned == "." || cleaned != name {
return fmt.Errorf("output file name %q must be clean", name)
}
return nil
}
func cloneMetadata(metadata map[string]any) map[string]any {
if len(metadata) == 0 {
return nil
}
out := make(map[string]any, len(metadata))
for key, value := range metadata {
out[key] = value
}
return out
}
func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
if len(profiles) == 0 {
return nil
}
return append([]artifacts.LLMProfileManifest(nil), profiles...)
}
func timePtr(t time.Time) *time.Time {
return &t
}
func pipelineUsesConfiguredValidators(pipeline ResolvedPipeline) bool {
for _, lane := range pipeline.ArtifactLanes {
if len(lane.Validators) > 0 {