perf(dash): normalize the request body once for lifecycle signals

cmd/context-guru-proxy/lifecycleflag_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"regexp"
+	"testing"
+	"time"
+)
+
+// -h prints each flag's default as the process computed it from the environment, so this
+// spans flag definition + envBool, which a helper-level test would not.
+func TestLifecycleSignalsFlagDefault(t *testing.T) {
+	re := regexp.MustCompile(`(?s)-dashboard-lifecycle-signals.*?\n\s*(.*?)\n  -`)
+	for _, tc := range []struct {
+		env  []string
+		want bool
+	}{{nil, true}, {[]string{"DASHBOARD_LIFECYCLE_SIGNALS=0"}, false}, {[]string{"DASHBOARD_LIFECYCLE_SIGNALS=false"}, false}} {
+		out, _, _ := runMain(t, 20*time.Second, append([]string{"CG_TEST_MAIN_ARGS=-h"}, tc.env...)...)
+		m := re.FindStringSubmatch(out)
+		if m == nil {
+			t.Fatalf("flag not in usage output:\n%s", out)
+		}
+		// Go prints "(default true)" only for a non-zero default.
+		if got := regexp.MustCompile(`\(default true\)`).MatchString(m[0]); got != tc.want {
+			t.Errorf("env %v: default on = %v, want %v\n%s", tc.env, got, tc.want, m[0])
+		}
+	}
+}

dash/lifecycle_retention_test.go

@@ -0,0 +1,59 @@
+package dash
+
+import (
+	"path/filepath"
+	"testing"
+)
+
+func lifeRows(t *testing.T, db *DB) int64 { return countRows(t, db, "request_lifecycle") }
+
+// A lifecycle row dies with its request (ON DELETE CASCADE), and the per-tenant quota counts
+// requests rows only: lifecycle rows neither count toward it nor survive the trim.
+func TestLifecycleRowsFollowRequestsAndQuotaCountsRequestsOnly(t *testing.T) {
+	rec, err := NewRecorder(Options{DBPath: filepath.Join(t.TempDir(), "d.db"), MaxRowsPerTenant: 10})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rec.Close()
+	seedSessions(t, rec.db, "heavy", 20, 2) // 40 requests
+	if _, err := rec.db.sql.Exec(`INSERT INTO request_lifecycle(request_id) SELECT id FROM requests`); err != nil {
+		t.Fatal(err)
+	}
+	counts, err := rec.db.TenantRowCounts()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if counts["heavy"] != 40 {
+		t.Fatalf("quota counted %d rows, want the 40 requests (lifecycle rows must not count)", counts["heavy"])
+	}
+	rec.enforceQuotas()
+	reqs, life := countRows(t, rec.db, "requests"), lifeRows(t, rec.db)
+	if reqs > 10 || life != reqs {
+		t.Errorf("after trim: %d requests, %d lifecycle rows; want lifecycle == requests <= 10", reqs, life)
+	}
+}
+
+// An existing database written before the table existed gets it from the additive DDL on open.
+func TestLifecycleTableAddedToExistingDatabase(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "d.db")
+	db, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	seedSessions(t, db, "t", 1, 1)
+	if _, err := db.sql.Exec(`DROP TABLE request_lifecycle`); err != nil {
+		t.Fatal(err)
+	}
+	db.Close()
+	db, err = Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if n := lifeRows(t, db); n != 0 {
+		t.Errorf("new table has %d rows", n)
+	}
+	if n := countRows(t, db, "requests"); n != 1 {
+		t.Errorf("upgrade lost request rows: %d", n)
+	}
+}

internal/waitcause/waitcause.go

@@ -224,10 +224,21 @@ func (t *Tracker) Observe(tenant, session, threadID string, provider bschemas.Mo
 	if t == nil {
 		return Observation{}
 	}
-	return t.observe(tenant, session, threadID, provider, body)
+	return t.observe(tenant, session, threadID, body, func() []bschemas.ChatMessage { return apply.NormalizeMessages(provider, body) })
 }
 
