Complete downstream semantic resume coverage
This commit is contained in:
@@ -207,12 +207,28 @@ runner marks the aggregate record stale and executes it. Analyze's validator
|
|||||||
can still accept a partial selection when only unrelated artifact records are
|
can still accept a partial selection when only unrelated artifact records are
|
||||||
stale.
|
stale.
|
||||||
|
|
||||||
The current production semantic contracts cover prepare, transcribe, merge,
|
Implemented reuse coverage is deliberately split between aggregate semantic
|
||||||
polish, normalize, trim, and render. They record resolved choices Narratio can
|
evidence and focused durable validators:
|
||||||
observe, not operational runner tuning. External model, module, prompt,
|
|
||||||
profile, and configuration-file contents that a tool privately loads remain
|
| Work | Reuse authority | Focused owners |
|
||||||
outside the manifest contract when their configured identifier is unchanged;
|
| --- | --- | --- |
|
||||||
operators must force the affected stage after such a private content change.
|
| prepare | aggregate semantic fingerprint | [prepare](stage-prepare.md) |
|
||||||
|
| transcribe | aggregate semantic fingerprint | [transcribe](stage-transcribe.md), [WhisperX](../integrations/whisperx.md) |
|
||||||
|
| merge | aggregate semantic fingerprint | [merge](stage-merge.md), [Seriatim](../integrations/seriatim.md) |
|
||||||
|
| polish | aggregate semantic fingerprint | [polish](stage-polish.md), [Audita](../integrations/audita.md) |
|
||||||
|
| normalize | aggregate semantic fingerprint | [normalize](stage-normalize.md), [Seriatim](../integrations/seriatim.md) |
|
||||||
|
| trim | aggregate semantic fingerprint | [trim](stage-trim.md), [Scriptorium](../integrations/scriptorium.md), [Seriatim](../integrations/seriatim.md) |
|
||||||
|
| render | aggregate semantic fingerprint | [render](stage-render.md), [Seriatim](../integrations/seriatim.md) |
|
||||||
|
| extract | aggregate semantic fingerprint plus reference/output validator | [extract](stage-extract.md), [Notarius](../integrations/notarius.md) |
|
||||||
|
| analyze artifacts | per-artifact fingerprint, reconciliation, and output validator | [analyze](stage-analyze.md), [Scriptorium](../integrations/scriptorium.md) |
|
||||||
|
| publish | aggregate semantic fingerprint plus immediate lock/commit checks | [publish](stage-publish.md), [storage adapter](adapters.md) |
|
||||||
|
| notify | aggregate delivery-mode fingerprint | [pipeline overview](overview.md), [configuration](../config.md#notifications) |
|
||||||
|
|
||||||
|
These contracts record resolved choices Narratio can observe, not operational
|
||||||
|
runner tuning. External model, module, prompt, profile, and configuration-file
|
||||||
|
contents that a tool privately loads remain outside the contract when their
|
||||||
|
configured identifier is unchanged; operators must force the affected work
|
||||||
|
after such a private content change.
|
||||||
|
|
||||||
Session manifest is the authoritative stage-progress ledger across invocations.
|
Session manifest is the authoritative stage-progress ledger across invocations.
|
||||||
Run manifest is invocation-scoped audit state.
|
Run manifest is invocation-scoped audit state.
|
||||||
@@ -222,8 +238,8 @@ comparison and invalidation rules to a cloned manifest. It predicts the runner
|
|||||||
decision without persisting session or invocation state. The shared helper
|
decision without persisting session or invocation state. The shared helper
|
||||||
hashes deterministic JSON from stage-owned typed structs; stage providers must
|
hashes deterministic JSON from stage-owned typed structs; stage providers must
|
||||||
exclude secrets, complete effective-configuration dumps, and operational
|
exclude secrets, complete effective-configuration dumps, and operational
|
||||||
values that cannot affect canonical results. Concrete stage coverage is owned
|
values that cannot affect canonical results. Concrete coverage is owned by the
|
||||||
by the focused stage documents as providers are added.
|
focused stage and integration documents linked above.
|
||||||
|
|
||||||
Before an explicitly bounded execution starts after `prepare`, the application
|
Before an explicitly bounded execution starts after `prepare`, the application
|
||||||
reads the session manifest and accepts only `succeeded` or `skipped` for every
|
reads the session manifest and accepts only `succeeded` or `skipped` for every
|
||||||
|
|||||||
@@ -68,8 +68,9 @@ The implemented canonical order is:
|
|||||||
11. `notify` (no-op)
|
11. `notify` (no-op)
|
||||||
|
|
||||||
`notify` currently has no persisted pipeline outputs and uses the explicit
|
`notify` currently has no persisted pipeline outputs and uses the explicit
|
||||||
`noop` notification mode. The focused stage documents own implementation
|
`noop` notification mode. Its versioned semantic evidence records that delivery
|
||||||
mechanics. The
|
mode and excludes adapter credentials and response data. The focused stage
|
||||||
|
documents own implementation mechanics. The
|
||||||
[CLI](../cli.md) and [Operations](../operations.md) own user-visible invocation
|
[CLI](../cli.md) and [Operations](../operations.md) own user-visible invocation
|
||||||
and execution semantics.
|
and execution semantics.
|
||||||
|
|
||||||
|
|||||||
@@ -21,8 +21,8 @@ procedures belong in [Operations](../operations.md).
|
|||||||
manifest-authoritative identity resolver before creating run-local output;
|
manifest-authoritative identity resolver before creating run-local output;
|
||||||
3. streams each verified reference into an invocation-local snapshot and
|
3. streams each verified reference into an invocation-local snapshot and
|
||||||
rejects any source change observed while copying;
|
rejects any source change observed while copying;
|
||||||
4. fingerprints the Notarius invocation contract, including sorted reference
|
4. fingerprints the byte- and provenance-bearing Notarius invocation evidence,
|
||||||
identities;
|
including sorted reference identities;
|
||||||
5. creates a run-local staging directory and invokes the injected
|
5. creates a run-local staging directory and invokes the injected
|
||||||
`notarius.Runner`;
|
`notarius.Runner`;
|
||||||
6. revalidates the reference snapshots, then validates the v2 successful
|
6. revalidates the reference snapshots, then validates the v2 successful
|
||||||
@@ -56,14 +56,20 @@ outputs is stable and does not repeatedly invalidate dependent stages.
|
|||||||
|
|
||||||
## Resume Validation
|
## Resume Validation
|
||||||
|
|
||||||
`internal/stage/extract_resume.go` permits a skip only when the existing stage
|
Before the focused validator runs, the application compares extract's versioned
|
||||||
record succeeded and still matches the current invocation fingerprint. The
|
semantic fingerprint. It covers enablement, Notarius pipeline identity, sorted
|
||||||
fingerprint covers the resolved executable and config paths, pipeline ID,
|
reference selector/source mappings, sorted declared output contracts, and each
|
||||||
timeout, working directory, sorted configured output contracts, the current
|
canonical `narratio.extraction.<key>` output identity. It excludes executable,
|
||||||
direct trimmed-transcript identity, and sorted prepared-reference identities.
|
timeout, working directory, config path, and private Notarius config contents.
|
||||||
The same reference helper and transcript identity are resolved again for
|
|
||||||
artifact evidence, so changing the current transcript bytes or producer
|
`internal/stage/extract_resume.go` then permits a skip only when the existing
|
||||||
identity makes the prior extraction obsolete.
|
stage record still matches the current byte- and provenance-bearing invocation
|
||||||
|
evidence. That evidence covers the current direct trimmed-transcript identity,
|
||||||
|
sorted prepared-reference identities, pipeline identity, and configured output
|
||||||
|
contracts. The same reference helper and transcript identity are resolved again
|
||||||
|
for artifact evidence, so changing current transcript bytes, reference bytes,
|
||||||
|
or producer identity makes the prior extraction obsolete. Operational runner
|
||||||
|
settings do not invalidate otherwise current durable evidence.
|
||||||
|
|
||||||
A valid prepared-reference change makes extraction non-resumable. Missing,
|
A valid prepared-reference change makes extraction non-resumable. Missing,
|
||||||
unsafe, or checksum-inconsistent prepared evidence is a hard validation error
|
unsafe, or checksum-inconsistent prepared evidence is a hard validation error
|
||||||
@@ -76,9 +82,10 @@ contracts and provenance, regular-file status, and stored checksums. Missing or
|
|||||||
obsolete results are non-resumable and run again; unsafe filesystem conditions
|
obsolete results are non-resumable and run again; unsafe filesystem conditions
|
||||||
return an error rather than silently accepting or replacing data.
|
return an error rather than silently accepting or replacing data.
|
||||||
|
|
||||||
The fingerprint cannot observe files imported by Notarius configuration,
|
Neither contract can observe files imported by Notarius configuration, profile
|
||||||
profile contents, prompt/module definitions, or other transitive inputs.
|
contents, prompt/module definitions, or other transitive inputs. Operators must
|
||||||
Operators must force extraction after changing any such input.
|
force extraction after changing any such private input behind a stable
|
||||||
|
identifier.
|
||||||
|
|
||||||
## Failure Behavior
|
## Failure Behavior
|
||||||
|
|
||||||
@@ -101,7 +108,8 @@ available for audit and recovery.
|
|||||||
|
|
||||||
- Stage execution, selection, and resume validation: `internal/stage/extract.go`,
|
- Stage execution, selection, and resume validation: `internal/stage/extract.go`,
|
||||||
`internal/stage/extract_resume.go`,
|
`internal/stage/extract_resume.go`,
|
||||||
`internal/stage/extract_test.go`
|
`internal/stage/extract_test.go`,
|
||||||
|
`internal/stage/semantic_contracts_delivery.go`
|
||||||
- Subprocess boundary: `internal/adapters/notarius/subprocess.go`,
|
- Subprocess boundary: `internal/adapters/notarius/subprocess.go`,
|
||||||
`internal/adapters/notarius/subprocess_test.go`
|
`internal/adapters/notarius/subprocess_test.go`
|
||||||
- Catalog hydration: `internal/artifacts/extraction_catalog.go`,
|
- Catalog hydration: `internal/artifacts/extraction_catalog.go`,
|
||||||
|
|||||||
@@ -27,8 +27,8 @@ Exact remote placement and the operator workflow belong in
|
|||||||
|
|
||||||
- when publishing or run upload is disabled, completes successfully with no
|
- when publishing or run upload is disabled, completes successfully with no
|
||||||
outputs and records explanatory metadata. This is not an explicit self-skip:
|
outputs and records explanatory metadata. This is not an explicit self-skip:
|
||||||
both manifests record success, and an ordinary later run reuses that result
|
both manifests record success. Enablement and upload policy are fingerprinted,
|
||||||
until publish is forced.
|
so changing either automatically makes the prior result non-resumable.
|
||||||
- validates prerequisite stage success and object-store availability.
|
- validates prerequisite stage success and object-store availability.
|
||||||
- derives a deterministic run-archive allowlist from the validated run
|
- derives a deterministic run-archive allowlist from the validated run
|
||||||
`manifest.json`: declared run-local outputs, logs, generated configs, and the
|
`manifest.json`: declared run-local outputs, logs, generated configs, and the
|
||||||
@@ -89,6 +89,16 @@ Includes counts/lists for:
|
|||||||
- post-commit local cleanup is authorized by the committed publish metadata and
|
- post-commit local cleanup is authorized by the committed publish metadata and
|
||||||
is durably recorded by the application lifecycle before any local deletion.
|
is durably recorded by the application lifecycle before any local deletion.
|
||||||
|
|
||||||
|
## Resume Semantics
|
||||||
|
|
||||||
|
The versioned semantic fingerprint covers enabled behavior, run-upload policy,
|
||||||
|
normalized source/destination/required output rules, static lock policy, and
|
||||||
|
the remote backend, bucket, region, endpoint, and root-prefix identity. Rule
|
||||||
|
and lock ordering is canonicalized. Credential environment names,
|
||||||
|
path-addressing transport mode, local workspace placement, and run identifiers
|
||||||
|
are excluded. Remote locks remain mutable state and are still revalidated at
|
||||||
|
the commit boundary; semantic evidence does not replace that safety check.
|
||||||
|
|
||||||
The commit boundary and cleanup gate are normative architecture invariants; see
|
The commit boundary and cleanup gate are normative architecture invariants; see
|
||||||
[Architecture](../policy/architecture.md#publish-commit-boundary).
|
[Architecture](../policy/architecture.md#publish-commit-boundary).
|
||||||
|
|
||||||
@@ -101,5 +111,7 @@ The commit boundary and cleanup gate are normative architecture invariants; see
|
|||||||
- [Artifact Internals](artifacts.md) explains source resolution and current-state
|
- [Artifact Internals](artifacts.md) explains source resolution and current-state
|
||||||
helpers.
|
helpers.
|
||||||
- Implementation and tests: `internal/stage/publish.go`,
|
- Implementation and tests: `internal/stage/publish.go`,
|
||||||
`internal/stage/publish_test.go`, `internal/app/operator_helpers_test.go`,
|
`internal/stage/publish_test.go`,
|
||||||
|
`internal/stage/semantic_contracts_delivery.go`,
|
||||||
|
`internal/app/operator_helpers_test.go`, and
|
||||||
`internal/app/post_publish_cleanup_test.go`
|
`internal/app/post_publish_cleanup_test.go`
|
||||||
|
|||||||
@@ -99,7 +99,9 @@ Execution rules:
|
|||||||
repeated self-skip with the same reason and no outputs is stable and does not
|
repeated self-skip with the same reason and no outputs is stable and does not
|
||||||
perpetually rerun dependent work.
|
perpetually rerun dependent work.
|
||||||
|
|
||||||
Prepare through render currently provide semantic-configuration evidence.
|
Every aggregate stage except analyze currently provides semantic-configuration
|
||||||
|
evidence; analyze retains its more precise per-artifact fingerprints and
|
||||||
|
validator.
|
||||||
Changing prepare selection semantics reruns all fixed descendants; changing
|
Changing prepare selection semantics reruns all fixed descendants; changing
|
||||||
WhisperX language/service identity reuses prepare; changing a Seriatim merge
|
WhisperX language/service identity reuses prepare; changing a Seriatim merge
|
||||||
transformation reuses prepare and transcribe; and changing an Audita model
|
transformation reuses prepare and transcribe; and changing an Audita model
|
||||||
@@ -297,6 +299,10 @@ narratio run-stage extract 2026-04-04 --force
|
|||||||
|
|
||||||
Narratio automatically reruns extraction when its recorded invocation contract,
|
Narratio automatically reruns extraction when its recorded invocation contract,
|
||||||
prepared Narratio reference identities, or durable output validation changes.
|
prepared Narratio reference identities, or durable output validation changes.
|
||||||
|
The semantic portion covers Notarius enablement, pipeline identity, declared
|
||||||
|
reference mapping, and output contracts. Executable, timeout, working directory,
|
||||||
|
and private config-file paths are operational and do not invalidate a current
|
||||||
|
result.
|
||||||
It cannot fingerprint configuration files, profiles, prompts, modules, or
|
It cannot fingerprint configuration files, profiles, prompts, modules, or
|
||||||
other references loaded transitively by Notarius itself. Force extraction after
|
other references loaded transitively by Notarius itself. Force extraction after
|
||||||
changing any of those inputs, even when the top-level Narratio and Notarius
|
changing any of those inputs, even when the top-level Narratio and Notarius
|
||||||
@@ -305,6 +311,12 @@ marks successful downstream stages stale. Ordinary extraction failures or
|
|||||||
outcome changes also stale affected downstream stages, while an identical
|
outcome changes also stale affected downstream stages, while an identical
|
||||||
repeated `notarius_disabled` self-skip does not repeatedly invalidate them.
|
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.
|
||||||
|
|
||||||
## Publish Workflow
|
## Publish Workflow
|
||||||
|
|
||||||
Run publish only:
|
Run publish only:
|
||||||
|
|||||||
@@ -392,7 +392,7 @@ prompt changes without invalidating unaffected upstream transcript work.
|
|||||||
|
|
||||||
## Stage 6 — Extract, Analyze, Publish, And Notify Resume Audit
|
## Stage 6 — Extract, Analyze, Publish, And Notify Resume Audit
|
||||||
|
|
||||||
**Status: Pending**
|
**Status: Completed**
|
||||||
|
|
||||||
### Goal
|
### Goal
|
||||||
|
|
||||||
|
|||||||
@@ -61,7 +61,7 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) {
|
|||||||
m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
|
m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
|
||||||
}
|
}
|
||||||
m.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled")
|
m.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled")
|
||||||
seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), m, "prepare", "transcribe", "merge", "polish", "normalize", "trim", "render")
|
seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), m, "prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "publish", "notify")
|
||||||
if err := store.Save(context.Background(), manifestPath, m); err != nil {
|
if err := store.Save(context.Background(), manifestPath, m); err != nil {
|
||||||
t.Fatalf("save manifest: %v", err)
|
t.Fatalf("save manifest: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -435,6 +435,122 @@ func TestRefinementSemanticChangesRespectDependencyBranches(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDeliverySemanticChangesInvalidateOnlyTheirFixedDependents(t *testing.T) {
|
||||||
|
t.Run("extract contract", func(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
configureDeliverySemanticConfig(cfg)
|
||||||
|
provider := canonicalSemanticProvider(t, "extract")
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
||||||
|
for _, name := range []string{"render", "extract", "analyze", "publish", "notify"} {
|
||||||
|
seed.MarkStageSucceeded(name, time.Now().UTC(), nil)
|
||||||
|
}
|
||||||
|
fingerprint, err := provider.SemanticConfigFingerprint(&stage.Env{Config: cfg})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
seed.Stages["extract"].SemanticConfig = &fingerprint
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
cfg.Pipeline.Notarius.PipelineID = "session-v2"
|
||||||
|
runs := 0
|
||||||
|
selected := semanticContractRunStub{name: "extract", provider: provider, runs: &runs}
|
||||||
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
loaded, err := store.Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if runs != 1 || loaded.Stages["render"].Status != manifest.StatusSucceeded {
|
||||||
|
t.Fatalf("runs=%d render=%q", runs, loaded.Stages["render"].Status)
|
||||||
|
}
|
||||||
|
for _, name := range []string{"analyze", "publish", "notify"} {
|
||||||
|
if loaded.Stages[name].Status != manifest.StatusStale {
|
||||||
|
t.Fatalf("%s status = %q, want stale", name, loaded.Stages[name].Status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mutate func(*config.Config)
|
||||||
|
wantRuns [2]int
|
||||||
|
}{
|
||||||
|
{name: "publish destination", mutate: func(cfg *config.Config) {
|
||||||
|
cfg.Pipeline.Publish.Outputs[0].Dest = "published/alternate.json"
|
||||||
|
}, wantRuns: [2]int{1, 1}},
|
||||||
|
{name: "publish credentials", mutate: func(cfg *config.Config) {
|
||||||
|
cfg.Pipeline.Storage.S3.AccessKeyIDEnv = "OTHER_ACCESS_KEY"
|
||||||
|
cfg.Pipeline.Storage.S3.SecretKeyEnv = "OTHER_SECRET_KEY"
|
||||||
|
}, wantRuns: [2]int{0, 0}},
|
||||||
|
{name: "notify mode", mutate: func(cfg *config.Config) {
|
||||||
|
cfg.Pipeline.Notification.Mode = "webhook"
|
||||||
|
}, wantRuns: [2]int{0, 1}},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
configureDeliverySemanticConfig(cfg)
|
||||||
|
names := []string{"publish", "notify"}
|
||||||
|
providers := []stage.SemanticConfigFingerprinter{
|
||||||
|
canonicalSemanticProvider(t, names[0]), canonicalSemanticProvider(t, names[1]),
|
||||||
|
}
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
||||||
|
for index, name := range names {
|
||||||
|
seed.MarkStageSucceeded(name, time.Now().UTC(), nil)
|
||||||
|
fingerprint, err := providers[index].SemanticConfigFingerprint(&stage.Env{Config: cfg})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
seed.Stages[name].SemanticConfig = &fingerprint
|
||||||
|
}
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
test.mutate(cfg)
|
||||||
|
runs := [2]int{}
|
||||||
|
selected := []stage.Stage{
|
||||||
|
semanticContractRunStub{name: names[0], provider: providers[0], runs: &runs[0]},
|
||||||
|
semanticContractRunStub{name: names[1], provider: providers[1], runs: &runs[1]},
|
||||||
|
}
|
||||||
|
if _, err := executeStages(context.Background(), cfg, selected, RunOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if runs != test.wantRuns {
|
||||||
|
t.Fatalf("runs = %v, want %v", runs, test.wantRuns)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func configureDeliverySemanticConfig(cfg *config.Config) {
|
||||||
|
enabled := true
|
||||||
|
disabled := false
|
||||||
|
cfg.Pipeline.Notarius = &config.NotariusConfig{
|
||||||
|
Enabled: true, PipelineID: "session",
|
||||||
|
References: map[string]string{"party": "narratio.input.party"},
|
||||||
|
Outputs: map[string]config.NotariusOutputConfig{
|
||||||
|
"encounters": {LaneID: "encounters", MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
cfg.Pipeline.Publish = &config.PublishConfig{
|
||||||
|
Enabled: &disabled, UploadRun: &enabled,
|
||||||
|
Outputs: []config.PublishOutputRule{{
|
||||||
|
Source: "narratio.transcript.final_trimmed", Dest: "transcripts/final.trimmed.json", Required: &enabled,
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
cfg.Pipeline.Storage = config.StorageConfig{
|
||||||
|
Backend: config.StorageBackendLocal,
|
||||||
|
S3: &config.StorageS3Config{
|
||||||
|
Bucket: "campaign", RootPrefix: "narratio", Region: "us-east-1",
|
||||||
|
Endpoint: "https://objects.example", AccessKeyIDEnv: "ACCESS_KEY", SecretKeyEnv: "SECRET_KEY",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
cfg.Pipeline.Notification.Mode = config.DefaultNotificationMode
|
||||||
|
}
|
||||||
|
|
||||||
func canonicalSemanticProvider(t *testing.T, name string) stage.SemanticConfigFingerprinter {
|
func canonicalSemanticProvider(t *testing.T, name string) stage.SemanticConfigFingerprinter {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
for _, candidate := range stage.All() {
|
for _, candidate := range stage.All() {
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"reflect"
|
"reflect"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -109,6 +110,86 @@ func TestAnalyzeResumeValidationRejectsChangedInputsTamperedOutputsAndLegacyStat
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAnalyzeResumeSemanticChangesRemainArtifactGranular(t *testing.T) {
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mutate func(*config.ScriptoriumConfig)
|
||||||
|
wantWork []string
|
||||||
|
}{
|
||||||
|
{name: "profile selected model", mutate: func(cfg *config.ScriptoriumConfig) {
|
||||||
|
artifact := cfg.Artifacts["player_handout"]
|
||||||
|
artifact.ProfileID = "production-model"
|
||||||
|
cfg.Artifacts["player_handout"] = artifact
|
||||||
|
}, wantWork: []string{"player_handout", "campaign_brief"}},
|
||||||
|
{name: "variable", mutate: func(cfg *config.ScriptoriumConfig) {
|
||||||
|
artifact := cfg.Artifacts["player_handout"]
|
||||||
|
artifact.Vars["audience"] = "players"
|
||||||
|
cfg.Artifacts["player_handout"] = artifact
|
||||||
|
}, wantWork: []string{"player_handout", "campaign_brief"}},
|
||||||
|
{name: "dependency", mutate: func(cfg *config.ScriptoriumConfig) {
|
||||||
|
artifact := cfg.Artifacts["player_handout"]
|
||||||
|
artifact.DependsOn = nil
|
||||||
|
cfg.Artifacts["player_handout"] = artifact
|
||||||
|
}, wantWork: []string{"player_handout", "campaign_brief"}},
|
||||||
|
{name: "input", mutate: func(cfg *config.ScriptoriumConfig) {
|
||||||
|
artifact := cfg.Artifacts["player_handout"]
|
||||||
|
artifact.Inputs["recap"] = config.ScriptoriumInputConfig{
|
||||||
|
Source: "narratio.transcript.final_trimmed", Required: true,
|
||||||
|
}
|
||||||
|
cfg.Artifacts["player_handout"] = artifact
|
||||||
|
}, wantWork: []string{"player_handout", "campaign_brief"}},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
env, m, _ := setupAnalyzeEnv(t)
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`)
|
||||||
|
addAnalyzeDependentArtifact(env)
|
||||||
|
env.Config.Pipeline.Scriptorium.Artifacts["player_handout"] = config.ScriptoriumArtifactConfig{
|
||||||
|
Enabled: true, DependsOn: []string{"session_recap"}, PromptID: "dnd.player_handout",
|
||||||
|
ProfileID: "testing-model", OutputPath: "artifacts/player_handout.md",
|
||||||
|
Inputs: map[string]config.ScriptoriumInputConfig{
|
||||||
|
"recap": {Source: "narratio.artifact.session_recap", Required: true},
|
||||||
|
},
|
||||||
|
Vars: map[string]any{"audience": "party"},
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Scriptorium.Artifacts["campaign_brief"] = config.ScriptoriumArtifactConfig{
|
||||||
|
Enabled: true, DependsOn: []string{"player_handout"}, PromptID: "dnd.campaign_brief",
|
||||||
|
OutputPath: "artifacts/campaign_brief.md",
|
||||||
|
Inputs: map[string]config.ScriptoriumInputConfig{
|
||||||
|
"handout": {Source: "narratio.artifact.player_handout", Required: true},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Scriptorium.Artifacts["unrelated"] = config.ScriptoriumArtifactConfig{
|
||||||
|
Enabled: true, PromptID: "dnd.unrelated", OutputPath: "artifacts/unrelated.md",
|
||||||
|
Inputs: map[string]config.ScriptoriumInputConfig{
|
||||||
|
"transcript": {Source: "narratio.transcript.final_trimmed", Required: true},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
result, err := (analyzeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
installAnalyzeProjection(m, result.AnalyzeState)
|
||||||
|
m.Stages["analyze"].Status = manifest.StatusSucceeded
|
||||||
|
|
||||||
|
test.mutate(env.Config.Pipeline.Scriptorium)
|
||||||
|
validation, err := (analyzeStage{}).ValidateResume(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if validation.Resumable || !resumeArtifactKeysEqual(validation.Analyze.ExecutionOrder, test.wantWork) {
|
||||||
|
t.Fatalf("validation = %#v, want only %v", validation, test.wantWork)
|
||||||
|
}
|
||||||
|
for _, reused := range validation.Analyze.ReusedCurrent {
|
||||||
|
if reused.Key == "unrelated" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t.Fatal("unrelated artifact was not preserved as current")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAnalyzeStaleAggregateRestoresSuccessWithoutAdapterWork(t *testing.T) {
|
func TestAnalyzeStaleAggregateRestoresSuccessWithoutAdapterWork(t *testing.T) {
|
||||||
env, m, _ := currentAnalyzeFixture(t, false)
|
env, m, _ := currentAnalyzeFixture(t, false)
|
||||||
m.Stages["analyze"].Status = manifest.StatusStale
|
m.Stages["analyze"].Status = manifest.StatusStale
|
||||||
|
|||||||
@@ -141,7 +141,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
return nil, fmt.Errorf("extract: materialize Notarius reference snapshots: %w", err)
|
return nil, fmt.Errorf("extract: materialize Notarius reference snapshots: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory, input, references.Identities)
|
fingerprint, err := extractionFingerprint(notariusConfig, input, references.Identities)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
|
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
|
||||||
}
|
}
|
||||||
@@ -394,21 +394,14 @@ type fingerprintOutput struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type fingerprintDocument struct {
|
type fingerprintDocument struct {
|
||||||
Binary string `json:"binary"`
|
|
||||||
ConfigPath string `json:"config_path"`
|
|
||||||
PipelineID string `json:"pipeline_id"`
|
PipelineID string `json:"pipeline_id"`
|
||||||
Timeout string `json:"timeout"`
|
|
||||||
WorkingDirectory string `json:"working_directory"`
|
|
||||||
Input artifacts.ExtractionInputIdentity `json:"input"`
|
Input artifacts.ExtractionInputIdentity `json:"input"`
|
||||||
References []extractReferenceIdentity `json:"references"`
|
References []extractReferenceIdentity `json:"references"`
|
||||||
Outputs []fingerprintOutput `json:"outputs"`
|
Outputs []fingerprintOutput `json:"outputs"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func extractionFingerprint(
|
func extractionFingerprint(
|
||||||
binary, configPath string,
|
|
||||||
cfg *config.NotariusConfig,
|
cfg *config.NotariusConfig,
|
||||||
timeout time.Duration,
|
|
||||||
workingDirectory string,
|
|
||||||
input artifacts.ExtractionInputIdentity,
|
input artifacts.ExtractionInputIdentity,
|
||||||
references []extractReferenceIdentity,
|
references []extractReferenceIdentity,
|
||||||
) (string, error) {
|
) (string, error) {
|
||||||
@@ -436,9 +429,7 @@ func extractionFingerprint(
|
|||||||
return sortedReferences[i].Path < sortedReferences[j].Path
|
return sortedReferences[i].Path < sortedReferences[j].Path
|
||||||
})
|
})
|
||||||
payload, err := json.Marshal(fingerprintDocument{
|
payload, err := json.Marshal(fingerprintDocument{
|
||||||
Binary: binary, ConfigPath: configPath, PipelineID: cfg.PipelineID,
|
PipelineID: cfg.PipelineID, Input: input, References: sortedReferences, Outputs: outputs,
|
||||||
Timeout: timeout.String(), WorkingDirectory: workingDirectory, Input: input,
|
|
||||||
References: sortedReferences, Outputs: outputs,
|
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
|
||||||
|
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
@@ -21,22 +20,6 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
|
|||||||
if m == nil {
|
if m == nil {
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: session manifest is required")
|
return ResumeValidation{}, fmt.Errorf("extract resume: session manifest is required")
|
||||||
}
|
}
|
||||||
timeout, err := time.ParseDuration(strings.TrimSpace(cfg.Timeout))
|
|
||||||
if err != nil || timeout <= 0 {
|
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: invalid Notarius timeout %q", cfg.Timeout)
|
|
||||||
}
|
|
||||||
resolvedBinary, err := resolveExecutable(cfg.Binary)
|
|
||||||
if err != nil {
|
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius binary: %w", err)
|
|
||||||
}
|
|
||||||
configPath, err := absolutePath(cfg.ConfigPath)
|
|
||||||
if err != nil {
|
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius config path: %w", err)
|
|
||||||
}
|
|
||||||
workingDirectory, err := absolutePath(cfg.WorkingDirectory)
|
|
||||||
if err != nil {
|
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius working directory: %w", err)
|
|
||||||
}
|
|
||||||
sessionID := strings.TrimSpace(m.SessionID)
|
sessionID := strings.TrimSpace(m.SessionID)
|
||||||
if sessionID == "" {
|
if sessionID == "" {
|
||||||
sessionID = strings.TrimSpace(env.Config.Session.SessionID)
|
sessionID = strings.TrimSpace(env.Config.Session.SessionID)
|
||||||
@@ -53,7 +36,7 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius references: %w", err)
|
return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius references: %w", err)
|
||||||
}
|
}
|
||||||
fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory, input, references.Identities)
|
fingerprint, err := extractionFingerprint(cfg, input, references.Identities)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err)
|
return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -530,11 +530,11 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
|
|||||||
{Selector: "zeta", SourceID: artifactpolicy.SourceInputPlayers, Path: "inputs/players.yml", Checksum: "players-checksum", SizeBytes: 12},
|
{Selector: "zeta", SourceID: artifactpolicy.SourceInputPlayers, Path: "inputs/players.yml", Checksum: "players-checksum", SizeBytes: 12},
|
||||||
{Selector: "alpha", SourceID: artifactpolicy.SourceInputParty, Path: "inputs/party.yml", Checksum: "party-checksum", SizeBytes: 34},
|
{Selector: "alpha", SourceID: artifactpolicy.SourceInputParty, Path: "inputs/party.yml", Checksum: "party-checksum", SizeBytes: 34},
|
||||||
}
|
}
|
||||||
one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work", input, references)
|
one, err := extractionFingerprint(first, input, references)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("extractionFingerprint(first) error = %v", err)
|
t.Fatalf("extractionFingerprint(first) error = %v", err)
|
||||||
}
|
}
|
||||||
two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input, []extractReferenceIdentity{references[1], references[0]})
|
two, err := extractionFingerprint(second, input, []extractReferenceIdentity{references[1], references[0]})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("extractionFingerprint(second) error = %v", err)
|
t.Fatalf("extractionFingerprint(second) error = %v", err)
|
||||||
}
|
}
|
||||||
@@ -554,7 +554,7 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
|
|||||||
t.Run(test.name, func(t *testing.T) {
|
t.Run(test.name, func(t *testing.T) {
|
||||||
changedInput := input
|
changedInput := input
|
||||||
test.mutate(&changedInput)
|
test.mutate(&changedInput)
|
||||||
changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", changedInput, references)
|
changed, err := extractionFingerprint(second, changedInput, references)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("extractionFingerprint(changed input) error = %v", err)
|
t.Fatalf("extractionFingerprint(changed input) error = %v", err)
|
||||||
}
|
}
|
||||||
@@ -576,7 +576,7 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
|
|||||||
t.Run("reference "+test.name, func(t *testing.T) {
|
t.Run("reference "+test.name, func(t *testing.T) {
|
||||||
changedReferences := append([]extractReferenceIdentity(nil), references...)
|
changedReferences := append([]extractReferenceIdentity(nil), references...)
|
||||||
test.mutate(&changedReferences[0])
|
test.mutate(&changedReferences[0])
|
||||||
changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input, changedReferences)
|
changed, err := extractionFingerprint(second, input, changedReferences)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("extractionFingerprint(changed reference) error = %v", err)
|
t.Fatalf("extractionFingerprint(changed reference) error = %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
217
internal/stage/semantic_contracts_delivery.go
Normal file
217
internal/stage/semantic_contracts_delivery.go
Normal file
@@ -0,0 +1,217 @@
|
|||||||
|
package stage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/notariusref"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
extractSemanticConfigVersion = 1
|
||||||
|
publishSemanticConfigVersion = 1
|
||||||
|
notifySemanticConfigVersion = 1
|
||||||
|
)
|
||||||
|
|
||||||
|
type extractSemanticConfig struct {
|
||||||
|
Enabled bool `json:"enabled"`
|
||||||
|
PipelineID string `json:"pipeline_id,omitempty"`
|
||||||
|
References []extractSemanticReference `json:"references"`
|
||||||
|
Outputs []extractSemanticOutput `json:"outputs"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type extractSemanticReference struct {
|
||||||
|
Selector string `json:"selector"`
|
||||||
|
SourceID string `json:"source_id"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type extractSemanticOutput struct {
|
||||||
|
Key string `json:"key"`
|
||||||
|
SourceID string `json:"source_id"`
|
||||||
|
LaneID string `json:"lane_id"`
|
||||||
|
MediaType string `json:"media_type"`
|
||||||
|
SchemaID string `json:"schema_id"`
|
||||||
|
SchemaVersion string `json:"schema_version"`
|
||||||
|
ModuleKey string `json:"module_key,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
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"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type publishSemanticRule struct {
|
||||||
|
Source string `json:"source"`
|
||||||
|
Dest string `json:"dest"`
|
||||||
|
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"`
|
||||||
|
Region string `json:"region,omitempty"`
|
||||||
|
Endpoint string `json:"endpoint,omitempty"`
|
||||||
|
RootPrefix string `json:"root_prefix,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type notifySemanticConfig struct {
|
||||||
|
Mode string `json:"mode"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (extractStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) {
|
||||||
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("extract semantic configuration requires resolved pipeline config")
|
||||||
|
}
|
||||||
|
cfg := env.Config.Pipeline.Notarius
|
||||||
|
payload := extractSemanticConfig{References: []extractSemanticReference{}, Outputs: []extractSemanticOutput{}}
|
||||||
|
if cfg == nil || !cfg.Enabled {
|
||||||
|
return FingerprintSemanticConfig(extractSemanticConfigVersion, payload)
|
||||||
|
}
|
||||||
|
payload.Enabled = true
|
||||||
|
payload.PipelineID = strings.TrimSpace(cfg.PipelineID)
|
||||||
|
for rawSelector, rawSourceID := range cfg.References {
|
||||||
|
selector, err := notariusref.NormalizeSelector(rawSelector)
|
||||||
|
if err != nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("extract semantic reference selector %q: %w", rawSelector, err)
|
||||||
|
}
|
||||||
|
payload.References = append(payload.References, extractSemanticReference{
|
||||||
|
Selector: selector,
|
||||||
|
SourceID: strings.TrimSpace(rawSourceID),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
sort.Slice(payload.References, func(i, j int) bool {
|
||||||
|
if payload.References[i].Selector != payload.References[j].Selector {
|
||||||
|
return payload.References[i].Selector < payload.References[j].Selector
|
||||||
|
}
|
||||||
|
return payload.References[i].SourceID < payload.References[j].SourceID
|
||||||
|
})
|
||||||
|
keys := make([]string, 0, len(cfg.Outputs))
|
||||||
|
for key := range cfg.Outputs {
|
||||||
|
keys = append(keys, key)
|
||||||
|
}
|
||||||
|
sort.Strings(keys)
|
||||||
|
for _, key := range keys {
|
||||||
|
output := cfg.Outputs[key]
|
||||||
|
payload.Outputs = append(payload.Outputs, extractSemanticOutput{
|
||||||
|
Key: strings.TrimSpace(key), SourceID: artifacts.ExtractionArtifactSourceID(strings.TrimSpace(key)),
|
||||||
|
LaneID: strings.TrimSpace(output.LaneID), MediaType: strings.TrimSpace(output.MediaType),
|
||||||
|
SchemaID: strings.TrimSpace(output.SchemaID), SchemaVersion: strings.TrimSpace(output.SchemaVersion),
|
||||||
|
ModuleKey: strings.TrimSpace(output.ModuleKey),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return FingerprintSemanticConfig(extractSemanticConfigVersion, payload)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (publishStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) {
|
||||||
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("publish semantic configuration requires resolved pipeline config")
|
||||||
|
}
|
||||||
|
pipeline := env.Config.Pipeline
|
||||||
|
payload := publishSemanticConfig{
|
||||||
|
Rules: []publishSemanticRule{}, StaticLocks: []publishSemanticLock{},
|
||||||
|
Destination: publishSemanticDestination{Backend: strings.ToLower(strings.TrimSpace(pipeline.Storage.Backend))},
|
||||||
|
}
|
||||||
|
if payload.Destination.Backend == "" {
|
||||||
|
payload.Destination.Backend = config.StorageBackendLocal
|
||||||
|
}
|
||||||
|
if pipeline.Storage.S3 != nil {
|
||||||
|
payload.Destination.Bucket = strings.TrimSpace(pipeline.Storage.S3.Bucket)
|
||||||
|
payload.Destination.Region = strings.TrimSpace(pipeline.Storage.S3.Region)
|
||||||
|
payload.Destination.Endpoint = strings.TrimSpace(pipeline.Storage.S3.Endpoint)
|
||||||
|
payload.Destination.RootPrefix = strings.Trim(strings.TrimSpace(pipeline.Storage.S3.RootPrefix), "/")
|
||||||
|
}
|
||||||
|
cfg := pipeline.Publish
|
||||||
|
if cfg == nil {
|
||||||
|
return FingerprintSemanticConfig(publishSemanticConfigVersion, payload)
|
||||||
|
}
|
||||||
|
payload.Enabled = cfg.Enabled == nil || *cfg.Enabled
|
||||||
|
payload.UploadRun = cfg.UploadRun == nil || *cfg.UploadRun
|
||||||
|
configured, extractions := publishSemanticSources(pipeline)
|
||||||
|
for _, rule := range cfg.Outputs {
|
||||||
|
source, err := artifactpolicy.ValidatePublishSourceWithExtractions(rule.Source, configured, extractions)
|
||||||
|
if err != nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("publish semantic source %q: %w", rule.Source, err)
|
||||||
|
}
|
||||||
|
dest, err := artifactpolicy.ResolvePublishedDestinationWithExtractions(
|
||||||
|
source.ID, rule.Dest, configured, extractions,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("publish semantic destination for %q: %w", source.ID, err)
|
||||||
|
}
|
||||||
|
dest, err = pathsafe.NormalizeRelativeDestination(dest)
|
||||||
|
if err != nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("publish semantic destination for %q: %w", source.ID, err)
|
||||||
|
}
|
||||||
|
payload.Rules = append(payload.Rules, publishSemanticRule{
|
||||||
|
Source: source.ID, Dest: dest, Required: rule.Required == nil || *rule.Required,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
sort.Slice(payload.Rules, func(i, j int) bool {
|
||||||
|
if payload.Rules[i].Source != payload.Rules[j].Source {
|
||||||
|
return payload.Rules[i].Source < payload.Rules[j].Source
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s placeholderStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) {
|
||||||
|
if s.name != "notify" {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("stage %q has no semantic configuration contract", s.name)
|
||||||
|
}
|
||||||
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil {
|
||||||
|
return manifest.SemanticConfigFingerprint{}, fmt.Errorf("notify semantic configuration requires resolved pipeline config")
|
||||||
|
}
|
||||||
|
mode := strings.ToLower(strings.TrimSpace(env.Config.Pipeline.Notification.Mode))
|
||||||
|
if mode == "" {
|
||||||
|
mode = config.DefaultNotificationMode
|
||||||
|
}
|
||||||
|
return FingerprintSemanticConfig(notifySemanticConfigVersion, notifySemanticConfig{Mode: mode})
|
||||||
|
}
|
||||||
|
|
||||||
|
func publishSemanticSources(pipeline *config.PipelineConfig) (map[string]string, map[string]struct{}) {
|
||||||
|
configured := map[string]string{}
|
||||||
|
if pipeline.Scriptorium != nil {
|
||||||
|
for key, artifact := range pipeline.Scriptorium.Artifacts {
|
||||||
|
configured[key] = artifact.OutputPath
|
||||||
|
}
|
||||||
|
}
|
||||||
|
extractions := map[string]struct{}{}
|
||||||
|
if pipeline.Notarius != nil {
|
||||||
|
for key := range pipeline.Notarius.Outputs {
|
||||||
|
extractions[key] = struct{}{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return configured, extractions
|
||||||
|
}
|
||||||
|
|
||||||
|
var _ SemanticConfigFingerprinter = extractStage{}
|
||||||
|
var _ SemanticConfigFingerprinter = publishStage{}
|
||||||
|
var _ SemanticConfigFingerprinter = placeholderStage{name: "notify"}
|
||||||
233
internal/stage/semantic_contracts_delivery_test.go
Normal file
233
internal/stage/semantic_contracts_delivery_test.go
Normal file
@@ -0,0 +1,233 @@
|
|||||||
|
package stage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestExtractSemanticConfigSensitivityAndOperationalExclusions(t *testing.T) {
|
||||||
|
baseline := downstreamSemanticEnv()
|
||||||
|
want, err := (extractStage{}).SemanticConfigFingerprint(baseline)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mutate func(*Env)
|
||||||
|
}{
|
||||||
|
{name: "enablement", mutate: func(env *Env) { env.Config.Pipeline.Notarius.Enabled = false }},
|
||||||
|
{name: "pipeline", mutate: func(env *Env) { env.Config.Pipeline.Notarius.PipelineID = "campaign-v2" }},
|
||||||
|
{name: "reference selector", mutate: func(env *Env) {
|
||||||
|
env.Config.Pipeline.Notarius.References = map[string]string{"party_v2": artifactpolicy.SourceInputParty}
|
||||||
|
}},
|
||||||
|
{name: "reference source", mutate: func(env *Env) {
|
||||||
|
env.Config.Pipeline.Notarius.References = map[string]string{"party": artifactpolicy.SourceInputPlayers}
|
||||||
|
}},
|
||||||
|
{name: "lane", mutate: func(env *Env) {
|
||||||
|
output := env.Config.Pipeline.Notarius.Outputs["encounters"]
|
||||||
|
output.LaneID = "encounters-v2"
|
||||||
|
env.Config.Pipeline.Notarius.Outputs["encounters"] = output
|
||||||
|
}},
|
||||||
|
{name: "contract", mutate: func(env *Env) {
|
||||||
|
output := env.Config.Pipeline.Notarius.Outputs["encounters"]
|
||||||
|
output.SchemaVersion = "2"
|
||||||
|
env.Config.Pipeline.Notarius.Outputs["encounters"] = output
|
||||||
|
}},
|
||||||
|
{name: "output identity", mutate: func(env *Env) {
|
||||||
|
env.Config.Pipeline.Notarius.Outputs["characters"] = config.NotariusOutputConfig{
|
||||||
|
LaneID: "characters", MediaType: "application/json", SchemaID: "characters", SchemaVersion: "1",
|
||||||
|
}
|
||||||
|
}},
|
||||||
|
} {
|
||||||
|
t.Run("semantic "+test.name, func(t *testing.T) {
|
||||||
|
env := downstreamSemanticEnv()
|
||||||
|
test.mutate(env)
|
||||||
|
got, err := (extractStage{}).SemanticConfigFingerprint(env)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if want.Equal(got) {
|
||||||
|
t.Fatalf("semantic change %q retained fingerprint", test.name)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
operational := downstreamSemanticEnv()
|
||||||
|
operational.Config.Pipeline.Notarius.Binary = "/opt/notarius-v2"
|
||||||
|
operational.Config.Pipeline.Notarius.Timeout = "2h"
|
||||||
|
operational.Config.Pipeline.Notarius.WorkingDirectory = "/var/tmp/notarius"
|
||||||
|
operational.Config.Pipeline.Notarius.ConfigPath = "/etc/notarius/private-v2.yml"
|
||||||
|
operational.Config.Pipeline.Workspace.Root = "/relocated/workspace"
|
||||||
|
got, err := (extractStage{}).SemanticConfigFingerprint(operational)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !want.Equal(got) {
|
||||||
|
t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
|
||||||
|
env, m, _ := setupExtractEnv(t)
|
||||||
|
seedSucceededExtractResult(t, env, m)
|
||||||
|
env.Config.Pipeline.Notarius.Binary = "/not/installed"
|
||||||
|
env.Config.Pipeline.Notarius.ConfigPath = "/not/present/private.yml"
|
||||||
|
env.Config.Pipeline.Notarius.Timeout = "not-a-duration"
|
||||||
|
env.Config.Pipeline.Notarius.WorkingDirectory = "/not/present/work"
|
||||||
|
|
||||||
|
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !validation.Resumable {
|
||||||
|
t.Fatalf("operational change validation = %#v, want reusable durable evidence", validation)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPublishSemanticConfigSensitivityAndOperationalExclusions(t *testing.T) {
|
||||||
|
want, err := (publishStage{}).SemanticConfigFingerprint(downstreamSemanticEnv())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mutate func(*Env)
|
||||||
|
}{
|
||||||
|
{name: "enablement", mutate: func(env *Env) { env.Config.Pipeline.Publish.Enabled = boolPointer(false) }},
|
||||||
|
{name: "run upload", mutate: func(env *Env) { env.Config.Pipeline.Publish.UploadRun = boolPointer(false) }},
|
||||||
|
{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" }},
|
||||||
|
{name: "endpoint", mutate: func(env *Env) { env.Config.Pipeline.Storage.S3.Endpoint = "https://objects-v2.example" }},
|
||||||
|
{name: "root prefix", mutate: func(env *Env) { env.Config.Pipeline.Storage.S3.RootPrefix = "archive-v2" }},
|
||||||
|
} {
|
||||||
|
t.Run("semantic "+test.name, func(t *testing.T) {
|
||||||
|
env := downstreamSemanticEnv()
|
||||||
|
test.mutate(env)
|
||||||
|
got, err := (publishStage{}).SemanticConfigFingerprint(env)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if want.Equal(got) {
|
||||||
|
t.Fatalf("semantic change %q retained fingerprint", test.name)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
operational := downstreamSemanticEnv()
|
||||||
|
operational.Config.Pipeline.Storage.S3.AccessKeyIDEnv = "OTHER_ACCESS_KEY"
|
||||||
|
operational.Config.Pipeline.Storage.S3.SecretKeyEnv = "OTHER_SECRET_KEY"
|
||||||
|
operational.Config.Pipeline.Storage.S3.ForcePathStyle = true
|
||||||
|
operational.Config.Pipeline.Workspace.Root = "/relocated/workspace"
|
||||||
|
got, err := (publishStage{}).SemanticConfigFingerprint(operational)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !want.Equal(got) {
|
||||||
|
t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeliverySemanticConfigCanonicalOrderingAndNotifyMode(t *testing.T) {
|
||||||
|
first := downstreamSemanticEnv()
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
secondPublish, err := (publishStage{}).SemanticConfigFingerprint(second)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !firstPublish.Equal(secondPublish) {
|
||||||
|
t.Fatalf("publish ordering changed fingerprint: first=%#v second=%#v", firstPublish, secondPublish)
|
||||||
|
}
|
||||||
|
|
||||||
|
defaultNotify := downstreamSemanticEnv()
|
||||||
|
defaultNotify.Config.Pipeline.Notification.Mode = ""
|
||||||
|
explicitNotify := downstreamSemanticEnv()
|
||||||
|
explicitNotify.Config.Pipeline.Notification.Mode = config.DefaultNotificationMode
|
||||||
|
defaultFingerprint, err := (placeholderStage{name: "notify"}).SemanticConfigFingerprint(defaultNotify)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
explicitFingerprint, err := (placeholderStage{name: "notify"}).SemanticConfigFingerprint(explicitNotify)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !defaultFingerprint.Equal(explicitFingerprint) {
|
||||||
|
t.Fatalf("default notify mode differs from explicit default")
|
||||||
|
}
|
||||||
|
explicitNotify.Config.Pipeline.Notification.Mode = "webhook"
|
||||||
|
changed, err := (placeholderStage{name: "notify"}).SemanticConfigFingerprint(explicitNotify)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if defaultFingerprint.Equal(changed) {
|
||||||
|
t.Fatal("notify mode change retained fingerprint")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCanonicalStageSemanticReuseCoverage(t *testing.T) {
|
||||||
|
for _, candidate := range All() {
|
||||||
|
if candidate.Name() == "analyze" {
|
||||||
|
if _, ok := candidate.(SemanticConfigFingerprinter); ok {
|
||||||
|
t.Fatal("analyze must retain artifact-level fingerprint authority")
|
||||||
|
}
|
||||||
|
if _, ok := candidate.(ResumeValidator); !ok {
|
||||||
|
t.Fatal("analyze must provide artifact-level resume validation")
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := candidate.(SemanticConfigFingerprinter); !ok {
|
||||||
|
t.Fatalf("canonical stage %q has no semantic reuse contract", candidate.Name())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func downstreamSemanticEnv() *Env {
|
||||||
|
return &Env{Config: &config.Config{
|
||||||
|
Pipeline: &config.PipelineConfig{
|
||||||
|
Workspace: config.WorkspaceConfig{Root: "/workspace"},
|
||||||
|
Storage: config.StorageConfig{Backend: config.StorageBackendS3, S3: &config.StorageS3Config{
|
||||||
|
Bucket: "campaign-bucket", RootPrefix: "narratio", Region: "us-east-1",
|
||||||
|
Endpoint: "https://objects.example", AccessKeyIDEnv: "ACCESS_KEY", SecretKeyEnv: "SECRET_KEY",
|
||||||
|
}},
|
||||||
|
Notarius: &config.NotariusConfig{
|
||||||
|
Enabled: true, Binary: "notarius", ConfigPath: "/etc/notarius/private.yml",
|
||||||
|
PipelineID: "campaign", Timeout: "1h", WorkingDirectory: "/tmp/notarius",
|
||||||
|
References: map[string]string{"party": artifactpolicy.SourceInputParty},
|
||||||
|
Outputs: map[string]config.NotariusOutputConfig{
|
||||||
|
"encounters": {
|
||||||
|
LaneID: "encounters", MediaType: "application/json", SchemaID: "encounters",
|
||||||
|
SchemaVersion: "1", ModuleKey: "encounters",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Publish: &config.PublishConfig{
|
||||||
|
Enabled: boolPointer(true), UploadRun: boolPointer(true),
|
||||||
|
Outputs: []config.PublishOutputRule{
|
||||||
|
{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},
|
||||||
|
},
|
||||||
|
Session: &config.SessionConfig{SessionID: "2026-08-30", Campaign: "campaign"},
|
||||||
|
}}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user