Add archive promotion locks
This commit is contained in:
@@ -154,6 +154,9 @@ archive:
|
|||||||
- source: narratio.artifact.session_recap
|
- source: narratio.artifact.session_recap
|
||||||
dest: artifacts/session_recap.md
|
dest: artifacts/session_recap.md
|
||||||
required: true
|
required: true
|
||||||
|
locks:
|
||||||
|
- source: narratio.artifact.session_recap
|
||||||
|
reason: Final recap was manually edited.
|
||||||
|
|
||||||
whisperx:
|
whisperx:
|
||||||
transcribe_url: "https://transcription.example.com/transcribe"
|
transcribe_url: "https://transcription.example.com/transcribe"
|
||||||
@@ -177,6 +180,7 @@ Operational notes:
|
|||||||
|
|
||||||
- archive promotion is explicit and source-based via `archive.promote_artifacts`.
|
- archive promotion is explicit and source-based via `archive.promote_artifacts`.
|
||||||
- `source` is required; `dest` is optional and derived when omitted.
|
- `source` is required; `dest` is optional and derived when omitted.
|
||||||
|
- `archive.locks` skips top-level promotion overwrites for locked sources while preserving run-local uploads.
|
||||||
- Narratio does not auto-promote all generated analyze artifacts.
|
- Narratio does not auto-promote all generated analyze artifacts.
|
||||||
- `restore` reads the same config/campaign/session inputs and restore scope is bounded by committed archive current state.
|
- `restore` reads the same config/campaign/session inputs and restore scope is bounded by committed archive current state.
|
||||||
|
|
||||||
@@ -205,6 +209,9 @@ Operational notes:
|
|||||||
| `pipeline.archive.promote_artifacts[].source` | string | Yes (per rule) | none |
|
| `pipeline.archive.promote_artifacts[].source` | string | Yes (per rule) | none |
|
||||||
| `pipeline.archive.promote_artifacts[].dest` | string | No | derived from source |
|
| `pipeline.archive.promote_artifacts[].dest` | string | No | derived from source |
|
||||||
| `pipeline.archive.promote_artifacts[].required` | bool | No | `true` |
|
| `pipeline.archive.promote_artifacts[].required` | bool | No | `true` |
|
||||||
|
| `pipeline.archive.locks[]` | list | No | empty |
|
||||||
|
| `pipeline.archive.locks[].source` | string | Yes (per lock) | none |
|
||||||
|
| `pipeline.archive.locks[].reason` | string | No | empty |
|
||||||
| `pipeline.whisperx.transcribe_url` | string | Yes | none |
|
| `pipeline.whisperx.transcribe_url` | string | Yes | none |
|
||||||
| `pipeline.whisperx.language` | string | No | `en` |
|
| `pipeline.whisperx.language` | string | No | `en` |
|
||||||
| `pipeline.whisperx.timeout` | duration string | No | `30m` |
|
| `pipeline.whisperx.timeout` | duration string | No | `30m` |
|
||||||
@@ -305,6 +312,8 @@ Allowed `pipeline.scriptorium.artifacts.<name>.inputs.<key>.source` values:
|
|||||||
- `narratio.bounds.session`
|
- `narratio.bounds.session`
|
||||||
- `narratio.artifact.<configured_artifact_key>`
|
- `narratio.artifact.<configured_artifact_key>`
|
||||||
|
|
||||||
|
`pipeline.archive.locks[].source` accepts the same source values as `pipeline.archive.promote_artifacts[].source`.
|
||||||
|
|
||||||
Archive promotion destination rules:
|
Archive promotion destination rules:
|
||||||
|
|
||||||
- `dest` must be a clean relative path (not absolute, no traversal).
|
- `dest` must be a clean relative path (not absolute, no traversal).
|
||||||
@@ -314,6 +323,14 @@ Archive promotion destination rules:
|
|||||||
- configured sources derive from `pipeline.scriptorium.artifacts.<name>.output_path`;
|
- configured sources derive from `pipeline.scriptorium.artifacts.<name>.output_path`;
|
||||||
- derivation failure is a config validation error.
|
- derivation failure is a config validation error.
|
||||||
|
|
||||||
|
Archive lock rules:
|
||||||
|
|
||||||
|
- locks are source-based and do not accept `dest`.
|
||||||
|
- duplicate lock sources are rejected.
|
||||||
|
- locked promotions are recorded as intentional skips in archive metadata.
|
||||||
|
- locked required promotions do not fail archive by default.
|
||||||
|
- ordinary `--force` reruns do not override locks.
|
||||||
|
|
||||||
Restore-related implications:
|
Restore-related implications:
|
||||||
|
|
||||||
- restore remote identity requires archive S3 identity to resolve (`pipeline.storage.s3.bucket` and session prefix derivation inputs).
|
- restore remote identity requires archive S3 identity to resolve (`pipeline.storage.s3.bucket` and session prefix derivation inputs).
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ Inputs:
|
|||||||
- session manifest and prerequisite stage records
|
- session manifest and prerequisite stage records
|
||||||
- run root contents under `runs/{run_id}/`
|
- run root contents under `runs/{run_id}/`
|
||||||
- promotion rules with artifact `source` IDs and archive `dest` paths (`archive.promote_artifacts`)
|
- promotion rules with artifact `source` IDs and archive `dest` paths (`archive.promote_artifacts`)
|
||||||
|
- source-based promotion locks (`archive.locks`)
|
||||||
- session-level `previous/**` cache files when present
|
- session-level `previous/**` cache files when present
|
||||||
|
|
||||||
Outputs:
|
Outputs:
|
||||||
@@ -23,6 +24,7 @@ Owns:
|
|||||||
- Prerequisite stage success enforcement
|
- Prerequisite stage success enforcement
|
||||||
- Run file collection and upload (excluding `audio/`)
|
- Run file collection and upload (excluding `audio/`)
|
||||||
- Promotion rule resolution and upload
|
- Promotion rule resolution and upload
|
||||||
|
- Promotion lock enforcement
|
||||||
- Session previous-cache file collection/upload
|
- Session previous-cache file collection/upload
|
||||||
- Commit pointer publish order
|
- Commit pointer publish order
|
||||||
|
|
||||||
@@ -34,6 +36,7 @@ Does not own:
|
|||||||
- `pipeline.archive.enabled`
|
- `pipeline.archive.enabled`
|
||||||
- `pipeline.archive.upload_run`
|
- `pipeline.archive.upload_run`
|
||||||
- `pipeline.archive.promote_artifacts`
|
- `pipeline.archive.promote_artifacts`
|
||||||
|
- `pipeline.archive.locks`
|
||||||
- `pipeline.storage.s3.bucket`
|
- `pipeline.storage.s3.bucket`
|
||||||
- `pipeline.storage.s3.root_prefix`
|
- `pipeline.storage.s3.root_prefix`
|
||||||
- `pipeline.workspace.root`
|
- `pipeline.workspace.root`
|
||||||
@@ -47,9 +50,11 @@ Does not own:
|
|||||||
- Requires `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, and `analyze` status `succeeded`.
|
- Requires `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, and `analyze` status `succeeded`.
|
||||||
- Resolves bucket/prefix from manifest identity first, then config fallback.
|
- Resolves bucket/prefix from manifest identity first, then config fallback.
|
||||||
- Uploads session `previous/**` files as durable session state when the local `previous/` directory exists.
|
- Uploads session `previous/**` files as durable session state when the local `previous/` directory exists.
|
||||||
|
- Skips top-level promotion uploads for sources listed in `pipeline.archive.locks`; run-local uploads still publish.
|
||||||
- Writes metadata including:
|
- Writes metadata including:
|
||||||
- upload counts/paths
|
- upload counts/paths
|
||||||
- `previous_files_uploaded` and `previous_uploaded_paths`
|
- `previous_files_uploaded` and `previous_uploaded_paths`
|
||||||
|
- `locked_promotion_count` and `locked_promotions`
|
||||||
- `current_manifest_key`
|
- `current_manifest_key`
|
||||||
- `current_run_id_key`
|
- `current_run_id_key`
|
||||||
- `current_pointer_written`
|
- `current_pointer_written`
|
||||||
@@ -60,7 +65,8 @@ Does not own:
|
|||||||
- Runner-level skip also applies for previously succeeded stage unless forced.
|
- Runner-level skip also applies for previously succeeded stage unless forced.
|
||||||
|
|
||||||
## Failure Behavior
|
## Failure Behavior
|
||||||
- Fails on missing prerequisite success, missing object store when required, missing run root, missing required promotion source, upload failures, or pointer write failures.
|
- Fails on missing prerequisite success, missing object store when required, missing run root, missing unlocked required promotion source, upload failures, or pointer write failures.
|
||||||
|
- Locked required promotions are intentional skips and do not fail archive.
|
||||||
- Pointer semantics are fail-safe: `current/run_id.txt` is not written if prior required uploads fail.
|
- Pointer semantics are fail-safe: `current/run_id.txt` is not written if prior required uploads fail.
|
||||||
|
|
||||||
## Tests to Inspect Before Changing
|
## Tests to Inspect Before Changing
|
||||||
@@ -70,5 +76,6 @@ Does not own:
|
|||||||
## Architectural Invariants
|
## Architectural Invariants
|
||||||
- Run upload excludes `audio/` subtree.
|
- Run upload excludes `audio/` subtree.
|
||||||
- Session `previous/**` is archiveable durable input/provenance state, not run-local output.
|
- Session `previous/**` is archiveable durable input/provenance state, not run-local output.
|
||||||
|
- Ordinary `--force` does not override archive locks.
|
||||||
- `current/manifest.json` uploads before `current/run_id.txt`.
|
- `current/manifest.json` uploads before `current/run_id.txt`.
|
||||||
- `current/run_id.txt` is the remote publish commit marker.
|
- `current/run_id.txt` is the remote publish commit marker.
|
||||||
|
|||||||
@@ -132,6 +132,8 @@ Archive promotion is explicit and source-based:
|
|||||||
- missing required promotion sources fail archive stage.
|
- missing required promotion sources fail archive stage.
|
||||||
- missing optional promotion sources are skipped.
|
- missing optional promotion sources are skipped.
|
||||||
- invalid resolved artifacts fail archive stage.
|
- invalid resolved artifacts fail archive stage.
|
||||||
|
- `archive.locks` skips top-level promotion overwrites for locked sources while run-local uploads still publish.
|
||||||
|
- locked required promotions are treated as intentional successful skips and are recorded in archive metadata.
|
||||||
|
|
||||||
## Resume, retry, restore, and safe rerun behavior
|
## Resume, retry, restore, and safe rerun behavior
|
||||||
|
|
||||||
@@ -149,6 +151,7 @@ Restore conflict policy:
|
|||||||
|
|
||||||
Forced reruns:
|
Forced reruns:
|
||||||
- force-rerunning an upstream succeeded stage marks downstream succeeded stages as `stale`.
|
- force-rerunning an upstream succeeded stage marks downstream succeeded stages as `stale`.
|
||||||
|
- ordinary `--force` does not override archive locks.
|
||||||
|
|
||||||
Safe rerun pattern:
|
Safe rerun pattern:
|
||||||
1. rerun the changed stage with `--force`.
|
1. rerun the changed stage with `--force`.
|
||||||
|
|||||||
@@ -169,13 +169,13 @@ When `prepare` materializes a remote session into `inputs/session.yml`, record t
|
|||||||
- local checksum;
|
- local checksum;
|
||||||
- downloaded temp/materialized path.
|
- downloaded temp/materialized path.
|
||||||
|
|
||||||
## Phase 3: Add Logical Archive Locks
|
## Phase 3: Add Logical Archive Locks (implemented)
|
||||||
|
|
||||||
Add source-based archive locks under `pipeline.archive.locks`. The current promotion system is already source-based, so the first implementation must not support destination-based locks.
|
Narratio supports source-based archive locks under `pipeline.archive.locks`. The promotion system is source-based; destination-based locks are not supported.
|
||||||
|
|
||||||
### Config Shape
|
### Config Shape
|
||||||
|
|
||||||
Add lock entries:
|
Lock entries:
|
||||||
|
|
||||||
archive:
|
archive:
|
||||||
locks:
|
locks:
|
||||||
@@ -194,7 +194,7 @@ Validation rules:
|
|||||||
|
|
||||||
### Archive Behavior
|
### Archive Behavior
|
||||||
|
|
||||||
Archive must continue uploading complete run-local outputs under `runs/{run_id}/`.
|
Archive continues uploading complete run-local outputs under `runs/{run_id}/`.
|
||||||
|
|
||||||
Promotion behavior:
|
Promotion behavior:
|
||||||
|
|
||||||
@@ -205,11 +205,11 @@ Promotion behavior:
|
|||||||
5. Continue archive commit when all run-local uploads and all non-locked required promotions succeed.
|
5. Continue archive commit when all run-local uploads and all non-locked required promotions succeed.
|
||||||
6. Upload `current/manifest.json` and `current/run_id.txt` in the existing order, with `current/run_id.txt` last.
|
6. Upload `current/manifest.json` and `current/run_id.txt` in the existing order, with `current/run_id.txt` last.
|
||||||
|
|
||||||
Ordinary `--force` must not override locks. Do not implement a lock-break override in this phase.
|
Ordinary `--force` does not override locks. A lock-break override remains out of scope.
|
||||||
|
|
||||||
### Metadata
|
### Metadata
|
||||||
|
|
||||||
Record locked promotion skips in archive metadata/reporting so operators can distinguish missing optional promotions from lock-protected promotions.
|
Archive records locked promotion skips in archive metadata/reporting so operators can distinguish missing optional promotions from lock-protected promotions.
|
||||||
|
|
||||||
Include:
|
Include:
|
||||||
|
|
||||||
|
|||||||
@@ -95,6 +95,7 @@ type ArchiveConfig struct {
|
|||||||
Enabled *bool `yaml:"enabled"`
|
Enabled *bool `yaml:"enabled"`
|
||||||
UploadRun *bool `yaml:"upload_run"`
|
UploadRun *bool `yaml:"upload_run"`
|
||||||
PromoteArtifacts []ArchivePromotionRule `yaml:"promote_artifacts"`
|
PromoteArtifacts []ArchivePromotionRule `yaml:"promote_artifacts"`
|
||||||
|
Locks []ArchiveLockRule `yaml:"locks"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// ArchivePromotionRule configures one artifact promotion mapping.
|
// ArchivePromotionRule configures one artifact promotion mapping.
|
||||||
@@ -104,6 +105,13 @@ type ArchivePromotionRule struct {
|
|||||||
Required *bool `yaml:"required"`
|
Required *bool `yaml:"required"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ArchiveLockRule prevents one source-based promotion from overwriting its
|
||||||
|
// top-level archive destination.
|
||||||
|
type ArchiveLockRule struct {
|
||||||
|
Source string `yaml:"source"`
|
||||||
|
Reason string `yaml:"reason"`
|
||||||
|
}
|
||||||
|
|
||||||
// WhisperXConfig configures WhisperX adapter settings.
|
// WhisperXConfig configures WhisperX adapter settings.
|
||||||
type WhisperXConfig struct {
|
type WhisperXConfig struct {
|
||||||
TranscribeURL string `yaml:"transcribe_url"`
|
TranscribeURL string `yaml:"transcribe_url"`
|
||||||
|
|||||||
@@ -287,6 +287,100 @@ archive:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestArchiveLockValidation(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
pipelineYML string
|
||||||
|
wantErr string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "valid built in source",
|
||||||
|
pipelineYML: testPipelineBaseYAML + `
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- source: narratio.transcript.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
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- source: narratio.artifact.session_recap
|
||||||
|
`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "missing source rejected",
|
||||||
|
pipelineYML: testPipelineBaseYAML + `
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- reason: no source
|
||||||
|
`,
|
||||||
|
wantErr: "pipeline.archive.locks[0].source is required",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "invalid source rejected",
|
||||||
|
pipelineYML: testPipelineBaseYAML + `
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- source: narratio.unknown
|
||||||
|
`,
|
||||||
|
wantErr: "pipeline.archive.locks[0].source \"narratio.unknown\" is unsupported",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "duplicate source rejected",
|
||||||
|
pipelineYML: testPipelineBaseYAML + `
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- source: narratio.transcript.trimmed
|
||||||
|
- source: " narratio.transcript.trimmed "
|
||||||
|
`,
|
||||||
|
wantErr: "duplicates another archive 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 TestArchiveLockUnknownFieldFailsStrictDecode(t *testing.T) {
|
||||||
|
pipelineYAML := testPipelineBaseYAML + `
|
||||||
|
archive:
|
||||||
|
locks:
|
||||||
|
- source: narratio.transcript.trimmed
|
||||||
|
dest: transcripts/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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestArchiveLegacyFromToFailsStrictDecode(t *testing.T) {
|
func TestArchiveLegacyFromToFailsStrictDecode(t *testing.T) {
|
||||||
pipelineYAML := testPipelineBaseYAML + `
|
pipelineYAML := testPipelineBaseYAML + `
|
||||||
archive:
|
archive:
|
||||||
|
|||||||
@@ -152,6 +152,23 @@ func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error {
|
|||||||
}
|
}
|
||||||
seenDest[normalizedDest] = struct{}{}
|
seenDest[normalizedDest] = struct{}{}
|
||||||
}
|
}
|
||||||
|
seenLocks := map[string]struct{}{}
|
||||||
|
for i, item := range cfg.Locks {
|
||||||
|
prefix := fmt.Sprintf("pipeline.archive.locks[%d]", i)
|
||||||
|
source := strings.TrimSpace(item.Source)
|
||||||
|
if source == "" {
|
||||||
|
return fmt.Errorf("%s.source is required", prefix)
|
||||||
|
}
|
||||||
|
if _, err := archiveSourceKnown(source, scriptorium); err != nil {
|
||||||
|
return fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err)
|
||||||
|
}
|
||||||
|
if _, ok := seenLocks[source]; ok {
|
||||||
|
return fmt.Errorf("%s.source %q duplicates another archive lock source", prefix, source)
|
||||||
|
}
|
||||||
|
seenLocks[source] = struct{}{}
|
||||||
|
cfg.Locks[i].Source = source
|
||||||
|
cfg.Locks[i].Reason = strings.TrimSpace(item.Reason)
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -126,7 +126,14 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
||||||
}
|
}
|
||||||
promotions, skippedOptional, err := resolveArchivePromotions(sessionPaths, m, runtimeCatalog, env.Config.Pipeline.Archive.PromoteArtifacts)
|
promotions, skippedOptional, lockedPromotions, err := resolveArchivePromotions(
|
||||||
|
sessionPaths,
|
||||||
|
m,
|
||||||
|
runtimeCatalog,
|
||||||
|
env.Config.Pipeline.Archive.PromoteArtifacts,
|
||||||
|
env.Config.Pipeline.Archive.Locks,
|
||||||
|
sessionPrefix,
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
||||||
}
|
}
|
||||||
@@ -166,6 +173,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
promotedUploaded,
|
promotedUploaded,
|
||||||
previousUploaded,
|
previousUploaded,
|
||||||
skippedOptional,
|
skippedOptional,
|
||||||
|
lockedPromotions,
|
||||||
currentManifestKey,
|
currentManifestKey,
|
||||||
))
|
))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -204,6 +212,8 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
"previous_files_uploaded": len(previousUploaded),
|
"previous_files_uploaded": len(previousUploaded),
|
||||||
"previous_uploaded_paths": previousUploaded,
|
"previous_uploaded_paths": previousUploaded,
|
||||||
"skipped_optional_promotions": skippedOptional,
|
"skipped_optional_promotions": skippedOptional,
|
||||||
|
"locked_promotion_count": len(lockedPromotions),
|
||||||
|
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||||
"current_manifest_key": currentManifestKey,
|
"current_manifest_key": currentManifestKey,
|
||||||
"current_run_id_key": currentRunPointerKey,
|
"current_run_id_key": currentRunPointerKey,
|
||||||
"current_pointer_written": true,
|
"current_pointer_written": true,
|
||||||
@@ -220,6 +230,16 @@ type archivePromotion struct {
|
|||||||
Provenance string
|
Provenance string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type archiveLockedPromotion struct {
|
||||||
|
Source string
|
||||||
|
Dest string
|
||||||
|
RemoteKey string
|
||||||
|
Reason string
|
||||||
|
Required bool
|
||||||
|
LocalPath string
|
||||||
|
Provenance string
|
||||||
|
}
|
||||||
|
|
||||||
func archiveDisabled(env *Env) bool {
|
func archiveDisabled(env *Env) bool {
|
||||||
cfg := env.Config.Pipeline.Archive
|
cfg := env.Config.Pipeline.Archive
|
||||||
if cfg == nil {
|
if cfg == nil {
|
||||||
@@ -319,26 +339,53 @@ func resolveArchivePromotions(
|
|||||||
m *manifest.Manifest,
|
m *manifest.Manifest,
|
||||||
catalog *artifacts.ArtifactCatalog,
|
catalog *artifacts.ArtifactCatalog,
|
||||||
rules []config.ArchivePromotionRule,
|
rules []config.ArchivePromotionRule,
|
||||||
) ([]archivePromotion, []string, error) {
|
locks []config.ArchiveLockRule,
|
||||||
|
sessionPrefix string,
|
||||||
|
) ([]archivePromotion, []string, []archiveLockedPromotion, error) {
|
||||||
out := make([]archivePromotion, 0, len(rules))
|
out := make([]archivePromotion, 0, len(rules))
|
||||||
skippedOptional := make([]string, 0)
|
skippedOptional := make([]string, 0)
|
||||||
|
lockedPromotions := make([]archiveLockedPromotion, 0)
|
||||||
|
lockSet := archiveLockSet(locks)
|
||||||
for _, rule := range rules {
|
for _, rule := range rules {
|
||||||
source := strings.TrimSpace(rule.Source)
|
source := strings.TrimSpace(rule.Source)
|
||||||
required := rule.Required == nil || *rule.Required
|
required := rule.Required == nil || *rule.Required
|
||||||
dest, err := resolveArchivePromotionDest(rule, catalog)
|
dest, err := resolveArchivePromotionDest(rule, catalog)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, fmt.Errorf("source %q: %w", source, err)
|
return nil, nil, nil, fmt.Errorf("source %q: %w", source, err)
|
||||||
}
|
}
|
||||||
|
lock, locked := lockSet[source]
|
||||||
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
|
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if locked {
|
||||||
|
lockedPromotions = append(lockedPromotions, archiveLockedPromotion{
|
||||||
|
Source: source,
|
||||||
|
Dest: dest,
|
||||||
|
RemoteKey: artifacts.S3PromotedArtifactKey(sessionPrefix, dest),
|
||||||
|
Reason: strings.TrimSpace(lock.Reason),
|
||||||
|
Required: required,
|
||||||
|
})
|
||||||
|
continue
|
||||||
|
}
|
||||||
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required {
|
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required {
|
||||||
skippedOptional = append(skippedOptional, dest)
|
skippedOptional = append(skippedOptional, dest)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
|
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
|
||||||
return nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
|
return nil, nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
|
||||||
}
|
}
|
||||||
return nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
|
return nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
|
||||||
|
}
|
||||||
|
if locked {
|
||||||
|
lockedPromotions = append(lockedPromotions, archiveLockedPromotion{
|
||||||
|
Source: source,
|
||||||
|
Dest: dest,
|
||||||
|
RemoteKey: artifacts.S3PromotedArtifactKey(sessionPrefix, dest),
|
||||||
|
Reason: strings.TrimSpace(lock.Reason),
|
||||||
|
Required: required,
|
||||||
|
LocalPath: resolved.Path,
|
||||||
|
Provenance: resolved.Provenance,
|
||||||
|
})
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
out = append(out, archivePromotion{
|
out = append(out, archivePromotion{
|
||||||
Source: source,
|
Source: source,
|
||||||
@@ -348,7 +395,21 @@ func resolveArchivePromotions(
|
|||||||
Provenance: resolved.Provenance,
|
Provenance: resolved.Provenance,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
return out, skippedOptional, nil
|
return out, skippedOptional, lockedPromotions, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func archiveLockSet(locks []config.ArchiveLockRule) map[string]config.ArchiveLockRule {
|
||||||
|
out := make(map[string]config.ArchiveLockRule, 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 resolveArchivePromotionDest(rule config.ArchivePromotionRule, catalog *artifacts.ArtifactCatalog) (string, error) {
|
func resolveArchivePromotionDest(rule config.ArchivePromotionRule, catalog *artifacts.ArtifactCatalog) (string, error) {
|
||||||
@@ -663,6 +724,7 @@ func archiveMetadataPreview(
|
|||||||
promotedUploaded []string,
|
promotedUploaded []string,
|
||||||
previousUploaded []string,
|
previousUploaded []string,
|
||||||
skippedOptional []string,
|
skippedOptional []string,
|
||||||
|
lockedPromotions []archiveLockedPromotion,
|
||||||
currentManifestKey string,
|
currentManifestKey string,
|
||||||
) map[string]any {
|
) map[string]any {
|
||||||
return map[string]any{
|
return map[string]any{
|
||||||
@@ -677,9 +739,27 @@ func archiveMetadataPreview(
|
|||||||
"previous_files_uploaded": len(previousUploaded),
|
"previous_files_uploaded": len(previousUploaded),
|
||||||
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
||||||
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
||||||
|
"locked_promotion_count": len(lockedPromotions),
|
||||||
|
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||||
"current_manifest_key": currentManifestKey,
|
"current_manifest_key": currentManifestKey,
|
||||||
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
||||||
"current_pointer_written": false,
|
"current_pointer_written": false,
|
||||||
"audio_upload_skipped": true,
|
"audio_upload_skipped": true,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func lockedPromotionMetadata(locked []archiveLockedPromotion) []map[string]any {
|
||||||
|
out := make([]map[string]any, 0, len(locked))
|
||||||
|
for _, item := range locked {
|
||||||
|
out = append(out, map[string]any{
|
||||||
|
"source": item.Source,
|
||||||
|
"dest": item.Dest,
|
||||||
|
"remote_key": item.RemoteKey,
|
||||||
|
"reason": item.Reason,
|
||||||
|
"required": item.Required,
|
||||||
|
"local_path": item.LocalPath,
|
||||||
|
"provenance": item.Provenance,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package stage
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -215,6 +216,128 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) {
|
||||||
|
env, m, _ := archiveFixture(t)
|
||||||
|
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||||
|
{Source: "narratio.transcript.trimmed", Reason: "human reviewed"},
|
||||||
|
}
|
||||||
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
|
||||||
|
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
trimmedKey := m.S3SessionPrefix + "transcripts/trimmed.json"
|
||||||
|
if _, ok := fake.Objects[trimmedKey]; ok {
|
||||||
|
t.Fatalf("locked promotion key %q should not be uploaded", trimmedKey)
|
||||||
|
}
|
||||||
|
recapKey := m.S3SessionPrefix + "artifacts/session_recap.md"
|
||||||
|
if _, ok := fake.Objects[recapKey]; !ok {
|
||||||
|
t.Fatalf("unlocked promotion key %q should be uploaded", recapKey)
|
||||||
|
}
|
||||||
|
runTrimmedKey := m.S3RunPrefix + "trim/outputs/transcripts/trimmed.json"
|
||||||
|
if _, ok := fake.Objects[runTrimmedKey]; !ok {
|
||||||
|
t.Fatalf("run-local locked source output %q should still be uploaded", runTrimmedKey)
|
||||||
|
}
|
||||||
|
|
||||||
|
currentRunIDKey := m.S3SessionPrefix + "current/run_id.txt"
|
||||||
|
if len(fake.Uploads) == 0 || fake.Uploads[len(fake.Uploads)-1].Key != currentRunIDKey {
|
||||||
|
t.Fatalf("last upload = %#v, want current run pointer %q", fake.Uploads, currentRunIDKey)
|
||||||
|
}
|
||||||
|
|
||||||
|
if result.Metadata["promoted_files_uploaded"] != 1 {
|
||||||
|
t.Fatalf("metadata promoted_files_uploaded = %#v, want 1", result.Metadata["promoted_files_uploaded"])
|
||||||
|
}
|
||||||
|
if result.Metadata["locked_promotion_count"] != 1 {
|
||||||
|
t.Fatalf("metadata locked_promotion_count = %#v, want 1", result.Metadata["locked_promotion_count"])
|
||||||
|
}
|
||||||
|
locked := result.Metadata["locked_promotions"].([]map[string]any)
|
||||||
|
if len(locked) != 1 {
|
||||||
|
t.Fatalf("locked_promotions = %#v, want one item", locked)
|
||||||
|
}
|
||||||
|
if locked[0]["source"] != "narratio.transcript.trimmed" ||
|
||||||
|
locked[0]["dest"] != "transcripts/trimmed.json" ||
|
||||||
|
locked[0]["remote_key"] != trimmedKey ||
|
||||||
|
locked[0]["reason"] != "human reviewed" ||
|
||||||
|
locked[0]["required"] != true ||
|
||||||
|
locked[0]["local_path"] == "" ||
|
||||||
|
locked[0]["provenance"] == "" {
|
||||||
|
t.Fatalf("locked promotion metadata = %#v", locked[0])
|
||||||
|
}
|
||||||
|
|
||||||
|
currentManifestKey := m.S3SessionPrefix + "current/manifest.json"
|
||||||
|
var current map[string]any
|
||||||
|
if err := json.Unmarshal(fake.Objects[currentManifestKey].Data, ¤t); err != nil {
|
||||||
|
t.Fatalf("unmarshal current manifest: %v", err)
|
||||||
|
}
|
||||||
|
stages := current["stages"].(map[string]any)
|
||||||
|
archive := stages["archive"].(map[string]any)
|
||||||
|
meta := archive["metadata"].(map[string]any)
|
||||||
|
if meta["locked_promotion_count"] != float64(1) {
|
||||||
|
t.Fatalf("current manifest locked_promotion_count = %#v, want 1", meta["locked_promotion_count"])
|
||||||
|
}
|
||||||
|
items := meta["locked_promotions"].([]any)
|
||||||
|
if len(items) != 1 {
|
||||||
|
t.Fatalf("current manifest locked_promotions = %#v, want one item", items)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveLockedRequiredMissingPromotionSucceeds(t *testing.T) {
|
||||||
|
env, m, _ := archiveFixture(t)
|
||||||
|
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
|
||||||
|
{Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", Required: boolPtr(true)},
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||||
|
{Source: "narratio.transcript.merged", Reason: "manual merge is locked"},
|
||||||
|
}
|
||||||
|
|
||||||
|
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
mergedKey := m.S3SessionPrefix + "transcripts/merged.json"
|
||||||
|
if _, ok := fake.Objects[mergedKey]; ok {
|
||||||
|
t.Fatalf("locked missing promotion key %q should not be uploaded", mergedKey)
|
||||||
|
}
|
||||||
|
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; !ok {
|
||||||
|
t.Fatalf("current run pointer should be written for locked missing promotion")
|
||||||
|
}
|
||||||
|
locked := result.Metadata["locked_promotions"].([]map[string]any)
|
||||||
|
if len(locked) != 1 {
|
||||||
|
t.Fatalf("locked_promotions = %#v, want one item", locked)
|
||||||
|
}
|
||||||
|
if locked[0]["local_path"] != "" || locked[0]["provenance"] != "" {
|
||||||
|
t.Fatalf("locked missing promotion metadata = %#v, want empty local path/provenance", locked[0])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveLockDoesNotOverwriteExistingPromotion(t *testing.T) {
|
||||||
|
env, m, _ := archiveFixture(t)
|
||||||
|
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||||
|
{Source: "narratio.transcript.trimmed", Reason: "already published"},
|
||||||
|
}
|
||||||
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
trimmedKey := m.S3SessionPrefix + "transcripts/trimmed.json"
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: trimmedKey, Data: []byte("previously published\n")})
|
||||||
|
|
||||||
|
if _, err := (archiveStage{}).Run(context.Background(), env, m); err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
got := string(fake.Objects[trimmedKey].Data)
|
||||||
|
if got != "previously published\n" {
|
||||||
|
t.Fatalf("locked promotion object contents = %q, want existing object preserved", got)
|
||||||
|
}
|
||||||
|
for _, upload := range fake.Uploads {
|
||||||
|
if upload.Key == trimmedKey {
|
||||||
|
t.Fatalf("locked promotion key %q was uploaded", trimmedKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
|
func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
|
||||||
env, m, _ := archiveFixture(t)
|
env, m, _ := archiveFixture(t)
|
||||||
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
|
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
|
||||||
|
|||||||
Reference in New Issue
Block a user