From b74e31e1ccde52f9b6edd29efa353313a8882205 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 12:52:55 +0000 Subject: [PATCH] Model publish output outcomes --- docs/internal/stage-publish.md | 4 +- docs/roadmap/implementation.md | 2 +- internal/stage/publish_output_plan.go | 257 +++++++++++++++++++++ internal/stage/publish_output_plan_test.go | 197 ++++++++++++++++ 4 files changed, 458 insertions(+), 2 deletions(-) create mode 100644 internal/stage/publish_output_plan.go create mode 100644 internal/stage/publish_output_plan_test.go diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 1655b73..45600ed 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -33,7 +33,8 @@ and optional rules, destination resolution, and collision inputs live in `publish_outputs.go`. Copying, hashing, immutable upload, reuse checks, and remote verification live in `publish_transfer.go`; committed-manifest, immutable-commit, and current-pointer encoding and writes live in -`publish_commit.go`. +`publish_commit.go`. `publish_output_plan.go` owns the deterministic, I/O-free +five-outcome model and its secret-free metadata projection. - when publishing or run upload is disabled, completes successfully with no outputs and records explanatory metadata. This is not an explicit self-skip: @@ -125,6 +126,7 @@ The commit boundary and cleanup gate are normative architecture invariants; see helpers. - Implementation and tests: `internal/stage/publish.go`, `internal/stage/publish_sources.go`, `internal/stage/publish_outputs.go`, + `internal/stage/publish_output_plan.go`, `internal/stage/publish_transfer.go`, `internal/stage/publish_commit.go`, `internal/stage/publish_test.go`, `internal/stage/semantic_contracts_delivery.go`, diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 3b66f19..4a879bf 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -222,7 +222,7 @@ partial-failure tests. Run `go test ./internal/stage -run Publish`, ## Stage 6 — Model Locked Output Carry-Forward -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/stage/publish_output_plan.go b/internal/stage/publish_output_plan.go new file mode 100644 index 0000000..071b21b --- /dev/null +++ b/internal/stage/publish_output_plan.go @@ -0,0 +1,257 @@ +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/pathsafe" +) + +type publishOutputStatus string + +const ( + publishOutputPublished publishOutputStatus = "published" + publishOutputCarriedLocked publishOutputStatus = "carried_locked" + publishOutputLockedUnpublished publishOutputStatus = "locked_unpublished" + publishOutputOptionalMissing publishOutputStatus = "optional_missing" + publishOutputUnselected publishOutputStatus = "unselected" +) + +type resolvedPublishRule struct { + Source string + Destination string + Required bool + Selected bool +} + +type discoveredPublishOutput struct { + Source string + LocalPath string + Provenance string +} + +type publishOutputOutcome struct { + Source string + Status publishOutputStatus + Destination string + DeclaredKey string + OriginRun string + Required bool + LocalPath string + Provenance string + Declaration artifacts.RemoteArtifact +} + +type publishOutputPlan struct { + Outcomes []publishOutputOutcome +} + +func buildPublishOutputPlan( + sessionPrefix string, + runPrefix string, + rules []resolvedPublishRule, + discovered []discoveredPublishOutput, + locks []artifacts.RemoteLock, + prior *artifacts.RemoteCommitManifest, +) (*publishOutputPlan, error) { + probeIdentity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, publishOutputRemoteKey(runPrefix, "validation/object")) + if err != nil || probeIdentity.RelativePath != "published/validation/object" { + return nil, fmt.Errorf("publish run prefix %q is not canonical for session %q", runPrefix, sessionPrefix) + } + sources, err := indexDiscoveredPublishOutputs(discovered) + if err != nil { + return nil, err + } + lockSet, err := indexPublishOutputLocks(locks) + if err != nil { + return nil, err + } + priorBySource, err := indexPriorPublishedOutputs(sessionPrefix, prior) + if err != nil { + return nil, err + } + + orderedRules := append([]resolvedPublishRule(nil), rules...) + sort.Slice(orderedRules, func(i, j int) bool { return orderedRules[i].Source < orderedRules[j].Source }) + seenSources := make(map[string]struct{}, len(orderedRules)) + configuredDestinations := make(map[string]string, len(orderedRules)) + outcomes := make([]publishOutputOutcome, 0, len(orderedRules)) + for _, rule := range orderedRules { + if err := validateResolvedPublishRule(rule); err != nil { + return nil, err + } + if _, duplicate := seenSources[rule.Source]; duplicate { + return nil, fmt.Errorf("publish output source %q is configured more than once", rule.Source) + } + seenSources[rule.Source] = struct{}{} + if other, collision := configuredDestinations[rule.Destination]; collision { + return nil, fmt.Errorf("publish output destination %q is configured for both %q and %q", rule.Destination, other, rule.Source) + } + configuredDestinations[rule.Destination] = rule.Source + + outcome := publishOutputOutcome{ + Source: rule.Source, Destination: rule.Destination, Required: rule.Required, + } + if !rule.Selected { + outcome.Status = publishOutputUnselected + outcomes = append(outcomes, outcome) + continue + } + if _, locked := lockSet[rule.Source]; locked { + if declaration, found := priorBySource[rule.Source]; found { + identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, declaration.DestinationKey) + if err != nil { + return nil, fmt.Errorf("carried publish output %q: %w", rule.Source, err) + } + outcome.Status = publishOutputCarriedLocked + outcome.Destination = strings.TrimPrefix(identity.RelativePath, "published/") + outcome.DeclaredKey = declaration.DestinationKey + outcome.OriginRun = identity.RunID + outcome.Declaration = declaration + } else { + outcome.Status = publishOutputLockedUnpublished + } + outcomes = append(outcomes, outcome) + continue + } + discoveredSource, found := sources[rule.Source] + if !found { + if rule.Required { + return nil, fmt.Errorf("required output source unavailable: %q", rule.Source) + } + outcome.Status = publishOutputOptionalMissing + outcomes = append(outcomes, outcome) + continue + } + declaredKey := publishOutputRemoteKey(runPrefix, rule.Destination) + identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, declaredKey) + if err != nil { + return nil, fmt.Errorf("publish output %q destination: %w", rule.Source, err) + } + outcome.Status = publishOutputPublished + outcome.DeclaredKey = declaredKey + outcome.OriginRun = identity.RunID + outcome.LocalPath = discoveredSource.LocalPath + outcome.Provenance = discoveredSource.Provenance + outcome.Declaration = artifacts.RemoteArtifact{ + Type: artifacts.RemoteArtifactTypePublishedOutput, Source: rule.Source, DestinationKey: declaredKey, + } + outcomes = append(outcomes, outcome) + } + if err := validateEffectivePublishDestinations(outcomes); err != nil { + return nil, err + } + return &publishOutputPlan{Outcomes: outcomes}, nil +} + +func validateResolvedPublishRule(rule resolvedPublishRule) error { + if strings.TrimSpace(rule.Source) != rule.Source || rule.Source == "" { + return fmt.Errorf("publish output source %q is not canonical", rule.Source) + } + if _, err := artifactpolicy.ClassifySource(rule.Source); err != nil { + return fmt.Errorf("publish output source %q: %w", rule.Source, err) + } + if strings.TrimSpace(rule.Destination) != rule.Destination || rule.Destination == "" { + return fmt.Errorf("publish output %q destination is required and canonical", rule.Source) + } + normalized, err := pathsafe.NormalizeRelativeDestination(rule.Destination) + if err != nil || normalized != rule.Destination { + return fmt.Errorf("publish output %q destination %q is not canonical", rule.Source, rule.Destination) + } + return nil +} + +func indexDiscoveredPublishOutputs(discovered []discoveredPublishOutput) (map[string]discoveredPublishOutput, error) { + out := make(map[string]discoveredPublishOutput, len(discovered)) + for _, source := range discovered { + if strings.TrimSpace(source.Source) != source.Source || source.Source == "" { + return nil, fmt.Errorf("discovered publish source %q is not canonical", source.Source) + } + if _, err := artifactpolicy.ClassifySource(source.Source); err != nil { + return nil, fmt.Errorf("discovered publish source %q: %w", source.Source, err) + } + if strings.TrimSpace(source.LocalPath) == "" { + return nil, fmt.Errorf("discovered publish source %q has no local path", source.Source) + } + if _, duplicate := out[source.Source]; duplicate { + return nil, fmt.Errorf("discovered publish source %q is ambiguous", source.Source) + } + out[source.Source] = source + } + return out, nil +} + +func indexPublishOutputLocks(locks []artifacts.RemoteLock) (map[string]artifacts.RemoteLock, error) { + out := make(map[string]artifacts.RemoteLock, len(locks)) + for _, lock := range locks { + if err := lock.Validate(); err != nil { + return nil, fmt.Errorf("publish output lock: %w", err) + } + if _, duplicate := out[lock.Source]; duplicate { + return nil, fmt.Errorf("publish output lock source %q is duplicated", lock.Source) + } + out[lock.Source] = lock + } + return out, nil +} + +func indexPriorPublishedOutputs(sessionPrefix string, prior *artifacts.RemoteCommitManifest) (map[string]artifacts.RemoteArtifact, error) { + out := map[string]artifacts.RemoteArtifact{} + if prior == nil { + return out, nil + } + if err := prior.ValidateForSessionPrefix(sessionPrefix); err != nil { + return nil, fmt.Errorf("prior remote commit: %w", err) + } + for _, declaration := range prior.Artifacts { + if declaration.Type != artifacts.RemoteArtifactTypePublishedOutput { + continue + } + if _, duplicate := out[declaration.Source]; duplicate { + return nil, fmt.Errorf("prior remote commit publishes source %q more than once", declaration.Source) + } + out[declaration.Source] = declaration + } + return out, nil +} + +func validateEffectivePublishDestinations(outcomes []publishOutputOutcome) error { + seen := map[string]string{} + for _, outcome := range outcomes { + if outcome.Status != publishOutputPublished && outcome.Status != publishOutputCarriedLocked { + continue + } + if other, collision := seen[outcome.Destination]; collision { + return fmt.Errorf("effective publish destination %q is declared by both %q and %q", outcome.Destination, other, outcome.Source) + } + seen[outcome.Destination] = outcome.Source + } + return nil +} + +func publishOutputOutcomeMetadata(plan *publishOutputPlan) []map[string]any { + if plan == nil { + return nil + } + out := make([]map[string]any, 0, len(plan.Outcomes)) + for _, outcome := range plan.Outcomes { + item := map[string]any{ + "source": outcome.Source, + "status": string(outcome.Status), + } + if outcome.Destination != "" { + item["destination"] = outcome.Destination + } + if outcome.DeclaredKey != "" { + item["declared_key"] = outcome.DeclaredKey + } + if outcome.OriginRun != "" { + item["origin_run"] = outcome.OriginRun + } + out = append(out, item) + } + return out +} diff --git a/internal/stage/publish_output_plan_test.go b/internal/stage/publish_output_plan_test.go new file mode 100644 index 0000000..9b13d5d --- /dev/null +++ b/internal/stage/publish_output_plan_test.go @@ -0,0 +1,197 @@ +package stage + +import ( + "reflect" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" +) + +func TestBuildPublishOutputPlanModelsEveryOutcome(t *testing.T) { + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + currentRun := "20260519T010203Z-a1b2c3d4" + runPrefix := artifacts.S3RunPrefix(sessionPrefix, currentRun) + carried := publishPlanDeclaration(sessionPrefix, "20260512T010203Z-11223344", "narratio.artifact.beta", "artifacts/beta-old.md") + prior := publishPlanCommit(carried) + locks := []artifacts.RemoteLock{ + publishPlanLock(t, "narratio.artifact.beta"), + publishPlanLock(t, "narratio.artifact.gamma"), + } + rules := []resolvedPublishRule{ + {Source: "narratio.artifact.gamma", Destination: "artifacts/gamma.md", Required: true, Selected: true}, + {Source: "narratio.artifact.alpha", Destination: "artifacts/alpha.md", Required: true, Selected: true}, + {Source: "narratio.artifact.epsilon", Destination: "artifacts/epsilon.md", Required: true, Selected: false}, + {Source: "narratio.artifact.beta", Destination: "artifacts/beta-new.md", Required: true, Selected: true}, + {Source: "narratio.artifact.delta", Destination: "artifacts/delta.md", Required: false, Selected: true}, + } + discovered := []discoveredPublishOutput{ + {Source: "narratio.artifact.alpha", LocalPath: "/workspace/alpha.md", Provenance: "current"}, + } + + plan, err := buildPublishOutputPlan(sessionPrefix, runPrefix, rules, discovered, locks, prior) + if err != nil { + t.Fatalf("buildPublishOutputPlan() error = %v", err) + } + wantSources := []string{ + "narratio.artifact.alpha", + "narratio.artifact.beta", + "narratio.artifact.delta", + "narratio.artifact.epsilon", + "narratio.artifact.gamma", + } + wantStatuses := []publishOutputStatus{ + publishOutputPublished, + publishOutputCarriedLocked, + publishOutputOptionalMissing, + publishOutputUnselected, + publishOutputLockedUnpublished, + } + if len(plan.Outcomes) != len(wantSources) { + t.Fatalf("outcomes = %#v", plan.Outcomes) + } + for index := range wantSources { + if plan.Outcomes[index].Source != wantSources[index] || plan.Outcomes[index].Status != wantStatuses[index] { + t.Fatalf("outcomes[%d] = %#v, want source %q status %q", index, plan.Outcomes[index], wantSources[index], wantStatuses[index]) + } + } + if !reflect.DeepEqual(plan.Outcomes[1].Declaration, carried) { + t.Fatalf("carried declaration = %#v, want exact %#v", plan.Outcomes[1].Declaration, carried) + } + if plan.Outcomes[1].Destination != "artifacts/beta-old.md" || plan.Outcomes[1].OriginRun != "20260512T010203Z-11223344" { + t.Fatalf("carried outcome = %#v", plan.Outcomes[1]) + } + if plan.Outcomes[4].Declaration != (artifacts.RemoteArtifact{}) { + t.Fatalf("locked unpublished declaration = %#v, want none", plan.Outcomes[4].Declaration) + } + + metadata := publishOutputOutcomeMetadata(plan) + if len(metadata) != len(plan.Outcomes) { + t.Fatalf("metadata = %#v", metadata) + } + if metadata[1]["status"] != "carried_locked" || metadata[1]["declared_key"] != carried.DestinationKey || metadata[1]["origin_run"] != "20260512T010203Z-11223344" { + t.Fatalf("carried metadata = %#v", metadata[1]) + } + for _, item := range metadata { + if _, exposed := item["local_path"]; exposed { + t.Fatalf("metadata exposes local path: %#v", item) + } + } +} + +func TestBuildPublishOutputPlanRejectsAmbiguities(t *testing.T) { + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + runPrefix := artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4") + available := []discoveredPublishOutput{{Source: "narratio.artifact.alpha", LocalPath: "/workspace/alpha.md"}} + tests := []struct { + name string + rules []resolvedPublishRule + discovered []discoveredPublishOutput + locks []artifacts.RemoteLock + prior *artifacts.RemoteCommitManifest + want string + }{ + { + name: "duplicate configured source", + rules: []resolvedPublishRule{ + {Source: "narratio.artifact.alpha", Destination: "artifacts/a.md", Selected: true}, + {Source: "narratio.artifact.alpha", Destination: "artifacts/b.md", Selected: true}, + }, + discovered: available, + want: "configured more than once", + }, + { + name: "configured destination collision", + rules: []resolvedPublishRule{ + {Source: "narratio.artifact.alpha", Destination: "artifacts/shared.md", Selected: true}, + {Source: "narratio.artifact.beta", Destination: "artifacts/shared.md", Selected: false}, + }, + discovered: available, + want: "configured for both", + }, + { + name: "ambiguous discovery", + rules: []resolvedPublishRule{{Source: "narratio.artifact.alpha", Destination: "artifacts/a.md", Selected: true}}, + discovered: []discoveredPublishOutput{ + {Source: "narratio.artifact.alpha", LocalPath: "/workspace/a.md"}, + {Source: "narratio.artifact.alpha", LocalPath: "/workspace/other.md"}, + }, + want: "ambiguous", + }, + { + name: "duplicate prior source", + rules: []resolvedPublishRule{{Source: "narratio.artifact.alpha", Destination: "artifacts/a.md", Selected: true}}, + locks: []artifacts.RemoteLock{publishPlanLock(t, "narratio.artifact.alpha")}, + prior: publishPlanCommit( + publishPlanDeclaration(sessionPrefix, "20260512T010203Z-11223344", "narratio.artifact.alpha", "artifacts/a.md"), + publishPlanDeclaration(sessionPrefix, "20260512T010203Z-11223344", "narratio.artifact.alpha", "artifacts/other.md"), + ), + want: "ambiguous published_output source", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := buildPublishOutputPlan(sessionPrefix, runPrefix, test.rules, test.discovered, test.locks, test.prior) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("buildPublishOutputPlan() error = %v, want %q", err, test.want) + } + }) + } +} + +func TestBuildPublishOutputPlanRejectsInvalidCarryAndEffectiveCollision(t *testing.T) { + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") + runPrefix := artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4") + locked := publishPlanLock(t, "narratio.artifact.beta") + rules := []resolvedPublishRule{ + {Source: "narratio.artifact.alpha", Destination: "artifacts/shared.md", Required: true, Selected: true}, + {Source: "narratio.artifact.beta", Destination: "artifacts/new-beta.md", Required: true, Selected: true}, + } + discovered := []discoveredPublishOutput{{Source: "narratio.artifact.alpha", LocalPath: "/workspace/alpha.md"}} + + crossSession := publishPlanDeclaration( + artifacts.S3SessionPrefix("dnd", "other-campaign", "2026-05-03"), + "20260512T010203Z-11223344", + "narratio.artifact.beta", + "artifacts/shared.md", + ) + if _, err := buildPublishOutputPlan(sessionPrefix, runPrefix, rules, discovered, []artifacts.RemoteLock{locked}, publishPlanCommit(crossSession)); err == nil || !strings.Contains(err.Error(), "outside session run prefix") { + t.Fatalf("cross-session carry error = %v", err) + } + + carried := publishPlanDeclaration(sessionPrefix, "20260512T010203Z-11223344", "narratio.artifact.beta", "artifacts/shared.md") + if _, err := buildPublishOutputPlan(sessionPrefix, runPrefix, rules, discovered, []artifacts.RemoteLock{locked}, publishPlanCommit(carried)); err == nil || !strings.Contains(err.Error(), "effective publish destination") { + t.Fatalf("effective destination collision error = %v", err) + } +} + +func publishPlanLock(t *testing.T, source string) artifacts.RemoteLock { + t.Helper() + lock, err := artifacts.NewRemoteLock(source, "review", time.Date(2026, 5, 18, 1, 2, 3, 0, time.UTC)) + if err != nil { + t.Fatalf("NewRemoteLock() error = %v", err) + } + return lock +} + +func publishPlanDeclaration(sessionPrefix, runID, source, destination string) artifacts.RemoteArtifact { + return artifacts.RemoteArtifact{ + Type: artifacts.RemoteArtifactTypePublishedOutput, + Source: source, + DestinationKey: publishOutputRemoteKey(artifacts.S3RunPrefix(sessionPrefix, runID), destination), + SHA256: strings.Repeat("a", 64), + Size: 17, + Generation: "generation-" + source, + } +} + +func publishPlanCommit(declarations ...artifacts.RemoteArtifact) *artifacts.RemoteCommitManifest { + return &artifacts.RemoteCommitManifest{ + FormatVersion: artifacts.RemoteCommitFormatVersion, + Campaign: "sample-campaign", + SessionID: "2026-05-03", + RunID: "20260512T010203Z-11223344", + Artifacts: declarations, + } +}