amadan

carlos / platform Public

release: Pipeline object — signed per-app channel list with per-channel rules

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 76fa3b5091f77a5277112cfa13ad77c8347f49e8 parent 09f820e
2 files changed, +416 −0
  • internal/release/pipeline.go +281 −0
  • internal/release/pipeline_test.go +135 −0
diff --git a/internal/release/pipeline.go b/internal/release/pipeline.go
new file mode 100644
index 000000000..e77b767ab
--- /dev/null
+++ b/internal/release/pipeline.go
@@ -0,0 +1,281 @@
+// The per-app pipeline object — the release pipelines v2 spec
+// (docs/superpowers/specs/2026-08-20-release-pipelines-v2-design.md, §2).
+//
+// A pipeline replaces the implicit edge/beta/stable ladder with an
+// account-declared, ordered list of channels: where a release enters
+// (Channels[0]), how it moves between the rest, and — per channel — how
+// long it bakes and how many approvals promoting into or changing it takes.
+// canary/* stays outside the pipeline entirely; it is the pre-pipeline
+// staging rung the ladder already gives every channel name (spec §2), never
+// itself a pipeline entry.
+//
+// Signing and verification mirror production.go exactly: the object is
+// root-signed in the pointer's own signedWrapper envelope and root-ONLY — a
+// wrapper carrying a key grant is refused before its signature is even
+// checked, because a scoped key that could redraw the pipeline could hand
+// itself a channel with no bake and no approvals.
+package release
+
+import (
+ "context"
+ "crypto/ed25519"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "strings"
+ "time"
+)
+
+// ChannelRules is a channel's opt-in policy. The zero value is the default:
+// no rule. Enforcement is the console's job (spec §4-§5); boxes only ever
+// read Bake — rules travel here so one signed object is the whole policy.
+type ChannelRules struct {
+ Passkey bool `json:"passkey,omitempty"`
+ PromoteApprovals int `json:"promote_approvals,omitempty"`
+ ChangeApprovals int `json:"change_approvals,omitempty"`
+}
+
+type PipelineChannel struct {
+ Name string `json:"name"`
+ Bake string `json:"bake,omitempty"` // Go duration string; "" means 0
+ Rules ChannelRules `json:"rules,omitzero"`
+}
+
+// BakeWindow parses Bake, treating "" as no bake at all. A negative
+// duration is rejected here rather than left to a caller that would
+// otherwise wait for a time that never comes.
+func (c PipelineChannel) BakeWindow() (time.Duration, error) {
+ if c.Bake == "" {
+ return 0, nil
+ }
+ d, err := time.ParseDuration(c.Bake)
+ if err != nil || d < 0 {
+ return 0, fmt.Errorf("channel %s: bake %q is not a non-negative duration", c.Name, c.Bake)
+ }
+ return d, nil
+}
+
+// Pipeline is the signed payload: one app's ordered channel list.
+type Pipeline struct {
+ Schema int `json:"schema"` // 1
+ Account string `json:"account"`
+ App string `json:"app"`
+ Channels []PipelineChannel `json:"channels"`
+ // Actor is who set it — a console member's address. Recorded for the
+ // same reason the pointer's audit trail exists: an invisible bypass is a
+ // broken gate.
+ Actor string `json:"actor,omitempty"`
+ SetAt time.Time `json:"set_at"`
+}
+
+// Entry is where a release lands first. Callers must hold a validated
+// pipeline (validatePipeline guarantees at least one channel) — there is no
+// defensive branch here for an empty list.
+func (p *Pipeline) Entry() string { return p.Channels[0].Name }
+
+// Index is a channel's position in the promotion order, -1 for anything
+// outside it — including every canary/* name, which is never a pipeline
+// channel (spec §2).
+func (p *Pipeline) Index(channel string) int {
+ if r, ok := Rung(channel); ok && r == RungCanary {
+ return -1
+ }
+ for i, c := range p.Channels {
+ if c.Name == channel {
+ return i
+ }
+ }
+ return -1
+}
+
+func (p *Pipeline) ChannelNamed(name string) (PipelineChannel, bool) {
+ if i := p.Index(name); i >= 0 {
+ return p.Channels[i], true
+ }
+ return PipelineChannel{}, false
+}
+
+// MaxApprovals is the largest quorum any channel demands — the console
+// refuses to store a pipeline whose MaxApprovals exceeds the account's
+// member count (spec §4).
+func (p *Pipeline) MaxApprovals() int {
+ m := 0
+ for _, c := range p.Channels {
+ if c.Rules.PromoteApprovals > m {
+ m = c.Rules.PromoteApprovals
+ }
+ if c.Rules.ChangeApprovals > m {
+ m = c.Rules.ChangeApprovals
+ }
+ }
+ return m
+}
+
+// PipelineKey is where an app's pipeline lives: beside production.json,
+// deliberately NEVER under ChannelsPrefix — those listings are parsed as
+// channel pointers by the console and the CLI.
+func PipelineKey(acct, app string) string { return appPrefix(acct, app) + "pipeline.json" }
+
+// DefaultPipelineChannels is what a fresh app gets before anyone declares a
+// pipeline of their own: a single entry channel, matching the ladder's
+// current edge-only default.
+func DefaultPipelineChannels() []PipelineChannel { return []PipelineChannel{{Name: "edge"}} }
+
+// SetPipeline writes the object, root key only, CAS over whatever is there
+// (a concurrent write surfaces as ErrCASConflict — re-ask, never overwrite
+// blind). SetAt is stamped here, strictly after whatever this key's own
+// last signed write recorded.
+func SetPipeline(ctx context.Context, s Store, priv ed25519.PrivateKey, acct, app string, channels []PipelineChannel, actor string) (*Pipeline, error) {
+ if len(priv) != ed25519.PrivateKeySize {
+ return nil, fmt.Errorf("set pipeline: no usable release signing key")
+ }
+ if !ValidAccount(acct) {
+ return nil, fmt.Errorf("set pipeline: invalid account %q", acct)
+ }
+ if !ValidApp(app) {
+ return nil, fmt.Errorf("set pipeline: invalid app %q", app)
+ }
+ // A trashed app cannot be given a pipeline — restore first. Unreadable
+ // refuses too: a caller about to write a pipeline must never act as
+ // though a marker it cannot verify simply isn't there.
+ tomb, err := ReadAppTombstone(ctx, s, acct, app)
+ if err != nil {
+ return nil, fmt.Errorf("set pipeline: cannot check the trash marker for %s/%s: %w", acct, app, err)
+ }
+ if tomb != nil {
+ return nil, fmt.Errorf("set pipeline: %s/%s is in the trash — restore it before changing its pipeline", acct, app)
+ }
+ key := PipelineKey(acct, app)
+ raw, etag, err := s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ etag = "" // first write: create-only
+ } else if err != nil {
+ return nil, fmt.Errorf("set pipeline: read %s: %w", key, err)
+ }
+ setAt := time.Now().UTC().Truncate(time.Second)
+ if err == nil {
+ // SetAt is truncated to the second; two writes inside one wall-clock
+ // second would otherwise share a value. Bump strictly past whatever
+ // this key's own last signed write recorded.
+ if existing, verr := verifyPipeline(priv.Public().(ed25519.PublicKey), raw, acct, app); verr == nil && !existing.SetAt.Before(setAt) {
+ setAt = existing.SetAt.Add(time.Second)
+ }
+ }
+ p := &Pipeline{
+ Schema: 1, Account: acct, App: app, Channels: channels,
+ Actor: actor, SetAt: setAt,
+ }
+ if err := validatePipeline(p, acct, app); err != nil {
+ return nil, fmt.Errorf("set pipeline: %w", err)
+ }
+ payload, err := json.MarshalIndent(p, "", " ")
+ if err != nil {
+ return nil, err
+ }
+ b64 := base64.StdEncoding.EncodeToString(payload)
+ wrapper, err := json.MarshalIndent(signedWrapper{
+ Payload: b64,
+ Sig: base64.StdEncoding.EncodeToString(ed25519.Sign(priv, []byte(b64))),
+ }, "", " ")
+ if err != nil {
+ return nil, err
+ }
+ if _, err := s.Put(ctx, key, wrapper, etag); err != nil {
+ return nil, err
+ }
+ return p, nil
+}
+
+// GetPipeline resolves an app's pipeline. A MISSING object is (nil, nil):
+// callers fall back to DefaultPipelineChannels() — but never let a nil
+// collapse the distinction: any OTHER failure (unreadable, unverifiable,
+// misnamed) is the error.
+func GetPipeline(ctx context.Context, s Store, pub ed25519.PublicKey, acct, app string) (*Pipeline, error) {
+ raw, _, err := s.Get(ctx, PipelineKey(acct, app))
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return verifyPipeline(pub, raw, acct, app)
+}
+
+// verifyPipeline is verifyProductionFlag's discipline for the pipeline:
+// refuse a grant outright, check the signature against the ROOT key alone,
+// decode after, validate the payload names the account/app whose key it
+// sits at.
+func verifyPipeline(pub ed25519.PublicKey, wrapper []byte, wantAcct, wantApp string) (*Pipeline, error) {
+ var sw signedWrapper
+ if err := json.Unmarshal(wrapper, &sw); err != nil {
+ return nil, fmt.Errorf("pipeline wrapper: %v", err)
+ }
+ if sw.Grant != nil {
+ return nil, fmt.Errorf("pipeline %s/%s: signed by a scoped key — the pipeline takes the root key, because a delegate that could redraw it could hand itself a channel with no bake and no approvals", wantAcct, wantApp)
+ }
+ if len(pub) != ed25519.PublicKeySize {
+ return nil, fmt.Errorf("pipeline: no usable release public key")
+ }
+ sig, err := base64.StdEncoding.DecodeString(sw.Sig)
+ if err != nil {
+ return nil, fmt.Errorf("pipeline sig: %v", err)
+ }
+ if !ed25519.Verify(pub, []byte(sw.Payload), sig) {
+ return nil, fmt.Errorf("pipeline signature verification failed")
+ }
+ payload, err := base64.StdEncoding.DecodeString(sw.Payload)
+ if err != nil {
+ return nil, fmt.Errorf("pipeline payload: %v", err)
+ }
+ var p Pipeline
+ if err := json.Unmarshal(payload, &p); err != nil {
+ return nil, fmt.Errorf("pipeline payload: %v", err)
+ }
+ if err := validatePipeline(&p, wantAcct, wantApp); err != nil {
+ return nil, err
+ }
+ return &p, nil
+}
+
+// validatePipeline checks a candidate pipeline before it is ever signed or
+// trusted after verification: schema, the copied-object trap (account/app
+// must name the key it sits at), channel count, and per-channel shape.
+func validatePipeline(p *Pipeline, wantAcct, wantApp string) error {
+ if p.Schema != 1 {
+ return fmt.Errorf("unsupported pipeline schema %d", p.Schema)
+ }
+ if p.Account != wantAcct || p.App != wantApp {
+ return fmt.Errorf("pipeline names %s/%s, expected %s/%s", p.Account, p.App, wantAcct, wantApp)
+ }
+ if len(p.Channels) == 0 {
+ return fmt.Errorf("pipeline must have at least one channel")
+ }
+ if len(p.Channels) > 16 {
+ return fmt.Errorf("pipeline has %d channels, at most 16 allowed", len(p.Channels))
+ }
+ seen := make(map[string]bool, len(p.Channels))
+ for _, c := range p.Channels {
+ if !ValidChannel(c.Name) {
+ return fmt.Errorf("invalid channel name %q", c.Name)
+ }
+ if strings.Contains(c.Name, "/") {
+ return fmt.Errorf("channel names may not contain \"/\" — only canary nests")
+ }
+ if c.Name == RungCanary {
+ return fmt.Errorf("%q is reserved — canary/* lives outside the pipeline", RungCanary)
+ }
+ if seen[c.Name] {
+ return fmt.Errorf("duplicate channel %q", c.Name)
+ }
+ seen[c.Name] = true
+ if _, err := c.BakeWindow(); err != nil {
+ return err
+ }
+ if c.Rules.PromoteApprovals < 0 || c.Rules.ChangeApprovals < 0 {
+ return fmt.Errorf("channel %s: approvals must be >= 0", c.Name)
+ }
+ }
+ return nil
+}
diff --git a/internal/release/pipeline_test.go b/internal/release/pipeline_test.go
new file mode 100644
index 000000000..8bf1c1c47
--- /dev/null
+++ b/internal/release/pipeline_test.go
@@ -0,0 +1,135 @@
+package release
+
+// The pipeline object (2026-08-20 release pipelines v2 spec §2): a
+// root-signed per-app ordered channel list with per-channel bake and
+// approval rules, replacing the implicit edge/beta/stable ladder assumption.
+// The properties tested here: validation catches the reserved/duplicate/
+// malformed-channel cases before anything is signed; the accessors read the
+// object the way later tasks will (Entry, Index, ChannelNamed,
+// MaxApprovals); the object round-trips through the same root-only signed
+// wrapper discipline as production.go, including monotonic SetAt and
+// refusal of a copied object at the wrong app.
+
+import (
+ "context"
+ "crypto/ed25519"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestPipelineValidation(t *testing.T) {
+ good := []PipelineChannel{{Name: "edge"}, {Name: "production", Bake: "24h",
+ Rules: ChannelRules{Passkey: true, PromoteApprovals: 2, ChangeApprovals: 2}}}
+ for _, tc := range []struct {
+ name string
+ channels []PipelineChannel
+ wantErr string // "" = valid
+ }{
+ {"default single edge", DefaultPipelineChannels(), ""},
+ {"two channels with rules", good, ""},
+ {"empty list", nil, "at least one channel"},
+ {"reserved canary", []PipelineChannel{{Name: "canary"}}, "reserved"},
+ {"slash in name", []PipelineChannel{{Name: "canary/x"}}, "may not contain"},
+ {"duplicate names", []PipelineChannel{{Name: "edge"}, {Name: "edge"}}, "duplicate"},
+ {"bad duration", []PipelineChannel{{Name: "edge", Bake: "soon"}}, "bake"},
+ {"negative bake", []PipelineChannel{{Name: "edge", Bake: "-1h"}}, "bake"},
+ {"negative approvals", []PipelineChannel{{Name: "edge",
+ Rules: ChannelRules{PromoteApprovals: -1}}}, "approvals"},
+ } {
+ p := &Pipeline{Schema: 1, Account: "bae", App: "web", Channels: tc.channels, SetAt: time.Now()}
+ err := validatePipeline(p, "bae", "web")
+ if tc.wantErr == "" && err != nil {
+ t.Errorf("%s: unexpected error %v", tc.name, err)
+ }
+ if tc.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tc.wantErr)) {
+ t.Errorf("%s: err = %v, want mention of %q", tc.name, err, tc.wantErr)
+ }
+ }
+}
+
+func TestPipelineAccessors(t *testing.T) {
+ p := &Pipeline{Schema: 1, Channels: []PipelineChannel{
+ {Name: "edge"}, {Name: "staging", Bake: "1h"}, {Name: "production", Bake: "24h"}}}
+ if p.Entry() != "edge" {
+ t.Errorf("Entry() = %q, want edge", p.Entry())
+ }
+ if i := p.Index("staging"); i != 1 {
+ t.Errorf("Index(staging) = %d, want 1", i)
+ }
+ if i := p.Index("canary/x"); i != -1 {
+ t.Errorf("Index(canary/x) = %d, want -1 (canary is never a pipeline channel)", i)
+ }
+ if i := p.Index("nope"); i != -1 {
+ t.Errorf("Index(nope) = %d, want -1", i)
+ }
+ c, ok := p.ChannelNamed("production")
+ if !ok || c.Bake != "24h" {
+ t.Errorf("ChannelNamed(production) = %+v, %v", c, ok)
+ }
+ if d, err := c.BakeWindow(); err != nil || d != 24*time.Hour {
+ t.Errorf("BakeWindow() = %v, %v", d, err)
+ }
+ var zero PipelineChannel
+ if d, err := zero.BakeWindow(); err != nil || d != 0 {
+ t.Errorf("zero BakeWindow() = %v, %v; want 0, nil", d, err)
+ }
+}
+
+func TestSetGetPipelineRoundTrip(t *testing.T) {
+ // Mirror TestSetProductionMonotonicSetAtOnRapidFlip's harness:
+ // dir store + root key. Missing object reads as (nil, nil).
+ s, priv, pub := pipelineTestHarness(t)
+ ctx := context.Background()
+ if p, err := GetPipeline(ctx, s, pub, "bae", "web"); p != nil || err != nil {
+ t.Fatalf("missing pipeline = %v, %v; want nil, nil", p, err)
+ }
+ set, err := SetPipeline(ctx, s, priv, "bae", "web", DefaultPipelineChannels(), "paul@rslw.com")
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := GetPipeline(ctx, s, pub, "bae", "web")
+ if err != nil || got == nil || got.Entry() != "edge" || got.Actor != "paul@rslw.com" {
+ t.Fatalf("round trip = %+v, %v", got, err)
+ }
+ // Monotonic SetAt: an immediate rewrite must stamp strictly later.
+ set2, err := SetPipeline(ctx, s, priv, "bae", "web",
+ []PipelineChannel{{Name: "edge"}, {Name: "production", Bake: "24h"}}, "paul@rslw.com")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !set2.SetAt.After(set.SetAt) {
+ t.Fatalf("SetAt not monotonic: %v then %v", set.SetAt, set2.SetAt)
+ }
+}
+
+func TestGetPipelineRefusesTamperAndWrongApp(t *testing.T) {
+ s, priv, pub := pipelineTestHarness(t)
+ ctx := context.Background()
+ if _, err := SetPipeline(ctx, s, priv, "bae", "web", DefaultPipelineChannels(), ""); err != nil {
+ t.Fatal(err)
+ }
+ // Copy the object to another app's key: signed payload names bae/web,
+ // so it must refuse at bae/other (same trap production.go documents).
+ raw, _, err := s.Get(ctx, PipelineKey("bae", "web"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(ctx, PipelineKey("bae", "other"), raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := GetPipeline(ctx, s, pub, "bae", "other"); err == nil {
+ t.Fatal("copied pipeline verified at the wrong app; want refusal")
+ }
+}
+
+// pipelineTestHarness builds a fresh store and root key pair, copying
+// exactly how production_test.go's tests build theirs (NewDirStore over a
+// temp dir, testKeys for the signing pair) — there is no in-memory store in
+// this package; NewDirStore(t.TempDir()) is the existing convention.
+func pipelineTestHarness(t *testing.T) (Store, ed25519.PrivateKey, ed25519.PublicKey) {
+ t.Helper()
+ s := NewDirStore(t.TempDir())
+ priv, pub := testKeys(t)
+ return s, priv, pub
+}