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

Snapshot verified references for extraction

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-29 16:18:40 +00:00
parent e7e3bef1e4
commit abfbe42d61
12 changed files with 265 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/config.md
View File

@@ -274,8 +274,9 @@ It validates only selector structure and the prepared source vocabulary;
Notarius owns target-slot declarations and media compatibility.
Before extraction, Narratio resolves every binding from the current prepared
session manifest and passes its canonical absolute `inputs/` path to Notarius.
Missing, unsafe, empty, or checksum-inconsistent prepared evidence fails with
session manifest and streams it into a verified invocation-local snapshot whose
absolute path is passed to Notarius. Missing, unsafe, empty,
changed-during-copy, or checksum-inconsistent prepared evidence fails with
guidance to force `prepare`. Bindings are sorted by normalized selector and are
part of extraction fingerprint and resume identity. See the
[Notarius integration contract](./integrations/notarius.md) for the subprocess

22
docs/integrations/notarius.md
View File

@@ -30,14 +30,16 @@ references are configured and invokes each binding as a separate argument
before `--json`:
```text
notarius run <pipeline_id> --config <config_path> --input <trimmed_json> --output-dir <staging_dir> [--reference <selector>=<prepared_path>]... --json
notarius run <pipeline_id> --config <config_path> --input <trimmed_json> --output-dir <staging_dir> [--reference <selector>=<verified_snapshot_path>]... --json
```
Reference paths are absolute canonical files prepared inside the current
Narratio session workspace. Narratio passes only configured bindings, ordered
lexically by normalized selector, as direct argument-vector entries without
shell interpretation. A CLI binding takes precedence over a matching external
path in Notarius configuration. Narratio never emits `--without-reference`.
Reference paths are absolute invocation-local snapshots streamed from the
manifest-verified canonical files prepared inside the current Narratio session
workspace. Narratio verifies snapshot checksum and size before and after the
subprocess, and passes only configured bindings, ordered lexically by normalized
selector, as direct argument-vector entries without shell interpretation. A CLI
binding takes precedence over a matching external path in Notarius
configuration. Narratio never emits `--without-reference`.
The maintained D&D boundary binds only the four campaign-owned external slots:
@@ -46,10 +48,10 @@ notarius run dnd-session \
--config <absolute config path> \
--input <absolute trimmed transcript path> \
--output-dir <absolute staging directory> \
--reference glossary=<absolute prepared glossary path> \
--reference party=<absolute prepared party path> \
--reference players=<absolute prepared players path> \
--reference spell_catalog=<absolute prepared spell catalog path> \
--reference glossary=<absolute verified glossary snapshot> \
--reference party=<absolute verified party snapshot> \
--reference players=<absolute verified players snapshot> \
--reference spell_catalog=<absolute verified spell catalog snapshot> \
--json
```

23
docs/internal/stage-extract.md
View File

@@ -19,17 +19,19 @@ procedures belong in [Operations](../operations.md).
1. resolves the final trimmed transcript from the shared artifact catalog;
2. resolves every configured prepared reference through the shared
manifest-authoritative identity resolver before creating run-local output;
3. fingerprints the Notarius invocation contract, including sorted reference
3. streams each verified reference into an invocation-local snapshot and
rejects any source change observed while copying;
4. fingerprints the Notarius invocation contract, including sorted reference
identities;
4. creates a run-local staging directory and invokes the injected
5. creates a run-local staging directory and invokes the injected
`notarius.Runner`;
5. validates the v2 successful receipt, confined index, management documents,
configured required lane descriptors, validation summaries, and regular
payload files;
6. atomically promotes the complete bundle to its immutable durable location;
7. records one non-selectable `notarius_index` output and one selectable
6. revalidates the reference snapshots, then validates the v2 successful
receipt, confined index, management documents, configured required lane
descriptors, validation summaries, and regular payload files;
7. atomically promotes the complete bundle to its immutable durable location;
8. records one non-selectable `notarius_index` output and one selectable
`notarius_lane` output per configured lane; and
8. registers each lane as `narratio.extraction.<output_key>` for downstream
9. registers each lane as `narratio.extraction.<output_key>` for downstream
Scriptorium and publish resolution.
Lane records retain checksum, contract, producer run ID, and Notarius system,
@@ -40,7 +42,10 @@ fingerprint. The input identity binds the exact transcript bytes, canonical
source ID, producer stage/output/run identity, and resolution provenance.
Reference metadata contains only selector, source ID, canonical session-relative
path, checksum, and size; adapter requests receive selector and absolute
prepared path, never payload contents.
invocation-local snapshot path, never payload contents. Snapshot bytes must
match the prepared identity both before and after Notarius runs, so a concurrent
prepared-file replacement cannot make recorded provenance describe different
bytes from those supplied to Notarius.
Validation completes before
promotion, so a rejected result cannot expose a partial durable bundle.

5
docs/operations.md
View File

@@ -143,7 +143,10 @@ are not selectable or published implicitly.
Configured Notarius references resolve only from the current manifest-backed
prepared inputs. Their canonical locations are `inputs/party.yml`,
`inputs/players.yml`, `inputs/glossary.yml`, and, when configured,
`inputs/spell_catalog.json`. Inspect the effective stable-input inventory and
`inputs/spell_catalog.json`. Extraction supplies Notarius with verified copies
under `runs/<run_id>/extract/references/` so a concurrent refresh of canonical
prepared files cannot change the bytes consumed by an in-flight invocation.
Inspect the effective stable-input inventory and
prepared-file readiness with:
```bash

