Durable pause recovery table
apps/backend/internal/office/repository/sqlite/base_migrations.go ↗The migration creates office_agent_pause_recoveries so the set of tasks to recover survives restarts and threshold changes.
New table
_, _ = 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)`)
Compare-and-set agent status helpers
apps/backend/internal/office/repository/sqlite/agents.go ↗The helpers only write when the agent still has the expected status, so a concurrent stop or manual change is not overwritten.
CAS update
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
}
Clear reason only
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
}
Snapshot capture and lookup
apps/backend/internal/office/repository/sqlite/failure.go ↗The repository builds a snapshot from the newest failed runs and stores it atomically for later recovery.
Replace snapshot
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()
}
Latest run check
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
}
MarkAgentPausedFixed now recovers from snapshot
apps/backend/internal/office/service/failure.go ↗The service loads the snapshot, clears pause with a CAS that respects current status, and requeues each task only when it is still assigned and still failed.
Load and guard
func (s *Service) MarkAgentPausedFixed(
ctx context.Context, userID, agentID string,
) error {
agent, err := s.repo.GetAgentInstance(ctx, agentID)
if err != nil {
return fmt.Errorf("get agent: %w", err)
}
autoPaused := strings.HasPrefix(agent.PauseReason, autoPauseReasonPrefix)
recoveries, err := s.loadPauseRecoveries(ctx, agent, autoPaused)
if err != nil {
return err
}
if !autoPaused && len(recoveries) == 0 {
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)
}
CAS unpause with retry
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)
}
Per-task recovery
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
}
Mark fixed for single run and idempotent requeue
apps/backend/internal/office/service/failure.go ↗The single-run path now requeues before dismiss and keeps the inbox entry when the queue fails, and requeues use a stable idempotency key.
Requeue before dismiss
func (s *Service) MarkAgentRunFailedFixed(
ctx context.Context, userID, runID string,
) error {
run, err := s.repo.GetRun(ctx, runID)
if err != nil {
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 s.repo.DismissInboxItem(ctx, userID, InboxKindAgentRunFailed, runID)
}
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
}
Idempotent key
func (s *Service) requeueRunForTask(
ctx context.Context, agentID, taskID, failedRunID string,
) error {
payload := mustJSONString(map[string]string{"task_id": taskID})
identity := failedRunID
if identity == "" {
identity = taskID
}
key := fmt.Sprintf("%s:%s:%s", RunReasonManualResumeAfterFailure, agentID, identity)
return s.QueueRun(ctx, agentID, RunReasonManualResumeAfterFailure, payload, key)
}
Capture at pause
func (s *Service) autoPauseAgent(
ctx context.Context, agentID string, count int, errorMessage string,
) error {
agent, err := s.repo.GetAgentInstance(ctx, agentID)
if err != nil {
return fmt.Errorf("get agent: %w", err)
}
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 {
return fmt.Errorf("set pause reason: %w", err)
}
s.logger.Warn("agent auto-paused",
zap.String("agent", agentID), zap.String("name", agent.Name),
zap.Int("consecutive_failures", count))
s.publishAgentAutoPaused(ctx, agent, count, errorMessage)
return nil
}
Cleanup and coalescing
apps/backend/internal/office/repository/sqlite/workspace_deletion.go ↗Workspace and agent deletion now remove snapshot rows, and manual recovery runs coalesce per task.
Workspace delete
`DELETE FROM office_agent_pause_recoveries WHERE agent_id IN (SELECT id FROM agent_profiles WHERE workspace_id = ?)`
Agent delete
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(
`DELETE FROM office_agent_pause_recoveries WHERE agent_id = ?`), id)
return err
}
Task-scoped coalescing
func isTaskScopedCoalescingReason(reason string) bool {
return reason == "task_assigned" || reason == "manual_resume_after_failure"
}
