Skip to content

Commit 9f73370

Browse files
authored
Show startup progress before the agent's first message (#1692)
## Summary After an investigation starts, the Activity pane showed only "Starting investigation…" and a timer for the 20+ seconds until the agent's first message. Three real signals exist before that message and none was shown. This PR reports the two Radar can observe honestly, with no timer-based labels: - **Connected to Radar's tools** the moment the agent completes the MCP handshake on the private investigation mount (agent-agnostic). - **Claude Code ready · \<model\> · \<n\> Radar tools** from Claude Code's stream-json `system`/`init` line. A non-connected MCP server becomes a visible warning in the Activity pane instead of a silent tool-less investigation. Cursor's init names the model only. Also on the same Activity rows: sub-2s tool durations are hidden on the collapsed row (whole seconds above that), the exact milliseconds stay in the expanded result header, and a failed tool row shows the producer's own failure reason inline. ## What changed - `internal/investigationrefs`: a scope now carries a `Connected()` channel; `Registry.MarkConnected(scope)` closes it once for a live scope. This is the seam between the MCP mount and the runner, so `internal/mcp` still does not import `internal/ai`. - `internal/mcp/server.go`: the investigation middleware marks the scope connected after `initialize` or `tools/list` returns **without error** — a rejected handshake never reports as connected. - `internal/ai/diagnoser.go`: `StreamEvent` gains optional `Model`, `ToolCount`, `MCPServers` (phase events were already persisted in the run log; no migration). The runner watches the scope and emits one `phase: connected` per turn, serialized with the parser's events and always before the turn returns — including when the handshake and the stream's end coincide. The watcher stays alive until process exit. `parseStream` handles `system`/`init` and emits `phase: ready` with the model, the `mcp__radar__` tool count and every MCP server status. Cursor emits the model. - `web/src/api/diagnose.ts`, `parts.tsx`, `InvestigationView.tsx`: the `phase` event folds into `turn.startup` (furthest phase wins, since the handshake and init are reported by different goroutines). The pending status line shows the startup label only while the turn is running with no items, so it never renders on a replayed finished turn and never becomes a transcript item. The MCP warning is attributed to startup and survives replay, since it is a fact of the run; its copy is conditional on the server's state and reads in past tense on a replayed run. - Tool rows: `toolDurationLabel` (hidden under 2s, whole seconds above), `toolErrorReason` (JSON `error` field or first non-empty line of the producer text, with Radar's retry hints after the em dash dropped), `middleTruncate` so a long reason keeps its identifier tail, arguments shrink before the reason on the collapsed row, `PayloadBlock` header detail for the exact ms. - Running status: the elapsed counter is a row-level "Ns elapsed" figure set apart from the label, so "Connected to Radar's tools" no longer reads as if the handshake took that long. ## Testing - `make tsc`, `make build`, `go test ./...` (both modules), `git diff --check` — pass. New Go tests: the Claude init parser, the handshake-emitted-once runner path including the short-turn race, the registry handshake, the middleware's success-only marking, and Cursor's init model. - `packages/k8s-ui`: 3549 passed. `web`: 1023 passed across 87 files, including the startup-signal merge (furthest phase wins, init facts kept), the bare `ready` label with no init facts, replay suppression, and the tool-row duration/reason rendering. - Visual test against a live GKE cluster (Deployment `dev/cloud-provisioner`, real Claude Code run). Observed from the click: 0.4s "Claude Code starting…", 0.9s "Connected to Radar's tools", 5.0s "Claude Code ready · claude-sonnet-5 · 23 Radar tools", 23.2s first tool row. Reloading a finished run shows no startup label. A follow-up turn on the same run repeated the progression without special-casing. - Tool rows on the same cluster: a call for a nonexistent Secret rendered `resource not found: secret "…" not found` inline after the arguments. Every live call finished under 2s, so the slow row was verified against a saved run whose `search` took 3754ms — the collapsed row reads `4s`, the expanded header `ORIGINAL RESULT · 3754ms`. ## Scope and tradeoffs - Codex (`codex exec --json`) only emits `thread.started`; it carries no model or MCP data, so it gets the agent-agnostic "connected" signal only. - The startup warning is hand-rolled rather than `AlertBanner`. That component is the drawer-scale renderer surface (`mb-4 p-3`, 14px title plus message); the Activity pane is an 11px transcript and every warning in `parts.tsx` (consent, apply, truncation notes) is hand-rolled at that scale, so the banner would be out of register there. - Radar Hub's hosted transport forwards only the phase name for `phase` events, so a hosted run shows the bare "\<agent\> ready" label with no model, tool count or warning. That is the same code path with fields absent — there is no version fallback. - The "starting" label uses the agent's display name ("Claude Code starting…") rather than the literal "Agent starting…"; the label is right there and reads better. - The 5s to 23s gap between "ready" and the first tool is the model's first thinking; removing it needs `--include-partial-messages`, which is the planned follow-up. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_0125QDX24TNuod7FKLwTDhvW <!-- CURSOR_SUMMARY --> --- > [!NOTE] > **Medium Risk** > Touches investigation streaming, MCP handshake signaling, and live UI state; race handling is new but heavily tested—no auth or data-model changes. > > **Overview** > Investigation Activity now surfaces **real startup progress** before the first agent message, instead of a generic “Starting investigation…” line. > > The backend emits new **`phase` stream events**: `investigating`, then **`connected`** when the agent completes MCP handshake on the private investigation mount (via `investigationrefs.MarkConnected` + middleware on successful `initialize`/`tools/list`), then **`ready`** from the CLI’s `system/init` line with **model**, **Radar tool count** (`mcp__radar__` prefix), and **MCP server statuses**. Events are **mutex-serialized** so handshake and parser callbacks don’t interleave; the handshake watcher still fires once on short turns. **Cursor** maps init to `ready` with **model only**. > > The UI folds phases into **`turn.startup`** (`mergeStartupSignal` keeps the furthest phase), shows agent-specific pending labels while the turn is running and the transcript is empty, and **warns** when MCP servers aren’t `connected` (warning persists on replay). **Tool rows** hide sub-2s durations on the collapsed row (seconds + exact ms in expanded header), show **inline failure reasons**, and move the elapsed timer to a separate **“Ns elapsed”** figure beside the status label. > > <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 8d78513. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> <!-- /CURSOR_SUMMARY -->
1 parent cc95f0d commit 9f73370

