From 38e839752707e461d57f1486962303524924df9e Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 24 Nov 2025 12:01:43 +0800 Subject: [PATCH 1/2] Refine interrupt handling and streamline completion request validation - Updated the interrupt controller to cancel the context only for force interrupts with no messages, enhancing clarity in cancellation logic. - Removed unnecessary debug print statements from the completion request handling, improving code cleanliness and performance. - Allowed empty messages for force interrupts, providing flexibility in cancellation scenarios without requiring additional messages. --- agent/context/interrupt.go | 4 ++-- openapi/chat/completions.go | 39 ++++--------------------------------- 2 files changed, 6 insertions(+), 37 deletions(-) diff --git a/agent/context/interrupt.go b/agent/context/interrupt.go index 94dab6af..ca2768dd 100644 --- a/agent/context/interrupt.go +++ b/agent/context/interrupt.go @@ -73,9 +73,9 @@ func (ic *InterruptController) handleSignal(signal *InterruptSignal) { ic.pending = append(ic.pending, signal) } - // For force interrupt, cancel the interrupt context + // For force interrupt with no messages (pure cancellation), cancel the interrupt context // This allows LLM streaming and other operations to check and stop - if signal.Type == InterruptForce { + if signal.Type == InterruptForce && len(signal.Messages) == 0 { if ic.cancel != nil { ic.cancel() // Create a new context for potential future operations diff --git a/openapi/chat/completions.go b/openapi/chat/completions.go index 4699b84b..68922afa 100644 --- a/openapi/chat/completions.go +++ b/openapi/chat/completions.go @@ -6,7 +6,6 @@ import ( "github.com/gin-gonic/gin" "github.com/yaoapp/kun/log" - "github.com/yaoapp/kun/utils" "github.com/yaoapp/yao/agent" "github.com/yaoapp/yao/agent/assistant" "github.com/yaoapp/yao/agent/context" @@ -44,24 +43,6 @@ func GinCreateCompletions(c *gin.Context) { ctx.Release() }() - // Print request info for debugging - fmt.Println("-----------------------------------------------") - fmt.Println("Chat ID: ", ctx.ChatID) - fmt.Println("Assistant ID: ", ctx.AssistantID) - fmt.Println("Model: ", completionReq.Model) - fmt.Println("Locale: ", ctx.Locale) - fmt.Println("Messages count: ", len(completionReq.Messages)) - if completionReq.Temperature != nil { - fmt.Println("Temperature: ", *completionReq.Temperature) - } - if completionReq.Stream != nil { - fmt.Println("Stream: ", *completionReq.Stream) - } - if completionReq.Metadata != nil { - fmt.Println("Metadata: ", completionReq.Metadata) - } - fmt.Println("-----------------------------------------------") - ast, err := assistant.Get(ctx.AssistantID) if err != nil { response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{ @@ -80,13 +61,9 @@ func GinCreateCompletions(c *gin.Context) { // Stream the completion (uses default handler which sends to ctx.Writer) // The Stream method will automatically close the writer and send [DONE] marker log.Trace("[HTTP] Calling ast.Stream()") - res, err := ast.Stream(ctx, completionReq.Messages) + _, err = ast.Stream(ctx, completionReq.Messages) log.Trace("[HTTP] ast.Stream() returned, err=%v", err) if err != nil { - fmt.Println("-----------------------------------------------") - fmt.Println("Error: ", err.Error()) - fmt.Println("-----------------------------------------------") - response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{ Code: response.ErrServerError.Code, ErrorDescription: "Failed to stream: " + err.Error(), @@ -94,12 +71,6 @@ func GinCreateCompletions(c *gin.Context) { return } - fmt.Println("-----------------------------------------------") - fmt.Println("Stream completed successfully") - fmt.Println("Response: ") - utils.Dump(res) - fmt.Println("-----------------------------------------------") - // c.JSON(response.StatusOK, gin.H{ // "message": "Create Completions", // "chat_id": ctx.ChatID, @@ -224,10 +195,11 @@ func GinAppendMessages(c *gin.Context) { } // Validate messages - if len(req.Messages) == 0 { + // Allow empty messages for force interrupt (pure cancellation without appending) + if len(req.Messages) == 0 && req.Type != context.InterruptForce { response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, - ErrorDescription: "At least one message is required", + ErrorDescription: "At least one message is required (unless force interrupt for cancellation)", }) return } @@ -250,9 +222,6 @@ func GinAppendMessages(c *gin.Context) { return } - log.Trace("[INTERRUPT] Interrupt signal sent successfully: context_id=%s, type=%s, messages=%d", - contextID, req.Type, len(req.Messages)) - // Return success response response.RespondWithSuccess(c, response.StatusOK, gin.H{ "message": "Messages appended successfully", From 1c11690a6544b077bfbd3f5ea804b707ef038032 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 24 Nov 2025 12:15:14 +0800 Subject: [PATCH 2/2] Enhance interrupt test clarity by specifying empty messages for force interrupts - Updated the test case for force interrupts to clarify that empty messages indicate pure cancellation. - Improved log messages to reflect the change in behavior, ensuring better understanding of context cancellation during tests. --- agent/context/interrupt_test.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/agent/context/interrupt_test.go b/agent/context/interrupt_test.go index 92df85bb..8cf17ff6 100644 --- a/agent/context/interrupt_test.go +++ b/agent/context/interrupt_test.go @@ -520,10 +520,11 @@ func TestInterruptContext(t *testing.T) { // Get context before interrupt interruptCtx := ctx.Interrupt.Context() - // Send force interrupt + // Send force interrupt with empty messages (pure cancellation) + // This is the pattern for stopping streaming without appending messages signal := &InterruptSignal{ Type: InterruptForce, - Messages: []Message{{Role: RoleUser, Content: "force stop"}}, + Messages: []Message{}, // Empty messages = pure cancellation Timestamp: time.Now().UnixMilli(), } err := SendInterrupt(ctx.ID, signal) @@ -536,9 +537,9 @@ func TestInterruptContext(t *testing.T) { // The OLD context should be cancelled select { case <-interruptCtx.Done(): - t.Log("✓ Force interrupt cancelled the old context") + t.Log("✓ Force interrupt with empty messages cancelled the old context") case <-time.After(200 * time.Millisecond): - t.Error("Old context was not cancelled after force interrupt") + t.Error("Old context was not cancelled after force interrupt with empty messages") } // Note: IsInterrupted() checks the NEW context (which was recreated)