diff --git a/docs/feature/task-timeout-classification.md b/docs/feature/task-timeout-classification.md index 62d36db7..3bcbb2f3 100644 --- a/docs/feature/task-timeout-classification.md +++ b/docs/feature/task-timeout-classification.md @@ -33,3 +33,9 @@ Replace the four task-level OR checks with this predicate. Child timeouts follow 4. Commit and push each phase; merge only after checks pass and remote main is rechecked. No release or deployment requested. Rollback: revert the fix commit; no data migration. Customer-specific fault attribution remains a separate evidence gap. + +## F4 — Implementation + +Added `taskDeadlineExceeded`, based on task context Err rather than operation error or cancellation cause. Single-agent, multi-agent, role workflow and batch task paths now use it. Existing terminal error details, trace persistence and retry policy are unchanged. + +Regression proof: before the fix, `TestBatchTaskDeadlineClassification` failed for both child DeadlineExceeded and wrapped HTTP timeout (got timeout, wanted failed). After the fix, targeted tests pass, including real expiry, unrelated returned error after expiry, custom deadline causes and user cancellation. diff --git a/internal/handler/batch_queue_executor.go b/internal/handler/batch_queue_executor.go index 34ea2c49..be950dfe 100644 --- a/internal/handler/batch_queue_executor.go +++ b/internal/handler/batch_queue_executor.go @@ -385,7 +385,7 @@ func (h *AgentHandler) handleBatchSubTaskRunError( strings.Contains(strings.ToLower(errStr), "context canceled") || strings.Contains(strings.ToLower(errStr), "context cancelled") || (partialResp != "" && (strings.Contains(partialResp, "任务已被取消") || strings.Contains(partialResp, "任务执行中断"))) - isTimeout := errors.Is(runErr, context.DeadlineExceeded) || errors.Is(context.Cause(taskCtx), context.DeadlineExceeded) + isTimeout := taskDeadlineExceeded(taskCtx) if isTimeout { *finishStatus = "timeout" diff --git a/internal/handler/eino_single_agent.go b/internal/handler/eino_single_agent.go index 88ea6691..fbfd6c9d 100644 --- a/internal/handler/eino_single_agent.go +++ b/internal/handler/eino_single_agent.go @@ -356,7 +356,7 @@ func (h *AgentHandler) EinoSingleAgentLoopStream(c *gin.Context) { return } - if errors.Is(runErr, context.DeadlineExceeded) || errors.Is(context.Cause(taskCtx), context.DeadlineExceeded) { + if taskDeadlineExceeded(taskCtx) { taskStatus = "timeout" h.tasks.UpdateTaskStatus(conversationID, taskStatus) timeoutMsg := "任务执行超时，已自动终止。" diff --git a/internal/handler/multi_agent.go b/internal/handler/multi_agent.go index 6e7a86bd..7ea97834 100644 --- a/internal/handler/multi_agent.go +++ b/internal/handler/multi_agent.go @@ -370,7 +370,7 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) { return } - if errors.Is(runErr, context.DeadlineExceeded) || errors.Is(context.Cause(taskCtx), context.DeadlineExceeded) { + if taskDeadlineExceeded(taskCtx) { taskStatus = "timeout" h.tasks.UpdateTaskStatus(conversationID, taskStatus) timeoutMsg := "任务执行超时，已自动终止。" diff --git a/internal/handler/task_deadline.go b/internal/handler/task_deadline.go new file mode 100644 index 00000000..37d7c8be --- /dev/null +++ b/internal/handler/task_deadline.go @@ -0,0 +1,13 @@ +package handler + +import ( + "context" + "errors" +) + +// taskDeadlineExceeded distinguishes the task's deadline from a child operation's +// timeout. A run error (including a cancellation cause) may wrap DeadlineExceeded +// while the task itself still has time remaining; it must not exhaust the task budget. +func taskDeadlineExceeded(taskCtx context.Context) bool { + return taskCtx != nil && errors.Is(taskCtx.Err(), context.DeadlineExceeded) +} diff --git a/internal/handler/task_deadline_test.go b/internal/handler/task_deadline_test.go new file mode 100644 index 00000000..a40579c6 --- /dev/null +++ b/internal/handler/task_deadline_test.go @@ -0,0 +1,111 @@ +package handler + +import ( + "context" + "errors" + "fmt" + "net/url" + "path/filepath" + "strings" + "testing" + "time" + + "cyberstrike-ai/internal/database" + "go.uber.org/zap" +) + +func TestBatchTaskDeadlineClassification(t *testing.T) { + for _, tc := range []struct { + name string + expired bool + runErr error + want string + }{ + {"child deadline", false, context.DeadlineExceeded, "failed"}, + {"wrapped HTTP timeout", false, &url.Error{Op: "Post", URL: "https://model.invalid", Err: fmt.Errorf("request timeout: %w", context.DeadlineExceeded)}, "failed"}, + {"task deadline", true, context.DeadlineExceeded, "timeout"}, + {"task deadline with unrelated error", true, errors.New("stream ended"), "timeout"}, + } { + t.Run(tc.name, func(t *testing.T) { + logger := zap.NewNop() + db, err := database.NewDB(filepath.Join(t.TempDir(), "deadline.db"), logger) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + conv, err := db.CreateConversation("deadline regression", database.ConversationCreateMeta{}) + if err != nil { + t.Fatal(err) + } + msg, err := db.AddMessage(conv.ID, "assistant", "", nil) + if err != nil { + t.Fatal(err) + } + manager := NewBatchTaskManager(logger) + queue, err := manager.CreateBatchQueue("deadline", "", "eino_single", "manual", "", "", nil, 1, []string{"test"}) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Hour) + if tc.expired { + deadline = time.Now().Add(-time.Second) + } + taskCtx, cancel := context.WithDeadline(context.Background(), deadline) + defer cancel() + h := &AgentHandler{db: db, logger: logger, batchTaskManager: manager} + status := "running" + h.handleBatchSubTaskRunError(queue.ID, queue.Tasks[0], conv.ID, msg.ID, context.Background(), taskCtx, nil, tc.runErr, &status) + if status != tc.want { + t.Fatalf("finish status = %q, want %q", status, tc.want) + } + messages, err := db.GetMessages(conv.ID) + if err != nil { + t.Fatal(err) + } + if len(messages) != 1 || !strings.Contains(messages[0].Content, tc.runErr.Error()) { + t.Fatalf("original diagnostic not preserved: %+v", messages) + } + updated, ok := manager.GetBatchQueue(queue.ID) + if !ok || updated.Tasks[0].Status != BatchTaskStatusFailed { + t.Fatalf("batch failure not recorded: %+v", updated) + } + }) + } +} + +func TestTaskDeadlineExceededContextOwnership(t *testing.T) { + taskCtx, stop := context.WithTimeout(context.Background(), 10*time.Hour) + defer stop() + child, stopChild := context.WithDeadline(taskCtx, time.Now().Add(-time.Second)) + defer stopChild() + if !errors.Is(child.Err(), context.DeadlineExceeded) { + t.Fatal("child deadline did not expire") + } + if taskDeadlineExceeded(taskCtx) { + t.Fatal("child expiry exhausted parent task") + } + if !taskDeadlineExceeded(child) { + t.Fatal("expired task deadline not recognized") + } + + cancelled, cancel := context.WithCancelCause(context.Background()) + cancel(fmt.Errorf("child failed: %w", context.DeadlineExceeded)) + if taskDeadlineExceeded(cancelled) { + t.Fatal("cancellation cause misclassified as task deadline") + } + + userCtx, userCancel := context.WithCancelCause(context.Background()) + userCancel(ErrTaskCancelled) + if taskDeadlineExceeded(userCtx) { + t.Fatal("user cancellation misclassified as task deadline") + } + + customDeadline, stopCustom := context.WithDeadlineCause(context.Background(), time.Now().Add(-time.Second), errors.New("budget exhausted")) + defer stopCustom() + if !taskDeadlineExceeded(customDeadline) { + t.Fatal("custom deadline cause hid real task expiry") + } + if taskDeadlineExceeded(nil) || taskDeadlineExceeded(context.Background()) { + t.Fatal("context without deadline classified as expired") + } +} diff --git a/internal/handler/workflow_integration.go b/internal/handler/workflow_integration.go index a154d9aa..d6627772 100644 --- a/internal/handler/workflow_integration.go +++ b/internal/handler/workflow_integration.go @@ -151,7 +151,7 @@ func (h *AgentHandler) runRoleWorkflowStreamIfBound( sendEvent("done", "", map[string]interface{}{"conversationId": conversationID}) return true } - if errors.Is(err, context.DeadlineExceeded) || errors.Is(context.Cause(taskCtx), context.DeadlineExceeded) { + if taskDeadlineExceeded(taskCtx) { taskStatus = "timeout" h.tasks.UpdateTaskStatus(conversationID, taskStatus) timeoutMsg := "任务执行超时，已自动终止。"