agent-exec transport 4838630332e6bb1220f0e63e947ddb81/599831d96200

runtime/llm/sse.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"strings"
 )
@@ -42,6 +43,7 @@ type usageWire struct {
 const (
 	maxObservedToolFragments     = 10000
 	maxObservedToolArgumentBytes = 4*1024*1024 + 1
+	maxObservedTailEvents        = 10000
 )
 
 // ParseSSE reads an OpenAI-compatible server-sent event body and emits
@@ -68,7 +70,14 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 	lastFrameJSONComplete := false
 	toolFragments := 0
 	toolArgumentBytes := 0
+	sc := bufio.NewScanner(r)
+	sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
+	reader := &sseEventReader{sc: sc}
 	observation := func() *TerminalObservation {
+		tailKind := reader.tailKind
+		if tailKind == "" {
+			tailKind = "unknown"
+		}
 		argumentsValid := len(order) > 0
 		for _, idx := range order {
 			var args map[string]json.RawMessage
@@ -81,6 +90,7 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 			ReadOutcome: readOutcome, LastFrameJSONComplete: lastFrameJSONComplete,
 			UsagePresent: sawUsage, ToolFragments: toolFragments,
 			ToolArgumentBytes: toolArgumentBytes, ToolArgumentsValid: argumentsValid,
+			TailEventKind: tailKind, TailEventCount: reader.tailCount,
 		}
 	}
 	sendTerminalFailure := func(why string) {
@@ -97,9 +107,6 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 		case out <- failed:
 		}
 	}
-	sc := bufio.NewScanner(r)
-	sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
-	reader := &sseEventReader{sc: sc}
 	for {
 		data, ok := reader.next()
 		if !ok {
@@ -111,12 +118,14 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 			return
 		}
 		if data == "[DONE]" {
+			reader.noteTail("done_literal")
 			sawDone = true
 			readOutcome = "done_marker"
 			break
 		}
 		var chunk sseChunk
 		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			reader.noteTail(badFrameKind(data, err))
 			readOutcome = "invalid_frame"
 			lastFrameJSONComplete = false
 			if sawUsage {
@@ -126,6 +135,7 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 			emitError(out, content.String(), false, "malformed stream chunk: "+err.Error(), observation())
 			return
 		}
+		reader.noteTail("json_frame")
 		lastFrameJSONComplete = true
 		if chunk.Error != nil && chunk.Error.Message != "" {
 			readOutcome = "provider_error"
@@ -300,9 +310,38 @@ func terminalStreamFailure(message Message, usage Usage, reason string, complete
 
 // sseEventReader 按空行切事件，并把同一事件里的多行 data: 用换行拼回。
 // 已经是完整 JSON 的 data 行单独成事件，这样单行流不用空行也能解析。
+// tailKind/tailCount 只记最后完成的结构事件类别与饱和计数，不存帧内容。
 type sseEventReader struct {
-	sc    *bufio.Scanner
-	carry string
+	sc        *bufio.Scanner
+	carry     string
+	tailKind  string
+	tailCount int
+}
+
+func (r *sseEventReader) noteTail(kind string) {
+	if r.tailCount < maxObservedTailEvents {
+		r.tailCount++
+	}
+	r.tailKind = kind
+}
+
+// badFrameKind classifies a data event that did not decode as a chunk using
+// only the decoder's error category and position, never its text or bytes.
+func badFrameKind(data string, err error) string {
+	if strings.TrimSpace(data) == "" {
+		return "empty_data"
+	}
+	var syntax *json.SyntaxError
+	if !errors.As(err, &syntax) {
+		return "unknown"
+	}
+	if syntax.Offset == int64(len(data)) {
+		var probe json.RawMessage
+		if errors.Is(json.NewDecoder(strings.NewReader(data)).Decode(&probe), io.ErrUnexpectedEOF) {
+			return "partial_frame"
+		}
+	}
+	return "unknown_non_json"
 }
 
 func (r *sseEventReader) next() (string, bool) {
@@ -331,6 +370,7 @@ func (r *sseEventReader) next() (string, bool) {
 		}
 		trimmed := strings.TrimSpace(line)
 		if strings.HasPrefix(trimmed, ":") {
+			r.noteTail("comment")
 			continue
 		}
 		if !strings.HasPrefix(trimmed, "data:") {