diff --git a/api/tangled/cbor_gen.go b/api/tangled/cbor_gen.go
index 6c9f483f..0335dbc8 100644
--- a/api/tangled/cbor_gen.go
+++ b/api/tangled/cbor_gen.go
@@ -662,7 +662,7 @@ func (t *ActorProfile) UnmarshalCBOR(r io.Reader) (err error) {
 
 	return nil
 }
-func (t *CiDefs_Pipeline) MarshalCBOR(w io.Writer) error {
+func (t *CiPipeline) MarshalCBOR(w io.Writer) error {
 	if t == nil {
 		_, err := w.Write(cbg.CborNull)
 		return err
@@ -761,7 +761,7 @@ func (t *CiDefs_Pipeline) MarshalCBOR(w io.Writer) error {
 		return err
 	}
 
-	// t.Trigger (tangled.CiDefs_Pipeline_Trigger) (struct)
+	// t.Trigger (tangled.CiPipeline_Trigger) (struct)
 	if len("trigger") > 1000000 {
 		return xerrors.Errorf("Value in field \"trigger\" was too long")
 	}
@@ -809,7 +809,7 @@ func (t *CiDefs_Pipeline) MarshalCBOR(w io.Writer) error {
 		}
 	}
 
-	// t.Workflows ([]*tangled.CiDefs_Workflow) (slice)
+	// t.Workflows ([]*tangled.CiPipeline_Workflow) (slice)
 	if len("workflows") > 1000000 {
 		return xerrors.Errorf("Value in field \"workflows\" was too long")
 	}
@@ -837,8 +837,8 @@ func (t *CiDefs_Pipeline) MarshalCBOR(w io.Writer) error {
 	return nil
 }
 
-func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
-	*t = CiDefs_Pipeline{}
+func (t *CiPipeline) UnmarshalCBOR(r io.Reader) (err error) {
+	*t = CiPipeline{}
 
 	cr := cbg.NewCborReader(r)
 
@@ -857,7 +857,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 	}
 
 	if extra > cbg.MaxLength {
-		return fmt.Errorf("CiDefs_Pipeline: map struct too large (%d)", extra)
+		return fmt.Errorf("CiPipeline: map struct too large (%d)", extra)
 	}
 
 	n := extra
@@ -921,7 +921,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 
 				t.Commit = string(sval)
 			}
-			// t.Trigger (tangled.CiDefs_Pipeline_Trigger) (struct)
+			// t.Trigger (tangled.CiPipeline_Trigger) (struct)
 		case "trigger":
 
 			{
@@ -934,7 +934,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 					if err := cr.UnreadByte(); err != nil {
 						return err
 					}
-					t.Trigger = new(CiDefs_Pipeline_Trigger)
+					t.Trigger = new(CiPipeline_Trigger)
 					if err := t.Trigger.UnmarshalCBOR(cr); err != nil {
 						return xerrors.Errorf("unmarshaling t.Trigger pointer: %w", err)
 					}
@@ -962,7 +962,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 					t.CreatedAt = (*string)(&sval)
 				}
 			}
-			// t.Workflows ([]*tangled.CiDefs_Workflow) (slice)
+			// t.Workflows ([]*tangled.CiPipeline_Workflow) (slice)
 		case "workflows":
 
 			maj, extra, err = cr.ReadHeader()