12 files changed

Lines changed: 1023 additions & 23 deletions

File tree

‎internal/ai/agent_cursor.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,7 @@ type cursorEvent struct {
197197
Type string `json:"type"` // system|user|assistant|thinking|tool_call|result
198198
Subtype string `json:"subtype"` // init | delta | started | completed | success | ...
199199
SessionID string `json:"session_id"`
200+
Model string `json:"model"` // on system/init
200201
Text string `json:"text"` // on thinking/delta
201202
Message *cursorMessage `json:"message"` // on assistant/user
202203
ToolCall *cursorToolCall `json:"tool_call"`
@@ -292,6 +293,11 @@ func (a *cursorAgent) parseStream(r io.Reader, onEvent func(StreamEvent)) Diagno
292293
sessionID = e.SessionID
293294
}
294295
switch e.Type {
296+
case "system":
297+
// Cursor's init names the model but not its MCP servers or tools.
298+
if e.Subtype == "init" {
299+
onEvent(StreamEvent{Type: "phase", Phase: "ready", Model: e.Model})
300+
}
295301
case "thinking":
296302
if e.Subtype == "delta" && e.Text != "" {
297303
onEvent(StreamEvent{Type: "thinking", Token: e.Text})

‎internal/ai/agent_cursor_test.go‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,10 +96,14 @@ func TestCursorParseStream_FormatPin(t *testing.T) {
9696

9797
var running, done bool
9898
var doneIsError *bool
99-
var thinking, doneResult, doneEvidenceRef, runningSummary string
99+
var thinking, doneResult, doneEvidenceRef, runningSummary, readyModel string
100100
agent := &cursorAgent{bin: "cursor-agent"}
101101
diag := agent.parseStream(strings.NewReader(stream), func(ev StreamEvent) {
102102
switch ev.Type {
103+
case "phase":
104+
if ev.Phase == "ready" {
105+
readyModel = ev.Model
106+
}
103107
case "thinking":
104108
thinking += ev.Token
105109
case "step":
@@ -146,6 +150,9 @@ func TestCursorParseStream_FormatPin(t *testing.T) {
146150
if diag.SessionID != "sess-abc" {
147151
t.Errorf("session id (system/init) not captured: %q", diag.SessionID)
148152
}
153+
if readyModel != "GPT-5.5" {
154+
t.Errorf("ready phase model (system/init) = %q, want GPT-5.5", readyModel)
155+
}
149156
}
150157

151158
func TestCursorParseStreamPreservesUncappedProducerResultForValidation(t *testing.T) {

‎internal/ai/diagnoser.go‎

Lines changed: 75 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -249,8 +249,13 @@ type Diagnosis struct {
249249
// "turn" marks the start of a new turn (carries Question/Apply) so a connecting
250250
// or reconnecting client can reconstruct turn boundaries from the event log.
251251
type StreamEvent struct {
252-
Type string `json:"type"` // "turn"|"phase"|"step"|"thinking"|"done"|"error"|"closed"
253-
Phase string `json:"phase,omitempty"`
252+
Type string `json:"type"` // "turn"|"phase"|"step"|"thinking"|"done"|"error"|"closed"
253+
Phase string `json:"phase,omitempty"` // "investigating"|"connected"|"ready"
254+
// Startup facts reported by the agent CLI on the "ready" phase: the model it
255+
// resolved, how many Radar tools it registered, and each MCP server's status.
256+
Model string `json:"model,omitempty"`
257+
ToolCount *int `json:"toolCount,omitempty"`
258+
MCPServers []MCPServerStatus `json:"mcpServers,omitempty"`
254259
Step *StepInfo `json:"step,omitempty"`
255260
Token string `json:"token,omitempty"`
256261
Diag *Diagnosis `json:"diagnosis,omitempty"`
@@ -277,6 +282,13 @@ const (
277282
ApplyMutationUnknown ApplyMutationOutcome = "unknown"
278283
)
279284

285+
// MCPServerStatus is one MCP server's connection state as the agent CLI
286+
// reported it at startup. Only "connected" means the server's tools are usable.
287+
type MCPServerStatus struct {
288+
Name string `json:"name"`
289+
Status string `json:"status"`
290+
}
291+
280292
// StepInfo describes one tool invocation (running → done).
281293
type StepInfo struct {
282294
ID string `json:"id"`
@@ -609,19 +621,54 @@ func (d *Diagnoser) DiagnoseStream(ctx context.Context, req Request, onEvent fun
609621
return Diagnosis{}, fmt.Errorf("ai: start %s: %w", agent.Name(), err)
610622
}
611623

612-
onEvent(StreamEvent{Type: "phase", Phase: "investigating"})
624+
// The handshake watcher and the stream parser both report to the caller;
625+
// serialize them so callback consumers never see interleaved events.
626+
var emitMu sync.Mutex
627+
emit := func(event StreamEvent) {
628+
emitMu.Lock()
629+
defer emitMu.Unlock()
630+
onEvent(event)
631+
}
632+
emit(StreamEvent{Type: "phase", Phase: "investigating"})
613633
validator := investigationEvidenceValidator{
614634
registry: d.evidenceRefs,
615635
scope: evidenceScope,
616636
claimed: make(map[string]struct{}),
617637
}
618638
streamEvent := func(event StreamEvent) {
619-
onEvent(validator.validate(event))
639+
emit(validator.validate(event))
640+
}
641+
streamDone := make(chan struct{})
642+
var handshake sync.WaitGroup
643+
if evidenceLease != nil {
644+
// The private mount marks the scope connected on the agent's MCP
645+
// handshake, which happens before its first message. Report it once.
646+
handshake.Add(1)
647+
go func() {
648+
defer handshake.Done()
649+
connected := evidenceLease.Connected()
650+
select {
651+
case <-connected:
652+
case <-streamDone:
653+
// Both may be ready when a short turn ends right after its
654+
// handshake; select picks arbitrarily, so re-check before giving up.
655+
select {
656+
case <-connected:
657+
default:
658+
return
659+
}
660+
}
661+
emit(StreamEvent{Type: "phase", Phase: "connected"})
662+
}()
620663
}
621664
diag := agent.parseStream(stdout, streamEvent)
622665
diag.evidenceScope = evidenceScope
623666

624667
waitErr := cmd.Wait()
668+
// The watcher outlives stdout so a handshake that lands between EOF and
669+
// process exit is still reported, and always before this method returns.
670+
close(streamDone)
671+
handshake.Wait()
625672
if evidenceLease != nil {
626673
// Closing before the diagnosis leaves this method prevents any late private
627674
// MCP call from minting evidence for a completed turn. The deferred close
@@ -944,6 +991,26 @@ type cliEvent struct {
944991
TotalCostUSD *float64 `json:"total_cost_usd"`
945992
NumTurns int `json:"num_turns"`
946993
SessionID string `json:"session_id"`
994+
// system/init only.
995+
Model string `json:"model"`
996+
Tools []string `json:"tools"`
997+
MCPServers []MCPServerStatus `json:"mcp_servers"`
998+
}
999+
1000+
// agentReadyEvent reports the CLI's init message as a startup phase. Radar
1001+
// tools are counted by their MCP prefix so the count reflects what the model can
1002+
// actually call, not what Radar offered.
1003+
func agentReadyEvent(model string, tools []string, servers []MCPServerStatus) StreamEvent {
1004+
count := 0
1005+
for _, tool := range tools {
1006+
if strings.HasPrefix(tool, "mcp__radar__") {
1007+
count++
1008+
}
1009+
}
1010+
return StreamEvent{
1011+
Type: "phase", Phase: "ready",
1012+
Model: model, ToolCount: &count, MCPServers: servers,
1013+
}
9471014
}
9481015

9491016
func parseStream(r io.Reader, onEvent func(StreamEvent)) Diagnosis {
@@ -982,6 +1049,10 @@ func parseStream(r io.Reader, onEvent func(StreamEvent)) Diagnosis {
9821049
continue
9831050
}
9841051
switch ev.Type {
1052+
case "system":
1053+
if ev.Subtype == "init" {
1054+
onEvent(agentReadyEvent(ev.Model, ev.Tools, ev.MCPServers))
1055+
}
9851056
case "assistant":
9861057
if ev.Message == nil {
9871058
continue

‎internal/ai/diagnoser_test.go‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,11 @@ import (
99
"net/url"
1010
"os"
1111
"os/exec"
12+
"path/filepath"
1213
"strings"
14+
"sync"
1315
"testing"
16+
"time"
1417

1518
"github.com/skyhook-io/radar/internal/investigationrefs"
1619
)
@@ -922,3 +925,186 @@ func TestDiagnoseStream_ProcessAndStreamErrors(t *testing.T) {
922925
t.Errorf("structured conclusion not preserved: %q", diag.RootCause)
923926
}
924927
}
928+
929+
// TestParseStream_InitReportsAgentReady pins the Claude Code system/init line:
930+
// the resolved model, the Radar tool count by MCP prefix, and every MCP
931+
// server's status, including a failed one the UI must surface as a warning.
932+
func TestParseStream_InitReportsAgentReady(t *testing.T) {
933+
stream := strings.Join([]string{
934+
`{"type":"system","subtype":"hook_started","hook_name":"SessionStart:startup","session_id":"s1"}`,
935+
`{"type":"system","subtype":"init","session_id":"s1","tools":["Bash","mcp__radar__diagnose","mcp__radar__get_resource","mcp__other__ping"],"mcp_servers":[{"name":"radar","status":"connected"},{"name":"other","status":"failed"}],"model":"claude-opus-5[1m]"}`,
936+
`{"type":"result","result":"done","num_turns":1}`,
937+
}, "\n")
938+
var ready []StreamEvent
939+
parseStream(strings.NewReader(stream), func(ev StreamEvent) {
940+
if ev.Type == "phase" {
941+
ready = append(ready, ev)
942+
}
943+
})
944+
if len(ready) != 1 || ready[0].Phase != "ready" {
945+
t.Fatalf("phase events = %+v, want exactly one ready phase", ready)
946+
}
947+
got := ready[0]
948+
if got.Model != "claude-opus-5[1m]" {
949+
t.Errorf("model = %q", got.Model)
950+
}
951+
if got.ToolCount == nil || *got.ToolCount != 2 {
952+
t.Errorf("radar tool count = %v, want 2", got.ToolCount)
953+
}
954+
want := []MCPServerStatus{{Name: "radar", Status: "connected"}, {Name: "other", Status: "failed"}}
955+
if len(got.MCPServers) != len(want) {
956+
t.Fatalf("mcp servers = %+v, want %+v", got.MCPServers, want)
957+
}
958+
for i := range want {
959+
if got.MCPServers[i] != want[i] {
960+
t.Errorf("mcp server %d = %+v, want %+v", i, got.MCPServers[i], want[i])
961+
}
962+
}
963+
raw, err := json.Marshal(got)
964+
if err != nil {
965+
t.Fatal(err)
966+
}
967+
for _, field := range []string{`"toolCount":2`, `"mcpServers":[`, `"model":"claude-opus-5[1m]"`} {
968+
if !strings.Contains(string(raw), field) {
969+
t.Errorf("serialized ready phase %s lacks %s", raw, field)
970+
}
971+
}
972+
zero := agentReadyEvent("m", nil, nil)
973+
if raw, _ := json.Marshal(zero); !strings.Contains(string(raw), `"toolCount":0`) {
974+
t.Errorf("a zero Radar tool count must serialize, got %s", raw)
975+
}
976+
}
977+
978+
// TestDiagnoseStreamReportsHandshakeOnce pins the agent-agnostic "connected"
979+
// phase: the private mount's handshake produces exactly one event per turn, no
980+
// matter how many times the scope is marked, and it lands before the stream ends.
981+
func TestDiagnoseStreamReportsHandshakeOnce(t *testing.T) {
982+
dir := t.TempDir()
983+
flag := filepath.Join(dir, "handshake-observed")
984+
bin := filepath.Join(dir, "claude")
985+
resultLine := `{"type":"result","result":"` + "```json\\n{\\\"healthy\\\":true}\\n```" + `","num_turns":1}`
986+
script := "#!/bin/sh\nwhile [ ! -f '" + flag + "' ]; do sleep 0.02; done\n" +
987+
"printf '%s\\n' '" + resultLine + "'\n"
988+
if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
989+
t.Fatal(err)
990+
}
991+
refs := investigationrefs.NewRegistry()
992+
d, err := New(bin, refs)
993+
if err != nil {
994+
t.Fatal(err)
995+
}
996+
scope := strings.Repeat("d", 26)
997+
started := make(chan struct{})
998+
var mu sync.Mutex
999+
var phases []string
1000+
type outcome struct {
1001+
diag Diagnosis
1002+
err error
1003+
}
1004+
finished := make(chan outcome, 1)
1005+
go func() {
1006+
diag, err := d.DiagnoseStream(context.Background(), Request{
1007+
Kind: "Pod", Namespace: "ns", Name: "p", MCPPort: 1, EvidenceScope: scope,
1008+
}, func(ev StreamEvent) {
1009+
if ev.Type != "phase" {
1010+
return
1011+
}
1012+
mu.Lock()
1013+
phases = append(phases, ev.Phase)
1014+
mu.Unlock()
1015+
if ev.Phase == "investigating" {
1016+
close(started)
1017+
}
1018+
})
1019+
finished <- outcome{diag, err}
1020+
}()
1021+
select {
1022+
case <-started:
1023+
case <-time.After(10 * time.Second):
1024+
t.Fatal("agent never started")
1025+
}
1026+
if !refs.MarkConnected(scope) || !refs.MarkConnected(scope) {
1027+
t.Fatal("live turn scope rejected its handshake")
1028+
}
1029+
deadline := time.Now().Add(5 * time.Second)
1030+
for {
1031+
mu.Lock()
1032+
seen := len(phases) >= 2
1033+
mu.Unlock()
1034+
if seen || time.Now().After(deadline) {
1035+
break
1036+
}
1037+
time.Sleep(10 * time.Millisecond)
1038+
}
1039+
if err := os.WriteFile(flag, nil, 0o644); err != nil {
1040+
t.Fatal(err)
1041+
}
1042+
var result outcome
1043+
select {
1044+
case result = <-finished:
1045+
case <-time.After(10 * time.Second):
1046+
t.Fatal("DiagnoseStream did not finish")
1047+
}
1048+
if result.err != nil || !result.diag.Healthy {
1049+
t.Fatalf("diagnosis = %+v, err = %v", result.diag, result.err)
1050+
}
1051+
mu.Lock()
1052+
defer mu.Unlock()
1053+
if strings.Join(phases, ",") != "investigating,connected" {
1054+
t.Fatalf("phases = %v, want investigating then exactly one connected", phases)
1055+
}
1056+
}
1057+
1058+
// handshakeAgent completes the private-mount handshake from inside command()
1059+
// and exits at once, so the stream can end in the same instant the scope is
1060+
// marked connected.
1061+
type handshakeAgent struct {
1062+
refs *investigationrefs.Registry
1063+
}
1064+
1065+
func (*handshakeAgent) Name() string { return "claude" }
1066+
func (*handshakeAgent) Path() string { return "printf" }
1067+
func (*handshakeAgent) SigninCmd() string { return "claude auth login" }
1068+
func (a *handshakeAgent) parseStream(r io.Reader, onEvent func(StreamEvent)) Diagnosis {
1069+
return parseStream(r, onEvent)
1070+
}
1071+
func (a *handshakeAgent) command(ctx context.Context, spec turnSpec) (*exec.Cmd, func(), error) {
1072+
u, err := url.Parse(spec.mcpURL)
1073+
if err != nil {
1074+
return nil, func() {}, err
1075+
}
1076+
if !a.refs.MarkConnected(u.Query().Get("scope")) {
1077+
return nil, func() {}, fmt.Errorf("test agent could not mark the scope connected")
1078+
}
1079+
stream := `{"type":"result","result":"` + "```json\\n{\\\"healthy\\\":true}\\n```" + `","num_turns":1}`
1080+
return exec.CommandContext(ctx, "printf", "%s\n", stream), func() {}, nil
1081+
}
1082+
1083+
// TestDiagnoseStreamReportsHandshakeOnShortTurn pins that a handshake which
1084+
// lands as the stream ends is still reported exactly once, before the turn
1085+
// returns, rather than lost to a select that happens to observe the end first.
1086+
func TestDiagnoseStreamReportsHandshakeOnShortTurn(t *testing.T) {
1087+
refs := investigationrefs.NewRegistry()
1088+
diagnoser := &Diagnoser{
1089+
agents: map[string]Agent{"claude": &handshakeAgent{refs: refs}},
1090+
defName: "claude",
1091+
evidenceRefs: refs,
1092+
}
1093+
for i := 0; i < 20; i++ {
1094+
var phases []string
1095+
diag, err := diagnoser.DiagnoseStream(context.Background(), Request{
1096+
Kind: "Pod", Namespace: "ns", Name: "p", MCPPort: 1,
1097+
EvidenceScope: strings.Repeat("e", 26),
1098+
}, func(ev StreamEvent) {
1099+
if ev.Type == "phase" {
1100+
phases = append(phases, ev.Phase)
1101+
}
1102+
})
1103+
if err != nil || !diag.Healthy {
1104+
t.Fatalf("diagnosis = %+v, err = %v", diag, err)
1105+
}
1106+
if strings.Join(phases, ",") != "investigating,connected" {
1107+
t.Fatalf("run %d phases = %v, want investigating then exactly one connected", i, phases)
1108+
}
1109+
}
1110+
}

0 commit comments

Comments
 (0)