mirror of
https://github.com/Ed1s0nZ/CyberStrikeAI.git
synced 2026-10-21 23:01:37 +02:00
fix: preserve task cancellation through final events
This commit is contained in:
1 parent
07066f9219
commit
054a77cdab
7 files changed
+133
-4
No files matched your search
@@ -0,0 +1,79 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"cyberstrike-ai/internal/multiagent"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCancelledAgentTaskCannotFinishSuccessfully(t *testing.T) {
|
||||
for _, lateStatus := range []string{"running", "completed", "failed"} {
|
||||
m := NewAgentTaskManager()
|
||||
ctx, cancel := context.WithCancelCause(context.Background())
|
||||
task, err := m.StartTask("conversation", "test", cancel)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ok, err := m.CancelTask("conversation", ErrTaskCancelled); err != nil || !ok {
|
||||
t.Fatalf("cancel: %v %v", ok, err)
|
||||
}
|
||||
if context.Cause(ctx) != ErrTaskCancelled {
|
||||
t.Fatal("cancellation not delivered")
|
||||
}
|
||||
m.UpdateTaskStatus("conversation", lateStatus)
|
||||
if err := m.FinishTaskRun("conversation", task.RunID, "completed"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := m.GetCompletedTasks()
|
||||
if len(done) != 1 || done[0].Status != "cancelled" {
|
||||
t.Fatalf("cancelled task became successful: %#v", done)
|
||||
}
|
||||
if len(m.GetActiveTasks()) != 0 {
|
||||
t.Fatal("cancelled task remains active")
|
||||
}
|
||||
}
|
||||
}
|
||||
func TestInterruptContinueDoesNotBecomeTaskCancellation(t *testing.T) {
|
||||
m := NewAgentTaskManager()
|
||||
_, cancel := context.WithCancelCause(context.Background())
|
||||
task, err := m.StartTask("conversation", "test", cancel)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := m.CancelTask("conversation", multiagent.ErrInterruptContinue); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m.UpdateTaskStatus("conversation", "running")
|
||||
if err := m.FinishTaskRun("conversation", task.RunID, "completed"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if done := m.GetCompletedTasks(); len(done) != 1 || done[0].Status != "completed" {
|
||||
t.Fatal("continue run lost success")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDoneEventCarriesTheSavedCancellationStatus(t *testing.T) {
|
||||
m := NewAgentTaskManager()
|
||||
task, err := m.StartTask("conversation", "test", func(error) {})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := m.CancelTask("conversation", ErrTaskCancelled); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
h := &AgentHandler{tasks: m}
|
||||
var result map[string]interface{}
|
||||
send := h.taskFinishingEventSender(func(kind, message string, data interface{}) {
|
||||
if kind == "done" {
|
||||
result = data.(map[string]interface{})
|
||||
}
|
||||
}, "conversation", task.RunID, func() string { return "completed" })
|
||||
original := map[string]interface{}{"messageId": "message"}
|
||||
send("done", "", original)
|
||||
if result["status"] != "cancelled" || result["messageId"] != "message" {
|
||||
t.Fatalf("incorrect done payload: %#v", result)
|
||||
}
|
||||
if _, exists := original["status"]; exists {
|
||||
t.Fatal("caller payload mutated")
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,22 @@ func (h *AgentHandler) taskFinishingEventSender(send func(string, string, interf
|
||||
}
|
||||
send("error", taskCleanupMessage(err)+": "+err.Error(), map[string]interface{}{"errorType": taskCleanupStatus(err)})
|
||||
data = map[string]interface{}{"conversationId": conversationID, "runId": runID, "status": taskCleanupStatus(err), "cleanupError": err.Error()}
|
||||
} else {
|
||||
payload := map[string]interface{}{}
|
||||
if existing, ok := data.(map[string]interface{}); ok {
|
||||
for key, value := range existing {
|
||||
payload[key] = value
|
||||
}
|
||||
}
|
||||
payload["conversationId"] = conversationID
|
||||
payload["runId"] = runID
|
||||
for _, task := range h.tasks.GetCompletedTasks() {
|
||||
if task.RunID == runID && task.ConversationID == conversationID {
|
||||
payload["status"] = task.Status
|
||||
break
|
||||
}
|
||||
}
|
||||
data = payload
|
||||
}
|
||||
}
|
||||
send(eventType, message, data)
|
||||
|
||||
@@ -37,6 +37,7 @@ type AgentTask struct {
|
||||
IsolationBackend string `json:"isolationBackend,omitempty"`
|
||||
finishing chan struct{}
|
||||
stopping chan struct{}
|
||||
cancelRequested bool
|
||||
finalStatus string
|
||||
ConversationID string `json:"conversationId"`
|
||||
Title string `json:"title,omitempty"`
|
||||
@@ -484,6 +485,11 @@ func (m *AgentTaskManager) CancelTask(conversationID string, cause error) (bool,
|
||||
return true, nil
|
||||
}
|
||||
|
||||
if cause == nil || errors.Is(cause, ErrTaskCancelled) {
|
||||
task.cancelRequested = true
|
||||
task.finalStatus = "cancelled"
|
||||
}
|
||||
|
||||
// ErrInterruptContinue:仅掐断当前推理步骤,随后由处理器续跑,不进入长时间「取消中」态。
|
||||
if cause != nil && errors.Is(cause, multiagent.ErrInterruptContinue) {
|
||||
task.Status = "running"
|
||||
@@ -559,6 +565,9 @@ func (m *AgentTaskManager) UpdateTaskStatus(conversationID string, status string
|
||||
if task.finishing != nil || task.Status == "cleanup_failed" {
|
||||
return
|
||||
}
|
||||
if task.cancelRequested {
|
||||
status = "cancelled"
|
||||
}
|
||||
switch status {
|
||||
case "completed", "cancelled", "failed", "timeout":
|
||||
task.finalStatus = status
|
||||
@@ -628,6 +637,9 @@ func (m *AgentTaskManager) FinishTaskRun(conversationID, runID, finalStatus stri
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if task.cancelRequested {
|
||||
finalStatus = "cancelled"
|
||||
}
|
||||
done := make(chan struct{})
|
||||
task.finishing = done
|
||||
task.finalStatus = finalStatus
|
||||
|
||||
Reference in new issue
Block a user