diff --git a/flake.nix b/flake.nix
--- a/flake.nix
+++ b/flake.nix
@@ -374,6 +374,7 @@
             pkgs.iproute2
           ];
         shellHook = ''
+          export CC=${pkgs.stdenv.cc}/bin/cc
           mkdir -p appview/pages/static
           # temporary self-heal for workspaces that copied static assets as read-only
           [ -d appview/pages/static/icons ] && [ ! -w appview/pages/static/icons ] && chmod -R u+rwX appview/pages/static
diff --git a/lexutil/client.go b/lexutil/client.go
--- a/lexutil/client.go
+++ b/lexutil/client.go
@@ -44,7 +44,9 @@
 func (c *Client) LexDo(ctx context.Context, method string, inputEncoding string, endpoint string, params map[string]any, bodyData any, out any) error {
 	switch method {
 	case Subscription:
-		if process, ok := out.(processFn); ok {
+		if process, ok := out.(func(context.Context, *cbg.CborReader) error); ok {
+			return c.LexSubscribe(ctx, endpoint, params, process)
+		} else if process, ok := out.(processFn); ok {
 			return c.LexSubscribe(ctx, endpoint, params, process)
 		} else if redialer, ok := out.(Redialer); ok {
 			return c.LexSubscribeWithRedialer(ctx, endpoint, params, redialer)
diff --git a/spindle/stream.go b/spindle/stream.go
--- a/spindle/stream.go
+++ b/spindle/stream.go
@@ -2,12 +2,10 @@
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
-	"os"
 	"time"
 
 	"tangled.org/core/eventstream"
@@ -93,35 +91,7 @@
 
 	filePath := models.LogFilePath(s.cfg.Server.LogDir, wid)
 
-	if status.Status == models.StatusKindFailed.String() && status.Error != nil {
-		if _, err := os.Stat(filePath); os.IsNotExist(err) {
-			msgs := []models.LogLine{
-				{
-					Kind:     models.LogKindControl,
-					Content:  "",
-					StepId:   0,
-					StepKind: models.StepKindUser,
-				},
-				{
-					Kind:    models.LogKindData,
-					Content: *status.Error,
-				},
-			}
 
-			for _, msg := range msgs {
-				b, err := json.Marshal(msg)
-				if err != nil {
-					return err
-				}
-
-				if err := conn.WriteMessage(websocket.TextMessage, b); err != nil {
-					return fmt.Errorf("failed to write to websocket: %w", err)
-				}
-			}
-
-			return nil
-		}
-	}
 
 	config := tail.Config{
 		Follow:    !isFinished,
diff --git a/spindle/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go
new file mode 100644
--- /dev/null
+++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go
@@ -0,0 +1,322 @@
+package xrpc
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/atclient"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/gorilla/websocket"
+	"github.com/hpcloud/tail"
+	"tangled.org/core/api/tangled"
+	"tangled.org/core/spindle/models"
+)
+
+func (x *Xrpc) HandleCiPipelineSubscribeLogs(w http.ResponseWriter, r *http.Request) {
+	var (
+		pipelineQuery = r.URL.Query().Get("pipeline")
+		workflows     = r.URL.Query()["workflows"]
+	)
+
+	pipeline, err := syntax.ParseTID(pipelineQuery)
+	if err != nil {
+		writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("pipeline parameter invalid: %s", pipelineQuery)})
+		return
+	}
+
+	x.handleSubscribeLogs(w, r, pipeline, workflows)
+}
+
+var wsUpgrader = websocket.Upgrader{
+	ReadBufferSize:  10_000,
+	WriteBufferSize: 10_000,
+}
+
+func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) {
+	l := x.Logger.With("pipeline", pipeline, "workflows", workflows)
+
+	// 1. query the event from database to get the knot
+	var eventJson string
+	err := x.Db.QueryRow(
+		`select event from events where nsid = ? and rkey = ?`,
+		tangled.PipelineNSID,
+		pipeline.String(),
+	).Scan(&eventJson)
+	if err != nil {
+		l.Error("failed to find pipeline event", "err", err)
+		writeJson(w, http.StatusNotFound, atclient.ErrorBody{Name: "NotFound", Message: fmt.Sprintf("pipeline not found: %s", pipeline.String())})
+		return
+	}
+
+	var tpl tangled.Pipeline
+	if err := json.Unmarshal([]byte(eventJson), &tpl); err != nil {
+		l.Error("failed to unmarshal pipeline event", "err", err)
+		writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "failed to parse pipeline event"})
+		return
+	}
+
+	if tpl.TriggerMetadata == nil || tpl.TriggerMetadata.Repo == nil {
+		l.Error("pipeline event trigger metadata is incomplete")
+		writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "pipeline event trigger metadata is incomplete"})
+		return
+	}
+	knot := tpl.TriggerMetadata.Repo.Knot
+
+	// 2. if workflows is empty, default to all workflows defined in the pipeline
+	if len(workflows) == 0 {
+		for _, wf := range tpl.Workflows {
+			if wf != nil && wf.Name != "" {
+				workflows = append(workflows, wf.Name)
+			}
+		}
+	}
+
+	if len(workflows) == 0 {
+		writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "no workflows specified or found"})
+		return
+	}
+
+	// 3. upgrade to websocket
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+
+	conn, err := wsUpgrader.Upgrade(w, r, w.Header())
+	if err != nil {
+		l.Error("websocket upgrade failed", "err", err)
+		return
+	}
+	defer conn.Close()
+
+	lastWriteLk := sync.Mutex{}
+	lastWrite := time.Now()
+
+	// Ping loop
+	go func() {
+		ticker := time.NewTicker(30 * time.Second)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				lastWriteLk.Lock()
+				lw := lastWrite
+				lastWriteLk.Unlock()
+
+				if time.Since(lw) < 30*time.Second {
+					continue
+				}
+
+				if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
+					l.Warn("failed to ping client", "err", err)
+					cancel()
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	conn.SetPingHandler(func(message string) error {
+		err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60))
+		if err == websocket.ErrCloseSent {
+			return nil
+		}
+		return err
+	})
+
+	// Read discard loop
+	go func() {
+		for {
+			_, _, err := conn.ReadMessage()
+			if err != nil {
+				l.Warn("failed to read message from client", "err", err)
+				cancel()
+				return
+			}
+		}
+	}()
+
+	eventsChan := make(chan tangled.CiPipelineSubscribeLogs_Event, 128)
+	wg := sync.WaitGroup{}
+
+	// 4. start a tail reader goroutine for each workflow
+	for _, wf := range workflows {
+		wg.Add(1)
+		go func(wfName string) {
+			defer wg.Done()
+
+			wid := models.WorkflowId{
+				PipelineId: models.PipelineId{
+					Knot: knot,
+					Rkey: pipeline.String(),
+				},
+				Name: wfName,
+			}
+
+			// check if finished, but poll database to know when it finishes
+			var isFinished bool
+			status, err := x.Db.GetStatus(wid)
+			if err == nil {
+				isFinished = models.StatusKind(status.Status).IsFinish()
+			}
+
+			filePath := models.LogFilePath(x.Config.Server.LogDir, wid)
+
+
+
+			tailConfig := tail.Config{
+				Follow:    !isFinished,
+				ReOpen:    !isFinished,
+				MustExist: false,
+				Location: &tail.SeekInfo{
+					Offset: 0,
+					Whence: io.SeekStart,
+				},
+			}
+
+			t, err := tail.TailFile(filePath, tailConfig)
+			if err != nil {
+				l.Error("failed to tail log file", "workflow", wfName, "err", err)
+				return
+			}
+			defer t.Stop()
+
+			// if we are following, poll status in database to stop tailing when finished
+			if !isFinished {
+				go func() {
+					ticker := time.NewTicker(2 * time.Second)
+					defer ticker.Stop()
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						case <-ticker.C:
+							status, err := x.Db.GetStatus(wid)
+							if err == nil && models.StatusKind(status.Status).IsFinish() {
+								t.Stop()
+								return
+							}
+						}
+					}
+				}()
+			}
+
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case line, ok := <-t.Lines:
+					if !ok || line == nil {
+						return
+					}
+
+					if line.Err != nil {
+						l.Warn("error tailing log file", "workflow", wfName, "err", line.Err)
+						return
+					}
+
+					var logLine models.LogLine
+					if err := json.Unmarshal([]byte(line.Text), &logLine); err != nil {
+						// if it's not JSON, treat it as a raw data line
+						logLine = models.NewDataLogLine(0, line.Text, "stdout")
+					}
+
+					var ev tangled.CiPipelineSubscribeLogs_Event
+					timeStr := logLine.Time.Format(time.RFC3339)
+					if logLine.Time.IsZero() {
+						timeStr = time.Now().Format(time.RFC3339)
+					}
+
+					if logLine.Kind == models.LogKindControl {
+						stepKindStr := "user"
+						if logLine.StepKind == models.StepKindSystem {
+							stepKindStr = "system"
+						}
+						ev = tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{
+							Time:     timeStr,
+							Workflow: wfName,
+							Step:     int64(logLine.StepId),
+							Content:  logLine.Content,
+							Command:  strptrOrNil(logLine.StepCommand),
+							Status:   strptrOrNil(string(logLine.StepStatus)),
+							Kind:     strptrOrNil(stepKindStr),
+						}}
+					} else {
+						streamType := logLine.Stream
+						if streamType != "stdout" && streamType != "stderr" {
+							streamType = "stdout"
+						}
+						ev = tangled.CiPipelineSubscribeLogs_Event{Data: &tangled.CiPipelineSubscribeLogs_Data{
+							Time:     timeStr,
+							Workflow: wfName,
+							Step:     int64(logLine.StepId),
+							Content:  logLine.Content + "\n", // Append newline back since logger trims it
+							Stream:   streamType,
+						}}
+					}
+
+					select {
+					case eventsChan <- ev:
+					case <-ctx.Done():
+						return
+					}
+				}
+			}
+		}(wf)
+	}
+
+	// Closer goroutine for eventsChan
+	go func() {
+		wg.Wait()
+		close(eventsChan)
+	}()
+
+	// Main writer loop
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case evt, ok := <-eventsChan:
+			if !ok {
+				return
+			}
+
+			wc, err := conn.NextWriter(websocket.BinaryMessage)
+			if err != nil {
+				l.Error("failed to get next writer", "err", err)
+				return
+			}
+
+			err = evt.Serialize(wc)
+			if err != nil {
+				l.Error("failed to serialize event", "err", err)
+				wc.Close()
+				return
+			}
+
+			if err := wc.Close(); err != nil {
+				l.Warn("failed to flush-close event write", "err", err)
+				return
+			}
+
+			lastWriteLk.Lock()
+			lastWrite = time.Now()
+			lastWriteLk.Unlock()
+		}
+	}
+}
+
+func strptr(s string) *string { return &s }
+
+func strptrOrNil(s string) *string {
+	if s == "" {
+		return nil
+	}
+	return &s
+}
diff --git a/spindle/xrpc/xrpc.go b/spindle/xrpc/xrpc.go
--- a/spindle/xrpc/xrpc.go
+++ b/spindle/xrpc/xrpc.go
@@ -48,6 +48,7 @@
 
 	// service query endpoints (no auth required)
 	r.Get("/"+tangled.OwnerNSID, x.Owner)
+	r.Get("/"+tangled.CiPipelineSubscribeLogsNSID, x.HandleCiPipelineSubscribeLogs)
 
 	return r
 }
@@ -59,4 +60,13 @@
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)
 	json.NewEncoder(w).Encode(e)
+}
+
+func writeJson(w http.ResponseWriter, status int, response any) error {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	if err := json.NewEncoder(w).Encode(response); err != nil {
+		return err
+	}
+	return nil
 }