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

Permit safe cross-run output declarations

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-05 12:37:58 +00:00
parent be59bff690
commit 145baac393
16 changed files with 363 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
internal/app/operator_artifact_rendering.go
View File

@@ -127,9 +127,8 @@ func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, cata
fmt.Fprintf(out, "- %s\n", strings.Join(parts, " "))
}
func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string {
func unresolvedRemotePublishedOutputAvailability(cfg *config.Config, catalog *artifacts.ArtifactCatalog, state string) map[string]string {
out := map[string]string{}
sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID)
for _, rule := range cfg.Pipeline.Publish.Outputs {
source := strings.TrimSpace(rule.Source)
dest, _, err := helperPublishedOutputDest(rule, catalog)
@@ -137,14 +136,7 @@ func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config,
out[publishedOutputRemoteStateKey(source, "")] = "remote=error"
continue
}
key := artifacts.S3PublishedOutputKey(sessionPrefix, dest)
if exists, err := store.Exists(ctx, key); err == nil && exists {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=published"
} else if err != nil {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=error"
} else {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing"
}
out[publishedOutputRemoteStateKey(source, dest)] = state
}
return out
}
@@ -157,10 +149,9 @@ func remotePublishedOutputAvailabilityForCurrent(
current *RemoteCurrentState,
) map[string]string {
if current == nil || current.Commit == nil {
return remotePublishedOutputAvailability(ctx, cfg, store, catalog)
return unresolvedRemotePublishedOutputAvailability(cfg, catalog, "remote=missing")
}
out := map[string]string{}
runPrefix := artifacts.S3RunPrefix(current.SessionPrefix, current.RunID)
for _, rule := range cfg.Pipeline.Publish.Outputs {
source := strings.TrimSpace(rule.Source)
dest, _, err := helperPublishedOutputDest(rule, catalog)
@@ -168,26 +159,44 @@ func remotePublishedOutputAvailabilityForCurrent(
out[publishedOutputRemoteStateKey(source, "")] = "remote=error"
continue
}
key := artifacts.S3RunRelativeDestinationKey(runPrefix, dest)
if committedPublishedOutput(current.Commit, key) {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=published"
} else {
declaration, ok := committedPublishedOutput(current.Commit, source)
if !ok {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing"
continue
}
identity, err := artifacts.ParseImmutableRunObjectKey(current.SessionPrefix, declaration.DestinationKey)
if err != nil {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=error"
continue
}
exists, err := store.Exists(ctx, declaration.DestinationKey)
if err != nil {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=error"
continue
}
if !exists {
out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing"
continue
}
state := "remote=published"
if identity.RunID != current.RunID {
state += " origin_run=" + identity.RunID
}
out[publishedOutputRemoteStateKey(source, dest)] = state
}
return out
}
func committedPublishedOutput(commit *artifacts.RemoteCommitManifest, key string) bool {
func committedPublishedOutput(commit *artifacts.RemoteCommitManifest, source string) (artifacts.RemoteArtifact, bool) {
if commit == nil {
return false
return artifacts.RemoteArtifact{}, false
}
for _, artifact := range commit.Artifacts {
if artifact.Type == artifacts.RemoteArtifactTypePublishedOutput && artifact.DestinationKey == key {
return true
if artifact.Type == artifacts.RemoteArtifactTypePublishedOutput && artifact.Source == source {
return artifact, true
}
}
return false
return artifacts.RemoteArtifact{}, false
}
func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog) (string, bool, error) {

7
internal/app/operator_artifacts_list.go
View File

@@ -33,7 +33,12 @@ func ArtifactsList(ctx context.Context, args []string, out io.Writer) error {
}
publishedRemoteState := map[string]string{}
if remote && store != nil {
publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog)
current := inspectRemoteCurrentState(ctx, cfg, store)
if current.Err != nil {
publishedRemoteState = unresolvedRemotePublishedOutputAvailability(cfg, catalog, "remote=error")
} else {
publishedRemoteState = remotePublishedOutputAvailabilityForCurrent(ctx, cfg, store, catalog, current.State)
}
}
writeArtifactList(out, cfg, catalog, locks, publishedRemoteState)
return nil

85
internal/app/operator_helpers_test.go
View File

@@ -7,6 +7,7 @@ import (
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"testing"
@@ -914,6 +915,55 @@ func addSessionTemplateToCampaign(t *testing.T, campaignPath, templateFile strin
}
}
func seedOperatorCommittedOutputs(t *testing.T, fake *storage.FakeBackend, outputs map[string]string) {
t.Helper()
const runID = "20260519T010203Z-a1b2c3d4"
sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")
manifestKey := artifacts.S3RunSessionManifestKey(sessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: manifestKey, Data: restoreManifestJSON(t, "2026-05-03", "sample-campaign")})
declarations := []artifacts.RemoteArtifact{
remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypeSessionManifest, "session.manifest", manifestKey),
}
sources := make([]string, 0, len(outputs))
for source := range outputs {
sources = append(sources, source)
}
sort.Strings(sources)
for _, source := range sources {
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), outputs[source])
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte(`{"available":true}`)})
declarations = append(declarations, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, source, key))
}
commit := artifacts.RemoteCommitManifest{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: "sample-campaign",
SessionID: "2026-05-03",
RunID: runID,
Artifacts: declarations,
}
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
if err != nil {
t.Fatalf("encode operator remote commit: %v", err)
}
commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: commitKey, Data: commitData})
commitObject := fake.Objects[commitKey]
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: "sample-campaign",
SessionID: "2026-05-03",
RunID: runID,
CommitKey: commitKey,
CommitSHA256: restoreCommitSHA256(commitData),
CommitSize: int64(len(commitData)),
CommitGeneration: commitObject.ETag,
})
if err != nil {
t.Fatalf("encode operator current pointer: %v", err)
}
fake.SeedObject(storage.FakeObject{Key: artifacts.S3CurrentCommitPointerKey(sessionPrefix), Data: pointerData})
}
func TestExecuteArtifactsListRemoteReportsPublishedAvailability(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
@@ -924,11 +974,9 @@ func TestExecuteArtifactsListRemoteReportsPublishedAvailability(t *testing.T) {
required: true
`)
fake := &storage.FakeBackend{}
trimmedKey := artifacts.S3PublishedOutputKey(
artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"),
"transcripts/final.trimmed.json",
)
fake.SeedObject(storage.FakeObject{Key: trimmedKey, Data: []byte(`{"segments":[]}`)})
seedOperatorCommittedOutputs(t, fake, map[string]string{
"narratio.transcript.final_trimmed": "transcripts/final.trimmed.json",
})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
@@ -961,11 +1009,9 @@ func TestExecuteArtifactsListReportsExtractionLifecycleWithoutPayload(t *testing
`)
lanePath := writeOperatorExtractionManifest(t, workspaceRoot)
fake := &storage.FakeBackend{}
publishedKey := artifacts.S3PublishedOutputKey(
artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"),
"artifacts/encounters.json",
)
fake.SeedObject(storage.FakeObject{Key: publishedKey, Data: []byte(`{"secret":"DO_NOT_PRINT"}`)})
seedOperatorCommittedOutputs(t, fake, map[string]string{
"narratio.extraction.encounters": "artifacts/encounters.json",
})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
@@ -1027,9 +1073,10 @@ func TestExecuteArtifactsListRemoteUsesPublishOutputDestinations(t *testing.T) {
required: true
`)
fake := &storage.FakeBackend{}
sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")
fake.SeedObject(storage.FakeObject{Key: artifacts.S3PublishedOutputKey(sessionPrefix, "transcripts/full.json"), Data: []byte(`{"segments":[]}`)})
fake.SeedObject(storage.FakeObject{Key: artifacts.S3PublishedOutputKey(sessionPrefix, "transcripts/bounds.json"), Data: []byte(`{}`)})
seedOperatorCommittedOutputs(t, fake, map[string]string{
"narratio.transcript.final": "transcripts/full.json",
"narratio.bounds.session": "transcripts/bounds.json",
})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
@@ -1078,15 +1125,11 @@ func TestExecuteStatusReportsRemoteArtifactCatalog(t *testing.T) {
`)
fake := &storage.FakeBackend{}
sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")
manifestKey := artifacts.S3RunSessionManifestKey(sessionPrefix, "20260519T010203Z-a1b2c3d4")
runIDKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)
trimmedKey := artifacts.S3PublishedOutputKey(sessionPrefix, "transcripts/final.trimmed.json")
fullKey := artifacts.S3PublishedOutputKey(sessionPrefix, "transcripts/full.json")
lockKey := artifacts.S3SessionLocksKey(sessionPrefix)
fake.SeedObject(storage.FakeObject{Key: runIDKey, Data: []byte("20260519T010203Z-a1b2c3d4\n")})
fake.SeedObject(storage.FakeObject{Key: manifestKey, Data: restoreManifestJSON(t, "2026-05-03", "sample-campaign")})
fake.SeedObject(storage.FakeObject{Key: trimmedKey, Data: []byte(`{"segments":[]}`)})
fake.SeedObject(storage.FakeObject{Key: fullKey, Data: []byte(`{"segments":[]}`)})
seedOperatorCommittedOutputs(t, fake, map[string]string{
"narratio.transcript.final_trimmed": "transcripts/final.trimmed.json",
"narratio.transcript.final": "transcripts/full.json",
})
fake.SeedObject(storage.FakeObject{Key: lockKey, Data: []byte("locks:\n - source: narratio.transcript.final_trimmed\n reason: remote review\n")})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})

