From 937f77a7ec378847a51999a063b1c892bd894613 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B4=AA=E6=B3=BD=E9=91=AB?= Date: Fri, 21 Aug 2026 22:13:01 +0800 Subject: [PATCH] fix(runtime): preserve tasks through reconnect and late failure Co-authored-by: multica-agent --- server/cmd/server/runtime_sweeper.go | 46 ++++-- server/cmd/server/runtime_sweeper_test.go | 185 ++++++++++++++++++++++ server/internal/service/task.go | 55 +++++-- server/pkg/db/generated/comment.sql.go | 26 +++ server/pkg/db/generated/runtime.sql.go | 12 +- server/pkg/db/queries/comment.sql | 18 +++ server/pkg/db/queries/runtime.sql | 8 +- 7 files changed, 319 insertions(+), 31 deletions(-) diff --git a/server/cmd/server/runtime_sweeper.go b/server/cmd/server/runtime_sweeper.go index 3bc09ed859e..e5ed68db21a 100644 --- a/server/cmd/server/runtime_sweeper.go +++ b/server/cmd/server/runtime_sweeper.go @@ -37,6 +37,11 @@ const ( // The dispatched→running transition should be near-instant, so 5 minutes // means something went wrong (e.g. StartTask API call failed silently). dispatchTimeoutSeconds = 300.0 + // offlineRuntimeRecoveryGraceSeconds gives a daemon one bounded reconnect + // window after its runtime row flips offline. The daemon's WebSocket retry + // backoff is capped at 30s, so 60s covers one failed reconnect plus the next + // attempt without immediately failing a healthy in-flight run. + offlineRuntimeRecoveryGraceSeconds = 60.0 // runningTimeoutSeconds fails tasks stuck in 'running' beyond this. It is a // coarse server-side backstop keyed on started_at, AND-gated by daemon // liveness (agent_runtime.last_seen_at freshness within @@ -48,9 +53,10 @@ const ( // so the server-side wall clock is only a defensive backstop for the // pathological case where a runtime row somehow retains status='online' // with a stale DB heartbeat for longer than this timeout. The primary - // "daemon died" path is `sweepStaleRuntimes` in the same tick (Redis - // liveness + DB stale + FailTasksForOfflineRuntimes), which typically - // reclaims orphaned tasks within ~180s. + // "daemon died" path is `sweepStaleRuntimes` followed by + // `sweepOfflineRuntimeTasks`: Redis/DB liveness marks the runtime offline, + // then the bounded recovery grace lets a reconnect preserve the in-flight + // task before orphan cleanup runs. runningTimeoutSeconds = 9000.0 // queuedTTLSeconds expires tasks that have been sitting in 'queued' // for longer than this without ever being claimed. This is the cleanup @@ -100,6 +106,7 @@ func runRuntimeSweeper(ctx context.Context, queries *db.Queries, liveness handle return case <-ticker.C: sweepStaleRuntimes(ctx, queries, liveness, taskSvc, bus) + sweepOfflineRuntimeTasks(ctx, queries, taskSvc) sweepStaleTasks(ctx, queries, taskSvc, bus) sweepExpiredQueuedTasks(ctx, queries, taskSvc) sweepDeferredChatFinalizations(ctx, queries, taskSvc) @@ -108,8 +115,9 @@ func runRuntimeSweeper(ctx context.Context, queries *db.Queries, liveness handle } } -// sweepStaleRuntimes marks runtimes offline if they haven't heartbeated, -// then fails any tasks belonging to those offline runtimes. +// sweepStaleRuntimes marks runtimes offline if they haven't heartbeated. +// Active tasks are handled separately by sweepOfflineRuntimeTasks after a +// bounded reconnect grace period. func sweepStaleRuntimes(ctx context.Context, queries *db.Queries, liveness handler.LivenessStore, taskSvc *service.TaskService, bus *events.Bus) { candidates, err := queries.SelectStaleOnlineRuntimes(ctx, staleThresholdSeconds) if err != nil { @@ -168,15 +176,6 @@ func sweepStaleRuntimes(ctx context.Context, queries *db.Queries, liveness handl slog.Info("runtime sweeper: marked stale runtimes offline", "count", len(staleRows), "workspaces", len(workspaces)) - // Fail orphaned tasks (dispatched/running) whose runtimes just went offline. - failedTasks, err := queries.FailTasksForOfflineRuntimes(ctx) - if err != nil { - slog.Warn("runtime sweeper: failed to clean up stale tasks", "error", err) - } else if len(failedTasks) > 0 { - slog.Info("runtime sweeper: failed orphaned tasks", "count", len(failedTasks)) - taskSvc.HandleFailedTasks(ctx, failedTasks) - } - // Notify frontend clients so they re-fetch runtime list. for wsID := range workspaces { bus.Publish(events.Event{ @@ -190,6 +189,25 @@ func sweepStaleRuntimes(ctx context.Context, queries *db.Queries, liveness handl } } +// sweepOfflineRuntimeTasks fails active tasks only after their runtime has +// remained offline for a full reconnect grace window. It runs every tick (not +// just the tick that first marks a runtime offline) so rows are revisited when +// their grace period expires. A daemon that reconnects in time flips the +// runtime online and keeps its task intact. +func sweepOfflineRuntimeTasks(ctx context.Context, queries *db.Queries, taskSvc *service.TaskService) { + failedTasks, err := queries.FailTasksForOfflineRuntimes(ctx, offlineRuntimeRecoveryGraceSeconds) + if err != nil { + slog.Warn("runtime sweeper: failed to clean up offline-runtime tasks", "error", err) + return + } + if len(failedTasks) == 0 { + return + } + + slog.Info("runtime sweeper: failed orphaned tasks after reconnect grace", "count", len(failedTasks)) + taskSvc.HandleFailedTasks(ctx, failedTasks) +} + // filterStaleRuntimesByLiveness narrows a SELECT-of-stale-candidates down to // the set that should actually be flipped offline. When liveness is available // and reports a candidate as alive, we skip it (DB is just lagging). When the diff --git a/server/cmd/server/runtime_sweeper_test.go b/server/cmd/server/runtime_sweeper_test.go index 55866dbbfb8..328d04e4022 100644 --- a/server/cmd/server/runtime_sweeper_test.go +++ b/server/cmd/server/runtime_sweeper_test.go @@ -104,6 +104,99 @@ func ageOutAgentRuntime(t *testing.T, agentID string, staleAgo time.Duration) { }) } +func TestOfflineRuntimeReconnectInsideGracePreservesTask(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + ctx := context.Background() + issueID, agentID, taskID := setupSweeperTestFixture(t, "running") + t.Cleanup(func() { cleanupSweeperFixture(t, issueID, agentID) }) + t.Cleanup(func() { + testPool.Exec(context.Background(), ` + UPDATE agent_runtime SET status = 'online', last_seen_at = now(), updated_at = now() + WHERE id = (SELECT runtime_id FROM agent WHERE id = $1) + `, agentID) + }) + + if _, err := testPool.Exec(ctx, ` + UPDATE agent_runtime SET status = 'offline', updated_at = now() + WHERE id = (SELECT runtime_id FROM agent WHERE id = $1) + `, agentID); err != nil { + t.Fatalf("mark runtime freshly offline: %v", err) + } + + queries := db.New(testPool) + failed, err := queries.FailTasksForOfflineRuntimes(ctx, offlineRuntimeRecoveryGraceSeconds) + if err != nil { + t.Fatalf("FailTasksForOfflineRuntimes inside grace: %v", err) + } + for _, task := range failed { + if task.ID.Bytes == parseUUIDBytes(taskID) { + t.Fatalf("task %s failed before reconnect grace elapsed", taskID) + } + } + + if _, err := testPool.Exec(ctx, ` + UPDATE agent_runtime SET status = 'online', last_seen_at = now(), updated_at = now() + WHERE id = (SELECT runtime_id FROM agent WHERE id = $1) + `, agentID); err != nil { + t.Fatalf("reconnect runtime: %v", err) + } + if _, err := queries.FailTasksForOfflineRuntimes(ctx, offlineRuntimeRecoveryGraceSeconds); err != nil { + t.Fatalf("FailTasksForOfflineRuntimes after reconnect: %v", err) + } + + var status string + if err := testPool.QueryRow(ctx, `SELECT status FROM agent_task_queue WHERE id = $1`, taskID).Scan(&status); err != nil { + t.Fatalf("load task after reconnect: %v", err) + } + if status != "running" { + t.Fatalf("task status after reconnect = %q, want running", status) + } +} + +func TestOfflineRuntimeBeyondGraceFailsTask(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + ctx := context.Background() + issueID, agentID, taskID := setupSweeperTestFixture(t, "running") + t.Cleanup(func() { cleanupSweeperFixture(t, issueID, agentID) }) + t.Cleanup(func() { + testPool.Exec(context.Background(), ` + UPDATE agent_runtime SET status = 'online', last_seen_at = now(), updated_at = now() + WHERE id = (SELECT runtime_id FROM agent WHERE id = $1) + `, agentID) + }) + + if _, err := testPool.Exec(ctx, ` + UPDATE agent_runtime + SET status = 'offline', updated_at = now() - make_interval(secs => $1) + WHERE id = (SELECT runtime_id FROM agent WHERE id = $2) + `, offlineRuntimeRecoveryGraceSeconds+1, agentID); err != nil { + t.Fatalf("age offline runtime beyond grace: %v", err) + } + + failed, err := db.New(testPool).FailTasksForOfflineRuntimes(ctx, offlineRuntimeRecoveryGraceSeconds) + if err != nil { + t.Fatalf("FailTasksForOfflineRuntimes beyond grace: %v", err) + } + found := false + for _, task := range failed { + if task.ID.Bytes == parseUUIDBytes(taskID) { + found = true + if !task.FailureReason.Valid || task.FailureReason.String != "runtime_offline" { + t.Fatalf("failure_reason = %v, want runtime_offline", task.FailureReason) + } + } + } + if !found { + t.Fatalf("task %s was not failed after reconnect grace elapsed", taskID) + } +} + func TestRefreshAgentStatusFromTasks(t *testing.T) { if testPool == nil { t.Skip("no database connection") @@ -648,6 +741,98 @@ func TestSweepResetsInProgressIssueToTodo(t *testing.T) { } } +// TestSweepPreservesDeliveredAgentOutput verifies the late-failure consistency +// contract: a source-linked agent comment is durable delivery evidence. The +// task remains failed, but reconciliation must not enqueue a duplicate retry or +// send the issue back to todo after the user-visible handoff already landed. +func TestSweepPreservesDeliveredAgentOutput(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + ctx := context.Background() + issueID, agentID, taskID := setupSweeperTestFixture(t, "running") + t.Cleanup(func() { cleanupSweeperFixture(t, issueID, agentID) }) + t.Cleanup(func() { + testPool.Exec(context.Background(), `DELETE FROM comment WHERE issue_id = $1`, issueID) + }) + + if _, err := testPool.Exec(ctx, `UPDATE issue SET status = 'in_progress' WHERE id = $1`, issueID); err != nil { + t.Fatalf("seed in_progress issue: %v", err) + } + queries := db.New(testPool) + if _, err := testPool.Exec(ctx, ` + INSERT INTO comment (issue_id, workspace_id, author_type, author_id, content, type, source_task_id) + VALUES ($1, $2, 'agent', $3, 'runtime went offline', 'system', $4) + `, issueID, testWorkspaceID, agentID, taskID); err != nil { + t.Fatalf("insert source-linked system failure: %v", err) + } + if delivered, err := queries.HasAgentOutputCommentForTask(ctx, parseUUID(taskID)); err != nil { + t.Fatalf("check system failure delivery evidence: %v", err) + } else if delivered { + t.Fatal("source-linked system failure was mistaken for delivered agent output") + } + if _, err := testPool.Exec(ctx, ` + INSERT INTO comment (issue_id, workspace_id, author_type, author_id, content, type, source_task_id) + VALUES ($1, $2, 'agent', $3, 'Delivered agent result', 'comment', $4) + `, issueID, testWorkspaceID, agentID, taskID); err != nil { + t.Fatalf("insert source-linked agent output: %v", err) + } + if delivered, err := queries.HasAgentOutputCommentForTask(ctx, parseUUID(taskID)); err != nil { + t.Fatalf("check agent output delivery evidence: %v", err) + } else if !delivered { + t.Fatal("source-linked agent output was not recognized as delivery evidence") + } + + ageOutAgentRuntime(t, agentID, 10*time.Minute) + failedTasks, err := queries.FailStaleTasks(ctx, db.FailStaleTasksParams{ + DispatchTimeoutSecs: 300.0, + RunningTimeoutSecs: 1.0, + RuntimeStaleSecs: staleThresholdSeconds, + }) + if err != nil { + t.Fatalf("FailStaleTasks: %v", err) + } + found := false + for _, task := range failedTasks { + if task.ID.Bytes == parseUUIDBytes(taskID) { + found = true + } + } + if !found { + t.Fatalf("expected task %s in failed batch", taskID) + } + + taskService := service.NewTaskService(queries, testPool, nil, events.New()) + if retried := taskService.HandleFailedTasks(ctx, failedTasks); retried != 0 { + t.Fatalf("HandleFailedTasks retried %d task(s), want 0 after delivered output", retried) + } + + var ( + issueStatus string + retryCount int + taskStatus string + ) + if err := testPool.QueryRow(ctx, `SELECT status FROM issue WHERE id = $1`, issueID).Scan(&issueStatus); err != nil { + t.Fatalf("load reconciled issue: %v", err) + } + if err := testPool.QueryRow(ctx, `SELECT count(*) FROM agent_task_queue WHERE retry_of_task_id = $1`, taskID).Scan(&retryCount); err != nil { + t.Fatalf("count retry children: %v", err) + } + if err := testPool.QueryRow(ctx, `SELECT status FROM agent_task_queue WHERE id = $1`, taskID).Scan(&taskStatus); err != nil { + t.Fatalf("load failed task: %v", err) + } + if issueStatus != "in_review" { + t.Fatalf("issue status = %q, want in_review", issueStatus) + } + if retryCount != 0 { + t.Fatalf("retry children = %d, want 0", retryCount) + } + if taskStatus != "failed" { + t.Fatalf("task status = %q, want failed", taskStatus) + } +} + // TestSweepDoesNotResetIssueAlreadyInReview verifies that the sweeper only resets // issues that are truly stuck in in_progress — it must not clobber issues whose // agents already moved them forward (e.g. to in_review) before the task timed out. diff --git a/server/internal/service/task.go b/server/internal/service/task.go index 3e94e1fc60a..22ed92c867e 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -4598,9 +4598,11 @@ func (s *TaskService) enqueueRerunTask(ctx context.Context, issue db.Issue, agen // HandleFailedTasks runs the post-failure side effects for a batch of // freshly-failed tasks: optional auto-retry, task:failed event broadcast, -// agent status reconciliation, and (when an issue has no remaining active -// task and isn't being retried) resetting the issue back to todo so the -// daemon can pick it up again. +// agent status reconciliation, and issue status recovery. An in_progress +// issue normally resets to todo when no task remains. If the failed task +// already wrote a source-linked agent output comment, it instead advances to +// in_review and is never auto-retried: the run failure stays truthful without +// duplicating delivered work or hiding the handoff from reviewers. // // All callers that surface a task as failed — sweepers, FailTask, // recover-orphans — funnel through here so the same UI-consistency @@ -4613,18 +4615,49 @@ func (s *TaskService) HandleFailedTasks(ctx context.Context, tasks []db.AgentTas affectedAgents := make(map[string]pgtype.UUID) processedIssues := make(map[string]bool) retriedIssues := make(map[string]bool) + issuesWithDeliveredOutput := make(map[string]bool) + tasksWithDeliveredOutput := make(map[string]bool) retried := 0 + // Gather delivery evidence before deciding retries or issue status. Do this + // for the whole batch so iteration order cannot let one failed task reset an + // issue to todo when another task in the same batch already delivered output. + for _, t := range tasks { + if !t.IssueID.Valid { + continue + } + delivered, err := s.Queries.HasAgentOutputCommentForTask(ctx, t.ID) + if err != nil { + slog.Warn("handle failed tasks: delivered output check failed", + "task_id", util.UUIDToString(t.ID), + "issue_id", util.UUIDToString(t.IssueID), + "error", err, + ) + continue + } + if delivered { + tasksWithDeliveredOutput[util.UUIDToString(t.ID)] = true + issuesWithDeliveredOutput[util.UUIDToString(t.IssueID)] = true + } + } + for _, t := range tasks { // Auto-retry first so the issue stays in_progress rather than // flapping todo → in_progress within a tick. retryPending := false - if child, _ := s.MaybeRetryFailedTask(ctx, t); child != nil { - retryPending = true - retried++ - if t.IssueID.Valid { - retriedIssues[util.UUIDToString(t.IssueID)] = true + if !tasksWithDeliveredOutput[util.UUIDToString(t.ID)] { + if child, _ := s.MaybeRetryFailedTask(ctx, t); child != nil { + retryPending = true + retried++ + if t.IssueID.Valid { + retriedIssues[util.UUIDToString(t.IssueID)] = true + } } + } else { + slog.Info("task auto-retry skipped: agent output already delivered", + "task_id", util.UUIDToString(t.ID), + "issue_id", util.UUIDToString(t.IssueID), + ) } failureReason := "agent_error" @@ -4649,9 +4682,13 @@ func (s *TaskService) HandleFailedTasks(ctx context.Context, tasks []db.AgentTas "error", checkErr, ) } else if !hasActive { + nextStatus := "todo" + if issuesWithDeliveredOutput[issueKey] { + nextStatus = "in_review" + } updatedIssue, updateErr := s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{ ID: t.IssueID, - Status: "todo", + Status: nextStatus, WorkspaceID: issue.WorkspaceID, }) if updateErr != nil { diff --git a/server/pkg/db/generated/comment.sql.go b/server/pkg/db/generated/comment.sql.go index afe1454e96f..caecd580af9 100644 --- a/server/pkg/db/generated/comment.sql.go +++ b/server/pkg/db/generated/comment.sql.go @@ -420,6 +420,32 @@ func (q *Queries) HasAgentCommentedSince(ctx context.Context, arg HasAgentCommen return commented, err } +const hasAgentOutputCommentForTask = `-- name: HasAgentOutputCommentForTask :one +SELECT EXISTS ( + SELECT 1 + FROM comment c + JOIN agent_task_queue t ON t.id = c.source_task_id + WHERE t.id = $1 + AND t.issue_id IS NOT NULL + AND c.issue_id = t.issue_id + AND c.author_type = 'agent' + AND c.author_id = t.agent_id + AND c.type = 'comment' +) AS delivered +` + +// A source-linked agent comment is durable proof that this run already +// delivered user-visible output. Failure reconciliation uses it to avoid +// duplicate retries and a misleading issue rollback after late task failure. +// Match the task's own issue/agent lineage and exclude generated system +// failure messages, which also carry source_task_id but are not agent output. +func (q *Queries) HasAgentOutputCommentForTask(ctx context.Context, sourceTaskID pgtype.UUID) (bool, error) { + row := q.db.QueryRow(ctx, hasAgentOutputCommentForTask, sourceTaskID) + var delivered bool + err := row.Scan(&delivered) + return delivered, err +} + const hasAgentRepliedInThread = `-- name: HasAgentRepliedInThread :one SELECT count(*) > 0 AS has_replied FROM comment WHERE parent_id = $1 AND author_type = 'agent' AND author_id = $2 diff --git a/server/pkg/db/generated/runtime.sql.go b/server/pkg/db/generated/runtime.sql.go index ef720f14a5a..4dee3b9c6ae 100644 --- a/server/pkg/db/generated/runtime.sql.go +++ b/server/pkg/db/generated/runtime.sql.go @@ -223,16 +223,18 @@ SET status = 'failed', completed_at = now(), error = 'runtime went offline', wait_reason = NULL WHERE status IN ('dispatched', 'running', 'waiting_local_directory') AND runtime_id IN ( - SELECT id FROM agent_runtime WHERE status = 'offline' + SELECT id FROM agent_runtime + WHERE status = 'offline' + AND updated_at < now() - make_interval(secs => $1::double precision) ) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` // Marks dispatched/running/waiting_local_directory tasks as failed when -// their runtime is offline. This cleans up orphaned tasks after a daemon -// crash or network partition. -func (q *Queries) FailTasksForOfflineRuntimes(ctx context.Context) ([]AgentTaskQueue, error) { - rows, err := q.db.Query(ctx, failTasksForOfflineRuntimes) +// their runtime has remained offline beyond a bounded recovery window. A +// reconnect sets the runtime online before this query can claim the task. +func (q *Queries) FailTasksForOfflineRuntimes(ctx context.Context, offlineGraceSeconds float64) ([]AgentTaskQueue, error) { + rows, err := q.db.Query(ctx, failTasksForOfflineRuntimes, offlineGraceSeconds) if err != nil { return nil, err } diff --git a/server/pkg/db/queries/comment.sql b/server/pkg/db/queries/comment.sql index 058864bdfa8..5ffd5571138 100644 --- a/server/pkg/db/queries/comment.sql +++ b/server/pkg/db/queries/comment.sql @@ -452,6 +452,24 @@ SELECT EXISTS ( AND created_at >= @since ) AS commented; +-- name: HasAgentOutputCommentForTask :one +-- A source-linked agent comment is durable proof that this run already +-- delivered user-visible output. Failure reconciliation uses it to avoid +-- duplicate retries and a misleading issue rollback after late task failure. +-- Match the task's own issue/agent lineage and exclude generated system +-- failure messages, which also carry source_task_id but are not agent output. +SELECT EXISTS ( + SELECT 1 + FROM comment c + JOIN agent_task_queue t ON t.id = c.source_task_id + WHERE t.id = @source_task_id + AND t.issue_id IS NOT NULL + AND c.issue_id = t.issue_id + AND c.author_type = 'agent' + AND c.author_id = t.agent_id + AND c.type = 'comment' +) AS delivered; + -- name: HasAgentRepliedInThread :one -- Returns true if the given agent has posted a reply in the thread rooted at -- the specified parent comment. Used to detect agent participation in a diff --git a/server/pkg/db/queries/runtime.sql b/server/pkg/db/queries/runtime.sql index 007639ed182..dc9e7ad1069 100644 --- a/server/pkg/db/queries/runtime.sql +++ b/server/pkg/db/queries/runtime.sql @@ -227,15 +227,17 @@ RETURNING id, workspace_id, owner_id, daemon_id, provider; -- name: FailTasksForOfflineRuntimes :many -- Marks dispatched/running/waiting_local_directory tasks as failed when --- their runtime is offline. This cleans up orphaned tasks after a daemon --- crash or network partition. +-- their runtime has remained offline beyond a bounded recovery window. A +-- reconnect sets the runtime online before this query can claim the task. UPDATE agent_task_queue SET status = 'failed', completed_at = now(), error = 'runtime went offline', failure_reason = 'runtime_offline', wait_reason = NULL WHERE status IN ('dispatched', 'running', 'waiting_local_directory') AND runtime_id IN ( - SELECT id FROM agent_runtime WHERE status = 'offline' + SELECT id FROM agent_runtime + WHERE status = 'offline' + AND updated_at < now() - make_interval(secs => @offline_grace_seconds::double precision) ) RETURNING *;