CHE-16: fix runtime reconnect and late delivery consistency

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

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.

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 {