8
internal/app/operator_status.go
View File

@@ -58,11 +58,13 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
store, storeErr := objectStoreIfConfigured(ctx, cfg)
var remoteCurrent *RemoteCurrentState
var remoteCurrentErr error
if storeErr != nil {
fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr)
} else if store != nil {
current := inspectRemoteCurrentState(ctx, cfg, store)
if current.Err != nil {
remoteCurrentErr = current.Err
fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", current.Err)
} else {
remoteCurrent = current.State
@@ -97,7 +99,11 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
}
publishedRemoteState := map[string]string{}
if store != nil {
publishedRemoteState = remotePublishedOutputAvailabilityForCurrent(ctx, cfg, store, catalog, remoteCurrent)
if remoteCurrentErr != nil {
publishedRemoteState = unresolvedRemotePublishedOutputAvailability(cfg, catalog, "remote=error")
} else {
publishedRemoteState = remotePublishedOutputAvailabilityForCurrent(ctx, cfg, store, catalog, remoteCurrent)
}
}
fmt.Fprintln(out, "Remote outputs:")
writeArtifactList(out, cfg, catalog, catalogLocks, publishedRemoteState)

60
internal/app/restore_commit_test.go
View File

@@ -43,6 +43,38 @@ func TestCommittedRestorePlanUsesOnlyDeclaredObjects(t *testing.T) {
}
}
func TestCommittedRestoreDownloadsCarriedOutputFromDeclaredRun(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("carried\n"),
})
declaration := &current.Commit.Artifacts[1]
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, "20260512T010203Z-11223344"), "transcripts/full.json")
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
oldObject := fake.Objects[oldKey]
declaration.DestinationKey = oldKey
declaration.Generation = oldObject.ETag
if err := current.Commit.ValidateForSessionPrefix(current.SessionPrefix); err != nil {
t.Fatalf("carried commit validation: %v", err)
}
plan, err := buildRestorePlan(context.Background(), cfg, current, fake, RestorePlanOptions{})
if err != nil {
t.Fatalf("buildRestorePlan() error = %v", err)
}
action := restoreActionForPath(t, plan, "transcripts/full.json")
if action.RemoteKey != oldKey {
t.Fatalf("restore remote key = %q, want exact carried key %q", action.RemoteKey, oldKey)
}
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err != nil {
t.Fatalf("executeRestorePlan() error = %v", err)
}
if got := fakeDownloadCount(fake, oldKey); got != 1 {
t.Fatalf("carried object downloads = %d, want one", got)
}
}
func TestCommittedRestoreReusesVerifiedManifestCandidate(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}
@@ -92,6 +124,7 @@ func TestCommittedStatusReportsOnlyDeclaredPublishedOutputs(t *testing.T) {
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("declared\n"),
})
current.Commit.Artifacts[1].Source = "narratio.transcript.final_trimmed"
fake.SeedObject(storage.FakeObject{Key: current.SessionPrefix + "transcripts/full.json", Data: []byte("mutable stale copy\n")})
availability := remotePublishedOutputAvailabilityForCurrent(context.Background(), cfg, fake, nil, current)
@@ -101,6 +134,33 @@ func TestCommittedStatusReportsOnlyDeclaredPublishedOutputs(t *testing.T) {
}
}
func TestCommittedStatusReportsCarriedOutputOriginAndChecksExactKey(t *testing.T) {
cfg := restorePlanConfig(t)
cfg.Pipeline.Publish = &config.PublishConfig{Outputs: []config.PublishOutputRule{{
Source: "narratio.transcript.final_trimmed",
Dest: "transcripts/full.json",
}}}
fake := &storage.FakeBackend{}
current := seedCommittedRestoreSnapshot(t, cfg, fake, "20260519T010203Z-a1b2c3d4", map[string][]byte{
"transcripts/full.json": []byte("carried\n"),
})
const originRun = "20260512T010203Z-11223344"
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, originRun), "transcripts/full.json")
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
current.Commit.Artifacts[1].Source = "narratio.transcript.final_trimmed"
current.Commit.Artifacts[1].DestinationKey = oldKey
availability := remotePublishedOutputAvailabilityForCurrent(context.Background(), cfg, fake, nil, current)
key := publishedOutputRemoteStateKey("narratio.transcript.final_trimmed", "transcripts/full.json")
want := "remote=published origin_run=" + originRun
if availability[key] != want {
t.Fatalf("availability = %#v, want %q", availability, want)
}
if len(fake.ExistsKeys) == 0 || fake.ExistsKeys[len(fake.ExistsKeys)-1] != oldKey {
t.Fatalf("exists keys = %#v, want final exact declaration key %q", fake.ExistsKeys, oldKey)
}
}
func TestCommittedRestoreKeepsSelectedSnapshotWhenPointerChanges(t *testing.T) {
cfg := restorePlanConfig(t)
fake := &storage.FakeBackend{}

16
internal/app/restore_plan.go
View File

@@ -148,13 +148,9 @@ func buildCommittedRestoreActions(
if current == nil || current.Commit == nil {
return nil, fmt.Errorf("committed remote current state is required")
}
runPrefix := artifacts.S3RunPrefix(current.SessionPrefix, current.RunID)
if runPrefix == "" {
return nil, fmt.Errorf("committed run prefix is required")
}
actions := make([]RestoreAction, 0, len(current.Commit.Artifacts))
for _, artifact := range current.Commit.Artifacts {
rel, include, err := restoreLocalRelativePathForCommittedArtifact(runPrefix, artifact, opts.IncludeAudio)
rel, include, err := restoreLocalRelativePathForCommittedArtifact(current.SessionPrefix, artifact, opts.IncludeAudio)
if err != nil {
return nil, fmt.Errorf("map committed object %q: %w", artifact.DestinationKey, err)
}
@@ -179,18 +175,18 @@ func buildCommittedRestoreActions(
return actions, nil
}
func restoreLocalRelativePathForCommittedArtifact(runPrefix string, artifact artifacts.RemoteArtifact, includeAudio bool) (string, bool, error) {
func restoreLocalRelativePathForCommittedArtifact(sessionPrefix string, artifact artifacts.RemoteArtifact, includeAudio bool) (string, bool, error) {
if artifact.Type == artifacts.RemoteArtifactTypeSessionManifest {
return config.PathManifestFile, true, nil
}
if artifact.Type != artifacts.RemoteArtifactTypePublishedOutput {
return "", false, nil
}
key := normalizeRemoteKey(artifact.DestinationKey)
if !strings.HasPrefix(key, runPrefix) {
return "", false, fmt.Errorf("object is outside committed run prefix %q", runPrefix)
identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, artifact.DestinationKey)
if err != nil {
return "", false, err
}
rel := strings.TrimPrefix(key, runPrefix)
rel := strings.TrimPrefix(identity.RelativePath, "published/")
cleanRel, err := pathsafe.NormalizeRelativeDestination(rel)
if err != nil {
return "", false, fmt.Errorf("committed destination is unsafe: %w", err)

2
internal/app/restore_plan_test.go
View File

@@ -345,7 +345,7 @@ func TestRestorePlanTraversalUnsafeKeyFails(t *testing.T) {
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "committed destination is unsafe") {
if !strings.Contains(err.Error(), "clean relative object key") {
t.Fatalf("error = %v, want traversal safety failure", err)
}
}