From 4e9556a1192b652c227b396f498f0990dc4fa90d Mon Sep 17 00:00:00 2001 From: Henry Date: Mon, 7 Sep 2026 20:47:30 +0800 Subject: [PATCH] fix(office): stop Mark fixed from leaving an auto-paused agent stuck (#3464) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../office/repository/sqlite/agents.go | 55 +++- .../office/repository/sqlite/agents_test.go | 70 +++++ .../repository/sqlite/base_migrations.go | 15 +- .../office/repository/sqlite/base_test.go | 1 + .../office/repository/sqlite/failure.go | 144 +++++++++- .../office/repository/sqlite/failure_test.go | 58 ++++ .../repository/sqlite/workspace_deletion.go | 1 + .../sqlite/workspace_deletion_test.go | 16 ++ .../internal/office/service/failure.go | 257 +++++++++++++---- .../internal/office/service/failure_test.go | 272 +++++++++++++++++- .../internal/runs/repository/sqlite/runs.go | 14 +- .../internal/runs/service/service_test.go | 26 ++ 12 files changed, 862 insertions(+), 67 deletions(-) diff --git a/apps/backend/internal/office/repository/sqlite/agents.go b/apps/backend/internal/office/repository/sqlite/agents.go index 941236c42f..986b340439 100644 --- a/apps/backend/internal/office/repository/sqlite/agents.go +++ b/apps/backend/internal/office/repository/sqlite/agents.go @@ -420,6 +420,55 @@ func (r *Repository) UpdateAgentStatusFields( return err } +// UpdateAgentStatusFieldsIfCurrent updates status fields only when the +// agent still has expectedStatus. The affected-row result makes a stale +// caller observable instead of allowing it to overwrite a newer status. +func (r *Repository) UpdateAgentStatusFieldsIfCurrent( + ctx context.Context, id, expectedStatus, status, pauseReason string, +) (bool, error) { + now := time.Now().UTC() + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE agent_profiles + SET status = CASE WHEN ? = 'working' THEN status ELSE ? END, + pause_reason = CASE WHEN ? = 'working' THEN pause_reason ELSE ? END, + working_run_id = CASE + WHEN status = 'working' AND working_run_id <> '' AND ? = 'working' THEN working_run_id + ELSE '' + END, + updated_at = ? + WHERE id = ? AND status = ? AND `+agentInstanceFilter+` + `), status, status, status, pauseReason, status, now, id, expectedStatus) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + if err != nil { + return false, err + } + return rows > 0, nil +} + +// ClearAgentPauseReasonIfCurrent clears only pause_reason when the agent +// still has expectedStatus. This preserves a concurrent working or stopped +// status while avoiding a stale status write. +func (r *Repository) ClearAgentPauseReasonIfCurrent( + ctx context.Context, id, expectedStatus string, +) (bool, error) { + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE agent_profiles + SET pause_reason = '', updated_at = ? + WHERE id = ? AND status = ? AND `+agentInstanceFilter+` + `), time.Now().UTC(), id, expectedStatus) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + if err != nil { + return false, err + } + return rows > 0, nil +} + // GetAgentInstanceByNameAny returns the first agent instance matching a name // across all workspaces. Used for ID-or-name lookups without workspace context. func (r *Repository) GetAgentInstanceByNameAny( @@ -535,8 +584,12 @@ func (r *Repository) DeleteAgentInstanceTx(ctx context.Context, tx *sqlx.Tx, id func (r *Repository) deleteAgentInstance(ctx context.Context, ext sqlx.ExtContext, id string) error { now := time.Now().UTC() + if _, err := ext.ExecContext(ctx, r.db.Rebind( + `UPDATE agent_profiles SET deleted_at = ?, updated_at = ? WHERE id = ?`), now, now, id); err != nil { + return err + } _, err := ext.ExecContext(ctx, r.db.Rebind( - `UPDATE agent_profiles SET deleted_at = ?, updated_at = ? WHERE id = ?`), now, now, id) + `DELETE FROM office_agent_pause_recoveries WHERE agent_id = ?`), id) return err } diff --git a/apps/backend/internal/office/repository/sqlite/agents_test.go b/apps/backend/internal/office/repository/sqlite/agents_test.go index 715fcfc7fd..120a0b44e3 100644 --- a/apps/backend/internal/office/repository/sqlite/agents_test.go +++ b/apps/backend/internal/office/repository/sqlite/agents_test.go @@ -645,6 +645,76 @@ func TestUpdateAgentStatusFields(t *testing.T) { } } +func TestUpdateAgentStatusFieldsIfCurrent_DoesNotOverwriteNewStatus(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + agent := fullAgentInstance("st-cas", "ws-1", "CAS") + if err := repo.CreateAgentInstance(ctx, agent); err != nil { + t.Fatalf("CreateAgentInstance: %v", err) + } + if err := repo.UpdateAgentStatusFields(ctx, agent.ID, "paused", "Auto-paused: test"); err != nil { + t.Fatalf("pause agent: %v", err) + } + + changed, err := repo.UpdateAgentStatusFieldsIfCurrent( + ctx, agent.ID, "paused", "idle", "", + ) + if err != nil { + t.Fatalf("compare-and-set paused: %v", err) + } + if !changed { + t.Fatal("compare-and-set paused = false, want true") + } + + if err := repo.UpdateAgentStatusFields(ctx, agent.ID, "stopped", "manual stop"); err != nil { + t.Fatalf("stop agent: %v", err) + } + changed, err = repo.UpdateAgentStatusFieldsIfCurrent( + ctx, agent.ID, "paused", "idle", "", + ) + if err != nil { + t.Fatalf("stale compare-and-set: %v", err) + } + if changed { + t.Fatal("stale compare-and-set = true, want false") + } + + got, err := repo.GetAgentInstance(ctx, agent.ID) + if err != nil { + t.Fatalf("GetAgentInstance: %v", err) + } + if got.Status != settingsmodels.AgentStatus("stopped") || got.PauseReason != "manual stop" { + t.Fatalf("status/reason = %q/%q, want stopped/manual stop", got.Status, got.PauseReason) + } +} + +func TestClearAgentPauseReasonIfCurrent_PreservesStatus(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + agent := fullAgentInstance("st-clear-reason", "ws-1", "Clear reason") + if err := repo.CreateAgentInstance(ctx, agent); err != nil { + t.Fatalf("CreateAgentInstance: %v", err) + } + if err := repo.UpdateAgentStatusFields(ctx, agent.ID, "stopped", "Auto-paused: stale"); err != nil { + t.Fatalf("set stale pause: %v", err) + } + + changed, err := repo.ClearAgentPauseReasonIfCurrent(ctx, agent.ID, "stopped") + if err != nil { + t.Fatalf("clear reason: %v", err) + } + if !changed { + t.Fatal("clear reason = false, want true") + } + got, err := repo.GetAgentInstance(ctx, agent.ID) + if err != nil { + t.Fatalf("GetAgentInstance: %v", err) + } + if got.Status != settingsmodels.AgentStatus("stopped") || got.PauseReason != "" { + t.Fatalf("status/reason = %q/%q, want stopped/empty", got.Status, got.PauseReason) + } +} + func TestUpdateAgentStatusFields_ClearsWorkingOwner(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/office/repository/sqlite/base_migrations.go b/apps/backend/internal/office/repository/sqlite/base_migrations.go index 5f16269736..2fbc6cf19d 100644 --- a/apps/backend/internal/office/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/office/repository/sqlite/base_migrations.go @@ -227,8 +227,9 @@ func (r *Repository) migrateProviderRouting() { )`) } -// migrateFailureColumns creates the auxiliary office_workspace_settings -// and office_inbox_dismissals tables used by office-agent-error-handling. +// migrateFailureColumns creates the auxiliary failure-handling tables: +// workspace settings, inbox dismissals, and durable auto-pause recovery +// snapshots. // The runs.error_message column is part of the canonical CREATE TABLE // in base.go, so no ALTER is needed here. func (r *Repository) migrateFailureColumns() { @@ -248,6 +249,16 @@ func (r *Repository) migrateFailureColumns() { PRIMARY KEY (user_id, item_kind, item_id) )`) _, _ = r.db.Exec(`CREATE INDEX IF NOT EXISTS idx_office_inbox_dismissals_kind ON office_inbox_dismissals(item_kind, item_id)`) + + _, _ = r.db.Exec(` + CREATE TABLE IF NOT EXISTS office_agent_pause_recoveries ( + agent_id TEXT NOT NULL, + task_id TEXT NOT NULL, + failed_run_id TEXT NOT NULL, + captured_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (agent_id, task_id) + )`) + _, _ = r.db.Exec(`CREATE INDEX IF NOT EXISTS idx_office_agent_pause_recoveries_agent ON office_agent_pause_recoveries(agent_id)`) } // migrateSchedulerColumns adds the atomic task-checkout columns to the diff --git a/apps/backend/internal/office/repository/sqlite/base_test.go b/apps/backend/internal/office/repository/sqlite/base_test.go index b58b69ad35..80345a83f3 100644 --- a/apps/backend/internal/office/repository/sqlite/base_test.go +++ b/apps/backend/internal/office/repository/sqlite/base_test.go @@ -67,6 +67,7 @@ func TestInitSchema_AllTablesExist(t *testing.T) { "office_agent_instructions", "office_labels", "office_task_labels", + "office_agent_pause_recoveries", "task_workspace_groups", "task_workspace_group_members", } diff --git a/apps/backend/internal/office/repository/sqlite/failure.go b/apps/backend/internal/office/repository/sqlite/failure.go index e423014038..cda1966c60 100644 --- a/apps/backend/internal/office/repository/sqlite/failure.go +++ b/apps/backend/internal/office/repository/sqlite/failure.go @@ -24,6 +24,144 @@ func (r *Repository) GetRun( return &req, nil } +// AgentPauseRecovery is the task snapshot captured when an agent enters +// auto-pause. The failed run identifies the failure streak entry that made +// the task eligible for recovery. +type AgentPauseRecovery struct { + AgentID string `db:"agent_id"` + TaskID string `db:"task_id"` + FailedRunID string `db:"failed_run_id"` +} + +// ReplaceAgentPauseRecoveries replaces the active recovery snapshot for an +// agent with its newest failed runs. The limit is the consecutive-failure +// count, so older failures from a previous streak are not selected. +func (r *Repository) ReplaceAgentPauseRecoveries( + ctx context.Context, agentID string, limit int, +) error { + snapshot, err := r.pauseRecoverySnapshot(ctx, agentID, limit) + if err != nil { + return err + } + + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.ExecContext(ctx, tx.Rebind(` + DELETE FROM office_agent_pause_recoveries WHERE agent_id = ? + `), agentID); err != nil { + return err + } + now := time.Now().UTC() + for _, recovery := range snapshot { + if _, err := tx.ExecContext(ctx, tx.Rebind(` + INSERT INTO office_agent_pause_recoveries + (agent_id, task_id, failed_run_id, captured_at) + VALUES (?, ?, ?, ?) + `), recovery.AgentID, recovery.TaskID, recovery.FailedRunID, now); err != nil { + return err + } + } + return tx.Commit() +} + +func (r *Repository) pauseRecoverySnapshot( + ctx context.Context, agentID string, limit int, +) ([]AgentPauseRecovery, error) { + if limit <= 0 { + return nil, nil + } + taskIDExpr := dialect.JSONExtract(r.ro.DriverName(), "payload", "task_id") + rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(fmt.Sprintf(` + SELECT id AS failed_run_id, COALESCE(%s, '') AS task_id + FROM runs + WHERE agent_profile_id = ? AND status = 'failed' + ORDER BY COALESCE(finished_at, requested_at) DESC, + requested_at DESC, id DESC + LIMIT ? + `, taskIDExpr)), agentID, limit) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + seenTasks := make(map[string]struct{}, limit) + var snapshot []AgentPauseRecovery + for rows.Next() { + var row AgentPauseRecovery + if err := rows.StructScan(&row); err != nil { + return nil, err + } + if row.TaskID == "" { + continue + } + if _, seen := seenTasks[row.TaskID]; seen { + continue + } + seenTasks[row.TaskID] = struct{}{} + row.AgentID = agentID + snapshot = append(snapshot, row) + } + if err := rows.Err(); err != nil { + return nil, err + } + return snapshot, nil +} + +// ListAgentPauseRecoveries returns the active task snapshot for an agent. +func (r *Repository) ListAgentPauseRecoveries( + ctx context.Context, agentID string, +) ([]AgentPauseRecovery, error) { + var rows []AgentPauseRecovery + err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(` + SELECT agent_id, task_id, failed_run_id + FROM office_agent_pause_recoveries + WHERE agent_id = ? + ORDER BY captured_at ASC, task_id ASC + `), agentID) + if rows == nil { + rows = []AgentPauseRecovery{} + } + return rows, err +} + +// DeleteAgentPauseRecovery removes one completed or obsolete snapshot row. +func (r *Repository) DeleteAgentPauseRecovery( + ctx context.Context, agentID, taskID string, +) error { + _, err := r.db.ExecContext(ctx, r.db.Rebind(` + DELETE FROM office_agent_pause_recoveries + WHERE agent_id = ? AND task_id = ? + `), agentID, taskID) + return err +} + +// GetLatestRunForAgentTask returns the newest run for an agent/task pair. +// It includes non-failed states so recovery can discard a snapshot when a +// newer run already succeeded or is already queued. +func (r *Repository) GetLatestRunForAgentTask( + ctx context.Context, agentID, taskID string, +) (*models.Run, error) { + taskIDExpr := dialect.JSONExtract(r.ro.DriverName(), "payload", "task_id") + var run models.Run + err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(fmt.Sprintf(` + SELECT * FROM runs + WHERE agent_profile_id = ? AND %s = ? + ORDER BY COALESCE(finished_at, requested_at) DESC, + requested_at DESC, id DESC + LIMIT 1 + `, taskIDExpr)), agentID, taskID).StructScan(&run) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &run, nil +} + // DefaultAgentFailureThreshold is the workspace-level default applied // when neither a per-workspace nor per-agent override is set. const DefaultAgentFailureThreshold = 3 @@ -374,10 +512,8 @@ func (r *Repository) HasPriorTasklessFailedRun( return err == nil, err } -// ListFailedRunsForAgent returns the runs for an agent whose -// status is `failed` and that haven't been dismissed by the user. -// Used by the FailureService when computing which prior per-task -// inbox entries to auto-dismiss on auto-pause. +// ListFailedRunsForAgent returns all failed runs for an agent. The assignee +// change handler uses this list to dismiss obsolete per-task inbox entries. func (r *Repository) ListFailedRunsForAgent( ctx context.Context, agentID string, ) ([]string, error) { diff --git a/apps/backend/internal/office/repository/sqlite/failure_test.go b/apps/backend/internal/office/repository/sqlite/failure_test.go index 30e05c5fb9..489b494108 100644 --- a/apps/backend/internal/office/repository/sqlite/failure_test.go +++ b/apps/backend/internal/office/repository/sqlite/failure_test.go @@ -601,6 +601,64 @@ func TestListFailedRunsForAgent(t *testing.T) { } } +func TestAgentPauseRecoveries_SnapshotsNewestFailures(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + seedSummaryRun(t, repo, "pause-old", "agent-a", "failed", + "2026-01-01 09:00:00", "", "2026-01-01 10:00:00", `{"task_id":"old"}`) + seedSummaryRun(t, repo, "pause-new-a", "agent-a", "failed", + "2026-01-02 09:00:00", "", "2026-01-02 10:00:00", `{"task_id":"new-a"}`) + seedSummaryRun(t, repo, "pause-new-b", "agent-a", "failed", + "2026-01-03 09:00:00", "", "2026-01-03 10:00:00", `{"task_id":"new-b"}`) + + if err := repo.ReplaceAgentPauseRecoveries(ctx, "agent-a", 2); err != nil { + t.Fatalf("replace pause recoveries: %v", err) + } + recoveries, err := repo.ListAgentPauseRecoveries(ctx, "agent-a") + if err != nil { + t.Fatalf("list pause recoveries: %v", err) + } + if len(recoveries) != 2 { + t.Fatalf("recoveries = %+v, want two newest tasks", recoveries) + } + if recoveries[0].TaskID != "new-a" || recoveries[0].FailedRunID != "pause-new-a" { + t.Errorf("first recovery = %+v, want new-a/pause-new-a", recoveries[0]) + } + if recoveries[1].TaskID != "new-b" || recoveries[1].FailedRunID != "pause-new-b" { + t.Errorf("second recovery = %+v, want new-b/pause-new-b", recoveries[1]) + } + + if err := repo.DeleteAgentPauseRecovery(ctx, "agent-a", "new-a"); err != nil { + t.Fatalf("delete pause recovery: %v", err) + } + recoveries, err = repo.ListAgentPauseRecoveries(ctx, "agent-a") + if err != nil { + t.Fatalf("list after delete: %v", err) + } + if len(recoveries) != 1 || recoveries[0].TaskID != "new-b" { + t.Fatalf("recoveries after delete = %+v, want only new-b", recoveries) + } +} + +func TestGetLatestRunForAgentTask_IncludesNewestState(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + seedSummaryRun(t, repo, "latest-failed", "agent-a", "failed", + "2026-01-01 09:00:00", "", "2026-01-01 10:00:00", `{"task_id":"task-a"}`) + seedSummaryRun(t, repo, "latest-queued", "agent-a", "queued", + "2026-01-02 09:00:00", "", "", `{"task_id":"task-a"}`) + + latest, err := repo.GetLatestRunForAgentTask(ctx, "agent-a", "task-a") + if err != nil { + t.Fatalf("get latest run: %v", err) + } + if latest == nil || latest.ID != "latest-queued" || latest.Status != "queued" { + t.Fatalf("latest = %+v, want latest-queued/queued", latest) + } +} + func TestHasPriorTasklessFailedRun(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/office/repository/sqlite/workspace_deletion.go b/apps/backend/internal/office/repository/sqlite/workspace_deletion.go index c4e741e3b3..7e2e151b4c 100644 --- a/apps/backend/internal/office/repository/sqlite/workspace_deletion.go +++ b/apps/backend/internal/office/repository/sqlite/workspace_deletion.go @@ -63,6 +63,7 @@ func (r *Repository) deleteWorkspaceDataTx(ctx context.Context, tx *sqlx.Tx, wor `DELETE FROM runs WHERE agent_profile_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, `DELETE FROM agent_wakeup_requests WHERE agent_profile_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, `DELETE FROM agent_continuation_summaries WHERE agent_profile_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, + `DELETE FROM office_agent_pause_recoveries WHERE agent_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, `DELETE FROM office_agent_memory WHERE agent_profile_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, `DELETE FROM office_agent_instructions WHERE agent_profile_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, `DELETE FROM office_agent_runtime WHERE agent_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`, diff --git a/apps/backend/internal/office/repository/sqlite/workspace_deletion_test.go b/apps/backend/internal/office/repository/sqlite/workspace_deletion_test.go index 9a48c46987..5a26a4e358 100644 --- a/apps/backend/internal/office/repository/sqlite/workspace_deletion_test.go +++ b/apps/backend/internal/office/repository/sqlite/workspace_deletion_test.go @@ -51,6 +51,21 @@ func TestDeleteWorkspaceDataDeletesOwnedOfficeRows(t *testing.T) { assertTableCount(t, db, table, "ws-delete", 0) assertTableCount(t, db, table, "ws-keep", 1) } + var deletedRecoveries, keptRecoveries int + if err := db.Get(&deletedRecoveries, + `SELECT COUNT(*) FROM office_agent_pause_recoveries WHERE agent_id = ?`, + "ws-delete-agent"); err != nil { + t.Fatalf("count deleted pause recoveries: %v", err) + } + if err := db.Get(&keptRecoveries, + `SELECT COUNT(*) FROM office_agent_pause_recoveries WHERE agent_id = ?`, + "ws-keep-agent"); err != nil { + t.Fatalf("count kept pause recoveries: %v", err) + } + if deletedRecoveries != 0 || keptRecoveries != 1 { + t.Fatalf("pause recoveries = deleted:%d kept:%d, want 0 and 1", + deletedRecoveries, keptRecoveries) + } } func TestGetWorkspaceDeletionCounts(t *testing.T) { @@ -120,6 +135,7 @@ func seedWorkspaceDeletionRows(t *testing.T, repo *sqlite.Repository, workspaceI execRaw(t, repo, `INSERT INTO office_run_skills (run_id, skill_id, version, content_hash, materialized_path) VALUES (?, ?, 'v1', 'hash', '/tmp/skill')`, runID, workspaceID+"-skill") execRaw(t, repo, `INSERT INTO agent_wakeup_requests (id, agent_profile_id, source, status, requested_at) VALUES (?, ?, 'test', 'queued', ?)`, workspaceID+"-wakeup", agentID, now) execRaw(t, repo, `INSERT INTO agent_continuation_summaries (agent_profile_id, scope, content) VALUES (?, 'workspace', 'summary')`, agentID) + execRaw(t, repo, `INSERT INTO office_agent_pause_recoveries (agent_id, task_id, failed_run_id) VALUES (?, ?, ?)`, agentID, taskID, runID) execRaw(t, repo, `INSERT INTO office_cost_events (id, agent_profile_id, project_id, occurred_at, created_at) VALUES (?, ?, ?, ?, ?)`, workspaceID+"-cost", agentID, projectID, now, now) execRaw(t, repo, `INSERT INTO office_budget_policies (id, workspace_id, scope_type, scope_id, limit_subcents, period, created_at, updated_at) VALUES (?, ?, 'workspace', ?, 100, 'month', ?, ?)`, workspaceID+"-budget", workspaceID, workspaceID, now, now) execRaw(t, repo, `INSERT INTO office_routines (id, workspace_id, name, created_at, updated_at) VALUES (?, ?, ?, ?, ?)`, routineID, workspaceID, "Routine "+workspaceID, now, now) diff --git a/apps/backend/internal/office/service/failure.go b/apps/backend/internal/office/service/failure.go index 3b4514dd8b..45a2c897e3 100644 --- a/apps/backend/internal/office/service/failure.go +++ b/apps/backend/internal/office/service/failure.go @@ -3,6 +3,7 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "strings" "time" @@ -11,6 +12,7 @@ import ( "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/office/models" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" ) // Inbox item kinds for office-agent-error-handling. @@ -89,28 +91,36 @@ func (s *Service) RecordAgentSuccess(ctx context.Context, agentID string) { } } -// MarkAgentRunFailedFixed dismisses the per-task inbox entry, clears -// the FAILED state on the (task, agent) session, and re-queues a -// run for that pair. Used by the inbox "Mark fixed" action. +// MarkAgentRunFailedFixed clears the FAILED state on the (task, agent) +// session, re-queues a run for that pair, and then dismisses the inbox entry. +// Used by the inbox "Mark fixed" action. func (s *Service) MarkAgentRunFailedFixed( ctx context.Context, userID, runID string, ) error { - if err := s.repo.DismissInboxItem(ctx, userID, InboxKindAgentRunFailed, runID); err != nil { - return fmt.Errorf("dismiss: %w", err) - } run, err := s.repo.GetRun(ctx, runID) if err != nil { // Run vanished (e.g. cancelled by reactivity) — dismissal // alone is the best we can do. No retry needed. + if dismissErr := s.repo.DismissInboxItem( + ctx, userID, InboxKindAgentRunFailed, runID, + ); dismissErr != nil { + return fmt.Errorf("dismiss: %w", dismissErr) + } s.logger.Info("mark fixed: run not found, dismissed only", zap.String("run_id", runID)) return nil } taskID := taskIDFromRunPayload(run.Payload) if taskID == "" { - return nil + return s.repo.DismissInboxItem(ctx, userID, InboxKindAgentRunFailed, runID) } - return s.requeueRunForTask(ctx, run.AgentProfileID, taskID) + if err := s.requeueRunForTask(ctx, run.AgentProfileID, taskID, runID); err != nil { + return fmt.Errorf("requeue run: %w", err) + } + if err := s.repo.DismissInboxItem(ctx, userID, InboxKindAgentRunFailed, runID); err != nil { + return fmt.Errorf("dismiss: %w", err) + } + return nil } // MarkAgentPausedFixed unpauses an auto-paused agent, clears the @@ -120,57 +130,192 @@ func (s *Service) MarkAgentRunFailedFixed( func (s *Service) MarkAgentPausedFixed( ctx context.Context, userID, agentID string, ) error { - if err := s.repo.DismissInboxItem(ctx, userID, InboxKindAgentPausedAfterFails, agentID); err != nil { - return fmt.Errorf("dismiss: %w", err) - } - agent, err := s.repo.GetAgentInstance(ctx, agentID) if err != nil { return fmt.Errorf("get agent: %w", err) } - if !strings.HasPrefix(agent.PauseReason, autoPauseReasonPrefix) { - // Already cleared by something else; nothing to do. - return nil - } - - if err := s.repo.UpdateAgentStatusFields(ctx, agentID, string(agent.Status), ""); err != nil { - return fmt.Errorf("clear pause reason: %w", err) - } - if err := s.repo.ResetAgentConsecutiveFailures(ctx, agentID); err != nil { - s.logger.Warn("reset counter on unpause failed", - zap.String("agent", agentID), zap.Error(err)) - } - - // Re-queue runs for the tasks affected by the pause AND - // auto-dismiss the prior per-task failure entries so unpausing - // doesn't resurface them in the inbox. The pause entry is the - // single source of truth for this batch of failures; once the - // user marks it fixed, the per-task entries are no longer - // actionable on their own. - runIDs, err := s.repo.ListFailedRunsForAgent(ctx, agentID) + autoPaused := strings.HasPrefix(agent.PauseReason, autoPauseReasonPrefix) + recoveries, err := s.loadPauseRecoveries(ctx, agent, autoPaused) if err != nil { - return fmt.Errorf("list failed runs: %w", err) + return err } - seenTasks := map[string]bool{} - for _, wID := range runIDs { - // Auto-dismiss every prior failed run row so it doesn't - // re-emerge in the inbox when the agent unpauses. - _ = s.repo.DismissInboxItem(ctx, autoDismissUserID, InboxKindAgentRunFailed, wID) - w, err := s.repo.GetRun(ctx, wID) + if !autoPaused && len(recoveries) == 0 { + // Both the pause marker and its durable recovery work are gone. + return s.repo.DismissInboxItem(ctx, userID, InboxKindAgentPausedAfterFails, agentID) + } + if err := s.repo.DismissInboxItem( + ctx, userID, InboxKindAgentPausedAfterFails, agentID, + ); err != nil { + return fmt.Errorf("dismiss: %w", err) + } + + if autoPaused { + if err := s.clearAutoPause(ctx, agent); err != nil { + return err + } + if err := s.repo.ResetAgentConsecutiveFailures(ctx, agentID); err != nil { + s.logger.Warn("reset counter on unpause failed", + zap.String("agent", agentID), zap.Error(err)) + } + } + return s.recoverPausedTasks(ctx, agentID, recoveries) +} + +func (s *Service) loadPauseRecoveries( + ctx context.Context, agent *models.AgentInstance, autoPaused bool, +) ([]officesqlite.AgentPauseRecovery, error) { + recoveries, err := s.repo.ListAgentPauseRecoveries(ctx, agent.ID) + if err != nil { + return nil, fmt.Errorf("list pause recoveries: %w", err) + } + if len(recoveries) == 0 && autoPaused { + // Populate the snapshot for auto-paused agents created before this + // table was introduced, then use the same safe recovery path. + if err := s.repo.ReplaceAgentPauseRecoveries( + ctx, agent.ID, agent.ConsecutiveFailures, + ); err != nil { + return nil, fmt.Errorf("capture pause recoveries: %w", err) + } + recoveries, err = s.repo.ListAgentPauseRecoveries(ctx, agent.ID) if err != nil { - continue - } - taskID := taskIDFromRunPayload(w.Payload) - if taskID == "" || seenTasks[taskID] { - continue - } - seenTasks[taskID] = true - if err := s.requeueRunForTask(ctx, agentID, taskID); err != nil { - s.logger.Warn("requeue on unpause failed", - zap.String("agent", agentID), zap.String("task_id", taskID), - zap.Error(err)) + return nil, fmt.Errorf("reload pause recoveries: %w", err) } } + return recoveries, nil +} + +func (s *Service) clearAutoPause( + ctx context.Context, agent *models.AgentInstance, +) error { + for attempt := 0; attempt < 2; attempt++ { + changed, err := s.clearAutoPauseAttempt(ctx, agent) + if err != nil { + return err + } + if changed { + return nil + } + + current, err := s.repo.GetAgentInstance(ctx, agent.ID) + if err != nil { + return fmt.Errorf("reload agent status: %w", err) + } + if !strings.HasPrefix(current.PauseReason, autoPauseReasonPrefix) { + return nil + } + agent = current + } + return fmt.Errorf("clear pause reason: agent status changed") +} + +func (s *Service) clearAutoPauseAttempt( + ctx context.Context, agent *models.AgentInstance, +) (bool, error) { + if agent.Status == models.AgentStatusPaused { + return s.unpauseAgentIfCurrent(ctx, agent) + } + return s.clearPauseReasonIfCurrent(ctx, agent) +} + +func (s *Service) unpauseAgentIfCurrent( + ctx context.Context, agent *models.AgentInstance, +) (bool, error) { + changed, err := s.repo.UpdateAgentStatusFieldsIfCurrent( + ctx, agent.ID, string(models.AgentStatusPaused), + string(models.AgentStatusIdle), "", + ) + if err != nil { + return false, fmt.Errorf("unpause agent: %w", err) + } + if changed { + s.publishAgentStatusChanged( + ctx, agent.ID, agent.WorkspaceID, string(models.AgentStatusIdle), + ) + } + return changed, nil +} + +func (s *Service) clearPauseReasonIfCurrent( + ctx context.Context, agent *models.AgentInstance, +) (bool, error) { + changed, err := s.repo.ClearAgentPauseReasonIfCurrent( + ctx, agent.ID, string(agent.Status), + ) + if err != nil { + return false, fmt.Errorf("clear pause reason: %w", err) + } + if changed { + s.publishAgentStatusChanged( + ctx, agent.ID, agent.WorkspaceID, string(agent.Status), + ) + } + return changed, nil +} + +func (s *Service) recoverPausedTasks( + ctx context.Context, agentID string, + recoveries []officesqlite.AgentPauseRecovery, +) error { + var recoveryErrs []error + for _, recovery := range recoveries { + if err := s.recoverPausedTask(ctx, agentID, recovery); err != nil { + recoveryErrs = append(recoveryErrs, + fmt.Errorf("recover task %s: %w", recovery.TaskID, err)) + } + } + return errors.Join(recoveryErrs...) +} + +func (s *Service) recoverPausedTask( + ctx context.Context, agentID string, + recovery officesqlite.AgentPauseRecovery, +) error { + fields, err := s.repo.GetTaskExecutionFields(ctx, recovery.TaskID) + if errors.Is(err, officesqlite.ErrTaskNotFound) { + return s.discardPauseRecovery(ctx, recovery) + } + if err != nil { + return fmt.Errorf("load task: %w", err) + } + if fields.AssigneeAgentProfileID != agentID { + return s.discardPauseRecovery(ctx, recovery) + } + + latest, err := s.repo.GetLatestRunForAgentTask(ctx, agentID, recovery.TaskID) + if err != nil { + return fmt.Errorf("load latest run: %w", err) + } + if latest == nil || latest.ID != recovery.FailedRunID || + latest.Status != models.RunStatusFailed { + return s.discardPauseRecovery(ctx, recovery) + } + if err := s.requeueRunForTask( + ctx, agentID, recovery.TaskID, recovery.FailedRunID, + ); err != nil { + return fmt.Errorf("requeue task: %w", err) + } + if err := s.repo.DeleteAgentPauseRecovery( + ctx, agentID, recovery.TaskID, + ); err != nil { + return fmt.Errorf("delete recovery: %w", err) + } + _ = s.repo.DismissInboxItem( + ctx, autoDismissUserID, InboxKindAgentRunFailed, recovery.FailedRunID, + ) + return nil +} + +func (s *Service) discardPauseRecovery( + ctx context.Context, recovery officesqlite.AgentPauseRecovery, +) error { + if err := s.repo.DeleteAgentPauseRecovery( + ctx, recovery.AgentID, recovery.TaskID, + ); err != nil { + return fmt.Errorf("delete recovery: %w", err) + } + _ = s.repo.DismissInboxItem( + ctx, autoDismissUserID, InboxKindAgentRunFailed, recovery.FailedRunID, + ) return nil } @@ -306,6 +451,9 @@ func (s *Service) autoPauseAgent( } reason := fmt.Sprintf("%s %d consecutive failures. Last error: %s", autoPauseReasonPrefix, count, truncateForReason(errorMessage)) + if err := s.repo.ReplaceAgentPauseRecoveries(ctx, agentID, count); err != nil { + return fmt.Errorf("capture pause recoveries: %w", err) + } if err := s.repo.UpdateAgentStatusFields( ctx, agentID, string(models.AgentStatusPaused), reason, ); err != nil { @@ -319,10 +467,15 @@ func (s *Service) autoPauseAgent( } func (s *Service) requeueRunForTask( - ctx context.Context, agentID, taskID string, + ctx context.Context, agentID, taskID, failedRunID string, ) error { payload := mustJSONString(map[string]string{"task_id": taskID}) - return s.QueueRun(ctx, agentID, RunReasonManualResumeAfterFailure, payload, "") + identity := failedRunID + if identity == "" { + identity = taskID + } + key := fmt.Sprintf("%s:%s:%s", RunReasonManualResumeAfterFailure, agentID, identity) + return s.QueueRun(ctx, agentID, RunReasonManualResumeAfterFailure, payload, key) } func (s *Service) publishRunFailed( diff --git a/apps/backend/internal/office/service/failure_test.go b/apps/backend/internal/office/service/failure_test.go index e59ce88e74..6487d12660 100644 --- a/apps/backend/internal/office/service/failure_test.go +++ b/apps/backend/internal/office/service/failure_test.go @@ -5,7 +5,6 @@ import ( "encoding/json" "strings" "testing" - "time" "github.com/kandev/kandev/internal/office/models" "github.com/kandev/kandev/internal/office/service" @@ -187,6 +186,275 @@ func TestHandleAgentFailure_RespectsPerAgentThreshold(t *testing.T) { } } +// autoPauseAgent drives HandleAgentFailure exactly `count` times across +// distinct tasks so the agent crosses whatever its effective threshold +// is, then returns the resulting paused agent for assertions. +func autoPauseAgent( + t *testing.T, svc *service.Service, wsID, agentID string, count int, +) *models.AgentInstance { + t.Helper() + ctx := context.Background() + for i := 0; i < count; i++ { + taskID := agentID + "-task-" + uuidish("p", i) + insertSyntheticTask(t, svc, taskID, wsID, agentID) + w := queueAndReadRun(t, svc, agentID, taskID) + if err := svc.HandleAgentFailure(ctx, w, "boom"); err != nil { + t.Fatalf("handle failure %d: %v", i, err) + } + } + agent, err := svc.GetAgentInstance(ctx, agentID) + if err != nil { + t.Fatalf("get agent: %v", err) + } + if !strings.HasPrefix(agent.PauseReason, "Auto-paused:") { + t.Fatalf("setup failed: expected auto-pause after %d failures, got pause_reason=%q", + count, agent.PauseReason) + } + if agent.Status != models.AgentStatusPaused { + t.Fatalf("setup failed: expected paused status, got %q", agent.Status) + } + return agent +} + +// Regression for "Mark fixed does not recover an auto-paused Office +// agent": MarkAgentPausedFixed must actually unpause the agent (write +// idle, not the pre-existing paused status back), zero the counter, +// and leave QueueRun able to succeed again. +func TestMarkAgentPausedFixed_RecoversAgent(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-recover") + autoPauseAgent(t, svc, "ws-1", "agent-recover", 3) + + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-recover"); err != nil { + t.Fatalf("mark fixed: %v", err) + } + + agent, err := svc.GetAgentInstance(ctx, "agent-recover") + if err != nil { + t.Fatalf("get agent: %v", err) + } + if agent.Status != models.AgentStatusIdle { + t.Fatalf("expected status idle after mark fixed, got %q", agent.Status) + } + if agent.ConsecutiveFailures != 0 { + t.Fatalf("expected counter reset, got %d", agent.ConsecutiveFailures) + } + if agent.PauseReason != "" { + t.Fatalf("expected pause reason cleared, got %q", agent.PauseReason) + } + + // Proves the guardAgentStatus rejection is gone: QueueRun must + // succeed now that the agent is actually idle. + if err := svc.QueueRun(ctx, "agent-recover", service.RunReasonTaskAssigned, + mustMarshalJSON(map[string]string{"task_id": "agent-recover-task-a"}), + "agent-recover:post-fix"); err != nil { + t.Fatalf("expected QueueRun to succeed after mark fixed, got: %v", err) + } +} + +func TestMarkAgentPausedFixed_RequeuesEachAffectedTask(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-multi-recover") + autoPauseAgent(t, svc, "ws-1", "agent-multi-recover", 3) + + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-multi-recover"); err != nil { + t.Fatalf("mark fixed: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + seen := map[string]bool{} + for _, run := range runs { + if run.AgentProfileID != "agent-multi-recover" || + run.Reason != service.RunReasonManualResumeAfterFailure { + continue + } + seen[taskIDFromPayload(t, run.Payload)] = true + } + for i := 0; i < 3; i++ { + taskID := "agent-multi-recover-task-" + uuidish("p", i) + if !seen[taskID] { + t.Errorf("missing recovery run for task %s; seen=%v", taskID, seen) + } + } +} + +func TestMarkAgentPausedFixed_RetainsPendingRecoveryForRetry(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-retry-recover") + autoPaused := autoPauseAgent(t, svc, "ws-1", "agent-retry-recover", 3) + if err := svc.UpdateAgentStatusFields(ctx, "agent-retry-recover", + string(models.AgentStatusStopped), autoPaused.PauseReason); err != nil { + t.Fatalf("stop agent: %v", err) + } + + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-retry-recover"); err == nil { + t.Fatal("mark fixed while stopped = nil error, want queue failure") + } + if err := svc.UpdateAgentStatusFields(ctx, "agent-retry-recover", + string(models.AgentStatusIdle), ""); err != nil { + t.Fatalf("restore agent: %v", err) + } + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-retry-recover"); err != nil { + t.Fatalf("retry mark fixed: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.AgentProfileID == "agent-retry-recover" && + run.Reason == service.RunReasonManualResumeAfterFailure { + return + } + } + t.Fatal("retry did not queue a manual recovery run") +} + +func TestMarkAgentRunFailedFixed_LeavesInboxWhenRequeueFails(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-task-retry") + taskID := "task-retry-after-queue-error" + insertSyntheticTask(t, svc, taskID, "ws-1", "agent-task-retry") + run := queueAndReadRun(t, svc, "agent-task-retry", taskID) + if err := svc.HandleAgentFailure(ctx, run, "boom"); err != nil { + t.Fatalf("handle failure: %v", err) + } + if err := svc.UpdateAgentStatusFields(ctx, "agent-task-retry", + string(models.AgentStatusStopped), "manual stop"); err != nil { + t.Fatalf("stop agent: %v", err) + } + + if err := svc.MarkAgentRunFailedFixed(ctx, "user-1", run.ID); err == nil { + t.Fatal("mark fixed while stopped = nil error, want queue failure") + } + dismissed, err := svc.IsInboxItemDismissed( + ctx, "user-1", service.InboxKindAgentRunFailed, run.ID, + ) + if err != nil { + t.Fatalf("check dismissal: %v", err) + } + if dismissed { + t.Fatal("failed requeue dismissed the inbox item before recovery succeeded") + } +} + +func TestMarkAgentPausedFixed_UsesPauseSnapshot(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-snapshot") + oldTaskID := "old-failure-task" + insertSyntheticTask(t, svc, oldTaskID, "ws-1", "agent-snapshot") + oldRun := queueAndReadRun(t, svc, "agent-snapshot", oldTaskID) + if err := svc.HandleAgentFailure(ctx, oldRun, "old failure"); err != nil { + t.Fatalf("old failure: %v", err) + } + svc.RecordAgentSuccess(ctx, "agent-snapshot") + + autoPauseAgent(t, svc, "ws-1", "agent-snapshot", 3) + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-snapshot"); err != nil { + t.Fatalf("mark fixed: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.AgentProfileID == "agent-snapshot" && + run.Reason == service.RunReasonManualResumeAfterFailure && + taskIDFromPayload(t, run.Payload) == oldTaskID { + t.Fatalf("queued stale recovery for task %s", oldTaskID) + } + } +} + +// Threshold-agnostic: the fix must not assume the default threshold +// of 3. A per-agent override to a different value must still recover. +func TestMarkAgentPausedFixed_ThresholdAgnostic(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-threshold") + override := 5 + agent, err := svc.GetAgentInstance(ctx, "agent-threshold") + if err != nil { + t.Fatalf("get agent: %v", err) + } + agent.FailureThreshold = &override + if err := svc.UpdateAgentInstance(ctx, agent); err != nil { + t.Skipf("update agent unsupported: %v", err) + } + + autoPauseAgent(t, svc, "ws-1", "agent-threshold", override) + + if err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-threshold"); err != nil { + t.Fatalf("mark fixed: %v", err) + } + + agent, err = svc.GetAgentInstance(ctx, "agent-threshold") + if err != nil { + t.Fatalf("get agent: %v", err) + } + if agent.Status != models.AgentStatusIdle { + t.Fatalf("expected status idle after mark fixed, got %q", agent.Status) + } + if agent.ConsecutiveFailures != 0 { + t.Fatalf("expected counter reset, got %d", agent.ConsecutiveFailures) + } + if agent.PauseReason != "" { + t.Fatalf("expected pause reason cleared, got %q", agent.PauseReason) + } +} + +// If the agent left "paused" through some other path (e.g. a manual +// stop) before the inbox dismissal is processed, mark-fixed must not +// resurrect it into idle — only the stale pause reason is cleared. +// The requeue attempts that follow then genuinely fail (the agent is +// stopped), and that failure must be returned, not swallowed. +func TestMarkAgentPausedFixed_NonPausedAgentGuardAndRequeueError(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "agent-guarded") + agent := autoPauseAgent(t, svc, "ws-1", "agent-guarded", 3) + + // Simulate the agent having moved to stopped via another path while + // the stale Auto-paused reason is still on the row. + if err := svc.UpdateAgentStatusFields(ctx, "agent-guarded", + string(models.AgentStatusStopped), agent.PauseReason); err != nil { + t.Fatalf("force stopped: %v", err) + } + + err := svc.MarkAgentPausedFixed(ctx, "user-1", "agent-guarded") + if err == nil || !strings.Contains(err.Error(), "requeue task") { + t.Fatalf("expected a requeue error, got: %v", err) + } + + after, getErr := svc.GetAgentInstance(ctx, "agent-guarded") + if getErr != nil { + t.Fatalf("get agent: %v", getErr) + } + if after.Status != models.AgentStatusStopped { + t.Fatalf("expected status to remain stopped, got %q", after.Status) + } + if after.PauseReason != "" { + t.Fatalf("expected stale pause reason cleared, got %q", after.PauseReason) + } +} + // Pins that reassigning a task auto-dismisses the per-task inbox // entry for the OLD agent without resetting that agent's counter. func TestOnAssigneeChanged_DismissesPriorEntryWithoutResettingCounter(t *testing.T) { @@ -226,6 +494,4 @@ func TestOnAssigneeChanged_DismissesPriorEntryWithoutResettingCounter(t *testing if !dismissed { t.Fatalf("expected run %s dismissed via _auto", w.ID) } - // Sanity: settling time so any async event handlers complete. - time.Sleep(10 * time.Millisecond) } diff --git a/apps/backend/internal/runs/repository/sqlite/runs.go b/apps/backend/internal/runs/repository/sqlite/runs.go index eb2d530de4..9438b5ce1f 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs.go +++ b/apps/backend/internal/runs/repository/sqlite/runs.go @@ -455,11 +455,11 @@ func (r *Repository) CoalesceRun( taskID := taskIDFromPayload(payload) taskPredicate := "" args := []interface{}{payload, agentInstanceID, reason, cutoff, commentkeys.TaskCommentPrefix + "%"} - // Assignment wakes are task-specific: merging two tasks for the same - // agent would replace the first task's payload and silently drop its - // launch. Other reasons, such as task comments, intentionally retain - // their existing cross-task coalescing behaviour. - if taskID != "" && reason == "task_assigned" { + // Task assignment and failure recovery wakes are task-specific: merging + // two tasks for the same agent would replace the first task's payload and + // silently drop its launch. Other reasons, such as task comments, + // intentionally retain their existing cross-task coalescing behaviour. + if taskID != "" && isTaskScopedCoalescingReason(reason) { taskPredicate = fmt.Sprintf(" AND %s = ?", dialect.JSONExtract(r.db.DriverName(), "payload", "task_id")) args = append(args, taskID) } @@ -487,6 +487,10 @@ func (r *Repository) CoalesceRun( return rows > 0, nil } +func isTaskScopedCoalescingReason(reason string) bool { + return reason == "task_assigned" || reason == "manual_resume_after_failure" +} + func taskIDFromPayload(payload string) string { var raw map[string]any if err := json.Unmarshal([]byte(payload), &raw); err != nil { diff --git a/apps/backend/internal/runs/service/service_test.go b/apps/backend/internal/runs/service/service_test.go index e01b8ce201..f7b31046e9 100644 --- a/apps/backend/internal/runs/service/service_test.go +++ b/apps/backend/internal/runs/service/service_test.go @@ -115,6 +115,32 @@ func TestQueueRun_DoesNotCoalesceTaskAssignedAcrossTasks(t *testing.T) { } } +func TestQueueRun_DoesNotCoalesceManualRecoveryAcrossTasks(t *testing.T) { + svc, _, repo := newTestServiceWithRepo(t) + ctx := context.Background() + + for _, taskID := range []string{"task-a", "task-b"} { + if _, err := svc.QueueRun(ctx, runsservice.QueueRunRequest{ + AgentProfileID: "agent-primary", + TaskID: taskID, + Reason: "manual_resume_after_failure", + IdempotencyKey: "manual_resume_after_failure:" + taskID, + }); err != nil { + t.Fatalf("queue %s: %v", taskID, err) + } + } + + var count int + if err := repo.Reader().GetContext(ctx, &count, + `SELECT COUNT(*) FROM runs WHERE agent_profile_id = ?`, + "agent-primary"); err != nil { + t.Fatalf("count queued runs: %v", err) + } + if count != 2 { + t.Fatalf("manual recovery runs = %d, want 2", count) + } +} + func TestQueueRun_UsesRequestAgentProfileIDAndAddsEnvelopePayload(t *testing.T) { svc, _, repo := newTestServiceWithRepo(t) ctx := context.Background()