From 4bc33773fa7094aae6934e7f773c89e4911bd553 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 13:04:08 +0000 Subject: [PATCH] Integrate artifact locks into publish --- docs/internal/stage-publish.md | 62 +++--- docs/roadmap/implementation.md | 2 +- internal/app/runner.go | 26 --- internal/app/runner_composition.go | 17 -- internal/app/runner_dependencies.go | 26 +-- internal/app/runner_test.go | 20 +- internal/stage/publish.go | 128 ++++------- internal/stage/publish_commit.go | 17 ++ internal/stage/publish_output_plan.go | 41 ++-- internal/stage/publish_outputs.go | 190 ++++++++-------- internal/stage/publish_test.go | 306 +++++++++++++++++++------- internal/stage/publish_transfer.go | 40 +++- internal/stage/stage.go | 4 - 13 files changed, 503 insertions(+), 376 deletions(-) diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 45600ed..c3a2d75 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -9,8 +9,8 @@ Upload run/session outputs to object storage and atomically advance remote curre - successful preceding stages from the [canonical stage set](overview.md#pipeline-stage-set) - declared run-local diagnostic files - resolved publish output rules -- effective publish locks (static + remote merged lock set), revalidated at the - remote commit boundary +- independent per-source remote lock objects read at planning start +- the validated prior current commit when an initially locked source may carry - durable previous-session cache files when present ## Outputs @@ -61,19 +61,28 @@ five-outcome model and its secret-free metadata projection. - publishes extraction lanes only through explicit configured output rules; neither run-local nor durable Notarius bundles are scanned or uploaded wholesale. - 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. +- plans every configured output into exactly one source-sorted outcome: + `published`, `carried_locked`, `locked_unpublished`, `optional_missing`, or + `unselected`. +- an initially locked selected output carries its exact prior + `published_output` declaration when one exists. The declaration may point to + an earlier run in the same session and retains its source, key, checksum, + size, and generation; the bytes are not copied. +- verifies a carried object's exact key, size, and generation through object + metadata without downloading or opening the local output source. +- an initially locked source with no prior declaration remains absent even + when the output rule is required. Optional missing unlocked outputs remain + absent; required missing unlocked outputs fail. - 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 -- writes only the selected `current/commit-pointer.json` control record. -- rechecks remote lock state immediately before the pointer update. A newly - committed lock aborts selection, leaving any uploaded immutable attempt - unselected. -- reads the mutable remote lock document through a direct limit-plus-one read - capped by `MaxRemoteLockStoreBytes` (1 MiB), retaining the generation returned - with the opened body for conditional updates. Oversized lock documents fail - before YAML decoding; published artifact payloads do not use this limit. +- writes the immutable commit from the combined new and exact carried + declarations. +- rereads every initially unlocked participating source after the immutable + commit. A newly present lock aborts selection, leaving the uploaded immutable + attempt unselected. Removing an initial lock may conservatively produce one + carry because the initial snapshot controls the plan. +- updates `current/commit-pointer.json` exactly once, last, and writes no other + mutable publication control record. ## Metadata Signals @@ -81,9 +90,8 @@ Includes counts/lists for: - run uploads - published output uploads - previous uploads -- skipped optional outputs -- skipped unselected outputs -- locked outputs +- one `publish_output_outcomes` collection containing all five output states, + exact declared keys, and key-derived origin runs where applicable - remote commit and current-pointer key paths - the run identifier selected by the commit @@ -98,28 +106,30 @@ Includes counts/lists for: - run-local diagnostics, including Notarius receipt and stderr files, are archived only when recorded by the session manifest. A run with no declared current-run diagnostics legitimately uploads none. -- publish locks are not overridden by `--force`; remote locks are revalidated - immediately before current-state selection. +- independent publish locks are not overridden by `--force`; every initially + unlocked relevant source is reread immediately before current-state + selection. - post-commit local cleanup is authorized by the committed publish metadata and is durably recorded by the application lifecycle before any local deletion. ## Resume Semantics The versioned semantic fingerprint covers enabled behavior, run-upload policy, -normalized source/destination/required output rules, static lock policy, and -the remote backend, bucket, region, endpoint, and root-prefix identity. Rule -and lock ordering is canonicalized. Credential environment names, -path-addressing transport mode, local workspace placement, and run identifiers -are excluded. Remote locks remain mutable state and are still revalidated at -the commit boundary; semantic evidence does not replace that safety check. +normalized source/destination/required output rules, the currently supported +configured lock field, and the remote backend, bucket, region, endpoint, and +root-prefix identity. Rule and lock ordering is canonicalized. Credential +environment names, path-addressing transport mode, local workspace placement, +and run identifiers are excluded. Independent remote locks remain mutable state +and are read directly by publish at planning and commit boundaries; semantic +evidence does not replace those safety checks. The commit boundary and cleanup gate are normative architecture invariants; see [Architecture](../policy/architecture.md#publish-commit-boundary). ## Related Contracts And Tests -- [Configuration](../config.md#publish-configuration-summary) owns output and - static-lock fields. +- [Configuration](../config.md#publish-configuration-summary) owns output + rules. - [Operations](../operations.md#publish-locks) owns remote lock lifecycle and physical remote state. - [Artifact Internals](artifacts.md) explains source resolution and current-state diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 4a879bf..1742350 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -254,7 +254,7 @@ One plan drives both declarations and metadata. ## Stage 7 — Integrate Locks Into Publish -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/app/runner.go b/internal/app/runner.go index cb683a6..7af8d11 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -529,32 +529,6 @@ func needsNotariusForRun(cfg *config.Config, stages []stage.Stage) bool { return false } -func needsRemoteLocksForRun(cfg *config.Config, stages []stage.Stage) bool { - if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { - return false - } - publishRequested := false - for _, s := range stages { - if s != nil && s.Name() == "publish" { - publishRequested = true - break - } - } - if !publishRequested { - return false - } - if cfg.Pipeline.Publish == nil { - return false - } - if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled { - return false - } - if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun { - return false - } - return cfg.Pipeline.Storage.S3 != nil -} - func configuredScriptoriumArtifacts(cfg *config.Config) map[string]config.ScriptoriumArtifactConfig { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Scriptorium == nil { return nil diff --git a/internal/app/runner_composition.go b/internal/app/runner_composition.go index 446f85f..627e64b 100644 --- a/internal/app/runner_composition.go +++ b/internal/app/runner_composition.go @@ -1,11 +1,9 @@ package app import ( - "context" "fmt" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" - "gitea.maximumdirect.net/eric/narratio/internal/config" ) func composeInvocation(runtime *invocationRuntime) error { @@ -47,21 +45,6 @@ func composeInvocation(runtime *invocationRuntime) error { } env.ObjectStore = objectStore } - if needsRemoteLocksForRun(env.Config, stages) { - locks, err := dependencies.loadEffectiveLocks(runtime.ctx, env.Config, env.ObjectStore) - if err != nil { - return runtime.persistTerminalFailure(fmt.Errorf("load remote publish locks: %w", err)) - } - applyEffectiveLocks(env.Config, locks.All) - staticLocks := append([]config.PublishLockRule(nil), locks.Static...) - env.RevalidatePublishLocks = func(recheckCtx context.Context) ([]config.PublishLockRule, error) { - remote, _, _, err := dependencies.loadRemoteLockStore(recheckCtx, env.Config, env.ObjectStore) - if err != nil { - return nil, err - } - return config.MergePublishLockRules(staticLocks, remote.Locks), nil - } - } if env.Notifier == nil && stagesContainAny(stages, "notify") { env.Notifier = ¬ify.NoopSender{} } diff --git a/internal/app/runner_dependencies.go b/internal/app/runner_dependencies.go index 59cfc1b..85545be 100644 --- a/internal/app/runner_dependencies.go +++ b/internal/app/runner_dependencies.go @@ -15,17 +15,15 @@ import ( ) type invocationDependencies struct { - now func() time.Time - newRunID func() (string, error) - newWhisperX func(*config.Config) (whisperx.Client, error) - newSeriatim func(*config.Config) (seriatim.Runner, error) - newAudita func(*config.Config) (audita.Runner, error) - newNotarius func() notarius.Runner - newScriptorium func() scriptorium.Runner - newObjectStore func(context.Context, *config.Config) (storage.ObjectStore, error) - loadEffectiveLocks func(context.Context, *config.Config, storage.ObjectStore) (*effectiveLocks, error) - loadRemoteLockStore func(context.Context, *config.Config, storage.ObjectStore) (*config.PublishLockStore, string, string, error) - removeScopedDir func(root, target, policy string) error + now func() time.Time + newRunID func() (string, error) + newWhisperX func(*config.Config) (whisperx.Client, error) + newSeriatim func(*config.Config) (seriatim.Runner, error) + newAudita func(*config.Config) (audita.Runner, error) + newNotarius func() notarius.Runner + newScriptorium func() scriptorium.Runner + newObjectStore func(context.Context, *config.Config) (storage.ObjectStore, error) + removeScopedDir func(root, target, policy string) error } func (dependencies invocationDependencies) normalized() invocationDependencies { @@ -55,12 +53,6 @@ func (dependencies invocationDependencies) normalized() invocationDependencies { return initializeObjectStoreBackend(ctx, cfg, storage.NewObjectStoreFromConfig) } } - if dependencies.loadEffectiveLocks == nil { - dependencies.loadEffectiveLocks = loadEffectiveLocks - } - if dependencies.loadRemoteLockStore == nil { - dependencies.loadRemoteLockStore = loadRemoteLockStore - } if dependencies.removeScopedDir == nil { dependencies.removeScopedDir = removeRunScopedDir } diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 6c17c9e..3c92964 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -537,16 +537,20 @@ func TestExecuteStagesPublishSkipsRequiredUnselectedConfiguredOutput(t *testing. t.Fatalf("Load manifest error = %v", err) } meta := loadedManifest.Stages["publish"].Metadata - skipped, ok := meta["skipped_unselected_outputs"].([]any) - if !ok || len(skipped) != 1 { - t.Fatalf("skipped_unselected_outputs = %#v, want one item", meta["skipped_unselected_outputs"]) - } - item, ok := skipped[0].(map[string]any) + outcomes, ok := meta["publish_output_outcomes"].([]any) if !ok { - t.Fatalf("skipped item = %#v, want object", skipped[0]) + t.Fatalf("publish_output_outcomes = %#v, want array", meta["publish_output_outcomes"]) } - if item["source"] != "narratio.artifact.session_recap" || item["dest"] != "artifacts/session_recap.md" || item["required"] != true { - t.Fatalf("skipped item = %#v, want required session_recap published output", item) + var item map[string]any + for _, raw := range outcomes { + candidate, candidateOK := raw.(map[string]any) + if candidateOK && candidate["source"] == "narratio.artifact.session_recap" { + item = candidate + break + } + } + if item == nil || item["status"] != "unselected" || item["destination"] != "artifacts/session_recap.md" { + t.Fatalf("publish output outcome = %#v, want unselected session_recap", item) } } diff --git a/internal/stage/publish.go b/internal/stage/publish.go index 83c541c..2ca0688 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -114,26 +114,45 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("publish: build runtime artifact catalog: %w", err) } - publishOutputs, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, err := resolvePublishOutputs( - sessionPaths, - m, + publishRules, err := resolvePublishRules( runtimeCatalog, env.Config.Pipeline.Publish.Outputs, - env.Config.Pipeline.Publish.Locks, env.SelectedArtifactKeys, - runPrefix, ) if err != nil { return nil, fmt.Errorf("publish: resolve publish output rules: %w", err) } + initialLocks, err := readPublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, publishRules) + if err != nil { + return nil, fmt.Errorf("publish: read publish output locks: %w", err) + } + var priorCommit *artifacts.RemoteCommitManifest + if publishRulesNeedPriorCommit(publishRules, initialLocks) { + priorCommit, err = loadPriorPublishCommit(ctx, env.ObjectStore, sessionPrefix, m.Campaign, m.SessionID) + if err != nil { + return nil, fmt.Errorf("publish: load prior current commit: %w", err) + } + } + discoveredOutputs, err := discoverPublishOutputs(sessionPaths, m, runtimeCatalog, publishRules, initialLocks) + if err != nil { + return nil, fmt.Errorf("publish: discover publish outputs: %w", err) + } + outputPlan, err := buildPublishOutputPlan(sessionPrefix, runPrefix, publishRules, discoveredOutputs, initialLocks, priorCommit) + if err != nil { + return nil, fmt.Errorf("publish: plan publish outputs: %w", err) + } runUploaded := publishUploadPaths(runFiles) - publishedUploaded := publishOutputPaths(publishOutputs) + publishedUploaded := publishOutputPaths(outputPlan) previousUploaded := publishUploadPaths(previousFiles) commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID) pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix) - transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, publishOutputs) + carriedDeclarations, err := verifyCarriedPublishOutputs(ctx, env.ObjectStore, outputPlan) + if err != nil { + return nil, fmt.Errorf("publish: %w", err) + } + transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, outputPlan) if err != nil { return nil, fmt.Errorf("publish: build immutable upload mapping: %w", err) } @@ -146,9 +165,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S runUploaded, publishedUploaded, previousUploaded, - skippedOptionalOutputs, - skippedUnselectedOutputs, - lockedOutputs, + outputPlan, commitKey, pointerKey, )) @@ -165,10 +182,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S // Durability boundary: transfer every declared immutable object, write the // immutable commit, perform pre-pointer checks, then write the sole mutable // current pointer exactly once. - declarations, err := uploadPublishTransfers(ctx, env.ObjectStore, uploader, transfers) + uploadedDeclarations, err := uploadPublishTransfers(ctx, env.ObjectStore, uploader, transfers) if err != nil { return nil, fmt.Errorf("publish: %w", err) } + declarations := combinePublishDeclarations(uploadedDeclarations, carriedDeclarations) committed, err := writeImmutablePublishCommit( ctx, env.ObjectStore, @@ -182,51 +200,24 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("publish: %w", err) } - if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil { + if err := revalidatePublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, outputPlan); err != nil { return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err) } if err := writePublishCurrentPointer(ctx, uploader, committed); err != nil { return nil, fmt.Errorf("publish: %w", err) } - return &StageResult{ - Metadata: map[string]any{ - "stage": "publish", - "uploaded": true, - "s3_bucket": bucket, - "s3_run_prefix": runPrefix, - "run_files_uploaded": len(runUploaded), - "run_uploaded_paths": runUploaded, - "published_files_uploaded": len(publishedUploaded), - "published_paths": publishedUploaded, - "previous_files_uploaded": len(previousUploaded), - "previous_uploaded_paths": previousUploaded, - "skipped_optional_outputs": skippedOptionalOutputs, - "skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs), - "locked_output_count": len(lockedOutputs), - "locked_outputs": lockedOutputMetadata(lockedOutputs), - "remote_commit_key": commitKey, - "current_commit_pointer_key": pointerKey, - "audio_upload_skipped": true, - }, - }, nil -} - -func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publishOutput) error { - if env == nil || env.RevalidatePublishLocks == nil || len(outputs) == 0 { - return nil - } - locks, err := env.RevalidatePublishLocks(ctx) - if err != nil { - return err - } - lockSet := publishLockSet(locks) - for _, output := range outputs { - if lock, locked := lockSet[output.Source]; locked { - return fmt.Errorf("source %q is locked: %s", output.Source, strings.TrimSpace(lock.Reason)) - } - } - return nil + return &StageResult{Metadata: publishMetadata( + bucket, + runPrefix, + runID, + runUploaded, + publishedUploaded, + previousUploaded, + outputPlan, + commitKey, + pointerKey, + )}, nil } func publishDisabled(env *Env) bool { @@ -310,9 +301,7 @@ func publishMetadata( runUploaded []string, publishedUploaded []string, previousUploaded []string, - skippedOptionalOutputs []string, - skippedUnselectedOutputs []publishSkippedUnselectedOutput, - lockedOutputs []publishLockedOutput, + outputPlan *publishOutputPlan, commitKey string, pointerKey string, ) map[string]any { @@ -328,40 +317,9 @@ func publishMetadata( "published_paths": append([]string(nil), publishedUploaded...), "previous_files_uploaded": len(previousUploaded), "previous_uploaded_paths": append([]string(nil), previousUploaded...), - "skipped_optional_outputs": append([]string(nil), skippedOptionalOutputs...), - "skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs), - "locked_output_count": len(lockedOutputs), - "locked_outputs": lockedOutputMetadata(lockedOutputs), + "publish_output_outcomes": publishOutputOutcomeMetadata(outputPlan), "remote_commit_key": commitKey, "current_commit_pointer_key": pointerKey, "audio_upload_skipped": true, } } - -func skippedUnselectedOutputMetadata(skipped []publishSkippedUnselectedOutput) []map[string]any { - out := make([]map[string]any, 0, len(skipped)) - for _, item := range skipped { - out = append(out, map[string]any{ - "source": item.Source, - "dest": item.Dest, - "required": item.Required, - }) - } - return out -} - -func lockedOutputMetadata(locked []publishLockedOutput) []map[string]any { - out := make([]map[string]any, 0, len(locked)) - for _, item := range locked { - out = append(out, map[string]any{ - "source": item.Source, - "dest": item.Dest, - "remote_key": item.RemoteKey, - "reason": item.Reason, - "required": item.Required, - "local_path": item.LocalPath, - "provenance": item.Provenance, - }) - } - return out -} diff --git a/internal/stage/publish_commit.go b/internal/stage/publish_commit.go index 6a35c1c..a0f268c 100644 --- a/internal/stage/publish_commit.go +++ b/internal/stage/publish_commit.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "os" "strings" @@ -13,6 +14,22 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) +func loadPriorPublishCommit(ctx context.Context, store storage.ObjectStore, sessionPrefix, campaign, sessionID string) (*artifacts.RemoteCommitManifest, error) { + current, err := artifacts.LoadCurrentState(ctx, store, sessionPrefix, artifacts.CurrentStateValidation{ + ExpectedCampaign: campaign, ExpectedSessionID: sessionID, ValidateRunID: true, + }) + if errors.Is(err, artifacts.ErrCurrentCommitPointerMissing) { + return nil, nil + } + if err != nil { + return nil, err + } + if current == nil || current.Commit == nil { + return nil, fmt.Errorf("prior current state has no remote commit") + } + return current.Commit, nil +} + type publishedCommit struct { Manifest artifacts.RemoteCommitManifest Key string diff --git a/internal/stage/publish_output_plan.go b/internal/stage/publish_output_plan.go index 071b21b..497dfa6 100644 --- a/internal/stage/publish_output_plan.go +++ b/internal/stage/publish_output_plan.go @@ -74,24 +74,12 @@ func buildPublishOutputPlan( return nil, err } - orderedRules := append([]resolvedPublishRule(nil), rules...) - sort.Slice(orderedRules, func(i, j int) bool { return orderedRules[i].Source < orderedRules[j].Source }) - seenSources := make(map[string]struct{}, len(orderedRules)) - configuredDestinations := make(map[string]string, len(orderedRules)) + orderedRules, err := normalizeResolvedPublishRules(rules) + if err != nil { + return nil, err + } outcomes := make([]publishOutputOutcome, 0, len(orderedRules)) for _, rule := range orderedRules { - if err := validateResolvedPublishRule(rule); err != nil { - return nil, err - } - if _, duplicate := seenSources[rule.Source]; duplicate { - return nil, fmt.Errorf("publish output source %q is configured more than once", rule.Source) - } - seenSources[rule.Source] = struct{}{} - if other, collision := configuredDestinations[rule.Destination]; collision { - return nil, fmt.Errorf("publish output destination %q is configured for both %q and %q", rule.Destination, other, rule.Source) - } - configuredDestinations[rule.Destination] = rule.Source - outcome := publishOutputOutcome{ Source: rule.Source, Destination: rule.Destination, Required: rule.Required, } @@ -147,6 +135,27 @@ func buildPublishOutputPlan( return &publishOutputPlan{Outcomes: outcomes}, nil } +func normalizeResolvedPublishRules(rules []resolvedPublishRule) ([]resolvedPublishRule, error) { + ordered := append([]resolvedPublishRule(nil), rules...) + sort.Slice(ordered, func(i, j int) bool { return ordered[i].Source < ordered[j].Source }) + seenSources := make(map[string]struct{}, len(ordered)) + configuredDestinations := make(map[string]string, len(ordered)) + for _, rule := range ordered { + if err := validateResolvedPublishRule(rule); err != nil { + return nil, err + } + if _, duplicate := seenSources[rule.Source]; duplicate { + return nil, fmt.Errorf("publish output source %q is configured more than once", rule.Source) + } + seenSources[rule.Source] = struct{}{} + if other, collision := configuredDestinations[rule.Destination]; collision { + return nil, fmt.Errorf("publish output destination %q is configured for both %q and %q", rule.Destination, other, rule.Source) + } + configuredDestinations[rule.Destination] = rule.Source + } + return ordered, nil +} + func validateResolvedPublishRule(rule resolvedPublishRule) error { if strings.TrimSpace(rule.Source) != rule.Source || rule.Source == "" { return fmt.Errorf("publish output source %q is not canonical", rule.Source) diff --git a/internal/stage/publish_outputs.go b/internal/stage/publish_outputs.go index a0cfa3d..60a0102 100644 --- a/internal/stage/publish_outputs.go +++ b/internal/stage/publish_outputs.go @@ -1,55 +1,25 @@ package stage import ( + "context" "errors" "fmt" "path/filepath" "strings" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) -type publishOutput struct { - Source string - Dest string - Required bool - LocalPath string - Provenance string -} - -type publishLockedOutput struct { - Source string - Dest string - RemoteKey string - Reason string - Required bool - LocalPath string - Provenance string -} - -type publishSkippedUnselectedOutput struct { - Source string - Dest string - Required bool -} - -func resolvePublishOutputs( - paths artifacts.SessionPaths, - m *manifest.Manifest, +func resolvePublishRules( catalog *artifacts.ArtifactCatalog, rules []config.PublishOutputRule, - locks []config.PublishLockRule, selectedArtifactKeys []string, - runPrefix string, -) ([]publishOutput, []string, []publishSkippedUnselectedOutput, []publishLockedOutput, error) { - out := make([]publishOutput, 0, len(rules)) - skippedOptionalOutputs := make([]string, 0) - skippedUnselectedOutputs := make([]publishSkippedUnselectedOutput, 0) - lockedOutputs := make([]publishLockedOutput, 0) - lockSet := publishLockSet(locks) +) ([]resolvedPublishRule, error) { + out := make([]resolvedPublishRule, 0, len(rules)) selectedSet := publishSelectedArtifactSet(selectedArtifactKeys) configuredOutputs := configuredOutputPathMapFromCatalog(catalog) extractionOutputs := extractionOutputSetFromCatalog(catalog) @@ -58,51 +28,102 @@ func resolvePublishOutputs( required := rule.Required == nil || *rule.Required dest, err := resolvePublishOutputDest(rule, configuredOutputs, extractionOutputs) if err != nil { - return nil, nil, nil, nil, fmt.Errorf("source %q: %w", source, err) + return nil, fmt.Errorf("source %q: %w", source, err) } + selected := true if len(selectedSet) > 0 { if key, ok := artifactpolicy.ParseConfiguredSource(source); ok { - if _, selected := selectedSet[key]; !selected { - skippedUnselectedOutputs = append(skippedUnselectedOutputs, publishSkippedUnselectedOutput{ - Source: source, Dest: dest, Required: required, - }) - continue - } + _, selected = selectedSet[key] } } - lock, locked := lockSet[source] - resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog) - if err != nil { - if locked { - lockedOutputs = append(lockedOutputs, publishLockedOutput{ - Source: source, Dest: dest, RemoteKey: publishOutputRemoteKey(runPrefix, dest), - Reason: strings.TrimSpace(lock.Reason), Required: required, - }) - continue - } - if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required { - skippedOptionalOutputs = append(skippedOptionalOutputs, dest) - continue - } - if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { - return nil, nil, nil, nil, fmt.Errorf("required output source unavailable: %q", source) - } - return nil, nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err) - } - if locked { - lockedOutputs = append(lockedOutputs, publishLockedOutput{ - Source: source, Dest: dest, RemoteKey: publishOutputRemoteKey(runPrefix, dest), - Reason: strings.TrimSpace(lock.Reason), Required: required, - LocalPath: resolved.Path, Provenance: resolved.Provenance, - }) + out = append(out, resolvedPublishRule{Source: source, Destination: dest, Required: required, Selected: selected}) + } + return normalizeResolvedPublishRules(out) +} + +func discoverPublishOutputs( + paths artifacts.SessionPaths, + m *manifest.Manifest, + catalog *artifacts.ArtifactCatalog, + rules []resolvedPublishRule, + locks []artifacts.RemoteLock, +) ([]discoveredPublishOutput, error) { + lockSet := make(map[string]struct{}, len(locks)) + for _, lock := range locks { + lockSet[lock.Source] = struct{}{} + } + out := make([]discoveredPublishOutput, 0, len(rules)) + for _, rule := range rules { + if !rule.Selected { continue } - out = append(out, publishOutput{ - Source: source, Dest: dest, Required: required, - LocalPath: resolved.Path, Provenance: resolved.Provenance, + if _, locked := lockSet[rule.Source]; locked { + continue + } + resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, rule.Source, catalog) + if err != nil { + if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { + continue + } + return nil, fmt.Errorf("resolve source %q: %w", rule.Source, err) + } + out = append(out, discoveredPublishOutput{ + Source: rule.Source, LocalPath: resolved.Path, Provenance: resolved.Provenance, }) } - return out, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, nil + return out, nil +} + +func readPublishOutputLocks(ctx context.Context, store storage.ObjectStore, sessionPrefix string, rules []resolvedPublishRule) ([]artifacts.RemoteLock, error) { + locks := make([]artifacts.RemoteLock, 0) + for _, rule := range rules { + if !rule.Selected { + continue + } + lookup, err := artifacts.LookupRemoteLock(ctx, store, sessionPrefix, rule.Source) + if err != nil { + return nil, fmt.Errorf("read lock for source %q: %w", rule.Source, err) + } + if lookup.Present { + locks = append(locks, lookup.Lock) + } + } + return locks, nil +} + +func publishRulesNeedPriorCommit(rules []resolvedPublishRule, locks []artifacts.RemoteLock) bool { + locked := make(map[string]struct{}, len(locks)) + for _, lock := range locks { + locked[lock.Source] = struct{}{} + } + for _, rule := range rules { + if !rule.Selected { + continue + } + if _, found := locked[rule.Source]; found { + return true + } + } + return false +} + +func revalidatePublishOutputLocks(ctx context.Context, store storage.ObjectStore, sessionPrefix string, plan *publishOutputPlan) error { + if plan == nil { + return fmt.Errorf("publish output plan is required") + } + for _, outcome := range plan.Outcomes { + if outcome.Status != publishOutputPublished && outcome.Status != publishOutputOptionalMissing { + continue + } + lookup, err := artifacts.LookupRemoteLock(ctx, store, sessionPrefix, outcome.Source) + if err != nil { + return fmt.Errorf("recheck source %q: %w", outcome.Source, err) + } + if lookup.Present { + return fmt.Errorf("source %q became locked before current commit selection: %s", outcome.Source, lookup.Lock.Reason) + } + } + return nil } func publishSelectedArtifactSet(selected []string) map[string]struct{} { @@ -119,20 +140,6 @@ func publishSelectedArtifactSet(selected []string) map[string]struct{} { return out } -func publishLockSet(locks []config.PublishLockRule) map[string]config.PublishLockRule { - out := make(map[string]config.PublishLockRule, len(locks)) - for _, lock := range locks { - source := strings.TrimSpace(lock.Source) - if source == "" { - continue - } - lock.Source = source - lock.Reason = strings.TrimSpace(lock.Reason) - out[source] = lock - } - return out -} - func resolvePublishOutputDest(rule config.PublishOutputRule, configured map[string]string, extractions map[string]struct{}) (string, error) { return artifactpolicy.ResolvePublishedDestinationWithExtractions(rule.Source, rule.Dest, configured, extractions) } @@ -193,10 +200,15 @@ func publishOutputRemoteKey(runPrefix, destination string) string { return artifacts.S3RunRelativeDestinationKey(runPrefix, filepath.ToSlash(filepath.Join("published", destination))) } -func publishOutputPaths(outputs []publishOutput) []string { - paths := make([]string, 0, len(outputs)) - for _, output := range outputs { - paths = append(paths, output.Dest) +func publishOutputPaths(plan *publishOutputPlan) []string { + if plan == nil { + return nil + } + paths := make([]string, 0, len(plan.Outcomes)) + for _, outcome := range plan.Outcomes { + if outcome.Status == publishOutputPublished { + paths = append(paths, outcome.Destination) + } } return paths } diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index e09fdbd..5ffa5b7 100644 --- a/internal/stage/publish_test.go +++ b/internal/stage/publish_test.go @@ -161,6 +161,15 @@ func TestPublishUploadsRunFilesPublishedOutputsAndCurrentPointer(t *testing.T) { 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) } + pointerWrites := 0 + for _, upload := range uploads { + if upload.Key == pointerKey { + pointerWrites++ + } + } + if pointerWrites != 1 { + t.Fatalf("current pointer writes = %d, want exactly one", pointerWrites) + } if result.Metadata["current_commit_pointer_key"] != pointerKey { t.Fatalf("metadata = %#v, want current commit pointer key", result.Metadata) @@ -562,8 +571,8 @@ func TestPublishOptionalMissingExtractionIsSkipped(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if got := result.Metadata["skipped_optional_outputs"].([]string); !reflect.DeepEqual(got, []string{"artifacts/encounters.json"}) { - t.Fatalf("skipped_optional_outputs = %#v", got) + if got := publishOutcomeForSource(t, result, artifacts.ExtractionArtifactSourceID("encounters")); got["status"] != string(publishOutputOptionalMissing) { + t.Fatalf("publish output outcome = %#v", got) } } @@ -594,10 +603,9 @@ func TestPublishSkipsOptionalMissingOutput(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - got, _ := result.Metadata["skipped_optional_outputs"].([]string) - want := []string{"transcripts/base.json"} - if !reflect.DeepEqual(got, want) { - t.Fatalf("skipped_optional_outputs = %#v, want %#v", got, want) + got := publishOutcomeForSource(t, result, "narratio.transcript.base") + if got["status"] != string(publishOutputOptionalMissing) || got["destination"] != "transcripts/base.json" { + t.Fatalf("publish output outcome = %#v", got) } } @@ -621,15 +629,9 @@ func TestPublishSkipsRequiredUnselectedConfiguredOutput(t *testing.T) { 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) - if len(skipped) != 1 { - t.Fatalf("skipped_unselected_outputs = %#v, want one item", skipped) - } - if skipped[0]["source"] != "narratio.artifact.session_recap" || skipped[0]["dest"] != "artifacts/session_recap.md" || skipped[0]["required"] != true { - t.Fatalf("skipped_unselected_outputs[0] = %#v, want session recap", skipped[0]) - } - if result.Metadata["locked_output_count"] != 0 { - t.Fatalf("locked_output_count = %#v, want 0", result.Metadata["locked_output_count"]) + outcome := publishOutcomeForSource(t, result, "narratio.artifact.session_recap") + if outcome["status"] != string(publishOutputUnselected) || outcome["destination"] != "artifacts/session_recap.md" { + t.Fatalf("publish output outcome = %#v, want unselected recap", outcome) } if _, ok := fake.Objects[artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix)]; !ok { t.Fatalf("missing current pointer") @@ -697,18 +699,16 @@ func TestPublishOmitsStaleConfiguredArtifactAndPublishesUnrelatedCurrentArtifact if _, ok := fake.Objects[publishOutputRemoteKey(m.S3RunPrefix, "artifacts/player_handout.md")]; !ok { t.Fatal("unrelated current configured artifact was not uploaded") } - if got := result.Metadata["skipped_optional_outputs"].([]string); !reflect.DeepEqual(got, []string{"artifacts/session_recap.md"}) { - t.Fatalf("skipped_optional_outputs = %#v", got) + if got := publishOutcomeForSource(t, result, "narratio.artifact.session_recap"); got["status"] != string(publishOutputOptionalMissing) { + t.Fatalf("publish output outcome = %#v", got) } } -func TestPublishLockedSelectedOutputSkipsAsLocked(t *testing.T) { +func TestPublishLockedSelectedOutputIsUnpublishedWithoutPriorDeclaration(t *testing.T) { env, m, _ := publishFixture(t) env.SelectedArtifactKeys = []string{"session_recap"} - env.Config.Pipeline.Publish.Locks = []config.PublishLockRule{ - {Source: "narratio.artifact.session_recap", Reason: "reviewed"}, - } fake := env.ObjectStore.(*storage.FakeBackend) + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.artifact.session_recap", "reviewed") result, err := publishStage{}.Run(context.Background(), env, m) if err != nil { @@ -717,12 +717,9 @@ func TestPublishLockedSelectedOutputSkipsAsLocked(t *testing.T) { 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 { - t.Fatalf("locked_output_count = %#v, want 1", result.Metadata["locked_output_count"]) - } - skipped := result.Metadata["skipped_unselected_outputs"].([]map[string]any) - if len(skipped) != 0 { - t.Fatalf("skipped_unselected_outputs = %#v, want empty", skipped) + outcome := publishOutcomeForSource(t, result, "narratio.artifact.session_recap") + if outcome["status"] != string(publishOutputLockedUnpublished) { + t.Fatalf("publish output outcome = %#v", outcome) } } @@ -734,29 +731,30 @@ func TestPublishLockedUnselectedConfiguredOutputSkipsAsUnselected(t *testing.T) PromptID: "dnd.player_handout", OutputPath: "artifacts/player_handout.md", } - env.Config.Pipeline.Publish.Locks = []config.PublishLockRule{ - {Source: "narratio.artifact.session_recap", Reason: "reviewed"}, - } + seedPublishRemoteLock(t, env.ObjectStore.(*storage.FakeBackend), m.S3SessionPrefix, "narratio.artifact.session_recap", "reviewed") result, err := publishStage{}.Run(context.Background(), env, m) if err != nil { t.Fatalf("Run() error = %v", err) } - if result.Metadata["locked_output_count"] != 0 { - t.Fatalf("locked_output_count = %#v, want 0", result.Metadata["locked_output_count"]) - } - skipped := result.Metadata["skipped_unselected_outputs"].([]map[string]any) - if len(skipped) != 1 || skipped[0]["source"] != "narratio.artifact.session_recap" { - t.Fatalf("skipped_unselected_outputs = %#v, want unselected recap", skipped) + outcome := publishOutcomeForSource(t, result, "narratio.artifact.session_recap") + if outcome["status"] != string(publishOutputUnselected) { + t.Fatalf("publish output outcome = %#v, want unselected recap", outcome) } } -func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) { +func TestPublishCarriesLockedRequiredOutputAndCommits(t *testing.T) { env, m, _ := publishFixture(t) - env.Config.Pipeline.Publish.Locks = []config.PublishLockRule{ - {Source: "narratio.transcript.final_trimmed", Reason: "human reviewed"}, - } fake := env.ObjectStore.(*storage.FakeBackend) + oldRunID := "20260515T010203Z-a1b2c3d4" + priorData := []byte("previously reviewed\n") + seedCommittedCurrentState(t, fake, m, oldRunID, committedPublishOutput{ + Source: "narratio.transcript.final_trimmed", Destination: "transcripts/final.trimmed.json", Data: priorData, + }) + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "human reviewed") + if err := os.Remove(filepath.Join(publishSessionPaths(env, m).TranscriptsDir, "final.trimmed.json")); err != nil { + t.Fatalf("remove locked local source: %v", err) + } result, err := publishStage{}.Run(context.Background(), env, m) if err != nil { @@ -765,8 +763,9 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) { 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) + t.Fatalf("locked published output key %q should not be uploaded for the new run", trimmedKey) } + priorTrimmedKey := publishOutputRemoteKey(artifacts.S3RunPrefix(m.S3SessionPrefix, oldRunID), "transcripts/final.trimmed.json") 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) @@ -784,21 +783,12 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) { if result.Metadata["published_files_uploaded"] != 1 { t.Fatalf("metadata published_files_uploaded = %#v, want 1", result.Metadata["published_files_uploaded"]) } - if result.Metadata["locked_output_count"] != 1 { - t.Fatalf("metadata locked_output_count = %#v, want 1", result.Metadata["locked_output_count"]) - } - locked := result.Metadata["locked_outputs"].([]map[string]any) - if len(locked) != 1 { - t.Fatalf("locked_outputs = %#v, want one item", locked) - } - if locked[0]["source"] != "narratio.transcript.final_trimmed" || - locked[0]["dest"] != "transcripts/final.trimmed.json" || - locked[0]["remote_key"] != trimmedKey || - locked[0]["reason"] != "human reviewed" || - locked[0]["required"] != true || - locked[0]["local_path"] == "" || - locked[0]["provenance"] == "" { - t.Fatalf("locked published output metadata = %#v", locked[0]) + outcome := publishOutcomeForSource(t, result, "narratio.transcript.final_trimmed") + if outcome["status"] != string(publishOutputCarriedLocked) || + outcome["destination"] != "transcripts/final.trimmed.json" || + outcome["declared_key"] != priorTrimmedKey || + outcome["origin_run"] != oldRunID { + t.Fatalf("carried published output metadata = %#v", outcome) } currentManifestKey := artifacts.S3RunSessionManifestKey(m.S3SessionPrefix, m.RunID) @@ -809,12 +799,29 @@ func TestPublishSkipsLockedRequiredOutputAndCommits(t *testing.T) { stages := current["stages"].(map[string]any) publishRecord := stages["publish"].(map[string]any) meta := publishRecord["metadata"].(map[string]any) - if meta["locked_output_count"] != float64(1) { - t.Fatalf("current manifest locked_output_count = %#v, want 1", meta["locked_output_count"]) + items := meta["publish_output_outcomes"].([]any) + foundCarry := false + for _, raw := range items { + item := raw.(map[string]any) + if item["source"] == "narratio.transcript.final_trimmed" && item["status"] == string(publishOutputCarriedLocked) { + foundCarry = item["declared_key"] == priorTrimmedKey + } } - items := meta["locked_outputs"].([]any) - if len(items) != 1 { - t.Fatalf("current manifest locked_outputs = %#v, want one item", items) + if !foundCarry { + t.Fatalf("current manifest publish_output_outcomes = %#v, want carried trimmed output", items) + } + state, err := artifacts.LoadCurrentState(context.Background(), fake, m.S3SessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true}) + if err != nil { + t.Fatalf("LoadCurrentState() error = %v", err) + } + carried := remoteArtifactForSource(t, state.Commit, "narratio.transcript.final_trimmed") + if carried.DestinationKey != priorTrimmedKey || carried.SHA256 != publishSHA256(priorData) { + t.Fatalf("carried declaration = %#v", carried) + } + for _, download := range fake.Downloads { + if download.Key == priorTrimmedKey { + t.Fatalf("carried object %q was downloaded", priorTrimmedKey) + } } } @@ -823,9 +830,7 @@ func TestPublishLockedRequiredMissingOutputSucceeds(t *testing.T) { env.Config.Pipeline.Publish.Outputs = []config.PublishOutputRule{ {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)}, } - env.Config.Pipeline.Publish.Locks = []config.PublishLockRule{ - {Source: "narratio.transcript.base", Reason: "manual merge is locked"}, - } + seedPublishRemoteLock(t, env.ObjectStore.(*storage.FakeBackend), m.S3SessionPrefix, "narratio.transcript.base", "manual merge is locked") result, err := publishStage{}.Run(context.Background(), env, m) if err != nil { @@ -840,32 +845,26 @@ func TestPublishLockedRequiredMissingOutputSucceeds(t *testing.T) { 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 { - t.Fatalf("locked_outputs = %#v, want one item", locked) - } - if locked[0]["local_path"] != "" || locked[0]["provenance"] != "" { - t.Fatalf("locked missing published output metadata = %#v, want empty local path/provenance", locked[0]) + outcome := publishOutcomeForSource(t, result, "narratio.transcript.base") + if outcome["status"] != string(publishOutputLockedUnpublished) { + t.Fatalf("locked missing published output metadata = %#v", outcome) } } -func TestPublishLockDoesNotOverwriteExistingOutput(t *testing.T) { +func TestPublishLockCarriesPriorOutputWithoutUploadingIt(t *testing.T) { env, m, _ := publishFixture(t) - env.Config.Pipeline.Publish.Locks = []config.PublishLockRule{ - {Source: "narratio.transcript.final_trimmed", Reason: "already published"}, - } fake := env.ObjectStore.(*storage.FakeBackend) + oldRunID := "20260515T010203Z-a1b2c3d4" + seedCommittedCurrentState(t, fake, m, oldRunID, committedPublishOutput{ + Source: "narratio.transcript.final_trimmed", Destination: "transcripts/final.trimmed.json", Data: []byte("previously published\n"), + }) + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "already published") 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 { t.Fatalf("Run() error = %v", err) } - got := string(fake.Objects[trimmedKey].Data) - if got != "previously published\n" { - t.Fatalf("locked published output object contents = %q, want existing object preserved", got) - } for _, upload := range fake.Uploads { if upload.Key == trimmedKey { t.Fatalf("locked published output key %q was uploaded", trimmedKey) @@ -873,6 +872,71 @@ func TestPublishLockDoesNotOverwriteExistingOutput(t *testing.T) { } } +func TestPublishContinuesToCarryWhenInitialLockIsRemoved(t *testing.T) { + env, m, _ := publishFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + oldRunID := "20260515T010203Z-a1b2c3d4" + seedCommittedCurrentState(t, fake, m, oldRunID, committedPublishOutput{ + Source: "narratio.transcript.final_trimmed", Destination: "transcripts/final.trimmed.json", Data: []byte("reviewed\n"), + }) + lockKey := seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "reviewed") + removed := false + fake.UploadHook = func(storage.FakeUploadCall) error { + if !removed { + removed = true + return fake.Delete(context.Background(), lockKey) + } + return nil + } + + result, err := (publishStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + outcome := publishOutcomeForSource(t, result, "narratio.transcript.final_trimmed") + if outcome["status"] != string(publishOutputCarriedLocked) || outcome["origin_run"] != oldRunID { + t.Fatalf("publish output outcome = %#v, want stable carry decision", outcome) + } +} + +func TestPublishRejectsInvalidPriorCurrentStateForLockedSource(t *testing.T) { + env, m, _ := publishFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "reviewed") + fake.SeedObject(storage.FakeObject{ + Key: artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix), + Data: []byte(`{"format_version":1,"invalid":true}`), + }) + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "load prior current commit") { + t.Fatalf("Run() error = %v, want invalid prior current state failure", err) + } + if len(fake.Uploads) != 0 { + t.Fatalf("uploads = %#v, want failure before uploads", fake.Uploads) + } +} + +func TestPublishRejectsChangedCarriedObjectMetadata(t *testing.T) { + env, m, _ := publishFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + oldRunID := "20260515T010203Z-a1b2c3d4" + seedCommittedCurrentState(t, fake, m, oldRunID, committedPublishOutput{ + Source: "narratio.transcript.final_trimmed", Destination: "transcripts/final.trimmed.json", Data: []byte("reviewed\n"), + }) + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "reviewed") + priorKey := publishOutputRemoteKey(artifacts.S3RunPrefix(m.S3SessionPrefix, oldRunID), "transcripts/final.trimmed.json") + fake.SeedObject(storage.FakeObject{Key: priorKey, Data: []byte("changed!\n")}) + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "verify carried output") { + t.Fatalf("Run() error = %v, want carried metadata verification failure", err) + } + if len(fake.Uploads) != 0 { + t.Fatalf("uploads = %#v, want verification before uploads", fake.Uploads) + } +} + func TestPublishFailsWhenRequiredOutputMissing(t *testing.T) { env, m, _ := publishFixture(t) env.Config.Pipeline.Publish.Outputs = []config.PublishOutputRule{ @@ -984,8 +1048,12 @@ func TestPublishDoesNotSelectCommitWhenLockAppearsAtCommitPoint(t *testing.T) { seedCommittedCurrentState(t, fake, m, oldRunID) pointerKey := artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix) priorPointer := append([]byte(nil), fake.Objects[pointerKey].Data...) - env.RevalidatePublishLocks = func(context.Context) ([]config.PublishLockRule, error) { - return []config.PublishLockRule{{Source: "narratio.transcript.final_trimmed", Reason: "manual review"}}, nil + commitKey := artifacts.S3RunCommitKey(m.S3SessionPrefix, m.RunID) + fake.UploadHook = func(call storage.FakeUploadCall) error { + if call.Key == commitKey { + seedPublishRemoteLock(t, fake, m.S3SessionPrefix, "narratio.transcript.final_trimmed", "manual review") + } + return nil } _, err := (publishStage{}).Run(context.Background(), env, m) @@ -995,6 +1063,9 @@ func TestPublishDoesNotSelectCommitWhenLockAppearsAtCommitPoint(t *testing.T) { if got := fake.Objects[pointerKey].Data; !reflect.DeepEqual(got, priorPointer) { t.Fatalf("current pointer changed after lock appeared: %q", got) } + if _, exists := fake.Objects[commitKey]; !exists { + t.Fatalf("immutable attempt commit %q missing after final lock appeared", commitKey) + } state, err := artifacts.LoadCurrentState(context.Background(), fake, m.S3SessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true}) if err != nil || state.RunID != oldRunID { t.Fatalf("current state after lock loss = %#v, %v; want old run %q", state, err, oldRunID) @@ -1026,7 +1097,7 @@ func TestPublishRejectsAmbiguousOrCollidingOutputMappings(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") { + if err == nil || !strings.Contains(err.Error(), "resolve publish output rules") { t.Fatalf("Run() error = %v, want invalid mapping", err) } if uploads := env.ObjectStore.(*storage.FakeBackend).Uploads; len(uploads) != 0 { @@ -1232,7 +1303,13 @@ func configurePublishExtractionFixture(t *testing.T, env *Env, m *manifest.Manif return lanePath } -func seedCommittedCurrentState(t *testing.T, fake *storage.FakeBackend, source *manifest.Manifest, runID string) { +type committedPublishOutput struct { + Source string + Destination string + Data []byte +} + +func seedCommittedCurrentState(t *testing.T, fake *storage.FakeBackend, source *manifest.Manifest, runID string, outputs ...committedPublishOutput) { t.Helper() copy := *source copy.RunID = runID @@ -1258,6 +1335,23 @@ func seedCommittedCurrentState(t *testing.T, fake *storage.FakeBackend, source * Generation: manifestObject.ETag, }}, } + priorRunPrefix := artifacts.S3RunPrefix(copy.S3SessionPrefix, runID) + for _, output := range outputs { + key := publishOutputRemoteKey(priorRunPrefix, output.Destination) + fake.SeedObject(storage.FakeObject{Key: key, Data: output.Data}) + object := fake.Objects[key] + commit.Artifacts = append(commit.Artifacts, artifacts.RemoteArtifact{ + Type: artifacts.RemoteArtifactTypePublishedOutput, + Source: output.Source, + DestinationKey: key, + SHA256: publishSHA256(output.Data), + Size: int64(len(output.Data)), + Generation: object.ETag, + }) + } + sort.Slice(commit.Artifacts, func(i, j int) bool { + return commit.Artifacts[i].DestinationKey < commit.Artifacts[j].DestinationKey + }) commitData, err := artifacts.EncodeRemoteCommitManifest(commit) if err != nil { t.Fatalf("encode remote commit: %v", err) @@ -1281,6 +1375,52 @@ func seedCommittedCurrentState(t *testing.T, fake *storage.FakeBackend, source * fake.SeedObject(storage.FakeObject{Key: artifacts.S3CurrentCommitPointerKey(copy.S3SessionPrefix), Data: pointerData}) } +func seedPublishRemoteLock(t *testing.T, fake *storage.FakeBackend, sessionPrefix, source, reason string) string { + t.Helper() + lock, err := artifacts.NewRemoteLock(source, reason, time.Date(2026, 5, 16, 0, 0, 0, 0, time.UTC)) + if err != nil { + t.Fatalf("NewRemoteLock() error = %v", err) + } + data, err := artifacts.EncodeRemoteLock(lock) + if err != nil { + t.Fatalf("EncodeRemoteLock() error = %v", err) + } + key, err := artifacts.RemoteLockKey(sessionPrefix, source) + if err != nil { + t.Fatalf("RemoteLockKey() error = %v", err) + } + fake.SeedObject(storage.FakeObject{Key: key, Data: data}) + return key +} + +func publishOutcomeForSource(t *testing.T, result *StageResult, source string) map[string]any { + t.Helper() + items, ok := result.Metadata["publish_output_outcomes"].([]map[string]any) + if !ok { + t.Fatalf("publish_output_outcomes = %#v", result.Metadata["publish_output_outcomes"]) + } + for _, item := range items { + if item["source"] == source { + return item + } + } + t.Fatalf("publish_output_outcomes = %#v, missing source %q", items, source) + return nil +} + +func remoteArtifactForSource(t *testing.T, commit *artifacts.RemoteCommitManifest, source string) artifacts.RemoteArtifact { + t.Helper() + if commit != nil { + for _, artifact := range commit.Artifacts { + if artifact.Source == source { + return artifact + } + } + } + t.Fatalf("remote commit %#v has no source %q", commit, source) + return artifacts.RemoteArtifact{} +} + func boolPtr(v bool) *bool { p := v return &p diff --git a/internal/stage/publish_transfer.go b/internal/stage/publish_transfer.go index a350da1..ee33c25 100644 --- a/internal/stage/publish_transfer.go +++ b/internal/stage/publish_transfer.go @@ -27,12 +27,15 @@ func buildPublishTransferPlan( runSources []*os.File, previousFiles []publishUploadFile, previousSources []*os.File, - publishOutputs []publishOutput, + outputPlan *publishOutputPlan, ) ([]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)) + if outputPlan == nil { + return nil, fmt.Errorf("publish output plan is required") + } + transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(outputPlan.Outcomes)) add := func(artifact artifacts.RemoteArtifact, source *os.File) error { if source == nil { return fmt.Errorf("source for %q is required", artifact.DestinationKey) @@ -60,7 +63,10 @@ func buildPublishTransferPlan( return nil, err } } - for _, output := range publishOutputs { + for _, output := range outputPlan.Outcomes { + if output.Status != publishOutputPublished { + continue + } source, err := openPublishOutputSource(output.LocalPath) if err != nil { closePublishTransfers(transfers) @@ -69,7 +75,7 @@ func buildPublishTransferPlan( if err := add(artifacts.RemoteArtifact{ Type: artifacts.RemoteArtifactTypePublishedOutput, Source: output.Source, - DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest), + DestinationKey: output.DeclaredKey, }, source); err != nil { _ = source.Close() closePublishTransfers(transfers) @@ -82,6 +88,32 @@ func buildPublishTransferPlan( return transfers, nil } +func verifyCarriedPublishOutputs(ctx context.Context, store storage.ObjectStore, plan *publishOutputPlan) ([]artifacts.RemoteArtifact, error) { + if plan == nil { + return nil, fmt.Errorf("publish output plan is required") + } + carried := make([]artifacts.RemoteArtifact, 0) + for _, outcome := range plan.Outcomes { + if outcome.Status != publishOutputCarriedLocked { + continue + } + declaration := outcome.Declaration + if _, err := verifyPublishedObject(ctx, store, declaration.DestinationKey, declaration.Size, declaration.Generation); err != nil { + return nil, fmt.Errorf("verify carried output %q at %q: %w", outcome.Source, declaration.DestinationKey, err) + } + carried = append(carried, declaration) + } + return carried, nil +} + +func combinePublishDeclarations(uploaded, carried []artifacts.RemoteArtifact) []artifacts.RemoteArtifact { + out := make([]artifacts.RemoteArtifact, 0, len(uploaded)+len(carried)) + out = append(out, uploaded...) + out = append(out, carried...) + sort.Slice(out, func(i, j int) bool { return out[i].DestinationKey < out[j].DestinationKey }) + return out +} + func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error { if transfers == nil { return fmt.Errorf("transfer plan is required") diff --git a/internal/stage/stage.go b/internal/stage/stage.go index 4d41dcc..b453a79 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -35,10 +35,6 @@ type Env struct { Scriptorium scriptorium.Runner ObjectStore storage.ObjectStore Notifier notify.Sender - - // RevalidatePublishLocks returns the effective lock set immediately before a - // publish commit selects a new remote snapshot. - RevalidatePublishLocks func(context.Context) ([]config.PublishLockRule, error) } // Stage is the pipeline unit contract.