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

Bind prepared references to extraction identity

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-29 15:24:31 +00:00
parent 51e0e8c5d0
commit 495f7bcde4
5 changed files with 376 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
internal/stage/extract.go
View File

@@ -80,6 +80,10 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("extract: resolve final-trimmed transcript identity: %w", err)
}
references, err := resolveExtractReferences(paths, m, notariusConfig, sessionID)
if err != nil {
return nil, fmt.Errorf("extract: resolve Notarius references: %w", err)
}
timeout, err := time.ParseDuration(strings.TrimSpace(notariusConfig.Timeout))
if err != nil || timeout <= 0 {
@@ -125,14 +129,14 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
}
}
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory, input)
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory, input, references.Identities)
if err != nil {
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
}
request := notarius.RunRequest{
Binary: resolvedBinary, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout,
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, References: references.Bindings,
}
adapterResult, err := env.Notarius.Run(ctx, request)
if err != nil {
@@ -249,6 +253,8 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
"narratio_run_id": runID,
"configuration_fingerprint": fingerprint,
"direct_input": input.Metadata(),
"reference_count": len(references.Identities),
"references": extractReferenceMetadata(references.Identities),
"receipt": map[string]any{
"run_id": adapterResult.Receipt.RunID, "pipeline_id": adapterResult.Receipt.PipelineID,
"normalized_output_count": adapterResult.Receipt.NormalizedOutputCount,
@@ -379,6 +385,7 @@ type fingerprintDocument struct {
Timeout string `json:"timeout"`
WorkingDirectory string `json:"working_directory"`
Input artifacts.ExtractionInputIdentity `json:"input"`
References []extractReferenceIdentity `json:"references"`
Outputs []fingerprintOutput `json:"outputs"`
}
@@ -388,6 +395,7 @@ func extractionFingerprint(
timeout time.Duration,
workingDirectory string,
input artifacts.ExtractionInputIdentity,
references []extractReferenceIdentity,
) (string, error) {
keys := make([]string, 0, len(cfg.Outputs))
for key := range cfg.Outputs {
@@ -402,9 +410,20 @@ func extractionFingerprint(
SchemaID: output.SchemaID, SchemaVersion: output.SchemaVersion, ModuleKey: output.ModuleKey,
})
}
sortedReferences := append([]extractReferenceIdentity(nil), references...)
sort.Slice(sortedReferences, func(i, j int) bool {
if sortedReferences[i].Selector != sortedReferences[j].Selector {
return sortedReferences[i].Selector < sortedReferences[j].Selector
}
if sortedReferences[i].SourceID != sortedReferences[j].SourceID {
return sortedReferences[i].SourceID < sortedReferences[j].SourceID
}
return sortedReferences[i].Path < sortedReferences[j].Path
})
payload, err := json.Marshal(fingerprintDocument{
Binary: binary, ConfigPath: configPath, PipelineID: cfg.PipelineID,
Timeout: timeout.String(), WorkingDirectory: workingDirectory, Input: input, Outputs: outputs,
Timeout: timeout.String(), WorkingDirectory: workingDirectory, Input: input,
References: sortedReferences, Outputs: outputs,
})
if err != nil {
return "", err

107
internal/stage/extract_references.go Normal file
View File

@@ -0,0 +1,107 @@
package stage
import (
"fmt"
"sort"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/notariusref"
)
type extractReferenceIdentity struct {
Selector string `json:"selector"`
SourceID string `json:"source_id"`
Path string `json:"path"`
Checksum string `json:"checksum"`
SizeBytes int64 `json:"size_bytes"`
}
type resolvedExtractReferences struct {
Bindings []notarius.ReferenceBinding
Identities []extractReferenceIdentity
}
func resolveExtractReferences(
paths artifacts.SessionPaths,
m *manifest.Manifest,
cfg *config.NotariusConfig,
sessionID string,
) (resolvedExtractReferences, error) {
if cfg == nil || len(cfg.References) == 0 {
return resolvedExtractReferences{
Bindings: []notarius.ReferenceBinding{},
Identities: []extractReferenceIdentity{},
}, nil
}
type configuredReference struct {
selector string
sourceID string
}
configured := make([]configuredReference, 0, len(cfg.References))
seen := make(map[string]struct{}, len(cfg.References))
for rawSelector, rawSourceID := range cfg.References {
selector, err := notariusref.NormalizeSelector(rawSelector)
if err != nil {
return resolvedExtractReferences{}, fmt.Errorf("reference selector %q is invalid: %w", rawSelector, err)
}
if _, duplicate := seen[selector]; duplicate {
return resolvedExtractReferences{}, fmt.Errorf("reference selector %q is configured more than once after normalization", selector)
}
seen[selector] = struct{}{}
configured = append(configured, configuredReference{selector: selector, sourceID: strings.TrimSpace(rawSourceID)})
}
sort.Slice(configured, func(i, j int) bool { return configured[i].selector < configured[j].selector })
resolved := resolvedExtractReferences{
Bindings: make([]notarius.ReferenceBinding, 0, len(configured)),
Identities: make([]extractReferenceIdentity, 0, len(configured)),
}
for _, reference := range configured {
selector := reference.selector
sourceID := reference.sourceID
identity, err := artifacts.ResolvePreparedInput(paths, m, sourceID)
if err != nil {
return resolvedExtractReferences{}, fmt.Errorf(
"reference %q source %q is unavailable or invalid; run narratio run-stage prepare %s --force: %w",
selector,
sourceID,
sessionID,
err,
)
}
resolved.Bindings = append(resolved.Bindings, notarius.ReferenceBinding{
Selector: selector,
Path: identity.Path,
})
resolved.Identities = append(resolved.Identities, extractReferenceIdentity{
Selector: selector,
SourceID: identity.SourceID,
Path: identity.RelativePath,
Checksum: identity.Checksum,
SizeBytes: identity.Size,
})
}
return resolved, nil
}
func extractReferenceMetadata(identities []extractReferenceIdentity) []map[string]any {
limit := len(identities)
if limit > config.MaxNotariusReferenceBindings {
limit = config.MaxNotariusReferenceBindings
}
metadata := make([]map[string]any, 0, limit)
for _, identity := range identities[:limit] {
metadata = append(metadata, map[string]any{
"selector": identity.Selector,
"source_id": identity.SourceID,
"path": identity.Path,
"checksum": identity.Checksum,
"size_bytes": identity.SizeBytes,
})
}
return metadata
}

6
internal/stage/extract_resume.go
View File

@@ -49,7 +49,11 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
if err != nil {
return ResumeValidation{}, fmt.Errorf("extract resume: resolve final-trimmed transcript identity: %w", err)
}
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory, input)
references, err := resolveExtractReferences(paths, m, cfg, sessionID)
if err != nil {
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius references: %w", err)
}
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory, input, references.Identities)
if err != nil {
return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err)
}

