diff --git a/api/org_tangled/cipipeline_ext.go b/api/org_tangled/cipipeline_ext.go
new file mode 100644
index 000000000..b4db60a64
--- /dev/null
+++ b/api/org_tangled/cipipeline_ext.go
@@ -0,0 +1,9 @@
+package org_tangled
+
+import lexutil "github.com/bluesky-social/indigo/lex/util"
+
+func init() {
+	lexutil.RegisterType(EventPushNSID, &EventPush{})
+	lexutil.RegisterType(EventPullRequestNSID, &EventPullRequest{})
+	lexutil.RegisterType(CiTriggerManualNSID, &CiTriggerManual{})
+}
diff --git a/api/org_tangled/cipipeline_test.go b/api/org_tangled/cipipeline_test.go
new file mode 100644
index 000000000..b2a78eea8
--- /dev/null
+++ b/api/org_tangled/cipipeline_test.go
@@ -0,0 +1,70 @@
+package org_tangled
+
+import (
+	"encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+)
+
+func TestCiPipelineTriggerJSON(t *testing.T) {
+	cases := []struct {
+		name    string
+		trigger CiPipeline_Trigger
+		typeID  string
+	}{
+		{"push", CiPipeline_Trigger{EventPush: &EventPush{}}, EventPushNSID},
+		{"pull_request", CiPipeline_Trigger{EventPullRequest: &EventPullRequest{}}, EventPullRequestNSID},
+		{"manual", CiPipeline_Trigger{CiTriggerManual: &CiTriggerManual{Repo: "did:plc:repo", Commit: strings.Repeat("1", 40)}}, CiTriggerManualNSID},
+	}
+	for _, test := range cases {
+		t.Run(test.name, func(t *testing.T) {
+			encoded, err := json.Marshal(&test.trigger)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !strings.Contains(string(encoded), `"$type":"`+test.typeID+`"`) {
+				t.Fatalf("trigger JSON = %s", encoded)
+			}
+		})
+	}
+}
+
+func TestCiPipelineTriggerRoundTripsUnknownType(t *testing.T) {
+	original := []byte(`{"$type":"org.tangled.event.future","detail":{"value":1}}`)
+	var trigger CiPipeline_Trigger
+	if err := json.Unmarshal(original, &trigger); err != nil {
+		t.Fatal(err)
+	}
+	if trigger.EventPush != nil || trigger.EventPullRequest != nil || trigger.CiTriggerManual != nil || trigger.Unknown == nil {
+		t.Fatalf("unknown trigger = %+v", trigger)
+	}
+	encoded, err := json.Marshal(&trigger)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var originalValue, encodedValue any
+	if err := json.Unmarshal(original, &originalValue); err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(encoded, &encodedValue); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(encodedValue, originalValue) {
+		t.Fatalf("round trip = %s, want %s", encoded, original)
+	}
+}
+
+func TestCanonicalTriggerTypesAreRegistered(t *testing.T) {
+	for _, typeID := range []string{EventPushNSID, EventPullRequestNSID, CiTriggerManualNSID} {
+		var decoded lexutil.LexiconTypeDecoder
+		if err := json.Unmarshal([]byte(`{"$type":"`+typeID+`","repo":"did:plc:repo","commit":"1111111111111111111111111111111111111111"}`), &decoded); err != nil {
+			t.Fatalf("decode %s: %v", typeID, err)
+		}
+		if decoded.Val == nil {
+			t.Fatalf("decode %s returned no value", typeID)
+		}
+	}
+}
diff --git a/api/org_tangled/cisubscribePipelineLogs_ext.go b/api/org_tangled/cisubscribePipelineLogs_ext.go
new file mode 100644
index 000000000..5a0bfdca7
--- /dev/null
+++ b/api/org_tangled/cisubscribePipelineLogs_ext.go
@@ -0,0 +1,98 @@
+// extending code generated from sh.tangled.ci.subscribePipelineLogs
+
+package org_tangled
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/bluesky-social/indigo/events"
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+	cbg "github.com/whyrusleeping/cbor-gen"
+	extlexutil "tangled.org/core/lexutil"
+)
+
+// TODO: generate codes below from lexicon
+type CiSubscribePipelineLogs_Event struct {
+	Error   *events.ErrorFrame
+	Control *CiSubscribePipelineLogs_Control
+	Data    *CiSubscribePipelineLogs_Data
+
+	// some private fields for internal routing perf
+	Preserialized []byte `json:"-" cborgen:"-"`
+}
+
+func (xevt *CiSubscribePipelineLogs_Event) Serialize(wc io.Writer) error {
+	header := events.EventHeader{Op: events.EvtKindMessage}
+	var obj lexutil.CBOR
+
+	switch {
+	case xevt.Error != nil:
+		header.Op = events.EvtKindErrorFrame
+		obj = xevt.Error
+	case xevt.Control != nil:
+		header.MsgType = "#control"
+		obj = xevt.Control
+	case xevt.Data != nil:
+		header.MsgType = "#data"
+		obj = xevt.Data
+	default:
+		return fmt.Errorf("unrecognized event kind")
+	}
+
+	cborWriter := cbg.NewCborWriter(wc)
+	if err := header.MarshalCBOR(cborWriter); err != nil {
+		return fmt.Errorf("failed to write header: %w", err)
+	}
+	return obj.MarshalCBOR(cborWriter)
+}
+
+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)
+	}
+	switch header.Op {
+	case events.EvtKindMessage:
+		switch header.MsgType {
+		case "#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 CiSubscribePipelineLogs_Data
+			if err := evt.UnmarshalCBOR(r); err != nil {
+				return fmt.Errorf("reading repoSync event: %w", err)
+			}
+			xevt.Data = &evt
+		}
+	case events.EvtKindErrorFrame:
+		var errframe events.ErrorFrame
+		if err := errframe.UnmarshalCBOR(r); err != nil {
+			return err
+		}
+		xevt.Error = &errframe
+	default:
+		return fmt.Errorf("unrecognized event stream type: %d", header.Op)
+	}
+	return nil
+}
+
+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, "", CiSubscribePipelineLogsNSID, params, nil, func(ctx context.Context, cr *cbg.CborReader) error {
+		var evt CiSubscribePipelineLogs_Event
+		if err := evt.Deserialize(cr); err != nil {
+			return err
+		}
+		return sched.AddWork(ctx, "", &evt)
+	})
+}
diff --git a/api/org_tangled/cisubscribePipelineLogs_ext_test.go b/api/org_tangled/cisubscribePipelineLogs_ext_test.go
new file mode 100644
index 000000000..3974e94db
--- /dev/null
+++ b/api/org_tangled/cisubscribePipelineLogs_ext_test.go
@@ -0,0 +1,53 @@
+package org_tangled
+
+import (
+	"bytes"
+	"testing"
+
+	"tangled.org/core/api/tangled"
+)
+
+func TestCiSubscribePipelineLogsEventRoundTrip(t *testing.T) {
+	roundTrip := func(want CiSubscribePipelineLogs_Event) CiSubscribePipelineLogs_Event {
+		t.Helper()
+		var buf bytes.Buffer
+		if err := want.Serialize(&buf); err != nil {
+			t.Fatal(err)
+		}
+		var got CiSubscribePipelineLogs_Event
+		if err := got.Deserialize(&buf); err != nil {
+			t.Fatal(err)
+		}
+		return got
+	}
+
+	kind := "user"
+	got := roundTrip(CiSubscribePipelineLogs_Event{Control: &CiSubscribePipelineLogs_Control{
+		Content: "building", Kind: &kind, Step: 2, Time: "2026-08-30T00:00:00Z", Workflow: "ci.yml",
+	}})
+	if got.Control == nil || got.Control.Content != "building" || got.Control.Workflow != "ci.yml" || got.Control.Step != 2 {
+		t.Fatalf("control = %+v", got.Control)
+	}
+
+	got = roundTrip(CiSubscribePipelineLogs_Event{Data: &CiSubscribePipelineLogs_Data{
+		Content: "hello", Step: 1, Stream: "stdout", Time: "2026-08-30T00:00:00Z", Workflow: "ci.yml",
+	}})
+	if got.Data == nil || got.Data.Content != "hello" || got.Data.Stream != "stdout" {
+		t.Fatalf("data = %+v", got.Data)
+	}
+
+	var buf bytes.Buffer
+	legacy := tangled.CiSubscribePipelineLogs_Event{Data: &tangled.CiSubscribePipelineLogs_Data{
+		Content: "legacy", Step: 3, Stream: "stderr", Time: "2026-08-30T00:00:00Z", Workflow: "ci.yml",
+	}}
+	if err := legacy.Serialize(&buf); err != nil {
+		t.Fatal(err)
+	}
+	got = CiSubscribePipelineLogs_Event{}
+	if err := got.Deserialize(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if got.Data == nil || got.Data.Content != "legacy" || got.Data.Stream != "stderr" {
+		t.Fatalf("legacy-compatible data = %+v", got.Data)
+	}
+}
diff --git a/flake.nix b/flake.nix
index 2af10e343..dc37dd472 100644
--- a/flake.nix
+++ b/flake.nix
@@ -661,8 +661,8 @@
               rootDir=$(jj --ignore-working-copy root || git rev-parse --show-toplevel) || (echo "error: can't find repo root?"; exit 1)
               cd "$rootDir"
 
-              # *_ext.go are hand-written extensions; never remove or mutate them
-              find api/tangled api/org_tangled -maxdepth 1 -type f -not -name '*_ext.go' -delete
+              # preserve hand-written extensions and tests
+              find api/tangled api/org_tangled -maxdepth 1 -type f -not -name '*_ext.go' -not -name '*_test.go' -delete
               lexgen --build-file lexicon-build-config.json lexicons
 
               # disable type registration temporarily while running cborgen