-func (t *Tracker) observe(tenant, session, threadID string, provider bschemas.ModelProvider, body []byte) Observation {
+// ObserveWith is Observe with the normalized messages supplied by the caller (norm is called
+// at most once, and not at all for a previous_response_id body), so a caller that also needs
+// the normalization can share one pass over the body.
+func (t *Tracker) ObserveWith(tenant, session, threadID string, body []byte, norm func() []bschemas.ChatMessage) (obs Observation) {
+	defer func() { recover() }() //nolint:errcheck // fail open; see package doc
+	if t == nil {
+		return Observation{}
+	}
+	return t.observe(tenant, session, threadID, body, norm)
+}
+
+func (t *Tracker) observe(tenant, session, threadID string, body []byte, norm func() []bschemas.ChatMessage) Observation {
 	if gjson.GetBytes(body, "previous_response_id").String() != "" {
 		// A Responses request continuing SERVER-HELD history: `input` here is only the
 		// newest turn, not the full conversation. apply.NormalizeMessages does not special-
@@ -240,7 +251,7 @@ func (t *Tracker) observe(tenant, session, threadID string, provider bschemas.Mo
 		// whole conversation" — not a second message parser.
 		return Observation{}
 	}
-	entries := apply.NormalizeMessages(provider, body)
+	entries := norm()
 	n := len(entries)
 	if n == 0 {
 		return Observation{}

proxy/dashcapture.go

@@ -417,16 +417,25 @@ func (c *capture) finish(usage Usage, usageOK bool, captureContent bool, content
 	toolUseNames, stopReason := usage.ToolUseNames, usage.StopReason
 	failedCall := status != 0 && status/100 != 2
 	provider := bschemas.ModelProvider(c.provider)
+	// Both closures run on the writer goroutine, one after the other: normalize once, lazily.
+	var normMsgs []bschemas.ChatMessage
+	var normDone bool
+	norm := func() []bschemas.ChatMessage {
+		if !normDone {
+			normMsgs, normDone = apply.NormalizeMessages(provider, waitBody), true
+		}
+		return normMsgs
+	}
 	if c.rec.LifecycleOn() {
 		ts, endTS, cacheW, cacheW1h := e.TS, time.Now().UnixMilli(), usage.CacheWrite-usage.CacheWrite1h, usage.CacheWrite1h
 		rec := c.rec
 		e.LifecycleFn = func() *dash.LifecycleRow {
 			return rec.ObserveLifecycle(tenantID, sessionID, threadID, ts, endTS,
-				scanLifecycle(provider, waitBody), cacheW, cacheW1h)
+				scanLifecycle(norm, waitBody), cacheW, cacheW1h)
 		}
 	}
 	e.WaitCause = func() dash.WaitCauseResult {
-		obs := waits.Observe(tenantID, sessionID, threadID, provider, waitBody)
+		obs := waits.ObserveWith(tenantID, sessionID, threadID, waitBody, norm)
 		waits.RecordOutcome(tenantID, sessionID, threadID, obs.MsgCount, status)
 		waitingFor, waitTool := waitcause.ClassifyWaitingFor(
 			toolUseNames, stopReason, failedCall, usageOK, obs.OpenBackground, obs.OpenTeammate)

proxy/lifecycle.go

@@ -5,7 +5,6 @@ import (
 
 	bschemas "github.com/maximhq/bifrost/core/schemas"
 
-	"github.com/rossoctl/context-guru/apply"
 	"github.com/rossoctl/context-guru/dash"
 	"github.com/rossoctl/context-guru/schema"
 )
@@ -15,12 +14,14 @@ import (
 // assistant turn. Count and bytes come from there; is_error is the one thing the normalizer
 // drops, so it is read straight off the last Anthropic user message. Runs on the dashboard
 // writer goroutine, never the request path.
-func scanLifecycle(provider bschemas.ModelProvider, body []byte) (q dash.LifecycleReq) {
+// norm yields the normalized conversation (shared with waitcause so the body is normalized
+// once per request).
+func scanLifecycle(norm func() []bschemas.ChatMessage, body []byte) (q dash.LifecycleReq) {
 	defer func() { recover() }() //nolint:errcheck // fail open: a body it cannot read yields no facts
 	if gjson.GetBytes(body, "previous_response_id").String() != "" {
 		return q // server-held history: `input` is only the newest turn
 	}
-	msgs := apply.NormalizeMessages(provider, body)
+	msgs := norm()
 	last := -1
 	for i, m := range msgs {
 		if m.Role == bschemas.ChatMessageRoleAssistant {

proxy/lifecycle_bench_test.go

@@ -0,0 +1,53 @@
+package proxy
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	bschemas "github.com/maximhq/bifrost/core/schemas"
+
+	"github.com/rossoctl/context-guru/apply"
+	"github.com/rossoctl/context-guru/internal/waitcause"
+)
+
+// bigAgentBody is an Anthropic transcript of about 300K tokens (~1.2 MB): 300 tool turns.
+func bigAgentBody() []byte {
+	var b strings.Builder
+	b.WriteString(`{"model":"m","messages":[{"role":"user","content":"start"}`)
+	blob := strings.Repeat("lorem ipsum dolor sit amet ", 150) // ~4 KB
+	for i := 0; i < 300; i++ {
+		fmt.Fprintf(&b, `,{"role":"assistant","content":[{"type":"tool_use","id":"t%d","name":"Bash","input":{"command":"ls"}}]}`, i)
+		fmt.Fprintf(&b, `,{"role":"user","content":[{"type":"tool_result","tool_use_id":"t%d","content":%q}]}`, i, blob)
+	}
+	b.WriteString(`]}`)
+	return []byte(b.String())
+}
+
+// The two per-request body scans the dashboard writer goroutine runs: wait-cause then
+// lifecycle. Twice = each normalizes (before this change); Once = they share one pass.
+func BenchmarkWriterBodyScansNormalizeTwice(b *testing.B) {
+	body := bigAgentBody()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		waitcause.New().Observe("t", "s", "th", bschemas.Anthropic, body)
+		scanLifecycleOld(bschemas.Anthropic, body)
+	}
+}
+
+func BenchmarkWriterBodyScansNormalizeOnce(b *testing.B) {
+	body := bigAgentBody()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		var msgs []bschemas.ChatMessage
+		var done bool
+		norm := func() []bschemas.ChatMessage {
+			if !done {
+				msgs, done = apply.NormalizeMessages(bschemas.Anthropic, body), true
+			}
+			return msgs
+		}
+		waitcause.New().ObserveWith("t", "s", "th", body, norm)
+		scanLifecycle(norm, body)
+	}
+}

proxy/lifecycle_equiv_test.go

@@ -0,0 +1,84 @@
+package proxy
+
+import (
+	"reflect"
+	"testing"
+
+	bschemas "github.com/maximhq/bifrost/core/schemas"
+	"github.com/tidwall/gjson"
+
+	"github.com/rossoctl/context-guru/apply"
+	"github.com/rossoctl/context-guru/dash"
+	"github.com/rossoctl/context-guru/internal/waitcause"
+	"github.com/rossoctl/context-guru/schema"
+)
+
+// scanLifecycleOld is scanLifecycle as it was before the normalization was shared: it
+// normalized the body itself. Kept here as the reference the new path must equal.
+func scanLifecycleOld(provider bschemas.ModelProvider, body []byte) (q dash.LifecycleReq) {
+	if gjson.GetBytes(body, "previous_response_id").String() != "" {
+		return q
+	}
+	msgs := apply.NormalizeMessages(provider, body)
+	last := -1
+	for i, m := range msgs {
+		if m.Role == bschemas.ChatMessageRoleAssistant {
+			last = i
+		}
+	}
+	for _, m := range msgs[last+1:] {
+		if m.ChatToolMessage != nil {
+			q.ResultCount++
+			q.ResultBytes += len(schema.MessageText(m))
+		}
+	}
+	gjson.GetBytes(body, "messages.@reverse.0.content").ForEach(func(_, b gjson.Result) bool {
+		if b.IsObject() && b.Get("type").String() == "tool_result" && b.Get("is_error").Bool() {
+			q.ResultError = true
+		}
+		return !q.ResultError
+	})
+	return q
+}
+
+// Sharing one normalization between waitcause and the lifecycle scan changes nothing: the
+// lifecycle facts and the wait-cause observation equal what each computed on its own.
+func TestSharedNormalizationIsByteIdentical(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		provider bschemas.ModelProvider
+		body     string
+	}{
+		{"anthropic", bschemas.Anthropic, `{"messages":[{"role":"user","content":"go"},
+			{"role":"assistant","content":[{"type":"tool_use","id":"1","name":"Bash","input":{}},{"type":"tool_use","id":"2","name":"Read","input":{}}]},
+			{"role":"user","content":[{"type":"tool_result","tool_use_id":"1","content":"0123456789","is_error":true},{"type":"tool_result","tool_use_id":"2","content":[{"type":"text","text":"abc"}]},{"type":"text","text":"more"}]}]}`},
+		{"openai chat", bschemas.OpenAI, `{"model":"gpt","messages":[{"role":"user","content":"go"},
+			{"role":"assistant","tool_calls":[{"id":"c1","type":"function","function":{"name":"sh","arguments":"{}"}}]},
+			{"role":"tool","tool_call_id":"c1","content":"hello world"}]}`},
+		{"responses", bschemas.OpenAI, `{"model":"gpt","input":[{"role":"user","content":"go"},
+			{"type":"function_call","call_id":"c1","name":"sh","arguments":"{}"},
+			{"type":"function_call_output","call_id":"c1","output":"result text"}]}`},
+		{"previous_response_id", bschemas.OpenAI, `{"model":"gpt","previous_response_id":"resp_1","input":[
+			{"type":"function_call_output","call_id":"c1","output":"result text"}]}`},
+		{"no tool results", bschemas.Anthropic, `{"messages":[{"role":"user","content":"hi"}]}`},
+		{"garbage", bschemas.Anthropic, `not json`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			body := []byte(tc.body)
+			norm := func() []bschemas.ChatMessage { return apply.NormalizeMessages(tc.provider, body) }
+			if got, want := scanLifecycle(norm, body), scanLifecycleOld(tc.provider, body); got != want {
+				t.Errorf("lifecycle: shared %+v, old %+v", got, want)
+			}
+			calls := 0
+			shared := func() []bschemas.ChatMessage { calls++; return norm() }
+			got := waitcause.New().ObserveWith("t", "s", "th", body, shared)
+			want := waitcause.New().Observe("t", "s", "th", tc.provider, body)
+			if !reflect.DeepEqual(got, want) {
+				t.Errorf("waitcause: shared %+v, old %+v", got, want)
+			}
+			if calls > 1 {
+				t.Errorf("norm called %d times by one Observe", calls)
+			}
+		})
+	}
+}

proxy/lifecycle_scan_test.go

@@ -4,26 +4,33 @@ import (
 	"testing"
 
 	bschemas "github.com/maximhq/bifrost/core/schemas"
+
+	"github.com/rossoctl/context-guru/apply"
+	"github.com/rossoctl/context-guru/dash"
 )
 
+func scanLifecycleBody(provider bschemas.ModelProvider, body []byte) dash.LifecycleReq {
+	return scanLifecycle(func() []bschemas.ChatMessage { return apply.NormalizeMessages(provider, body) }, body)
+}
+
 func TestScanLifecycle(t *testing.T) {
-	q := scanLifecycle(bschemas.Anthropic, []byte(`{"messages":[
+	q := scanLifecycleBody(bschemas.Anthropic, []byte(`{"messages":[
 	{"role":"user","content":"fix it"},
 	{"role":"assistant","content":[{"type":"tool_use","id":"1","name":"Bash","input":{"c":"SECRET"}},{"type":"tool_use","id":"2","name":"Read","input":{}}]},
 	{"role":"user","content":[{"type":"tool_result","tool_use_id":"1","content":"0123456789","is_error":true},{"type":"tool_result","tool_use_id":"2","content":"abc"},{"type":"text","text":"also this"}]}]}`))
 	if q.ResultCount != 2 || q.ResultBytes != 13 || !q.ResultError {
 		t.Errorf("%+v", q)
 	}
 	// Results answered earlier in the transcript are not counted again.
-	q = scanLifecycle(bschemas.Anthropic, []byte(`{"messages":[
+	q = scanLifecycleBody(bschemas.Anthropic, []byte(`{"messages":[
 	{"role":"assistant","content":[{"type":"tool_use","id":"1","name":"Bash","input":{}}]},
 	{"role":"user","content":[{"type":"tool_result","tool_use_id":"1","content":"x","is_error":true}]},
 	{"role":"assistant","content":[{"type":"text","text":"done"}]},
 	{"role":"user","content":"thanks"}]}`))
 	if q.ResultCount != 0 || q.ResultError {
 		t.Errorf("stale results counted: %+v", q)
 	}
-	if q = scanLifecycle(bschemas.Anthropic, []byte(`not json`)); q.ResultCount != 0 {
+	if q = scanLifecycleBody(bschemas.Anthropic, []byte(`not json`)); q.ResultCount != 0 {
 		t.Errorf("garbage: %+v", q)
 	}
 }