refactor(daemon): give stream events proper types
Validate known events before delivery and leave the cursor untouched on malformed batches.
31 unmodified lines3233343536373839404129 unmodified lines31 unmodified lines`ctrl+o` or `/agents` in a session shows its whole tree live: running agents pulse with their token rate, mail travels the edges as blocks. tab picks an agent, enter opens it (typing there is you, as the user), text + enter sends to it, `/spawn <name> <task>` starts a child under it.
routes: `GET /agents?session=<id>` (the tree), `GET /agents/stream` (batched events), `POST /sessions/:id/children`, `POST /sessions/:id/mail`.
The Go client decodes known agent events into `daemon.AgentEvent` before delivery to the view. Unknown event kinds are ignored. Malformed known events return a protocol error instead of supplying empty fields to the view. Tool progress uses the same typed value as the session stream.
## swarm overhead
kernel boots queue behind four slots. each boot owns only its session's composition,and each kernel's host callback captures only its routes, store handle, and session29 unmodified lines2 unmodified lines345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485138687881689909192939423959625979899100101102103104105106107108109110111112281131141151161172 unmodified lines2 unmodified linesimport ( "bufio" "context" "encoding/json" "fmt" "net/http" "strings")
// AgentEvent is one event from the daemon's agent bus.type AgentEvent struct { Progress *ToolProgress `json:"progress"` Type string `json:"type"` Session string `json:"session"` Parent string `json:"parent"` Name string `json:"name"` Model string `json:"model"` From string `json:"from"` To string `json:"to"` FromName string `json:"fromName"` Kind string `json:"kind"` Text string `json:"text"` CallID string `json:"callId"` Output string `json:"output"` Source string `json:"source"` Depth int `json:"depth"` Bytes int `json:"bytes"` Running bool `json:"running"`}
func decodeAgentEvent(raw json.RawMessage) (*AgentEvent, error) { var header struct { Type string `json:"type"` } if err := json.Unmarshal(raw, &header); err != nil { return nil, fmt.Errorf("invalid agent event: %w", err) } switch header.Type { case "spawn", "gone", "mail", "running", "text", "thinking", "arguments_delta", "tool_progress", "tool", "user", "message", "error", "interrupted", "progress", "renamed", "closed": default: return nil, nil } var event AgentEvent if err := json.Unmarshal(raw, &event); err != nil { return nil, fmt.Errorf("invalid %s agent event: %w", header.Type, err) } var fields map[string]json.RawMessage if err := json.Unmarshal(raw, &fields); err != nil { return nil, err } required := []string{"session"} switch event.Type { case "mail": required = []string{"to", "bytes", "kind"} case "spawn": required = append(required, "parent", "name", "model", "depth") case "running": required = append(required, "running") case "text", "thinking", "arguments_delta", "user", "message", "error", "progress": required = append(required, "text") if event.Type == "arguments_delta" { required = append(required, "callId", "name") } case "tool_progress": if err := validateToolProgress(fields["progress"]); err != nil { return nil, err } required = append(required, "progress") case "tool": required = append(required, "output") case "renamed": required = append(required, "name") } for _, name := range required { if value, exists := fields[name]; !exists || (string(value) == "null" && name != "progress") { return nil, fmt.Errorf("invalid %s agent event: missing %s", event.Type, name) } } return &event, nil}
// StreamAgents delivers nonempty batches until the stream ends or onBatch fails.// The caller controls cancellation, including any blocking work in onBatch.func StreamAgents(ctx context.Context, conn *Connection, onBatch func([]map[string]any) error) error {func StreamAgents(ctx context.Context, conn *Connection, onBatch func([]AgentEvent) error) error { operation := Operation{Name: "stream agents", Method: http.MethodGet, Path: "/agents/stream", Policy: ReadRecovery} return scanEventStream(ctx, conn, operation, streamLimits{lineBytes: 8 * 1024 * 1024}, func(scanner *bufio.Scanner) error { // The daemon sends each JSON batch on one data: line; batches do not span lines. for scanner.Scan() { line := scanner.Text() if !strings.HasPrefix(line, "data:") { continue } var batch struct { Events []map[string]any `json:"events"` Events []json.RawMessage `json:"events"` } if json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &batch) != nil || len(batch.Events) == 0 { if err := json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &batch); err != nil { return fmt.Errorf("invalid agent batch: %w", err) } events := make([]AgentEvent, 0, len(batch.Events)) for _, raw := range batch.Events { event, err := decodeAgentEvent(raw) if err != nil { return err } if event != nil { events = append(events, *event) } } if len(events) == 0 { continue } if err := onBatch(batch.Events); err != nil { if err := onBatch(events); err != nil { return err } } return nil2 unmodified lines12123456789101112131415161718192023 unmodified lines4445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107// Stalled health checks and startup locks need controlled cancellation that// daemon E2E cannot force reliably.// Controlled cancellation and malformed wire batches cannot be forced reliably// through a scripted provider; these tests exercise the public transport boundaries.package daemon
import ( "bufio" "context" "encoding/json" "errors" "fmt" "net" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "testing" "time")
23 unmodified lines } }) }}
func TestMalformedBatchDoesNotAdvanceCursorOrDeliverPartialEvents(t *testing.T) { client := NewChatClient(NewConnection(ConnectionSnapshot{}, ""), "s") delivered := 0 scan := func(payload string) error { return client.readStream(context.Background(), bufio.NewScanner(strings.NewReader(payload)), func(event StreamEvent) error { if event.Type == EventText { delivered++ } return nil }) } if err := scan("data: {\"cursor\":7,\"events\":[{\"type\":\"text\",\"text\":\"ok\"}]}\n\n"); err != nil { t.Fatal(err) } if err := scan("data: {\"cursor\":8,\"events\":[{\"type\":\"text\",\"text\":\"partial\"},{\"type\":\"message\",\"text\":false}]}\n\n"); err == nil { t.Fatal("malformed known event accepted") } if client.afterSeq != 7 || delivered != 1 { t.Fatalf("malformed batch changed stream state: cursor=%d delivered=%d", client.afterSeq, delivered) } if err := scan("data: {\"cursor\":9,\"events\":[{\"type\":\"future\",\"text\":false}]}\n\n"); err != nil { t.Fatal(err) } if client.afterSeq != 9 || delivered != 1 { t.Fatal("unknown event should advance cursor without delivery") }}
func TestTypedEventFormatsAndValidation(t *testing.T) { for _, body := range []string{ `{"type":"turn_membership","turnId":"turn","submissionIds":["one","two"]}`, `{"type":"turn_completed","turnId":"turn"}`, `{"type":"tool","name":"python","args":"{\"code\":\"print(1)\"}","result":"1"}`, `{"type":"usage","promptTokens":10,"elapsedMs":1.5,"cacheFade":[{"at":1,"cached":2}]}`, `{"type":"tool_progress","progress":null}`, } { if _, err := decodeChatEvent(json.RawMessage(body)); err != nil { t.Fatalf("valid format %s: %v", body, err) } } for _, body := range []string{ `{"type":"turn_membership","turnId":"turn","submissionIds":[1]}`, `{"type":"usage","promptTokens":"10"}`, `{"type":"thinking","text":"thought","elapsedMs":1.5}`, `{"type":"committed","seq":0}`, `{"type":"compacted","evicted":2}`, } { if _, err := decodeChatEvent(json.RawMessage(body)); err == nil { t.Fatalf("malformed event accepted: %s", body) } } event, err := decodeAgentEvent(json.RawMessage(`{"type":"mail","from":null,"fromName":"","to":"s","kind":"message","bytes":2}`)) if err != nil || event == nil || event.To != "s" { t.Fatalf("nullable bus sender rejected: %v", err) } if _, err := decodeAgentEvent(json.RawMessage(`{"type":"running","session":"s","running":"false"}`)); err == nil { t.Fatal("malformed bus boolean accepted") }}146 unmodified linesExpand all14714814915015115215315115215315415515615715815916016116215815916016116216316416516616716816917017117217317454 unmodified lines22923023123222622722822923023323423523623723823924023323423523623723823924024124124224324424524624724824925025125225325425525625725825926026126226326426526626724424524624724824925025125225326826925625725825926026126226326426526626727027127227327427527626927027127727827927327427527627728028128228328428528628728828929027928028128228328428528629129229329429528828929029129229329429529629729829930030130230330429629729829930030130230330430530630730830931031130830931031131231331431531631731231331431531631932032132232332432532631731831932032132832933033133233333433533633733833934034132232332434334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039132532632732839339439539639739839940040140232933033133233333440440540640740833533633733833934034134234334434541034641241341441541641741841942042142242342442534734834935035135235335442742842943043143243343443543643743843944044144244344444544635535635735835936036136236336436536636736836937037137229 unmodified lines402403404405485486406407408488489490491492409410494411496497498499412413414501415416417418419507420421422423424425426515427428429430520521522523524525431527528529530531532533432433434435436437438439440441442443444535445446537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520659521522523524525526527528529530531532533534535536537538539540541542543544545546547146 unmodified linesExpand all return nil, err }
var data struct { Events []map[string]any `json:"events"` Before int64 `json:"before"` More bool `json:"more"` Before *int64 `json:"before"` More *bool `json:"more"` Events []json.RawMessage `json:"events"` } if err := json.Unmarshal(body, &data); err != nil { return nil, err } page := &HistoryPage{Before: data.Before, More: data.More} for _, evMap := range data.Events { if ev := normalizeEvent(evMap); ev != nil { ev.Replayed = true page.Events = append(page.Events, *ev) if data.Events == nil || data.Before == nil || data.More == nil || *data.Before < 0 { return nil, errors.New("invalid history page: missing events, before, or more") } page := &HistoryPage{Before: *data.Before, More: *data.More} for _, raw := range data.Events { event, err := decodeChatEvent(raw) if err != nil { return nil, err } if event != nil { event.Replayed = true page.Events = append(page.Events, event.StreamEvent) } } return page, nil}
54 unmodified lines Phase: phase, }, nil}
func int64Field(raw map[string]any, key string) int64 { if n, ok := raw[key].(float64); ok && n > 0 { return int64(n) } return 0// wireChatEvent keeps arbitrary tool arguments separate from protocol fields.type wireChatEvent struct { CallID string `json:"callId"` Args json.RawMessage `json:"args"` Elapsed json.RawMessage `json:"elapsedMs"` StreamEvent}
// normalizeEvent parses one daemon event map, decoding a tool's JSON-encoded args.func normalizeEvent(evMap map[string]any) *StreamEvent { if evMap["type"] == "tool" { if strArgs, ok := evMap["args"].(string); ok { var parsedArgs map[string]any if err := json.Unmarshal([]byte(strArgs), &parsedArgs); err == nil { evMap["args"] = parsedArgs } else { evMap["args"] = make(map[string]any)func decodeChatEvent(data json.RawMessage) (*wireChatEvent, error) { var header struct { Type EventType `json:"type"` } if err := json.Unmarshal(data, &header); err != nil { return nil, fmt.Errorf("invalid event: %w", err) } switch header.Type { case "", EventText, EventReset, EventRetry, EventError, EventInterrupted, EventMessage, EventUser, EventThinking, EventNote, EventCompacted, EventToolProgress, EventTool, EventUsage, EventCommitted, EventTurnMembership, EventTurnCompleted, "arguments_delta", "turn_started": default: return nil, nil } var event wireChatEvent if err := json.Unmarshal(data, &event); err != nil { return nil, fmt.Errorf("invalid %s event: %w", header.Type, err) } var fields map[string]json.RawMessage if err := json.Unmarshal(data, &fields); err != nil { return nil, err } require := func(names ...string) error { for _, name := range names { value, exists := fields[name] if !exists || string(value) == "null" { return fmt.Errorf("invalid %s event: missing %s", header.Type, name) } } } return parseStreamEvent(evMap)}
func parseStreamEvent(raw map[string]any) *StreamEvent { typ, _ := raw["type"].(string) if typ == "" { if txt, ok := raw["text"].(string); ok { return &StreamEvent{Type: EventText, Text: txt} } return nil }
var timestamp *int64 if ts, ok := raw["timestamp"].(float64); ok && ts >= 0 { t := int64(ts) timestamp = &t }
switch typ { case "tool_progress": progVal := raw["progress"] if progVal == nil { return &StreamEvent{Type: EventToolProgress, Progress: nil} var err error switch header.Type { case "", EventText, EventMessage, EventUser, EventThinking, EventNote, EventError: err = require("text") if header.Type == "" { event.Type = EventText } progMap, ok := progVal.(map[string]any) if !ok { return nil if header.Type == EventMessage { event.Role = "assistant" } callID, _ := progMap["callId"].(string) name, _ := progMap["name"].(string) phase, _ := progMap["phase"].(string) if len(callID) > 200 || len(name) > 100 || (phase != "generating" && phase != "running") { return nil case "arguments_delta": err = require("callId", "name", "text") case EventTurnMembership: err = require("turnId", "submissionIds") case EventTurnCompleted: err = require("turnId") case EventCommitted: err = require("seq") if event.Seq <= 0 { err = errors.New("invalid committed event: seq must be positive") }
var intent *ToolIntent if im, ok := progMap["intent"].(map[string]any); ok { k, _ := im["kind"].(string) t, _ := im["target"].(string) if (k == "write" || k == "edit" || k == "read" || k == "run") && len(t) <= 300 { intent = &ToolIntent{Kind: k, Target: t} } case EventCompacted: err = require("evicted", "summary") if event.Evicted < 0 || event.Evicted > 1000000 || len(event.Summary) > 60000 { err = errors.New("invalid compacted event") }
var code *ToolCodePreview if cm, ok := progMap["code"].(map[string]any); ok { off, hasOff := cm["offset"].(float64) txt, hasTxt := cm["text"].(string) if hasOff && hasTxt && off >= 0 { switch { case phase == "generating" && len(txt) <= 512: code = &ToolCodePreview{Offset: int(off), Text: txt} case phase == "running" && len(txt) <= 16000: // a running call's code is its start: what it does, and // the first line to name it by if intent == nil && name == "python" { intent = ParsePythonIntent(txt) } head, _, _ := strings.Cut(strings.TrimSpace(txt), "\n") code = &ToolCodePreview{Text: sanitizeControlRunes(head)} case EventTool: err = require("name", "result") if len(event.Args) != 0 && string(event.Args) != "null" { args := event.Args if args[0] == '"' { var encoded string if decodeErr := json.Unmarshal(args, &encoded); decodeErr != nil { return nil, decodeErr } args = []byte(encoded) } // Invalid tool argument JSON is display-only, and has historically rendered empty. if json.Unmarshal(args, &event.ToolArgs) != nil { event.ToolArgs = map[string]any{} } }
return &StreamEvent{ Type: EventToolProgress, Progress: &ToolProgress{ CallID: callID, Name: cleanLabel(name), Phase: phase, Intent: intent, Code: code, }, case EventUsage: var usage Usage if decodeErr := json.Unmarshal(data, &usage); decodeErr != nil { return nil, fmt.Errorf("invalid usage event: %w", decodeErr) }
case "message": txt, _ := raw["text"].(string) return &StreamEvent{ Type: EventMessage, Role: "assistant", Text: txt, Timestamp: timestamp, event.Usage = &usage case EventToolProgress: if fieldErr := validateToolProgress(fields["progress"]); fieldErr != nil { return nil, fieldErr }
case "interrupted": return &StreamEvent{Type: EventInterrupted}
case "retry": return &StreamEvent{Type: EventRetry}
case "reset": return &StreamEvent{Type: EventReset, Before: int64Field(raw, "before"), More: raw["more"] == true}
case "committed": seq := int64Field(raw, "seq") if seq <= 0 { return nil if _, exists := fields["progress"]; !exists { err = errors.New("invalid tool progress event: missing progress") } return &StreamEvent{Type: EventCommitted, Seq: seq}
case "note": txt, _ := raw["text"].(string) return &StreamEvent{Type: EventNote, Text: txt}
case "user": txt, _ := raw["text"].(string) source, _ := raw["source"].(string) trig, _ := raw["triggeredAt"].(string) clientID, _ := raw["clientId"].(string) img := ParseImageMetadata(raw["image"]) return &StreamEvent{ Type: EventUser, Text: txt, Source: source, TriggeredAt: trig, ClientID: clientID, Timestamp: timestamp, Image: img, }
case "thinking": txt, _ := raw["text"].(string) return &StreamEvent{Type: EventThinking, Text: txt, ElapsedMs: int64Field(raw, "elapsedMs")}
case "error": txt, _ := raw["text"].(string) return &StreamEvent{Type: EventError, Text: txt}
case "tool": name, _ := raw["name"].(string) result, _ := raw["result"].(string) args, _ := raw["args"].(map[string]any) trace := ParseToolTrace(raw["trace"]) return &StreamEvent{ Type: EventTool, ToolName: name, ToolArgs: args, ToolResult: result, ToolTrace: trace, }
case "usage": model, _ := raw["model"].(string) var cacheFade []CacheStep if data, err := json.Marshal(raw["cacheFade"]); err == nil { if err := json.Unmarshal(data, &cacheFade); err != nil { cacheFade = nil if progress := event.Progress; progress != nil { if len(progress.CallID) > 200 || len(progress.Name) > 100 || (progress.Phase != "generating" && progress.Phase != "running") { return nil, errors.New("invalid tool progress event") } } var recordedAt *int64 if ra, ok := raw["recordedAt"].(float64); ok { r := int64(ra) recordedAt = &r } getInt := func(k string) *int { if v, ok := raw[k].(float64); ok { i := int(v) return &i progress.Name = cleanLabel(progress.Name) if intent := progress.Intent; intent != nil { if (intent.Kind != "write" && intent.Kind != "edit" && intent.Kind != "read" && intent.Kind != "run") || len(intent.Target) > 300 { progress.Intent = nil } } return nil } getFloat := func(k string) *float64 { if v, ok := raw[k].(float64); ok { return &v if code := progress.Code; code != nil { if code.Offset < 0 || (progress.Phase == "generating" && len(code.Text) > 512) || len(code.Text) > 16000 { progress.Code = nil } else if progress.Phase == "running" { if progress.Intent == nil && progress.Name == "python" { progress.Intent = ParsePythonIntent(code.Text) } head, _, _ := strings.Cut(strings.TrimSpace(code.Text), "\n") progress.Code = &ToolCodePreview{Text: sanitizeControlRunes(head)} } } return nil } return &StreamEvent{ Type: EventUsage, Usage: &Usage{ Model: model, RecordedAt: recordedAt, PromptTokens: getInt("promptTokens"), CachedPromptTokens: getInt("cachedPromptTokens"), CacheWriteTokens: getInt("cacheWriteTokens"), CompletionTokens: getInt("completionTokens"), TotalTokens: getInt("totalTokens"), ElapsedMs: getFloat("elapsedMs"), TokensPerSecond: getFloat("tokensPerSecond"), CacheFade: cacheFade, }, } if err != nil { return nil, err } if event.Type != EventUsage && len(event.Elapsed) != 0 { if err := json.Unmarshal(event.Elapsed, &event.ElapsedMs); err != nil { return nil, fmt.Errorf("invalid elapsedMs: %w", err) }
case "compacted": evicted, hasEv := raw["evicted"].(float64) summary, hasSum := raw["summary"].(string) if hasEv && hasSum && evicted >= 0 && evicted <= 1000000 && len(summary) <= 60000 { strategy, _ := raw["strategy"].(string) return &StreamEvent{ Type: EventCompacted, Evicted: int(evicted), Summary: summary, Strategy: strategy, } } return nil
default: if txt, ok := raw["text"].(string); ok { return &StreamEvent{Type: EventText, Text: txt} } return nil } if event.Timestamp != nil && *event.Timestamp < 0 { event.Timestamp = nil } if event.ElapsedMs < 0 { event.ElapsedMs = 0 } if event.Image != nil { event.Image = ParseImageMetadata(event.Image) } if event.ToolTrace != nil { event.ToolTrace = ParseToolTrace(event.ToolTrace) } return &event, nil}
// ResetStream discards unfinished tool arguments and starts the next stream from// a fresh snapshot. Call it after the subscription has stopped writing.29 unmodified lines })}
func (c *ChatClient) readStream(ctx context.Context, scanner *bufio.Scanner, onEvent func(StreamEvent) error) error { reporter := NewToolProgressReporter(func(prog *ToolProgress) error { return onEvent(StreamEvent{Type: EventToolProgress, Progress: prog}) reporter := NewToolProgressReporter(func(progress *ToolProgress) error { return onEvent(StreamEvent{Type: EventToolProgress, Progress: progress}) })
defer func() { _ = reporter.Report(nil, "running") }()
defer func() { _ = reporter.Report(nil, "running") }() var eventType string
for scanner.Scan() { select { case <-ctx.Done(): return ctx.Err() default: if err := ctx.Err(); err != nil { return err }
line := scanner.Text() if line == "" { eventType = "" continue }
if strings.HasPrefix(line, "event:") { eventType = strings.TrimSpace(line[6:]) continue } if !strings.HasPrefix(line, "data:") { continue }
payload := strings.TrimSpace(line[5:]) if payload == "" { continue }
var raw any if err := json.Unmarshal([]byte(payload), &raw); err != nil { continue }
if eventType == "error" { errStr := payload if m, ok := raw.(map[string]any); ok { if e, ok := m["error"].(string); ok { errStr = e } else if t, ok := m["text"].(string); ok { errStr = t } var failure struct { Error string `json:"error"` Text string `json:"text"` } if err := json.Unmarshal([]byte(payload), &failure); err != nil { return fmt.Errorf("invalid stream error: %w", err) } if failure.Error != "" { return errors.New(failure.Error) } if failure.Text != "" { return errors.New(failure.Text) } return errors.New(errStr) return errors.New(payload) }
rawMap, isMap := raw.(map[string]any) if isMap { if cursorVal, hasCursor := rawMap["cursor"]; hasCursor { if cursorNum, ok := cursorVal.(float64); ok { if eventsArr, ok := rawMap["events"].([]any); ok { // a reset opens the snapshot, and the snapshot is the rest of its page snapshot := false for _, evRaw := range eventsArr { evMap, ok := evRaw.(map[string]any) if !ok { continue } itemType, _ := evMap["type"].(string) switch itemType { case "reset", "retry", "turn_started", "message", "interrupted", "error": c.mu.Lock() c.argumentsByCall = make(map[string]*strings.Builder) c.mu.Unlock() if err := reporter.Report(nil, "running"); err != nil { return err } }
if itemType == "reset" { snapshot = true if err := onEvent(*parseStreamEvent(evMap)); err != nil { return err } continue } if itemType == "turn_started" { continue } if itemType == "tool" { if callID, ok := evMap["callId"].(string); ok { c.mu.Lock() delete(c.argumentsByCall, callID) c.mu.Unlock() } if err := reporter.Report(nil, "running"); err != nil { return err } } if itemType == "tool_progress" { if prog, ok := evMap["progress"].(map[string]any); ok { ph, _ := prog["phase"].(string) cid, _ := prog["callId"].(string) if ph == "running" && cid != "" { c.mu.Lock() delete(c.argumentsByCall, cid) c.mu.Unlock() if err := reporter.Report(nil, "running"); err != nil { return err } } } } if itemType == "arguments_delta" { callID, _ := evMap["callId"].(string) name, _ := evMap["name"].(string) text, _ := evMap["text"].(string) if callID != "" && text != "" { c.mu.Lock() arguments, exists := c.argumentsByCall[callID] fresh := !exists
retained := 0 for k, v := range c.argumentsByCall { retained += len(k) + v.Len() }
extraCallID := 0 if fresh { extraCallID = len(callID) }
if (fresh && len(c.argumentsByCall) >= 32) || retained+len(text)+extraCallID > 2000000 { c.argumentsByCall = make(map[string]*strings.Builder) c.afterSeq = -1 c.mu.Unlock() return errors.New("too many tool argument previews for this client to display") }
if fresh { arguments = new(strings.Builder) c.argumentsByCall[callID] = arguments } // Append-only builders keep earlier String snapshots immutable. arguments.WriteString(text) args := arguments.String() c.mu.Unlock()
call := &ToolCallAssembly{ ID: callID, } call.Function.Name = name call.Function.Arguments = args if err := reporter.Report(call, "generating"); err != nil { return err } continue } }
if ev := normalizeEvent(evMap); ev != nil { ev.Replayed = snapshot if err := onEvent(*ev); err != nil { return err } } }
c.mu.Lock() c.afterSeq = int(cursorNum) c.mu.Unlock() continue var batch struct { Cursor *int `json:"cursor"` Events []json.RawMessage `json:"events"` } if err := json.Unmarshal([]byte(payload), &batch); err != nil { return fmt.Errorf("invalid stream batch: %w", err) } if batch.Cursor == nil || *batch.Cursor < 0 || batch.Events == nil { return errors.New("invalid stream batch: missing cursor or events") } // Validate the whole batch before delivering events or changing preview state. events := make([]wireChatEvent, 0, len(batch.Events)) for _, raw := range batch.Events { event, err := decodeChatEvent(raw) if err != nil { return err } if event != nil { events = append(events, *event) } } snapshot := false for _, event := range events { switch event.Type { case EventReset, EventRetry, "turn_started", EventMessage, EventInterrupted, EventError: c.mu.Lock() c.argumentsByCall = make(map[string]*strings.Builder) c.mu.Unlock() if err := reporter.Report(nil, "running"); err != nil { return err } } if event.Type == EventReset { snapshot = true } if event.Type == "turn_started" { continue } if event.Type == EventTool || (event.Type == EventToolProgress && event.Progress != nil && event.Progress.Phase == "running" && event.Progress.CallID != "") { callID := event.CallID if event.Progress != nil { callID = event.Progress.CallID } c.mu.Lock() delete(c.argumentsByCall, callID) c.mu.Unlock() if err := reporter.Report(nil, "running"); err != nil { return err } } if event.Type == "arguments_delta" { if event.CallID != "" && event.Text != "" { arguments, err := c.appendArguments(event.CallID, event.Text) if err != nil { return err } call := &ToolCallAssembly{ID: event.CallID} call.Function.Name = event.ToolName call.Function.Arguments = arguments if err := reporter.Report(call, "generating"); err != nil { return err } } continue } event.Replayed = snapshot && event.Type != EventReset if err := onEvent(event.StreamEvent); err != nil { return err } } c.mu.Lock() c.afterSeq = *batch.Cursor c.mu.Unlock() }
return nil}
func (c *ChatClient) appendArguments(callID, text string) (string, error) { c.mu.Lock() defer c.mu.Unlock() arguments, exists := c.argumentsByCall[callID] retained := 0 for key, value := range c.argumentsByCall { retained += len(key) + value.Len() } extraCallID := 0 if !exists { extraCallID = len(callID) } if (!exists && len(c.argumentsByCall) >= 32) || retained+len(text)+extraCallID > 2000000 { c.argumentsByCall = make(map[string]*strings.Builder) c.afterSeq = -1 return "", errors.New("too many tool argument previews for this client to display") } if !exists { arguments = new(strings.Builder) c.argumentsByCall[callID] = arguments } arguments.WriteString(text) return arguments.String(), nil}1234567891094 unmodified lines10510610710810710810911011111211311411511611711811910911011111211311411511611711811912012112212312412512612719 unmodified lines14714814915014714814915015115215315415515615715815916016116216315115215315415515615715815916016116216316416516616716816917017117217316 unmodified lines190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229package daemon
import ( "encoding/json" "errors" "fmt" "strings" "unicode")
94 unmodified lines
type EventType string
const ( EventReset EventType = "reset" EventRetry EventType = "retry" EventText EventType = "text" EventError EventType = "error" EventInterrupted EventType = "interrupted" EventMessage EventType = "message" EventUser EventType = "user" EventThinking EventType = "thinking" EventNote EventType = "note" EventCompacted EventType = "compacted" EventToolProgress EventType = "tool_progress" EventTool EventType = "tool" EventUsage EventType = "usage" EventReset EventType = "reset" EventRetry EventType = "retry" EventText EventType = "text" EventError EventType = "error" EventInterrupted EventType = "interrupted" EventMessage EventType = "message" EventUser EventType = "user" EventTurnMembership EventType = "turn_membership" EventTurnCompleted EventType = "turn_completed" EventThinking EventType = "thinking" EventNote EventType = "note" EventCompacted EventType = "compacted" EventToolProgress EventType = "tool_progress" EventTool EventType = "tool" EventUsage EventType = "usage" // EventCommitted says transcript rows up to Seq now cover what was shown. EventCommitted EventType = "committed")
19 unmodified lines At int64 `json:"at"`}
type StreamEvent struct { Image *ImageMetadata `json:"image,omitempty"` Usage *Usage `json:"usage,omitempty"` ToolTrace *ToolTrace `json:"trace,omitempty"` Timestamp *int64 `json:"timestamp,omitempty"` ToolArgs map[string]any `json:"args,omitempty"` Progress *ToolProgress `json:"progress,omitempty"` Type EventType `json:"type"` Text string `json:"text,omitempty"` Role string `json:"role,omitempty"` // "assistant" Source string `json:"source,omitempty"` TriggeredAt string `json:"triggeredAt,omitempty"` ClientID string `json:"clientId,omitempty"` Summary string `json:"summary,omitempty"` Strategy string `json:"strategy,omitempty"` ToolName string `json:"name,omitempty"` ToolResult string `json:"result,omitempty"` Evicted int `json:"evicted,omitempty"` Image *ImageMetadata `json:"image,omitempty"` Usage *Usage `json:"usage,omitempty"` ToolTrace *ToolTrace `json:"trace,omitempty"` Timestamp *int64 `json:"timestamp,omitempty"` ToolArgs map[string]any `json:"args,omitempty"` Progress *ToolProgress `json:"progress,omitempty"` TurnID string `json:"turnId,omitempty"` Type EventType `json:"type"` Text string `json:"text,omitempty"` Role string `json:"role,omitempty"` // "assistant" Source string `json:"source,omitempty"` TriggeredAt string `json:"triggeredAt,omitempty"` ClientID string `json:"clientId,omitempty"` Summary string `json:"summary,omitempty"` Strategy string `json:"strategy,omitempty"` ToolName string `json:"name,omitempty"` ToolResult string `json:"result,omitempty"` SubmissionIDs []string `json:"submissionIds,omitempty"` Evicted int `json:"evicted,omitempty"` // ElapsedMs is how long a replayed thought took, when the daemon timed it. ElapsedMs int64 `json:"elapsedMs,omitempty"` // Seq is the newest transcript row an EventCommitted covers. Seq int64 `json:"seq,omitempty"`16 unmodified lines } } return sb.String()}
// validateToolProgress distinguishes a display reset from malformed known fields.func validateToolProgress(raw json.RawMessage) error { if string(raw) == "null" { return nil } var progress struct { CallID *string `json:"callId"` Name *string `json:"name"` Phase *string `json:"phase"` Intent *struct { Kind *string `json:"kind"` Target *string `json:"target"` } `json:"intent"` Code *struct { Text *string `json:"text"` Offset *int `json:"offset"` } `json:"code"` } if err := json.Unmarshal(raw, &progress); err != nil { return fmt.Errorf("invalid tool progress: %w", err) } if progress.CallID == nil || progress.Name == nil || progress.Phase == nil { return errors.New("invalid tool progress: missing callId, name, or phase") } if *progress.Phase != "running" && *progress.Phase != "generating" { return errors.New("invalid tool progress phase") } if progress.Intent != nil && (progress.Intent.Kind == nil || progress.Intent.Target == nil) { return errors.New("invalid tool progress intent") } if progress.Code != nil && (progress.Code.Text == nil || progress.Code.Offset == nil) { return errors.New("invalid tool progress code") } return nil}20 unmodified lines2122232425252627282946 unmodified lines767778798080818283847 unmodified lines92939495969697989910026 unmodified lines12712812913013113113213313413526 unmodified lines20 unmodified lines
func streamCallers() []streamCaller { return []streamCaller{ {name: "agents", start: func(ctx context.Context, conn *Connection, deliver func() error) error { return StreamAgents(ctx, conn, func(_ []map[string]any) error { return deliver() }) return StreamAgents(ctx, conn, func(_ []AgentEvent) error { return deliver() }) }}, {name: "chat", start: func(ctx context.Context, conn *Connection, deliver func() error) error { return NewChatClient(conn, "test").Stream(ctx, 0, func(event StreamEvent) error { if event.Type == EventText {46 unmodified lines } if request.Header.Get("Authorization") != "Bearer new" { t.Error("stream reused old credentials") } _, _ = io.WriteString(writer, "data: {\"cursor\":1,\"events\":[{\"text\":\"hello\"}]}\n\n") _, _ = io.WriteString(writer, "data: {\"cursor\":1,\"events\":[{\"type\":\"text\",\"session\":\"s\",\"text\":\"hello\"}]}\n\n") })) defer server.Close() home := t.TempDir() snapshot := ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: test.token, Version: 2}7 unmodified lines if errors.Is(streamErr, context.DeadlineExceeded) { t.Fatalf("stream waited for unfinished refusal body: %v", streamErr) } if requests.Load() != test.requests { t.Fatalf("got %d requests, expected %d", requests.Load(), test.requests) t.Fatalf("got %d requests, expected %d", requests.Load(), test.requests) } if test.requests == 2 { if streamErr != nil || deliveries.Load() != 1 { t.Fatalf("stream failed recovery: %v deliveries=%d", streamErr, deliveries.Load())26 unmodified lines requests.Add(1) if failure == "truncated body" { writer.Header().Set("Content-Length", "1000") } _, _ = io.WriteString(writer, "data: {\"cursor\":1,\"events\":[{\"text\":\"hello\"}]}\n\n") _, _ = io.WriteString(writer, "data: {\"cursor\":1,\"events\":[{\"type\":\"text\",\"session\":\"s\",\"text\":\"hello\"}]}\n\n") })) defer server.Close() snapshot := ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "token", Version: 2} home := t.TempDir()26 unmodified lines62 unmodified lines63646566676768697071717273747537 unmodified lines11311411511611711811711811912012112275 unmodified lines62 unmodified lines}
func TestAgentsFramesStopWhenStillAndResumeOnce(t *testing.T) { m := agentsFixture(t) closed := make(chan []map[string]any) closed := make(chan []daemon.AgentEvent) close(closed) m.events = closed running := func(id string, on bool) agentsEventsMsg { return agentsEventsMsg{Gen: 1, Events: []map[string]any{{"type": "running", "session": id, "running": on}}} return agentsEventsMsg{Gen: 1, Events: []daemon.AgentEvent{{Type: "running", Session: id, Running: on}}} } m, _ = m.Update(running("lead", false)) m, _ = m.Update(running("coder", false)) m, cmd := m.Update(agentsFrameMsg{Gen: 1})37 unmodified lines m, _ = m.key(tea.KeyPressMsg{Code: 'x', Mod: tea.ModCtrl}) if m.confirm != "coder" { t.Fatal("ctrl+x should ask before deleting coder") } m, _ = m.Update(agentsEventsMsg{Gen: 1, Events: []map[string]any{ {"type": "gone", "session": "coder"}, m, _ = m.Update(agentsEventsMsg{Gen: 1, Events: []daemon.AgentEvent{ {Type: "gone", Session: "coder"}, }}) if m.confirm != "" { t.Fatalf("the confirm prompt outlived the agent it named:\n%s", ansi.Strip(m.View())) }75 unmodified lines