feat: 完成SubAgent DAG调度器并修复infra稳定性

internal/session/todo.go

@@ -9,7 +9,7 @@ import (
 )
 
 // CurrentTodoVersion 表示当前 Todo 结构版本。
-const CurrentTodoVersion = 2
+const CurrentTodoVersion = 3
 
 // TodoStatus 表示 Todo 项的状态枚举。
 type TodoStatus string
@@ -260,30 +260,42 @@ func (s *Session) UpdateTodo(id string, patch TodoPatch, expectedRevision int64)
 // ClaimTodo 用于 SubAgent 领取 Todo，并设置 owner 与执行中状态。
 func (s *Session) ClaimTodo(id string, ownerType string, ownerID string, expectedRevision int64) error {
 	status := TodoStatusInProgress
+	failureReason := ""
+	nextRetryAt := time.Time{}
 	patch := TodoPatch{
-		Status:    &status,
-		OwnerType: &ownerType,
-		OwnerID:   &ownerID,
+		Status:        &status,
+		OwnerType:     &ownerType,
+		OwnerID:       &ownerID,
+		FailureReason: &failureReason,
+		NextRetryAt:   &nextRetryAt,
 	}
 	return s.UpdateTodo(id, patch, expectedRevision)
 }
 
 // CompleteTodo 将 Todo 标记为完成并写入产物列表。
 func (s *Session) CompleteTodo(id string, artifacts []string, expectedRevision int64) error {
 	status := TodoStatusCompleted
+	failureReason := ""
+	retryCount := 0
+	nextRetryAt := time.Time{}
 	patch := TodoPatch{
-		Status:    &status,
-		Artifacts: &artifacts,
+		Status:        &status,
+		Artifacts:     &artifacts,
+		FailureReason: &failureReason,
+		RetryCount:    &retryCount,
+		NextRetryAt:   &nextRetryAt,
 	}
 	return s.UpdateTodo(id, patch, expectedRevision)
 }
 
 // FailTodo 将 Todo 标记为失败并记录失败原因。
 func (s *Session) FailTodo(id string, reason string, expectedRevision int64) error {
 	status := TodoStatusFailed
+	nextRetryAt := time.Time{}
 	patch := TodoPatch{
 		Status:        &status,
 		FailureReason: &reason,
+		NextRetryAt:   &nextRetryAt,
 	}
 	return s.UpdateTodo(id, patch, expectedRevision)
 }
