diff --git a/docs/config.md b/docs/config.md index 2208e81..1d52119 100644 --- a/docs/config.md +++ b/docs/config.md @@ -255,9 +255,6 @@ publish: - source: narratio.artifact.session_recap dest: artifacts/session_recap.md required: true - locks: - - source: narratio.artifact.session_recap - reason: manual post-publish edits ``` Rules: @@ -268,7 +265,8 @@ Rules: a rule names that source; the Notarius index and complete bundle are not publish sources. - `outputs[].required` defaults to `true`. -- static locks (`pipeline.publish.locks`) merge with remote locks (`{session_prefix}/locks.yml`), with static locks taking precedence on duplicates. +- publish locks are independent remote session objects managed with + `narratio session locks`; they are not pipeline configuration fields. ## Full Schema @@ -302,9 +300,6 @@ Rules: | `pipeline.publish.outputs[].source` | string | Yes (per rule) | must reference built-in or configured artifact source | | `pipeline.publish.outputs[].dest` | string | Conditional | derived if omitted and source supports derivation | | `pipeline.publish.outputs[].required` | bool | No | `true` | -| `pipeline.publish.locks[]` | list | No | empty | -| `pipeline.publish.locks[].source` | string | Yes (per lock) | must reference supported publish source | -| `pipeline.publish.locks[].reason` | string | No | empty | | `pipeline.whisperx.transcribe_url` | string | Yes | absolute `http` or `https` URL | | `pipeline.whisperx.language` | string | No | `en` | | `pipeline.whisperx.timeout` | duration | No | `30m` | diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index c3a2d75..77c6e05 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -115,9 +115,9 @@ Includes counts/lists for: ## Resume Semantics The versioned semantic fingerprint covers enabled behavior, run-upload policy, -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 +normalized source/destination/required output rules, and the remote backend, +bucket, region, endpoint, and root-prefix identity. Rule 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 diff --git a/docs/operations.md b/docs/operations.md index e491f51..4551199 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -345,11 +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, 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. +output rules, 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 diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 16634e3..1c6cf7e 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -318,7 +318,7 @@ accurate lock and inspection commands. ## Stage 9 — Remove Retired Lock Authorities -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/adapters/storage/fake_test.go b/internal/adapters/storage/fake_test.go index 25394f2..d8cc6b2 100644 --- a/internal/adapters/storage/fake_test.go +++ b/internal/adapters/storage/fake_test.go @@ -105,16 +105,16 @@ func TestFakeBackendDeleteHonorsContextAndInjectedError(t *testing.T) { func TestFakeBackendConditionalUploadRejectsStaleGeneration(t *testing.T) { fake := &FakeBackend{} - fake.SeedObject(FakeObject{Key: "locks.yml", Data: []byte("old")}) - old := fake.Objects["locks.yml"].ETag - if _, err := fake.UploadConditional(context.Background(), strings.NewReader("new"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: old}); err != nil { + fake.SeedObject(FakeObject{Key: "control.json", Data: []byte("old")}) + old := fake.Objects["control.json"].ETag + if _, err := fake.UploadConditional(context.Background(), strings.NewReader("new"), "control.json", UploadOptions{}, WriteCondition{MatchETag: old}); err != nil { t.Fatalf("UploadConditional() error = %v", err) } - if _, err := fake.UploadConditional(context.Background(), strings.NewReader("lost"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: old}); !errors.Is(err, ErrConditionNotMet) { + if _, err := fake.UploadConditional(context.Background(), strings.NewReader("lost"), "control.json", UploadOptions{}, WriteCondition{MatchETag: old}); !errors.Is(err, ErrConditionNotMet) { t.Fatalf("UploadConditional() error = %v, want ErrConditionNotMet", err) } - if got := string(fake.Objects["locks.yml"].Data); got != "new" { - t.Fatalf("locks object = %q, want successful replacement preserved", got) + if got := string(fake.Objects["control.json"].Data); got != "new" { + t.Fatalf("control object = %q, want successful replacement preserved", got) } } diff --git a/internal/adapters/storage/s3_backend_test.go b/internal/adapters/storage/s3_backend_test.go index c748f4b..76b511d 100644 --- a/internal/adapters/storage/s3_backend_test.go +++ b/internal/adapters/storage/s3_backend_test.go @@ -187,14 +187,14 @@ func TestS3BackendDownloadCreatesParentDirectory(t *testing.T) { func TestS3BackendReadReturnsOpenedObjectMetadata(t *testing.T) { lastModified := time.Date(2026, 8, 11, 1, 2, 3, 0, time.UTC) client := &fakeS3API{ - getBody: io.NopCloser(strings.NewReader("locks")), + getBody: io.NopCloser(strings.NewReader("state")), getSize: int64Ptr(5), getETag: strPtr(`"generation"`), getLastModified: &lastModified, } backend := &S3Backend{bucket: "bucket-1", client: client} - info, body, err := backend.Read(context.Background(), `sessions\locks.yml`) + info, body, err := backend.Read(context.Background(), `sessions\control.json`) if err != nil { t.Fatalf("Read() error = %v", err) } @@ -203,10 +203,10 @@ func TestS3BackendReadReturnsOpenedObjectMetadata(t *testing.T) { if readErr != nil || closeErr != nil { t.Fatalf("read body error=%v close error=%v", readErr, closeErr) } - if info.Key != "sessions/locks.yml" || info.Size != 5 || info.ETag != "generation" || info.LastModified == nil || !info.LastModified.Equal(lastModified) { + if info.Key != "sessions/control.json" || info.Size != 5 || info.ETag != "generation" || info.LastModified == nil || !info.LastModified.Equal(lastModified) { t.Fatalf("Read() info = %#v, want opened object metadata", info) } - if string(data) != "locks" || client.lastGet == nil || *client.lastGet.Key != "sessions/locks.yml" { + if string(data) != "state" || client.lastGet == nil || *client.lastGet.Key != "sessions/control.json" { t.Fatalf("Read() data=%q request=%#v", data, client.lastGet) } } @@ -248,14 +248,14 @@ func TestS3BackendUploadAndExists(t *testing.T) { func TestS3BackendConditionalUploadUsesProviderPrecondition(t *testing.T) { client := &fakeS3API{putOut: &s3.PutObjectOutput{ETag: strPtr(`"etag123"`)}} backend := &S3Backend{bucket: "bucket-1", client: client} - if _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: "before"}); err != nil { + if _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "control.json", UploadOptions{}, WriteCondition{MatchETag: "before"}); err != nil { t.Fatalf("UploadConditional() error = %v", err) } if client.lastPut == nil || client.lastPut.IfMatch == nil || *client.lastPut.IfMatch != "before" || client.lastPut.IfNoneMatch != nil { t.Fatalf("PutObject conditional input = %#v", client.lastPut) } client.putErr = &smithy.GenericAPIError{Code: "PreconditionFailed", Message: "changed"} - _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "locks.yml", UploadOptions{}, WriteCondition{RequireAbsent: true}) + _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "control.json", UploadOptions{}, WriteCondition{RequireAbsent: true}) if !errors.Is(err, ErrConditionNotMet) { t.Fatalf("UploadConditional() error = %v, want ErrConditionNotMet", err) } diff --git a/internal/app/operator_helpers.go b/internal/app/operator_helpers.go index 2424367..3f2bcd0 100644 --- a/internal/app/operator_helpers.go +++ b/internal/app/operator_helpers.go @@ -5,11 +5,14 @@ import ( "flag" "fmt" "io" + "os" + "path/filepath" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) @@ -148,3 +151,24 @@ func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.O } return store, nil } + +func writeLocalFile(path string, data []byte, force bool) error { + cleaned := filepath.Clean(strings.TrimSpace(path)) + if cleaned == "" || cleaned == "." { + return fmt.Errorf("output path is required") + } + if !force { + if _, err := os.Stat(cleaned); err == nil { + return fmt.Errorf("output file %q already exists; pass --force to overwrite", cleaned) + } else if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("check output file %q: %w", cleaned, err) + } + } + if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(cleaned)); err != nil { + return fmt.Errorf("create output directory: %w", err) + } + if err := os.WriteFile(cleaned, data, fileops.WorkspaceFileMode); err != nil { + return err + } + return os.Chmod(cleaned, fileops.WorkspaceFileMode) +} diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index 3140c62..92f2b64 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -3,13 +3,11 @@ package app import ( "bytes" "context" - "errors" "fmt" "os" "path/filepath" "sort" "strings" - "sync" "testing" "time" @@ -626,123 +624,6 @@ func TestExecuteLocksListIsSortedAndRejectsMalformedObjects(t *testing.T) { } } -func TestMutateRemoteLockStoreRetainsConcurrentUpdates(t *testing.T) { - cfg := &config.Config{ - Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, - Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, - } - fake := &storage.FakeBackend{} - arrived := make(chan struct{}, 2) - release := make(chan struct{}) - var hookMu sync.Mutex - hookCalls := 0 - fake.UploadHook = func(storage.FakeUploadCall) error { - hookMu.Lock() - hookCalls++ - call := hookCalls - hookMu.Unlock() - if call <= 2 { - arrived <- struct{}{} - <-release - } - return nil - } - - mutate := func(source string) error { - return mutateRemoteLockStore(context.Background(), cfg, fake, func(lockStore *config.PublishLockStore) error { - set := lockSourceSet(lockStore.Locks) - set[source] = config.PublishLockRule{Source: source} - lockStore.Locks = lockMapValues(set) - return nil - }) - } - errs := make(chan error, 2) - go func() { errs <- mutate("narratio.transcript.final") }() - go func() { errs <- mutate("narratio.transcript.final_trimmed") }() - <-arrived - <-arrived - close(release) - if err := <-errs; err != nil { - t.Fatalf("first concurrent mutation error = %v", err) - } - if err := <-errs; err != nil { - t.Fatalf("second concurrent mutation error = %v", err) - } - - locks, _, _, err := loadRemoteLockStore(context.Background(), cfg, fake) - if err != nil { - t.Fatalf("loadRemoteLockStore() error = %v", err) - } - if len(locks.Locks) != 2 || locks.Locks[0].Source != "narratio.transcript.final" || locks.Locks[1].Source != "narratio.transcript.final_trimmed" { - t.Fatalf("remote locks = %#v, want both concurrent updates", locks.Locks) - } -} - -func TestLoadRemoteLockStoreAcceptsExactLimitAndRejectsLimitPlusOne(t *testing.T) { - cfg := &config.Config{ - Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, - Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, - } - key, err := remoteLocksKey(cfg) - if err != nil { - t.Fatalf("remoteLocksKey() error = %v", err) - } - encoded, err := config.MarshalPublishLockStore(&config.PublishLockStore{}) - if err != nil { - t.Fatalf("MarshalPublishLockStore() error = %v", err) - } - exact := append(append([]byte(nil), encoded...), bytes.Repeat([]byte(" "), int(MaxRemoteLockStoreBytes)-len(encoded))...) - store := &storage.FakeBackend{} - store.SeedObject(storage.FakeObject{Key: key, Data: exact, ETag: "lock-generation"}) - - locks, gotKey, generation, err := loadRemoteLockStore(context.Background(), cfg, store) - if err != nil { - t.Fatalf("loadRemoteLockStore() exact-limit error = %v", err) - } - if locks == nil || gotKey != key || generation != "lock-generation" { - t.Fatalf("loadRemoteLockStore() = (%#v, %q, %q), want decoded locks and opened generation", locks, gotKey, generation) - } - if len(store.Downloads) != 0 || len(store.Reads) != 1 || store.Reads[0].Key != key { - t.Fatalf("lock transfers reads=%#v downloads=%#v, want one direct read", store.Reads, store.Downloads) - } - - store.SeedObject(storage.FakeObject{Key: key, Data: append(exact, ' '), ETag: "new-generation"}) - _, _, _, err = loadRemoteLockStore(context.Background(), cfg, store) - if err == nil || !strings.Contains(err.Error(), "remote lock control object") || !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), fmt.Sprint(MaxRemoteLockStoreBytes)) { - t.Fatalf("loadRemoteLockStore() limit-plus-one error = %v, want category, key, and limit", err) - } -} - -func TestLoadRemoteLockStoreRejectsMalformedYAML(t *testing.T) { - cfg := &config.Config{ - Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, - Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, - } - key, err := remoteLocksKey(cfg) - if err != nil { - t.Fatalf("remoteLocksKey() error = %v", err) - } - store := &storage.FakeBackend{} - store.SeedObject(storage.FakeObject{Key: key, Data: []byte("locks: [\n"), ETag: "lock-generation"}) - - if _, _, _, err := loadRemoteLockStore(context.Background(), cfg, store); err == nil { - t.Fatal("loadRemoteLockStore() error = nil, want malformed YAML failure") - } -} - -func TestMutateRemoteLockStoreHonorsCancellation(t *testing.T) { - cfg := &config.Config{ - Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, - Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, - } - ctx, cancel := context.WithCancel(context.Background()) - cancel() - err := mutateRemoteLockStore(ctx, cfg, &storage.FakeBackend{}, func(*config.PublishLockStore) error { return nil }) - if !errors.Is(err, context.Canceled) { - t.Fatalf("mutateRemoteLockStore() error = %v, want context cancellation", err) - } -} - func TestExecuteLocksAddDuplicateRequiresForce(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) @@ -908,39 +789,6 @@ func TestExecuteLocksMutationRejectsSessionIDMismatch(t *testing.T) { } } -func TestExecuteLocksDoNotTreatStaticConfigAsAuthority(t *testing.T) { - workspaceRoot := t.TempDir() - pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - addStaticPublishLockToPipelineConfig(t, pipelinePath, "narratio.transcript.final_trimmed") - fake := &storage.FakeBackend{} - var storeInitCalls int - restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) - - var stdout bytes.Buffer - var stderr bytes.Buffer - code := Execute([]string{ - "session", "locks", "add", "2026-05-03", "narratio.transcript.final_trimmed", - "--config", pipelinePath, - "--campaign-file", campaignPath, - "--session", sessionPath, - }, &stdout, &stderr) - if code != 0 { - t.Fatalf("locks add exit code = %d, want 0; stderr=%q", code, stderr.String()) - } - - 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 { - t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String()) - } -} - func TestExecuteTopLevelLockAndUnlockAreRemoved(t *testing.T) { tests := []string{"lock", "unlock"} for _, cmd := range tests { @@ -1385,11 +1233,12 @@ func TestExecuteSessionValidateReportsPreviousStateFindingAndReturnsFindingError } } -func TestExecutePublishLoadsRemoteLocks(t *testing.T) { +func TestExecutePublishIgnoresRetiredSharedLockObject(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidPublishRunConfigFiles(t, workspaceRoot) fake := &storage.FakeBackend{} - lockKey := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + lockKey := sessionPrefix + "locks.yml" fake.SeedObject(storage.FakeObject{Key: lockKey, Data: []byte("locks:\n - source: narratio.transcript.final_trimmed\n reason: remote review\n")}) workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") @@ -1441,9 +1290,18 @@ func TestExecutePublishLoadsRemoteLocks(t *testing.T) { if storeInitCalls != 1 { t.Fatalf("object store initialization calls = %d, want 1", storeInitCalls) } - publishedKey := artifacts.S3PublishedOutputKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "transcripts/final.trimmed.json") - if _, ok := fake.Objects[publishedKey]; ok { - t.Fatalf("locked published key %q was uploaded", publishedKey) + current, err := artifacts.LoadCurrentState(context.Background(), fake, sessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true}) + if err != nil { + t.Fatalf("LoadCurrentState() error = %v", err) + } + found := false + for _, declaration := range current.Commit.Artifacts { + if declaration.Type == artifacts.RemoteArtifactTypePublishedOutput && declaration.Source == "narratio.transcript.final_trimmed" { + found = true + } + } + if !found { + t.Fatal("retired shared lock object affected publication") } } @@ -1607,23 +1465,3 @@ func writeValidPublishRunConfigFiles(t *testing.T, workspaceRoot string) (string } return pipelinePath, campaignPath, sessionPath } - -func addStaticPublishLockToPipelineConfig(t *testing.T, pipelinePath, source string) { - t.Helper() - data, err := os.ReadFile(pipelinePath) - if err != nil { - t.Fatalf("read pipeline: %v", err) - } - updated := strings.Replace( - string(data), - "publish:\n enabled: true\n upload_run: false\n", - "publish:\n enabled: true\n upload_run: false\n locks:\n - source: "+source+"\n reason: static review\n", - 1, - ) - if updated == string(data) { - t.Fatalf("publish section not found in pipeline config") - } - if err := os.WriteFile(pipelinePath, []byte(updated), 0o644); err != nil { - t.Fatalf("write pipeline: %v", err) - } -} diff --git a/internal/app/remote_locks.go b/internal/app/remote_locks.go deleted file mode 100644 index cdef96a..0000000 --- a/internal/app/remote_locks.go +++ /dev/null @@ -1,197 +0,0 @@ -package app - -import ( - "bytes" - "context" - "errors" - "fmt" - "os" - "path/filepath" - "sort" - "strings" - - "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" - "gitea.maximumdirect.net/eric/narratio/internal/artifacts" - "gitea.maximumdirect.net/eric/narratio/internal/config" - "gitea.maximumdirect.net/eric/narratio/internal/fileops" -) - -type effectiveLocks struct { - Static []config.PublishLockRule - Remote []config.PublishLockRule - All []config.PublishLockRule - Key string -} - -const ( - remoteLockMutationAttempts = 4 - // MaxRemoteLockStoreBytes bounds the mutable remote publish-lock document. - MaxRemoteLockStoreBytes int64 = 1 << 20 -) - -func remoteLocksKey(cfg *config.Config) (string, error) { - if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { - return "", fmt.Errorf("resolved config is required") - } - if cfg.Pipeline.Storage.S3 == nil { - return "", fmt.Errorf("pipeline.storage.s3 configuration is required") - } - sessionPrefix := artifacts.S3SessionPrefix( - cfg.Pipeline.Storage.S3.RootPrefix, - cfg.Session.Campaign, - cfg.Session.SessionID, - ) - return artifacts.S3SessionLocksKey(sessionPrefix), nil -} - -func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*config.PublishLockStore, string, string, error) { - key, err := remoteLocksKey(cfg) - if err != nil { - return nil, "", "", err - } - if store == nil { - return nil, key, "", fmt.Errorf("remote lock store is required") - } - info, data, err := storage.ReadObjectBounded(ctx, store, key, MaxRemoteLockStoreBytes) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return &config.PublishLockStore{}, key, "", nil - } - var limitErr *storage.ReadLimitError - if errors.As(err, &limitErr) { - return nil, key, "", fmt.Errorf("read remote lock control object %q with %d-byte limit: %w", key, MaxRemoteLockStoreBytes, err) - } - return nil, key, "", fmt.Errorf("read remote locks %q: %w", key, err) - } - if strings.TrimSpace(info.ETag) == "" { - return nil, key, "", fmt.Errorf("read remote locks %q: object has no generation", key) - } - lockStore, err := config.LoadPublishLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius) - if err != nil { - return nil, key, "", err - } - return lockStore, key, info.ETag, nil -} - -func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*effectiveLocks, error) { - staticLocks := staticPublishLocks(cfg) - if store == nil { - return &effectiveLocks{ - Static: staticLocks, - All: append([]config.PublishLockRule(nil), staticLocks...), - }, nil - } - lockStore, key, _, err := loadRemoteLockStore(ctx, cfg, store) - if err != nil { - return nil, err - } - remoteLocks := append([]config.PublishLockRule(nil), lockStore.Locks...) - return &effectiveLocks{ - Static: staticLocks, - Remote: remoteLocks, - All: config.MergePublishLockRules(staticLocks, remoteLocks), - Key: key, - }, nil -} - -func staticPublishLocks(cfg *config.Config) []config.PublishLockRule { - if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil { - return nil - } - return append([]config.PublishLockRule(nil), cfg.Pipeline.Publish.Locks...) -} - -func applyEffectiveLocks(cfg *config.Config, locks []config.PublishLockRule) { - if cfg == nil || cfg.Pipeline == nil { - return - } - if cfg.Pipeline.Publish == nil { - cfg.Pipeline.Publish = &config.PublishConfig{} - } - cfg.Pipeline.Publish.Locks = append([]config.PublishLockRule(nil), locks...) -} - -func mutateRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore, mutate func(*config.PublishLockStore) error) error { - for attempt := 0; attempt < remoteLockMutationAttempts; attempt++ { - if err := ctx.Err(); err != nil { - return err - } - lockStore, key, generation, err := loadRemoteLockStore(ctx, cfg, store) - if err != nil { - return err - } - if err := mutate(lockStore); err != nil { - return err - } - data, err := config.MarshalPublishLockStore(lockStore) - if err != nil { - return err - } - condition := storage.WriteCondition{MatchETag: generation} - if generation == "" { - condition = storage.WriteCondition{RequireAbsent: true} - } - _, err = store.UploadConditional(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}, condition) - if err == nil { - return nil - } - if !errors.Is(err, storage.ErrConditionNotMet) { - return fmt.Errorf("upload remote locks %q: %w", key, err) - } - } - if err := ctx.Err(); err != nil { - return err - } - return fmt.Errorf("update remote locks: concurrent updates prevented a conditional write after %d attempts", remoteLockMutationAttempts) -} - -func lockSourceSet(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 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 == "." { - return fmt.Errorf("output path is required") - } - if !force { - if _, err := os.Stat(cleaned); err == nil { - return fmt.Errorf("output file %q already exists; pass --force to overwrite", cleaned) - } else if err != nil && !os.IsNotExist(err) { - return fmt.Errorf("check output file %q: %w", cleaned, err) - } - } - if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(cleaned)); err != nil { - return fmt.Errorf("create output directory: %w", err) - } - if err := os.WriteFile(cleaned, data, fileops.WorkspaceFileMode); err != nil { - return err - } - return os.Chmod(cleaned, fileops.WorkspaceFileMode) -} diff --git a/internal/artifacts/s3_keys.go b/internal/artifacts/s3_keys.go index c7940d9..13fd322 100644 --- a/internal/artifacts/s3_keys.go +++ b/internal/artifacts/s3_keys.go @@ -40,12 +40,6 @@ func S3SessionConfigKey(sessionPrefix string) string { return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "session.yml") } -// S3SessionLocksKey returns the mutable session lock store key. -// Format: {session_prefix}/locks.yml -func S3SessionLocksKey(sessionPrefix string) string { - return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "locks.yml") -} - // S3RunCommitKey returns the immutable commit manifest key for one run. // Format: {session_prefix}/runs/{run_id}/commit.json func S3RunCommitKey(sessionPrefix, runID string) string { diff --git a/internal/artifacts/s3_keys_test.go b/internal/artifacts/s3_keys_test.go index a31ee6e..ad34d47 100644 --- a/internal/artifacts/s3_keys_test.go +++ b/internal/artifacts/s3_keys_test.go @@ -22,11 +22,6 @@ func TestS3KeyConstruction(t *testing.T) { t.Fatalf("session config key = %q", sessionConfigKey) } - locksKey := S3SessionLocksKey(`dnd\campaigns\forsaken\sessions\2026-04-19\`) - if locksKey != "dnd/campaigns/forsaken/sessions/2026-04-19/locks.yml" { - t.Fatalf("locks key = %q", locksKey) - } - runPrefix := S3RunPrefix(sessionPrefix, runID) wantRunPrefix := "dnd/campaigns/forsaken/sessions/2026-04-19/runs/" + runID + "/" if runPrefix != wantRunPrefix { diff --git a/internal/config/config.go b/internal/config/config.go index 0b0d2bd..d24cb3d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -130,7 +130,6 @@ type PublishConfig struct { Enabled *bool `yaml:"enabled"` UploadRun *bool `yaml:"upload_run"` Outputs []PublishOutputRule `yaml:"outputs"` - Locks []PublishLockRule `yaml:"locks"` } // PublishOutputRule configures one source-to-destination publish mapping. @@ -140,18 +139,6 @@ type PublishOutputRule struct { Required *bool `yaml:"required"` } -// PublishLockRule prevents one source from overwriting its top-level published -// destination. -type PublishLockRule struct { - Source string `yaml:"source"` - Reason string `yaml:"reason"` -} - -// PublishLockStore is the mutable per-session remote lock store. -type PublishLockStore struct { - Locks []PublishLockRule `yaml:"locks"` -} - // WhisperXConfig configures WhisperX adapter settings. type WhisperXConfig struct { TranscribeURL string `yaml:"transcribe_url"` diff --git a/internal/config/effective_digest.go b/internal/config/effective_digest.go index cceef3b..9e9fde2 100644 --- a/internal/config/effective_digest.go +++ b/internal/config/effective_digest.go @@ -10,7 +10,10 @@ import ( "gopkg.in/yaml.v3" ) -const pipelineDefaultOwnershipSource = "default" +const ( + pipelineDefaultOwnershipSource = "default" + effectivePipelineDigestVersion = "narratio-effective-pipeline-v2\n" +) // PipelineProfileProvenance identifies the profile selected while resolving a // pipeline. It contains only non-secret selection metadata. @@ -98,7 +101,8 @@ func recomputePipelineEffectiveDigest(cfg *PipelineConfig) error { if err != nil { return err } - digest := sha256.Sum256(canonical) + digestInput := append([]byte(effectivePipelineDigestVersion), canonical...) + digest := sha256.Sum256(digestInput) cfg.resolution.effectiveDigest = hex.EncodeToString(digest[:]) return nil } diff --git a/internal/config/load.go b/internal/config/load.go index 139e419..fb807f7 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -177,38 +177,6 @@ func LoadSessionBytesWithOptions(label string, data []byte, opts SessionLoadOpti return &cfg, nil } -// LoadPublishLockStoreBytes loads a mutable session lock store with strict -// field checking and source validation. -func LoadPublishLockStoreBytes( - label string, - data []byte, - scriptorium *ScriptoriumConfig, - notarius *NotariusConfig, -) (*PublishLockStore, error) { - var store PublishLockStore - if err := decodeStrictYAMLFromReader("publish lock store", label, strings.NewReader(string(data)), &store); err != nil { - return nil, fmt.Errorf("load publish lock store: %w", err) - } - locks, err := ValidatePublishLockRules(store.Locks, scriptorium, notarius, "locks") - if err != nil { - return nil, fmt.Errorf("load publish lock store: %w", err) - } - store.Locks = locks - return &store, nil -} - -// MarshalPublishLockStore serializes a mutable lock store as strict-compatible YAML. -func MarshalPublishLockStore(store *PublishLockStore) ([]byte, error) { - if store == nil { - store = &PublishLockStore{} - } - data, err := yaml.Marshal(store) - if err != nil { - return nil, fmt.Errorf("marshal publish lock store: %w", err) - } - return data, nil -} - // Load loads and resolves combined pipeline, campaign, and session configuration. // Passing only a session path is supported for package-internal compatibility; // in that form campaign.yml is expected next to the session file. diff --git a/internal/config/loader_strict_test.go b/internal/config/loader_strict_test.go index 3557e9d..6754b8a 100644 --- a/internal/config/loader_strict_test.go +++ b/internal/config/loader_strict_test.go @@ -43,13 +43,6 @@ func TestLoadersRejectTrailingYAMLDocuments(t *testing.T) { return err }, }, - { - name: "publish lock store", - load: func() error { - _, err := LoadPublishLockStoreBytes("s3://bucket/locks.yml", []byte("locks: []\n---\n{}\n"), nil, nil) - return err - }, - }, } for _, tt := range tests { diff --git a/internal/config/notarius_test.go b/internal/config/notarius_test.go index b13f24d..ed83ced 100644 --- a/internal/config/notarius_test.go +++ b/internal/config/notarius_test.go @@ -571,13 +571,6 @@ func TestExtractionReferencesRequireDeclaredOutput(t *testing.T) { `, wantErr: []string{"pipeline.publish.outputs[0].source", "not defined"}, }, - { - name: "publish lock declared extraction", - body: declared + `publish: - locks: - - source: narratio.extraction.npc_registry -`, - }, } for _, tt := range tests { diff --git a/internal/config/storage_publish_test.go b/internal/config/storage_publish_test.go index 1a90493..ad11e87 100644 --- a/internal/config/storage_publish_test.go +++ b/internal/config/storage_publish_test.go @@ -492,124 +492,16 @@ func TestScriptoriumOutputPathsIgnoreInvalidKeys(t *testing.T) { } } -func TestPublishLockValidation(t *testing.T) { - tests := []struct { - name string - pipelineYML string - wantErr string - }{ - { - name: "valid built in source", - pipelineYML: testPipelineBaseYAML + ` -publish: - locks: - - source: narratio.transcript.final_trimmed - reason: reviewed transcript -`, - }, - { - name: "valid configured source", - pipelineYML: testPipelineBaseYAML + ` -scriptorium: - artifacts: - session_recap: - enabled: true - prompt_id: dnd.session_recap - output_path: artifacts/session_recap.md -publish: - locks: - - source: narratio.artifact.session_recap -`, - }, - { - name: "missing source rejected", - pipelineYML: testPipelineBaseYAML + ` -publish: - locks: - - reason: no source -`, - wantErr: "pipeline.publish.locks[0].source is required", - }, - { - name: "invalid source rejected", - pipelineYML: testPipelineBaseYAML + ` -publish: - locks: - - source: narratio.unknown -`, - wantErr: "pipeline.publish.locks[0].source \"narratio.unknown\" is unsupported", - }, - { - name: "prepared input source rejected", - pipelineYML: testPipelineBaseYAML + ` -publish: - locks: - - source: narratio.input.players -`, - wantErr: "pipeline.publish.locks[0].source \"narratio.input.players\" is unsupported", - }, - { - name: "duplicate source rejected", - pipelineYML: testPipelineBaseYAML + ` -publish: - locks: - - source: narratio.transcript.final_trimmed - - source: " narratio.transcript.final_trimmed " -`, - wantErr: "duplicates another publish lock source", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - pipelinePath, sessionPath := writeConfigFiles(t, tt.pipelineYML, testSessionBaseYAML) - cfg, err := Load(pipelinePath, sessionPath) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - err = Validate(cfg) - if tt.wantErr == "" { - if err != nil { - t.Fatalf("Validate() error = %v", err) - } - return - } - if err == nil || !strings.Contains(err.Error(), tt.wantErr) { - t.Fatalf("Validate() error = %v, want to contain %q", err, tt.wantErr) - } - }) - } -} - -func TestPublishLockUnsupportedTranscriptSourceRejected(t *testing.T) { - source := "narratio.transcript." + "merged" - pipelineYAML := testPipelineBaseYAML + ` -publish: - locks: - - source: ` + source + ` -` - pipelinePath, sessionPath := writeConfigFiles(t, pipelineYAML, testSessionBaseYAML) - cfg, err := Load(pipelinePath, sessionPath) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - err = Validate(cfg) - if err == nil || !strings.Contains(err.Error(), "pipeline.publish.locks[0].source") || !strings.Contains(err.Error(), "unsupported") { - t.Fatalf("Validate() error = %v, want source field and unsupported category", err) - } -} - -func TestPublishLockUnknownFieldFailsStrictDecode(t *testing.T) { +func TestPublishLocksFieldFailsStrictDecode(t *testing.T) { pipelineYAML := testPipelineBaseYAML + ` publish: locks: - source: narratio.transcript.final_trimmed - dest: transcripts/final.trimmed.json ` pipelinePath, sessionPath := writeConfigFiles(t, pipelineYAML, testSessionBaseYAML) _, err := Load(pipelinePath, sessionPath) - if err == nil || !strings.Contains(err.Error(), "strict decode failed") { - t.Fatalf("Load() error = %v, want strict decode failed", err) + if err == nil || !strings.Contains(err.Error(), "strict decode failed") || !strings.Contains(err.Error(), "field locks not found") { + t.Fatalf("Load() error = %v, want retired locks field rejected", err) } } @@ -627,54 +519,6 @@ publish: } } -func TestPublishLockStoreBytesStrictDecodeAndValidation(t *testing.T) { - store, err := LoadPublishLockStoreBytes("locks.yml", []byte(`locks: - - source: narratio.transcript.final_trimmed - reason: reviewed -`), nil, nil) - if err != nil { - t.Fatalf("LoadPublishLockStoreBytes() error = %v", err) - } - if len(store.Locks) != 1 || store.Locks[0].Source != "narratio.transcript.final_trimmed" || store.Locks[0].Reason != "reviewed" { - t.Fatalf("locks = %#v", store.Locks) - } - - _, err = LoadPublishLockStoreBytes("locks.yml", []byte(`locks: - - source: narratio.transcript.final_trimmed - dest: transcripts/final.trimmed.json -`), nil, nil) - if err == nil || !strings.Contains(err.Error(), "strict decode failed") { - t.Fatalf("unknown field error = %v, want strict decode failed", err) - } - - _, err = LoadPublishLockStoreBytes("locks.yml", []byte(`locks: - - source: narratio.transcript.final_trimmed - - source: narratio.transcript.final_trimmed -`), nil, nil) - if err == nil || !strings.Contains(err.Error(), "duplicates another publish lock source") { - t.Fatalf("duplicate error = %v", err) - } -} - -func TestMergePublishLockRulesStaticWins(t *testing.T) { - merged := MergePublishLockRules( - []PublishLockRule{{Source: "narratio.transcript.final_trimmed", Reason: "static"}}, - []PublishLockRule{ - {Source: "narratio.transcript.final_trimmed", Reason: "remote"}, - {Source: "narratio.transcript.final", Reason: "remote full"}, - }, - ) - if len(merged) != 2 { - t.Fatalf("merged len = %d, want 2: %#v", len(merged), merged) - } - if merged[0].Source != "narratio.transcript.final_trimmed" || merged[0].Reason != "static" { - t.Fatalf("merged[0] = %#v, want static lock", merged[0]) - } - if merged[1].Source != "narratio.transcript.final" { - t.Fatalf("merged[1] = %#v, want remote full lock", merged[1]) - } -} - func TestSessionAudioS3Validation(t *testing.T) { tests := []struct { name string diff --git a/internal/config/validate_publish.go b/internal/config/validate_publish.go index a091caa..2558051 100644 --- a/internal/config/validate_publish.go +++ b/internal/config/validate_publish.go @@ -83,71 +83,9 @@ func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig, notariu } seenDest[normalizedDest] = struct{}{} } - locks, err := ValidatePublishLockRules(cfg.Locks, scriptorium, notarius, "pipeline.publish.locks") - if err != nil { - return err - } - cfg.Locks = locks return nil } -// ValidatePublishLockRules validates and normalizes source-based publish locks. -func ValidatePublishLockRules(locks []PublishLockRule, scriptorium *ScriptoriumConfig, notarius *NotariusConfig, label string) ([]PublishLockRule, error) { - seenLocks := map[string]struct{}{} - out := make([]PublishLockRule, 0, len(locks)) - configuredOutputs := scriptoriumOutputPathMap(scriptorium) - extractionOutputs := notariusOutputKeySet(notarius) - if strings.TrimSpace(label) == "" { - label = "publish.locks" - } - for i, item := range locks { - prefix := fmt.Sprintf("%s[%d]", label, i) - source := strings.TrimSpace(item.Source) - if source == "" { - return nil, fmt.Errorf("%s.source is required", prefix) - } - if _, err := artifactpolicy.ValidatePublishSourceWithExtractions(source, configuredOutputs, extractionOutputs); err != nil { - return nil, fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err) - } - if _, ok := seenLocks[source]; ok { - return nil, fmt.Errorf("%s.source %q duplicates another publish lock source", prefix, source) - } - seenLocks[source] = struct{}{} - out = append(out, PublishLockRule{ - Source: source, - Reason: strings.TrimSpace(item.Reason), - }) - } - return out, nil -} - -// MergePublishLockRules returns the union of static and remote locks. Static -// locks win when both sources contain the same lock. -func MergePublishLockRules(staticLocks, remoteLocks []PublishLockRule) []PublishLockRule { - out := make([]PublishLockRule, 0, len(staticLocks)+len(remoteLocks)) - seen := map[string]struct{}{} - for _, item := range staticLocks { - source := strings.TrimSpace(item.Source) - if source == "" { - continue - } - out = append(out, PublishLockRule{Source: source, Reason: strings.TrimSpace(item.Reason)}) - seen[source] = struct{}{} - } - for _, item := range remoteLocks { - source := strings.TrimSpace(item.Source) - if source == "" { - continue - } - if _, ok := seen[source]; ok { - continue - } - out = append(out, PublishLockRule{Source: source, Reason: strings.TrimSpace(item.Reason)}) - seen[source] = struct{}{} - } - return out -} - func scriptoriumOutputPathMap(scriptorium *ScriptoriumConfig) map[string]string { out := map[string]string{} if scriptorium == nil || len(scriptorium.Artifacts) == 0 { diff --git a/internal/stage/semantic_contracts_delivery.go b/internal/stage/semantic_contracts_delivery.go index 869564d..efa6d58 100644 --- a/internal/stage/semantic_contracts_delivery.go +++ b/internal/stage/semantic_contracts_delivery.go @@ -15,7 +15,7 @@ import ( const ( extractSemanticConfigVersion = 1 - publishSemanticConfigVersion = 1 + publishSemanticConfigVersion = 2 notifySemanticConfigVersion = 1 ) @@ -45,7 +45,6 @@ type publishSemanticConfig struct { Enabled bool `json:"enabled"` UploadRun bool `json:"upload_run"` Rules []publishSemanticRule `json:"rules"` - StaticLocks []publishSemanticLock `json:"static_locks"` Destination publishSemanticDestination `json:"destination"` } @@ -55,11 +54,6 @@ type publishSemanticRule struct { Required bool `json:"required"` } -type publishSemanticLock struct { - Source string `json:"source"` - Reason string `json:"reason,omitempty"` -} - type publishSemanticDestination struct { Backend string `json:"backend"` Bucket string `json:"bucket,omitempty"` @@ -122,7 +116,7 @@ func (publishStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfig } pipeline := env.Config.Pipeline payload := publishSemanticConfig{ - Rules: []publishSemanticRule{}, StaticLocks: []publishSemanticLock{}, + Rules: []publishSemanticRule{}, Destination: publishSemanticDestination{Backend: strings.ToLower(strings.TrimSpace(pipeline.Storage.Backend))}, } if payload.Destination.Backend == "" { @@ -166,19 +160,6 @@ func (publishStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfig } return payload.Rules[i].Dest < payload.Rules[j].Dest }) - for _, lock := range cfg.Locks { - source, err := artifactpolicy.ValidatePublishSourceWithExtractions(lock.Source, configured, extractions) - if err != nil { - return manifest.SemanticConfigFingerprint{}, fmt.Errorf("publish semantic lock source %q: %w", lock.Source, err) - } - payload.StaticLocks = append(payload.StaticLocks, publishSemanticLock{ - Source: source.ID, - Reason: strings.TrimSpace(lock.Reason), - }) - } - sort.Slice(payload.StaticLocks, func(i, j int) bool { - return payload.StaticLocks[i].Source < payload.StaticLocks[j].Source - }) return FingerprintSemanticConfig(publishSemanticConfigVersion, payload) } diff --git a/internal/stage/semantic_contracts_delivery_test.go b/internal/stage/semantic_contracts_delivery_test.go index ba12a9d..5195d7f 100644 --- a/internal/stage/semantic_contracts_delivery_test.go +++ b/internal/stage/semantic_contracts_delivery_test.go @@ -101,7 +101,6 @@ func TestPublishSemanticConfigSensitivityAndOperationalExclusions(t *testing.T) {name: "source", mutate: func(env *Env) { env.Config.Pipeline.Publish.Outputs[0].Source = "narratio.transcript.final_markdown" }}, {name: "destination", mutate: func(env *Env) { env.Config.Pipeline.Publish.Outputs[0].Dest = "published/other.json" }}, {name: "required", mutate: func(env *Env) { env.Config.Pipeline.Publish.Outputs[0].Required = boolPointer(false) }}, - {name: "static lock", mutate: func(env *Env) { env.Config.Pipeline.Publish.Locks = nil }}, {name: "backend", mutate: func(env *Env) { env.Config.Pipeline.Storage.Backend = config.StorageBackendLocal }}, {name: "bucket", mutate: func(env *Env) { env.Config.Pipeline.Storage.S3.Bucket = "other-bucket" }}, {name: "region", mutate: func(env *Env) { env.Config.Pipeline.Storage.S3.Region = "eu-west-1" }}, @@ -140,8 +139,6 @@ func TestDeliverySemanticConfigCanonicalOrderingAndNotifyMode(t *testing.T) { second := downstreamSemanticEnv() second.Config.Pipeline.Publish.Outputs[0], second.Config.Pipeline.Publish.Outputs[1] = second.Config.Pipeline.Publish.Outputs[1], second.Config.Pipeline.Publish.Outputs[0] - second.Config.Pipeline.Publish.Locks[0], second.Config.Pipeline.Publish.Locks[1] = - second.Config.Pipeline.Publish.Locks[1], second.Config.Pipeline.Publish.Locks[0] firstPublish, err := (publishStage{}).SemanticConfigFingerprint(first) if err != nil { t.Fatal(err) @@ -221,10 +218,6 @@ func downstreamSemanticEnv() *Env { {Source: "narratio.transcript.final_trimmed", Dest: "transcripts/final.trimmed.json", Required: boolPointer(true)}, {Source: "narratio.transcript.final_trimmed_markdown", Dest: "transcripts/final.trimmed.md", Required: boolPointer(true)}, }, - Locks: []config.PublishLockRule{ - {Source: "narratio.transcript.final_trimmed", Reason: "curated"}, - {Source: "narratio.transcript.final_trimmed_markdown", Reason: "approved"}, - }, }, Notification: config.NotificationConfig{Mode: config.DefaultNotificationMode}, },