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

Publish immutable remote commits

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 20:05:24 +00:00
parent d6deccf3e8
commit 361dbb4ca8
15 changed files with 725 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
internal/adapters/storage/fake.go
View File

@@ -2,6 +2,8 @@ package storage
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"os"
@@ -21,6 +23,7 @@ type FakeBackend struct {
DownloadErr error
UploadErr error
ExistsErr error
UploadHook func(FakeUploadCall) error
}
// FakeUploadCall captures one upload invocation in call order.
@@ -54,6 +57,9 @@ func (f *FakeBackend) SeedObject(obj FakeObject) {
obj.Key = key
obj.Data = append([]byte(nil), obj.Data...)
obj.Metadata = copyMetadata(obj.Metadata)
if obj.ETag == "" {
obj.ETag = fakeObjectETag(obj.Data)
}
f.Objects[key] = obj
}
@@ -175,14 +181,20 @@ func (f *FakeBackend) uploadReader(ctx context.Context, source io.Reader, key st
}
normalizedKey := normalizeObjectKey(key)
f.Uploads = append(f.Uploads, FakeUploadCall{
call := FakeUploadCall{
LocalPath: localPath,
Key: normalizedKey,
Options: UploadOptions{
Metadata: copyMetadata(opts.Metadata),
ContentType: opts.ContentType,
},
})
}
f.Uploads = append(f.Uploads, call)
if f.UploadHook != nil {
if err := f.UploadHook(call); err != nil {
return ObjectInfo{}, err
}
}
now := time.Now().UTC()
obj := FakeObject{
Key: normalizedKey,
@@ -194,10 +206,16 @@ func (f *FakeBackend) uploadReader(ctx context.Context, source io.Reader, key st
return ObjectInfo{
Key: normalizedKey,
Size: int64(len(data)),
ETag: fakeObjectETag(data),
LastModified: &now,
}, nil
}
func fakeObjectETag(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
// Exists checks object presence.
func (f *FakeBackend) Exists(ctx context.Context, key string) (bool, error) {
if err := ctx.Err(); err != nil {

8
internal/app/post_publish_cleanup.go
View File

@@ -144,11 +144,11 @@ func publishCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool,
if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded {
return false, "publish did not upload run record"
}
if pointer, _ := sr.Metadata["current_pointer_written"].(bool); !pointer {
return false, "publish did not write current pointer"
if strings.TrimSpace(asString(sr.Metadata["remote_commit_key"])) == "" {
return false, "publish remote commit key is missing"
}
if strings.TrimSpace(asString(sr.Metadata["current_run_id_key"])) == "" {
return false, "publish current run pointer key is missing"
if strings.TrimSpace(asString(sr.Metadata["current_commit_pointer_key"])) == "" {
return false, "publish current commit pointer key is missing"
}
return true, ""
}

30
internal/app/post_publish_cleanup_test.go
View File

@@ -25,10 +25,10 @@ func (publishSuccessStage) Name() string { return "publish" }
func (publishSuccessStage) Declares() stage.IODecl { return stage.IODecl{} }
func (s publishSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
md := map[string]any{
"stage": "publish",
"uploaded": true,
"current_pointer_written": true,
"current_run_id_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/run_id.txt",
"stage": "publish",
"uploaded": true,
"remote_commit_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/20260519T010203Z-a1b2c3d4/commit.json",
"current_commit_pointer_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/commit-pointer.json",
}
for k, v := range s.metadata {
md[k] = v
@@ -130,12 +130,12 @@ func TestPostPublishCleanupNotRunWhenPublishSkipped(t *testing.T) {
assertExists(t, seed.runWorkDir)
}
func TestPostPublishCleanupNotRunWhenCurrentPointerMissing(t *testing.T) {
func TestPostPublishCleanupNotRunWhenCommitPointerMissing(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterPublish = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"current_pointer_written": false}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"current_commit_pointer_key": ""}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
@@ -216,11 +216,11 @@ func TestPostPublishCleanupNotRunWhenOutputIsMissing(t *testing.T) {
assertExists(t, artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID))
}
func TestPostPublishCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) {
func TestPostPublishCleanupNotRunWhenCommittedManifestUploadFails(t *testing.T) {
cfg, seed, _ := publishStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterPublish = true
failKey := seed.sessionPrefix + "current/manifest.json"
failKey := artifacts.S3RunSessionManifestKey(seed.sessionPrefix, seed.runID)
publishStageImpl, err := stage.Select("publish")
if err != nil {
@@ -229,19 +229,19 @@ func TestPostPublishCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) {
_, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
})
if err == nil || !strings.Contains(err.Error(), "current manifest") {
t.Fatalf("executeStages() error = %v, want current-manifest failure", err)
if err == nil || !strings.Contains(err.Error(), "immutable object") {
t.Fatalf("executeStages() error = %v, want committed-manifest failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostPublishCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) {
func TestPostPublishCleanupNotRunWhenCommitPointerUploadFails(t *testing.T) {
cfg, seed, _ := publishStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterPublish = true
failKey := seed.sessionPrefix + "current/run_id.txt"
failKey := artifacts.S3CurrentCommitPointerKey(seed.sessionPrefix)
publishStageImpl, err := stage.Select("publish")
if err != nil {
@@ -250,8 +250,8 @@ func TestPostPublishCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) {
_, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
})
if err == nil || !strings.Contains(err.Error(), "current run pointer") {
t.Fatalf("executeStages() error = %v, want current-run-pointer failure", err)
if err == nil || !strings.Contains(err.Error(), "current commit pointer") {
t.Fatalf("executeStages() error = %v, want current-commit-pointer failure", err)
}
assertExists(t, seed.spoolAudioDir)
@@ -259,6 +259,7 @@ func TestPostPublishCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) {
}
type cleanupSeed struct {
runID string
runWorkDir string
otherRunDir string
spoolAudioDir string
@@ -312,6 +313,7 @@ func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) {
}
return cfg, cleanupSeed{
runID: runID,
runWorkDir: runWorkDir,
otherRunDir: otherRunDir,
spoolAudioDir: spoolAudioDir,

53
internal/artifacts/remote_commit.go
View File

@@ -19,6 +19,7 @@ type RemoteArtifactType string
const (
RemoteArtifactTypeSessionManifest RemoteArtifactType = "session_manifest"
RemoteArtifactTypeRunManifest RemoteArtifactType = "run_manifest"
RemoteArtifactTypeRunFile RemoteArtifactType = "run_file"
RemoteArtifactTypePublishedOutput RemoteArtifactType = "published_output"
RemoteArtifactTypePreviousArtifact RemoteArtifactType = "previous_artifact"
)
@@ -68,15 +69,38 @@ func (m RemoteCommitManifest) Validate() error {
return fmt.Errorf("remote commit artifacts are required")
}
destinations := make(map[string]struct{}, len(m.Artifacts))
if err := ValidateRemoteArtifactMapping(m.Artifacts); err != nil {
return err
}
for index, artifact := range m.Artifacts {
if err := artifact.Validate(); err != nil {
return fmt.Errorf("remote commit artifact %d: %w", index, err)
}
}
return nil
}
// ValidateRemoteArtifactMapping verifies that source and destination identities
// form one unambiguous immutable-object mapping.
func ValidateRemoteArtifactMapping(artifacts []RemoteArtifact) error {
if len(artifacts) == 0 {
return fmt.Errorf("remote commit artifacts are required")
}
destinations := make(map[string]struct{}, len(artifacts))
sources := make(map[string]struct{}, len(artifacts))
for index, artifact := range artifacts {
if err := validateRemoteArtifactIdentity(artifact); err != nil {
return fmt.Errorf("remote commit artifact %d: %w", index, err)
}
if _, exists := destinations[artifact.DestinationKey]; exists {
return fmt.Errorf("remote commit declares duplicate destination %q", artifact.DestinationKey)
}
destinations[artifact.DestinationKey] = struct{}{}
sourceKey := string(artifact.Type) + "\x00" + artifact.Source
if _, exists := sources[sourceKey]; exists {
return fmt.Errorf("remote commit declares ambiguous %s source %q", artifact.Type, artifact.Source)
}
sources[sourceKey] = struct{}{}
}
return nil
}
@@ -98,16 +122,8 @@ func (m RemoteCommitManifest) ValidateForSessionPrefix(sessionPrefix string) err
}
func (a RemoteArtifact) Validate() error {
switch a.Type {
case RemoteArtifactTypeSessionManifest, RemoteArtifactTypeRunManifest, RemoteArtifactTypePublishedOutput, RemoteArtifactTypePreviousArtifact:
default:
return fmt.Errorf("unsupported artifact type %q", a.Type)
}
if strings.TrimSpace(a.Source) == "" {
return fmt.Errorf("source is required")
}
if err := validateRemoteObjectKey(a.DestinationKey); err != nil {
return fmt.Errorf("destination key: %w", err)
if err := validateRemoteArtifactIdentity(a); err != nil {
return err
}
if err := validateSHA256(a.SHA256); err != nil {
return fmt.Errorf("sha256: %w", err)
@@ -121,6 +137,21 @@ func (a RemoteArtifact) Validate() error {
return nil
}
func validateRemoteArtifactIdentity(a RemoteArtifact) error {
switch a.Type {
case RemoteArtifactTypeSessionManifest, RemoteArtifactTypeRunManifest, RemoteArtifactTypeRunFile, RemoteArtifactTypePublishedOutput, RemoteArtifactTypePreviousArtifact:
default:
return fmt.Errorf("unsupported artifact type %q", a.Type)
}
if strings.TrimSpace(a.Source) == "" {
return fmt.Errorf("source is required")
}
if err := validateRemoteObjectKey(a.DestinationKey); err != nil {
return fmt.Errorf("destination key: %w", err)
}
return nil
}
func (p CurrentCommitPointer) Validate() error {
if p.FormatVersion != RemoteCommitFormatVersion {
return fmt.Errorf("unsupported current commit pointer format version %d", p.FormatVersion)

6
internal/artifacts/s3_keys.go
View File

@@ -64,6 +64,12 @@ func S3RunCommitKey(sessionPrefix, runID string) string {
return path.Join(strings.TrimSuffix(S3RunPrefix(sessionPrefix, runID), "/"), config.S3CommitFile)
}
// S3RunSessionManifestKey returns the immutable session-manifest snapshot key for one run.
// Format: {session_prefix}/runs/{run_id}/session-manifest.json
func S3RunSessionManifestKey(sessionPrefix, runID string) string {
return path.Join(strings.TrimSuffix(S3RunPrefix(sessionPrefix, runID), "/"), config.S3SessionSnapshotFile)
}
// S3CurrentCommitPointerKey returns the sole mutable selector for a committed run.
// Format: {session_prefix}/current/commit-pointer.json
func S3CurrentCommitPointerKey(sessionPrefix string) string {

3
internal/artifacts/s3_keys_test.go
View File

@@ -35,6 +35,9 @@ func TestS3KeyConstruction(t *testing.T) {
if got, want := S3RunCommitKey(sessionPrefix, runID), wantRunPrefix+"commit.json"; got != want {
t.Fatalf("commit key = %q, want %q", got, want)
}
if got, want := S3RunSessionManifestKey(sessionPrefix, runID), wantRunPrefix+"session-manifest.json"; got != want {
t.Fatalf("session manifest key = %q, want %q", got, want)
}
runPointer := S3CurrentRunPointerKey(sessionPrefix)
if runPointer != "dnd/campaigns/forsaken/sessions/2026-04-19/current/run_id.txt" {

17
internal/config/defaults.go
View File

@@ -83,14 +83,15 @@ const (
PathTranscriptFinal = artifactmodel.TranscriptPathFinal
PathTranscriptFinalTrimmed = artifactmodel.TranscriptPathFinalTrimmed
S3CampaignsSegment = "campaigns"
S3SessionsSegment = "sessions"
S3RunsSegment = "runs"
S3CurrentSegment = "current"
S3ManifestFile = "manifest.json"
S3RunIDFile = "run_id.txt"
S3CommitFile = "commit.json"
S3CommitPointerFile = "commit-pointer.json"
S3CampaignsSegment = "campaigns"
S3SessionsSegment = "sessions"
S3RunsSegment = "runs"
S3CurrentSegment = "current"
S3ManifestFile = "manifest.json"
S3RunIDFile = "run_id.txt"
S3CommitFile = "commit.json"
S3CommitPointerFile = "commit-pointer.json"
S3SessionSnapshotFile = "session-manifest.json"
)
// DefaultPublishOutputs defines the default publish output rules.

472
internal/stage/publish.go
View File

@@ -1,6 +1,7 @@
package stage
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
@@ -12,7 +13,6 @@ import (
"path/filepath"
"sort"
"strings"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
@@ -59,22 +59,20 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if publishDisabled(env) {
return &StageResult{
Metadata: map[string]any{
"stage": "publish",
"skipped": true,
"publish_enabled": false,
"audio_upload_skipped": true,
"current_pointer_written": false,
"stage": "publish",
"skipped": true,
"publish_enabled": false,
"audio_upload_skipped": true,
},
}, nil
}
if publishRunUploadDisabled(env) {
return &StageResult{
Metadata: map[string]any{
"stage": "publish",
"skipped": true,
"upload_run_enabled": false,
"audio_upload_skipped": true,
"current_pointer_written": false,
"stage": "publish",
"skipped": true,
"upload_run_enabled": false,
"audio_upload_skipped": true,
},
}, nil
}
@@ -86,10 +84,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled")
}
runRoot, err := resolvePublishRunRoot(env, m)
runManifestPath, err := resolvePublishRunManifestPath(env, m)
if err != nil {
return nil, fmt.Errorf("publish: resolve run root: %w", err)
return nil, fmt.Errorf("publish: resolve run manifest path: %w", err)
}
runRoot := filepath.Dir(runManifestPath)
runRootInfo, err := os.Lstat(runRoot)
if err != nil {
return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err)
@@ -115,7 +114,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
return nil, fmt.Errorf("publish: run id is required")
}
runManifestFile, err := fileops.OpenConfinedRegularFile(runRoot, "manifest.json")
runManifestFile, err := fileops.OpenConfinedRegularFile(runRoot, filepath.Base(runManifestPath))
if err != nil {
return nil, fmt.Errorf("publish: open run manifest: %w", err)
}
@@ -170,72 +169,83 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
env.Config.Pipeline.Publish.Outputs,
env.Config.Pipeline.Publish.Locks,
env.SelectedArtifactKeys,
sessionPrefix,
runPrefix,
)
if err != nil {
return nil, fmt.Errorf("publish: resolve publish output rules: %w", err)
}
runUploaded := make([]string, 0, len(runFiles))
for index, file := range runFiles {
key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath)
if _, err := uploader.UploadReader(ctx, runSources[index], key, storage.UploadOptions{}); err != nil {
return nil, fmt.Errorf("publish: upload run file %q to %q: %w", file.RelativePath, key, err)
}
runUploaded = append(runUploaded, file.RelativePath)
}
publishedUploaded := make([]string, 0, len(publishOutputs))
for _, publishedOutput := range publishOutputs {
key := artifacts.S3PublishedOutputKey(sessionPrefix, publishedOutput.Dest)
if _, err := env.ObjectStore.Upload(ctx, publishedOutput.LocalPath, key, storage.UploadOptions{}); err != nil {
return nil, fmt.Errorf("publish: upload published output source %q to %q: %w", publishedOutput.Source, key, err)
}
publishedUploaded = append(publishedUploaded, publishedOutput.Dest)
}
runUploaded := publishUploadPaths(runFiles)
publishedUploaded := publishOutputPaths(publishOutputs)
previousUploaded := publishUploadPaths(previousFiles)
commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)
previousUploaded := make([]string, 0, len(previousFiles))
for index, file := range previousFiles {
key := artifacts.S3PublishedOutputKey(sessionPrefix, file.RelativePath)
if _, err := uploader.UploadReader(ctx, previousSources[index], key, storage.UploadOptions{}); err != nil {
return nil, fmt.Errorf("publish: upload previous file %q to %q: %w", file.RelativePath, key, err)
}
previousUploaded = append(previousUploaded, file.RelativePath)
transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, publishOutputs)
if err != nil {
return nil, fmt.Errorf("publish: build immutable upload mapping: %w", err)
}
defer func() { closePublishTransfers(transfers) }()
currentManifestKey, currentRunPointerKey := artifacts.ResolveCurrentStateKeys(sessionPrefix)
manifestTempPath, err := writeCurrentManifestSnapshot(m, publishMetadataPreview(
manifestTempPath, err := writeCommittedSessionManifest(m, publishMetadata(
bucket,
runPrefix,
sessionPrefix,
runUploaded,
publishedUploaded,
previousUploaded,
skippedOptionalOutputs,
skippedUnselectedOutputs,
lockedOutputs,
currentManifestKey,
commitKey,
pointerKey,
))
if err != nil {
return nil, fmt.Errorf("publish: build current manifest snapshot: %w", err)
return nil, fmt.Errorf("publish: build committed session manifest: %w", err)
}
defer func() { _ = os.Remove(manifestTempPath) }()
if _, err := env.ObjectStore.Upload(ctx, manifestTempPath, currentManifestKey, storage.UploadOptions{
ContentType: "application/json",
}); err != nil {
return nil, fmt.Errorf("publish: upload current manifest to %q: %w", currentManifestKey, err)
if err := appendCommittedSessionManifest(&transfers, manifestTempPath, artifacts.S3RunSessionManifestKey(sessionPrefix, runID)); err != nil {
return nil, fmt.Errorf("publish: add committed session manifest: %w", err)
}
if err := validatePublishTransferPlan(transfers, commitKey); err != nil {
return nil, fmt.Errorf("publish: validate immutable upload mapping: %w", err)
}
runIDTempPath, err := writeCurrentRunIDPointer(runID)
commit := artifacts.RemoteCommitManifest{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: strings.TrimSpace(m.Campaign),
SessionID: strings.TrimSpace(m.SessionID),
RunID: runID,
Artifacts: make([]artifacts.RemoteArtifact, 0, len(transfers)),
}
for index := range transfers {
artifact, err := uploadVerifiedPublishTransfer(ctx, env.ObjectStore, uploader, &transfers[index])
if err != nil {
return nil, fmt.Errorf("publish: upload immutable object %q: %w", transfers[index].Artifact.DestinationKey, err)
}
commit.Artifacts = append(commit.Artifacts, artifact)
}
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
if err != nil {
return nil, fmt.Errorf("publish: build current run id pointer: %w", err)
return nil, fmt.Errorf("publish: encode immutable commit: %w", err)
}
defer func() { _ = os.Remove(runIDTempPath) }()
if _, err := env.ObjectStore.Upload(ctx, runIDTempPath, currentRunPointerKey, storage.UploadOptions{
ContentType: "text/plain; charset=utf-8",
}); err != nil {
return nil, fmt.Errorf("publish: upload current run pointer to %q: %w", currentRunPointerKey, err)
commitInfo, err := uploadVerifiedPublishBytes(ctx, env.ObjectStore, uploader, commitKey, commitData, "application/json")
if err != nil {
return nil, fmt.Errorf("publish: upload immutable commit %q: %w", commitKey, err)
}
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: commit.Campaign,
SessionID: commit.SessionID,
RunID: commit.RunID,
CommitKey: commitKey,
CommitSHA256: publishSHA256(commitData),
CommitSize: int64(len(commitData)),
CommitGeneration: commitInfo.ETag,
})
if err != nil {
return nil, fmt.Errorf("publish: encode current commit pointer: %w", err)
}
if _, err := uploader.UploadReader(ctx, bytes.NewReader(pointerData), pointerKey, storage.UploadOptions{ContentType: "application/json"}); err != nil {
return nil, fmt.Errorf("publish: update current commit pointer %q: %w", pointerKey, err)
}
return &StageResult{
@@ -254,9 +264,8 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
"locked_output_count": len(lockedOutputs),
"locked_outputs": lockedOutputMetadata(lockedOutputs),
"current_manifest_key": currentManifestKey,
"current_run_id_key": currentRunPointerKey,
"current_pointer_written": true,
"remote_commit_key": commitKey,
"current_commit_pointer_key": pointerKey,
"audio_upload_skipped": true,
},
}, nil
@@ -270,6 +279,12 @@ type publishOutput struct {
Provenance string
}
type publishTransfer struct {
Artifact artifacts.RemoteArtifact
Source *os.File
TemporaryPath string
}
type publishLockedOutput struct {
Source string
Dest string
@@ -286,6 +301,292 @@ type publishSkippedUnselectedOutput struct {
Required bool
}
func buildPublishTransferPlan(
runPrefix string,
runFiles []publishUploadFile,
runSources []*os.File,
previousFiles []publishUploadFile,
previousSources []*os.File,
publishOutputs []publishOutput,
) ([]publishTransfer, error) {
if len(runFiles) != len(runSources) || len(previousFiles) != len(previousSources) {
return nil, fmt.Errorf("verified publish sources do not match their declared files")
}
transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(publishOutputs))
add := func(artifact artifacts.RemoteArtifact, source *os.File) error {
if source == nil {
return fmt.Errorf("source for %q is required", artifact.DestinationKey)
}
transfers = append(transfers, publishTransfer{Artifact: artifact, Source: source})
return nil
}
for index, file := range runFiles {
artifactType := artifacts.RemoteArtifactTypeRunFile
source := "run:" + file.RelativePath
if file.RelativePath == config.S3ManifestFile {
artifactType = artifacts.RemoteArtifactTypeRunManifest
source = "run.manifest"
}
if err := add(artifacts.RemoteArtifact{
Type: artifactType,
Source: source,
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
}, runSources[index]); err != nil {
closePublishTransfers(transfers)
return nil, err
}
}
for index, file := range previousFiles {
if err := add(artifacts.RemoteArtifact{
Type: artifacts.RemoteArtifactTypePreviousArtifact,
Source: "previous:" + file.SourceRelativePath,
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
}, previousSources[index]); err != nil {
closePublishTransfers(transfers)
return nil, err
}
}
for _, output := range publishOutputs {
source, err := openPublishOutputSource(output.LocalPath)
if err != nil {
closePublishTransfers(transfers)
return nil, fmt.Errorf("open output source %q: %w", output.Source, err)
}
if err := add(artifacts.RemoteArtifact{
Type: artifacts.RemoteArtifactTypePublishedOutput,
Source: output.Source,
DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest),
}, source); err != nil {
_ = source.Close()
closePublishTransfers(transfers)
return nil, err
}
}
sort.Slice(transfers, func(i, j int) bool {
return transfers[i].Artifact.DestinationKey < transfers[j].Artifact.DestinationKey
})
return transfers, nil
}
func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error {
if transfers == nil {
return fmt.Errorf("transfer plan is required")
}
source, err := os.Open(localPath)
if err != nil {
return err
}
*transfers = append(*transfers, publishTransfer{
Artifact: artifacts.RemoteArtifact{
Type: artifacts.RemoteArtifactTypeSessionManifest,
Source: "session.manifest",
DestinationKey: destinationKey,
},
Source: source,
TemporaryPath: localPath,
})
sort.Slice(*transfers, func(i, j int) bool {
return (*transfers)[i].Artifact.DestinationKey < (*transfers)[j].Artifact.DestinationKey
})
return nil
}
func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) error {
artifactsToUpload := make([]artifacts.RemoteArtifact, 0, len(transfers))
for _, transfer := range transfers {
if transfer.Artifact.DestinationKey == commitKey {
return fmt.Errorf("artifact destination %q is reserved for the immutable commit", commitKey)
}
artifactsToUpload = append(artifactsToUpload, transfer.Artifact)
}
return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload)
}
func openPublishOutputSource(localPath string) (*os.File, error) {
path := filepath.Clean(strings.TrimSpace(localPath))
if path == "" || path == "." {
return nil, fmt.Errorf("local path is required")
}
return fileops.OpenConfinedRegularFile(filepath.Dir(path), filepath.Base(path))
}
func publishOutputRemoteKey(runPrefix, destination string) string {
return artifacts.S3RunRelativeDestinationKey(runPrefix, filepath.ToSlash(filepath.Join("published", destination)))
}
func uploadVerifiedPublishTransfer(
ctx context.Context,
store storage.ObjectStore,
uploader storage.ReaderUploader,
transfer *publishTransfer,
) (artifacts.RemoteArtifact, error) {
if transfer == nil || transfer.Source == nil {
return artifacts.RemoteArtifact{}, fmt.Errorf("transfer source is required")
}
if err := preparePublishTransfer(transfer); err != nil {
return artifacts.RemoteArtifact{}, err
}
if existing, found, err := verifyExistingImmutableObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.SHA256, transfer.Artifact.Size); err != nil {
return artifacts.RemoteArtifact{}, err
} else if found {
transfer.Artifact.Generation = existing.ETag
return transfer.Artifact, nil
}
info, err := uploader.UploadReader(ctx, transfer.Source, transfer.Artifact.DestinationKey, storage.UploadOptions{})
if err != nil {
return artifacts.RemoteArtifact{}, err
}
verified, err := verifyPublishedObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.Size, info.ETag)
if err != nil {
return artifacts.RemoteArtifact{}, err
}
transfer.Artifact.Generation = verified.ETag
return transfer.Artifact, nil
}
func uploadVerifiedPublishBytes(
ctx context.Context,
store storage.ObjectStore,
uploader storage.ReaderUploader,
key string,
data []byte,
contentType string,
) (storage.ObjectInfo, error) {
checksum := publishSHA256(data)
if existing, found, err := verifyExistingImmutableObject(ctx, store, key, checksum, int64(len(data))); err != nil {
return storage.ObjectInfo{}, err
} else if found {
return existing, nil
}
info, err := uploader.UploadReader(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: contentType})
if err != nil {
return storage.ObjectInfo{}, err
}
return verifyPublishedObject(ctx, store, key, int64(len(data)), info.ETag)
}
func verifyExistingImmutableObject(ctx context.Context, store storage.ObjectStore, key, wantSHA256 string, wantSize int64) (storage.ObjectInfo, bool, error) {
exists, err := store.Exists(ctx, key)
if err != nil {
return storage.ObjectInfo{}, false, fmt.Errorf("check immutable object %q: %w", key, err)
}
if !exists {
return storage.ObjectInfo{}, false, nil
}
path, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-verify-immutable-*.bin")
if err != nil {
return storage.ObjectInfo{}, false, fmt.Errorf("download existing immutable object %q: %w", key, err)
}
defer func() { _ = os.Remove(path) }()
file, err := os.Open(path)
if err != nil {
return storage.ObjectInfo{}, false, fmt.Errorf("open existing immutable object %q: %w", key, err)
}
hash := sha256.New()
size, copyErr := io.Copy(hash, file)
closeErr := file.Close()
if copyErr != nil {
return storage.ObjectInfo{}, false, fmt.Errorf("checksum existing immutable object %q: %w", key, copyErr)
}
if closeErr != nil {
return storage.ObjectInfo{}, false, fmt.Errorf("close existing immutable object %q: %w", key, closeErr)
}
if size != wantSize || hex.EncodeToString(hash.Sum(nil)) != wantSHA256 {
return storage.ObjectInfo{}, false, fmt.Errorf("immutable object %q already exists with different content", key)
}
info, err := verifyPublishedObject(ctx, store, key, wantSize, "")
if err != nil {
return storage.ObjectInfo{}, false, err
}
return info, true, nil
}
func preparePublishTransfer(transfer *publishTransfer) error {
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
return fmt.Errorf("rewind source %q: %w", transfer.Artifact.Source, err)
}
info, err := transfer.Source.Stat()
if err != nil {
return fmt.Errorf("stat source %q: %w", transfer.Artifact.Source, err)
}
if !info.Mode().IsRegular() {
return fmt.Errorf("source %q is not a regular file", transfer.Artifact.Source)
}
hash := sha256.New()
if _, err := io.Copy(hash, transfer.Source); err != nil {
return fmt.Errorf("checksum source %q: %w", transfer.Artifact.Source, err)
}
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
return fmt.Errorf("rewind source %q after checksum: %w", transfer.Artifact.Source, err)
}
transfer.Artifact.Size = info.Size()
transfer.Artifact.SHA256 = hex.EncodeToString(hash.Sum(nil))
return nil
}
func verifyPublishedObject(ctx context.Context, store storage.ObjectStore, key string, wantSize int64, uploadedGeneration string) (storage.ObjectInfo, error) {
objects, err := store.List(ctx, key)
if err != nil {
return storage.ObjectInfo{}, fmt.Errorf("list uploaded object %q: %w", key, err)
}
var found *storage.ObjectInfo
for _, object := range objects {
if object.Key != key {
continue
}
if found != nil {
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is ambiguous", key)
}
copy := object
found = &copy
}
if found == nil {
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is missing", key)
}
if found.Size != wantSize {
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q size is %d, want %d", key, found.Size, wantSize)
}
if strings.TrimSpace(found.ETag) == "" {
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q has no generation", key)
}
if strings.TrimSpace(uploadedGeneration) != "" && found.ETag != uploadedGeneration {
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q generation changed during verification", key)
}
return *found, nil
}
func closePublishTransfers(transfers []publishTransfer) {
for _, transfer := range transfers {
if transfer.Source != nil {
_ = transfer.Source.Close()
}
if transfer.TemporaryPath != "" {
_ = os.Remove(transfer.TemporaryPath)
}
}
}
func publishUploadPaths(files []publishUploadFile) []string {
paths := make([]string, 0, len(files))
for _, file := range files {
paths = append(paths, file.RelativePath)
}
return paths
}
func publishOutputPaths(outputs []publishOutput) []string {
paths := make([]string, 0, len(outputs))
for _, output := range outputs {
paths = append(paths, output.Dest)
}
return paths
}
func publishSHA256(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
func publishDisabled(env *Env) bool {
cfg := env.Config.Pipeline.Publish
if cfg == nil {
@@ -318,7 +619,7 @@ func validatePublishPrerequisites(m *manifest.Manifest) error {
return nil
}
func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) {
func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, error) {
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
if sessionID == "" && m != nil {
sessionID = strings.TrimSpace(m.SessionID)
@@ -338,13 +639,13 @@ func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) {
return "", fmt.Errorf("run id is required")
}
canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
canonicalExists, err := directoryExists(canonical)
canonical := filepath.Clean(artifacts.SessionRunManifestPathForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
canonicalExists, err := directoryExists(filepath.Dir(canonical))
if err != nil {
return "", fmt.Errorf("check canonical run root %q: %w", canonical, err)
return "", fmt.Errorf("check canonical run manifest directory %q: %w", filepath.Dir(canonical), err)
}
if !canonicalExists {
return "", fmt.Errorf("run root not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, canonical)
return "", fmt.Errorf("run manifest directory not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, filepath.Dir(canonical))
}
return canonical, nil
}
@@ -387,7 +688,7 @@ func resolvePublishOutputs(
rules []config.PublishOutputRule,
locks []config.PublishLockRule,
selectedArtifactKeys []string,
sessionPrefix string,
runPrefix string,
) ([]publishOutput, []string, []publishSkippedUnselectedOutput, []publishLockedOutput, error) {
out := make([]publishOutput, 0, len(rules))
skippedOptionalOutputs := make([]string, 0)
@@ -423,7 +724,7 @@ func resolvePublishOutputs(
lockedOutputs = append(lockedOutputs, publishLockedOutput{
Source: source,
Dest: dest,
RemoteKey: artifacts.S3PublishedOutputKey(sessionPrefix, dest),
RemoteKey: publishOutputRemoteKey(runPrefix, dest),
Reason: strings.TrimSpace(lock.Reason),
Required: required,
})
@@ -442,7 +743,7 @@ func resolvePublishOutputs(
lockedOutputs = append(lockedOutputs, publishLockedOutput{
Source: source,
Dest: dest,
RemoteKey: artifacts.S3PublishedOutputKey(sessionPrefix, dest),
RemoteKey: publishOutputRemoteKey(runPrefix, dest),
Reason: strings.TrimSpace(lock.Reason),
Required: required,
LocalPath: resolved.Path,
@@ -810,7 +1111,7 @@ func directoryExists(path string) (bool, error) {
return false, err
}
func writeCurrentManifestSnapshot(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
if m == nil {
return "", fmt.Errorf("manifest is required")
}
@@ -840,8 +1141,7 @@ func writeCurrentManifestSnapshot(m *manifest.Manifest, publishMetadata map[stri
clone.Stages[name] = &stageCopy
}
now := time.Now().UTC()
clone.MarkStageSucceeded("publish", now, nil)
clone.MarkStageSucceeded("publish", clone.UpdatedAt, nil)
if sr := clone.Stages["publish"]; sr != nil {
sr.Metadata = publishMetadata
}
@@ -852,7 +1152,7 @@ func writeCurrentManifestSnapshot(m *manifest.Manifest, publishMetadata map[stri
}
data = append(data, '\n')
tmp, err := os.CreateTemp("", "narratio-current-manifest-*.json")
tmp, err := os.CreateTemp("", "narratio-committed-session-manifest-*.json")
if err != nil {
return "", fmt.Errorf("create temp file: %w", err)
}
@@ -867,31 +1167,16 @@ func writeCurrentManifestSnapshot(m *manifest.Manifest, publishMetadata map[stri
return path, nil
}
func writeCurrentRunIDPointer(runID string) (string, error) {
tmp, err := os.CreateTemp("", "narratio-current-run-id-*.txt")
if err != nil {
return "", fmt.Errorf("create temp file: %w", err)
}
path := tmp.Name()
if _, err := tmp.WriteString(runID + "\n"); err != nil {
_ = tmp.Close()
return "", fmt.Errorf("write temp run id pointer: %w", err)
}
if err := tmp.Close(); err != nil {
return "", fmt.Errorf("close temp run id pointer: %w", err)
}
return path, nil
}
func publishMetadataPreview(
bucket, runPrefix, sessionPrefix string,
func publishMetadata(
bucket, runPrefix string,
runUploaded []string,
publishedUploaded []string,
previousUploaded []string,
skippedOptionalOutputs []string,
skippedUnselectedOutputs []publishSkippedUnselectedOutput,
lockedOutputs []publishLockedOutput,
currentManifestKey string,
commitKey string,
pointerKey string,
) map[string]any {
return map[string]any{
"stage": "publish",
@@ -908,9 +1193,8 @@ func publishMetadataPreview(
"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
"locked_output_count": len(lockedOutputs),
"locked_outputs": lockedOutputMetadata(lockedOutputs),
"current_manifest_key": currentManifestKey,
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
"current_pointer_written": false,
"remote_commit_key": commitKey,
"current_commit_pointer_key": pointerKey,
"audio_upload_skipped": true,
}
}

303
internal/stage/publish_test.go
View File

@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"errors"
"io"
"os"
"path/filepath"
"reflect"
@@ -127,8 +126,8 @@ func TestPublishUploadsRunRecordPublishedOutputsAndCurrentPointer(t *testing.T)
}
}
trimmedKey := sessionPrefix + "transcripts/final.trimmed.json"
recapKey := sessionPrefix + "artifacts/session_recap.md"
trimmedKey := publishOutputRemoteKey(runPrefix, "transcripts/final.trimmed.json")
recapKey := publishOutputRemoteKey(runPrefix, "artifacts/session_recap.md")
if _, ok := fake.Objects[trimmedKey]; !ok {
t.Fatalf("missing published output key %q", trimmedKey)
}
@@ -136,16 +135,20 @@ func TestPublishUploadsRunRecordPublishedOutputsAndCurrentPointer(t *testing.T)
t.Fatalf("missing published output key %q", recapKey)
}
currentManifestKey := sessionPrefix + "current/manifest.json"
currentRunIDKey := sessionPrefix + "current/run_id.txt"
if _, ok := fake.Objects[currentManifestKey]; !ok {
t.Fatalf("missing current manifest key %q", currentManifestKey)
commitKey := artifacts.S3RunCommitKey(sessionPrefix, m.RunID)
pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)
if _, ok := fake.Objects[commitKey]; !ok {
t.Fatalf("missing immutable commit key %q", commitKey)
}
if _, ok := fake.Objects[currentRunIDKey]; !ok {
t.Fatalf("missing current run pointer key %q", currentRunIDKey)
if _, ok := fake.Objects[pointerKey]; !ok {
t.Fatalf("missing current commit pointer key %q", pointerKey)
}
if got := string(fake.Objects[currentRunIDKey].Data); got != m.RunID+"\n" {
t.Fatalf("run pointer contents = %q, want %q", got, m.RunID+"\\n")
pointer, err := artifacts.DecodeCurrentCommitPointer(fake.Objects[pointerKey].Data)
if err != nil {
t.Fatalf("DecodeCurrentCommitPointer() error = %v", err)
}
if pointer.CommitKey != commitKey || pointer.RunID != m.RunID {
t.Fatalf("pointer = %#v, want selected commit %q", pointer, commitKey)
}
audioKey := runPrefix + "audio/speaker.flac"
@@ -157,12 +160,12 @@ func TestPublishUploadsRunRecordPublishedOutputsAndCurrentPointer(t *testing.T)
if len(uploads) == 0 {
t.Fatal("expected uploads")
}
if uploads[len(uploads)-1].Key != currentRunIDKey {
t.Fatalf("last upload key = %q, want current run pointer key %q", uploads[len(uploads)-1].Key, currentRunIDKey)
if uploads[len(uploads)-1].Key != pointerKey {
t.Fatalf("last upload key = %q, want current commit pointer key %q", uploads[len(uploads)-1].Key, pointerKey)
}
if result.Metadata["current_pointer_written"] != true {
t.Fatalf("metadata = %#v, want current_pointer_written=true", result.Metadata)
if result.Metadata["current_commit_pointer_key"] != pointerKey {
t.Fatalf("metadata = %#v, want current commit pointer key", result.Metadata)
}
if result.Metadata["published_files_uploaded"] != 2 {
t.Fatalf("metadata published_files_uploaded = %#v, want 2", result.Metadata["published_files_uploaded"])
@@ -195,8 +198,8 @@ func TestPublishUploadsPreviousCacheWhenPresent(t *testing.T) {
t.Fatalf("Run() error = %v", err)
}
previousManifestKey := m.S3SessionPrefix + "previous/manifest.json"
previousRecapKey := m.S3SessionPrefix + "previous/artifacts/session_recap.md"
previousManifestKey := artifacts.S3RunRelativeDestinationKey(m.S3RunPrefix, "previous/manifest.json")
previousRecapKey := artifacts.S3RunRelativeDestinationKey(m.S3RunPrefix, "previous/artifacts/session_recap.md")
if _, ok := fake.Objects[previousManifestKey]; !ok {
t.Fatalf("missing published previous manifest key %q", previousManifestKey)
}
@@ -405,10 +408,10 @@ func TestPublishUsesCustomOutputRules(t *testing.T) {
}
fake := env.ObjectStore.(*storage.FakeBackend)
if _, ok := fake.Objects[m.S3SessionPrefix+"published/trimmed.json"]; !ok {
if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "published/trimmed.json")]; !ok {
t.Fatalf("missing custom published trimmed key")
}
if _, ok := fake.Objects[m.S3SessionPrefix+"published/recap.md"]; !ok {
if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "published/recap.md")]; !ok {
t.Fatalf("missing custom published recap key")
}
}
@@ -427,7 +430,7 @@ func TestPublishUploadsExplicitExtractionAndPreservesManifestMetadata(t *testing
}
fake := env.ObjectStore.(*storage.FakeBackend)
durableBundleRoot := filepath.Dir(filepath.Dir(lanePath))
publishedKey := m.S3SessionPrefix + "artifacts/encounters.json"
publishedKey := publishOutputRemoteKey(m.S3RunPrefix, "artifacts/encounters.json")
if got := string(fake.Objects[publishedKey].Data); got != `{"encounters":[]}` {
t.Fatalf("published extraction = %q", got)
}
@@ -448,13 +451,13 @@ func TestPublishUploadsExplicitExtractionAndPreservesManifestMetadata(t *testing
if result.Metadata["published_files_uploaded"] != 1 {
t.Fatalf("published_files_uploaded = %#v, want 1", result.Metadata["published_files_uploaded"])
}
if uploads := fake.Uploads; len(uploads) == 0 || uploads[len(uploads)-1].Key != m.S3SessionPrefix+"current/run_id.txt" {
if uploads := fake.Uploads; len(uploads) == 0 || uploads[len(uploads)-1].Key != artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix) {
t.Fatalf("last upload = %#v, want current run pointer", uploads)
}
var current manifest.Manifest
if err := json.Unmarshal(fake.Objects[m.S3SessionPrefix+"current/manifest.json"].Data, &current); err != nil {
t.Fatalf("unmarshal current manifest: %v", err)
if err := json.Unmarshal(fake.Objects[artifacts.S3RunSessionManifestKey(m.S3SessionPrefix, m.RunID)].Data, &current); err != nil {
t.Fatalf("unmarshal committed session manifest: %v", err)
}
extractRecord := current.Stages["extract"]
if extractRecord == nil || len(extractRecord.Outputs) != 2 {
@@ -528,7 +531,7 @@ func TestPublishArtifactSelectionDoesNotFilterExtractionOutputs(t *testing.T) {
if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("Run() error = %v", err)
}
if _, ok := env.ObjectStore.(*storage.FakeBackend).Objects[m.S3SessionPrefix+"artifacts/encounters.json"]; !ok {
if _, ok := env.ObjectStore.(*storage.FakeBackend).Objects[publishOutputRemoteKey(m.S3RunPrefix, "artifacts/encounters.json")]; !ok {
t.Fatal("explicit extraction output was filtered by Scriptorium artifact selection")
}
}
@@ -565,10 +568,10 @@ func TestPublishSkipsRequiredUnselectedConfiguredOutput(t *testing.T) {
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if _, ok := fake.Objects[m.S3SessionPrefix+"transcripts/final.trimmed.json"]; !ok {
if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "transcripts/final.trimmed.json")]; !ok {
t.Fatalf("missing built-in published trimmed key")
}
if _, ok := fake.Objects[m.S3SessionPrefix+"artifacts/session_recap.md"]; ok {
if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "artifacts/session_recap.md")]; ok {
t.Fatalf("unexpected unselected recap published output upload")
}
skipped := result.Metadata["skipped_unselected_outputs"].([]map[string]any)
@@ -581,7 +584,7 @@ func TestPublishSkipsRequiredUnselectedConfiguredOutput(t *testing.T) {
if result.Metadata["locked_output_count"] != 0 {
t.Fatalf("locked_output_count = %#v, want 0", result.Metadata["locked_output_count"])
}
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; !ok {
if _, ok := fake.Objects[artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)]; !ok {
t.Fatalf("missing current pointer")
}
}
@@ -616,7 +619,7 @@ func TestPublishLockedSelectedOutputSkipsAsLocked(t *testing.T) {
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if _, ok := fake.Objects[m.S3SessionPrefix+"artifacts/session_recap.md"]; ok {
if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "artifacts/session_recap.md")]; ok {
t.Fatalf("unexpected locked recap published output upload")
}
if result.Metadata["locked_output_count"] != 1 {
@@ -665,11 +668,11 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) {
t.Fatalf("Run() error = %v", err)
}
trimmedKey := m.S3SessionPrefix + "transcripts/final.trimmed.json"
trimmedKey := publishOutputRemoteKey(m.S3RunPrefix, "transcripts/final.trimmed.json")
if _, ok := fake.Objects[trimmedKey]; ok {
t.Fatalf("locked published output key %q should not be uploaded", trimmedKey)
}
recapKey := m.S3SessionPrefix + "artifacts/session_recap.md"
recapKey := publishOutputRemoteKey(m.S3RunPrefix, "artifacts/session_recap.md")
if _, ok := fake.Objects[recapKey]; !ok {
t.Fatalf("unlocked published output key %q should be uploaded", recapKey)
}
@@ -678,9 +681,9 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) {
t.Fatalf("run-local locked source output %q should still be uploaded", runTrimmedKey)
}
currentRunIDKey := m.S3SessionPrefix + "current/run_id.txt"
if len(fake.Uploads) == 0 || fake.Uploads[len(fake.Uploads)-1].Key != currentRunIDKey {
t.Fatalf("last upload = %#v, want current run pointer %q", fake.Uploads, currentRunIDKey)
pointerKey := artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)
if len(fake.Uploads) == 0 || fake.Uploads[len(fake.Uploads)-1].Key != pointerKey {
t.Fatalf("last upload = %#v, want current commit pointer %q", fake.Uploads, pointerKey)
}
if result.Metadata["published_files_uploaded"] != 1 {
@@ -703,7 +706,7 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) {
t.Fatalf("locked published output metadata = %#v", locked[0])
}
currentManifestKey := m.S3SessionPrefix + "current/manifest.json"
currentManifestKey := artifacts.S3RunSessionManifestKey(m.S3SessionPrefix, m.RunID)
var current map[string]any
if err := json.Unmarshal(fake.Objects[currentManifestKey].Data, &current); err != nil {
t.Fatalf("unmarshal current manifest: %v", err)
@@ -735,12 +738,12 @@ func TestPublishLockedRequiredMissingOutputSucceeds(t *testing.T) {
}
fake := env.ObjectStore.(*storage.FakeBackend)
mergedKey := m.S3SessionPrefix + "transcripts/base.json"
mergedKey := publishOutputRemoteKey(m.S3RunPrefix, "transcripts/base.json")
if _, ok := fake.Objects[mergedKey]; ok {
t.Fatalf("locked missing published output key %q should not be uploaded", mergedKey)
}
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; !ok {
t.Fatalf("current run pointer should be written for locked missing output")
if _, ok := fake.Objects[artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)]; !ok {
t.Fatalf("current commit pointer should be written for locked missing output")
}
locked := result.Metadata["locked_outputs"].([]map[string]any)
if len(locked) != 1 {
@@ -757,7 +760,7 @@ func TestPublishLockDoesNotOverwriteExistingOutput(t *testing.T) {
{Source: "narratio.transcript.final_trimmed", Reason: "already published"},
}
fake := env.ObjectStore.(*storage.FakeBackend)
trimmedKey := m.S3SessionPrefix + "transcripts/final.trimmed.json"
trimmedKey := publishOutputRemoteKey(m.S3RunPrefix, "transcripts/final.trimmed.json")
fake.SeedObject(storage.FakeObject{Key: trimmedKey, Data: []byte("previously published\n")})
if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
@@ -797,49 +800,145 @@ func TestPublishFailsWhenCanonicalRunRootMissing(t *testing.T) {
if err == nil {
t.Fatal("expected missing run-root error, got nil")
}
if !strings.Contains(err.Error(), "run root not found") {
t.Fatalf("error = %v, want missing run-root error", err)
if !strings.Contains(err.Error(), "run manifest directory not found") {
t.Fatalf("error = %v, want missing run-manifest directory error", err)
}
}
func TestPublishDoesNotWriteCurrentPointerWhenOutputUploadFails(t *testing.T) {
func TestPublishFailurePreservesPreviousCommittedSnapshotAndRetryIsDeterministic(t *testing.T) {
env, m, _ := publishFixture(t)
fake := env.ObjectStore.(*storage.FakeBackend)
trimmedKey := m.S3SessionPrefix + "transcripts/final.trimmed.json"
oldRunID := "20260515T010203Z-a1b2c3d4"
seedCommittedCurrentState(t, fake, m, oldRunID)
pointerKey := artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)
priorPointer := append([]byte(nil), fake.Objects[pointerKey].Data...)
failingKey := publishOutputRemoteKey(m.S3RunPrefix, "transcripts/final.trimmed.json")
fake.UploadHook = func(call storage.FakeUploadCall) error {
if call.Key == failingKey {
return errors.New("forced upload failure")
}
return nil
}
origUploadErr := fake.UploadErr
fake.UploadErr = nil
failingKey := trimmedKey
_, err := publishStage{}.Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "forced upload failure") {
t.Fatalf("Run() error = %v, want immutable object upload failure", err)
}
if got := fake.Objects[pointerKey].Data; !reflect.DeepEqual(got, priorPointer) {
t.Fatalf("current pointer changed after failed publication: %q", got)
}
state, err := artifacts.LoadCurrentState(context.Background(), fake, m.S3SessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true})
if err != nil {
t.Fatalf("LoadCurrentState() error = %v", err)
}
if state.RunID != oldRunID {
t.Fatalf("readable current run = %q, want %q", state.RunID, oldRunID)
}
for _, upload := range fake.Uploads {
if upload.Key == pointerKey {
t.Fatalf("current pointer was uploaded after a pre-commit failure")
}
}
fake.UploadHook = nil
fake.Uploads = nil
originalUpload := fake.Upload
_ = originalUpload
// Use UploadErr toggle by checking call sequence in postcondition.
// First failure point is published output upload; simulate by setting error immediately before published output key write.
// We cannot hook FakeBackend per-key without changing public behavior; use dedicated backend wrapper instead.
env.ObjectStore = &publishedOutputFailingStore{delegate: fake, failKey: failingKey}
_, err := publishStage{}.Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "upload published output source") {
t.Fatalf("Run() error = %v, want published output upload failure", err)
if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("retry Run() error = %v", err)
}
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok {
t.Fatalf("unexpected current pointer write on published output failure")
if len(fake.Uploads) == 0 || fake.Uploads[len(fake.Uploads)-1].Key != pointerKey {
t.Fatalf("retry uploads = %#v, want pointer as final mutation", fake.Uploads)
}
state, err = artifacts.LoadCurrentState(context.Background(), fake, m.S3SessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true})
if err != nil {
t.Fatalf("LoadCurrentState() after retry error = %v", err)
}
if state.RunID != m.RunID {
t.Fatalf("retried current run = %q, want %q", state.RunID, m.RunID)
}
if state.Commit == nil {
t.Fatal("retried current state has no immutable commit")
}
destinations := make([]string, 0, len(state.Commit.Artifacts))
for _, artifact := range state.Commit.Artifacts {
destinations = append(destinations, artifact.DestinationKey)
}
if !sort.StringsAreSorted(destinations) {
t.Fatalf("retried immutable mapping is not deterministic: %v", destinations)
}
if _, exists := fake.Objects[m.S3SessionPrefix+"current/manifest.json"]; exists {
t.Fatal("new publication wrote legacy current manifest")
}
fake.UploadErr = origUploadErr
}
func TestPublishDoesNotWriteCurrentPointerWhenCurrentManifestUploadFails(t *testing.T) {
func TestPublishRejectsConflictingImmutableObject(t *testing.T) {
env, m, _ := publishFixture(t)
fake := env.ObjectStore.(*storage.FakeBackend)
env.ObjectStore = &publishedOutputFailingStore{delegate: fake, failKey: m.S3SessionPrefix + "current/manifest.json"}
conflictingKey := publishOutputRemoteKey(m.S3RunPrefix, "transcripts/final.trimmed.json")
fake.SeedObject(storage.FakeObject{Key: conflictingKey, Data: []byte("different content\n")})
_, err := publishStage{}.Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "current manifest") {
t.Fatalf("Run() error = %v, want current manifest upload failure", err)
_, err := (publishStage{}).Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "already exists with different content") {
t.Fatalf("Run() error = %v, want immutable conflict", err)
}
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok {
t.Fatalf("unexpected current pointer write when current manifest upload fails")
if _, exists := fake.Objects[artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)]; exists {
t.Fatal("current pointer was updated after immutable object conflict")
}
}
func TestPublishRejectsAmbiguousOrCollidingOutputMappings(t *testing.T) {
tests := []struct {
name string
outputs []config.PublishOutputRule
}{
{
name: "ambiguous source",
outputs: []config.PublishOutputRule{
{Source: "narratio.transcript.final_trimmed", Dest: "published/first.json", Required: boolPtr(true)},
{Source: "narratio.transcript.final_trimmed", Dest: "published/second.json", Required: boolPtr(true)},
},
},
{
name: "colliding destination",
outputs: []config.PublishOutputRule{
{Source: "narratio.transcript.final_trimmed", Dest: "published/shared.json", Required: boolPtr(true)},
{Source: "narratio.artifact.session_recap", Dest: "published/shared.json", Required: boolPtr(true)},
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
env, m, _ := publishFixture(t)
env.Config.Pipeline.Publish.Outputs = test.outputs
_, err := (publishStage{}).Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "immutable upload mapping") {
t.Fatalf("Run() error = %v, want invalid mapping", err)
}
if uploads := env.ObjectStore.(*storage.FakeBackend).Uploads; len(uploads) != 0 {
t.Fatalf("uploads = %#v, want mapping failure before upload", uploads)
}
})
}
}
func TestPublishKeepsSameBasenameSourcesDistinct(t *testing.T) {
env, m, _ := publishFixture(t)
writeStageTestFile(t, filepath.Join(env.Config.Pipeline.Workspace.Root, "work", m.Campaign, m.SessionID, "reports", "session_recap.md"), "# other recap\n")
env.Config.Pipeline.Scriptorium.Artifacts["other_recap"] = config.ScriptoriumArtifactConfig{
Enabled: true, PromptID: "dnd.other_recap", OutputPath: "reports/session_recap.md",
}
env.Config.Pipeline.Publish.Outputs = []config.PublishOutputRule{
{Source: "narratio.artifact.session_recap", Dest: "published/first/session_recap.md", Required: boolPtr(true)},
{Source: "narratio.artifact.other_recap", Dest: "published/second/session_recap.md", Required: boolPtr(true)},
}
if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("Run() error = %v", err)
}
fake := env.ObjectStore.(*storage.FakeBackend)
first := publishOutputRemoteKey(m.S3RunPrefix, "published/first/session_recap.md")
second := publishOutputRemoteKey(m.S3RunPrefix, "published/second/session_recap.md")
if string(fake.Objects[first].Data) != "# recap\n" || string(fake.Objects[second].Data) != "# other recap\n" {
t.Fatalf("same-basename outputs were not kept distinct: first=%q second=%q", fake.Objects[first].Data, fake.Objects[second].Data)
}
}
@@ -1022,35 +1121,53 @@ func configurePublishExtractionFixture(t *testing.T, env *Env, m *manifest.Manif
return lanePath
}
type publishedOutputFailingStore struct {
delegate *storage.FakeBackend
failKey string
}
func (s *publishedOutputFailingStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) {
return s.delegate.List(ctx, prefix)
}
func (s *publishedOutputFailingStore) Download(ctx context.Context, key, localPath string) error {
return s.delegate.Download(ctx, key, localPath)
}
func (s *publishedOutputFailingStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) {
return storage.ObjectInfo{}, errors.New("forced upload failure")
func seedCommittedCurrentState(t *testing.T, fake *storage.FakeBackend, source *manifest.Manifest, runID string) {
t.Helper()
copy := *source
copy.RunID = runID
manifestData, err := json.Marshal(&copy)
if err != nil {
t.Fatalf("marshal committed manifest: %v", err)
}
return s.delegate.Upload(ctx, localPath, key, opts)
}
func (s *publishedOutputFailingStore) UploadReader(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) {
return storage.ObjectInfo{}, errors.New("forced upload failure")
manifestData = append(manifestData, '\n')
manifestKey := artifacts.S3RunSessionManifestKey(copy.S3SessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: manifestKey, Data: manifestData})
manifestObject := fake.Objects[manifestKey]
commit := artifacts.RemoteCommitManifest{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: copy.Campaign,
SessionID: copy.SessionID,
RunID: runID,
Artifacts: []artifacts.RemoteArtifact{{
Type: artifacts.RemoteArtifactTypeSessionManifest,
Source: "session.manifest",
DestinationKey: manifestKey,
SHA256: publishSHA256(manifestData),
Size: int64(len(manifestData)),
Generation: manifestObject.ETag,
}},
}
return s.delegate.UploadReader(ctx, source, key, opts)
}
func (s *publishedOutputFailingStore) Exists(ctx context.Context, key string) (bool, error) {
return s.delegate.Exists(ctx, key)
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
if err != nil {
t.Fatalf("encode remote commit: %v", err)
}
commitKey := artifacts.S3RunCommitKey(copy.S3SessionPrefix, runID)
fake.SeedObject(storage.FakeObject{Key: commitKey, Data: commitData})
commitObject := fake.Objects[commitKey]
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
FormatVersion: artifacts.RemoteCommitFormatVersion,
Campaign: copy.Campaign,
SessionID: copy.SessionID,
RunID: runID,
CommitKey: commitKey,
CommitSHA256: publishSHA256(commitData),
CommitSize: int64(len(commitData)),
CommitGeneration: commitObject.ETag,
})
if err != nil {
t.Fatalf("encode current commit pointer: %v", err)
}
fake.SeedObject(storage.FakeObject{Key: artifacts.S3CurrentCommitPointerKey(copy.S3SessionPrefix), Data: pointerData})
}
func boolPtr(v bool) *bool {