14
internal/app/extract_lifecycle_test.go
View File

@@ -148,7 +148,7 @@ func TestExtractLifecycleDisabledThenEnabled(t *testing.T) {
}
}
func TestExtractLifecyclePrepareBindsCanonicalPreparedReferences(t *testing.T) {
func TestExtractLifecyclePrepareBindsVerifiedReferenceSnapshots(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
originalPaths := configureLifecycleReferences(t, cfg)
@@ -177,15 +177,21 @@ func TestExtractLifecyclePrepareBindsCanonicalPreparedReferences(t *testing.T) {
if len(request.References) != len(want) {
t.Fatalf("references = %#v", request.References)
}
loaded := loadLifecycleManifest(t, cfg)
for index, expected := range want {
binding := request.References[index]
canonical := filepath.Join(paths.InputsDir, expected.filename)
if binding.Selector != expected.selector || binding.Path != canonical || binding.Path == originalPaths[expected.sourceID] {
t.Fatalf("reference[%d] = %#v, want selector %q canonical %q and not source %q", index, binding, expected.selector, canonical, originalPaths[expected.sourceID])
snapshot := filepath.Join(
artifacts.SessionRunNotariusReferencesDirForCampaign(
cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, loaded.RunID,
),
expected.filename,
)
if binding.Selector != expected.selector || binding.Path != snapshot || binding.Path == canonical || binding.Path == originalPaths[expected.sourceID] {
t.Fatalf("reference[%d] = %#v, want selector %q snapshot %q and not prepared/source paths", index, binding, expected.selector, snapshot)
}
}
loaded := loadLifecycleManifest(t, cfg)
extract := loaded.Stages["extract"]
if extract == nil || extract.Status != manifest.StatusSucceeded || extract.Metadata["reference_count"] != float64(len(want)) {
t.Fatalf("extract record = %#v", extract)

6
internal/artifacts/paths.go
View File

@@ -113,6 +113,12 @@ func SessionRunNotariusLogPathForCampaign(rootDir, campaign, sessionID, runID st
return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius.stderr.log")
}
// SessionRunNotariusReferencesDirForCampaign returns the invocation-local
// directory containing verified reference snapshots supplied to Notarius.
func SessionRunNotariusReferencesDirForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "references")
}
// SessionRunNotariusOutputRootForCampaign returns the invocation-local Notarius output root.
func SessionRunNotariusOutputRootForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius-output")

1
internal/artifacts/paths_model_test.go
View File

@@ -64,6 +64,7 @@ func TestSessionNotariusPathsForCampaign(t *testing.T) {
{name: "extract directory", got: SessionRunExtractDirForCampaign(root, campaign, sessionID, runID), want: extractDir},
{name: "receipt", got: SessionRunNotariusReceiptPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.receipt.json")},
{name: "stderr", got: SessionRunNotariusLogPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.stderr.log")},
{name: "references", got: SessionRunNotariusReferencesDirForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "references")},
{name: "output root", got: SessionRunNotariusOutputRootForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius-output")},
{name: "durable bundle", got: SessionNotariusBundleDirForCampaign(root, campaign, sessionID, runID), want: filepath.Join(root, "work", campaign, sessionID, "artifacts", "notarius", runID)},
}

20
internal/fileops/fileops.go
View File

@@ -67,17 +67,29 @@ func CopyFileAtomicWithChecksum(src, dst string, perm os.FileMode) (string, erro
}
defer func() { _ = in.Close() }()
return WriteReaderAtomicWithChecksum(dst, in, perm)
}
// WriteReaderAtomicWithChecksum streams src through the durable replacement
// sequence and returns the SHA-256 checksum of the installed bytes. The caller
// retains ownership of src.
func WriteReaderAtomicWithChecksum(dst string, src io.Reader, perm os.FileMode) (string, error) {
if strings.TrimSpace(dst) == "" {
return "", fmt.Errorf("destination path is required")
}
if src == nil {
return "", fmt.Errorf("source reader is required")
}
if err := EnsureWorkspaceDirectory(filepath.Dir(dst)); err != nil {
return "", fmt.Errorf("create destination directory: %w", err)
}
digest := sha256.New()
err = replaceFileFromReaderConfined(
if err := replaceFileFromReaderConfined(
dst,
io.TeeReader(in, digest),
io.TeeReader(src, digest),
ReplaceFileOptions{Mode: perm},
)
if err != nil {
); err != nil {
return "", err
}
return hex.EncodeToString(digest.Sum(nil)), nil

24
internal/fileops/fileops_test.go
View File

@@ -1,6 +1,8 @@
package fileops
import (
"crypto/sha256"
"encoding/hex"
"errors"
"io"
"os"
@@ -74,6 +76,28 @@ func TestCopyFileAtomicWithChecksumMatchesDestination(t *testing.T) {
assertNoMatchingTempFiles(t, filepath.Dir(dst), ".copied.txt.tmp-")
}
func TestWriteReaderAtomicWithChecksumMatchesDestination(t *testing.T) {
root := t.TempDir()
dst := filepath.Join(root, "nested", "snapshot.yml")
payload := "verified reference bytes\n"
checksum, err := WriteReaderAtomicWithChecksum(dst, strings.NewReader(payload), WorkspaceFileMode)
if err != nil {
t.Fatalf("WriteReaderAtomicWithChecksum() error = %v", err)
}
wantChecksum := sha256.Sum256([]byte(payload))
if checksum != hex.EncodeToString(wantChecksum[:]) {
t.Fatalf("checksum = %q, want %q", checksum, hex.EncodeToString(wantChecksum[:]))
}
data, err := os.ReadFile(dst)
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
if string(data) != payload {
t.Fatalf("destination = %q, want %q", data, payload)
}
}
func TestCopyFileAtomicCleansTempFileOnInstallFailure(t *testing.T) {
root := t.TempDir()
src := filepath.Join(root, "source.txt")

19
internal/stage/extract.go
View File

@@ -123,11 +123,23 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("extract: resolve durable bundle path: %w", err)
}
for _, directory := range []string{filepath.Dir(receiptPath), outputRoot, filepath.Dir(durableBundle)} {
referenceSnapshotRoot, err := absolutePath(artifacts.SessionRunNotariusReferencesDirForCampaign(workspaceRoot, campaign, sessionID, runID))
if err != nil {
return nil, fmt.Errorf("extract: resolve reference snapshot path: %w", err)
}
directories := []string{filepath.Dir(receiptPath), outputRoot, filepath.Dir(durableBundle)}
if len(references.Bindings) > 0 {
directories = append(directories, referenceSnapshotRoot)
}
for _, directory := range directories {
if err := fileops.EnsureWorkspaceDirectory(directory); err != nil {
return nil, fmt.Errorf("extract: create directory %q: %w", directory, err)
}
}
referenceBindings, err := materializeExtractReferenceSnapshots(paths, references, referenceSnapshotRoot)
if err != nil {
return nil, fmt.Errorf("extract: materialize Notarius reference snapshots: %w", err)
}
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory, input, references.Identities)
if err != nil {
@@ -136,7 +148,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
request := notarius.RunRequest{
Binary: resolvedBinary, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, References: references.Bindings,
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, References: referenceBindings,
}
adapterResult, err := env.Notarius.Run(ctx, request)
if err != nil {
@@ -148,6 +160,9 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if adapterResult.Receipt.RunID == "" || adapterResult.Receipt.PipelineID != notariusConfig.PipelineID {
return nil, fmt.Errorf("extract: notarius receipt identity is missing or incompatible")
}
if err := verifyExtractReferenceSnapshots(referenceSnapshotRoot, references); err != nil {
return nil, fmt.Errorf("extract: verify Notarius reference snapshots: %w", err)
}
selected, err := selectRequiredNotariusLanes(env.ArtifactStore, notariusConfig.Outputs, adapterResult)
if err != nil {

86
internal/stage/extract_references.go
View File

@@ -1,13 +1,19 @@
package stage
import (
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"os"
"path/filepath"
"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/fileops"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/notariusref"
)
@@ -20,6 +26,86 @@ type extractReferenceIdentity struct {
SizeBytes int64 `json:"size_bytes"`
}
func materializeExtractReferenceSnapshots(
paths artifacts.SessionPaths,
references resolvedExtractReferences,
snapshotRoot string,
) ([]notarius.ReferenceBinding, error) {
if len(references.Bindings) != len(references.Identities) {
return nil, fmt.Errorf("reference bindings and identities have different lengths")
}
bindings := make([]notarius.ReferenceBinding, len(references.Bindings))
copied := make(map[string]string, len(references.Bindings))
destinationOwners := make(map[string]string, len(references.Bindings))
for index, binding := range references.Bindings {
identity := references.Identities[index]
destination, ok := copied[identity.SourceID]
if !ok {
filename := filepath.Base(filepath.FromSlash(identity.Path))
if filename == "" || filename == "." || filename == string(filepath.Separator) {
return nil, fmt.Errorf("reference %q source %q has invalid prepared filename", identity.Selector, identity.SourceID)
}
destination = filepath.Join(snapshotRoot, filename)
if owner, duplicate := destinationOwners[destination]; duplicate && owner != identity.SourceID {
return nil, fmt.Errorf("reference sources %q and %q resolve to the same snapshot path", owner, identity.SourceID)
}
destinationOwners[destination] = identity.SourceID
file, err := fileops.OpenConfinedRegularFile(paths.Root, identity.Path)
if err != nil {
return nil, fmt.Errorf("open reference %q source %q for snapshot: %w", identity.Selector, identity.SourceID, err)
}
checksum, copyErr := fileops.WriteReaderAtomicWithChecksum(destination, file, fileops.WorkspaceFileMode)
closeErr := file.Close()
if copyErr != nil {
return nil, fmt.Errorf("snapshot reference %q source %q: %w", identity.Selector, identity.SourceID, copyErr)
}
if closeErr != nil {
return nil, fmt.Errorf("close reference %q source %q: %w", identity.Selector, identity.SourceID, closeErr)
}
info, err := os.Lstat(destination)
if err != nil {
return nil, fmt.Errorf("inspect reference %q snapshot: %w", identity.Selector, err)
}
if !info.Mode().IsRegular() || !strings.EqualFold(checksum, identity.Checksum) || info.Size() != identity.SizeBytes {
return nil, fmt.Errorf("reference %q source %q changed while its verified snapshot was created", identity.Selector, identity.SourceID)
}
copied[identity.SourceID] = destination
}
bindings[index] = notarius.ReferenceBinding{Selector: binding.Selector, Path: destination}
}
return bindings, nil
}
func verifyExtractReferenceSnapshots(snapshotRoot string, references resolvedExtractReferences) error {
verified := make(map[string]struct{}, len(references.Identities))
for _, identity := range references.Identities {
if _, ok := verified[identity.SourceID]; ok {
continue
}
filename := filepath.Base(filepath.FromSlash(identity.Path))
file, err := fileops.OpenConfinedRegularFile(snapshotRoot, filename)
if err != nil {
return fmt.Errorf("reopen reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, err)
}
digest := sha256.New()
size, readErr := io.Copy(digest, file)
closeErr := file.Close()
if readErr != nil {
return fmt.Errorf("revalidate reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, readErr)
}
if closeErr != nil {
return fmt.Errorf("close reference %q source %q snapshot: %w", identity.Selector, identity.SourceID, closeErr)
}
checksum := hex.EncodeToString(digest.Sum(nil))
if !strings.EqualFold(checksum, identity.Checksum) || size != identity.SizeBytes {
return fmt.Errorf("reference %q source %q snapshot changed while Notarius was running", identity.Selector, identity.SourceID)
}
verified[identity.SourceID] = struct{}{}
}
return nil
}
type resolvedExtractReferences struct {
Bindings []notarius.ReferenceBinding
Identities []extractReferenceIdentity

76
internal/stage/extract_test.go
View File

@@ -100,12 +100,12 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
{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))
preparedPaths := 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)
preparedPaths[selector] = recordPreparedExtractInput(t, env, m, reference.sourceID, reference.contents)
wantSources[selector] = reference.sourceID
}
@@ -119,8 +119,14 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
}
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])
wantPath := filepath.Join(
artifacts.SessionRunNotariusReferencesDirForCampaign(
env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID,
),
filepath.Base(preparedPaths[selector]),
)
if binding.Selector != selector || binding.Path != wantPath || binding.Path == preparedPaths[selector] || !filepath.IsAbs(binding.Path) {
t.Fatalf("reference[%d] = %#v, want selector %q verified snapshot %q", index, binding, selector, wantPath)
}
}
@@ -154,6 +160,68 @@ func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T)
}
}
type referenceMutationRunner struct {
delegate notarius.Runner
preparedPath string
preparedBytes []byte
mutateSnapshot bool
snapshotBytes []byte
}
func (r *referenceMutationRunner) Run(ctx context.Context, req notarius.RunRequest) (notarius.RunResult, error) {
if err := os.WriteFile(r.preparedPath, r.preparedBytes, 0o664); err != nil {
return notarius.RunResult{}, err
}
if len(req.References) != 1 {
return notarius.RunResult{}, fmt.Errorf("got %d references, want one", len(req.References))
}
data, err := os.ReadFile(req.References[0].Path)
if err != nil {
return notarius.RunResult{}, err
}
r.snapshotBytes = data
if r.mutateSnapshot {
if err := os.WriteFile(req.References[0].Path, []byte("tampered snapshot\n"), 0o664); err != nil {
return notarius.RunResult{}, err
}
}
return r.delegate.Run(ctx, req)
}
func TestExtractStageInvokesNotariusWithVerifiedReferenceSnapshot(t *testing.T) {
env, m, fake := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "verified party\n")
runner := &referenceMutationRunner{
delegate: fake, preparedPath: preparedPath, preparedBytes: []byte("changed after snapshot\n"),
}
env.Notarius = runner
if _, err := (extractStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("Run() error = %v", err)
}
if string(runner.snapshotBytes) != "verified party\n" {
t.Fatalf("Notarius reference bytes = %q, want verified snapshot", runner.snapshotBytes)
}
if len(fake.Requests) != 1 || fake.Requests[0].References[0].Path == preparedPath {
t.Fatalf("adapter requests = %#v, want a run-local reference snapshot", fake.Requests)
}
}
func TestExtractStageRejectsReferenceSnapshotChangedDuringNotariusRun(t *testing.T) {
env, m, fake := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputParty}
preparedPath := recordPreparedExtractInput(t, env, m, artifactpolicy.SourceInputParty, "verified party\n")
env.Notarius = &referenceMutationRunner{
delegate: fake, preparedPath: preparedPath, preparedBytes: []byte("changed after snapshot\n"), mutateSnapshot: true,
}
result, err := (extractStage{}).Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "snapshot changed while Notarius was running") || result != nil {
t.Fatalf("Run() result = %#v error = %v, want changed-snapshot failure", result, err)
}
}
func TestExtractStageRejectsUnavailableReferenceBeforeInvocationOrRunDirectoryCreation(t *testing.T) {
env, m, fake := setupExtractEnv(t)
env.Config.Pipeline.Notarius.References = map[string]string{