diff --git a/agent/assistant/handlers/stream.go b/agent/assistant/handlers/stream.go index 639a3002..ce28fe70 100644 --- a/agent/assistant/handlers/stream.go +++ b/agent/assistant/handlers/stream.go @@ -1,6 +1,9 @@ package handlers import ( + "fmt" + "time" + jsoniter "github.com/json-iterator/go" "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/context" @@ -15,9 +18,10 @@ func DefaultStreamHandler(ctx *context.Context) message.StreamFunc { // Create stream state manager state := &streamState{ - ctx: ctx, - inGroup: false, - currentID: "", + ctx: ctx, + inGroup: false, + currentGroupID: "", + messageSeq: 0, } return func(chunkType message.StreamChunkType, data []byte) int { @@ -64,11 +68,14 @@ func DefaultStreamHandler(ctx *context.Context) message.StreamFunc { // streamState manages the state of the streaming process type streamState struct { - ctx *context.Context - inGroup bool - currentID string - currentType string // Track the current message type (text, thinking, tool_call) - buffer []byte + ctx *context.Context + inGroup bool + currentGroupID string // Current group ID (shared by all chunks in the group) + currentType string // Track the current message type (text, thinking, tool_call) + buffer []byte + chunkCount int // Track number of chunks in current group + messageSeq int // Message sequence number (for generating readable IDs) + groupStartTime time.Time // Track when group started } // handleStreamStart handles stream start event @@ -87,9 +94,32 @@ func (s *streamState) handleStreamStart(data []byte) int { // handleGroupStart handles group start event func (s *streamState) handleGroupStart(data []byte) int { + // Parse group start data first to get the group ID + var startData message.GroupStartData + if err := jsoniter.Unmarshal(data, &startData); err != nil { + log.Error("Failed to unmarshal group start data: %v", err) + return 0 + } + + // Use the group ID from the start data, or generate one if not provided + groupID := startData.GroupID + if groupID == "" { + groupID = generateMessageID() + startData.GroupID = groupID + } + + // Initialize group state with the correct group ID s.inGroup = true - s.currentID = generateMessageID() + s.currentGroupID = groupID s.buffer = []byte{} + s.chunkCount = 0 + s.messageSeq = 0 // Reset message sequence for each group + s.groupStartTime = time.Now() + + // Send group_start event + msg := output.NewEventMessage(message.EventGroupStart, "Group started", startData) + s.ctx.Send(msg) + return 0 // Continue } @@ -99,22 +129,22 @@ func (s *streamState) handleText(data []byte) int { return 0 } - // Ensure we have a message ID - if s.currentID == "" { - s.currentID = generateMessageID() - } - // Track current message type s.currentType = message.TypeText // Append to buffer s.buffer = append(s.buffer, data...) + s.chunkCount++ + s.messageSeq++ // Send delta message + // - ID: Sequential message ID (e.g., msg_001, msg_002) for readability + // - GroupID: Same for all chunks of this logical message (frontend merges by group_id) msg := &message.Message{ - ID: s.currentID, - Type: message.TypeText, - Delta: true, + ID: s.generateSequentialID(), // Sequential ID for this chunk + GroupID: s.currentGroupID, // Group ID for merging (all chunks share this) + Type: message.TypeText, + Delta: true, Props: map[string]interface{}{ "content": string(data), }, @@ -134,22 +164,22 @@ func (s *streamState) handleThinking(data []byte) int { return 0 } - // Ensure we have a message ID - if s.currentID == "" { - s.currentID = generateMessageID() - } - // Track current message type s.currentType = message.TypeThinking // Append to buffer s.buffer = append(s.buffer, data...) + s.chunkCount++ + s.messageSeq++ // Send delta message + // - ID: Sequential message ID (e.g., msg_001, msg_002) for readability + // - GroupID: Same for all chunks of this logical message (frontend merges by group_id) msg := &message.Message{ - ID: s.currentID, - Type: message.TypeThinking, - Delta: true, + ID: s.generateSequentialID(), // Sequential ID for this chunk + GroupID: s.currentGroupID, // Group ID for merging (all chunks share this) + Type: message.TypeThinking, + Delta: true, Props: map[string]interface{}{ "content": string(data), }, @@ -202,30 +232,38 @@ func (s *streamState) handleGroupEnd(data []byte) int { return 0 } - // Send done message with complete content - if s.currentID != "" && len(s.buffer) > 0 { - // Use the tracked message type (thinking, text, tool_call, etc.) - msgType := s.currentType - if msgType == "" { - msgType = message.TypeText // Fallback to text if type not set - } + // Calculate duration + durationMs := time.Since(s.groupStartTime).Milliseconds() - msg := &message.Message{ - ID: s.currentID, - Type: msgType, // Use the actual message type from the group - Done: true, - Props: map[string]interface{}{ - "content": string(s.buffer), - }, - } - s.ctx.Send(msg) + // Use the tracked message type (thinking, text, tool_call, etc.) + msgType := s.currentType + if msgType == "" { + msgType = message.TypeText // Fallback to text if type not set } + // Build GroupEndData with complete content + endData := message.GroupEndData{ + GroupID: s.currentGroupID, // Use the group ID, not message ID + Type: msgType, + Timestamp: time.Now().UnixMilli(), + DurationMs: durationMs, + ChunkCount: s.chunkCount, + Status: "completed", + Extra: map[string]interface{}{ + "content": string(s.buffer), // Include complete content in the event + }, + } + + // Send group_end event + msg := output.NewEventMessage(message.EventGroupEnd, "Group completed", endData) + s.ctx.Send(msg) + // Reset state s.inGroup = false - s.currentID = "" + s.currentGroupID = "" s.currentType = "" s.buffer = []byte{} + s.chunkCount = 0 return 0 // Continue } @@ -249,6 +287,13 @@ func (s *streamState) handleStreamEnd(data []byte) int { return 0 // Continue (stream will end naturally) } +// generateSequentialID generates a sequential message ID for better readability +func (s *streamState) generateSequentialID() string { + // Format: 1, 2, 3, etc. + // This makes it easier for developers to track message order in logs + return fmt.Sprintf("%d", s.messageSeq) +} + // generateMessageID generates a unique message ID func generateMessageID() string { // TODO: Implement proper ID generation diff --git a/agent/context/JSAPI_OUTPUT.md b/agent/context/JSAPI_OUTPUT.md index 91abdfb2..f556e0fe 100644 --- a/agent/context/JSAPI_OUTPUT.md +++ b/agent/context/JSAPI_OUTPUT.md @@ -1,6 +1,6 @@ # Context Output JS API -The Context object provides `Send`, `SendGroup`, and `Flush` methods for sending messages to clients from JavaScript within Agent Hook functions. +The Context object provides `Send`, `SendGroup`, `SendGroupStart`, and `SendGroupEnd` methods for sending messages to clients from JavaScript within Agent Hook functions. ## Hook Functions Overview @@ -23,16 +23,14 @@ Agent Hook functions are lifecycle callbacks that allow you to customize the beh * Create hook - send initial messages to client */ function Create(ctx, messages) { - // Send welcome message (string shorthand) + // Send welcome message (string shorthand, auto-flushes) ctx.Send("Welcome! Let me help you with that..."); - ctx.Flush(); - // Send loading indicator + // Send loading indicator (auto-flushes) ctx.Send({ type: "loading", props: { message: "Analyzing your request..." }, }); - ctx.Flush(); // Continue with normal processing return { messages }; @@ -154,13 +152,43 @@ ctx.SendGroup({ }); ``` -### ctx.Flush() +### ctx.SendGroupStart(type?, id?) -Flush the output buffer to ensure messages are sent immediately. +Start a message group and return the group ID. Messages sent after this should include the returned `group_id`. + +**Parameters:** + +- `type` (optional): Group type (`"text"`, `"thinking"`, `"tool_call"`, `"mixed"`), defaults to `"mixed"` +- `id` (optional): Custom group ID, auto-generates if not provided + +**Returns:** Group ID (string) ```javascript -ctx.Send("Processing..."); -ctx.Flush(); // Send immediately to client +// Auto-generate ID with default type +const groupId = ctx.SendGroupStart(); + +// Specify type, auto-generate ID +const groupId = ctx.SendGroupStart("text"); + +// Specify both type and custom ID +const groupId = ctx.SendGroupStart("thinking", "my-group-123"); +``` + +### ctx.SendGroupEnd(id, chunkCount?) + +End a message group. + +**Parameters:** + +- `id` (required): Group ID returned from `SendGroupStart` +- `chunkCount` (optional): Number of messages in the group + +```javascript +// Basic usage +ctx.SendGroupEnd(groupId); + +// With chunk count +ctx.SendGroupEnd(groupId, 5); ``` ## Complete Hook Examples @@ -172,9 +200,8 @@ ctx.Flush(); // Send immediately to client * Send welcome message when conversation starts */ function Create(ctx, messages) { - // Send welcome message + // Send welcome message (auto-flushes) ctx.Send("Welcome to AI Assistant! How can I help you today?"); - ctx.Flush(); // Return messages to continue processing return { messages }; @@ -409,16 +436,15 @@ function Create(ctx, messages) { } ``` -### 9. Message Groups for Related Content +### 9. Message Groups for Related Content (High-level API) ```javascript /** - * Send groups of related messages together + * Send groups of related messages together using SendGroup (auto-handles events) */ function Before(ctx, messages, response) { - // Send a group of context information + // SendGroup automatically sends group_start and group_end events ctx.SendGroup({ - id: "context_info", messages: [ { type: "text", @@ -442,12 +468,86 @@ function Before(ctx, messages, response) { type: "context", }, }); - ctx.Flush(); return { response }; } ``` +### 10. Manual Group Control (Low-level API) + +```javascript +/** + * Manually control group boundaries with SendGroupStart and SendGroupEnd + */ +function Create(ctx, messages) { + // Start a text group + const groupId = ctx.SendGroupStart("text"); + + // Send messages with group_id + ctx.Send({ + type: "text", + props: { content: "First message in group" }, + group_id: groupId, + }); + + ctx.Send({ + type: "text", + props: { content: "Second message in group" }, + group_id: groupId, + }); + + // End the group + ctx.SendGroupEnd(groupId, 2); + + return { messages }; +} +``` + +### 11. Streaming with Groups + +```javascript +/** + * Stream delta updates within a group + */ +function Create(ctx, messages) { + // Start thinking group + const thinkingId = ctx.SendGroupStart("thinking"); + + // Stream thinking process + const steps = ["Analyzing", "Processing", "Generating"]; + const msgId = "thinking_msg"; + + steps.forEach((step, i) => { + if (i === 0) { + // First message + ctx.Send({ + type: "thinking", + props: { content: step }, + id: msgId, + group_id: thinkingId, + delta: false, + }); + } else { + // Delta updates + ctx.Send({ + type: "thinking", + props: { content: ` → ${step}` }, + id: msgId, + group_id: thinkingId, + delta: true, + delta_path: "content", + delta_action: "append", + }); + } + }); + + // End thinking group + ctx.SendGroupEnd(thinkingId, steps.length); + + return { messages }; +} +``` + ## Message Types Built-in message types supported: @@ -617,9 +717,8 @@ ctx.Send({ ```javascript function Create(ctx, messages) { - ctx.Send("Starting processing..."); - ctx.Flush(); - // No need to wait, continue processing + ctx.Send("Starting processing..."); // Auto-flushes + // Continue processing immediately return { messages }; } ``` @@ -630,8 +729,7 @@ function Create(ctx, messages) { function Create(ctx, messages) { const stages = ["validate", "analyze", "prepare"]; stages.forEach((stage) => { - ctx.Send({ type: "loading", props: { message: `${stage}...` } }); - ctx.Flush(); + ctx.Send({ type: "loading", props: { message: `${stage}...` } }); // Auto-flushes performStage(stage); }); return { messages }; @@ -647,8 +745,7 @@ function Before(ctx, messages, response) { ctx.Send({ type: "thinking", props: { content: "Analyzing complex query..." }, - }); - ctx.Flush(); + }); // Auto-flushes } return { response }; } @@ -659,8 +756,7 @@ function Before(ctx, messages, response) { ```javascript function Error(ctx, messages, error) { if (error.code === "RATE_LIMIT") { - ctx.Send("Service is busy, retrying..."); - ctx.Flush(); + ctx.Send("Service is busy, retrying..."); // Auto-flushes time.Sleep(1000); return { retry: true }; } @@ -668,8 +764,7 @@ function Error(ctx, messages, error) { ctx.Send({ type: "error", props: { message: "Sorry, something went wrong.", code: error.code }, - }); - ctx.Flush(); + }); // Auto-flushes return { error }; } ``` @@ -686,16 +781,16 @@ Each hook receives different parameters: - `Done(ctx, messages, response)` - Context, messages, and final response - `Error(ctx, messages, error)` - Context, messages, and error object -### 2. Always Flush for Real-time Updates +### 2. Messages Auto-Flush for Real-time Updates ```javascript -// Good - user sees message immediately -ctx.Send("Processing..."); -ctx.Flush(); +// Messages are automatically flushed after each Send +ctx.Send("Processing..."); // Sent immediately to client -// Bad - message buffered until hook returns -ctx.Send("Processing..."); -// ... hook continues ... +// Multiple sends work seamlessly +ctx.Send("Step 1"); // Flushed +ctx.Send("Step 2"); // Flushed +ctx.Send("Step 3"); // Flushed ``` ### 3. Delta Updates Require Unique IDs @@ -722,9 +817,10 @@ ctx.Send({ ### 5. Performance Considerations -- Use `Flush()` sparingly - only when immediate delivery is needed -- Batch related messages with `SendGroup()` when possible -- Avoid sending too many small updates (combine them) +- Messages auto-flush after each `Send()` for real-time delivery +- Batch related messages with `SendGroup()` when possible for better performance +- Avoid sending too many small updates (combine them when feasible) +- Use `SendGroupStart`/`SendGroupEnd` for fine-grained control over grouping ### 6. Context Information Available @@ -757,9 +853,8 @@ function Create(ctx, messages) { ```javascript function Create(ctx, messages) { - ctx.Send("Hello"); - ctx.SendGroup({ id: "grp1", messages: [...] }); - ctx.Flush(); + ctx.Send("Hello"); // Auto-flushes + ctx.SendGroup({ messages: [...] }); // Auto-handles events and flushing return { messages }; } ``` @@ -768,17 +863,24 @@ function Create(ctx, messages) { 1. **Use String Shorthand**: `ctx.Send("Hello")` is simpler than `ctx.Send({ type: "text", props: { content: "Hello" } })` -2. **Flush After Each Step**: Ensure users see progress in real-time +2. **Messages Auto-Flush**: Each `Send()` automatically flushes for real-time delivery - no manual flushing needed -3. **Handle Errors Gracefully**: Always provide user-friendly error messages +3. **Choose the Right API Level**: -4. **Show Progress for Long Operations**: Use loading indicators for better UX + - **High-level**: Use `SendGroup()` for simple grouped messages (auto-handles events) + - **Low-level**: Use `SendGroupStart()`/`SendGroupEnd()` for fine-grained control -5. **Return Hook Results**: Always return required objects from hooks: +4. **Handle Errors Gracefully**: Always provide user-friendly error messages + +5. **Show Progress for Long Operations**: Use loading indicators for better UX + +6. **Return Hook Results**: Always return required objects from hooks: - `Create`: `{ messages }` - `Before/After`: `{ response }` - `Done`: `{}` or `{ response }` - `Error`: `{ error }` or `{ retry: true }` -6. **Test with Different Clients**: Verify behavior with both OpenAI and CUI clients +7. **Test with Different Clients**: Verify behavior with both OpenAI and CUI clients + +8. **Group Related Messages**: Use groups to organize related content for better frontend rendering diff --git a/agent/context/jsapi.go b/agent/context/jsapi.go index 9c42713b..59528bff 100644 --- a/agent/context/jsapi.go +++ b/agent/context/jsapi.go @@ -1,7 +1,11 @@ package context import ( + "time" + "github.com/yaoapp/gou/runtime/v8/bridge" + "github.com/yaoapp/yao/agent/output" + "github.com/yaoapp/yao/agent/output/message" traceJsapi "github.com/yaoapp/yao/trace/jsapi" "rogchap.com/v8go" ) @@ -47,7 +51,8 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) { jsObject.Set("Trace", ctx.traceMethod(v8ctx.Isolate())) jsObject.Set("Send", ctx.sendMethod(v8ctx.Isolate())) jsObject.Set("SendGroup", ctx.sendGroupMethod(v8ctx.Isolate())) - jsObject.Set("Flush", ctx.flushMethod(v8ctx.Isolate())) + jsObject.Set("SendGroupStart", ctx.sendGroupStartMethod(v8ctx.Isolate())) + jsObject.Set("SendGroupEnd", ctx.sendGroupEndMethod(v8ctx.Isolate())) // Create instance instance, err := jsObject.NewInstance(v8ctx) @@ -164,6 +169,7 @@ func (ctx *Context) traceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { // sendMethod implements ctx.Send(message) // Usage: ctx.Send({ type: "text", props: { content: "Hello" } }) // Usage: ctx.Send("Hello") // shorthand for text message +// Automatically generates ID and flushes output func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { v8ctx := info.Context() @@ -179,17 +185,28 @@ func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return bridge.JsException(v8ctx, "invalid message: "+err.Error()) } + // Generate unique ID if not provided + if msg.ID == "" { + msg.ID = output.GenerateID() + } + // Call ctx.Send if err := ctx.Send(msg); err != nil { return bridge.JsException(v8ctx, "Send failed: "+err.Error()) } + // Automatically flush after sending + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed: "+err.Error()) + } + return v8go.Undefined(iso) }) } // sendGroupMethod implements ctx.SendGroup(group) // Usage: ctx.SendGroup({ id: "group1", messages: [...] }) +// Automatically generates IDs, sends group_start/group_end events, and flushes output func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { v8ctx := info.Context() @@ -205,24 +222,159 @@ func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return bridge.JsException(v8ctx, "invalid group: "+err.Error()) } + // Generate group ID if not provided + if group.ID == "" { + group.ID = output.GenerateID() + } + + // Send group_start event + startTime := time.Now() + startEvent := output.NewEventMessage( + message.EventGroupStart, + "Group started", + message.GroupStartData{ + GroupID: group.ID, + Type: "mixed", // Mixed types in group + Timestamp: startTime.UnixMilli(), + }, + ) + if err := ctx.Send(startEvent); err != nil { + return bridge.JsException(v8ctx, "Failed to send group_start event: "+err.Error()) + } + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed after group_start: "+err.Error()) + } + + // Generate IDs for messages and set group_id + for _, msg := range group.Messages { + if msg.ID == "" { + msg.ID = output.GenerateID() + } + if msg.GroupID == "" { + msg.GroupID = group.ID + } + } + // Call ctx.SendGroup if err := ctx.SendGroup(group); err != nil { return bridge.JsException(v8ctx, "SendGroup failed: "+err.Error()) } - - return v8go.Undefined(iso) - }) -} - -// flushMethod implements ctx.Flush() -// Usage: ctx.Flush() -func (ctx *Context) flushMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { - return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { - v8ctx := info.Context() - - // Call ctx.Flush if err := ctx.Flush(); err != nil { - return bridge.JsException(v8ctx, "Flush failed: "+err.Error()) + return bridge.JsException(v8ctx, "Flush failed after SendGroup: "+err.Error()) + } + + // Send group_end event + endEvent := output.NewEventMessage( + message.EventGroupEnd, + "Group completed", + message.GroupEndData{ + GroupID: group.ID, + Type: "mixed", + Timestamp: time.Now().UnixMilli(), + DurationMs: time.Since(startTime).Milliseconds(), + ChunkCount: len(group.Messages), + Status: "completed", + }, + ) + if err := ctx.Send(endEvent); err != nil { + return bridge.JsException(v8ctx, "Failed to send group_end event: "+err.Error()) + } + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed after group_end: "+err.Error()) + } + + return v8go.Undefined(iso) + }) +} + +// sendGroupStartMethod implements ctx.SendGroupStart(type?, id?) +// Usage: const groupId = ctx.SendGroupStart() // type="mixed", auto-generate ID +// Usage: const groupId = ctx.SendGroupStart("text") // type="text", auto-generate ID +// Usage: const groupId = ctx.SendGroupStart("text", "my-group-id") // type="text", use provided ID +// Returns the group ID (generated or provided) +func (ctx *Context) sendGroupStartMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + // Get type (default: "mixed") + groupType := "mixed" + if len(args) > 0 && args[0].IsString() { + groupType = args[0].String() + } + + // Get or generate group ID + var groupID string + if len(args) > 1 && args[1].IsString() { + groupID = args[1].String() + } else { + groupID = output.GenerateID() + } + + // Send group_start event + startEvent := output.NewEventMessage( + message.EventGroupStart, + "Group started", + message.GroupStartData{ + GroupID: groupID, + Type: groupType, + Timestamp: time.Now().UnixMilli(), + }, + ) + if err := ctx.Send(startEvent); err != nil { + return bridge.JsException(v8ctx, "Failed to send group_start event: "+err.Error()) + } + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed after group_start: "+err.Error()) + } + + // Return the group ID + groupIDVal, err := v8go.NewValue(iso, groupID) + if err != nil { + return bridge.JsException(v8ctx, "Failed to create return value: "+err.Error()) + } + return groupIDVal + }) +} + +// sendGroupEndMethod implements ctx.SendGroupEnd(id, chunkCount?) +// Usage: ctx.SendGroupEnd(groupId) +// Usage: ctx.SendGroupEnd(groupId, 10) // With chunk count +func (ctx *Context) sendGroupEndMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + // Group ID is required + if len(args) < 1 || !args[0].IsString() { + return bridge.JsException(v8ctx, "SendGroupEnd requires a group ID (string) as first argument") + } + groupID := args[0].String() + + // Optional chunk count + chunkCount := 0 + if len(args) > 1 && args[1].IsNumber() { + chunkCount = int(args[1].Integer()) + } + + // Send group_end event + endEvent := output.NewEventMessage( + message.EventGroupEnd, + "Group completed", + message.GroupEndData{ + GroupID: groupID, + Type: "mixed", + Timestamp: time.Now().UnixMilli(), + DurationMs: 0, // Duration not tracked at this level + ChunkCount: chunkCount, + Status: "completed", + }, + ) + if err := ctx.Send(endEvent); err != nil { + return bridge.JsException(v8ctx, "Failed to send group_end event: "+err.Error()) + } + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed after group_end: "+err.Error()) } return v8go.Undefined(iso) diff --git a/agent/context/jsapi_helpers.go b/agent/context/jsapi_helpers.go index 0563be9c..c70708ae 100644 --- a/agent/context/jsapi_helpers.go +++ b/agent/context/jsapi_helpers.go @@ -58,9 +58,6 @@ func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, e if delta, ok := msgMap["delta"].(bool); ok { msg.Delta = delta } - if done, ok := msgMap["done"].(bool); ok { - msg.Done = done - } if deltaPath, ok := msgMap["delta_path"].(string); ok { msg.DeltaPath = deltaPath } @@ -73,12 +70,6 @@ func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, e if groupID, ok := msgMap["group_id"].(string); ok { msg.GroupID = groupID } - if groupStart, ok := msgMap["group_start"].(bool); ok { - msg.GroupStart = groupStart - } - if groupEnd, ok := msgMap["group_end"].(bool); ok { - msg.GroupEnd = groupEnd - } // Metadata (optional) if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { @@ -158,9 +149,6 @@ func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error if delta, ok := msgMap["delta"].(bool); ok { msg.Delta = delta } - if done, ok := msgMap["done"].(bool); ok { - msg.Done = done - } if deltaPath, ok := msgMap["delta_path"].(string); ok { msg.DeltaPath = deltaPath } @@ -173,12 +161,6 @@ func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error if groupID, ok := msgMap["group_id"].(string); ok { msg.GroupID = groupID } - if groupStart, ok := msgMap["group_start"].(bool); ok { - msg.GroupStart = groupStart - } - if groupEnd, ok := msgMap["group_end"].(bool); ok { - msg.GroupEnd = groupEnd - } // Metadata (optional) if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { diff --git a/agent/context/jsapi_output_test.go b/agent/context/jsapi_output_test.go index 98d09ba9..99f1a824 100644 --- a/agent/context/jsapi_output_test.go +++ b/agent/context/jsapi_output_test.go @@ -163,8 +163,8 @@ func TestJsValueSendGroup(t *testing.T) { assert.Equal(t, true, result["success"], "SendGroup should succeed") } -// TestJsValueFlush test the Flush method on Context -func TestJsValueFlush(t *testing.T) { +// TestJsValueSendGroupStartEnd test the SendGroupStart and SendGroupEnd methods +func TestJsValueSendGroupStartEnd(t *testing.T) { test.Prepare(t, config.Conf) defer test.Clean() @@ -181,13 +181,17 @@ func TestJsValueFlush(t *testing.T) { res, err := v8.Call(v8.CallOptions{}, ` function test(ctx) { try { - // Send a message - ctx.Send("Processing..."); + // Start a group with auto-generated ID + const groupId = ctx.SendGroupStart("text"); - // Flush output - ctx.Flush(); + // Send messages in the group + ctx.Send({ type: "text", props: { content: "Message 1" }, group_id: groupId }); + ctx.Send({ type: "text", props: { content: "Message 2" }, group_id: groupId }); - return { success: true }; + // End the group + ctx.SendGroupEnd(groupId, 2); + + return { success: true, groupId: groupId }; } catch (error) { return { success: false, error: error.message }; } @@ -200,7 +204,8 @@ func TestJsValueFlush(t *testing.T) { if !ok { t.Fatalf("Expected map result, got %T", res) } - assert.Equal(t, true, result["success"], "Flush should succeed") + assert.Equal(t, true, result["success"], "SendGroupStart/End should succeed") + assert.NotEmpty(t, result["groupId"], "Should return group ID") } // TestJsValueSendDeltaUpdates test delta updates in Send @@ -239,13 +244,7 @@ func TestJsValueSendDeltaUpdates(t *testing.T) { delta_action: "append" }); - // Mark as complete - ctx.Send({ - type: "text", - props: {}, - id: "msg_1", - done: true - }); + // Send completion (no done field needed) return { success: true }; } catch (error) { @@ -329,8 +328,6 @@ func TestJsValueSendMultipleTypes(t *testing.T) { } }); - ctx.Flush(); - return { success: true }; } catch (error) { return { success: false, error: error.message }; @@ -467,7 +464,6 @@ func TestJsValueSendWithCUIAccept(t *testing.T) { type: "text", props: { content: "Hello CUI" } }); - ctx.Flush(); return { success: true }; } catch (error) { return { success: false, error: error.message }; @@ -566,15 +562,11 @@ func TestJsValueSendChainedCalls(t *testing.T) { res, err := v8.Call(v8.CallOptions{}, ` function test(ctx) { try { - // Multiple sequential sends + // Multiple sequential sends (each auto-flushes) ctx.Send("Step 1"); ctx.Send("Step 2"); ctx.Send("Step 3"); - ctx.Flush(); - - // Send after flush ctx.Send("Step 4"); - ctx.Flush(); return { success: true }; } catch (error) { diff --git a/agent/output/README.md b/agent/output/README.md index b3dcb631..3dc8414a 100644 --- a/agent/output/README.md +++ b/agent/output/README.md @@ -39,7 +39,6 @@ type Message struct { // Streaming control ID string `json:"id,omitempty"` // Unique message ID (for merging in streaming) Delta bool `json:"delta,omitempty"` // Whether this is an incremental update - Done bool `json:"done,omitempty"` // Whether the message is complete // Delta update control (for incremental props updates) DeltaPath string `json:"delta_path,omitempty"` // Which field to update (e.g., "content", "items.0.name") @@ -85,11 +84,7 @@ type Message struct { - `true`: Append/update to existing message with same ID - `false`: Complete message (default) - Used for streaming LLM responses - -- **`Done`** (optional): Marks message as complete - - `true`: No more updates will come for this message ID - - `false`: More updates may follow - - Typically sent as final message in a delta sequence + - Message completion is signaled via `group_end` event instead #### Delta Update Control @@ -379,11 +374,22 @@ msg := &message.Message{ } output.Send(ctx, msg) -// Mark as complete -msg.Delta = false -msg.Done = true -msg.Props["content"] = "Hello world!" // Full content +// Send more delta updates... +msg.Props["content"] = " world" output.Send(ctx, msg) + +// Mark completion with group_end event +endData := message.GroupEndData{ + GroupID: "msg_123", + Type: "text", + Status: "completed", + ChunkCount: 2, + Extra: map[string]interface{}{ + "content": "Hello world!", // Full content + }, +} +eventMsg := output.NewEventMessage(message.EventGroupEnd, "Group completed", endData) +output.Send(ctx, eventMsg) ``` ### Custom Writers diff --git a/agent/output/adapters/openai/adapter.go b/agent/output/adapters/openai/adapter.go index 08e9f281..3190ae79 100644 --- a/agent/output/adapters/openai/adapter.go +++ b/agent/output/adapters/openai/adapter.go @@ -90,13 +90,6 @@ func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) { return []interface{}{}, nil // Return empty array, nothing to send } - // Skip "done" messages that are not delta updates - // These are final confirmation messages for CUI clients - // OpenAI clients don't need them - they use finish_reason instead - if msg.Done && !msg.Delta { - return []interface{}{}, nil // Return empty array, nothing to send - } - // Get converter for this message type converter, exists := a.registry.GetConverter(msg.Type) if !exists { diff --git a/agent/output/message/types.go b/agent/output/message/types.go index 227906ad..b80a716f 100644 --- a/agent/output/message/types.go +++ b/agent/output/message/types.go @@ -37,9 +37,8 @@ type Message struct { Props map[string]interface{} `json:"props,omitempty"` // Message properties (passed to frontend component) // Streaming control - ID string `json:"id,omitempty"` // Message ID (used for merging messages in streaming scenarios) + ID string `json:"id,omitempty"` // Unique chunk/message ID (each chunk has unique ID; use group_id for merging) Delta bool `json:"delta,omitempty"` // Whether this is an incremental update - Done bool `json:"done,omitempty"` // Whether the message is complete // Delta update control DeltaPath string `json:"delta_path,omitempty"` // Update path (e.g., "content", "data", "items.0.name") @@ -49,9 +48,7 @@ type Message struct { TypeChange bool `json:"type_change,omitempty"` // Marks this as a type correction message // Message group - GroupID string `json:"group_id,omitempty"` // Parent message group ID - GroupStart bool `json:"group_start,omitempty"` // Marks the start of a message group - GroupEnd bool `json:"group_end,omitempty"` // Marks the end of a message group + GroupID string `json:"group_id,omitempty"` // Group ID (all delta chunks of same logical message share this; used for merging) // Metadata Metadata *Metadata `json:"metadata,omitempty"` // Additional metadata