@@ -395,6 +407,15 @@ func normalizeTodoItem(item TodoItem) (TodoItem, error) {
 	item.Acceptance = normalizeTodoTextList(item.Acceptance)
 	item.Artifacts = normalizeTodoTextList(item.Artifacts)
 	item.FailureReason = strings.TrimSpace(item.FailureReason)
+	if item.RetryCount < 0 {
+		item.RetryCount = 0
+	}
+	if item.RetryLimit < 0 {
+		item.RetryLimit = 0
+	}
+	if !item.NextRetryAt.IsZero() {
+		item.NextRetryAt = item.NextRetryAt.UTC()
+	}
 	if item.Status == "" {
 		item.Status = TodoStatusPending
 	}
@@ -413,9 +434,12 @@ func normalizeTodoItem(item TodoItem) (TodoItem, error) {
 		return TodoItem{}, fmt.Errorf("session: invalid todo owner_type %q", item.OwnerType)
 	}
 
-	if item.Status != TodoStatusFailed {
+	if item.Status != TodoStatusFailed && item.Status != TodoStatusPending && item.Status != TodoStatusBlocked {
 		item.FailureReason = ""
 	}
+	if item.Status != TodoStatusPending && item.Status != TodoStatusFailed {
+		item.NextRetryAt = time.Time{}
+	}
 	return item, nil
 }
 

internal/session/todo_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestTodoStatusValidAndTransition(t *testing.T) {
@@ -194,9 +195,12 @@ func TestSessionClaimCompleteFail(t *testing.T) {
 		t.Fatalf("AddTodo(base) error = %v", err)
 	}
 	if err := session.AddTodo(TodoItem{
-		ID:           "task",
-		Content:      "task",
-		Dependencies: []string{"base"},
+		ID:            "task",
+		Content:       "task",
+		Dependencies:  []string{"base"},
+		FailureReason: "previous failure",
+		RetryCount:    2,
+		NextRetryAt:   time.Now().Add(2 * time.Minute),
 	}); err != nil {
 		t.Fatalf("AddTodo(task) error = %v", err)
 	}
@@ -208,6 +212,17 @@ func TestSessionClaimCompleteFail(t *testing.T) {
 	if task.OwnerType != TodoOwnerTypeSubAgent || task.OwnerID != "worker-1" {
 		t.Fatalf("unexpected owner after claim: %+v", task)
 	}
+	if task.FailureReason != "" || !task.NextRetryAt.IsZero() {
+		t.Fatalf("claim should clear failure/next_retry_at, got %+v", task)
+	}
+	retryCount := 5
+	if err := session.UpdateTodo("task", TodoPatch{RetryCount: &retryCount}, task.Revision); err != nil {
+		t.Fatalf("UpdateTodo(retry_count) error = %v", err)
+	}
+	task, _ = session.FindTodo("task")
+	if task.RetryCount != 5 {
+		t.Fatalf("retry_count = %d, want 5", task.RetryCount)
+	}
 
 	if err := session.FailTodo("task", "compile failed", task.Revision); err != nil {
 		t.Fatalf("FailTodo error = %v", err)
@@ -216,6 +231,9 @@ func TestSessionClaimCompleteFail(t *testing.T) {
 	if task.Status != TodoStatusFailed || task.FailureReason != "compile failed" {
 		t.Fatalf("unexpected fail state: %+v", task)
 	}
+	if !task.NextRetryAt.IsZero() {
+		t.Fatalf("FailTodo should clear next_retry_at, got %+v", task)
+	}
 
 	if err := session.SetTodoStatus("task", TodoStatusInProgress, task.Revision); err == nil || !errors.Is(err, ErrInvalidTransition) {
 		t.Fatalf("terminal transition error = %v, want invalid transition", err)
@@ -356,6 +374,24 @@ func TestTodoInternalHelpers(t *testing.T) {
 	if got := normalizeTodoDependencies([]string{" a ", "a", "b"}); len(got) != 2 {
 		t.Fatalf("normalizeTodoDependencies unexpected result: %+v", got)
 	}
+
+	normalized, err := normalizeTodoItem(TodoItem{
+		ID:            "n1",
+		Content:       "content",
+		Status:        TodoStatusBlocked,
+		FailureReason: " retry failed ",
+		RetryCount:    -3,
+		RetryLimit:    -2,
+	})
+	if err != nil {
+		t.Fatalf("normalizeTodoItem error = %v", err)
+	}
+	if normalized.FailureReason != "retry failed" {
+		t.Fatalf("blocked todo should keep failure reason, got %q", normalized.FailureReason)
+	}
+	if normalized.RetryCount != 0 || normalized.RetryLimit != 0 {
+		t.Fatalf("negative retry fields should be normalized to 0, got count=%d limit=%d", normalized.RetryCount, normalized.RetryLimit)
+	}
 }
 
 func TestApplyTodoPatchCoverage(t *testing.T) {
@@ -381,6 +417,9 @@ func TestApplyTodoPatchCoverage(t *testing.T) {
 	acceptance := []string{"ok"}
 	artifacts := []string{"a.txt"}
 	reason := "boom"
+	retryCount := 1
+	retryLimit := 3
+	nextRetryAt := time.Now().Add(3 * time.Minute).UTC()
 	status := TodoStatusInProgress
 	patch := TodoPatch{
 		Content:       &content,
@@ -391,6 +430,9 @@ func TestApplyTodoPatchCoverage(t *testing.T) {
 		Acceptance:    &acceptance,
 		Artifacts:     &artifacts,
 		FailureReason: &reason,
+		RetryCount:    &retryCount,
+		RetryLimit:    &retryLimit,
+		NextRetryAt:   &nextRetryAt,
 		Status:        &status,
 	}
 
@@ -407,6 +449,26 @@ func TestApplyTodoPatchCoverage(t *testing.T) {
 	if next.FailureReason != "" {
 		t.Fatalf("non-failed status should clear failure reason, got %q", next.FailureReason)
 	}
+	if next.RetryCount != 1 || next.RetryLimit != 3 {
+		t.Fatalf("retry fields not applied, got %+v", next)
+	}
+	if !next.NextRetryAt.IsZero() {
+		t.Fatalf("in_progress should clear next_retry_at, got %+v", next)
+	}
+
+	blocked := TodoStatusBlocked
+	blockedReason := "retry later"
+	blockedPatch := TodoPatch{
+		Status:        &blocked,
+		FailureReason: &blockedReason,
+	}
+	blockedNext, err := applyTodoPatch(base, blockedPatch)
+	if err != nil {
+		t.Fatalf("applyTodoPatch(blocked) error = %v", err)
+	}
+	if blockedNext.FailureReason != blockedReason {
+		t.Fatalf("blocked status should preserve failure reason, got %q", blockedNext.FailureReason)
+	}
 
 	invalidStatus := TodoStatus("paused")
 	if _, err := applyTodoPatch(base, TodoPatch{Status: &invalidStatus}); err == nil {