From 361dbb4ca86c92e9624d10503fc958f3ec962d77 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 20:05:24 +0000 Subject: [PATCH] Publish immutable remote commits --- docs/internal/artifacts.md | 9 +- docs/internal/stage-publish.md | 16 +- docs/internal/workspace.md | 3 +- docs/operations.md | 20 +- docs/policy/architecture.md | 12 +- docs/roadmap/implementation.md | 2 +- internal/adapters/storage/fake.go | 22 +- internal/app/post_publish_cleanup.go | 8 +- internal/app/post_publish_cleanup_test.go | 30 +- internal/artifacts/remote_commit.go | 53 ++- internal/artifacts/s3_keys.go | 6 + internal/artifacts/s3_keys_test.go | 3 + internal/config/defaults.go | 17 +- internal/stage/publish.go | 472 +++++++++++++++++----- internal/stage/publish_test.go | 303 +++++++++----- 15 files changed, 725 insertions(+), 251 deletions(-) diff --git a/docs/internal/artifacts.md b/docs/internal/artifacts.md index c6b5ac9..aaf81ac 100644 --- a/docs/internal/artifacts.md +++ b/docs/internal/artifacts.md @@ -102,10 +102,9 @@ restore, status and validation checks, and previous-cache planning. Core helpers: -- `LoadCurrentRunPointer` -- `LoadCurrentManifest` - `LoadCurrentState` - `ValidateCurrentStateIdentity` +- `RemoteCommitManifest` and `CurrentCommitPointer` Typed missing-state errors: @@ -133,6 +132,12 @@ Caller policy is intentionally outside artifacts helpers: - spool/cache paths; - S3 session/run/current-state key layout. +New publication creates run-scoped immutable objects, including +`runs/{run_id}/commit.json` and `runs/{run_id}/session-manifest.json`. The sole +mutable selector is `current/commit-pointer.json`; readers verify its selected +commit and declared object generations/checksums. Legacy current-pair loading +is confined to `current_state_legacy.go` for migration only. + Campaign, session, and Narratio run IDs are validated as portable opaque segments at configuration and artifact boundaries before they can be used in a workspace or S3 namespace. Previous-artifact destinations remain typed, diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 685a034..19fbb95 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -16,8 +16,8 @@ Upload run/session outputs to object storage and atomically advance remote curre - uploaded invocation record and selected publish outputs; - uploaded durable previous-session cache files when present; -- updated remote current manifest; and -- remote current-run commit marker, written last. +- immutable run-scoped commit manifest; and +- current commit pointer, written last. Exact remote placement and the operator workflow belong in [Operations](../operations.md#publish-workflow). @@ -40,7 +40,10 @@ Exact remote placement and the operator workflow belong in - selected artifact filter applies to configured artifact sources only. - locked outputs are skipped intentionally (including required ones). - optional missing outputs are skipped; required missing unlocked outputs fail. -- writes remote current manifest before current run pointer. +- creates one complete immutable source-to-destination mapping before upload; +- uploads and verifies every declared immutable object and the commit manifest; +- updates `current/commit-pointer.json` exactly once, last; and +- does not write the legacy `current/manifest.json` or `current/run_id.txt` pair. ## Metadata Signals @@ -51,12 +54,13 @@ Includes counts/lists for: - skipped optional outputs - skipped unselected outputs - locked outputs -- current-state key paths -- `current_pointer_written` +- remote commit and current-pointer key paths ## Invariants -- `current/run_id.txt` is the remote commit marker and is written last. +- `current/commit-pointer.json` is the remote commit marker and is written last. +- run files, selected outputs, previous-cache files, and the committed session + manifest are all declared by an immutable commit under the run prefix. - run and previous uploads contain only manifest-declared regular files opened from verified descriptors; symlinks, special files, replacement races, and undeclared entries are rejected or ignored before uploads begin. diff --git a/docs/internal/workspace.md b/docs/internal/workspace.md index 36192ae..1b507e3 100644 --- a/docs/internal/workspace.md +++ b/docs/internal/workspace.md @@ -55,7 +55,8 @@ lock only while their context remains active, and report a release failure. Automatic post-publish cleanup: - only runs when publish actually executed and succeeded; -- requires `uploaded=true` and `current_pointer_written=true` metadata; +- requires `uploaded=true`, a remote commit key, and a current commit-pointer + key in publish metadata; - consumes the resolved cleanup policy described in [Configuration](../config.md); - refuses unsafe deletes (root delete, out-of-root delete, and symlinked diff --git a/docs/operations.md b/docs/operations.md index 839a1d9..42210fe 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -200,13 +200,15 @@ Publish commit model: - uploads eligible run files under `{session_prefix}/runs/{run_id}/`, excluding audio and the run-local Notarius staging bundle; -- uploads configured published outputs, including only explicitly configured - extraction lanes; -- uploads `previous/**` cache files when present; -- writes `current/manifest.json`; -- writes `current/run_id.txt` last. +- uploads configured published outputs and `previous/**` cache files into the + same immutable run scope, including only explicitly configured extraction + lanes; +- writes `{session_prefix}/runs/{run_id}/commit.json` after all declared + immutable objects are uploaded and verified; and +- writes `{session_prefix}/current/commit-pointer.json` once, last. -`current/run_id.txt` is the remote current-state commit marker. +`current/commit-pointer.json` is the remote current-state commit marker. It +selects exactly one immutable commit, which declares the complete object set. ## Remote Commit Migration @@ -217,10 +219,10 @@ commit by version, checksum, size, and storage generation; committed artifacts are also checksum- and generation-bound. Readers accept this contract now and strictly reject mismatched or unknown data. -The publish workflow above remains the legacy writer until its planned cutover. Legacy reads are limited to a coherent `current/manifest.json` and -`current/run_id.txt` pair; a torn pair is rejected. New remote commit state does -not carry local `current_pointer_written` metadata. +`current/run_id.txt` pair; a torn pair is rejected. New publication does not +write that pair and remote commit state does not carry local +`current_pointer_written` metadata. ## Publish Locks diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index fd27f73..75d4345 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -167,13 +167,13 @@ contracts belong under [Integrations](../integrations/). ## Publish Commit Boundary Publish has one explicit remote commit boundary. A remote run becomes current -only after Narratio has successfully uploaded the run record, required published -outputs, `current/manifest.json`, and finally `current/run_id.txt`. +only after Narratio has successfully uploaded its immutable run-scoped objects, +the immutable commit manifest, and finally the current commit pointer. -`current/run_id.txt` is the commit marker and must be written last. Failed, -incomplete, skipped, or uncommitted publish attempts must not be presented as -current remote state. Publish locks remain authoritative and are not bypassed by -a forced run. +`current/commit-pointer.json` is the sole mutable selector and must be written +exactly once, last. Failed, incomplete, skipped, or uncommitted publish attempts +must not be presented as current remote state. Publish locks remain authoritative +and are not bypassed by a forced run. Automatic local cleanup is permitted only after a successful publish commit, only when explicitly configured, and only through the path-safety guardrails. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 8d15533..391d131 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -29,7 +29,7 @@ All stages are pending when this plan is created. | 11 | Make manifest and run identity singular | COR-001, TST-006 | Completed | | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Completed | | 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Completed | -| 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Pending | +| 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Completed | | 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Pending | | 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Pending | | 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Pending | diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index bce4ceb..f6b336b 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -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 { diff --git a/internal/app/post_publish_cleanup.go b/internal/app/post_publish_cleanup.go index f758703..791f9dd 100644 --- a/internal/app/post_publish_cleanup.go +++ b/internal/app/post_publish_cleanup.go @@ -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, "" } diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index 76d8df7..965180d 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -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, diff --git a/internal/artifacts/remote_commit.go b/internal/artifacts/remote_commit.go index 0c5951f..f614e88 100644 --- a/internal/artifacts/remote_commit.go +++ b/internal/artifacts/remote_commit.go @@ -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) diff --git a/internal/artifacts/s3_keys.go b/internal/artifacts/s3_keys.go index ef5454f..b7501dc 100644 --- a/internal/artifacts/s3_keys.go +++ b/internal/artifacts/s3_keys.go @@ -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 { diff --git a/internal/artifacts/s3_keys_test.go b/internal/artifacts/s3_keys_test.go index 4418aef..5362d6f 100644 --- a/internal/artifacts/s3_keys_test.go +++ b/internal/artifacts/s3_keys_test.go @@ -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" { diff --git a/internal/config/defaults.go b/internal/config/defaults.go index f0123cb..143be10 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -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. diff --git a/internal/stage/publish.go b/internal/stage/publish.go index 702b049..4a28f3f 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -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 = © + } + 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, } } diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index 0470bc7..279d2d9 100644 --- a/internal/stage/publish_test.go +++ b/internal/stage/publish_test.go @@ -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, ¤t); err != nil { - t.Fatalf("unmarshal current manifest: %v", err) + if err := json.Unmarshal(fake.Objects[artifacts.S3RunSessionManifestKey(m.S3SessionPrefix, m.RunID)].Data, ¤t); 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, ¤t); 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(©) + 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 {