diff --git a/docs/cli.md b/docs/cli.md index 364d71a..0b3dd1a 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -281,7 +281,8 @@ execution; the preview does not predict output identities. narratio session validate [...common config flags] ``` -Read-only preflight checks for config validity, required inputs, audio mode, previous-session requirements, publish outputs, and effective locks. +Read-only preflight checks for config validity, required inputs, audio mode, +previous-session requirements, publish outputs, and independent remote locks. ### `session status` @@ -290,7 +291,10 @@ narratio session status [...common config flags] ``` Prints local manifest state and, when storage is available, status for the -pointer-selected remote commit and its declared published outputs. +pointer-selected remote commit and its declared published outputs. Current +publish metadata reports `published`, `carried_locked`, `locked_unpublished`, +`optional_missing`, and `unselected`; carried output lines include the run +derived from their exact declared key. ### `session init` @@ -348,7 +352,8 @@ narratio session artifacts [--remote] [...common config flags] Lists effective built-in, configured Scriptorium, and configured extraction sources; reports planned, available, unavailable, and published state without reading payload bodies; and includes publish rules, lock state, and optional -remote published-state availability. +remote published-state availability. Remote inspection uses exact commit +declarations and displays the five current publish outcomes when recorded. ### `session locks` @@ -360,12 +365,14 @@ narratio session locks remove [...common config flags] Behavior: -- list mode reports the effective merge of static and remote locks; -- add/remove mutate only remote locks; -- static locks from pipeline config cannot be removed by CLI commands. +- list mode validates and reports independent remote locks sorted by source; +- add creates one lock and reports an existing lock unless `--force` is used; +- `--force` performs one generation-matched replacement and reports a + concurrent change for operator retry; and +- remove is idempotent and reports when the source was not locked. See [Operations: Publish Locks](./operations.md#publish-locks) for lock storage -and precedence. +and publication behavior. ## `--artifacts` Selection Rules diff --git a/docs/operations.md b/docs/operations.md index 29d3c98..e491f51 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -345,10 +345,11 @@ outcome changes also stale affected downstream stages, while an identical repeated `notarius_disabled` self-skip does not repeatedly invalidate them. Publish reuse additionally tracks enabled/run-upload behavior, normalized -output rules, static locks, and remote backend/bucket/region/endpoint/root -identity. Credential environment names, local workspace placement, and run IDs -are excluded. Regardless of semantic reuse evidence, executing publish still -revalidates mutable remote locks immediately before commit selection. +output rules, the currently supported configured lock field, and remote +backend/bucket/region/endpoint/root identity. Credential environment names, +local workspace placement, and run IDs are excluded. Regardless of semantic +reuse evidence, executing publish reads independent remote locks directly and +rechecks initially unlocked sources before commit selection. ## Publish Workflow @@ -399,24 +400,27 @@ local `current_pointer_written` metadata. ## Publish Locks -Lock sources: +Each mutable lock is an independent strict JSON object at +`{session_prefix}/locks/{sha256(source)}.lock`. The lowercase digest is computed +from the exact canonical source. Listing validates every object in that +reserved prefix and reports locks sorted by source. -- static locks in `pipeline.publish.locks` -- mutable remote locks in `{session_prefix}/locks.yml` +Operational rules: -Effective lock rules: - -- static and remote locks are merged; -- static locks win on source collisions; -- locked outputs are intentional skips; -- lock add/remove commands mutate only remote lock state through generation-bound - conditional writes. A command retries a bounded number of concurrent - conflicts while its invocation context remains active, so it never replaces a - different lock-document generation; and -- a publish re-reads remote locks immediately before it writes the current - commit pointer. A lock committed before that recheck prevents selecting the - new snapshot, even though its already-uploaded immutable objects may remain - available for a later retry. +- add uses conditional creation and reports an existing lock; +- add with `--force` observes the current generation and attempts one + generation-matched replacement. A concurrent change is reported for an + explicit operator retry; +- remove is idempotent and reports whether a lock was present; +- the initial publish lock snapshot controls output planning. A locked source + carries its exact declaration from the prior current commit when available; + a never-published locked source remains absent even when required; +- carried bytes are not copied. Publish verifies the prior object's exact key, + size, and generation without downloading it; and +- after immutable uploads and the immutable commit, publish re-reads every + initially unlocked participating source. A newly added lock prevents the + current pointer from advancing, although the immutable attempt may remain. + A removed initial lock may conservatively cause one carry. Examples: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 1742350..16634e3 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -289,7 +289,7 @@ locked-source work, and exactly one pointer write. Run ## Stage 8 — Migrate Commands And Inspection -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/app/operator_artifact_rendering.go b/internal/app/operator_artifact_rendering.go index 62d68c7..345dfc1 100644 --- a/internal/app/operator_artifact_rendering.go +++ b/internal/app/operator_artifact_rendering.go @@ -35,8 +35,8 @@ func buildHelperArtifactCatalog(cfg *config.Config, m *manifest.Manifest) (*arti return catalog, nil } -func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.ArtifactCatalog, locks *effectiveLocks, publishedRemoteState map[string]string) { - lockSet := lockSourceSet(locks.All) +func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.ArtifactCatalog, locks []artifacts.RemoteLock, publishedRemoteState map[string]string) { + lockSet := remoteLockSourceSet(locks) fmt.Fprintln(out, "Built-in:") for _, transcript := range artifacts.RuntimeTranscriptArtifacts() { writeArtifactLine(out, transcript.SourceID, lockSet) @@ -71,7 +71,7 @@ func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.Art } } -func writeConfiguredArtifactLine(out io.Writer, source, state, provenance string, lockSet map[string]config.PublishLockRule, origins map[string]config.ArtifactFamilyMemberOrigin) { +func writeConfiguredArtifactLine(out io.Writer, source, state, provenance string, lockSet map[string]struct{}, origins map[string]config.ArtifactFamilyMemberOrigin) { parts := []string{source, "planned", state} if key, ok := artifactpolicy.ParseConfiguredSource(source); ok { if origin, family := origins[key]; family { @@ -87,7 +87,7 @@ func writeConfiguredArtifactLine(out io.Writer, source, state, provenance string fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) } -func writeExtractionArtifactLine(out io.Writer, source, state, provenance string, lockSet map[string]config.PublishLockRule) { +func writeExtractionArtifactLine(out io.Writer, source, state, provenance string, lockSet map[string]struct{}) { parts := []string{source, "planned", state} if strings.TrimSpace(provenance) != "" { parts = append(parts, "provenance="+strings.TrimSpace(provenance)) @@ -98,7 +98,7 @@ func writeExtractionArtifactLine(out io.Writer, source, state, provenance string fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) } -func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.PublishLockRule) { +func writeArtifactLine(out io.Writer, source string, lockSet map[string]struct{}) { parts := []string{source} if _, ok := lockSet[source]; ok { parts = append(parts, "locked") @@ -106,7 +106,7 @@ func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.P fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) } -func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog, lockSet map[string]config.PublishLockRule, remoteState map[string]string) { +func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog, lockSet map[string]struct{}, remoteState map[string]string) { source := strings.TrimSpace(rule.Source) parts := []string{source} if _, ok := lockSet[source]; ok { @@ -127,6 +127,14 @@ func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, cata fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) } +func remoteLockSourceSet(locks []artifacts.RemoteLock) map[string]struct{} { + out := make(map[string]struct{}, len(locks)) + for _, lock := range locks { + out[lock.Source] = struct{}{} + } + return out +} + func unresolvedRemotePublishedOutputAvailability(cfg *config.Config, catalog *artifacts.ArtifactCatalog, state string) map[string]string { out := map[string]string{} for _, rule := range cfg.Pipeline.Publish.Outputs { @@ -151,6 +159,7 @@ func remotePublishedOutputAvailabilityForCurrent( if current == nil || current.Commit == nil { return unresolvedRemotePublishedOutputAvailability(cfg, catalog, "remote=missing") } + outcomes := committedPublishOutputOutcomes(current.Manifest) out := map[string]string{} for _, rule := range cfg.Pipeline.Publish.Outputs { source := strings.TrimSpace(rule.Source) @@ -159,26 +168,31 @@ func remotePublishedOutputAvailabilityForCurrent( out[publishedOutputRemoteStateKey(source, "")] = "remote=error" continue } + outcome := outcomes[source] + state := "" + if outcome != "" { + state = "outcome=" + outcome + " " + } declaration, ok := committedPublishedOutput(current.Commit, source) if !ok { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing" + out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=missing" continue } identity, err := artifacts.ParseImmutableRunObjectKey(current.SessionPrefix, declaration.DestinationKey) if err != nil { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=error" + out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=error" continue } exists, err := store.Exists(ctx, declaration.DestinationKey) if err != nil { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=error" + out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=error" continue } if !exists { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing" + out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=missing" continue } - state := "remote=published" + state += "remote=published" if identity.RunID != current.RunID { state += " origin_run=" + identity.RunID } @@ -187,6 +201,43 @@ func remotePublishedOutputAvailabilityForCurrent( return out } +func committedPublishOutputOutcomes(m *manifest.Manifest) map[string]string { + out := map[string]string{} + if m == nil || m.Stages["publish"] == nil { + return out + } + raw := m.Stages["publish"].Metadata["publish_output_outcomes"] + appendOutcome := func(item map[string]any) { + source, sourceOK := item["source"].(string) + status, statusOK := item["status"].(string) + if sourceOK && statusOK && isPublishOutputOutcome(status) { + out[source] = status + } + } + switch items := raw.(type) { + case []any: + for _, rawItem := range items { + if item, ok := rawItem.(map[string]any); ok { + appendOutcome(item) + } + } + case []map[string]any: + for _, item := range items { + appendOutcome(item) + } + } + return out +} + +func isPublishOutputOutcome(status string) bool { + switch status { + case "published", "carried_locked", "locked_unpublished", "optional_missing", "unselected": + return true + default: + return false + } +} + func committedPublishedOutput(commit *artifacts.RemoteCommitManifest, source string) (artifacts.RemoteArtifact, bool) { if commit == nil { return artifacts.RemoteArtifact{}, false diff --git a/internal/app/operator_helpers.go b/internal/app/operator_helpers.go index 4143083..2424367 100644 --- a/internal/app/operator_helpers.go +++ b/internal/app/operator_helpers.go @@ -94,7 +94,7 @@ func Artifacts(ctx context.Context, args []string, out io.Writer) error { } } -func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, *effectiveLocks, *manifest.Manifest, func(), error) { +func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, []artifacts.RemoteLock, *manifest.Manifest, func(), error) { loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return nil, nil, nil, nil, nil, err @@ -118,9 +118,16 @@ func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore b } else { store, _ = objectStoreIfConfigured(ctx, cfg) } - locks, err := loadEffectiveLocks(ctx, cfg, store) - if err != nil { - return nil, nil, nil, nil, nil, err + var locks []artifacts.RemoteLock + if store != nil { + sessionPrefix, prefixErr := artifacts.ResolvePublishSessionPrefix(cfg, nil) + if prefixErr != nil { + return nil, nil, nil, nil, nil, prefixErr + } + locks, err = artifacts.ListRemoteLocks(ctx, store, sessionPrefix) + if err != nil { + return nil, nil, nil, nil, nil, err + } } paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) m, err := loadLocalManifest(ctx, paths.ManifestPath) diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index 3cf97de..3140c62 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -515,7 +515,7 @@ func TestOperatorCommandsReportConfiguredSpellCatalog(t *testing.T) { } } -func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) { +func TestExecuteLocksAddListAndRemoveUseIndependentObjects(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) fake := &storage.FakeBackend{} @@ -534,13 +534,17 @@ func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) { if code != 0 { t.Fatalf("locks add exit code = %d, want 0; stderr=%q", code, stderr.String()) } - key := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) + key, err := artifacts.RemoteLockKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "narratio.transcript.final_trimmed") + if err != nil { + t.Fatal(err) + } obj, ok := fake.Objects[key] if !ok { t.Fatalf("remote locks key %q not uploaded", key) } - if !strings.Contains(string(obj.Data), "source: narratio.transcript.final_trimmed") || !strings.Contains(string(obj.Data), "reason: manual edit") { - t.Fatalf("lock store data = %q", string(obj.Data)) + lock, err := artifacts.DecodeRemoteLock(obj.Data) + if err != nil || lock.Source != "narratio.transcript.final_trimmed" || lock.Reason != "manual edit" { + t.Fatalf("lock object = %#v, %v", lock, err) } stdout.Reset() @@ -554,7 +558,7 @@ func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) { if code != 0 { t.Fatalf("locks list exit code = %d, want 0; stderr=%q", code, stderr.String()) } - if !strings.Contains(stdout.String(), "- narratio.transcript.final_trimmed origin=remote") { + if !strings.Contains(stdout.String(), "- narratio.transcript.final_trimmed published reason=manual edit created_at=") { t.Fatalf("stdout = %q, want remote lock", stdout.String()) } @@ -569,15 +573,56 @@ func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) { if code != 0 { t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String()) } - store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil, nil) + if _, exists := fake.Objects[key]; exists { + t.Fatalf("lock object %q remains after remove", key) + } + stdout.Reset() + stderr.Reset() + code = Execute([]string{ + "session", "locks", "remove", "2026-05-03", "narratio.transcript.final_trimmed", + "--config", pipelinePath, + "--campaign-file", campaignPath, + "--session", sessionPath, + }, &stdout, &stderr) + if code != 0 || !strings.Contains(stdout.String(), "was not locked") { + t.Fatalf("idempotent remove exit code = %d, stdout=%q stderr=%q", code, stdout.String(), stderr.String()) + } + if storeInitCalls != 4 { + t.Fatalf("object store init calls = %d, want 4", storeInitCalls) + } +} + +func TestExecuteLocksListIsSortedAndRejectsMalformedObjects(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final_trimmed", "second") + seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final", "first") + args := []string{"session", "locks", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath} + + var stdout bytes.Buffer + var stderr bytes.Buffer + if code := Execute(args, &stdout, &stderr); code != 0 { + t.Fatalf("locks list exit code = %d; stderr=%q", code, stderr.String()) + } + first := strings.Index(stdout.String(), "narratio.transcript.final ") + second := strings.Index(stdout.String(), "narratio.transcript.final_trimmed ") + if first < 0 || second < 0 || first >= second { + t.Fatalf("locks list is not source sorted: %q", stdout.String()) + } + + badKey, err := artifacts.RemoteLockKey(sessionPrefix, "narratio.transcript.base") if err != nil { - t.Fatalf("LoadPublishLockStoreBytes() error = %v", err) + t.Fatal(err) } - if len(store.Locks) != 0 { - t.Fatalf("locks after remove = %#v, want empty", store.Locks) - } - if storeInitCalls != 3 { - t.Fatalf("object store init calls = %d, want 3", storeInitCalls) + fake.SeedObject(storage.FakeObject{Key: badKey, Data: []byte(`{"format_version":1}`)}) + stdout.Reset() + stderr.Reset() + if code := Execute(args, &stdout, &stderr); code == 0 { + t.Fatalf("malformed lock list exit code = 0; stdout=%q", stdout.String()) } } @@ -747,9 +792,47 @@ func TestExecuteLocksAddDuplicateRequiresForce(t *testing.T) { if code != 0 { t.Fatalf("forced locks add exit code = %d, want 0; stderr=%q", code, stderr.String()) } - key := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) - if !strings.Contains(string(fake.Objects[key].Data), "reason: second") { - t.Fatalf("lock store data = %q, want updated reason", string(fake.Objects[key].Data)) + key, err := artifacts.RemoteLockKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "narratio.transcript.final_trimmed") + if err != nil { + t.Fatal(err) + } + lock, err := artifacts.DecodeRemoteLock(fake.Objects[key].Data) + if err != nil || lock.Reason != "second" { + t.Fatalf("lock object = %#v, %v; want updated reason", lock, err) + } +} + +func TestExecuteLocksForceReportsConcurrentReplacement(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + key := seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final_trimmed", "first") + replacement, err := artifacts.NewRemoteLock("narratio.transcript.final_trimmed", "concurrent", time.Date(2026, 5, 20, 1, 2, 3, 0, time.UTC)) + if err != nil { + t.Fatal(err) + } + replacementData, err := artifacts.EncodeRemoteLock(replacement) + if err != nil { + t.Fatal(err) + } + fake.UploadHook = func(call storage.FakeUploadCall) error { + if call.Key == key { + fake.SeedObject(storage.FakeObject{Key: key, Data: replacementData}) + } + return nil + } + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{ + "session", "locks", "add", "2026-05-03", "narratio.transcript.final_trimmed", "--force", + "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, + }, &stdout, &stderr) + if code == 0 || !strings.Contains(stderr.String(), "changed concurrently; retry") { + t.Fatalf("forced conflict exit code = %d, stderr=%q", code, stderr.String()) } } @@ -825,7 +908,7 @@ func TestExecuteLocksMutationRejectsSessionIDMismatch(t *testing.T) { } } -func TestExecuteLocksCannotModifyStaticLocks(t *testing.T) { +func TestExecuteLocksDoNotTreatStaticConfigAsAuthority(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) addStaticPublishLockToPipelineConfig(t, pipelinePath, "narratio.transcript.final_trimmed") @@ -841,11 +924,8 @@ func TestExecuteLocksCannotModifyStaticLocks(t *testing.T) { "--campaign-file", campaignPath, "--session", sessionPath, }, &stdout, &stderr) - if code == 0 { - t.Fatal("locks add static lock exit code = 0, want non-zero") - } - if !strings.Contains(stderr.String(), "locked by pipeline config") { - t.Fatalf("stderr = %q, want static lock error", stderr.String()) + if code != 0 { + t.Fatalf("locks add exit code = %d, want 0; stderr=%q", code, stderr.String()) } stdout.Reset() @@ -856,11 +936,8 @@ func TestExecuteLocksCannotModifyStaticLocks(t *testing.T) { "--campaign-file", campaignPath, "--session", sessionPath, }, &stdout, &stderr) - if code == 0 { - t.Fatal("locks remove static lock exit code = 0, want non-zero") - } - if !strings.Contains(stderr.String(), "locked by pipeline config") { - t.Fatalf("stderr = %q, want static lock error", stderr.String()) + if code != 0 { + t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String()) } } @@ -1125,12 +1202,11 @@ func TestExecuteStatusReportsRemoteArtifactCatalog(t *testing.T) { `) fake := &storage.FakeBackend{} sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") - lockKey := artifacts.S3SessionLocksKey(sessionPrefix) seedOperatorCommittedOutputs(t, fake, map[string]string{ "narratio.transcript.final_trimmed": "transcripts/final.trimmed.json", "narratio.transcript.final": "transcripts/full.json", }) - fake.SeedObject(storage.FakeObject{Key: lockKey, Data: []byte("locks:\n - source: narratio.transcript.final_trimmed\n reason: remote review\n")}) + seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final_trimmed", "remote review") var storeInitCalls int restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) @@ -1484,6 +1560,24 @@ func replaceInFileOrFatal(t *testing.T, path, old, new string) { } } +func seedOperatorRemoteLock(t *testing.T, fake *storage.FakeBackend, sessionPrefix, source, reason string) string { + t.Helper() + lock, err := artifacts.NewRemoteLock(source, reason, time.Date(2026, 5, 19, 1, 2, 3, 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 writeValidPublishRunConfigFiles(t *testing.T, workspaceRoot string) (string, string, string) { t.Helper() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) diff --git a/internal/app/operator_inspection.go b/internal/app/operator_inspection.go index 96edc7b..3d038f9 100644 --- a/internal/app/operator_inspection.go +++ b/internal/app/operator_inspection.go @@ -46,8 +46,8 @@ type remoteCurrentStateCheck struct { Err error } -type effectiveLocksCheck struct { - Locks *effectiveLocks +type remoteLocksCheck struct { + Locks []artifacts.RemoteLock Err error } @@ -194,12 +194,19 @@ func inspectRemoteCurrentState(ctx context.Context, cfg *config.Config, store st return remoteCurrentStateCheck{State: current} } -func inspectEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) effectiveLocksCheck { - locks, err := loadEffectiveLocks(ctx, cfg, store) - if err != nil { - return effectiveLocksCheck{Err: err} +func inspectRemoteLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) remoteLocksCheck { + if store == nil { + return remoteLocksCheck{} } - return effectiveLocksCheck{Locks: locks} + sessionPrefix, err := artifacts.ResolvePublishSessionPrefix(cfg, nil) + if err != nil { + return remoteLocksCheck{Err: err} + } + locks, err := artifacts.ListRemoteLocks(ctx, store, sessionPrefix) + if err != nil { + return remoteLocksCheck{Err: err} + } + return remoteLocksCheck{Locks: locks} } func resolveLocalInspectionAudioPaths(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) { diff --git a/internal/app/operator_locks.go b/internal/app/operator_locks.go index f4426d5..38b091e 100644 --- a/internal/app/operator_locks.go +++ b/internal/app/operator_locks.go @@ -2,12 +2,15 @@ package app import ( "context" + "errors" "flag" "fmt" "io" - "sort" "strings" + "time" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" ) @@ -26,7 +29,7 @@ func Locks(ctx context.Context, args []string, out io.Writer) error { return LocksList(ctx, args, out) } -// LocksList lists effective publish locks. +// LocksList lists independent publish locks. func LocksList(ctx context.Context, args []string, out io.Writer) error { fs := flag.NewFlagSet("locks", flag.ContinueOnError) fs.SetOutput(io.Discard) @@ -38,11 +41,15 @@ func LocksList(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks: session_id is required") } - cfg, _, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) + cfg, store, sessionPrefix, cleanup, err := loadLockCommandContext(ctx, flags) if err != nil { return fmt.Errorf("locks: %w", err) } defer cleanup() + locks, err := artifacts.ListRemoteLocks(ctx, store, sessionPrefix) + if err != nil { + return fmt.Errorf("locks: %w", err) + } writeLocks(out, cfg, locks) return nil } @@ -64,31 +71,28 @@ func LocksAdd(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks add: session_id is required") } - cfg, store, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) + _, store, sessionPrefix, cleanup, err := loadLockCommandContext(ctx, flags) if err != nil { return fmt.Errorf("locks add: %w", err) } defer cleanup() - if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks add"); err != nil { + lock, err := artifacts.NewRemoteLock(source, strings.TrimSpace(reason), nowUTC()) + if err != nil { return fmt.Errorf("locks add: %w", err) } - if _, ok := lockSourceSet(locks.Static)[source]; ok { - return fmt.Errorf("locks add: source %q is locked by pipeline config and cannot be modified remotely", source) - } - if err := mutateRemoteLockStore(ctx, cfg, store, func(lockStore *config.PublishLockStore) error { - remoteSet := lockSourceSet(lockStore.Locks) - if _, exists := remoteSet[source]; exists && !force { - return fmt.Errorf("remote lock for %q already exists; pass --force to update", source) - } - remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)} - lockStore.Locks = lockMapValues(remoteSet) - normalized, err := config.ValidatePublishLockRules(lockStore.Locks, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks") + if !force { + current, err := artifacts.LookupRemoteLock(ctx, store, sessionPrefix, source) if err != nil { - return err + return fmt.Errorf("locks add: %w", err) + } + if current.Present { + return fmt.Errorf("locks add: remote lock for %q already exists; pass --force to update", source) + } + } + if err := artifacts.AddRemoteLock(ctx, store, sessionPrefix, lock, force); err != nil { + if errors.Is(err, storage.ErrConditionNotMet) { + return fmt.Errorf("locks add: remote lock for %q changed concurrently; retry: %w", source, err) } - lockStore.Locks = normalized - return nil - }); err != nil { return fmt.Errorf("locks add: %w", err) } _, err = fmt.Fprintf(out, "narratio session locks add: locked %s\n", source) @@ -108,34 +112,55 @@ func LocksRemove(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks remove: session_id is required") } - cfg, store, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) + _, store, sessionPrefix, cleanup, err := loadLockCommandContext(ctx, flags) if err != nil { return fmt.Errorf("locks remove: %w", err) } defer cleanup() - if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks remove"); err != nil { + current, err := artifacts.LookupRemoteLock(ctx, store, sessionPrefix, source) + if err != nil { return fmt.Errorf("locks remove: %w", err) } - if err := mutateRemoteLockStore(ctx, cfg, store, func(lockStore *config.PublishLockStore) error { - remoteSet := lockSourceSet(lockStore.Locks) - if _, ok := remoteSet[source]; !ok { - if _, static := lockSourceSet(locks.Static)[source]; static { - return fmt.Errorf("source %q is locked by pipeline config and cannot be unlocked remotely", source) - } - return fmt.Errorf("remote lock for %q does not exist", source) - } - delete(remoteSet, source) - lockStore.Locks = lockMapValues(remoteSet) - return nil - }); err != nil { + if err := artifacts.RemoveRemoteLock(ctx, store, sessionPrefix, source); err != nil { return fmt.Errorf("locks remove: %w", err) } + if !current.Present { + _, err = fmt.Fprintf(out, "narratio session locks remove: %s was not locked\n", source) + return err + } _, err = fmt.Fprintf(out, "narratio session locks remove: unlocked %s\n", source) return err } -func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { - if locks == nil || len(locks.All) == 0 { +func loadLockCommandContext(ctx context.Context, flags commonConfigFlags) (*config.Config, storage.ObjectStore, string, func(), error) { + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + return nil, nil, "", nil, err + } + release := true + defer func() { + if release { + _ = loaded.Close() + } + }() + cfg := loaded.Config + if err := config.Validate(cfg); err != nil { + return nil, nil, "", nil, err + } + store, err := newCommandObjectStore(ctx, cfg, nil) + if err != nil { + return nil, nil, "", nil, err + } + sessionPrefix, err := artifacts.ResolvePublishSessionPrefix(cfg, nil) + if err != nil { + return nil, nil, "", nil, err + } + release = false + return cfg, store, sessionPrefix, func() { _ = loaded.Close() }, nil +} + +func writeLocks(out io.Writer, cfg *config.Config, locks []artifacts.RemoteLock) { + if len(locks) == 0 { fmt.Fprintln(out, "Publish locks: none") return } @@ -146,12 +171,7 @@ func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { published[strings.TrimSpace(rule.Source)] = rule } } - staticSet := lockSourceSet(locks.Static) - for _, lock := range locks.All { - origin := "remote" - if _, ok := staticSet[lock.Source]; ok { - origin = "pipeline" - } + for _, lock := range locks { promo := "not-published" if _, ok := published[lock.Source]; ok { promo = "published" @@ -160,22 +180,6 @@ func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { if reason == "" { reason = "(no reason)" } - fmt.Fprintf(out, "- %s origin=%s %s reason=%s\n", lock.Source, origin, promo, reason) + fmt.Fprintf(out, "- %s %s reason=%s created_at=%s\n", lock.Source, promo, reason, lock.CreatedAt.UTC().Format(time.RFC3339Nano)) } } - -func lockMapValues(in map[string]config.PublishLockRule) []config.PublishLockRule { - keys := make([]string, 0, len(in)) - for key := range in { - keys = append(keys, key) - } - sort.Strings(keys) - out := make([]config.PublishLockRule, 0, len(keys)) - for _, key := range keys { - item := in[key] - item.Source = key - item.Reason = strings.TrimSpace(item.Reason) - out = append(out, item) - } - return out -} diff --git a/internal/app/operator_session_validate.go b/internal/app/operator_session_validate.go index 64e1fd0..5419aad 100644 --- a/internal/app/operator_session_validate.go +++ b/internal/app/operator_session_validate.go @@ -76,13 +76,13 @@ func SessionValidate(ctx context.Context, args []string, out io.Writer) error { } } - locks := inspectEffectiveLocks(ctx, cfg, store) + locks := inspectRemoteLocks(ctx, cfg, store) if locks.Err != nil { findings = append(findings, errorFinding("locks", locks.Err.Error())) - } else if len(locks.Locks.All) == 0 { - findings = append(findings, okFinding("locks", "no effective publish locks")) + } else if len(locks.Locks) == 0 { + findings = append(findings, okFinding("locks", "no publish locks")) } else { - for _, lock := range locks.Locks.All { + for _, lock := range locks.Locks { findings = append(findings, warnFinding("locks", fmt.Sprintf("%s locked: %s", lock.Source, strings.TrimSpace(lock.Reason)))) } } diff --git a/internal/app/operator_status.go b/internal/app/operator_status.go index 999e094..0300b6f 100644 --- a/internal/app/operator_status.go +++ b/internal/app/operator_status.go @@ -84,7 +84,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error { artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg), effective), )) - lockChecks := inspectEffectiveLocks(ctx, cfg, store) + lockChecks := inspectRemoteLocks(ctx, cfg, store) locks := lockChecks.Locks lockErr := lockChecks.Err if catalog, catalogErr := buildHelperArtifactCatalog(cfg, localManifest); catalogErr != nil { @@ -92,10 +92,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error { } else if storeErr == nil { catalogLocks := locks if lockErr != nil { - catalogLocks = &effectiveLocks{ - Static: staticPublishLocks(cfg), - All: staticPublishLocks(cfg), - } + catalogLocks = nil } publishedRemoteState := map[string]string{} if store != nil { diff --git a/internal/app/remote_locks.go b/internal/app/remote_locks.go index e640100..cdef96a 100644 --- a/internal/app/remote_locks.go +++ b/internal/app/remote_locks.go @@ -7,6 +7,7 @@ import ( "fmt" "os" "path/filepath" + "sort" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" @@ -158,6 +159,22 @@ func lockSourceSet(locks []config.PublishLockRule) map[string]config.PublishLock return out } +func lockMapValues(in map[string]config.PublishLockRule) []config.PublishLockRule { + keys := make([]string, 0, len(in)) + for key := range in { + keys = append(keys, key) + } + sort.Strings(keys) + out := make([]config.PublishLockRule, 0, len(keys)) + for _, key := range keys { + item := in[key] + item.Source = key + item.Reason = strings.TrimSpace(item.Reason) + out = append(out, item) + } + return out +} + func writeLocalFile(path string, data []byte, force bool) error { cleaned := filepath.Clean(strings.TrimSpace(path)) if cleaned == "" || cleaned == "." { diff --git a/internal/app/restore_commit_test.go b/internal/app/restore_commit_test.go index 713af3a..05efd27 100644 --- a/internal/app/restore_commit_test.go +++ b/internal/app/restore_commit_test.go @@ -149,10 +149,16 @@ func TestCommittedStatusReportsCarriedOutputOriginAndChecksExactKey(t *testing.T fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")}) current.Commit.Artifacts[1].Source = "narratio.transcript.final_trimmed" current.Commit.Artifacts[1].DestinationKey = oldKey + current.Manifest.Stages["publish"] = &manifest.StageRecord{ + Name: "publish", Status: manifest.StatusSucceeded, + Metadata: map[string]any{"publish_output_outcomes": []any{map[string]any{ + "source": "narratio.transcript.final_trimmed", "status": "carried_locked", + }}}, + } availability := remotePublishedOutputAvailabilityForCurrent(context.Background(), cfg, fake, nil, current) key := publishedOutputRemoteStateKey("narratio.transcript.final_trimmed", "transcripts/full.json") - want := "remote=published origin_run=" + originRun + want := "outcome=carried_locked remote=published origin_run=" + originRun if availability[key] != want { t.Fatalf("availability = %#v, want %q", availability, want) } @@ -161,6 +167,24 @@ func TestCommittedStatusReportsCarriedOutputOriginAndChecksExactKey(t *testing.T } } +func TestCommittedPublishOutputOutcomesRecognizesAllStates(t *testing.T) { + statuses := []string{"published", "carried_locked", "locked_unpublished", "optional_missing", "unselected"} + items := make([]any, 0, len(statuses)) + for _, status := range statuses { + items = append(items, map[string]any{"source": "source." + status, "status": status}) + } + m := &manifest.Manifest{Stages: map[string]*manifest.StageRecord{ + "publish": {Metadata: map[string]any{"publish_output_outcomes": items}}, + }} + + got := committedPublishOutputOutcomes(m) + for _, status := range statuses { + if got["source."+status] != status { + t.Fatalf("outcomes = %#v, missing %q", got, status) + } + } +} + func TestCommittedRestoreKeepsSelectedSnapshotWhenPointerChanges(t *testing.T) { cfg := restorePlanConfig(t) fake := &storage.FakeBackend{} diff --git a/internal/app/session_oriented_cli_test.go b/internal/app/session_oriented_cli_test.go index f07c685..e3c6843 100644 --- a/internal/app/session_oriented_cli_test.go +++ b/internal/app/session_oriented_cli_test.go @@ -324,9 +324,13 @@ func TestExecuteSessionLocksMutationAcceptsPositionalSessionID(t *testing.T) { if code != 0 { t.Fatalf("locks add exit code = %d, want 0; stderr=%q", code, stderr.String()) } - key := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) - if !strings.Contains(string(fake.Objects[key].Data), "reason: review") { - t.Fatalf("lock store data = %q, want reason", string(fake.Objects[key].Data)) + key, err := artifacts.RemoteLockKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "narratio.transcript.final_trimmed") + if err != nil { + t.Fatal(err) + } + lock, err := artifacts.DecodeRemoteLock(fake.Objects[key].Data) + if err != nil || lock.Reason != "review" { + t.Fatalf("lock object = %#v, %v; want reason", lock, err) } stdout.Reset() @@ -340,12 +344,8 @@ func TestExecuteSessionLocksMutationAcceptsPositionalSessionID(t *testing.T) { if code != 0 { t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String()) } - store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil, nil) - if err != nil { - t.Fatalf("LoadPublishLockStoreBytes() error = %v", err) - } - if len(store.Locks) != 0 { - t.Fatalf("locks after remove = %#v, want empty", store.Locks) + if _, exists := fake.Objects[key]; exists { + t.Fatalf("lock object %q remains after remove", key) } }