@@ -979,7 +979,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 			}
 
 			if extra > 0 {
-				t.Workflows = make([]*CiDefs_Workflow, extra)
+				t.Workflows = make([]*CiPipeline_Workflow, extra)
 			}
 
 			for i := 0; i < int(extra); i++ {
@@ -1001,7 +1001,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 							if err := cr.UnreadByte(); err != nil {
 								return err
 							}
-							t.Workflows[i] = new(CiDefs_Workflow)
+							t.Workflows[i] = new(CiPipeline_Workflow)
 							if err := t.Workflows[i].UnmarshalCBOR(cr); err != nil {
 								return xerrors.Errorf("unmarshaling t.Workflows[i] pointer: %w", err)
 							}
@@ -1022,7 +1022,7 @@ func (t *CiDefs_Pipeline) UnmarshalCBOR(r io.Reader) (err error) {
 
 	return nil
 }
-func (t *CiDefs_Pipeline_Trigger) MarshalCBOR(w io.Writer) error {
+func (t *CiPipeline_Trigger) MarshalCBOR(w io.Writer) error {
 	if t == nil {
 		_, err := w.Write(cbg.CborNull)
 		return err
@@ -1084,8 +1084,8 @@ func (t *CiDefs_Pipeline_Trigger) MarshalCBOR(w io.Writer) error {
 	return nil
 }
 
-func (t *CiDefs_Pipeline_Trigger) UnmarshalCBOR(r io.Reader) (err error) {
-	*t = CiDefs_Pipeline_Trigger{}
+func (t *CiPipeline_Trigger) UnmarshalCBOR(r io.Reader) (err error) {
+	*t = CiPipeline_Trigger{}
 
 	cr := cbg.NewCborReader(r)
 
@@ -1104,7 +1104,7 @@ func (t *CiDefs_Pipeline_Trigger) UnmarshalCBOR(r io.Reader) (err error) {
 	}
 
 	if extra > cbg.MaxLength {
-		return fmt.Errorf("CiDefs_Pipeline_Trigger: map struct too large (%d)", extra)
+		return fmt.Errorf("CiPipeline_Trigger: map struct too large (%d)", extra)
 	}
 
 	n := extra
@@ -1196,7 +1196,7 @@ func (t *CiDefs_Pipeline_Trigger) UnmarshalCBOR(r io.Reader) (err error) {
 
 	return nil
 }
-func (t *CiDefs_Workflow) MarshalCBOR(w io.Writer) error {
+func (t *CiPipeline_Workflow) MarshalCBOR(w io.Writer) error {
 	if t == nil {
 		_, err := w.Write(cbg.CborNull)
 		return err
@@ -1388,8 +1388,8 @@ func (t *CiDefs_Workflow) MarshalCBOR(w io.Writer) error {
 	return nil
 }
 
-func (t *CiDefs_Workflow) UnmarshalCBOR(r io.Reader) (err error) {
-	*t = CiDefs_Workflow{}
+func (t *CiPipeline_Workflow) UnmarshalCBOR(r io.Reader) (err error) {
+	*t = CiPipeline_Workflow{}
 
 	cr := cbg.NewCborReader(r)
 
@@ -1408,7 +1408,7 @@ func (t *CiDefs_Workflow) UnmarshalCBOR(r io.Reader) (err error) {
 	}
 
 	if extra > cbg.MaxLength {
-		return fmt.Errorf("CiDefs_Workflow: map struct too large (%d)", extra)
+		return fmt.Errorf("CiPipeline_Workflow: map struct too large (%d)", extra)
 	}
 
 	n := extra
@@ -1536,7 +1536,7 @@ func (t *CiDefs_Workflow) UnmarshalCBOR(r io.Reader) (err error) {
 
 	return nil
 }
-func (t *CiPipelineSubscribeLogs_Control) MarshalCBOR(w io.Writer) error {
+func (t *CiSubscribePipelineLogs_Control) MarshalCBOR(w io.Writer) error {
 	if t == nil {
 		_, err := w.Write(cbg.CborNull)
 		return err
@@ -1750,8 +1750,8 @@ func (t *CiPipelineSubscribeLogs_Control) MarshalCBOR(w io.Writer) error {
 	return nil
 }
 
-func (t *CiPipelineSubscribeLogs_Control) UnmarshalCBOR(r io.Reader) (err error) {
-	*t = CiPipelineSubscribeLogs_Control{}
+func (t *CiSubscribePipelineLogs_Control) UnmarshalCBOR(r io.Reader) (err error) {
+	*t = CiSubscribePipelineLogs_Control{}
 
 	cr := cbg.NewCborReader(r)
 
@@ -1770,7 +1770,7 @@ func (t *CiPipelineSubscribeLogs_Control) UnmarshalCBOR(r io.Reader) (err error)
 	}
 
 	if extra > cbg.MaxLength {
-		return fmt.Errorf("CiPipelineSubscribeLogs_Control: map struct too large (%d)", extra)
+		return fmt.Errorf("CiSubscribePipelineLogs_Control: map struct too large (%d)", extra)
 	}
 
 	n := extra
@@ -1924,7 +1924,7 @@ func (t *CiPipelineSubscribeLogs_Control) UnmarshalCBOR(r io.Reader) (err error)
 
 	return nil
 }
-func (t *CiPipelineSubscribeLogs_Data) MarshalCBOR(w io.Writer) error {
+func (t *CiSubscribePipelineLogs_Data) MarshalCBOR(w io.Writer) error {
 	if t == nil {
 		_, err := w.Write(cbg.CborNull)
 		return err
@@ -2052,8 +2052,8 @@ func (t *CiPipelineSubscribeLogs_Data) MarshalCBOR(w io.Writer) error {
 	return nil
 }
 
-func (t *CiPipelineSubscribeLogs_Data) UnmarshalCBOR(r io.Reader) (err error) {
-	*t = CiPipelineSubscribeLogs_Data{}
+func (t *CiSubscribePipelineLogs_Data) UnmarshalCBOR(r io.Reader) (err error) {
+	*t = CiSubscribePipelineLogs_Data{}
 
 	cr := cbg.NewCborReader(r)
 
@@ -2072,7 +2072,7 @@ func (t *CiPipelineSubscribeLogs_Data) UnmarshalCBOR(r io.Reader) (err error) {
 	}
 
 	if extra > cbg.MaxLength {
-		return fmt.Errorf("CiPipelineSubscribeLogs_Data: map struct too large (%d)", extra)
+		return fmt.Errorf("CiSubscribePipelineLogs_Data: map struct too large (%d)", extra)
 	}
 
 	n := extra
diff --git a/api/tangled/cicancelPipeline.go b/api/tangled/cicancelPipeline.go
new file mode 100644
index 00000000..47ab7e76
--- /dev/null
+++ b/api/tangled/cicancelPipeline.go
@@ -0,0 +1,34 @@
+// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT.
+
+package tangled
+
+// schema: sh.tangled.ci.cancelPipeline
+
+import (
+	"context"
+
+	"github.com/bluesky-social/indigo/lex/util"
+)
+
+const (
+	CiCancelPipelineNSID = "sh.tangled.ci.cancelPipeline"
+)
+
+// CiCancelPipeline_Input is the input argument to a sh.tangled.ci.cancelPipeline call.
+type CiCancelPipeline_Input struct {
+	// pipeline: pipeline TID
+	Pipeline string `json:"pipeline" cborgen:"pipeline"`
+	// repo: git repository DID
+	Repo string `json:"repo" cborgen:"repo"`
+	// workflows: Workflow names to filter. When not provided, entire pipeline will be canceled.
+	Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"`
+}
+
+// CiCancelPipeline calls the XRPC method "sh.tangled.ci.cancelPipeline".
+func CiCancelPipeline(ctx context.Context, c util.LexClient, input *CiCancelPipeline_Input) error {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.cancelPipeline", nil, input, nil); err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/api/tangled/cigetPipeline.go b/api/tangled/cigetPipeline.go
index 37ec1d89..b5e48b70 100644
--- a/api/tangled/cigetPipeline.go
+++ b/api/tangled/cigetPipeline.go
@@ -17,8 +17,8 @@ const (
 // CiGetPipeline calls the XRPC method "sh.tangled.ci.getPipeline".
 //
 // pipeline: Spindle-local pipeline id
-func CiGetPipeline(ctx context.Context, c util.LexClient, pipeline string) (*CiDefs_Pipeline, error) {
-	var out CiDefs_Pipeline
+func CiGetPipeline(ctx context.Context, c util.LexClient, pipeline string) (*CiPipeline, error) {
+	var out CiPipeline
 
 	params := map[string]interface{}{}
 	params["pipeline"] = pipeline
diff --git a/api/tangled/cidefs.go b/api/tangled/cipipeline.go
similarity index 77%
rename from api/tangled/cidefs.go
rename to api/tangled/cipipeline.go
index 8edcbe31..4368e1bb 100644
--- a/api/tangled/cidefs.go
+++ b/api/tangled/cipipeline.go
@@ -2,7 +2,7 @@
 
 package tangled
 
-// schema: sh.tangled.ci.defs
+// schema: sh.tangled.ci.pipeline
 
 import (
 	"encoding/json"
@@ -11,10 +11,14 @@ import (
 	"github.com/bluesky-social/indigo/lex/util"
 )
 
-const ()
+const (
+	CiPipelineNSID = "sh.tangled.ci.pipeline"
+)
 
-// CiDefs_Pipeline is a "pipeline" in the sh.tangled.ci.defs schema.
-type CiDefs_Pipeline struct {
+// CiPipeline is a "main" in the sh.tangled.ci.pipeline schema.
+//
+// A CI pipeline. Record-like, but owned by the spindle rather than a PDS.
+type CiPipeline struct {
 	// commit: Commit Id this pipeline is running on
 	Commit    string  `json:"commit" cborgen:"commit"`
 	CreatedAt *string `json:"createdAt,omitempty" cborgen:"createdAt,omitempty"`
@@ -23,19 +27,19 @@ type CiDefs_Pipeline struct {
 	// repo: Repository DID
 	Repo *string `json:"repo,omitempty" cborgen:"repo,omitempty"`
 	// trigger: Trigger event metadata
-	Trigger *CiDefs_Pipeline_Trigger `json:"trigger" cborgen:"trigger"`
+	Trigger *CiPipeline_Trigger `json:"trigger" cborgen:"trigger"`
 	// workflows: Triggered workflows
-	Workflows []*CiDefs_Workflow `json:"workflows" cborgen:"workflows"`
+	Workflows []*CiPipeline_Workflow `json:"workflows" cborgen:"workflows"`
 }
 
 // Trigger event metadata
-type CiDefs_Pipeline_Trigger struct {
+type CiPipeline_Trigger struct {
 	CiTrigger_Push        *CiTrigger_Push
 	CiTrigger_PullRequest *CiTrigger_PullRequest
 	CiTrigger_Manual      *CiTrigger_Manual
 }
 
-func (t *CiDefs_Pipeline_Trigger) MarshalJSON() ([]byte, error) {
+func (t *CiPipeline_Trigger) MarshalJSON() ([]byte, error) {
 	if t.CiTrigger_Push != nil {
 		t.CiTrigger_Push.LexiconTypeID = "sh.tangled.ci.trigger#push"
 		return json.Marshal(t.CiTrigger_Push)
@@ -50,7 +54,7 @@ func (t *CiDefs_Pipeline_Trigger) MarshalJSON() ([]byte, error) {
 	}
 	return nil, fmt.Errorf("cannot marshal empty enum")
 }
-func (t *CiDefs_Pipeline_Trigger) UnmarshalJSON(b []byte) error {
+func (t *CiPipeline_Trigger) UnmarshalJSON(b []byte) error {
 	typ, err := util.TypeExtract(b)
 	if err != nil {
 		return err
@@ -72,8 +76,8 @@ func (t *CiDefs_Pipeline_Trigger) UnmarshalJSON(b []byte) error {
 	}
 }
 
-// CiDefs_Workflow is a "workflow" in the sh.tangled.ci.defs schema.
-type CiDefs_Workflow struct {
+// CiPipeline_Workflow is a "workflow" in the sh.tangled.ci.pipeline schema.
+type CiPipeline_Workflow struct {
 	Error      *string `json:"error,omitempty" cborgen:"error,omitempty"`
 	FinishedAt *string `json:"finishedAt,omitempty" cborgen:"finishedAt,omitempty"`
 	// id: Spindle-local workflow id. Unique per pipeline, usually same as name.
diff --git a/api/tangled/ciqueryPipelines.go b/api/tangled/ciqueryPipelines.go
index 8f6a336b..a78dc0c1 100644
--- a/api/tangled/ciqueryPipelines.go
+++ b/api/tangled/ciqueryPipelines.go
@@ -16,8 +16,8 @@ const (
 
 // CiQueryPipelines_Output is the output of a sh.tangled.ci.queryPipelines call.
 type CiQueryPipelines_Output struct {
-	Cursor    *string            `json:"cursor,omitempty" cborgen:"cursor,omitempty"`
-	Pipelines []*CiDefs_Pipeline `json:"pipelines" cborgen:"pipelines"`
+	Cursor    *string       `json:"cursor,omitempty" cborgen:"cursor,omitempty"`
+	Pipelines []*CiPipeline `json:"pipelines" cborgen:"pipelines"`
 	// total: Maximum number of pipelines
 	Total int64 `json:"total" cborgen:"total"`
 }
diff --git a/api/tangled/pipelinesubscribeLogs.go b/api/tangled/cisubscribePipelineLogs.go
similarity index 69%
rename from api/tangled/pipelinesubscribeLogs.go
rename to api/tangled/cisubscribePipelineLogs.go
index 451d90e9..2955d03d 100644
--- a/api/tangled/pipelinesubscribeLogs.go
+++ b/api/tangled/cisubscribePipelineLogs.go
@@ -2,14 +2,14 @@
 
 package tangled
 
-// schema: sh.tangled.ci.pipeline.subscribeLogs
+// schema: sh.tangled.ci.subscribePipelineLogs
 
 const (
-	CiPipelineSubscribeLogsNSID = "sh.tangled.ci.pipeline.subscribeLogs"
+	CiSubscribePipelineLogsNSID = "sh.tangled.ci.subscribePipelineLogs"
 )
 
-// CiPipelineSubscribeLogs_Control is a "control" in the sh.tangled.ci.pipeline.subscribeLogs schema.
-type CiPipelineSubscribeLogs_Control struct {
+// CiSubscribePipelineLogs_Control is a "control" in the sh.tangled.ci.subscribePipelineLogs schema.
+type CiSubscribePipelineLogs_Control struct {
 	// command: Step command
 	Command *string `json:"command,omitempty" cborgen:"command,omitempty"`
 	Content string  `json:"content" cborgen:"content"`
@@ -24,8 +24,8 @@ type CiPipelineSubscribeLogs_Control struct {
 	Workflow string `json:"workflow" cborgen:"workflow"`
 }
 
-// CiPipelineSubscribeLogs_Data is a "data" in the sh.tangled.ci.pipeline.subscribeLogs schema.
-type CiPipelineSubscribeLogs_Data struct {
+// CiSubscribePipelineLogs_Data is a "data" in the sh.tangled.ci.subscribePipelineLogs schema.
+type CiSubscribePipelineLogs_Data struct {
 	Content string `json:"content" cborgen:"content"`
 	// step: Step ID
 	Step   int64  `json:"step" cborgen:"step"`
diff --git a/api/tangled/pipelinesubscribeLogs_ext.go b/api/tangled/cisubscribePipelineLogs_ext.go
similarity index 75%
rename from api/tangled/pipelinesubscribeLogs_ext.go
rename to api/tangled/cisubscribePipelineLogs_ext.go
index b0e53d7f..d40e0b45 100644
--- a/api/tangled/pipelinesubscribeLogs_ext.go
+++ b/api/tangled/cisubscribePipelineLogs_ext.go
@@ -1,4 +1,4 @@
-// extending code generated from sh.tangled.ci.pipeline.subscribeLogs
+// extending code generated from sh.tangled.ci.subscribePipelineLogs
 
 package tangled
 
@@ -14,16 +14,16 @@ import (
 )
 
 // TODO: generate codes below from lexicon
-type CiPipelineSubscribeLogs_Event struct {
+type CiSubscribePipelineLogs_Event struct {
 	Error   *events.ErrorFrame
-	Control *CiPipelineSubscribeLogs_Control
-	Data    *CiPipelineSubscribeLogs_Data
+	Control *CiSubscribePipelineLogs_Control
+	Data    *CiSubscribePipelineLogs_Data
 
 	// some private fields for internal routing perf
 	Preserialized []byte `json:"-" cborgen:"-"`
 }
 
-func (xevt *CiPipelineSubscribeLogs_Event) Serialize(wc io.Writer) error {
+func (xevt *CiSubscribePipelineLogs_Event) Serialize(wc io.Writer) error {
 	header := events.EventHeader{Op: events.EvtKindMessage}
 	var obj lexutil.CBOR
 
@@ -48,7 +48,7 @@ func (xevt *CiPipelineSubscribeLogs_Event) Serialize(wc io.Writer) error {
 	return obj.MarshalCBOR(cborWriter)
 }
 
-func (xevt *CiPipelineSubscribeLogs_Event) Deserialize(r io.Reader) error {
+func (xevt *CiSubscribePipelineLogs_Event) Deserialize(r io.Reader) error {
 	var header events.EventHeader
 	if err := header.UnmarshalCBOR(r); err != nil {
 		return fmt.Errorf("reading header: %w", err)
@@ -57,13 +57,13 @@ func (xevt *CiPipelineSubscribeLogs_Event) Deserialize(r io.Reader) error {
 	case events.EvtKindMessage:
 		switch header.MsgType {
 		case "#control":
-			var evt CiPipelineSubscribeLogs_Control
+			var evt CiSubscribePipelineLogs_Control
 			if err := evt.UnmarshalCBOR(r); err != nil {
 				return fmt.Errorf("reading repoCommit event: %w", err)
 			}
 			xevt.Control = &evt
 		case "#data":
-			var evt CiPipelineSubscribeLogs_Data
+			var evt CiSubscribePipelineLogs_Data
 			if err := evt.UnmarshalCBOR(r); err != nil {
 				return fmt.Errorf("reading repoSync event: %w", err)
 			}
@@ -81,15 +81,15 @@ func (xevt *CiPipelineSubscribeLogs_Event) Deserialize(r io.Reader) error {
 	return nil
 }
 
-func CiPipelineSubscribeLogs(ctx context.Context, c extlexutil.LexClient, pipeline string, workflows []string, sched extlexutil.Scheduler[CiPipelineSubscribeLogs_Event]) error {
+func CiSubscribePipelineLogs(ctx context.Context, c extlexutil.LexClient, pipeline string, workflows []string, sched extlexutil.Scheduler[CiSubscribePipelineLogs_Event]) error {
 	defer sched.Shutdown()
 
 	params := map[string]any{}
 	params["pipeline"] = pipeline
 	params["workflows"] = workflows
 
-	return c.LexDo(ctx, extlexutil.Subscription, "", CiPipelineSubscribeLogsNSID, params, nil, func(ctx context.Context, cr *cbg.CborReader) error {
-		var evt CiPipelineSubscribeLogs_Event
+	return c.LexDo(ctx, extlexutil.Subscription, "", CiSubscribePipelineLogsNSID, params, nil, func(ctx context.Context, cr *cbg.CborReader) error {
+		var evt CiSubscribePipelineLogs_Event
 		if err := evt.Deserialize(cr); err != nil {
 			return err
 		}
diff --git a/api/tangled/pipelinetriggerPipeline.go b/api/tangled/citriggerPipeline.go
similarity index 51%
rename from api/tangled/pipelinetriggerPipeline.go
rename to api/tangled/citriggerPipeline.go
index a45e2754..d372886a 100644
--- a/api/tangled/pipelinetriggerPipeline.go
+++ b/api/tangled/citriggerPipeline.go
@@ -2,7 +2,7 @@
 
 package tangled
 
-// schema: sh.tangled.ci.pipeline.triggerPipeline
+// schema: sh.tangled.ci.triggerPipeline
 
 import (
 	"context"
@@ -11,11 +11,11 @@ import (
 )
 
 const (
-	CiPipelineTriggerPipelineNSID = "sh.tangled.ci.pipeline.triggerPipeline"
+	CiTriggerPipelineNSID = "sh.tangled.ci.triggerPipeline"
 )
 
-// CiPipelineTriggerPipeline_Input is the input argument to a sh.tangled.ci.pipeline.triggerPipeline call.
-type CiPipelineTriggerPipeline_Input struct {
+// CiTriggerPipeline_Input is the input argument to a sh.tangled.ci.triggerPipeline call.
+type CiTriggerPipeline_Input struct {
 	// ref: optional ref the SHA was resolved from, for display
 	Ref *string `json:"ref,omitempty" cborgen:"ref,omitempty"`
 	// repo: AT-URI of the sh.tangled.repo record
@@ -26,16 +26,16 @@ type CiPipelineTriggerPipeline_Input struct {
 	Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"`
 }
 
-// CiPipelineTriggerPipeline_Output is the output of a sh.tangled.ci.pipeline.triggerPipeline call.
-type CiPipelineTriggerPipeline_Output struct {
+// CiTriggerPipeline_Output is the output of a sh.tangled.ci.triggerPipeline call.
+type CiTriggerPipeline_Output struct {
 	// pipeline: AT-URI of the created pipeline
 	Pipeline string `json:"pipeline" cborgen:"pipeline"`
 }
 
-// CiPipelineTriggerPipeline calls the XRPC method "sh.tangled.ci.pipeline.triggerPipeline".
-func CiPipelineTriggerPipeline(ctx context.Context, c util.LexClient, input *CiPipelineTriggerPipeline_Input) (*CiPipelineTriggerPipeline_Output, error) {
-	var out CiPipelineTriggerPipeline_Output
-	if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.pipeline.triggerPipeline", nil, input, &out); err != nil {
+// CiTriggerPipeline calls the XRPC method "sh.tangled.ci.triggerPipeline".
+func CiTriggerPipeline(ctx context.Context, c util.LexClient, input *CiTriggerPipeline_Input) (*CiTriggerPipeline_Output, error) {
+	var out CiTriggerPipeline_Output
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.triggerPipeline", nil, input, &out); err != nil {
 		return nil, err
 	}
 
diff --git a/api/tangled/pipelinestatus.go b/api/tangled/pipelinestatus.go
index 89984f52..10cd2c70 100644
--- a/api/tangled/pipelinestatus.go
+++ b/api/tangled/pipelinestatus.go
@@ -15,7 +15,7 @@ const (
 func init() {
 	util.RegisterType("sh.tangled.pipeline.status", &PipelineStatus{})
 } //
-// DEPRECATED: use sh.tangled.ci.defs#pipeline instead
+// DEPRECATED: use sh.tangled.ci.pipeline instead
 //
 // RECORDTYPE: PipelineStatus
 type PipelineStatus struct {
diff --git a/api/tangled/tangledpipeline.go b/api/tangled/tangledpipeline.go
index b36ebf62..b51b1127 100644
--- a/api/tangled/tangledpipeline.go
+++ b/api/tangled/tangledpipeline.go
@@ -15,7 +15,7 @@ const (
 func init() {
 	util.RegisterType("sh.tangled.pipeline", &Pipeline{})
 } //
-// DEPRECATED: use sh.tangled.ci.defs#pipeline instead
+// DEPRECATED: use sh.tangled.ci.pipeline instead
 //
 // RECORDTYPE: Pipeline
 type Pipeline struct {
diff --git a/appview/oauth/scopes.go b/appview/oauth/scopes.go
index 657953fb..59a6e413 100644
--- a/appview/oauth/scopes.go
+++ b/appview/oauth/scopes.go
@@ -29,8 +29,8 @@ var TangledScopes = []string{
 
 	"rpc:sh.tangled.knot.addMember?aud=*",
 	"rpc:sh.tangled.knot.removeMember?aud=*",
-	"rpc:sh.tangled.ci.pipeline.triggerPipeline?aud=*",
-	"rpc:sh.tangled.pipeline.cancelPipeline?aud=*",
+	"rpc:sh.tangled.ci.triggerPipeline?aud=*",
+	"rpc:sh.tangled.ci.cancelPipeline?aud=*",
 	"rpc:sh.tangled.repo.addCollaborator?aud=*",
 	"rpc:sh.tangled.repo.addSecret?aud=*",
 	"rpc:sh.tangled.repo.create?aud=*",
diff --git a/appview/pipelines/pipelines.go b/appview/pipelines/pipelines.go
index cf7ef87c..48fa2619 100644
--- a/appview/pipelines/pipelines.go
+++ b/appview/pipelines/pipelines.go
@@ -142,7 +142,7 @@ func (p *Pipelines) Index(w http.ResponseWriter, r *http.Request) {
 
 	var pipelines []types.Pipeline
 	for _, pipeline := range out.Pipelines {
-		pipelines = append(pipelines, types.Pipeline{CiDefs_Pipeline: pipeline})
+		pipelines = append(pipelines, types.Pipeline{CiPipeline: pipeline})
 	}
 
 	p.pages.Pipelines(w, pages.PipelinesParams{
@@ -222,7 +222,7 @@ func (p *Pipelines) Workflow(w http.ResponseWriter, r *http.Request) {
 	p.pages.Workflow(w, pages.WorkflowParams{
 		BaseParams: pages.BaseParamsFromContext(r.Context()),
 		RepoInfo:   p.repoResolver.GetRepoInfo(r, user),
-		Pipeline:   types.Pipeline{CiDefs_Pipeline: out},
+		Pipeline:   types.Pipeline{CiPipeline: out},
 		Workflow:   workflowName,
 	})
 }
@@ -233,13 +233,13 @@ var upgrader = websocket.Upgrader{
 }
 
 type webLogScheduler struct {
-	ch chan *tangled.CiPipelineSubscribeLogs_Event
+	ch chan *tangled.CiSubscribePipelineLogs_Event
 }
 
-var _ lexutil.Scheduler[tangled.CiPipelineSubscribeLogs_Event] = (*webLogScheduler)(nil)
+var _ lexutil.Scheduler[tangled.CiSubscribePipelineLogs_Event] = (*webLogScheduler)(nil)
 
 // AddWork implements [lexutil.Scheduler].
-func (w *webLogScheduler) AddWork(ctx context.Context, _ string, val *tangled.CiPipelineSubscribeLogs_Event) error {
+func (w *webLogScheduler) AddWork(ctx context.Context, _ string, val *tangled.CiSubscribePipelineLogs_Event) error {
 	select {
 	case w.ch <- val:
 		return nil
@@ -296,12 +296,12 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	evChan := make(chan *tangled.CiPipelineSubscribeLogs_Event, 100)
+	evChan := make(chan *tangled.CiSubscribePipelineLogs_Event, 100)
 	done := make(chan error, 1)
 	sched := &webLogScheduler{ch: evChan}
 	xrpcc := &lexutil.Client{Client: indigoxrpc.Client{Host: spindleUrl}}
 	go func() {
-		done <- tangled.CiPipelineSubscribeLogs(ctx, xrpcc, pipelineId.String(), []string{workflowName}, sched)
+		done <- tangled.CiSubscribePipelineLogs(ctx, xrpcc, pipelineId.String(), []string{workflowName}, sched)
 	}()
 
 	var lastWriteLk sync.Mutex
@@ -570,16 +570,16 @@ func (p *Pipelines) retry(w http.ResponseWriter, r *http.Request, only string) {
 	}
 	redirectWf := workflows[0]
 
-	spindleClient, err := p.spindleServiceClient(r, f.Spindle, tangled.CiPipelineTriggerPipelineNSID)
+	spindleClient, err := p.spindleServiceClient(r, f.Spindle, tangled.CiTriggerPipelineNSID)
 	if err != nil {
 		fail("failed to authorize with spindle", err)
 		return
 	}
 
-	out, err := tangled.CiPipelineTriggerPipeline(
+	out, err := tangled.CiTriggerPipeline(
 		r.Context(),
 		spindleClient,
-		&tangled.CiPipelineTriggerPipeline_Input{
+		&tangled.CiTriggerPipeline_Input{
 			Repo:      string(f.RepoAt()),
 			Sha:       orig.Commit,
 			Workflows: workflows,
diff --git a/appview/pipelines/ssh/cihelpers.go b/appview/pipelines/ssh/cihelpers.go
index 49dce14b..e0297b1d 100644
--- a/appview/pipelines/ssh/cihelpers.go
+++ b/appview/pipelines/ssh/cihelpers.go
@@ -8,7 +8,7 @@ import (
 
 // helper functions against generated code
 
-func workflowElapsed(wf *tangled.CiDefs_Workflow, now time.Time) time.Duration {
+func workflowElapsed(wf *tangled.CiPipeline_Workflow, now time.Time) time.Duration {
 	if wf.StartedAt == nil {
 		return 0
 	}
@@ -33,7 +33,7 @@ var finishedStatuses = map[string]bool{
 	"success":   true,
 }
 
-func pipelineFinished(p *tangled.CiDefs_Pipeline) bool {
+func pipelineFinished(p *tangled.CiPipeline) bool {
 	for _, wf := range p.Workflows {
 		if !finishedStatuses[wf.Status] {
 			return false
diff --git a/appview/pipelines/ssh/logstream.go b/appview/pipelines/ssh/logstream.go
index f92486ac..1525c325 100644
--- a/appview/pipelines/ssh/logstream.go
+++ b/appview/pipelines/ssh/logstream.go
@@ -22,20 +22,20 @@ type logDoneMsg struct {
 }
 
 type logEventMsg struct {
-	ev     *tangled.CiPipelineSubscribeLogs_Event
-	events chan *tangled.CiPipelineSubscribeLogs_Event
+	ev     *tangled.CiSubscribePipelineLogs_Event
+	events chan *tangled.CiSubscribePipelineLogs_Event
 	done   chan error
 }
 
 type eventScheduler struct {
-	ch chan *tangled.CiPipelineSubscribeLogs_Event
+	ch chan *tangled.CiSubscribePipelineLogs_Event
 }
 
 func newEventScheduler() *eventScheduler {
-	return &eventScheduler{ch: make(chan *tangled.CiPipelineSubscribeLogs_Event, 1024)}
+	return &eventScheduler{ch: make(chan *tangled.CiSubscribePipelineLogs_Event, 1024)}
 }
 
-func (s *eventScheduler) AddWork(ctx context.Context, _ string, v *tangled.CiPipelineSubscribeLogs_Event) error {
+func (s *eventScheduler) AddWork(ctx context.Context, _ string, v *tangled.CiSubscribePipelineLogs_Event) error {
 	select {
 	case s.ch <- v:
 		return nil
diff --git a/appview/pipelines/ssh/tui.go b/appview/pipelines/ssh/tui.go
index 6be0ea87..b06f8441 100644
--- a/appview/pipelines/ssh/tui.go
+++ b/appview/pipelines/ssh/tui.go
@@ -25,7 +25,7 @@ var (
 type tickMsg time.Time
 
 type statusUpdateMsg struct {
-	pipeline *tangled.CiDefs_Pipeline
+	pipeline *tangled.CiPipeline
 }
 
 type statusUpdateErrMsg struct{ err error }
@@ -33,7 +33,7 @@ type statusUpdateErrMsg struct{ err error }
 type pipelineModel struct {
 	renderer *lipgloss.Renderer
 	xrpcc    *extlexutil.Client
-	pipeline *tangled.CiDefs_Pipeline
+	pipeline *tangled.CiPipeline
 	selected int
 	logs     map[string]*workflowLogs
 
@@ -55,7 +55,7 @@ type workflowLogs struct {
 	ready     bool
 }
 
-func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *tangled.CiDefs_Pipeline, width, height int) *pipelineModel {
+func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *tangled.CiPipeline, width, height int) *pipelineModel {
 	logs := make(map[string]*workflowLogs, len(pipeline.Workflows))
 	for _, wf := range pipeline.Workflows {
 		logs[wf.Name] = &workflowLogs{stepIndex: make(map[int64]int)}
@@ -95,14 +95,14 @@ func (m *pipelineModel) subscribeCmd() tea.Cmd {
 
 	pipelineId := m.pipeline.Id
 	go func() {
-		err := tangled.CiPipelineSubscribeLogs(ctx, m.xrpcc, pipelineId, nil, sched)
+		err := tangled.CiSubscribePipelineLogs(ctx, m.xrpcc, pipelineId, nil, sched)
 		done <- err
 	}()
 
 	return readEventCmd(sched.ch, done)
 }
 
-func readEventCmd(events chan *tangled.CiPipelineSubscribeLogs_Event, done chan error) tea.Cmd {
+func readEventCmd(events chan *tangled.CiSubscribePipelineLogs_Event, done chan error) tea.Cmd {
 	return func() tea.Msg {
 		ev, ok := <-events
 		if !ok {
@@ -295,7 +295,7 @@ func (m *pipelineModel) refreshRunning() {
 }
 
 // applyEvent routes a decoded subscribeLogs event into the matching workflow.
-func (m *pipelineModel) applyEvent(ev *tangled.CiPipelineSubscribeLogs_Event) {
+func (m *pipelineModel) applyEvent(ev *tangled.CiSubscribePipelineLogs_Event) {
 	switch {
 	case ev.Error != nil:
 		if ev.Error.Message != "" {
@@ -447,7 +447,7 @@ func shortSha(sha string) string {
 	return sha
 }
 
-func triggerLine(r *lipgloss.Renderer, t *tangled.CiDefs_Pipeline_Trigger, sha string) string {
+func triggerLine(r *lipgloss.Renderer, t *tangled.CiPipeline_Trigger, sha string) string {
 	hash := shortSha(sha)
 	dim := r.NewStyle().Faint(true)
 	if t == nil {
diff --git a/appview/pulls/list.go b/appview/pulls/list.go
index 50925bc8..a9b7ef8d 100644
--- a/appview/pulls/list.go
+++ b/appview/pulls/list.go
@@ -286,7 +286,7 @@ func (s *Pulls) RepoPulls(w http.ResponseWriter, r *http.Request) {
 			if pipeline == nil {
 				continue
 			}
-			m[pipeline.Commit] = types.Pipeline{CiDefs_Pipeline: pipeline}
+			m[pipeline.Commit] = types.Pipeline{CiPipeline: pipeline}
 		}
 		return m
 	}(r.Context(), shas)
diff --git a/appview/pulls/single.go b/appview/pulls/single.go
index 3f6e0b57..4c1427c8 100644
--- a/appview/pulls/single.go
+++ b/appview/pulls/single.go
@@ -182,7 +182,7 @@ func (s *Pulls) repoPullHelper(w http.ResponseWriter, r *http.Request, interdiff
 			if pipeline == nil {
 				continue
 			}
-			m[pipeline.Commit] = types.Pipeline{CiDefs_Pipeline: pipeline}
+			m[pipeline.Commit] = types.Pipeline{CiPipeline: pipeline}
 		}
 		return m
 	}(r.Context())
diff --git a/appview/repo/repo_util.go b/appview/repo/repo_util.go
index 6dc96bae..6d0d5092 100644
--- a/appview/repo/repo_util.go
+++ b/appview/repo/repo_util.go
@@ -116,7 +116,7 @@ func getPipelineStatuses(
 	}
 
 	for _, p := range out.Pipelines {
-		m[p.Commit] = types.Pipeline{CiDefs_Pipeline: p}
+		m[p.Commit] = types.Pipeline{CiPipeline: p}
 	}
 
 	return m, nil
diff --git a/cmd/cborgen/cborgen.go b/cmd/cborgen/cborgen.go
index 3cbd17e2..50eceb01 100644
--- a/cmd/cborgen/cborgen.go
+++ b/cmd/cborgen/cborgen.go
@@ -15,11 +15,11 @@ func main() {
 		"api/tangled/cbor_gen.go",
 		"tangled",
 		tangled.ActorProfile{},
-		tangled.CiDefs_Pipeline{},
-		tangled.CiDefs_Pipeline_Trigger{},
-		tangled.CiDefs_Workflow{},
-		tangled.CiPipelineSubscribeLogs_Control{},
-		tangled.CiPipelineSubscribeLogs_Data{},
+		tangled.CiPipeline{},
+		tangled.CiPipeline_Trigger{},
+		tangled.CiPipeline_Workflow{},
+		tangled.CiSubscribePipelineLogs_Control{},
+		tangled.CiSubscribePipelineLogs_Data{},
 		tangled.CiTrigger_Manual{},
 		tangled.CiTrigger_PullRequest{},
 		tangled.CiTrigger_Push{},
diff --git a/eventconsumer/upgrade_test.go b/eventconsumer/upgrade_test.go
index 3c90f34b..6376645b 100644
--- a/eventconsumer/upgrade_test.go
+++ b/eventconsumer/upgrade_test.go
@@ -22,7 +22,7 @@ func sqliteCursorStore(t *testing.T) cursor.Store {
 	return store
 }
 
-func drainProcessed(t *testing.T, store cursor.Store, source Source) []int64 {
+func drainProcessed(t *testing.T, store cursor.Store, source Source, expected int) []int64 {
 	t.Helper()
 
 	var mu sync.Mutex
@@ -43,26 +43,23 @@ func drainProcessed(t *testing.T, store cursor.Store, source Source) []int64 {
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
-	t.Cleanup(cancel)
+	defer cancel()
 	c.Start(ctx)
 	c.AddSource(ctx, source)
 
 	deadline := time.Now().Add(3 * time.Second)
-	last, stable := -1, 0
 	for time.Now().Before(deadline) {
-		time.Sleep(100 * time.Millisecond)
 		mu.Lock()
 		n := len(seen)
 		mu.Unlock()
-		if n == last {
-			if stable++; stable >= 3 && n > 0 {
-				break
-			}
-		} else {
-			last, stable = n, 0
+		if n >= expected {
+			break
 		}
+		time.Sleep(20 * time.Millisecond)
 	}
 
+	c.Stop()
+
 	mu.Lock()
 	defer mu.Unlock()
 	return append([]int64(nil), seen...)
@@ -78,7 +75,7 @@ func TestSpindleUpgrade_OrphanedCursorReplaysFromZero(t *testing.T) {
 	store := sqliteCursorStore(t)
 	store.Set(source.Host, 5)
 
-	seen := drainProcessed(t, store, source)
+	seen := drainProcessed(t, store, source, 8)
 
 	if len(seen) != 8 {
 		t.Fatalf("orphaned bare-host cursor processed %d events, want a full replay of 8: %v", len(seen), seen)
@@ -97,7 +94,7 @@ func TestSpindleUpgrade_MigratedCursorResumesNoReplay(t *testing.T) {
 
 	MigrateLegacyCursor(store, source)
 
-	seen := drainProcessed(t, store, source)
+	seen := drainProcessed(t, store, source, 3)
 
 	if len(seen) != 3 {
 		t.Fatalf("migrated cursor processed %d events, want a resume of 3: %v", len(seen), seen)
diff --git a/lexicons/ci/pipeline/cancelPipeline.json b/lexicons/ci/cancelPipeline.json
similarity index 95%
rename from lexicons/ci/pipeline/cancelPipeline.json
rename to lexicons/ci/cancelPipeline.json
index 69e26999..81dc576c 100644
--- a/lexicons/ci/pipeline/cancelPipeline.json
+++ b/lexicons/ci/cancelPipeline.json
@@ -1,6 +1,6 @@
 {
   "lexicon": 1,
-  "id": "sh.tangled.ci.pipeline.cancelPipeline",
+  "id": "sh.tangled.ci.cancelPipeline",
   "defs": {
     "main": {
       "type": "procedure",
diff --git a/lexicons/ci/getPipeline.json b/lexicons/ci/getPipeline.json
index 9e66e14e..379ce71f 100644
--- a/lexicons/ci/getPipeline.json
+++ b/lexicons/ci/getPipeline.json
@@ -19,7 +19,7 @@
         "encoding": "application/json",
         "schema": {
           "type": "ref",
-          "ref": "sh.tangled.ci.defs#pipeline"
+          "ref": "sh.tangled.ci.pipeline"
         }
       },
       "errors": [
diff --git a/lexicons/ci/defs.json b/lexicons/ci/pipeline.json
similarity index 93%
rename from lexicons/ci/defs.json
rename to lexicons/ci/pipeline.json
index 06a9f8c0..ef89839f 100644
--- a/lexicons/ci/defs.json
+++ b/lexicons/ci/pipeline.json
@@ -1,9 +1,10 @@
 {
   "lexicon": 1,
-  "id": "sh.tangled.ci.defs",
+  "id": "sh.tangled.ci.pipeline",
   "defs": {
-    "pipeline": {
+    "main": {
       "type": "object",
+      "description": "A CI pipeline. Record-like, but owned by the spindle rather than a PDS.",
       "required": ["id", "trigger", "commit", "workflows"],
       "properties": {
         "id": {
diff --git a/lexicons/ci/queryPipelines.json b/lexicons/ci/queryPipelines.json
index deb80b05..15fcf92a 100644
--- a/lexicons/ci/queryPipelines.json
+++ b/lexicons/ci/queryPipelines.json
@@ -48,7 +48,7 @@
             },
             "pipelines": {
               "type": "array",
-              "items": { "type": "ref", "ref": "sh.tangled.ci.defs#pipeline" }
+              "items": { "type": "ref", "ref": "sh.tangled.ci.pipeline" }
             }
           }
         }
diff --git a/lexicons/ci/pipeline/subscribeLogs.json b/lexicons/ci/subscribePipelineLogs.json
similarity index 97%
rename from lexicons/ci/pipeline/subscribeLogs.json
rename to lexicons/ci/subscribePipelineLogs.json
index 4d63e834..f82e848a 100644
--- a/lexicons/ci/pipeline/subscribeLogs.json
+++ b/lexicons/ci/subscribePipelineLogs.json
@@ -1,6 +1,6 @@
 {
   "lexicon": 1,
-  "id": "sh.tangled.ci.pipeline.subscribeLogs",
+  "id": "sh.tangled.ci.subscribePipelineLogs",
   "defs": {
     "main": {
       "type": "subscription",
diff --git a/lexicons/ci/pipeline/triggerPipeline.json b/lexicons/ci/triggerPipeline.json
similarity index 97%
rename from lexicons/ci/pipeline/triggerPipeline.json
rename to lexicons/ci/triggerPipeline.json
index d40238db..2abe12b6 100644
--- a/lexicons/ci/pipeline/triggerPipeline.json
+++ b/lexicons/ci/triggerPipeline.json
@@ -1,6 +1,6 @@
 {
   "lexicon": 1,
-  "id": "sh.tangled.ci.pipeline.triggerPipeline",
+  "id": "sh.tangled.ci.triggerPipeline",
   "defs": {
     "main": {
       "type": "procedure",
diff --git a/lexicons/pipeline/cancelPipeline.json b/lexicons/pipeline/cancelPipeline.json
index 2feda77a..12b9b951 100644
--- a/lexicons/pipeline/cancelPipeline.json
+++ b/lexicons/pipeline/cancelPipeline.json
@@ -4,7 +4,7 @@
   "defs": {
     "main": {
       "type": "procedure",
-      "description": "DEPRECATED: use sh.tangled.ci.pipeline.cancelPipeline instead - Cancel a running pipeline",
+      "description": "DEPRECATED: use sh.tangled.ci.cancelPipeline instead - Cancel a running pipeline",
       "input": {
         "encoding": "application/json",
         "schema": {
diff --git a/lexicons/pipeline/pipeline.json b/lexicons/pipeline/pipeline.json
index 04d2e5b9..5a6e4954 100644
--- a/lexicons/pipeline/pipeline.json
+++ b/lexicons/pipeline/pipeline.json
@@ -9,7 +9,7 @@
       "key": "tid",
       "record": {
         "type": "object",
-        "description": "DEPRECATED: use sh.tangled.ci.defs#pipeline instead",
+        "description": "DEPRECATED: use sh.tangled.ci.pipeline instead",
         "required": [
           "triggerMetadata",
           "workflows"
diff --git a/lexicons/pipeline/status.json b/lexicons/pipeline/status.json
index 3106837f..a9689f71 100644
--- a/lexicons/pipeline/status.json
+++ b/lexicons/pipeline/status.json
@@ -9,7 +9,7 @@
       "key": "tid",
       "record": {
         "type": "object",
-        "description": "DEPRECATED: use sh.tangled.ci.defs#pipeline instead",
+        "description": "DEPRECATED: use sh.tangled.ci.pipeline instead",
         "required": ["pipeline", "workflow", "status", "createdAt"],
         "properties": {
           "pipeline": {
diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go
index 4c6cebed..d2f4556b 100644
--- a/spindle/db/pipelines.go
+++ b/spindle/db/pipelines.go
@@ -11,7 +11,7 @@ import (
 	"tangled.org/core/spindle/models"
 )
 
-func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []string, cursor string, limit int) ([]*tangled.CiDefs_Pipeline, string, int64, error) {
+func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []string, cursor string, limit int) ([]*tangled.CiPipeline, string, int64, error) {
 	if limit <= 0 {
 		limit = 30
 	}
@@ -59,7 +59,7 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []strin
 	}
 	defer rows.Close()
 
-	var pipelines []*tangled.CiDefs_Pipeline
+	var pipelines []*tangled.CiPipeline
 	var lastCreated int64
 
 	for rows.Next() {
@@ -75,7 +75,7 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []strin
 			continue
 		}
 
-		p, err := d.mapToCiDefsPipeline(ctx, rkey, created, rawPipeline)
+		p, err := d.mapToCiPipeline(ctx, rkey, created, rawPipeline)
 		if err != nil {
 			return nil, "", 0, err
 		}
@@ -90,7 +90,7 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []strin
 	return pipelines, nextCursor, total, nil
 }
 
-func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiDefs_Pipeline, error) {
+func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiPipeline, error) {
 	var eventJson string
 	var created int64
 	err := d.QueryRowContext(ctx,
@@ -113,10 +113,10 @@ func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiDefs_Pipe
 		return nil, err
 	}
 
-	return d.mapToCiDefsPipeline(ctx, rkey, created, rawPipeline)
+	return d.mapToCiPipeline(ctx, rkey, created, rawPipeline)
 }
 
-func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64, raw tangled.Pipeline) (*tangled.CiDefs_Pipeline, error) {
+func (d *DB) mapToCiPipeline(ctx context.Context, rkey string, created int64, raw tangled.Pipeline) (*tangled.CiPipeline, error) {
 	createdAtStr := time.Unix(0, created).Format(time.RFC3339)
 
 	var repoDidStr string
@@ -129,7 +129,7 @@ func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64
 	}
 
 	commitSha := ""
-	var trigger tangled.CiDefs_Pipeline_Trigger
+	var trigger tangled.CiPipeline_Trigger
 
 	if raw.TriggerMetadata != nil {
 		switch raw.TriggerMetadata.Kind {
@@ -160,7 +160,7 @@ func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64
 		}
 	}
 
-	var workflows []*tangled.CiDefs_Workflow
+	var workflows []*tangled.CiPipeline_Workflow
 	for _, wf := range raw.Workflows {
 		status := "pending"
 		var startedAt, finishedAt, wfError *string
@@ -182,7 +182,7 @@ func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64
 			}
 		}
 
-		workflows = append(workflows, &tangled.CiDefs_Workflow{
+		workflows = append(workflows, &tangled.CiPipeline_Workflow{
 			Id:         wf.Name,
 			Name:       wf.Name,
 			Status:     status,
@@ -192,7 +192,7 @@ func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64
 		})
 	}
 
-	return &tangled.CiDefs_Pipeline{
+	return &tangled.CiPipeline{
 		Id:        rkey,
 		Commit:    commitSha,
 		Repo:      &repoDidStr,
diff --git a/spindle/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go
index 7b4810ed..e0b32fc0 100644
--- a/spindle/xrpc/ci_pipeline_subscribe_logs.go
+++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go
@@ -17,7 +17,7 @@ import (
 	"tangled.org/core/spindle/models"
 )
 
-func (x *Xrpc) HandleCiPipelineSubscribeLogs(w http.ResponseWriter, r *http.Request) {
+func (x *Xrpc) HandleCiSubscribePipelineLogs(w http.ResponseWriter, r *http.Request) {
 	var (
 		pipelineQuery = r.URL.Query().Get("pipeline")
 		workflows     = r.URL.Query()["workflows"]
@@ -142,7 +142,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel
 		}
 	}()
 
-	eventsChan := make(chan tangled.CiPipelineSubscribeLogs_Event, 128)
+	eventsChan := make(chan tangled.CiSubscribePipelineLogs_Event, 128)
 	wg := sync.WaitGroup{}
 
 	// 4. start a tail reader goroutine for each workflow
@@ -225,7 +225,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel
 						logLine = models.NewDataLogLine(0, line.Text, "stdout")
 					}
 
-					var ev tangled.CiPipelineSubscribeLogs_Event
+					var ev tangled.CiSubscribePipelineLogs_Event
 					timeStr := logLine.Time.Format(time.RFC3339)
 					if logLine.Time.IsZero() {
 						timeStr = time.Now().Format(time.RFC3339)
@@ -236,7 +236,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel
 						if logLine.StepKind == models.StepKindSystem {
 							stepKindStr = "system"
 						}
-						ev = tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{
+						ev = tangled.CiSubscribePipelineLogs_Event{Control: &tangled.CiSubscribePipelineLogs_Control{
 							Time:     timeStr,
 							Workflow: wfName,
 							Step:     int64(logLine.StepId),
@@ -250,7 +250,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel
 						if streamType != "stdout" && streamType != "stderr" {
 							streamType = "stdout"
 						}
-						ev = tangled.CiPipelineSubscribeLogs_Event{Data: &tangled.CiPipelineSubscribeLogs_Data{
+						ev = tangled.CiSubscribePipelineLogs_Event{Data: &tangled.CiSubscribePipelineLogs_Data{
 							Time:     timeStr,
 							Workflow: wfName,
 							Step:     int64(logLine.StepId),
diff --git a/spindle/xrpc/ci_pipeline_trigger_pipeline.go b/spindle/xrpc/ci_pipeline_trigger_pipeline.go
index 93da3386..669c3180 100644
--- a/spindle/xrpc/ci_pipeline_trigger_pipeline.go
+++ b/spindle/xrpc/ci_pipeline_trigger_pipeline.go
@@ -30,7 +30,7 @@ func (x *Xrpc) TriggerPipeline(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var input tangled.CiPipelineTriggerPipeline_Input
+	var input tangled.CiTriggerPipeline_Input
 	if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
 		fail(xrpcerr.GenericError(err))
 		return
@@ -62,7 +62,7 @@ func (x *Xrpc) TriggerPipeline(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if err := writeJson(w, http.StatusOK, tangled.CiPipelineTriggerPipeline_Output{
+	if err := writeJson(w, http.StatusOK, tangled.CiTriggerPipeline_Output{
 		Pipeline: pipelineAt.String(),
 	}); err != nil {
 		l.Error("failed to write response", "err", err)
diff --git a/spindle/xrpc/xrpc.go b/spindle/xrpc/xrpc.go
index a2c5739f..11bcff93 100644
--- a/spindle/xrpc/xrpc.go
+++ b/spindle/xrpc/xrpc.go
@@ -59,13 +59,13 @@ func (x *Xrpc) Router() http.Handler {
 		r.Post("/"+tangled.RepoAddSecretNSID, x.AddSecret)
 		r.Post("/"+tangled.RepoRemoveSecretNSID, x.RemoveSecret)
 		r.Get("/"+tangled.RepoListSecretsNSID, x.ListSecrets)
-		r.Post("/"+tangled.CiPipelineCancelPipelineNSID, x.CancelPipeline)
-		r.Post("/"+tangled.CiPipelineTriggerPipelineNSID, x.TriggerPipeline)
+		r.Post("/"+tangled.CiCancelPipelineNSID, x.CancelPipeline)
+		r.Post("/"+tangled.CiTriggerPipelineNSID, x.TriggerPipeline)
 	})
 
 	// service query endpoints (no auth required)
 	r.Get("/"+tangled.OwnerNSID, x.Owner)
-	r.Get("/"+tangled.CiPipelineSubscribeLogsNSID, x.HandleCiPipelineSubscribeLogs)
+	r.Get("/"+tangled.CiSubscribePipelineLogsNSID, x.HandleCiSubscribePipelineLogs)
 	r.Get("/"+tangled.CiQueryPipelinesNSID, x.HandleCiQueryPipelines)
 	r.Get("/"+tangled.CiGetPipelineNSID, x.HandleCiGetPipeline)
 
diff --git a/types/pipeline.go b/types/pipeline.go
index f1b56519..68e0cfb3 100644
--- a/types/pipeline.go
+++ b/types/pipeline.go
@@ -31,7 +31,7 @@ func (s StatusKind) IsStart() bool {
 }
 
 type WorkflowStatus struct {
-	*tangled.CiDefs_Workflow
+	*tangled.CiPipeline_Workflow
 	PipelineCreatedAt *string
 }
 
@@ -40,21 +40,21 @@ func (w WorkflowStatus) Latest() WorkflowStatus {
 }
 
 func (w WorkflowStatus) Error() string {
-	if w.CiDefs_Workflow == nil || w.CiDefs_Workflow.Error == nil {
+	if w.CiPipeline_Workflow == nil || w.CiPipeline_Workflow.Error == nil {
 		return ""
 	}
-	return *w.CiDefs_Workflow.Error
+	return *w.CiPipeline_Workflow.Error
 }
 
 func (w WorkflowStatus) Status() StatusKind {
-	if w.CiDefs_Workflow == nil {
+	if w.CiPipeline_Workflow == nil {
 		return ""
 	}
-	return StatusKind(w.CiDefs_Workflow.Status)
+	return StatusKind(w.CiPipeline_Workflow.Status)
 }
 
 func (w WorkflowStatus) TimeTaken() time.Duration {
-	if w.CiDefs_Workflow == nil || w.StartedAt == nil || w.FinishedAt == nil || *w.StartedAt == "" || *w.FinishedAt == "" {
+	if w.CiPipeline_Workflow == nil || w.StartedAt == nil || w.FinishedAt == nil || *w.StartedAt == "" || *w.FinishedAt == "" {
 		return 0
 	}
 	t1, err1 := time.Parse(time.RFC3339, *w.StartedAt)
@@ -67,9 +67,9 @@ func (w WorkflowStatus) TimeTaken() time.Duration {
 
 func (w WorkflowStatus) Created() time.Time {
 	var timeStr string
-	if w.CiDefs_Workflow != nil && w.StartedAt != nil && *w.StartedAt != "" {
+	if w.CiPipeline_Workflow != nil && w.StartedAt != nil && *w.StartedAt != "" {
 		timeStr = *w.StartedAt
-	} else if w.CiDefs_Workflow != nil && w.FinishedAt != nil && *w.FinishedAt != "" {
+	} else if w.CiPipeline_Workflow != nil && w.FinishedAt != nil && *w.FinishedAt != "" {
 		timeStr = *w.FinishedAt
 	} else if w.PipelineCreatedAt != nil && *w.PipelineCreatedAt != "" {
 		timeStr = *w.PipelineCreatedAt
@@ -85,27 +85,27 @@ func (w WorkflowStatus) Created() time.Time {
 }
 
 type Trigger struct {
-	*tangled.CiDefs_Pipeline_Trigger
+	*tangled.CiPipeline_Trigger
 }
 
 func (t Trigger) IsPush() bool {
-	return t.CiDefs_Pipeline_Trigger != nil && t.CiDefs_Pipeline_Trigger.CiTrigger_Push != nil
+	return t.CiPipeline_Trigger != nil && t.CiPipeline_Trigger.CiTrigger_Push != nil
 }
 
 func (t Trigger) IsPullRequest() bool {
-	return t.CiDefs_Pipeline_Trigger != nil && t.CiDefs_Pipeline_Trigger.CiTrigger_PullRequest != nil
+	return t.CiPipeline_Trigger != nil && t.CiPipeline_Trigger.CiTrigger_PullRequest != nil
 }
 
 func (t Trigger) IsManual() bool {
-	return t.CiDefs_Pipeline_Trigger != nil && t.CiDefs_Pipeline_Trigger.CiTrigger_Manual != nil
+	return t.CiPipeline_Trigger != nil && t.CiPipeline_Trigger.CiTrigger_Manual != nil
 }
 
 func (t Trigger) TargetRef() string {
-	if t.CiDefs_Pipeline_Trigger == nil {
+	if t.CiPipeline_Trigger == nil {
 		return ""
 	}
-	if t.CiDefs_Pipeline_Trigger.CiTrigger_Push != nil {
-		ref := t.CiDefs_Pipeline_Trigger.CiTrigger_Push.Ref
+	if t.CiPipeline_Trigger.CiTrigger_Push != nil {
+		ref := t.CiPipeline_Trigger.CiTrigger_Push.Ref
 		if strings.HasPrefix(ref, "refs/heads/") {
 			return strings.TrimPrefix(ref, "refs/heads/")
 		}
@@ -114,17 +114,17 @@ func (t Trigger) TargetRef() string {
 		}
 		return ref
 	}
-	if t.CiDefs_Pipeline_Trigger.CiTrigger_PullRequest != nil {
-		return t.CiDefs_Pipeline_Trigger.CiTrigger_PullRequest.TargetBranch
+	if t.CiPipeline_Trigger.CiTrigger_PullRequest != nil {
+		return t.CiPipeline_Trigger.CiTrigger_PullRequest.TargetBranch
 	}
 	return ""
 }
 
 func (t Trigger) PRSourceBranch() string {
-	if t.CiDefs_Pipeline_Trigger == nil || t.CiDefs_Pipeline_Trigger.CiTrigger_PullRequest == nil {
+	if t.CiPipeline_Trigger == nil || t.CiPipeline_Trigger.CiTrigger_PullRequest == nil {
 		return ""
 	}
-	sb := t.CiDefs_Pipeline_Trigger.CiTrigger_PullRequest.SourceBranch
+	sb := t.CiPipeline_Trigger.CiTrigger_PullRequest.SourceBranch
 	if sb == nil {
 		return ""
 	}
@@ -132,27 +132,27 @@ func (t Trigger) PRSourceBranch() string {
 }
 
 type Pipeline struct {
-	*tangled.CiDefs_Pipeline
+	*tangled.CiPipeline
 }
 
 func (p Pipeline) Valid() bool {
-	return p.CiDefs_Pipeline != nil
+	return p.CiPipeline != nil
 }
 
 func (p Pipeline) Id() string {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return ""
 	}
-	return p.CiDefs_Pipeline.Id
+	return p.CiPipeline.Id
 }
 
 func (p Pipeline) Statuses() map[string]WorkflowStatus {
 	m := make(map[string]WorkflowStatus)
-	if p.CiDefs_Pipeline != nil {
-		for _, w := range p.CiDefs_Pipeline.Workflows {
+	if p.CiPipeline != nil {
+		for _, w := range p.CiPipeline.Workflows {
 			m[w.Name] = WorkflowStatus{
-				CiDefs_Workflow:   w,
-				PipelineCreatedAt: p.CreatedAt,
+				CiPipeline_Workflow: w,
+				PipelineCreatedAt:   p.CreatedAt,
 			}
 		}
 	}
@@ -161,8 +161,8 @@ func (p Pipeline) Statuses() map[string]WorkflowStatus {
 
 func (p Pipeline) Counts() map[string]int {
 	m := make(map[string]int)
-	if p.CiDefs_Pipeline != nil {
-		for _, w := range p.CiDefs_Pipeline.Workflows {
+	if p.CiPipeline != nil {
+		for _, w := range p.CiPipeline.Workflows {
 			m[w.Status]++
 		}
 	}
@@ -177,21 +177,21 @@ func (p Pipeline) InProgress() bool {
 }
 
 func (p Pipeline) ShortStatusSummary() string {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return ""
 	}
 	counts := p.Counts()
-	total := len(p.CiDefs_Pipeline.Workflows)
+	total := len(p.CiPipeline.Workflows)
 	successes := counts["success"]
 	return fmt.Sprintf("%d/%d", successes, total)
 }
 
 func (p Pipeline) LongStatusSummary() string {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return ""
 	}
 	counts := p.Counts()
-	total := len(p.CiDefs_Pipeline.Workflows)
+	total := len(p.CiPipeline.Workflows)
 	var parts []string
 	states := []string{"success", "failed", "timeout", "cancelled", "running", "pending"}
 	for _, state := range states {
@@ -203,18 +203,18 @@ func (p Pipeline) LongStatusSummary() string {
 }
 
 func (p Pipeline) TimeTaken() time.Duration {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return 0
 	}
 	var s time.Duration
-	for _, w := range p.CiDefs_Pipeline.Workflows {
-		s += WorkflowStatus{CiDefs_Workflow: w}.TimeTaken()
+	for _, w := range p.CiPipeline.Workflows {
+		s += WorkflowStatus{CiPipeline_Workflow: w}.TimeTaken()
 	}
 	return s
 }
 
 func (p Pipeline) Created() time.Time {
-	if p.CiDefs_Pipeline == nil || p.CreatedAt == nil || *p.CreatedAt == "" {
+	if p.CiPipeline == nil || p.CreatedAt == nil || *p.CreatedAt == "" {
 		return time.Time{}
 	}
 	t, err := time.Parse(time.RFC3339, *p.CreatedAt)
@@ -225,27 +225,27 @@ func (p Pipeline) Created() time.Time {
 }
 
 func (p Pipeline) Trigger() Trigger {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return Trigger{nil}
 	}
-	return Trigger{p.CiDefs_Pipeline.Trigger}
+	return Trigger{p.CiPipeline.Trigger}
 }
 
 func (p Pipeline) IsResponding() bool {
-	return p.CiDefs_Pipeline != nil && len(p.CiDefs_Pipeline.Workflows) > 0
+	return p.CiPipeline != nil && len(p.CiPipeline.Workflows) > 0
 }
 
 func (p Pipeline) Sha() string {
-	if p.CiDefs_Pipeline == nil {
+	if p.CiPipeline == nil {
 		return ""
 	}
-	return p.CiDefs_Pipeline.Commit
+	return p.CiPipeline.Commit
 }
 
 func (p Pipeline) Workflows() []string {
 	var ws []string
-	if p.CiDefs_Pipeline != nil {
-		for _, w := range p.CiDefs_Pipeline.Workflows {
+	if p.CiPipeline != nil {
+		for _, w := range p.CiPipeline.Workflows {
 			ws = append(ws, w.Name)
 		}
 	}