Migrate lock commands to independent objects

This commit is contained in:
2026-09-05 13:11:11 +00:00
parent 4bc33773fa
commit 610eaefb27
13 changed files with 368 additions and 156 deletions

View File

@@ -281,7 +281,8 @@ execution; the preview does not predict output identities.
narratio session validate <session_id> [...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 <session_id> [...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 <session_id> [--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 <session_id> <source> [...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

View File

@@ -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:

View File

@@ -289,7 +289,7 @@ locked-source work, and exactly one pointer write. Run
## Stage 8 — Migrate Commands And Inspection
**Status:** Pending
**Status:** Completed
### Goal

View File

@@ -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

View File

@@ -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)

View File

@@ -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)

View File

@@ -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) {

View File

@@ -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
}

View File

@@ -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))))
}
}

View File

@@ -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 {

View File

@@ -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 == "." {

View File

@@ -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{}

View File

@@ -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)
}
}