244
internal/stage/extract_test.go
View File

@@ -15,6 +15,7 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
@@ -75,11 +76,108 @@ func TestExtractStageResolvesManifestInputAndBuildsExactRequest(t *testing.T) {
if !filepath.IsAbs(req.Binary) {
t.Fatalf("binary = %q, want absolute resolved path", req.Binary)
}
if len(req.References) != 0 {
t.Fatalf("references = %#v, want none", req.References)
}
if result.Metadata["reference_count"] != 0 || len(result.Metadata["references"].([]map[string]any)) != 0 {
t.Fatalf("reference metadata = count %#v references %#v", result.Metadata["reference_count"], result.Metadata["references"])
}
if len(result.Outputs) != 2 {
t.Fatalf("outputs = %#v, want lane and index", result.Outputs)
}
}
func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T) {
env, m, fake := setupExtractEnv(t)
configured := []struct {
selector string
sourceID string
contents string
}{
{selector: " players ", sourceID: artifactpolicy.SourceInputPlayers, contents: "players-secret\n"},
{selector: "spells.spell_catalog", sourceID: artifactpolicy.SourceInputSpellCatalog, contents: "spell-secret\n"},
{selector: "glossary", sourceID: artifactpolicy.SourceInputGlossary, contents: "glossary-secret\n"},
{selector: "party", sourceID: artifactpolicy.SourceInputParty, contents: "party-secret\n"},
}
env.Config.Pipeline.Notarius.References = make(map[string]string, len(configured))
wantPaths := make(map[string]string, len(configured))
wantSources := make(map[string]string, len(configured))
for _, reference := range configured {
env.Config.Pipeline.Notarius.References[reference.selector] = reference.sourceID
selector := strings.TrimSpace(reference.selector)
wantPaths[selector] = recordPreparedExtractInput(t, env, m, reference.sourceID, reference.contents)
wantSources[selector] = reference.sourceID
}
result, err := (extractStage{}).Run(context.Background(), env, m)
if err != nil {
t.Fatalf("Run() error = %v", err)
}
wantSelectors := []string{"glossary", "party", "players", "spells.spell_catalog"}
if len(fake.Requests) != 1 || len(fake.Requests[0].References) != len(wantSelectors) {
t.Fatalf("requests = %#v", fake.Requests)
}
for index, selector := range wantSelectors {
binding := fake.Requests[0].References[index]
if binding.Selector != selector || binding.Path != wantPaths[selector] || !filepath.IsAbs(binding.Path) {
t.Fatalf("reference[%d] = %#v, want selector %q path %q", index, binding, selector, wantPaths[selector])
}
}
if result.Metadata["reference_count"] != len(wantSelectors) {
t.Fatalf("reference_count = %#v", result.Metadata["reference_count"])
}
metadata, ok := result.Metadata["references"].([]map[string]any)
if !ok || len(metadata) != len(wantSelectors) {
t.Fatalf("references metadata = %#v", result.Metadata["references"])
}
for index, selector := range wantSelectors {
entry := metadata[index]
if len(entry) != 5 || entry["selector"] != selector || entry["source_id"] != wantSources[selector] {
t.Fatalf("reference metadata[%d] = %#v", index, entry)
}
if path, _ := entry["path"].(string); !strings.HasPrefix(path, "inputs/") || filepath.IsAbs(path) {
t.Fatalf("reference metadata path = %#v", entry["path"])
}
if entry["checksum"] == "" || entry["size_bytes"] == int64(0) {
t.Fatalf("reference identity metadata = %#v", entry)
}
}
encoded, err := json.Marshal(result.Metadata)
if err != nil {
t.Fatalf("json.Marshal(metadata) error = %v", err)
}
for _, reference := range configured {
if strings.Contains(string(encoded), strings.TrimSpace(reference.contents)) {
t.Fatalf("metadata contains reference payload %q: %s", reference.contents, encoded)
}
}
}
func TestExtractStageRejectsUnavailableReferenceBeforeInvocationOrRunDirectoryCreation(t *testing.T) {
env, m, fake := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{
"spell_catalog": artifactpolicy.SourceInputSpellCatalog,
}
runRoot := artifacts.SessionRunRootForCampaign(
env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID,
)
if err := os.RemoveAll(runRoot); err != nil {
t.Fatalf("RemoveAll(run root) error = %v", err)
}
result, err := (extractStage{}).Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "run narratio run-stage prepare ") || !strings.Contains(err.Error(), "--force") {
t.Fatalf("Run() result = %#v error = %v", result, err)
}
if len(fake.Requests) != 0 {
t.Fatalf("adapter requests = %#v", fake.Requests)
}
if _, statErr := os.Stat(runRoot); !os.IsNotExist(statErr) {
t.Fatalf("run root stat error = %v, want not exist", statErr)
}
}
func TestExtractStageProducesImmutableManifestReadyOutputs(t *testing.T) {
env, m, fake := setupExtractEnv(t)
fixture := extractFixtureFromEnv(t, env, m)
@@ -360,11 +458,15 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
second := &config.NotariusConfig{PipelineID: "pipeline", Outputs: map[string]config.NotariusOutputConfig{
"alpha": first.Outputs["alpha"], "zeta": first.Outputs["zeta"],
}}
one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work", input)
references := []extractReferenceIdentity{
{Selector: "zeta", SourceID: artifactpolicy.SourceInputPlayers, Path: "inputs/players.yml", Checksum: "players-checksum", SizeBytes: 12},
{Selector: "alpha", SourceID: artifactpolicy.SourceInputParty, Path: "inputs/party.yml", Checksum: "party-checksum", SizeBytes: 34},
}
one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work", input, references)
if err != nil {
t.Fatalf("extractionFingerprint(first) error = %v", err)
}
two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input)
two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input, []extractReferenceIdentity{references[1], references[0]})
if err != nil {
t.Fatalf("extractionFingerprint(second) error = %v", err)
}
@@ -384,7 +486,7 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
t.Run(test.name, func(t *testing.T) {
changedInput := input
test.mutate(&changedInput)
changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", changedInput)
changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", changedInput, references)
if err != nil {
t.Fatalf("extractionFingerprint(changed input) error = %v", err)
}
@@ -393,10 +495,36 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
}
})
}
for _, test := range []struct {
name string
mutate func(*extractReferenceIdentity)
}{
{name: "selector", mutate: func(identity *extractReferenceIdentity) { identity.Selector = "changed" }},
{name: "source", mutate: func(identity *extractReferenceIdentity) { identity.SourceID = artifactpolicy.SourceInputGlossary }},
{name: "path", mutate: func(identity *extractReferenceIdentity) { identity.Path = "inputs/changed.yml" }},
{name: "checksum", mutate: func(identity *extractReferenceIdentity) { identity.Checksum = "changed-checksum" }},
{name: "size", mutate: func(identity *extractReferenceIdentity) { identity.SizeBytes++ }},
} {
t.Run("reference "+test.name, func(t *testing.T) {
changedReferences := append([]extractReferenceIdentity(nil), references...)
test.mutate(&changedReferences[0])
changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input, changedReferences)
if err != nil {
t.Fatalf("extractionFingerprint(changed reference) error = %v", err)
}
if one == changed {
t.Fatalf("fingerprint did not change with reference %s", test.name)
}
})
}
}
func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T) {
env, m, _ := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{
"party": artifactpolicy.SourceInputParty,
}
recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "party-reference\n")
seedSucceededExtractResult(t, env, m)
m.RunID = "20260810T020304Z-fedcba98"
@@ -409,6 +537,93 @@ func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T)
}
}
func TestExtractStageResumeValidationRejectsChangedPreparedReferenceIdentity(t *testing.T) {
env, m, _ := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{
"party": artifactpolicy.SourceInputParty,
}
path := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "original-party\n")
seedSucceededExtractResult(t, env, m)
if err := os.WriteFile(path, []byte("changed-party\n"), 0o644); err != nil {
t.Fatalf("WriteFile(reference) error = %v", err)
}
checksum, err := artifacts.SHA256File(path)
if err != nil {
t.Fatalf("SHA256File(reference) error = %v", err)
}
descriptor, _ := artifactpolicy.DescribePreparedInputSource(artifactpolicy.SourceInputParty)
for index := range m.Inputs {
if m.Inputs[index].Kind == descriptor.ManifestKind {
m.Inputs[index].Checksum = checksum
}
}
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
if err != nil {
t.Fatalf("ValidateResume() error = %v", err)
}
if validation.Resumable || !strings.Contains(validation.Reason, "invocation contract changed") {
t.Fatalf("validation = %#v, want changed invocation contract", validation)
}
}
func TestExtractStageResumeValidationErrorsForUnavailablePreparedReference(t *testing.T) {
for _, test := range []struct {
name string
mutate func(*testing.T, string, *manifest.Manifest)
}{
{name: "missing", mutate: func(t *testing.T, path string, _ *manifest.Manifest) {
if err := os.Remove(path); err != nil {
t.Fatalf("Remove(reference) error = %v", err)
}
}},
{name: "unsafe manifest path", mutate: func(t *testing.T, _ string, m *manifest.Manifest) {
descriptor, _ := artifactpolicy.DescribePreparedInputSource(artifactpolicy.SourceInputParty)
for index := range m.Inputs {
if m.Inputs[index].Kind == descriptor.ManifestKind {
m.Inputs[index].Path = filepath.Join(t.TempDir(), "outside.yml")
}
}
}},
{name: "stale checksum", mutate: func(t *testing.T, path string, _ *manifest.Manifest) {
if err := os.WriteFile(path, []byte("tampered-party\n"), 0o644); err != nil {
t.Fatalf("WriteFile(reference) error = %v", err)
}
}},
} {
t.Run(test.name, func(t *testing.T) {
env, m, _ := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{
"party": artifactpolicy.SourceInputParty,
}
path := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "original-party\n")
seedSucceededExtractResult(t, env, m)
test.mutate(t, path, m)
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "run narratio run-stage prepare ") || !strings.Contains(err.Error(), "--force") {
t.Fatalf("ValidateResume() validation = %#v error = %v", validation, err)
}
})
}
}
func TestExtractStageResumeValidationDoesNotResolveReferencesWhenDisabled(t *testing.T) {
env, m, _ := setupExtractEnv(t)
env.Config.Pipeline.Notarius.Enabled = false
env.Config.Pipeline.Notarius.References = map[string]string{
"missing": artifactpolicy.SourceInputSpellCatalog,
}
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
if err != nil {
t.Fatalf("ValidateResume() error = %v", err)
}
if validation.Resumable || !strings.Contains(validation.Reason, "disabled") {
t.Fatalf("validation = %#v", validation)
}
}
func TestExtractStageAdapterResultIsImmediatelyReusableAndCatalogVisible(t *testing.T) {
env, m, fake := setupExtractEnv(t)
producerRunID := m.RunID
@@ -622,6 +837,29 @@ func extractFixtureFromEnv(t *testing.T, env *Env, m *manifest.Manifest) extract
}
}
func recordPreparedExtractInput(t *testing.T, env *Env, m *manifest.Manifest, sourceID, contents string) string {
t.Helper()
descriptor, ok := artifactpolicy.DescribePreparedInputSource(sourceID)
if !ok {
t.Fatalf("DescribePreparedInputSource(%q) ok = false", sourceID)
}
path := filepath.Join(sessionPathsForEnv(env, m.SessionID).InputsDir, descriptor.Filename)
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
t.Fatalf("MkdirAll(prepared input) error = %v", err)
}
if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
t.Fatalf("WriteFile(prepared input) error = %v", err)
}
checksum, err := artifacts.SHA256File(path)
if err != nil {
t.Fatalf("SHA256File(prepared input) error = %v", err)
}
m.Inputs = append(m.Inputs, manifest.InputRecord{
Kind: descriptor.ManifestKind, Path: path, Checksum: checksum,
})
return path
}
func setupExtractEnv(t *testing.T) (*Env, *manifest.Manifest, *notarius.FakeRunner) {
t.Helper()
workspace := t.TempDir()