diff --git a/cmd/ccd/server.go b/cmd/ccd/server.go index 0a672b6c..9ff0bef5 100644 --- a/cmd/ccd/server.go +++ b/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 diff --git a/cmd/ccd/transcript_route.go b/cmd/ccd/transcript_route.go new file mode 100644 index 00000000..9545e754 --- /dev/null +++ b/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 +} diff --git a/cmd/ccd/transcript_route_test.go b/cmd/ccd/transcript_route_test.go new file mode 100644 index 00000000..cf808e65 --- /dev/null +++ b/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) } diff --git a/docs/officers.md b/docs/officers.md index 0bd1c1d7..3c358cd2 100644 --- a/docs/officers.md +++ b/docs/officers.md @@ -60,6 +60,7 @@ MUSTER_GITHUB_APP_ACCESS=read # with the GitHub App: officers o | `repo` | none | The GitHub repository (`owner/name`, the same shape check as a task's repository) the instances may **read** through the `github` connector — see [Reading a repository](#reading-a-repository-repo). Not checked out. A value that is not `owner/name` is a boot refusal naming the officer. | | `connectors` | none | Connector names, besides `github`, the instances may request — e.g. a [postgres connector](connectors.md#postgres-connectors) for read-only SQL through `postgres_query`. Declaring one grants nothing: each instance's request is approved by hand, never from a shared approval ([below](#reading-a-repository-repo)). `github` itself, and a name that is not a connector name, is a boot refusal. | | `ship` | none | Lets the officer push, merge its own pull requests and tag releases in `repo`, through muster's gates: [Shipping officers](#shipping-officers). | +| `transcript` | off | `{"enabled": true}` forwards each instance's live session transcript to HQ while it holds a claimed item: [Transcript to HQ](#transcript-to-hq). Strict: any other key is a boot refusal. | Each officer's token is read from `MUSTER_OFFICER_TOKEN_`: the id upper-cased with every other character written as `_`. A declared officer with @@ -935,6 +936,134 @@ In order of reach. The first two do not depend on HQ. None of these undoes a tag already pushed. If a bad release is live, a human releases the fix. +## Transcript to HQ + +An officer declared with `"transcript": {"enabled": true}` has its instances' +live session transcripts forwarded to HQ — the FULL transcript, for HQ's +officer pages — while an instance holds a claimed item. It is off unless an +officer enables it. + +```json +{"id": "release:example-web", "hq_url": "https://hq.example.com/api/v1", + "transcript": {"enabled": true}} +``` + +**The source** is the instance's own Claude Code transcript (the JSONL the CLI +writes), never the screen. The instance pod's `ccd` serves it incrementally on +`GET /v1/transcript?after=&session=` (`after=-1&backfill=` +for the tail), decoded by `internal/cctranscript` exactly as the rendered view +decodes a mirrored session: complete lines only, a line still being written is +left for the next read. It is the PRIMARY CLI's transcript — the one an +officer's turns run in. muster-server reads it through the agent gateway with +the instance's own turn bearer (`agentgateway.Gateway.Transcript`). + +**The sink** is `POST {hq_url}/officers/{id}/items/{item_id}/transcript` with +the officer's token: + +```json +{"instance": "officer-release-example-web-1", + "events": [{"seq": 2192, "at": "2026-11-01T12:01:00Z", "kind": "tool_use", "text": "Bash: go test ./..."}]} +``` + +HQ answers `{"stored": n, "last_seq": s}`; it skips a `seq` at or below its +highest stored (idempotent). At most 100 events a call and 8 KiB a text. +Before the first events for an item, muster sends one post with +`"events": []` to learn HQ's `last_seq` for it (the PROBE); an HQ that answers +that with 400 is handled (muster then sends without it), but loses the +resume and re-claim handling below. + +| transcript entry | `kind` | `text` | +|---|---|---| +| the prompt an instance is given | `user` | the prompt | +| assistant text | `assistant` | the text (thinking is never sent) | +| a tool call | `tool_use` | `ToolName: ` — the file, the command, the pattern; a compact argument summary when there is none | +| a tool result | `tool_result` | its output (as `ccd` caps it, 4 KiB), `[error] ` first when the tool failed | +| an API error, a harness notice | `system` | the message | + +**When.** `officer_claim` succeeding starts forwarding for that item, from the +claim on (the first read takes the tail and keeps the lines written since the +claim, less a 5 s clock slack; a line with no timestamp there is not sent). +`officer_complete` or `officer_release` succeeding stops it. Every wake-loop +pass also reconciles: an item one of the officer's instances holds under a +live lease that muster is not forwarding is RESUMED, and forwarding stops for +an item no instance holds any more. A resume after a muster restart re-reads a +256 KiB backfill window and sends only what is above HQ's `last_seq` (the +probe); when HQ holds NOTHING for the item, that window is the instance's +earlier work, not this item's, and the resume starts live instead. A resume +of an instance this muster has already forwarded for (it held another item, +completed one) always starts live: what its transcript holds since is another +item's. HQ answering a +decision — 409 `not_claimed` (the item is not this instance's), 409 +`officer_disabled`, 404, 401/403 — stops forwarding for that item; a pass does +not restart it until a new claim (or until the item leaves the held list). An +instance that claims another item while forwarding one switches to it. + +**Redaction.** Every text goes through, in order: the officer's own token +replaced (`officers.Client.Scrub` — a shape filter cannot know an opaque +token); the CI-log secret filter (`officership.RedactText`, the same +`RedactLines` `officer_ci_logs` uses: a line that looks like it carries a +credential is withheld whole, a PEM block is withheld from BEGIN to END, a URL's +userinfo is masked; it knows GitHub, GitLab `glpat-`, AWS `AKIA`/`ASIA` and +`aws_secret_access_key`, Slack `xox?-`, Stripe `sk_`/`rk_` live/test, Google +`AIza…`, Hugging Face `hf_`, npm `npm_`, PyPI `pypi-`, `sk-`/`sk-proj-`/`sk-ant-` +and JWT shapes, and a kubeconfig's `client-key-data`/`client-certificate-data`); +then a line in which a word cut by a cap (`…` right after it — `ccd` caps long +fields before they reach muster) starts like one of those shapes is withheld +(`officership.RedactCutTail`), and a cut word ending in the officer token's +first bytes has that part replaced (`Client.Scrub`, which runs first); only then the 8 KiB cut, marked `… [truncated N bytes]`. No +cut of muster's own happens before the filters. Transcript text is never +logged. + +**`seq`** is the event's position in the transcript FILE: the end offset of +its line × 32768 + its index within the line (15 bits: ccd never serves a +line over 1 MiB, so no line carries that many events, and none is dropped). It strictly increases along the +file, so a restarted muster derives the same `seq` for the same event, and its +resume — which re-reads the backfill window — is deduplicated by HQ's rule +rather than by memory. It stays below 2^53 (a JSON-safe integer) for any file +under 256 GiB (2^53 / 2^15 = 2^38 bytes). A new conversation mid-item (a new file, offsets from zero: an +operator's `/clear`; `ccd` resumes the primary's own conversation by id on a +restart) is shifted above the seqs before it; a muster that restarts INTO such +a conversation has lost that shift, sees HQ answer `stored: 0` with a +`last_seq` above a batch it read live, and re-sequences that batch and the ones +after it above `last_seq` — on a batch's FIRST send only: a re-send answered +`stored: 0` is a lost acknowledgement (HQ committed, then failed to answer) +and is NOT re-sequenced, so it is stored once. What that double edge can lose +is the part of the resume's window HQ skipped as old. A batch whose seqs are +not above HQ's `last_seq` from the probe — a RE-CLAIM of the item by another +instance, whose transcript is its own file with its own offsets — is shifted +above it rather than skipped (HQ's `last_seq` is per item). + +**Cadence and back-pressure.** Each forwarding instance is read and posted +every 2 s, in batches of at most 100. An HQ outage (5xx, 429, no answer) backs +off — 2 s doubling to 1 min — and a batch is dropped after 5 failed re-sends. +A 400 drops the batch. A 413 drops that batch and halves the batch size for +the rest of the item. The forwarder never blocks the officer: the officer +tools and the wake loop only record state; reads and posts run on the +forwarder's own loop. Each instance's buffer holds at most 1000 events +waiting for HQ; past that the OLDEST are dropped. Every drop is logged and +counted on `muster_officer_transcript_events_total{result}`: `posted` (what HQ +answered as STORED), `skipped_by_hq` (sent, HQ already held it — a re-send +after a lost answer), `skipped_resume` (a resume's re-read HQ already held, or +earlier work), `dropped_buffer`, `dropped_retries`, `dropped_refused` (400), +`dropped_too_large` (413), `dropped_superseded`, `dropped_unsequenced`. + +**Deploy order.** HQ's transcript route must be LIVE before `transcript` is +enabled: an HQ without it answers 404, which muster reads as a decision and +stops forwarding for that item (until its next claim) — every item, quietly. +`MUSTER_OFFICERS` refuses an unknown field, so the +`transcript` key must reach a deployment only with (or after) a muster image +that knows it — the other way round, muster-server refuses to boot. The +instance pods need a `ccd` that serves `GET /v1/transcript`: an older pod +answers 404, the forwarder logs that the transcript cannot be read and retries +every tick, and nothing is sent until the pod runs the new image. An officer +that enables `transcript` on a server whose agent gateway cannot read +transcripts is a boot refusal. + +**Rollback order** is the reverse: remove the `transcript` key from +`MUSTER_OFFICERS` (and let muster-server restart on that config) BEFORE rolling +muster-server back to an image that predates it — an older image refuses the +unknown field and does not boot. + ## Security model - **The token never leaves muster-server.** It is read from the server's @@ -988,7 +1117,9 @@ From the officers contract, v1 (base `/api/v1`, bearer token of scope shipping officer also uses `GET /officers/{id}` (its `enabled`), `POST /officers/{id}/notices` (or its configured `notice.path`) and `POST /officers/{id}/preflight` (`{"wal": {"last_archived_time", "last_failed_time", -"failed_count"}, "last_backup_time"}`). Where the contract leaves a shape open, muster accepts: +"failed_count"}, "last_backup_time"}`). An officer with `transcript` enabled +uses `POST /officers/{id}/items/{item}/transcript` +([Transcript to HQ](#transcript-to-hq)). Where the contract leaves a shape open, muster accepts: - the items list as HQ sends it, `{"items": […], "next": "…"|null}`; the wake loop follows `next` for up to 5 pages of 100; @@ -998,7 +1129,8 @@ shipping officer also uses `GET /officers/{id}` (its `enabled`), `POST implements the claim/lease rules as HQ does (409 to ANY claim while the lease is live — the holder's own included, which `officer_claim` turns into a heartbeat — an expired lease claimable by anyone, 410 on a finished item, 403 -across officers) and the captain route's body rules. +across officers), the captain route's body rules, and the transcript route's +(limits, 409 `not_claimed` / `officer_disabled`, idempotency on `seq`). ## Deploying diff --git a/internal/agentgateway/transcript.go b/internal/agentgateway/transcript.go new file mode 100644 index 00000000..a2781396 --- /dev/null +++ b/internal/agentgateway/transcript.go @@ -0,0 +1,72 @@ +package agentgateway + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/ZacxDev/muster/internal/agents" + "github.com/ZacxDev/muster/internal/cctranscript" +) + +// ErrTranscriptUnsupported is an agent runtime without GET /v1/transcript: a +// ccd image older than the route, or a runtime that is not ccd. +var ErrTranscriptUnsupported = errors.New("agentgateway: the agent's runtime does not serve its transcript (GET /v1/transcript); its image predates it") + +// transcriptTimeout bounds one transcript read. +const transcriptTimeout = 10 * time.Second + +// maxTranscriptBody bounds one transcript page's body: ccd reads at most +// 1 MiB of transcript per page, and an event's fields are capped +// (cctranscript), so this is a backstop. +const maxTranscriptBody = 16 << 20 + +// Transcript reads agent a's session transcript from its runtime (ccd's GET +// /v1/transcript): the complete lines after offset after in session (after -1: +// the tail, its last backfill bytes). It is the officer transcript +// forwarder's source (internal/officers/transcript). The bearer is the turn +// bearer, the same credential a turn carries; it never appears in an error. +func (g *Gateway) Transcript(ctx context.Context, a agents.Agent, session string, after int64, backfill int) (cctranscript.Page, error) { + t, err := g.Resolve(ctx, a) + if err != nil { + return cctranscript.Page{}, err + } + q := url.Values{} + q.Set("after", strconv.FormatInt(after, 10)) + if backfill > 0 { + q.Set("backfill", strconv.Itoa(backfill)) + } + if session != "" { + q.Set("session", session) + } + ctx, cancel := context.WithTimeout(ctx, transcriptTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSuffix(t.ep.URL(), "/")+"/v1/transcript?"+q.Encode(), nil) + if err != nil { + return cctranscript.Page{}, fmt.Errorf("agentgateway: transcript request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+t.bearer) + resp, err := g.client.Do(req) + if err != nil { + return cctranscript.Page{}, fmt.Errorf("agentgateway: read agent %q's transcript: %w", a.Name, err) + } + defer resp.Body.Close() + switch { + case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusMethodNotAllowed: + return cctranscript.Page{}, ErrTranscriptUnsupported + case resp.StatusCode != http.StatusOK: + return cctranscript.Page{}, fmt.Errorf("agentgateway: agent %q's transcript: its runtime answered %d", a.Name, resp.StatusCode) + } + var p cctranscript.Page + if err := json.NewDecoder(io.LimitReader(resp.Body, maxTranscriptBody)).Decode(&p); err != nil { + return cctranscript.Page{}, fmt.Errorf("agentgateway: agent %q's transcript is not a page: %v", a.Name, err) + } + return p, nil +} diff --git a/internal/agentgateway/transcript_test.go b/internal/agentgateway/transcript_test.go new file mode 100644 index 00000000..a0ab3fcf --- /dev/null +++ b/internal/agentgateway/transcript_test.go @@ -0,0 +1,50 @@ +package agentgateway + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/ZacxDev/muster/internal/cctranscript" +) + +// TestTranscriptReadsTheAgentsTranscriptWithItsTurnBearer: the GET carries +// the agent's own turn bearer and the query ccd's route takes, and the page +// decodes; a runtime without the route is ErrTranscriptUnsupported. +func TestTranscriptReadsTheAgentsTranscriptWithItsTurnBearer(t *testing.T) { + var gotAuth, gotQuery, gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth, gotQuery, gotPath = r.Header.Get("Authorization"), r.URL.RawQuery, r.URL.Path + _ = json.NewEncoder(w).Encode(cctranscript.Page{Session: "s-9", Next: 512, + Lines: []cctranscript.Line{{End: 512, At: 1700000000000, Events: []cctranscript.Event{{Kind: cctranscript.KindText, Text: "hi"}}}}}) + })) + defer srv.Close() + gw, err := New(Config{Driver: &fixedResolver{ep: endpointOf(t, srv.URL)}, Runtime: HooksSHA256(), Model: testSentinel, Client: srv.Client()}) + if err != nil { + t.Fatal(err) + } + a := fixtureAgent() + p, err := gw.Transcript(context.Background(), a, "s-8", 256, 4096) + if err != nil { + t.Fatal(err) + } + if gotPath != "/v1/transcript" || gotQuery != "after=256&backfill=4096&session=s-8" { + t.Fatalf("GET %s?%s", gotPath, gotQuery) + } + if gotAuth != "Bearer "+HooksSHA256().Bearer(a.HooksToken) { + t.Fatalf("authorization %q is not the agent's turn bearer", gotAuth) + } + if p.Session != "s-9" || p.Next != 512 || len(p.Lines) != 1 || p.Lines[0].Events[0].Text != "hi" || p.Lines[0].At != 1700000000000 { + t.Fatalf("page %+v", p) + } + + old := httptest.NewServer(http.NotFoundHandler()) + defer old.Close() + gw, _ = New(Config{Driver: &fixedResolver{ep: endpointOf(t, old.URL)}, Runtime: HooksSHA256(), Model: testSentinel, Client: old.Client()}) + if _, err := gw.Transcript(context.Background(), a, "", -1, 0); !errors.Is(err, ErrTranscriptUnsupported) { + t.Fatalf("an image without the route: %v, want ErrTranscriptUnsupported", err) + } +} diff --git a/internal/api/officer_transcript_test.go b/internal/api/officer_transcript_test.go new file mode 100644 index 00000000..64e3ccbd --- /dev/null +++ b/internal/api/officer_transcript_test.go @@ -0,0 +1,125 @@ +package api + +import ( + "bytes" + "context" + "log" + "strings" + "sync" + "testing" + "time" + + "github.com/ZacxDev/muster/internal/agents" + "github.com/ZacxDev/muster/internal/cctranscript" + "github.com/ZacxDev/muster/internal/officers" + "github.com/ZacxDev/muster/internal/officers/hqtest" + "github.com/ZacxDev/muster/internal/provision" +) + +// --------------------------------------------------------------------------- +// THE TRANSCRIPT FORWARDER, BOUND TO THE SERVER (officers.go): the seam +// between the officer tools / the wake loop and internal/officers/transcript. +// Each half is tested on its own there; this pins that officer_claim starts +// forwarding, officer_complete stops it, a wake pass resumes a held item, and +// the instance's transcript is read through the agent gateway BY ITS OWN +// AGENT ROW. +// --------------------------------------------------------------------------- + +// transcriptGateway is the reprompt chat fake plus the gateway's transcript +// read, serving one line per read for whichever agent it is asked about. +type transcriptGateway struct { + repromptChat + tmu sync.Mutex + asked []string + end int64 + clock func() time.Time +} + +func (g *transcriptGateway) Transcript(_ context.Context, a agents.Agent, _ string, after int64, _ int) (cctranscript.Page, error) { + g.tmu.Lock() + defer g.tmu.Unlock() + g.asked = append(g.asked, a.Name) + g.end += 100 + return cctranscript.Page{Session: "s-1", Next: g.end, Lines: []cctranscript.Line{{End: g.end, At: g.clock().UnixMilli(), + Events: []cctranscript.Event{{Kind: cctranscript.KindText, Text: "working " + a.Name}}}}}, nil +} + +func transcriptServer(t *testing.T, gw Gateway) (*Server, *officerAgents, *hqtest.Server, officers.Officer, error) { + t.Helper() + hq := hqtest.New(map[string]string{offATok: offA}) + t.Cleanup(hq.Close) + store := &officerAgents{} + prov := &officerProv{store: store, phase: map[string]provision.Phase{}} + s := New(nil, AuthConfig{UIPassword: testUIPassword, HookToken: testHookToken}, log.New(&bytes.Buffer{}, "", 0)) + s.UseExtensions(Extensions{Agents: store, Provisioner: prov, Gateway: gw, SessionLiveness: stubLiveness{}, Kinds: stubKinds{pick: "acct-one"}}) + o := officers.Officer{ID: offA, HQURL: hq.BaseURL(), MaxInstances: 1, IdleAfter: 30 * time.Minute, PollInterval: time.Minute, Owner: 1, + Transcript: &officers.Transcript{Enabled: true}}.WithToken(offATok) + _, err := s.EnableOfficers(OfficersConfig{Officers: []officers.Officer{o}, InstallSession: discardSession, NoTranscriptRun: true}) + if err == nil { + a := agents.Agent{ID: 901, Name: officers.InstanceName(o.ID, 1), Kind: agents.KindClaudeCode, Status: agents.StatusRunning, OwnerID: 1, Namespace: "ns"} + store.rows = append(store.rows, a) + prov.phase[a.Name] = provision.PhaseRunning + } + return s, store, hq, o, err +} + +func TestTheOfficerToolsAndTheWakeLoopDriveTheTranscriptForwarder(t *testing.T) { + gw := &transcriptGateway{clock: time.Now} + s, store, hq, o, err := transcriptServer(t, gw) + if err != nil { + t.Fatal(err) + } + inst, _ := store.GetByName(context.Background(), officers.InstanceName(o.ID, 1)) + fwd := s.officerRT.fwd + if fwd == nil { + t.Fatal("no forwarder for a transcript-enabled officer") + } + ctx := context.Background() + item := hq.AddItem(offA, "release.evaluate", "svc:v1") + + if out := s.dispatchAgentTool(ctx, inst, officers.ToolClaim, `{"item_id":"`+item+`"}`); isToolErr(out) { + t.Fatalf("claim: %s", out) + } + if got := fwd.Active(); got[inst.Name] != officers.ItemID(item) { + t.Fatalf("after officer_claim, forwarding %v", got) + } + fwd.Tick(ctx) + if got := hq.Transcript(item); len(got) != 1 || got[0].Text != "working "+inst.Name || got[0].Instance != inst.Name { + t.Fatalf("HQ stored %+v", got) + } + gw.tmu.Lock() + asked := strings.Join(gw.asked, ",") + gw.tmu.Unlock() + if asked != inst.Name { + t.Fatalf("the gateway was asked for %q, want the instance's own row %q", asked, inst.Name) + } + + if out := s.dispatchAgentTool(ctx, inst, officers.ToolComplete, `{"item_id":"`+item+`","outcome":"done","summary":"ok"}`); isToolErr(out) { + t.Fatalf("complete: %s", out) + } + if got := fwd.Active(); len(got) != 0 { + t.Fatalf("after officer_complete, still forwarding %v", got) + } + + // A claim muster did not see (a restart): the wake pass resumes it. + next := hq.AddItem(offA, "release.evaluate", "svc:v2") + hq.SetClaim(next, inst.Name, time.Now().Add(10*time.Minute)) + s.officerRT.rec.Check(ctx, offA, true) + if got := fwd.Active(); got[inst.Name] != officers.ItemID(next) { + t.Fatalf("after a wake pass seeing the held item, forwarding %v", got) + } + // And a pass that no longer sees it held stops it. + hq.Complete(next) + time.Sleep(2 * time.Millisecond) // the pass reads after the resume's start + s.officerRT.rec.Check(ctx, offA, true) + if got := fwd.Active(); len(got) != 0 { + t.Fatalf("after a pass with nothing held, forwarding %v", got) + } +} + +func TestATranscriptOfficerWithoutAGatewayTranscriptIsABootRefusal(t *testing.T) { + _, _, _, _, err := transcriptServer(t, &repromptChat{}) + if err == nil || !strings.Contains(err.Error(), "transcript") { + t.Fatalf("err = %v, want a refusal naming transcript forwarding", err) + } +} diff --git a/internal/api/officers.go b/internal/api/officers.go index f82086af..14c58080 100644 --- a/internal/api/officers.go +++ b/internal/api/officers.go @@ -16,8 +16,10 @@ import ( "github.com/ZacxDev/muster/internal/agents" "github.com/ZacxDev/muster/internal/agentspec" "github.com/ZacxDev/muster/internal/authz" + "github.com/ZacxDev/muster/internal/cctranscript" "github.com/ZacxDev/muster/internal/connectors" "github.com/ZacxDev/muster/internal/officers" + "github.com/ZacxDev/muster/internal/officers/transcript" "github.com/ZacxDev/muster/internal/officership" "github.com/ZacxDev/muster/internal/provision" ) @@ -60,6 +62,9 @@ type officerRuntime struct { // ships are the shipping officers' runtimes, by officer id // (officer_ship.go). ships map[string]*shipRuntime + // fwd forwards instances' transcripts to HQ (internal/officers/transcript); + // nil when no declared officer enables `transcript`. + fwd *transcript.Forwarder } // OfficersConfig is EnableOfficers' input. @@ -87,6 +92,70 @@ type OfficersConfig struct { ShipCI func(officers.Officer) officership.CIReader // NoShipResume skips resuming rollout watches at boot (tests). NoShipResume bool + // TranscriptSource replaces where a transcript-enabled officer's + // instances' transcripts are read (tests); nil is each instance's ccd, + // through the agent gateway (agentgateway.Gateway.Transcript). + TranscriptSource transcript.Source + // TranscriptInterval replaces the forwarder's cadence (tests); 0 is + // transcript.DefaultInterval. NoTranscriptRun skips running its loop + // (tests drive Tick). + TranscriptInterval time.Duration + NoTranscriptRun bool +} + +// transcriptReader is the agent gateway's transcript read +// (agentgateway.Gateway.Transcript), asserted rather than added to Gateway: +// it is the officer transcript forwarder's alone. +type transcriptReader interface { + Transcript(ctx context.Context, a agents.Agent, session string, after int64, backfill int) (cctranscript.Page, error) +} + +var _ transcriptReader = (*agentgateway.Gateway)(nil) + +// gatewayTranscripts is transcript.Source over the agent gateway: the +// instance agent by name (as its officer's owner), then its ccd's +// GET /v1/transcript. +type gatewayTranscripts struct { + s *Server + r transcriptReader +} + +func (g gatewayTranscripts) Read(ctx context.Context, o officers.Officer, instance string, q transcript.Query) (cctranscript.Page, error) { + octx := g.s.officerPrincipal(ctx, o.Owner) + a, err := g.s.ext.Agents.GetByName(octx, instance) + if err != nil { + return cctranscript.Page{}, fmt.Errorf("officers: instance %s: %w", instance, err) + } + return g.r.Transcript(octx, a, q.Session, q.After, q.Backfill) +} + +// newTranscriptForwarder builds the forwarder when a declared officer enables +// `transcript`, and nil otherwise. An enabled officer with no way to read a +// transcript is a boot refusal, not a silent no-op. +func (s *Server) newTranscriptForwarder(cfg OfficersConfig, hq *officers.ClientHQ) (*transcript.Forwarder, error) { + var on []string + for _, o := range cfg.Officers { + if o.TranscriptEnabled() { + on = append(on, o.ID) + } + } + if len(on) == 0 { + return nil, nil + } + src := cfg.TranscriptSource + if src == nil { + r, ok := s.ext.Gateway.(transcriptReader) + if !ok { + return nil, fmt.Errorf("officers: %s enable transcript forwarding, but the agent gateway cannot read an instance's transcript", strings.Join(on, ", ")) + } + src = gatewayTranscripts{s: s, r: r} + } + f, err := transcript.New(transcript.Config{Source: src, Poster: transcript.NewClientPoster(hq), Logger: s.logger, Now: cfg.Now, Interval: cfg.TranscriptInterval}) + if err != nil { + return nil, err + } + s.logger.Printf("officers: forwarding instance transcripts to HQ for %s while an instance holds a claimed item", strings.Join(on, ", ")) + return f, nil } // EnableOfficers turns the officer runtime on and returns its wake loop for the @@ -108,17 +177,30 @@ func (s *Server) EnableOfficers(cfg OfficersConfig) (*officers.Reconciler, error reg := officers.NewRegistry(cfg.Officers) hc := cfg.HTTPClient hq := officers.NewClientHQ(func(o officers.Officer) *officers.Client { return officers.NewClient(o, hc) }) - rec, err := officers.NewReconciler(officers.Config{Registry: reg, HQ: hq, Fleet: officerFleet{s: s}, Logger: s.logger, Now: cfg.Now}) + fwd, err := s.newTranscriptForwarder(cfg, hq) + if err != nil { + return nil, err + } + rcfg := officers.Config{Registry: reg, HQ: hq, Fleet: officerFleet{s: s}, Logger: s.logger, Now: cfg.Now} + if fwd != nil { + rcfg.Observed = fwd.Sync + } + rec, err := officers.NewReconciler(rcfg) if err != nil { return nil, err } host, _ := os.Hostname() - s.officerRT = &officerRuntime{reg: reg, hq: hq, rec: rec, claimOwner: fmt.Sprintf("officer-reprompt/%s/%d", host, os.Getpid())} + s.officerRT = &officerRuntime{reg: reg, hq: hq, rec: rec, fwd: fwd, claimOwner: fmt.Sprintf("officer-reprompt/%s/%d", host, os.Getpid())} if err := s.enableShips(cfg); err != nil { s.officerRT = nil return nil, err } cfg.InstallSession(s.OfficerClaudeCodeSession) + if fwd != nil && !cfg.NoTranscriptRun { + // Ends with the server's streams (EndStreams, at shutdown). + ctx := s.streams.context() + safeGo(s.logger, "officer transcript forwarder", func() { fwd.Run(ctx) }) + } for _, o := range reg.All() { repo := "none" if o.Repo != "" { @@ -709,6 +791,9 @@ func (s *Server) dispatchOfficerTool(ctx context.Context, a agents.Agent, name, return bad, true } body, err = s.officerClaim(ctx, c, in.ItemID, instance, in.LeaseSeconds) + if err == nil && s.officerRT.fwd != nil { + s.officerRT.fwd.Claimed(o, instance, in.ItemID) + } case officers.ToolHeartbeat: if bad := needItem(); bad != "" { return bad, true @@ -732,11 +817,17 @@ func (s *Server) dispatchOfficerTool(ctx context.Context, a agents.Agent, name, return toolErr("at most %d links", officers.MaxLinks), true } body, err = c.Complete(ctx, in.ItemID, instance, officers.Completion{Outcome: in.Outcome, Summary: in.Summary, Links: in.Links}) + if err == nil && s.officerRT.fwd != nil { + s.officerRT.fwd.Ended(o, instance, in.ItemID) + } case officers.ToolRelease: if bad := needItem(); bad != "" { return bad, true } body, err = c.Release(ctx, in.ItemID, instance) + if err == nil && s.officerRT.fwd != nil { + s.officerRT.fwd.Ended(o, instance, in.ItemID) + } case officers.ToolReleaseEvaluation: post, perr := officers.ParseCaptainPost([]byte(orEmptyJSON(arguments))) if perr != nil { diff --git a/internal/api/routes_golden_test.go b/internal/api/routes_golden_test.go index 47ccb0e7..090e3534 100644 --- a/internal/api/routes_golden_test.go +++ b/internal/api/routes_golden_test.go @@ -703,9 +703,11 @@ func moduleRoot(t *testing.T) string { // internal/agentgateway; /hook/{event} is served only on ccd's hook socket (a // unix socket, each hook authenticated with a credential only ccd's CLIs hold); // GET /{$} is the same health check as /healthz, at muster's default agent -// health path ("/"). +// health path ("/"); GET /v1/transcript is the primary CLI's transcript, +// read incrementally by muster-server's officer transcript forwarder +// (internal/officers/transcript), bearer-authenticated like /v1/responses. var otherServerRoutes = map[string][]string{ - "cmd/ccd": {"GET /healthz", "GET /{$}", "POST /hook/{event}", "POST /v1/responses"}, + "cmd/ccd": {"GET /healthz", "GET /{$}", "GET /v1/transcript", "POST /hook/{event}", "POST /v1/responses"}, } // TestOtherServersRouteLedger is the other half of the exemption above. diff --git a/internal/cctranscript/page.go b/internal/cctranscript/page.go new file mode 100644 index 00000000..ebe931fc --- /dev/null +++ b/internal/cctranscript/page.go @@ -0,0 +1,17 @@ +package cctranscript + +// Page is one incremental read of a session's transcript, as ccd serves it on +// GET /v1/transcript (cmd/ccd/transcript_route.go) and muster-server's officer +// transcript forwarder reads it (internal/officers/transcript). +// +// - Session is the CLI session id whose transcript this is. A caller that +// sent another session's offset was served this one from its start. +// - Next is the offset to read from next: always a line boundary +// (ReadLines), never past a partial last line. +// - Lines are the complete lines read, each with the events it produced, its +// end offset and its own time. +type Page struct { + Session string `json:"session"` + Next int64 `json:"next"` + Lines []Line `json:"lines"` +} diff --git a/internal/cctranscript/tail.go b/internal/cctranscript/tail.go index 3a492131..50c3a06a 100644 --- a/internal/cctranscript/tail.go +++ b/internal/cctranscript/tail.go @@ -15,10 +15,12 @@ import ( // point: every byte before it was a complete line, decoded exactly once. // Line is the events of one complete transcript line, and the offset just past -// its newline: a resume point that comes AFTER every one of its events. +// its newline: a resume point that comes AFTER every one of its events. At is +// the record's own timestamp (Unix ms; 0 when it has none). type Line struct { - Events []Event - End int64 + Events []Event `json:"events"` + End int64 `json:"end"` + At int64 `json:"at,omitempty"` } // ReadLines decodes the complete lines in r between off and size, reading at @@ -66,7 +68,7 @@ func ReadLines(r io.ReaderAt, off, size int64, maxBytes int) ([]Line, int64, err evs[i].Line = pos // what Prompts names it by } } - out = append(out, Line{Events: evs, End: pos}) + out = append(out, Line{Events: evs, End: pos, At: rec.Time()}) } } } @@ -191,12 +193,12 @@ func Backfill(r io.ReaderAt, size int64, window, maxEvents int) ([]Line, int64, } if n+len(evs) > maxEvents { if keep := maxEvents - n; keep > 0 { - out = append(out, Line{Events: evs[len(evs)-keep:], End: end}) + out = append(out, Line{Events: evs[len(evs)-keep:], End: end, At: rec.Time()}) } break } n += len(evs) - out = append(out, Line{Events: evs, End: end}) + out = append(out, Line{Events: evs, End: end, At: rec.Time()}) } } end = start + int64(j) + 1 diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index d8a69de2..04d6575f 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -314,6 +314,17 @@ var ( Name: "muster_officer_hq_faults", Help: "Officers whose HQ refuses them permanently (401/403/404); their instances are scaled to zero.", }) + // OfficerTranscriptEvents counts officer transcript events by what became + // of them (internal/officers/transcript): posted (HQ answered 2xx), + // dropped_buffer (an instance's buffer was full: the oldest went), + // dropped_retries (HQ kept failing past the retry cap), dropped_refused + // (HQ answered 400), dropped_superseded (the instance claimed another + // item before they were sent), dropped_unsequenced (no valid seq could be + // derived from the transcript position). + OfficerTranscriptEvents = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "muster_officer_transcript_events_total", + Help: "Officer transcript events forwarded to HQ, by result (posted or a drop reason).", + }, []string{"result"}) ) // ObserveHTTP records one HTTP request's matched route pattern, method, status, diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 85dbb0fe..7d59087f 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -43,6 +43,7 @@ import ( var wantMetrics = map[string][]string{ "muster_officer_error_slots": nil, "muster_officer_hq_faults": nil, + "muster_officer_transcript_events_total": {"result"}, "muster_build_info": {"version"}, "muster_http_requests_total": {"code", "method", "route"}, "muster_http_request_duration_seconds": {"method", "route"}, @@ -94,6 +95,7 @@ func gather(t *testing.T) map[string][]string { MemoRelayAgentChecks.WithLabelValues("eligible") MemoRelayUpstream.WithLabelValues("ok") MemoRelayDeliveries.WithLabelValues("delivered") + OfficerTranscriptEvents.WithLabelValues("posted") ObserveHTTP("GET /api/tasks/{id}", "GET", 200, 0) families, err := prometheus.DefaultGatherer.Gather() diff --git a/internal/officers/config.go b/internal/officers/config.go index 5a0e31b9..2ca0fc2d 100644 --- a/internal/officers/config.go +++ b/internal/officers/config.go @@ -93,6 +93,10 @@ type Officer struct { // failing tests and error lines of the officer's own pull request's CI, // read by muster-server from a Tekton namespace (cilogs.go). CILogs *CILogs `json:"-"` + // Transcript, when set and enabled, forwards each instance's live session + // transcript to HQ while it holds a claimed item + // (internal/officers/transcript). nil (the default) forwards nothing. + Transcript *Transcript `json:"-"` // Connectors are the connectors, besides `github`, the officer's instances // may REQUEST (internal/api officerConnectorRefusal), by name — e.g. a // postgres connector for read-only SQL. Declaring one grants nothing: each @@ -138,6 +142,31 @@ type declared struct { Ship json.RawMessage `json:"ship"` // CILogs is parsed by parseCILogs (cilogs.go); it needs Ship. CILogs json.RawMessage `json:"ci_logs"` + // Transcript is parsed by parseTranscript. + Transcript json.RawMessage `json:"transcript"` +} + +// Transcript is an officer's `transcript` block: whether its instances' live +// session transcripts are forwarded to HQ (POST …/items/{item}/transcript) +// while they hold a claimed item. +type Transcript struct { + Enabled bool `json:"enabled"` +} + +// TranscriptEnabled reports whether o's instances' transcripts are forwarded +// to HQ. THE one switch: the forwarder asks it before every read and post. +func (o Officer) TranscriptEnabled() bool { return o.Transcript != nil && o.Transcript.Enabled } + +// parseTranscript reads a `transcript` block strictly: an unknown key is a +// refusal, as everywhere in MUSTER_OFFICERS. +func parseTranscript(raw json.RawMessage) (*Transcript, error) { + dec := json.NewDecoder(strings.NewReader(string(raw))) + dec.DisallowUnknownFields() + var t Transcript + if err := dec.Decode(&t); err != nil { + return nil, fmt.Errorf("transcript must be {\"enabled\": true|false}: %v", err) + } + return &t, nil } // idPart is one half of an officer id: lowercase alphanumerics and single @@ -326,6 +355,13 @@ func (d declared) officer(getenv func(string) string, maxName int) (Officer, err } o.CILogs = ci } + if len(d.Transcript) > 0 && string(d.Transcript) != "null" { + tr, err := parseTranscript(d.Transcript) + if err != nil { + return Officer{}, fmt.Errorf("officer %s: %w", o.ID, err) + } + o.Transcript = tr + } tok := strings.TrimSpace(getenv(TokenEnvPrefix + TokenEnvSuffix(o.ID))) if tok == "" { return Officer{}, fmt.Errorf("officer %s: no token: set %s%s", o.ID, TokenEnvPrefix, TokenEnvSuffix(o.ID)) diff --git a/internal/officers/config_test.go b/internal/officers/config_test.go index acca44a7..f3cf36d7 100644 --- a/internal/officers/config_test.go +++ b/internal/officers/config_test.go @@ -167,3 +167,31 @@ func TestParseReadsTheRepo(t *testing.T) { } } } + +// TestParseReadsTheTranscriptSwitch: transcript forwarding is OFF unless an +// officer says `"transcript": {"enabled": true}`; the block is strict (an +// unknown key is a boot refusal, as everywhere in MUSTER_OFFICERS). +func TestParseReadsTheTranscriptSwitch(t *testing.T) { + tok := env(map[string]string{"MUSTER_OFFICER_TOKEN_A_B": "x"}) + for raw, want := range map[string]bool{ + `[{"id":"a:b","hq_url":"http://h/api/v1"}]`: false, + `[{"id":"a:b","hq_url":"http://h/api/v1","transcript":null}]`: false, + `[{"id":"a:b","hq_url":"http://h/api/v1","transcript":{}}]`: false, + `[{"id":"a:b","hq_url":"http://h/api/v1","transcript":{"enabled":false}}]`: false, + `[{"id":"a:b","hq_url":"http://h/api/v1","transcript":{"enabled":true}}]`: true, + } { + list, err := officers.Parse(raw, tok, "muster-agent-") + if err != nil { + t.Fatalf("%s: %v", raw, err) + } + if got := list[0].TranscriptEnabled(); got != want { + t.Fatalf("%s: TranscriptEnabled = %v, want %v", raw, got, want) + } + } + for _, bad := range []string{`{"enabled":true,"url":"x"}`, `{"enabled":"yes"}`, `true`} { + raw := `[{"id":"a:b","hq_url":"http://h/api/v1","transcript":` + bad + `}]` + if _, err := officers.Parse(raw, tok, "muster-agent-"); err == nil || !strings.Contains(err.Error(), "transcript") { + t.Errorf("transcript %s: err = %v, want a refusal naming transcript", bad, err) + } + } +} diff --git a/internal/officers/hq.go b/internal/officers/hq.go index 3a27c9c9..63503501 100644 --- a/internal/officers/hq.go +++ b/internal/officers/hq.go @@ -88,7 +88,7 @@ const ( // 🔴 IT IS THE ONLY CODE THAT READS Officer.Token(). The token goes in the // Authorization header of requests to the officer's own HQ base URL, and // nowhere else: not in an error, not in a log line (HQError carries HQ's body, -// never the request). +// never the request). (Scrub reads it too, only to REMOVE it from text.) type Client struct { officer Officer http *http.Client @@ -314,6 +314,103 @@ func (c *Client) Release(ctx context.Context, item ItemID, instance string) ([]b return c.Do(ctx, http.MethodPost, c.itemPath(item, "release"), map[string]any{"instance": instance}) } +// --- the transcript route (POST …/items/{item}/transcript) ------------------- +// +// HQ's contract: {"instance", "events": [{"seq", "at", "kind", "text"}]}, at +// most MaxTranscriptEvents events, each text at most MaxTranscriptText bytes, +// seq strictly increasing per item and idempotent (HQ skips a seq at or below +// its highest stored). It answers {"stored", "last_seq"}; 409 officer_disabled, +// 404 (not this officer's item), 409 not_claimed (not claimed by this +// instance), 400 on a limit. + +// Transcript route bounds, as HQ enforces them. +const ( + MaxTranscriptEvents = 100 + MaxTranscriptText = 8 << 10 // bytes +) + +// Transcript event kinds, as the contract names them. +const ( + TranscriptUser = "user" + TranscriptAssistant = "assistant" + TranscriptToolUse = "tool_use" + TranscriptToolResult = "tool_result" + TranscriptSystem = "system" +) + +// TranscriptEvent is one transcript event posted to HQ. +type TranscriptEvent struct { + Seq int64 `json:"seq"` + At string `json:"at"` + Kind string `json:"kind"` + Text string `json:"text"` +} + +// TranscriptAck is HQ's answer to a transcript post. +type TranscriptAck struct { + Stored int `json:"stored"` + LastSeq int64 `json:"last_seq"` +} + +// PostTranscript is POST …/items/{item}/transcript. +func (c *Client) PostTranscript(ctx context.Context, item ItemID, instance string, events []TranscriptEvent) (TranscriptAck, error) { + raw, err := c.Do(ctx, http.MethodPost, c.itemPath(item, "transcript"), struct { + Instance string `json:"instance"` + Events []TranscriptEvent `json:"events"` + }{instance, events}) + if err != nil { + return TranscriptAck{}, err + } + var ack TranscriptAck + if err := json.Unmarshal(raw, &ack); err != nil { + return TranscriptAck{}, fmt.Errorf("%w: transcript: HQ's answer is not {stored, last_seq}: %v", ErrHQUnavailable, err) + } + return ack, nil +} + +// tokenWithheld is what Scrub writes in place of the officer token. +const tokenWithheld = "[officer token withheld]" + +// Scrub is s with every occurrence of the officer's own token replaced — and +// every word a cap cut short ("…" right after it) that ends in the token's +// first 4 or more bytes. It is how text built from an instance's session +// (which could echo anything) is cleared of the one credential this client +// holds, without the token leaving the client: a shape-based secret filter +// cannot know an opaque token's shape. (officership.RedactCutTail is the same +// cut rule for the token SHAPES; it cannot be given this token.) +func (c *Client) Scrub(s string) string { + t := c.officer.Token() + if t == "" { + return s + } + s = strings.ReplaceAll(s, t, tokenWithheld) + const cut = "…" + if !strings.Contains(s, cut) || len(t) < 8 { + return s + } + var b strings.Builder + for rest := s; ; { + j := strings.Index(rest, cut) + if j < 0 { + b.WriteString(rest) + break + } + before := rest[:j] + k := strings.LastIndexAny(before, " \t\n\"'=:") + 1 + word := before[k:] + for i := 0; len(word)-i >= 4; i++ { + if strings.HasPrefix(t, word[i:]) { + before = before[:k] + word[:i] + tokenWithheld + break + } + } + b.WriteString(before) + b.WriteString(cut) + rest = rest[j+len(cut):] + } + return b.String() +} + // --- the release captain route (POST /releases/captain) --------------------- // // HQ's body (civ-hq docs/RELEASES.md §9.3, internal/store/release_captain.go): diff --git a/internal/officers/hqtest/hqtest.go b/internal/officers/hqtest/hqtest.go index 9d2d0bee..afd02f42 100644 --- a/internal/officers/hqtest/hqtest.go +++ b/internal/officers/hqtest/hqtest.go @@ -62,6 +62,52 @@ type Server struct { NoticeStatus int // Preflight is POST /officers/{id}/preflight's answer (404 when nil). Preflight map[string]any + // transcripts are the transcript events stored per item, in seq order + // (POST /officers/{id}/items/{item}/transcript); transcriptStatus, when + // non-zero, is answered instead (e.g. 503). + transcripts map[string][]TranscriptEvent + transcriptStatus int + // failNext answers the next failN transcript posts with failCode, after + // STORING them when failCommit (an HQ that commits and then fails to + // answer: a lost acknowledgement). + failN, failCode int + failCommit bool +} + +// FailTranscript makes the next n transcript posts answer code; commit stores +// each one first (the acknowledgement is lost, the write is not). +func (s *Server) FailTranscript(code, n int, commit bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.failCode, s.failN, s.failCommit = code, n, commit +} + +// TranscriptEvent is one stored transcript event. +type TranscriptEvent struct { + Seq int64 + At string + Kind string + Text string + Instance string +} + +// Transcript returns the events stored for item, in seq order. +func (s *Server) Transcript(item string) []TranscriptEvent { + s.mu.Lock() + defer s.mu.Unlock() + return append([]TranscriptEvent(nil), s.transcripts[item]...) +} + +// SetTranscriptStatus makes the transcript route answer code (0: serve). +func (s *Server) SetTranscriptStatus(code int) { s.mu.Lock(); s.transcriptStatus = code; s.mu.Unlock() } + +// Complete marks item done (as an instance's officer_complete would), so a +// later transcript post for it answers 409 not_claimed. +func (s *Server) Complete(id string) { + s.mu.Lock() + defer s.mu.Unlock() + it := s.items[id] + it.Status, it.ClaimedBy, it.LeaseExpiresAt = "done", "", nil } // Notice is one accepted sensitive-path notice. @@ -334,6 +380,8 @@ func (s *Server) itemVerb(w http.ResponseWriter, officer, id, verb string, body } it.Status, it.Outcome, it.Summary, it.ClaimedBy, it.LeaseExpiresAt, it.UpdatedAt = st, outcome, summary, "", nil, now s.write(w, http.StatusOK, map[string]any{"item": *it}) + case "transcript": + s.transcript(w, officer, it, instance, live, body) case "release": if it.Status != "claimed" || it.ClaimedBy != instance { s.write(w, http.StatusConflict, map[string]any{"error": "lease lost", "code": "lease_lost"}) @@ -345,3 +393,102 @@ func (s *Server) itemVerb(w http.ResponseWriter, officer, id, verb string, body s.write(w, http.StatusNotFound, map[string]any{"error": "no such verb"}) } } + +// Transcript route limits, as HQ states them. +const ( + maxTranscriptEvents = 100 + maxTranscriptText = 8 << 10 +) + +// transcript is POST /officers/{id}/items/{item}/transcript, to the contract: +// {instance, events: [{seq, at, kind, text}]}; ≤100 events, text ≤ 8 KiB, seq +// strictly increasing within the call; idempotent on seq (a seq ≤ the highest +// stored is skipped); 409 officer_disabled, 409 not_claimed (the item is not +// claimed by this instance under a live lease), 400 on a limit. Answers +// {stored, last_seq}. +func (s *Server) transcript(w http.ResponseWriter, officer string, it *Item, instance string, live bool, body map[string]any) { + if s.transcriptStatus != 0 { + s.write(w, s.transcriptStatus, map[string]any{"error": "transcript refused by the fake"}) + return + } + if s.failN > 0 && !s.failCommit { + s.failN-- + s.write(w, s.failCode, map[string]any{"error": "transcript failed by the fake"}) + return + } + for k := range body { + if k != "instance" && k != "events" { + s.write(w, http.StatusBadRequest, map[string]any{"error": "unknown field " + k, "code": "bad_request"}) + return + } + } + if s.Disabled[officer] { + s.write(w, http.StatusConflict, map[string]any{"error": "officer disabled", "code": "officer_disabled"}) + return + } + if !live || it.ClaimedBy != instance || instance == "" { + s.write(w, http.StatusConflict, map[string]any{"error": "item not claimed by this instance", "code": "not_claimed"}) + return + } + raw, ok := body["events"].([]any) + if !ok || len(raw) > maxTranscriptEvents { + s.write(w, http.StatusBadRequest, map[string]any{"error": "events must be an array of at most 100", "code": "limits"}) + return + } + if s.transcripts == nil { + s.transcripts = map[string][]TranscriptEvent{} + } + var last int64 + if have := s.transcripts[it.ID]; len(have) > 0 { + last = have[len(have)-1].Seq + } + var batch []TranscriptEvent + prev := int64(-1 << 62) + for _, e := range raw { + m, ok := e.(map[string]any) + if !ok { + s.write(w, http.StatusBadRequest, map[string]any{"error": "an event is not an object", "code": "limits"}) + return + } + seqF, _ := m["seq"].(float64) + seq := int64(seqF) + at, _ := m["at"].(string) + kind, _ := m["kind"].(string) + text, _ := m["text"].(string) + if _, err := time.Parse(time.RFC3339, at); err != nil { + s.write(w, http.StatusBadRequest, map[string]any{"error": "at is not RFC3339", "code": "limits"}) + return + } + switch kind { + case "user", "assistant", "tool_use", "tool_result", "system": + default: + s.write(w, http.StatusBadRequest, map[string]any{"error": "unknown kind " + kind, "code": "limits"}) + return + } + if len(text) > maxTranscriptText { + s.write(w, http.StatusBadRequest, map[string]any{"error": "text over 8 KiB", "code": "limits"}) + return + } + if seq <= prev { + s.write(w, http.StatusBadRequest, map[string]any{"error": "seq not strictly increasing", "code": "limits"}) + return + } + prev = seq + batch = append(batch, TranscriptEvent{Seq: seq, At: at, Kind: kind, Text: text, Instance: instance}) + } + stored := 0 + for _, e := range batch { + if e.Seq <= last { + continue + } + s.transcripts[it.ID] = append(s.transcripts[it.ID], e) + last = e.Seq + stored++ + } + if s.failN > 0 && s.failCommit { + s.failN-- + s.write(w, s.failCode, map[string]any{"error": "committed, then failed to answer"}) + return + } + s.write(w, http.StatusOK, map[string]any{"stored": stored, "last_seq": last}) +} diff --git a/internal/officers/transcript/forward.go b/internal/officers/transcript/forward.go new file mode 100644 index 00000000..4ede0056 --- /dev/null +++ b/internal/officers/transcript/forward.go @@ -0,0 +1,807 @@ +// Package transcript forwards an officer instance's LIVE session transcript to +// HQ while the instance holds a claimed item (docs/officers.md, "Transcript to +// HQ"), so HQ's officer pages can show what the officer is doing. +// +// THE SOURCE is the instance's own Claude Code transcript — the structured +// JSONL the CLI writes, decoded by internal/cctranscript exactly as every other +// reader of a session decodes it — served incrementally by the instance pod's +// ccd on GET /v1/transcript (cmd/ccd/transcript_route.go). Never the screen. +// +// THE SINK is HQ's POST /officers/{id}/items/{item}/transcript, sent with the +// officer's own token by officers.Client (the token never leaves it). +// +// 🔴 EVERY TEXT IS REDACTED BEFORE IT LEAVES: the officer's own token is +// replaced (officers.Client.Scrub — a shape filter cannot know an opaque +// token), then the CI-log secret filter (officership.RedactText, the same +// RedactLines officer_ci_logs uses) withholds every line that looks like it +// carries a credential, then the text is cut to the contract's 8 KiB with a +// marker. Transcript text is never logged. +// +// 🔴 THE SEQ IS A FUNCTION OF THE TRANSCRIPT POSITION, NOT OF MEMORY (Seq): +// the end offset of the event's line in the transcript file times 32768, plus the +// event's index within that line. It strictly increases along the file, so a +// restarted muster derives the SAME seq for the same event, and HQ's +// idempotency rule (a seq at or below its highest stored is skipped) turns a +// resume's re-read into no duplicates. A resume re-reads a backfill window +// (Config.Backfill) of the transcript, so events written while muster was +// down are sent too, as long as they fit in it. It stays below 2^53 for any +// file under 256 GiB, so it survives a JSON reader that parses numbers as +// doubles. +// +// The one case the file position cannot order is a NEW CONVERSATION mid-item +// (a new transcript file, offsets from zero — ccd resumes the primary's own +// conversation by id, so this is an operator's /clear, not a restart). A +// forwarder that sees it shifts every later seq above the highest it had +// (worker.shift). A forwarder that restarted INTO such a file has lost the +// shift; HQ then answers stored 0 with a last_seq above a batch read LIVE +// (past the resume's re-read, so it cannot be a duplicate), and the forwarder +// re-sequences that batch and every later one above it (heal). What that case +// can lose is the part of the resume's re-read HQ skipped as old. +// +// 🔴 IT NEVER BLOCKS THE OFFICER. Claimed, Ended and Sync — called from the +// officer tools and the wake loop — only record state under a mutex; every +// read and post runs on the forwarder's own loop (Run → Tick). An instance's +// buffer is capped (Config.MaxBuffer): past it the OLDEST events are dropped +// and counted (muster_officer_transcript_events_total{result="dropped_buffer"}), +// never held without bound. +package transcript + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "sort" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/ZacxDev/muster/internal/cctranscript" + "github.com/ZacxDev/muster/internal/metrics" + "github.com/ZacxDev/muster/internal/officers" + "github.com/ZacxDev/muster/internal/officership" +) + +// Query is one read of an instance's transcript. +type Query struct { + // Session is the session the reader's offset belongs to ("" for none yet). + Session string + // After is the offset to read from; -1 is the tail (the last Backfill + // bytes; Backfill 0 starts live, at the end). + After int64 + Backfill int +} + +// Source reads an officer instance's transcript (production: the instance +// pod's ccd, GET /v1/transcript). +type Source interface { + Read(ctx context.Context, o officers.Officer, instance string, q Query) (cctranscript.Page, error) +} + +// Poster is the HQ side: the transcript route, and the officer-token scrub. +type Poster interface { + Post(ctx context.Context, o officers.Officer, item officers.ItemID, instance string, events []officers.TranscriptEvent) (officers.TranscriptAck, error) + Scrub(o officers.Officer, s string) string +} + +type clientPoster struct{ hq *officers.ClientHQ } + +// NewClientPoster is a Poster over each officer's own HQ client. +func NewClientPoster(hq *officers.ClientHQ) Poster { return clientPoster{hq: hq} } + +func (p clientPoster) Post(ctx context.Context, o officers.Officer, item officers.ItemID, instance string, events []officers.TranscriptEvent) (officers.TranscriptAck, error) { + return p.hq.Client(o).PostTranscript(ctx, item, instance, events) +} + +func (p clientPoster) Scrub(o officers.Officer, s string) string { return p.hq.Client(o).Scrub(s) } + +// Defaults. +const ( + DefaultInterval = 2 * time.Second + DefaultMaxBuffer = 1000 + DefaultMaxRetries = 5 + DefaultMaxBackoff = time.Minute + DefaultBackfill = 256 << 10 + DefaultMaxPostsPerTick = 10 + DefaultRequestTimeout = 15 * time.Second +) + +// Config is New's input. Zero values take the defaults. +type Config struct { + Source Source + Poster Poster + Logger *log.Logger + Now func() time.Time + // Interval is the cadence: how often each instance's transcript is read + // and its new events posted. + Interval time.Duration + // MaxBuffer caps one instance's events read but not yet accepted by HQ. + MaxBuffer int + // MaxRetries is how many times one batch is re-sent after an HQ outage + // (5xx, 429, no answer) before it is dropped (counted). + MaxRetries int + // MaxBackoff caps the wait between re-sends (Interval doubled per failure). + MaxBackoff time.Duration + // Backfill is the window, in bytes, a RESUME (a held item muster was not + // forwarding — after a restart) re-reads from the end of the transcript. + Backfill int + // MaxPostsPerTick bounds one instance's posts per tick. + MaxPostsPerTick int + // RequestTimeout bounds one read or post. + RequestTimeout time.Duration +} + +// Results, the label values of muster_officer_transcript_events_total. +const ( + resultPosted = "posted" + resultBuffer = "dropped_buffer" + resultRetries = "dropped_retries" + resultRefused = "dropped_refused" + resultSuperseded = "dropped_superseded" + resultUnsequenced = "dropped_unsequenced" + resultTooLarge = "dropped_too_large" + resultSkippedByHQ = "skipped_by_hq" + resultSkipResume = "skipped_resume" + truncatedMarkerFmt = "… [truncated %d bytes]" +) + +// Forwarder is the transcript forwarder: one worker per officer instance that +// holds a claimed item of a transcript-enabled officer. +type Forwarder struct { + cfg Config + log *log.Logger + now func() time.Time + + tickMu sync.Mutex // one Tick at a time + + mu sync.Mutex + workers map[string]*worker // by instance agent name + // stopped is (officer, instance, item) HQ refused (409 not_claimed …): not + // restarted by a pass until a new claim, or until the item leaves the + // held list. + stopped map[key]bool + // seen is every instance this forwarder has had a worker for. A resume + // of an instance it has SEEN starts live: what the transcript holds since + // it last forwarded is another item's work (it claimed another, or + // completed one), not this one's. + seen map[string]bool +} + +type key struct { + officer, instance string + item officers.ItemID +} + +// worker is one instance's forwarding state. The control fields are guarded +// by Forwarder.mu; the data fields belong to the tick loop alone. +type worker struct { + o officers.Officer + instance string + + // control (Forwarder.mu) + item officers.ItemID + gen int // bumped when item changes + since time.Time // when this item's forwarding began + dead bool + + // data (tick only) + seenGen int + session string + cursor int64 + backfill int + // notBefore, set, skips lines older than it on the first read: a claim's + // forwarding starts at the claim (less ClaimSlack), not at whatever the + // backfill window holds of the instance's earlier work. + notBefore time.Time + buf []officers.TranscriptEvent + lastSeq int64 + // shift is added to every seq read from the current transcript file (a + // new conversation mid-item: Seq's doc). + shift int64 + // maxSeq is the highest seq this worker has assigned. + maxSeq int64 + // replayMax is the highest seq its FIRST read (a claim's or a resume's + // window) assigned. + replayMax int64 + reads int + // resume: this worker began on a pass's view of a held item, not on a + // claim. probed: HQ's last_seq for the item is known (floor), learned by + // an empty post before the first events are sent. + resume bool + probed bool + floor int64 + // ackedOwn is the highest seq of this worker's own events HQ has + // answered for. + ackedOwn int64 + // batchMax is the most events one post carries (halved by a 413). + batchMax int + failures int + nextAt time.Time + readFailing bool +} + +// New builds a forwarder. +func New(cfg Config) (*Forwarder, error) { + if cfg.Source == nil || cfg.Poster == nil { + return nil, errors.New("transcript: a forwarder needs a source and a poster") + } + if cfg.Logger == nil { + cfg.Logger = log.Default() + } + if cfg.Now == nil { + cfg.Now = time.Now + } + if cfg.Interval <= 0 { + cfg.Interval = DefaultInterval + } + if cfg.MaxBuffer <= 0 { + cfg.MaxBuffer = DefaultMaxBuffer + } + if cfg.MaxRetries <= 0 { + cfg.MaxRetries = DefaultMaxRetries + } + if cfg.MaxBackoff <= 0 { + cfg.MaxBackoff = DefaultMaxBackoff + } + if cfg.Backfill <= 0 { + cfg.Backfill = DefaultBackfill + } + if cfg.MaxPostsPerTick <= 0 { + cfg.MaxPostsPerTick = DefaultMaxPostsPerTick + } + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = DefaultRequestTimeout + } + return &Forwarder{cfg: cfg, log: cfg.Logger, now: cfg.Now, workers: map[string]*worker{}, stopped: map[key]bool{}, seen: map[string]bool{}}, nil +} + +// ClaimSlack is how far before a claim's own time its first read reaches: the +// pod's clock and muster's can differ, and the officer_claim call itself (the +// tool_use, written just before the claim) belongs to the item. +const ClaimSlack = 5 * time.Second + +// Claimed is officer_claim succeeding: instance now holds item. Forwarding +// starts AT THE CLAIM: the first read takes the transcript's backfill window +// and keeps only the lines written since the claim (less ClaimSlack) — what +// the instance does from the claim on is the item's, and nothing written +// between the claim and the forwarder's next tick is lost. An instance already +// forwarding another item switches to this one, keeping its place in the +// transcript. +// +// An untimed line in that first read cannot be shown to follow the claim and +// is not sent. +func (f *Forwarder) Claimed(o officers.Officer, instance string, item officers.ItemID) { + if !o.TranscriptEnabled() { + return + } + f.mu.Lock() + defer f.mu.Unlock() + delete(f.stopped, key{o.ID, instance, item}) + now := f.now() + if w := f.workers[instance]; w != nil && !w.dead { + if w.item != item { + w.item, w.since = item, now + w.gen++ + } + return + } + f.workers[instance] = &worker{o: o, instance: instance, item: item, since: now, cursor: -1, backfill: f.cfg.Backfill, + notBefore: now.Add(-ClaimSlack), batchMax: officers.MaxTranscriptEvents} + f.seen[instance] = true +} + +// Ended is officer_complete or officer_release succeeding: the item is no +// longer the instance's, and forwarding for it stops. +func (f *Forwarder) Ended(o officers.Officer, instance string, item officers.ItemID) { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.stopped, key{o.ID, instance, item}) + if w := f.workers[instance]; w != nil && w.item == item { + w.dead = true + delete(f.workers, instance) + } +} + +// Sync is one wake-loop pass's view of o's claimed items, read at at: every +// item one of o's instances holds under a live lease is forwarded (a RESUME, +// from the backfill window, when muster was not forwarding it — after a +// restart), and forwarding stops for an item no instance of o holds any more. +// A claim made after at (Claimed) is not undone by this pass. +func (f *Forwarder) Sync(o officers.Officer, claimed []officers.Item, at time.Time) { + if !o.TranscriptEnabled() { + return + } + reg := officers.NewRegistry([]officers.Officer{o}) + held := map[string][]officers.ItemID{} + heldKey := map[key]bool{} + for _, it := range claimed { + if !it.LeaseLive(at) { + continue + } + if _, _, ok := reg.ForAgentName(it.ClaimedBy); !ok { + continue + } + held[it.ClaimedBy] = append(held[it.ClaimedBy], it.ID) + heldKey[key{o.ID, it.ClaimedBy, it.ID}] = true + } + f.mu.Lock() + defer f.mu.Unlock() + for k := range f.stopped { + if k.officer == o.ID && !heldKey[k] { + delete(f.stopped, k) + } + } + for inst, w := range f.workers { + if w.o.ID != o.ID || heldKey[key{o.ID, inst, w.item}] || w.since.After(at) { + continue + } + w.dead = true + delete(f.workers, inst) + } + insts := make([]string, 0, len(held)) + for inst := range held { + insts = append(insts, inst) + } + sort.Strings(insts) + for _, inst := range insts { + if f.workers[inst] != nil { + continue // forwarding an item it still holds + } + items := held[inst] + sort.Slice(items, func(i, j int) bool { return items[i] < items[j] }) + for _, item := range items { + if f.stopped[key{o.ID, inst, item}] { + continue + } + backfill := f.cfg.Backfill + if f.seen[inst] { + backfill = 0 // live: see Forwarder.seen + } + f.workers[inst] = &worker{o: o, instance: inst, item: item, since: f.now(), cursor: -1, backfill: backfill, resume: true, + batchMax: officers.MaxTranscriptEvents} + f.seen[inst] = true + f.log.Printf("officers: %s: forwarding instance %s's transcript for item %s to HQ (resumed: the instance holds it)", o.ID, inst, item) + break + } + } +} + +// Active is each forwarding instance's item. +func (f *Forwarder) Active() map[string]officers.ItemID { + f.mu.Lock() + defer f.mu.Unlock() + out := map[string]officers.ItemID{} + for inst, w := range f.workers { + out[inst] = w.item + } + return out +} + +// Run ticks every Interval until ctx ends. +func (f *Forwarder) Run(ctx context.Context) { + t := time.NewTicker(f.cfg.Interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + f.tick(ctx) + } + } +} + +func (f *Forwarder) tick(ctx context.Context) { + defer func() { + if rec := recover(); rec != nil { + f.log.Printf("officers: PANIC in the transcript forwarder: %v", rec) + } + }() + f.Tick(ctx) +} + +// Tick runs one step for every worker: read its transcript's new lines, then +// post what is buffered (unless it is backing off). +func (f *Forwarder) Tick(ctx context.Context) { + f.tickMu.Lock() + defer f.tickMu.Unlock() + f.mu.Lock() + ws := make([]*worker, 0, len(f.workers)) + for _, w := range f.workers { + ws = append(ws, w) + } + f.mu.Unlock() + sort.Slice(ws, func(i, j int) bool { return ws[i].instance < ws[j].instance }) + for _, w := range ws { + if ctx.Err() != nil { + return + } + f.step(ctx, w) + } +} + +func (f *Forwarder) step(ctx context.Context, w *worker) { + f.mu.Lock() + dead, item, gen := w.dead, w.item, w.gen + f.mu.Unlock() + if dead { + return + } + if gen != w.seenGen { + // A new item: what was buffered for the old one is not sent to it. + f.count(resultSuperseded, len(w.buf)) + w.buf, w.lastSeq, w.failures, w.nextAt, w.seenGen, w.replayMax = nil, 0, 0, time.Time{}, gen, 0 + w.probed, w.resume, w.floor, w.ackedOwn = false, false, 0, 0 + } + f.read(ctx, w) + f.flush(ctx, w, item, gen) +} + +func (f *Forwarder) count(result string, n int) { + if n > 0 { + metrics.OfficerTranscriptEvents.WithLabelValues(result).Add(float64(n)) + } +} + +// read appends the transcript's new events to w's buffer, mapped, redacted, +// cut and sequenced, then applies the buffer cap. +func (f *Forwarder) read(ctx context.Context, w *worker) { + rctx, cancel := context.WithTimeout(ctx, f.cfg.RequestTimeout) + page, err := f.cfg.Source.Read(rctx, w.o, w.instance, Query{Session: w.session, After: w.cursor, Backfill: w.backfill}) + cancel() + if err != nil { + if !w.readFailing { + f.log.Printf("officers: %s: instance %s's transcript cannot be read (retrying every %s): %v", w.o.ID, w.instance, f.cfg.Interval, err) + } + w.readFailing = true + return + } + if w.readFailing { + f.log.Printf("officers: %s: instance %s's transcript reads again", w.o.ID, w.instance) + w.readFailing = false + } + if w.session != "" && page.Session != w.session { + // A new conversation: its offsets start again from zero, so its seqs + // are shifted above every seq this worker assigned. + w.shift = max(w.maxSeq, w.lastSeq) + 1 + f.log.Printf("officers: %s: instance %s's session changed; its transcript seqs continue above %d", w.o.ID, w.instance, w.shift) + } + w.session, w.cursor, w.backfill = page.Session, page.Next, 0 + notBefore := w.notBefore + w.notBefore = time.Time{} + w.reads++ + // Only seqs THIS worker assigned are skipped here; HQ's last_seq is not + // (a live event below it is the heal's case, in flush). + last := w.maxSeq + now := f.now() + unsequenced := 0 + for _, l := range page.Lines { + at := now + if l.At > 0 { + at = time.UnixMilli(l.At) + } + if !notBefore.IsZero() && (l.At <= 0 || at.Before(notBefore)) { + continue // the instance's work before this claim (or not shown to follow it) + } + for i, ev := range l.Events { + kind, text, ok := mapEvent(ev) + if !ok { + continue + } + seq, ok := Seq(l.End, i) + if seq += w.shift; !ok || seq >= maxSeq { + unsequenced++ + continue + } + if seq <= last { + continue // already sent (a resume's re-read, a re-served line) + } + w.buf = append(w.buf, officers.TranscriptEvent{Seq: seq, At: at.UTC().Format(time.RFC3339Nano), Kind: kind, + Text: f.clean(w.o, text)}) + last = seq + w.maxSeq = seq + if w.reads == 1 { + w.replayMax = seq + } + } + } + if unsequenced > 0 { + f.count(resultUnsequenced, unsequenced) + f.log.Printf("officers: %s: instance %s: dropped %d transcript event(s) with no valid seq (a line with more than %d events)", + w.o.ID, w.instance, unsequenced, maxIdx+1) + } + if over := len(w.buf) - f.cfg.MaxBuffer; over > 0 { + w.buf = append([]officers.TranscriptEvent(nil), w.buf[over:]...) + f.count(resultBuffer, over) + f.log.Printf("officers: %s: instance %s: transcript buffer full (%d events waiting for HQ): dropped the oldest %d", + w.o.ID, w.instance, f.cfg.MaxBuffer, over) + } +} + +// flush posts w's buffer in batches of at most batchMax (MaxTranscriptEvents, +// halved by each 413) — after, once per item, the PROBE: an empty post +// whose answer is HQ's last_seq for the item (the floor). +// +// - A RESUME drops what HQ already holds (seq at or below the floor); with +// nothing at HQ (floor 0) it drops the whole first read, which is the +// instance's EARLIER work (the claim's own time is unknown), and starts +// live. +// - Any batch whose seqs are not above the floor — a re-claim by ANOTHER +// instance (its own transcript, offsets from zero), or a conversation +// this restarted forwarder did not see begin — is shifted above it, with +// everything after it, instead of being skipped as old. +// - A probe HQ refuses with 400 (an HQ that does not take an empty post) +// leaves the floor at 0 and a resume starting live; the heal below still +// covers a shifted file. +func (f *Forwarder) flush(ctx context.Context, w *worker, item officers.ItemID, gen int) { + now := f.now() + if now.Before(w.nextAt) { + return + } + if !w.probed && !f.probe(ctx, w, item, gen, now) { + return + } + for posts := 0; len(w.buf) > 0 && posts < f.cfg.MaxPostsPerTick; posts++ { + n := min(len(w.buf), max(w.batchMax, 1)) + if w.buf[0].Seq <= w.floor { + f.resequence(w, w.floor-w.buf[0].Seq+1) + } + batch := w.buf[:n] + first := w.failures == 0 + pctx, cancel := context.WithTimeout(ctx, f.cfg.RequestTimeout) + ack, err := f.cfg.Poster.Post(pctx, w.o, item, w.instance, batch) + cancel() + var hqErr *officers.HQError + switch { + case err == nil && first && ack.Stored == 0 && ack.LastSeq >= batch[n-1].Seq && ack.LastSeq > w.ackedOwn: + // A batch's FIRST send, skipped whole, with HQ past every seq this + // worker ever had answered: HQ's seqs for the item came from a file + // this forwarder never saw (a new conversation before a restart, + // with no usable probe). Move it and everything after it above + // HQ's last_seq and send it again. NOT on a re-send: a re-send + // answered stored 0 is a lost acknowledgement (HQ committed, then + // failed to answer), and re-sequencing it would duplicate it. + // (ackedOwn is defence in depth: no path today re-sends a batch HQ + // answered as a FIRST send, so the first-send rule alone decides — + // a mutation removing ackedOwn survives the suite for that reason.) + f.resequence(w, ack.LastSeq-batch[0].Seq+1) + w.floor = max(w.floor, ack.LastSeq) + f.log.Printf("officers: %s: instance %s: HQ holds item %s's transcript past this position; re-sequenced above %d", + w.o.ID, w.instance, item, ack.LastSeq) + case err == nil: + stored := min(max(ack.Stored, 0), n) + f.count(resultPosted, stored) + if skipped := n - stored; skipped > 0 { + f.count(resultSkippedByHQ, skipped) + f.log.Printf("officers: %s: instance %s: HQ already held %d of a transcript batch of %d for item %s (a re-send after a lost answer, or a resume)", + w.o.ID, w.instance, skipped, n, item) + } + w.lastSeq = max(w.lastSeq, batch[n-1].Seq, ack.LastSeq) + w.ackedOwn = max(w.ackedOwn, batch[n-1].Seq) + w.buf = w.buf[n:] + w.failures, w.nextAt = 0, time.Time{} + case errors.As(err, &hqErr) && hqErr.Status == http.StatusRequestEntityTooLarge: + // Too large for HQ: drop THAT batch, send the rest at half the size + // (for the rest of this item: HQ's size limit does not shrink). + f.count(resultTooLarge, n) + w.buf = w.buf[n:] + w.batchMax = max(n/2, 1) + w.failures = 0 + f.log.Printf("officers: %s: instance %s: HQ answered 413 to a transcript batch of %d for item %s; dropped, next batches at most %d", + w.o.ID, w.instance, n, item, w.batchMax) + case errors.As(err, &hqErr) && hqErr.Status == http.StatusBadRequest: + f.count(resultRefused, n) + w.buf = w.buf[n:] + w.failures = 0 + f.log.Printf("officers: %s: instance %s: HQ refused a transcript batch of %d for item %s (400%s); dropped", + w.o.ID, w.instance, n, item, hqCode(hqErr)) + case errors.As(err, &hqErr) && hqErr.Status != http.StatusTooManyRequests: + // 409 not_claimed / officer_disabled, 404 not this officer's item, + // 401/403: a decision, not an outage — stop for this item. + f.log.Printf("officers: %s: instance %s: stopped forwarding item %s's transcript: HQ answered %d%s", + w.o.ID, w.instance, item, hqErr.Status, hqCode(hqErr)) + f.stop(w, gen, item) + return + default: + // An outage (5xx, 429, no answer): back off; retries are capped. + w.failures++ + if w.failures > f.cfg.MaxRetries { + f.count(resultRetries, n) + f.log.Printf("officers: %s: instance %s: HQ did not take a transcript batch of %d for item %s after %d attempts; dropped (%v)", + w.o.ID, w.instance, n, item, w.failures, err) + w.buf = w.buf[n:] + w.failures = 0 + } + w.nextAt = now.Add(f.backoff(w.failures)) + return + } + } +} + +// probe learns HQ's last_seq for item (an empty post) and applies it to a +// resume's first read (flush's doc). It reports whether posting may go on. +func (f *Forwarder) probe(ctx context.Context, w *worker, item officers.ItemID, gen int, now time.Time) bool { + pctx, cancel := context.WithTimeout(ctx, f.cfg.RequestTimeout) + ack, err := f.cfg.Poster.Post(pctx, w.o, item, w.instance, []officers.TranscriptEvent{}) + cancel() + var hqErr *officers.HQError + switch { + case err == nil: + w.floor = max(ack.LastSeq, 0) + case errors.As(err, &hqErr) && hqErr.Status == http.StatusBadRequest: + w.floor = 0 + f.log.Printf("officers: %s: instance %s: HQ does not answer an empty transcript post for item %s (400%s); sending without its last_seq", + w.o.ID, w.instance, item, hqCode(hqErr)) + case errors.As(err, &hqErr) && hqErr.Status != http.StatusTooManyRequests && hqErr.Status != http.StatusRequestEntityTooLarge: + f.log.Printf("officers: %s: instance %s: stopped forwarding item %s's transcript: HQ answered %d%s", + w.o.ID, w.instance, item, hqErr.Status, hqCode(hqErr)) + f.stop(w, gen, item) + return false + default: + w.failures++ + w.nextAt = now.Add(f.backoff(w.failures)) + return false + } + w.probed, w.failures, w.nextAt = true, 0, time.Time{} + if w.resume { + drop := w.floor + if drop == 0 { + drop = w.replayMax // nothing at HQ: the first read is earlier work + } + kept := w.buf[:0] + skipped := 0 + for _, e := range w.buf { + if e.Seq <= drop { + skipped++ + continue + } + kept = append(kept, e) + } + w.buf = kept + f.count(resultSkipResume, skipped) + } + return true +} + +// resequence moves every buffered seq, and every later one, up by delta. +func (f *Forwarder) resequence(w *worker, delta int64) { + for i := range w.buf { + w.buf[i].Seq += delta + } + w.shift += delta + w.maxSeq += delta +} + +func (f *Forwarder) backoff(failures int) time.Duration { + d := f.cfg.Interval + for i := 0; i < failures && d < f.cfg.MaxBackoff; i++ { + d *= 2 + } + return min(d, f.cfg.MaxBackoff) +} + +// stop ends w for item after an HQ refusal, unless the worker moved on. +func (f *Forwarder) stop(w *worker, gen int, item officers.ItemID) { + f.mu.Lock() + defer f.mu.Unlock() + f.stopped[key{w.o.ID, w.instance, item}] = true + if w.gen == gen && !w.dead { + w.dead = true + if f.workers[w.instance] == w { + delete(f.workers, w.instance) + } + } +} + +// hqCode is HQ's error code from a refusal body, as " code" or "". The body's +// other fields are not logged: only HQ's own vocabulary. +func hqCode(e *officers.HQError) string { + var b struct { + Code string `json:"code"` + } + if json.Unmarshal(e.Body, &b) == nil && b.Code != "" && len(b.Code) <= 64 && !strings.ContainsAny(b.Code, " \n") { + return " " + b.Code + } + return "" +} + +// --- mapping ----------------------------------------------------------------- + +// mapEvent is THE mapping from a transcript event to the contract's kinds: +// the human's prompt → user; assistant text → assistant; a tool call → +// tool_use as "ToolName: "; a tool result → tool_result; an API error +// or a harness notice → system. Injected context draws nothing and is not +// sent. +func mapEvent(ev cctranscript.Event) (kind, text string, ok bool) { + switch ev.Kind { + case cctranscript.KindPrompt: + return officers.TranscriptUser, ev.Text, true + case cctranscript.KindText: + return officers.TranscriptAssistant, ev.Text, true + case cctranscript.KindToolCall: + return officers.TranscriptToolUse, toolSummary(ev), true + case cctranscript.KindToolResult: + if !ev.OK { + return officers.TranscriptToolResult, "[error] " + ev.Output, true + } + return officers.TranscriptToolResult, ev.Output, true + case cctranscript.KindError: + return officers.TranscriptSystem, "API error: " + ev.Text, true + case cctranscript.KindNotice: + if ev.Status != "" { + return officers.TranscriptSystem, ev.Text + " (" + ev.Status + ")", true + } + return officers.TranscriptSystem, ev.Text, true + } + return "", "", false +} + +func toolSummary(ev cctranscript.Event) string { + arg := strings.TrimSpace(ev.Target) + if arg == "" { + // Not cut here: the whole input (ccd caps it at 2 KiB) is scrubbed and + // redacted first, and only then cut, by clean. + arg = strings.Join(strings.Fields(ev.Input), " ") + } + if arg == "" { + return ev.Tool + } + return ev.Tool + ": " + arg +} + +// clean is THE text pipeline, in this order: the officer's token replaced +// (whole, and a cut prefix of it), the CI-log secret filter, a line ending in +// a cut token prefix withheld (ccd caps long fields with "…" BEFORE they reach +// muster, so a token can arrive cut short of its full pattern), and only then +// the 8 KiB cut — no cut of muster's own happens before the filters. +func (f *Forwarder) clean(o officers.Officer, s string) string { + return Truncate(officership.RedactCutTail(officership.RedactText(f.cfg.Poster.Scrub(o, s)), nil), officers.MaxTranscriptText) +} + +// Truncate cuts s to at most n bytes on a character boundary, ending it with +// "… [truncated N bytes]" (N the bytes removed) when it cuts. +func Truncate(s string, n int) string { + if len(s) <= n { + return s + } + // The marker for the most bytes that could be removed is the longest. + keep := n - len(fmt.Sprintf(truncatedMarkerFmt, len(s))) + if keep < 0 { + keep = 0 + } + for keep > 0 && !utf8.RuneStart(s[keep]) { + keep-- + } + return s[:keep] + fmt.Sprintf(truncatedMarkerFmt, len(s)-keep) +} + +// --- the seq ------------------------------------------------------------------- + +// The seq's layout: the line's end offset in its transcript file, then 15 bits +// of the event's index within that line. The end offset strictly increases +// from line to line along one file and the index within one, so seq does. +// maxSeq keeps it a JSON-safe integer (below 2^53): 2^53 / 2^15 = 2^38, so an +// end offset below 256 GiB (a transcript far past any session's). +// +// 🔴 15 BITS SO NO LINE'S EVENTS ARE EVER DROPPED. ccd never serves a line +// longer than its 1 MiB read window (cctranscript.ReadLines skips one), and an +// event is at least one content block of well over 32 bytes, so a served line +// carries fewer than 2^20 / 32 = 32768 events. (4 bits dropped every event +// past the 16th of a user record carrying many parallel tool results.) +const ( + idxBits = 15 + maxIdx = 1<= maxSeq>>idxBits || idx < 0 || idx > maxIdx { + return 0, false + } + return end< from { + p.Lines = append(p.Lines, l) + } + } + return p, nil +} + +type rig struct { + t *testing.T + hq *hqtest.Server + src *fakeSource + o officers.Officer + fwd *transcript.Forwarder + now time.Time + log *bytes.Buffer + cfg transcript.Config +} + +func officer(t *testing.T, hq *hqtest.Server, enabled bool) officers.Officer { + t.Helper() + tr := "" + if enabled { + tr = `,"transcript":{"enabled":true}` + } + list, err := officers.Parse(`[{"id":"`+officerID+`","hq_url":"`+hq.BaseURL()+`","max_instances":2`+tr+`}]`, + func(k string) string { + if k == officers.TokenEnvPrefix+officers.TokenEnvSuffix(officerID) { + return officerToken + } + return "" + }, "muster-agent-") + if err != nil { + t.Fatal(err) + } + return list[0] +} + +func newRig(t *testing.T, enabled bool, tweak func(*transcript.Config)) *rig { + t.Helper() + hq := hqtest.New(map[string]string{officerToken: officerID}) + t.Cleanup(hq.Close) + r := &rig{t: t, hq: hq, src: newSource(), now: time.Date(2026, 11, 1, 12, 0, 0, 0, time.UTC), log: &bytes.Buffer{}} + hq.Now = func() time.Time { return r.now } + r.o = officer(t, hq, enabled) + r.cfg = transcript.Config{ + Source: r.src, + Poster: transcript.NewClientPoster(officers.NewClientHQ(nil)), + Logger: log.New(r.log, "", 0), + Now: func() time.Time { return r.now }, + } + if tweak != nil { + tweak(&r.cfg) + } + r.restart() + return r +} + +// restart is a fresh Forwarder over the same source and HQ: what a muster +// restart leaves. +func (r *rig) restart() { + f, err := transcript.New(r.cfg) + if err != nil { + r.t.Fatal(err) + } + r.fwd = f +} + +// claim makes HQ hold item for the instance under a live lease. +func (r *rig) claim() string { + id := r.hq.AddItem(officerID, "release", "subject-1") + r.hq.SetClaim(id, instance, r.now.Add(24*time.Hour)) // outlives every test clock + return id +} + +func (r *rig) tick() { + r.fwd.Tick(context.Background()) +} + +// advance moves the clock past any backoff and ticks. +func (r *rig) advance(d time.Duration) { + r.now = r.now.Add(d) + r.tick() +} + +// posts is every request to the transcript route that carried events. +func (r *rig) posts() []hqtest.Request { + var out []hqtest.Request + for _, q := range r.hq.Requests() { + if strings.HasSuffix(q.Path, "/transcript") && len(eventsOf(q)) > 0 { + out = append(out, q) + } + } + return out +} + +// probes is every transcript post with NO events: the forwarder learning +// HQ's last_seq for an item before it sends any. +func (r *rig) probes() []hqtest.Request { + var out []hqtest.Request + for _, q := range r.hq.Requests() { + if strings.HasSuffix(q.Path, "/transcript") && len(eventsOf(q)) == 0 { + out = append(out, q) + } + } + return out +} + +func eventsOf(q hqtest.Request) []any { + evs, _ := q.Body["events"].([]any) + return evs +} + +func at(min int) time.Time { return time.Date(2026, 11, 1, 12, min, 0, 0, time.UTC) } + +func TestForwardsNewEventsMappedToKindsInStrictSeqOrder(t *testing.T) { + r := newRig(t, true, nil) + // Before the claim: the officer's earlier work, which is NOT this item's. + r.src.add(r.now.Add(-time.Minute), cctranscript.Event{Kind: cctranscript.KindText, Text: "before the claim"}) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindPrompt, Text: "work the queue"}) + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "Looking at the release."}, + cctranscript.Event{Kind: cctranscript.KindToolCall, ToolID: "t1", Tool: "Bash", Target: "go test ./...", Input: `{"command":"go test ./..."}`}) + r.src.add(at(3), cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "t1", OK: true, Output: "ok \tpkg\t0.1s"}) + r.src.add(at(4), cctranscript.Event{Kind: cctranscript.KindError, Text: "rate limited"}) + r.src.add(at(5), cctranscript.Event{Kind: cctranscript.KindNotice, Notice: cctranscript.NoticeTask, Text: "background task finished"}) + r.src.add(at(5), cctranscript.Event{Kind: cctranscript.KindInjected, Text: "diagnostics"}) + r.tick() + + got := r.hq.Transcript(item) + want := []struct{ kind, text string }{ + {"user", "work the queue"}, + {"assistant", "Looking at the release."}, + {"tool_use", "Bash: go test ./..."}, + {"tool_result", "ok \tpkg\t0.1s"}, + {"system", "rate limited"}, + {"system", "background task finished"}, + } + if len(got) != len(want) { + t.Fatalf("stored %d events, want %d: %+v (log: %s)", len(got), len(want), got, r.log) + } + for i, w := range want { + if got[i].Kind != w.kind || !strings.Contains(got[i].Text, w.text) { + t.Errorf("event %d = %s %q, want %s containing %q", i, got[i].Kind, got[i].Text, w.kind, w.text) + } + if got[i].Instance != instance { + t.Errorf("event %d posted for instance %q", i, got[i].Instance) + } + if i > 0 && got[i].Seq <= got[i-1].Seq { + t.Errorf("seq not strictly increasing: %d then %d", got[i-1].Seq, got[i].Seq) + } + } + if got[0].At != at(1).Format(time.RFC3339) { + t.Errorf("at = %q, want the record's own time %q", got[0].At, at(1).Format(time.RFC3339)) + } + // The officer token is the bearer, and nowhere in a body. + for _, q := range r.posts() { + if q.Auth != "Bearer "+officerToken { + t.Fatalf("auth %q", q.Auth) + } + } + + // A second tick with nothing new posts nothing. + n := len(r.posts()) + r.advance(3 * time.Second) + if len(r.posts()) != n { + t.Fatalf("a tick with no new event posted (%d -> %d)", n, len(r.posts())) + } +} + +func TestBatchesAtMostOneHundredEventsPerCall(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + for i := 0; i < 250; i++ { + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "line"}) + } + r.tick() + posts := r.posts() + if len(posts) != 3 { + t.Fatalf("%d posts, want 3 (100+100+50)", len(posts)) + } + total := 0 + for _, q := range posts { + n := len(eventsOf(q)) + if n > 100 || n == 0 { + t.Fatalf("a post carried %d events", n) + } + total += n + } + if total != 250 || len(r.hq.Transcript(item)) != 250 { + t.Fatalf("posted %d, stored %d, want 250", total, len(r.hq.Transcript(item))) + } +} + +func TestTruncatesTextAtEightKiBWithAMarker(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + long := strings.Repeat("é", 10000) // 20000 bytes, two per rune: a cut must not split one + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: long}) + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "short and whole"}) + r.tick() + got := r.hq.Transcript(item) + if len(got) != 2 { + t.Fatalf("stored %d events: %+v", len(got), got) + } + cut := got[0].Text + if len(cut) > officers.MaxTranscriptText { + t.Fatalf("text is %d bytes, over %d", len(cut), officers.MaxTranscriptText) + } + i := strings.LastIndex(cut, "… [truncated ") + if i < 0 { + t.Fatalf("no truncation marker in a %d-byte text", len(cut)) + } + kept := cut[:i] + marker := "… [truncated " + itoa(len(long)-len(kept)) + " bytes]" + if !strings.HasSuffix(cut, marker) { + t.Fatalf("text ends %q, want the marker %q", cut[len(cut)-40:], marker) + } + if !strings.HasPrefix(long, kept) || len(kept) < 8000 { + t.Fatalf("kept %d bytes, not a prefix of the text or far too short", len(kept)) + } + if got[1].Text != "short and whole" { + t.Fatalf("a short text was changed: %q", got[1].Text) + } +} + +func itoa(n int) string { b, _ := json.Marshal(n); return string(b) } + +func TestRedactsASecretInAToolResultBeforePosting(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + // Built at runtime: a token-shaped literal in the source trips the leak gate. + secret := "ghp_" + strings.Repeat("A1b2", 9) + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "t1", OK: true, + Output: "fetched the release notes\nGITHUB_TOKEN=" + secret + "\nall 12 checks passed"}) + // The officer's own token, echoed by a tool, is never sent either — even + // in a line no secret SHAPE matches (only the scrub can know it). + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "t2", OK: true, + Output: "the value was " + officerToken + " today"}) + r.tick() + + got := r.hq.Transcript(item) + if len(got) != 2 { + t.Fatalf("stored %d events: %+v", len(got), got) + } + // POSITIVE CONTROL: the ordinary lines pass unchanged, so a filter that + // withheld everything would fail here. + if !strings.Contains(got[0].Text, "fetched the release notes") || !strings.Contains(got[0].Text, "all 12 checks passed") { + t.Fatalf("the clean lines did not pass unchanged: %q", got[0].Text) + } + if !strings.Contains(got[0].Text, "withheld") { + t.Fatalf("the secret line was not marked withheld: %q", got[0].Text) + } + for _, q := range r.posts() { + b, _ := json.Marshal(q.Body) + if strings.Contains(string(b), secret) || strings.Contains(string(b), strings.Repeat("A1b2", 9)) { + t.Fatalf("the secret reached HQ: %s", b) + } + if strings.Contains(string(b), officerToken) { + t.Fatalf("the officer token reached HQ in a body: %s", b) + } + } + if strings.Contains(r.log.String(), secret) || strings.Contains(r.log.String(), "fetched the release notes") { + t.Fatalf("transcript text was logged: %s", r.log.String()) + } +} + +func TestStopsForAnItemOnNotClaimed(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "one"}) + r.tick() + if len(r.hq.Transcript(item)) != 1 { + t.Fatalf("first event not stored") + } + r.hq.Complete(item) // the item is no longer claimed by the instance + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "two"}) + r.advance(3 * time.Second) + refused := len(r.posts()) + if refused != 2 { + t.Fatalf("%d posts, want 2 (one stored, one answered not_claimed)", refused) + } + for i := 0; i < 5; i++ { + r.src.add(at(3), cctranscript.Event{Kind: cctranscript.KindText, Text: "more"}) + r.advance(time.Minute) + } + if len(r.posts()) != refused { + t.Fatalf("kept posting after not_claimed: %d posts", len(r.posts())) + } + if len(r.fwd.Active()) != 0 { + t.Fatalf("still forwarding: %v", r.fwd.Active()) + } + // A pass that still lists the item as held (HQ's list lags) does not + // restart it: the stop holds until a new claim. + r.fwd.Sync(r.o, []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: instance, LeaseExpiresAt: ptr(r.now.Add(time.Minute))}}, r.now) + r.advance(time.Minute) + if len(r.posts()) != refused { + t.Fatalf("a pass restarted a stopped item: %d posts", len(r.posts())) + } + + // The next claim starts again. + next := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(next)) + r.src.add(r.now.Add(time.Second), cctranscript.Event{Kind: cctranscript.KindText, Text: "the next item"}) + r.advance(3 * time.Second) + if got := r.hq.Transcript(next); len(got) != 1 || got[0].Text != "the next item" { + t.Fatalf("the next claim's transcript = %+v", got) + } +} + +func ptr(t time.Time) *time.Time { return &t } + +func TestResumesAfterARestartWithoutDuplicates(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + for i := 0; i < 3; i++ { + r.src.add(at(1+i), cctranscript.Event{Kind: cctranscript.KindText, Text: "before restart " + itoa(i)}) + } + r.tick() + if len(r.hq.Transcript(item)) != 3 { + t.Fatalf("stored %d before the restart", len(r.hq.Transcript(item))) + } + + r.restart() // muster restarts: nothing in memory + // Written while muster was down. + r.src.add(at(5), cctranscript.Event{Kind: cctranscript.KindText, Text: "during the restart"}) + // The wake loop's first pass sees the instance still holding the item. + r.fwd.Sync(r.o, []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: instance, LeaseExpiresAt: ptr(r.now.Add(5 * time.Minute))}}, r.now) + r.tick() + r.src.add(at(6), cctranscript.Event{Kind: cctranscript.KindText, Text: "after restart"}) + r.advance(3 * time.Second) + + got := r.hq.Transcript(item) + var texts []string + seen := map[int64]bool{} + for i, e := range got { + if seen[e.Seq] { + t.Fatalf("seq %d stored twice", e.Seq) + } + seen[e.Seq] = true + if i > 0 && e.Seq <= got[i-1].Seq { + t.Fatalf("seq went backwards") + } + texts = append(texts, e.Text) + } + want := "before restart 0|before restart 1|before restart 2|during the restart|after restart" + if strings.Join(texts, "|") != want { + t.Fatalf("stored %q, want %q", strings.Join(texts, "|"), want) + } + // The restarted forwarder learned HQ's last_seq (an empty post) and + // sent only what HQ did not hold: nothing stored is re-sent. + if len(r.probes()) < 2 { + t.Fatalf("%d probes, want one per forwarder start", len(r.probes())) + } + sent := 0 + for _, q := range r.posts() { + for _, e := range eventsOf(q) { + if m, _ := e.(map[string]any); m["text"] == "before restart 0" { + sent++ + } + } + } + if sent != 1 { + t.Fatalf("an event HQ held was sent %d times", sent) + } +} + +func TestANewSessionKeepsSeqIncreasing(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + for i := 0; i < 5; i++ { + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "old session"}) + } + r.tick() + // The CLI starts a new conversation: a new file, offsets from zero. + r.src.newSession("99999999-2222-3333-4444-555555555555") + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "new session"}) + r.advance(3 * time.Second) + got := r.hq.Transcript(item) + if len(got) != 6 || got[5].Text != "new session" { + t.Fatalf("stored %+v", got) + } +} + +func TestADisabledOfficerSendsNothing(t *testing.T) { + r := newRig(t, false, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.fwd.Sync(r.o, []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: instance, LeaseExpiresAt: ptr(r.now.Add(time.Minute))}}, r.now) + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "x"}) + r.tick() + r.advance(time.Minute) + if n := len(r.posts()); n != 0 { + t.Fatalf("a disabled officer posted %d times", n) + } + if r.src.reads != 0 { + t.Fatalf("a disabled officer's transcript was read %d times", r.src.reads) + } + if r.o.TranscriptEnabled() { + t.Fatal("the rig's officer reads as enabled") + } + // POSITIVE CONTROL: the same steps on an enabled officer do post. + on := newRig(t, true, nil) + item = on.claim() + on.fwd.Claimed(on.o, instance, officers.ItemID(item)) + on.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "x"}) + on.tick() + if len(on.posts()) != 1 { + t.Fatalf("the enabled control posted %d times", len(on.posts())) + } +} + +// droppedBuffer is muster_officer_transcript_events_total{result="dropped_buffer"}. +func droppedBuffer(t *testing.T) float64 { t.Helper(); return counter(t, "dropped_buffer") } + +// counter is muster_officer_transcript_events_total{result=result}. +func counter(t *testing.T, result string) float64 { + t.Helper() + fams, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatal(err) + } + for _, f := range fams { + if f.GetName() != "muster_officer_transcript_events_total" { + continue + } + for _, m := range f.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == "result" && l.GetValue() == result { + return m.GetCounter().GetValue() + } + } + } + } + return 0 +} + +func TestTheBufferIsCappedAndADropIsCounted(t *testing.T) { + r := newRig(t, true, func(c *transcript.Config) { c.MaxBuffer = 10 }) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + before := droppedBuffer(t) + r.hq.SetTranscriptStatus(http.StatusServiceUnavailable) + for i := 0; i < 25; i++ { + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "e" + itoa(i)}) + } + r.tick() + if d := droppedBuffer(t) - before; d != 15 { + t.Fatalf("dropped_buffer moved by %v, want 15 (25 read, 10 kept)", d) + } + if !strings.Contains(r.log.String(), "dropped") { + t.Fatalf("a drop was not logged: %q", r.log.String()) + } + // Backing off: an immediate tick does not post again. + failed := len(r.posts()) + len(r.probes()) + if failed == 0 { + t.Fatal("nothing reached HQ: the backoff below would be vacuous") + } + r.now = r.now.Add(10 * time.Millisecond) + r.tick() + if n := len(r.posts()) + len(r.probes()); n != failed { + t.Fatalf("posted again inside the backoff (%d -> %d)", failed, n) + } + r.hq.SetTranscriptStatus(0) + r.advance(5 * time.Minute) + got := r.hq.Transcript(item) + if len(got) != 10 || got[0].Text != "e15" || got[9].Text != "e24" { + t.Fatalf("after recovery stored %d events, first %+v: want the newest 10 (e15..e24)", len(got), got) + } +} + +func TestRetriesAreCappedThenTheBatchIsDropped(t *testing.T) { + r := newRig(t, true, func(c *transcript.Config) { c.MaxRetries = 2 }) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.tick() // the probe, while HQ answers + r.hq.SetTranscriptStatus(http.StatusServiceUnavailable) + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "lost"}) + r.tick() + for i := 0; i < 6; i++ { + r.advance(5 * time.Minute) + } + if n := len(r.posts()); n != 3 { + t.Fatalf("%d attempts, want 3 (1 + MaxRetries 2)", n) + } + r.hq.SetTranscriptStatus(0) + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "kept"}) + r.advance(5 * time.Minute) + if got := r.hq.Transcript(item); len(got) != 1 || got[0].Text != "kept" { + t.Fatalf("stored %+v, want only the event after the dropped batch (log: %s)", got, r.log) + } +} + +func TestAPassStartsAHeldItemAndStopsOneNoLongerHeld(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + held := []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: instance, LeaseExpiresAt: ptr(r.now.Add(time.Minute))}} + // Another officer's instance, and an expired lease: neither is started. + other := []officers.Item{ + {ID: "77", Status: officers.StatusClaimed, ClaimedBy: officers.InstanceName("other:thing", 1), LeaseExpiresAt: ptr(r.now.Add(time.Minute))}, + {ID: "78", Status: officers.StatusClaimed, ClaimedBy: officers.InstanceName(officerID, 1), LeaseExpiresAt: ptr(r.now.Add(-time.Minute))}, + } + r.fwd.Sync(r.o, other, r.now) + if len(r.fwd.Active()) != 0 { + t.Fatalf("started for %v", r.fwd.Active()) + } + r.fwd.Sync(r.o, held, r.now) + if got := r.fwd.Active(); got[instance] != officers.ItemID(item) { + t.Fatalf("active = %v", got) + } + r.now = r.now.Add(time.Second) + r.fwd.Sync(r.o, nil, r.now) + if len(r.fwd.Active()) != 0 { + t.Fatalf("still active after the item left the held list: %v", r.fwd.Active()) + } + // A claim made AFTER a pass's read is not undone by that pass. + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.fwd.Sync(r.o, nil, r.now.Add(-time.Second)) + if len(r.fwd.Active()) != 1 { + t.Fatalf("a stale pass stopped a fresh claim") + } +} + +// TestALineWithManyEventsPostsEveryOne: one user record carrying 40 parallel +// tool results is one transcript line with 40 events; every one is posted, +// in strictly increasing seq, none dropped as unsequenced. +func TestALineWithManyEventsPostsEveryOne(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + var evs []cctranscript.Event + for i := 0; i < 40; i++ { + evs = append(evs, cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "t" + itoa(i), OK: true, Output: "result " + itoa(i)}) + } + r.src.add(at(1), evs...) + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "the next line"}) + r.tick() + got := r.hq.Transcript(item) + if len(got) != 41 { + t.Fatalf("stored %d events, want all 41 (log: %s)", len(got), r.log) + } + for i := 0; i < 40; i++ { + if got[i].Text != "result "+itoa(i) { + t.Fatalf("event %d = %q", i, got[i].Text) + } + } + for i := 1; i < len(got); i++ { + if got[i].Seq <= got[i-1].Seq { + t.Fatalf("seq not strictly increasing at %d: %d then %d", i, got[i-1].Seq, got[i].Seq) + } + } + if got[40].Text != "the next line" { + t.Fatalf("the following line: %q", got[40].Text) + } +} + +// TestARestartIntoANewConversationHeals: the one case the file position +// cannot order. A new conversation mid-item is shifted above the old one's +// seqs by the forwarder that saw it; a forwarder restarted INTO that +// conversation has lost the shift, so HQ skips its live events as old — and +// the forwarder re-sequences them above HQ's last_seq instead of losing every +// event until the new file outgrows the old. +func TestARestartIntoANewConversationHeals(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + for i := 0; i < 20; i++ { + r.src.add(at(1), cctranscript.Event{Kind: cctranscript.KindText, Text: "old conversation"}) + } + r.tick() + r.src.newSession("99999999-2222-3333-4444-555555555555") + r.src.add(at(2), cctranscript.Event{Kind: cctranscript.KindText, Text: "new conversation"}) + r.advance(3 * time.Second) + if n := len(r.hq.Transcript(item)); n != 21 { + t.Fatalf("stored %d before the restart, want 21", n) + } + + r.restart() + r.fwd.Sync(r.o, []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: instance, LeaseExpiresAt: ptr(r.now.Add(time.Hour))}}, r.now) + r.tick() // the resume's re-read: all already stored (HQ skips it) + r.src.add(at(3), cctranscript.Event{Kind: cctranscript.KindText, Text: "after the restart"}) + r.advance(3 * time.Second) + r.src.add(at(4), cctranscript.Event{Kind: cctranscript.KindText, Text: "and after that"}) + r.advance(3 * time.Second) + + got := r.hq.Transcript(item) + if len(got) != 23 || got[21].Text != "after the restart" || got[22].Text != "and after that" { + var texts []string + for _, e := range got { + texts = append(texts, e.Text) + } + t.Fatalf("stored %d: %q (log: %s)", len(got), texts, r.log) + } + for i := 1; i < len(got); i++ { + if got[i].Seq <= got[i-1].Seq { + t.Fatalf("seq not increasing at %d", i) + } + } +} diff --git a/internal/officers/transcript/round1_test.go b/internal/officers/transcript/round1_test.go new file mode 100644 index 00000000..d5105418 --- /dev/null +++ b/internal/officers/transcript/round1_test.go @@ -0,0 +1,217 @@ +package transcript_test + +import ( + "net/http" + "strings" + "testing" + "time" + + "github.com/ZacxDev/muster/internal/cctranscript" + "github.com/ZacxDev/muster/internal/officers" + "github.com/ZacxDev/muster/internal/officers/hqtest" +) + +func stored(r *rig, item string) []hqtest.TranscriptEvent { return r.hq.Transcript(item) } + +func texts(evs []hqtest.TranscriptEvent) string { + var out []string + for _, e := range evs { + out = append(out, e.Text) + } + return strings.Join(out, "|") +} + +// addUntimed appends a line whose record carries no timestamp (At 0). +func (s *fakeSource) addUntimed(evs ...cctranscript.Event) { + s.mu.Lock() + defer s.mu.Unlock() + s.end += 137 + s.lines = append(s.lines, cctranscript.Line{Events: evs, End: s.end}) +} + +func text(s string) cctranscript.Event { + return cctranscript.Event{Kind: cctranscript.KindText, Text: s} +} + +func held(item string, inst string, until time.Time) []officers.Item { + return []officers.Item{{ID: officers.ItemID(item), Status: officers.StatusClaimed, ClaimedBy: inst, LeaseExpiresAt: ptr(until)}} +} + +// TestALostAcknowledgementIsNotResentAsNew: HQ stores a batch and then fails +// to answer (a 5xx after its commit). The re-send is answered stored 0, +// last_seq = the batch's end — which is NOT a reason to re-sequence: the +// events are stored exactly once. +func TestALostAcknowledgementIsNotResentAsNew(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.tick() // the probe + posted, skipped := counter(t, "posted"), counter(t, "skipped_by_hq") + r.hq.FailTranscript(http.StatusBadGateway, 1, true) + r.src.add(at(1), text("one"), text("two")) + r.tick() // stored, then 502 + r.advance(5 * time.Minute) // the re-send: stored 0 + r.src.add(at(2), text("three")) + r.advance(5 * time.Minute) + if got := texts(stored(r, item)); got != "one|two|three" { + t.Fatalf("stored %q, want each event once (log: %s)", got, r.log) + } + if d := counter(t, "posted") - posted; d != 1 { + t.Fatalf("posted moved by %v, want 1 (only what HQ stored as new on an answered post)", d) + } + if d := counter(t, "skipped_by_hq") - skipped; d != 2 { + t.Fatalf("skipped_by_hq moved by %v, want 2 (the re-send HQ already held)", d) + } +} + +// TestAResumeWithNothingAtHQForwardsOnlyNewLines: muster resumes an item HQ +// holds no transcript for (first enable, or a restart before the first +// post). The backfill window is the instance's EARLIER work — not this +// item's — and is not sent. +func TestAResumeWithNothingAtHQForwardsOnlyNewLines(t *testing.T) { + r := newRig(t, true, nil) + for i := 0; i < 3; i++ { + r.src.add(r.now.Add(-time.Minute), text("another item's work")) + } + item := r.claim() + r.fwd.Sync(r.o, held(item, instance, r.now.Add(time.Hour)), r.now) + r.tick() + r.src.add(at(1), text("new")) + r.advance(3 * time.Second) + if got := texts(stored(r, item)); got != "new" { + t.Fatalf("stored %q, want only the line written after the resume (log: %s)", got, r.log) + } +} + +// TestResumingAnItemAfterAnotherCompletesStartsLive: an instance holding A +// claims B (forwarding switches to B), completes B, and a pass resumes A. +// What the transcript holds since is B's work: none of it goes to A. +func TestResumingAnItemAfterAnotherCompletesStartsLive(t *testing.T) { + r := newRig(t, true, nil) + a := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(a)) + r.src.add(at(1), text("a1")) + r.tick() + b := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(b)) + r.src.add(at(2), text("b1")) + r.advance(3 * time.Second) + r.fwd.Ended(r.o, instance, officers.ItemID(b)) + r.fwd.Sync(r.o, held(a, instance, r.now.Add(time.Hour)), r.now) + r.advance(3 * time.Second) + r.src.add(at(3), text("a2")) + r.advance(3 * time.Second) + if got := texts(stored(r, a)); got != "a1|a2" { + t.Fatalf("item A stored %q, want a1|a2 (B's work must not land in A) (log: %s)", got, r.log) + } + if got := texts(stored(r, b)); got != "b1" { + t.Fatalf("item B stored %q", got) + } +} + +// TestAFreshClaimSkipsUntimedLinesBeforeIt: a line with no timestamp in the +// first read cannot be shown to follow the claim, so it is not sent. +func TestAFreshClaimSkipsUntimedLinesBeforeIt(t *testing.T) { + r := newRig(t, true, nil) + r.src.addUntimed(text("untimed, before the claim")) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.tick() + r.src.add(at(1), text("after")) + r.advance(3 * time.Second) + if got := texts(stored(r, item)); got != "after" { + t.Fatalf("stored %q", got) + } +} + +// TestAReclaimByAnotherInstanceLandsAboveHQsLastSeq: HQ's last_seq is per +// ITEM. Instance 1 forwards and releases; instance 2 (its own transcript, +// offsets from zero) claims the same item: its events are stored, not +// skipped as old. +func TestAReclaimByAnotherInstanceLandsAboveHQsLastSeq(t *testing.T) { + r := newRig(t, true, nil) + two := officers.InstanceName(officerID, 2) + other := newSource() + other.session = "22222222-2222-3333-4444-555555555555" + r.src.others = map[string]*fakeSource{two: other} + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + // Two lines from instance one; five from instance two, whose first two + // offsets fall at or below HQ's last_seq and the rest above it: a batch + // that STRADDLES HQ's last_seq, which HQ would store only in part. + for i := 0; i < 2; i++ { + r.src.add(at(1), text("instance one")) + } + r.tick() + r.fwd.Ended(r.o, instance, officers.ItemID(item)) + r.hq.SetClaim(item, two, r.now.Add(time.Hour)) + r.fwd.Claimed(r.o, two, officers.ItemID(item)) + for i := 0; i < 5; i++ { + other.add(at(2), text("instance two "+itoa(i))) + } + r.advance(3 * time.Second) + got := stored(r, item) + if len(got) != 7 || got[6].Text != "instance two 4" || got[2].Text != "instance two 0" || got[6].Instance != two { + t.Fatalf("stored %q (log: %s)", texts(got), r.log) + } +} + +// TestATooLargeAnswerDropsTheBatchAndHalvesTheNext: a 413 is not a decision +// about the item: that batch is dropped (counted) and the next is sent at +// half the size; forwarding goes on. +func TestATooLargeAnswerDropsTheBatchAndHalvesTheNext(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + r.tick() // the probe + before := counter(t, "dropped_too_large") + r.hq.FailTranscript(http.StatusRequestEntityTooLarge, 1, false) + for i := 0; i < 250; i++ { + r.src.add(at(1), text("e"+itoa(i))) + } + r.tick() + if d := counter(t, "dropped_too_large") - before; d != 100 { + t.Fatalf("dropped_too_large moved by %v, want 100", d) + } + posts := r.posts() + for _, q := range posts[1:] { + if n := len(eventsOf(q)); n > 50 { + t.Fatalf("a post after the 413 carried %d events, want at most 50", n) + } + } + got := stored(r, item) + if len(got) != 150 || got[0].Text != "e100" { + t.Fatalf("stored %d events after the 413, first %+v", len(got), got) + } + if len(r.fwd.Active()) != 1 { + t.Fatal("a 413 stopped forwarding") + } +} + +// TestATokenCutByACapIsWithheld: ccd caps a long field with "…"; a token cut +// there leaves a PREFIX no full-token pattern matches. The line it ends is +// withheld. An ordinary capped line passes unchanged (positive control). +func TestATokenCutByACapIsWithheld(t *testing.T) { + r := newRig(t, true, nil) + item := r.claim() + r.fwd.Claimed(r.o, instance, officers.ItemID(item)) + cutGitHub := "ghp_" + strings.Repeat("A1b2", 3) // 16 characters: below the full pattern's 20 + r.src.add(at(1), + cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "a", OK: true, Output: "first line\nthe header was " + officerToken[:12] + "…"}, + cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "b", OK: true, Output: "clone with " + cutGitHub + "…"}, + cctranscript.Event{Kind: cctranscript.KindToolResult, ToolID: "c", OK: true, Output: "an ordinary long output, cut…"}) + r.tick() + got := stored(r, item) + if len(got) != 3 { + t.Fatalf("stored %d", len(got)) + } + if strings.Contains(got[0].Text, officerToken[:12]) || !strings.HasPrefix(got[0].Text, "first line\n") { + t.Fatalf("a cut officer token: %q", got[0].Text) + } + if strings.Contains(got[1].Text, cutGitHub) { + t.Fatalf("a cut GitHub token: %q", got[1].Text) + } + if got[2].Text != "an ordinary long output, cut…" { + t.Fatalf("an ordinary capped line was changed: %q", got[2].Text) + } +} diff --git a/internal/officers/wake.go b/internal/officers/wake.go index 22386f35..73eec2da 100644 --- a/internal/officers/wake.go +++ b/internal/officers/wake.go @@ -330,6 +330,8 @@ type Reconciler struct { fleet Fleet log *log.Logger now func() time.Time + // observed is Config.Observed (nil: none). + observed func(o Officer, claimed []Item, at time.Time) mu sync.Mutex state map[string]*officerState @@ -344,6 +346,11 @@ type Config struct { Logger *log.Logger // Now is the clock; nil is time.Now. Now func() time.Time + // Observed, when set, is told every pass's claimed items for o, read no + // earlier than at (the transcript forwarder's Sync). It runs only after + // a SUCCESSFUL read — a pass that saw nothing tells it nothing, so an HQ + // outage never reads as "no item is held". It must not block. + Observed func(o Officer, claimed []Item, at time.Time) } // NewReconciler returns a loop over cfg's officers. @@ -357,7 +364,7 @@ func NewReconciler(cfg Config) (*Reconciler, error) { if cfg.Now == nil { cfg.Now = time.Now } - return &Reconciler{reg: cfg.Registry, hq: cfg.HQ, fleet: cfg.Fleet, log: cfg.Logger, now: cfg.Now, + return &Reconciler{reg: cfg.Registry, hq: cfg.HQ, fleet: cfg.Fleet, log: cfg.Logger, now: cfg.Now, observed: cfg.Observed, state: map[string]*officerState{}, done: make(chan struct{})}, nil } @@ -444,6 +451,9 @@ func (r *Reconciler) Check(ctx context.Context, id string, force bool) PassResul return r.scaleDownForFault(ctx, o, err, refusal) } r.noteErrors(o, obs.Instances) + if r.observed != nil { + r.observed(o, obs.Claimed, now) + } r.mu.Lock() prompted := make(map[int]time.Time, len(s.promptedAt)) for n, at := range s.promptedAt { diff --git a/internal/officership/cilogs.go b/internal/officership/cilogs.go index 3c13edc6..0d6441e9 100644 --- a/internal/officership/cilogs.go +++ b/internal/officership/cilogs.go @@ -340,13 +340,18 @@ var ( const withheld = "[line withheld: it looks like it carries a credential]" // tokenRE are lines carrying a credential by its SHAPE or its framing: a -// GitHub, AWS, Slack, OpenAI-style or JWT token, a PEM header, an +// GitHub, GitLab, AWS (AKIA/ASIA, aws_secret_access_key), Slack, Stripe +// (sk_/rk_ live/test), Google (AIza…), Hugging Face (hf_), npm (npm_), PyPI +// (pypi-), OpenAI/Anthropic-style (sk-, sk-proj-, sk-ant-) or JWT token, a +// kubeconfig's client-key-data / client-certificate-data, a PEM header, an // Authorization header or a Bearer token, a Cookie / Set-Cookie header, a // signed URL's sig= / signature= / token= query parameter, a CLI flag // --password/--token/--secret/--api-key with a value, a registry or database // login's -p VALUE / -pVALUE, and `basic auth: `. var tokenRE = regexp.MustCompile(`(?i)` + - `\bgh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|\bAKIA[0-9A-Z]{16}\b|\bxox[abprs]-[A-Za-z0-9-]{10,}|\bsk-[A-Za-z0-9_-]{20,}|` + + `\bgh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|\b(AKIA|ASIA)[0-9A-Z]{16}\b|\bxox[abprs]-[A-Za-z0-9-]{10,}|\bsk-[A-Za-z0-9_-]{20,}|` + + `\b[sr]k_(live|test)_[A-Za-z0-9]{10,}|\bAIza[0-9A-Za-z_-]{35}|\bglpat-[A-Za-z0-9_-]{20,}|\bhf_[A-Za-z0-9]{30,}|\bnpm_[A-Za-z0-9]{36}\b|` + + `\bpypi-[A-Za-z0-9_-]{50,}|\baws_secret_access_key\b|\b(client-key-data|client-certificate-data)\s*:\s*\S|` + `\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.|-----BEGIN [A-Z ]*(PRIVATE KEY|CERTIFICATE)|\bauthorization:\s*\S|\bbearer\s+[A-Za-z0-9._~+/=-]{12,}|` + `\b(set-)?cookie:\s*\S|[?&](sig|signature|x-amz-signature|x-goog-signature|token|access_token)=[^&\s]{6,}|` + `--(password|passwd|token|secret|api[-_]?key|access[-_]?token|client[-_]?secret)(=|\s+)\S{4,}|` + @@ -421,6 +426,73 @@ func redactText(s string) string { return strings.Join(RedactLines(splitLines(s)), "\n") } +// RedactText is THE CI-log secret filter (RedactLines, unchanged) over any +// multi-line text — an officer's session transcript forwarded to HQ +// (internal/officers/transcript) — split at the same separators as a log +// (\n, \r\n, a lone \r) but KEEPING empty lines, so prose keeps its +// paragraphs. Every line is joined back with \n. +func RedactText(s string) string { + lines := strings.Split(strings.ReplaceAll(strings.ReplaceAll(s, "\r\n", "\n"), "\r", "\n"), "\n") + return strings.Join(RedactLines(lines), "\n") +} + +// cutPrefixRE is a word that STARTS like a credential the shapes above know: +// what is left of one when a cap cut it short of the length its full pattern +// needs. +var cutPrefixRE = regexp.MustCompile(`(?i)^(gh[pousr]_|github_pat_|sk[-_]|rk_(live|test)_|xox[abprs]-|akia|asia|aiza|glpat-|hf_|npm_|pypi-|eyj)`) + +// cutMark is the cap mark a cut field ends in (cctranscript.Cap and every +// capped field it feeds). +const cutMark = "…" + +// RedactCutTail withholds every line in which a word cut by a cap — the word +// right before a cutMark — could be a credential's prefix: it starts like a +// known token shape (cutPrefixRE), or it ends in the first 4 or more bytes of +// one of known (a credential the caller holds and can name). A shape pattern +// needs the whole token; a cap can leave only its start, and that start is +// still the credential's. +func RedactCutTail(s string, known []string) string { + if !strings.Contains(s, cutMark) { + return s + } + lines := strings.Split(s, "\n") + for i, l := range lines { + for rest := l; ; { + j := strings.Index(rest, cutMark) + if j < 0 { + break + } + before := rest[:j] + rest = rest[j+len(cutMark):] + f := strings.Fields(before) + if len(f) == 0 || strings.HasSuffix(before, " ") { + continue + } + if w := f[len(f)-1]; cutPrefixRE.MatchString(w) || endsInPrefixOf(w, known) { + lines[i] = withheld + break + } + } + } + return strings.Join(lines, "\n") +} + +// endsInPrefixOf reports whether w ends in the first 4 or more bytes of one +// of known. +func endsInPrefixOf(w string, known []string) bool { + for _, k := range known { + if len(k) < 8 { + continue + } + for i := 0; len(w)-i >= 4; i++ { + if strings.HasPrefix(k, w[i:]) { + return true + } + } + } + return false +} + // splitLines splits a log at \n, \r\n and a lone \r (a progress line // overwritten in place). func splitLines(s string) []string { diff --git a/internal/officership/redact_shapes_test.go b/internal/officership/redact_shapes_test.go new file mode 100644 index 00000000..97cb3918 --- /dev/null +++ b/internal/officership/redact_shapes_test.go @@ -0,0 +1,78 @@ +package officership + +import ( + "strings" + "testing" +) + +// TestRedactWithholdsMoreCredentialShapes: the provider token shapes the +// filter knows by their PREFIX, each in a line no key=value or env rule would +// catch, so only the shape can withhold it. Every fixture is built at run +// time: a token-shaped literal in the source trips the repository's leak +// gate. +func TestRedactWithholdsMoreCredentialShapes(t *testing.T) { + r := strings.Repeat + for name, secret := range map[string]string{ + "stripe live": "sk_live_" + r("x9Y8", 6), + "stripe test": "sk_test_" + r("x9Y8", 6), + "stripe restrict": "rk_live_" + r("x9Y8", 6), + "google": "AIza" + "Sy" + r("aB3-", 8) + "c", + "gitlab": "glpat-" + r("Ab1_", 5), + "hugging face": "hf_" + r("Ab1c", 8), + "aws session": "ASIA" + r("QWER", 4), + "npm": "npm_" + r("Ab1C", 9), + "pypi": "pypi-" + r("AgEI", 14), + "openai project": "sk-proj-" + r("Ab1c", 6), + "anthropic": "sk-ant-" + r("Ab1c", 6), + "slack": "xoxb-" + r("1234-", 4), + } { + line := "the request went out with " + secret + " attached" + if got := Redact(line); strings.Contains(got, secret) || got != withheld { + t.Errorf("%s: %q -> %q, want it withheld", name, line, got) + } + } + for name, line := range map[string]string{ + "kubeconfig client key": " client-key-data: " + r("TFMw", 12), + "kubeconfig client cert": " client-certificate-data: " + r("TFMw", 12), + "aws secret key": "aws_secret_access_key = " + r("Ab1/", 10), + } { + if got := Redact(line); got != withheld { + t.Errorf("%s: %q -> %q, want it withheld", name, line, got) + } + } + // POSITIVE CONTROLS: ordinary lines naming the same words pass unchanged. + for _, l := range []string{ + "the token was refreshed at 12:00", + "rotate the API key monthly", + "read /etc/ssl/private/server.key", + "key: value pairs are parsed in order", + "client-key-data is set in the kubeconfig", + "npm_config_cache points at the shared cache", + "hf_hub_download fetched the weights", + "AIzawa is not a key", + "see sk_live_ prefixes in the docs", + } { + if got := Redact(l); got != l { + t.Errorf("an ordinary line was changed: %q -> %q", l, got) + } + } +} + +// TestRedactCutTailWithholdsATokenPrefix: a field capped with "…" can end in +// the PREFIX of a token, shorter than any full-token pattern. That last line +// is withheld; an ordinary capped line is not. +func TestRedactCutTailWithholdsATokenPrefix(t *testing.T) { + cut := "ghp_" + strings.Repeat("A1b2", 3) + if got := RedactCutTail("ok\nclone with "+cut+"…", nil); strings.Contains(got, cut) || got != "ok\n"+withheld { + t.Fatalf("a cut token prefix: %q", got) + } + tok := "opaque-" + strings.Repeat("9z", 10) + if got := RedactCutTail("the value was "+tok[:9]+"…", []string{tok}); strings.Contains(got, tok[:9]) { + t.Fatalf("a cut known token: %q", got) + } + for _, l := range []string{"an ordinary long output, cut…", "no cut at all ghp_", "skipped … in the middle"} { + if got := RedactCutTail(l, []string{tok}); got != l { + t.Errorf("an ordinary line was changed: %q -> %q", l, got) + } + } +}