feat(officers): forward an officer instance's live session transcript to HQ

cmd/ccd/server.go

@@ -209,6 +209,8 @@ func newServer(cfg serverConfig, term terminal, auth *authTracker) *server {
 func (s *server) gatewayHandler() http.Handler {
 	mux := http.NewServeMux()
 	mux.HandleFunc("POST /v1/responses", s.handleResponses)
+	// The primary's transcript, incrementally (transcript_route.go).
+	mux.HandleFunc("GET /v1/transcript", s.handleTranscript)
 	// `/` is the same check: muster's default agent health path is "/"
 	// (agentspec.DefaultGatewayHealthPath), which its renderer uses for the
 	// startup and liveness probes. A per-kind health path for this image is muster

cmd/ccd/transcript_route.go

@@ -0,0 +1,134 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"os"
+	"strconv"
+
+	"github.com/ZacxDev/muster/internal/cctranscript"
+)
+
+// GET /v1/transcript: THE PRIMARY CLI'S OWN TRANSCRIPT, INCREMENTALLY.
+//
+// muster-server's officer transcript forwarder (internal/officers/transcript)
+// polls it to forward what an officer instance is doing to HQ while it holds
+// an item. It is the same structured record every other reader of a session
+// uses — the CLI's transcript JSONL, decoded by internal/cctranscript
+// (ReadLines / Backfill: complete lines only, a partial last line left for the
+// next poll) — never the screen.
+//
+// Query:
+//
+//   - after: the offset to read from (a Page.Next this route returned). -1 is
+//     the TAIL: the last `backfill` bytes (0: none — start live at the end).
+//   - backfill: the tail's window in bytes, at most transcriptBackfillMax.
+//   - session: the session the caller's offset belongs to. When the primary is
+//     now in another session (a new conversation), or the offset is past the
+//     file's end (the file was replaced), the read starts at 0 of the current
+//     one: every line of a conversation that began after the caller's last
+//     read is new to it.
+//
+// The answer is a cctranscript.Page. Bearer-authenticated like /v1/responses:
+// it serves tool output, so it is exactly as private as a turn's reply.
+//
+// The PRIMARY's CLI, found by its own pid and ITS sessions file
+// (cliSessionOf), never by the last hook's session: an officer's turns (its
+// first turn and every re-prompt) carry no lane header, so they run in the
+// primary. A lane's transcript is not served.
+
+const (
+	// transcriptReadMax bounds one forward read.
+	transcriptReadMax = 1 << 20
+	// transcriptBackfillMax bounds a tail's window.
+	transcriptBackfillMax = 1 << 20
+	// transcriptBackfillEvents bounds a tail's events.
+	transcriptBackfillEvents = 2000
+)
+
+func (s *server) handleTranscript(w http.ResponseWriter, r *http.Request) {
+	if !bearerOK(r, s.cfg.Bearer) {
+		writeFailure(w, &failure{Status: http.StatusUnauthorized, Type: failUnauthorized, Message: "missing or wrong bearer"})
+		return
+	}
+	q := r.URL.Query()
+	after := int64(0)
+	if v := q.Get("after"); v != "" {
+		n, err := strconv.ParseInt(v, 10, 64)
+		if err != nil || n < -1 {
+			writeFailure(w, &failure{Status: http.StatusBadRequest, Type: failBadRequest, Message: "after must be an offset, or -1 for the tail"})
+			return
+		}
+		after = n
+	}
+	backfill := 0
+	if v := q.Get("backfill"); v != "" {
+		n, err := strconv.Atoi(v)
+		if err != nil || n < 0 {
+			writeFailure(w, &failure{Status: http.StatusBadRequest, Type: failBadRequest, Message: "backfill must be a byte count"})
+			return
+		}
+		backfill = min(n, transcriptBackfillMax)
+	}
+	cs, path, ok := cliSessionOf(s.cfg.ConfigDir, procRoot, s.cliPIDOf(r.Context()))
+	if !ok || path == "" {
+		writeFailure(w, &failure{Status: http.StatusServiceUnavailable, Type: failNotReady, Message: "no running CLI session with a transcript"})
+		return
+	}
+	f, err := os.Open(path)
+	if err != nil {
+		writeFailure(w, &failure{Status: http.StatusServiceUnavailable, Type: failNotReady, Message: "the session's transcript cannot be opened"})
+		return
+	}
+	defer f.Close()
+	st, err := f.Stat()
+	if err != nil {
+		writeFailure(w, &failure{Status: http.StatusServiceUnavailable, Type: failNotReady, Message: "the session's transcript cannot be read"})
+		return
+	}
+	size := st.Size()
+	if sess := q.Get("session"); (sess != "" && sess != cs.SessionID) || after > size {
+		after = 0
+	}
+	page := cctranscript.Page{Session: cs.SessionID, Lines: []cctranscript.Line{}}
+	switch {
+	case after < 0 && backfill == 0:
+		page.Next, err = lastLineEnd(f, size)
+	case after < 0:
+		var lines []cctranscript.Line
+		lines, page.Next, err = cctranscript.Backfill(f, size, backfill, transcriptBackfillEvents)
+		page.Lines = append(page.Lines, lines...)
+	default:
+		var lines []cctranscript.Line
+		lines, page.Next, err = cctranscript.ReadLines(f, after, size, transcriptReadMax)
+		page.Lines = append(page.Lines, lines...)
+	}
+	if err != nil {
+		writeFailure(w, &failure{Status: http.StatusServiceUnavailable, Type: failNotReady, Message: "the session's transcript cannot be read"})
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(page)
+}
+
+// lastLineEnd is the offset just past the last complete line of r (size
+// bytes): where a live tail starts, so a record still being written is read
+// whole on the next poll rather than skipped from its middle.
+func lastLineEnd(r io.ReaderAt, size int64) (int64, error) {
+	buf := make([]byte, 64<<10)
+	for end := size; end > 0; {
+		start := max(end-int64(len(buf)), 0)
+		got, err := r.ReadAt(buf[:end-start], start)
+		if err != nil && !errors.Is(err, io.EOF) {
+			return 0, err
+		}
+		if i := bytes.LastIndexByte(buf[:got], '\n'); i >= 0 {
+			return start + int64(i) + 1, nil
+		}
+		end = start
+	}
+	return 0, nil
+}

cmd/ccd/transcript_route_test.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ZacxDev/muster/internal/cctranscript"
+)
+
+// GET /v1/transcript: the primary CLI's own transcript, read incrementally from
+// a line-boundary offset (cctranscript.ReadLines), for muster-server's officer
+// transcript forwarder.
+
+const transcriptBearer = "bearer-for-the-test"
+
+func transcriptRig(t *testing.T) (*server, string) {
+	t.Helper()
+	config := t.TempDir()
+	const pid = 4242
+	root := fakeProcFS(t, map[int]fakeProcess{pid: {ppid: 1, start: "777", argv: []string{"claude"}}})
+	old := procRoot
+	procRoot = root
+	t.Cleanup(func() { procRoot = old })
+	sessionFile(t, config, pid, fixtureSession, "busy", "777")
+	path := projectTranscript(t, config, fixtureSession)
+	s := newServer(serverConfig{ConfigDir: config, Bearer: transcriptBearer}, nil, newAuthTracker())
+	s.cliPID = func(context.Context) (int, error) { return pid, nil }
+	return s, path
+}
+
+func getTranscript(t *testing.T, s *server, query string, auth bool) (int, cctranscript.Page) {
+	t.Helper()
+	r := httptest.NewRequest(http.MethodGet, "/v1/transcript"+query, nil)
+	if auth {
+		r.Header.Set("Authorization", "Bearer "+transcriptBearer)
+	}
+	w := httptest.NewRecorder()
+	s.gatewayHandler().ServeHTTP(w, r)
+	var p cctranscript.Page
+	if w.Code == http.StatusOK {
+		if err := json.Unmarshal(w.Body.Bytes(), &p); err != nil {
+			t.Fatalf("body %q: %v", w.Body, err)
+		}
+	}
+	return w.Code, p
+}
+
+func TestTheTranscriptRouteServesThePrimarysTranscriptIncrementally(t *testing.T) {
+	s, path := transcriptRig(t)
+	if code, _ := getTranscript(t, s, "?after=0", false); code != http.StatusUnauthorized {
+		t.Fatalf("no bearer: %d, want 401", code)
+	}
+	code, p := getTranscript(t, s, "?after=0", true)
+	if code != http.StatusOK {
+		t.Fatalf("status %d", code)
+	}
+	st, _ := os.Stat(path)
+	if p.Session != fixtureSession || p.Next != st.Size() || len(p.Lines) == 0 {
+		t.Fatalf("page session %q next %d (size %d) lines %d", p.Session, p.Next, st.Size(), len(p.Lines))
+	}
+	if p.Lines[0].At != time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC).UnixMilli() || p.Lines[0].Events[0].Text != "fan the work out" {
+		t.Fatalf("first line %+v", p.Lines[0])
+	}
+	for i := 1; i < len(p.Lines); i++ {
+		if p.Lines[i].End <= p.Lines[i-1].End {
+			t.Fatalf("line ends not increasing")
+		}
+	}
+
+	// Nothing new: nothing served, the same offset back.
+	if _, again := getTranscript(t, s, "?session="+fixtureSession+"&after="+itoa64(p.Next), true); len(again.Lines) != 0 || again.Next != p.Next {
+		t.Fatalf("re-read at the end: %+v", again)
+	}
+	// A record appended is served from the offset, alone.
+	f, _ := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
+	_, _ = f.WriteString(`{"type":"assistant","timestamp":"2026-10-01T12:09:00Z","message":{"content":[{"type":"text","text":"appended"}]}}` + "\n" +
+		`{"type":"assistant","timestamp":"2026-10-01T12:09:01Z","message":{"content":[{"type":"text","text":"half wri`)
+	f.Close()
+	_, more := getTranscript(t, s, "?session="+fixtureSession+"&after="+itoa64(p.Next), true)
+	if len(more.Lines) != 1 || more.Lines[0].Events[0].Text != "appended" {
+		t.Fatalf("appended read: %+v", more)
+	}
+	// The partial last line is never consumed.
+	st, _ = os.Stat(path)
+	if more.Next >= st.Size() {
+		t.Fatalf("next %d consumed the partial line (size %d)", more.Next, st.Size())
+	}
+
+	// The tail with no backfill: nothing, positioned at the last complete line.
+	if _, tail := getTranscript(t, s, "?after=-1&backfill=0", true); len(tail.Lines) != 0 || tail.Next != more.Next {
+		t.Fatalf("live tail: %+v (want next %d)", tail, more.Next)
+	}
+	// The tail with a backfill window: only the lines inside it.
+	if _, tail := getTranscript(t, s, "?after=-1&backfill=1000", true); len(tail.Lines) == 0 || len(tail.Lines) >= len(p.Lines) {
+		t.Fatalf("backfill window served %d lines", len(tail.Lines))
+	}
+	// A caller holding another session's offset reads THIS session from 0.
+	if _, other := getTranscript(t, s, "?session=99999999-2222-4333-8444-555555555555&after="+itoa64(more.Next), true); len(other.Lines) != len(p.Lines)+1 {
+		t.Fatalf("another session's offset: %d lines, want the whole file (%d)", len(other.Lines), len(p.Lines)+1)
+	}
+	// An offset past the end (the file was replaced) reads from 0 too.
+	if _, past := getTranscript(t, s, "?session="+fixtureSession+"&after=999999", true); len(past.Lines) != len(p.Lines)+1 {
+		t.Fatalf("offset past the end: %d lines", len(past.Lines))
+	}
+}
+
+func TestTheTranscriptRouteWithNoCLIIsUnavailable(t *testing.T) {
+	s, _ := transcriptRig(t)
+	s.cliPID = func(context.Context) (int, error) { return 0, nil }
+	if code, _ := getTranscript(t, s, "?after=0", true); code != http.StatusServiceUnavailable {
+		t.Fatalf("no CLI: %d, want 503", code)
+	}
+}
+
+func itoa64(n int64) string { b, _ := json.Marshal(n); return string(b) }