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

Serialize typed artifacts at durable boundaries

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 07:33:49 +00:00
parent 66de1a5520
commit 814fcdc6ba
23 changed files with 624 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -56,12 +56,12 @@ type RunInput struct {
}
type RunOutput struct {
Manifest artifacts.RunManifest `json:"manifest"`
NormalizeOutputs []contracts.NormalizeOutput `json:"normalize_outputs,omitempty"`
Rejected []contracts.RejectedOutput `json:"rejected,omitempty"`
Warnings []contracts.Warning `json:"warnings,omitempty"`
OutputFiles []contracts.OutputFile `json:"-"`
CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"`
Manifest artifacts.RunManifest `json:"manifest"`
NormalizeOutputs []contracts.SerializedOutput `json:"normalize_outputs,omitempty"`
Rejected []contracts.RejectedOutput `json:"rejected,omitempty"`
Warnings []contracts.Warning `json:"warnings,omitempty"`
OutputFiles []contracts.OutputFile `json:"-"`
CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"`
}
func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) {
@@ -335,7 +335,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
StartedAt: outputStarted,
Payload: map[string]any{
"manifest": output.Manifest,
"normalize_outputs": debugNormalizeOutputEnvelopes(output.NormalizeOutputs),
"normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs),
"rejected": debugRejectedOutputEnvelopes(output.Rejected),
"warnings": output.Warnings,
"options": redactSensitiveMap(input.pipeline.Output.Options),
@@ -346,7 +346,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
}
encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
Manifest: output.Manifest,
NormalizeOutputs: cloneNormalizeOutputs(output.NormalizeOutputs),
NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs),
Rejected: cloneRejectedOutputs(output.Rejected),
Warnings: output.Warnings,
LLMProfile: input.pipeline.Output.LLMProfile,
@@ -865,7 +865,7 @@ func (r *Runner) runLegacyLane(ctx context.Context, input RunInput, checkpoints
}); err != nil {
return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err)
}
output.NormalizeOutputs = append(output.NormalizeOutputs, acceptedNormalize)
output.NormalizeOutputs = append(output.NormalizeOutputs, serializedOutputFromLegacy(acceptedNormalize))
return nil
}
@@ -1282,7 +1282,7 @@ func populateRawOutputManifest(output *RunOutput) {
output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected)
}
func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts.NormalizedOutputManifest {
func normalizedOutputManifests(outputs []contracts.SerializedOutput) []artifacts.NormalizedOutputManifest {
if len(outputs) == 0 {
return nil
}
@@ -1292,11 +1292,11 @@ func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts.
LaneID: output.LaneID,
ModuleKey: output.NormalizerKey,
SourceID: output.SourceID,
MediaType: output.Payload.MediaType,
MediaType: output.Artifact.MediaType,
Schema: artifacts.OutputSchemaProvenance{
ID: output.Schema.ID,
Name: output.Schema.Name,
Version: output.Schema.Version,
ID: output.Artifact.Schema.ID,
Name: output.Artifact.Schema.Name,
Version: output.Artifact.Schema.Version,
},
})
}
@@ -1637,17 +1637,27 @@ func cloneNormalizeOutput(output contracts.NormalizeOutput) contracts.NormalizeO
return output
}
func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput {
func cloneSerializedOutputs(outputs []contracts.SerializedOutput) []contracts.SerializedOutput {
if len(outputs) == 0 {
return nil
}
out := make([]contracts.NormalizeOutput, 0, len(outputs))
out := make([]contracts.SerializedOutput, 0, len(outputs))
for _, output := range outputs {
out = append(out, cloneNormalizeOutput(output))
out = append(out, contracts.CloneSerializedOutput(output))
}
return out
}
func serializedOutputFromLegacy(output contracts.NormalizeOutput) contracts.SerializedOutput {
return contracts.SerializedOutput{
LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID,
Artifact: contracts.SerializedArtifact{
Schema: contracts.ArtifactSchema{ID: output.Schema.ID, Name: output.Schema.Name, Version: output.Schema.Version, JSONSchema: append([]byte(nil), output.Schema.JSONSchema...)},
MediaType: output.Payload.MediaType, Content: append([]byte(nil), output.Payload.Content...), Metadata: cloneMetadata(output.Payload.Metadata),
},
}
}
func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput {
if len(rejected) == 0 {
return nil