diff --git a/apps/backend/internal/backendapp/adapters_office.go b/apps/backend/internal/backendapp/adapters_office.go index b07b6c09a0..b4e1669a45 100644 --- a/apps/backend/internal/backendapp/adapters_office.go +++ b/apps/backend/internal/backendapp/adapters_office.go @@ -3,6 +3,7 @@ package backendapp import ( "context" "database/sql" + "errors" "fmt" "github.com/kandev/kandev/internal/github" @@ -263,13 +264,23 @@ func (a *routineWakeupAdapter) CreateWakeupRequest( if req.IdempotencyKey != "" { row.IdempotencyKey = sql.NullString{String: req.IdempotencyKey, Valid: true} } - return a.repo.CreateWakeupRequest(ctx, row) + if err := a.repo.CreateWakeupRequest(ctx, row); err != nil { + if errors.Is(err, officesqlite.ErrWakeupIdempotencyConflict) { + return officeroutines.ErrWakeupAlreadyRequested + } + return err + } + return nil } func (a *routineWakeupAdapter) Dispatch(ctx context.Context, requestID string) error { return a.dispatcher.Dispatch(ctx, requestID) } +func (a *routineWakeupAdapter) FailWakeupRequest(ctx context.Context, requestID, reason string) error { + return a.repo.MarkWakeupRequestFailed(ctx, requestID, reason) +} + // configSyncerAdapter bridges config.ConfigService to the onboarding.ConfigSyncer // interface by converting config.ImportResult to onboarding.ApplyResult. type configSyncerAdapter struct { diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 9355d8ad07..20dfbbb20f 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -2129,6 +2129,12 @@ func buildOfficeFeatureServices( }) routineSvc.SetWorkflowEnsurer(&workflowEnsurerAdapter{repo: taskRepo}) routineSvc.SetTaskCreator(&taskCreatorAdapter{taskSvc: services.Task}) + // office-routine-runs: closes out a heavy routine run when its + // linked task reaches a terminal step, so the routine's next fire + // isn't gated by a task that already finished. + if services.Office != nil { + services.Office.SetRoutineRunSyncer(routineSvc) + } approvalSvc := officeapprovals.NewApprovalService(repo, log, activity, services.Office) approvalSvc.SetAgentWriter(agentSvc) channelSvc := officechannels.NewChannelService(repo, log, activity, agentSvc) diff --git a/apps/backend/internal/office/repository/sqlite/base.go b/apps/backend/internal/office/repository/sqlite/base.go index 0127cebf2d..8afa9c7097 100644 --- a/apps/backend/internal/office/repository/sqlite/base.go +++ b/apps/backend/internal/office/repository/sqlite/base.go @@ -462,6 +462,12 @@ func (r *Repository) createRoutineTables() error { created_at TIMESTAMP NOT NULL, FOREIGN KEY (routine_id) REFERENCES office_routines(id) ON DELETE CASCADE ); + CREATE INDEX IF NOT EXISTS idx_office_routine_runs_active_fingerprint + ON office_routine_runs(routine_id, dispatch_fingerprint, created_at DESC) + WHERE status = 'task_created'; + CREATE INDEX IF NOT EXISTS idx_office_routine_runs_linked_task + ON office_routine_runs(linked_task_id, created_at DESC) + WHERE linked_task_id != ''; `) return err } @@ -677,10 +683,10 @@ func (r *Repository) createContinuationSummaryTable() error { // dispatcher coalesces / claims / drops them per the agent's policy // before creating the corresponding runs row. // -// idempotency_key carries source-level dedup (e.g. heartbeat:: -// ) — duplicates within the window land on the partial -// UNIQUE index and are rejected. The "" sentinel keeps the index free -// for rows without a key. +// idempotency_key carries source-level dedup (for example, a cron routine +// trigger and minute bucket). Duplicates with the same source identity land +// on the partial UNIQUE index and are rejected. The "" sentinel keeps the +// index free for rows without a key. func (r *Repository) createAgentWakeupRequestTable() error { _, err := r.db.Exec(` CREATE TABLE IF NOT EXISTS agent_wakeup_requests ( diff --git a/apps/backend/internal/office/repository/sqlite/blockers.go b/apps/backend/internal/office/repository/sqlite/blockers.go index 4cd8c929a4..65247c065e 100644 --- a/apps/backend/internal/office/repository/sqlite/blockers.go +++ b/apps/backend/internal/office/repository/sqlite/blockers.go @@ -197,6 +197,15 @@ func (r *Repository) DeleteTaskBlockersForTask(ctx context.Context, taskID strin return err } +// Task lifecycle state values as stored in the shared `tasks` table, +// shared by every office-repo reader of terminal task state +// (IsTaskInTerminalStep, GetTaskTerminalStatus). +const ( + taskStateCompleted = "COMPLETED" + taskStateFailed = "FAILED" + taskStateCancelled = "CANCELLED" +) + func (r *Repository) IsTaskInTerminalStep(ctx context.Context, taskID string) (bool, error) { var state string err := r.ro.QueryRowxContext(ctx, r.ro.Rebind( @@ -204,7 +213,7 @@ func (r *Repository) IsTaskInTerminalStep(ctx context.Context, taskID string) (b if err != nil { return false, err } - return state == "COMPLETED" || state == "CANCELLED", nil + return state == taskStateCompleted || state == taskStateFailed || state == taskStateCancelled, nil } // GetTaskAssignee returns the agent currently driving a task. Resolves diff --git a/apps/backend/internal/office/repository/sqlite/routines.go b/apps/backend/internal/office/repository/sqlite/routines.go index 3bbfee28b1..db61f8d320 100644 --- a/apps/backend/internal/office/repository/sqlite/routines.go +++ b/apps/backend/internal/office/repository/sqlite/routines.go @@ -321,7 +321,11 @@ func (r *Repository) ListAllRuns(ctx context.Context, workspaceID string, limit return runs, nil } -// GetActiveRunForFingerprint returns an active run matching the fingerprint. +// GetActiveRunForFingerprint returns the newest active run matching the +// fingerprint, if any. "Active" means status = task_created (the only +// status a run can gate another fire from — see +// routines.RoutineService's materialise* methods). The caller repairs +// terminal, archived, or missing linked tasks before it applies policy. func (r *Repository) GetActiveRunForFingerprint( ctx context.Context, routineID, fingerprint string, ) (*models.RoutineRun, error) { @@ -341,6 +345,77 @@ func (r *Repository) GetActiveRunForFingerprint( return &run, nil } +// GetRoutineRunByLinkedTaskID returns the most recent run linked to +// taskID, or nil if no run is linked to it (most tasks aren't +// routine-created). Used by SyncRunStatus to find the run to close out +// when its task reaches a terminal step. taskID must be non-empty: +// linked_task_id defaults to "" for every lightweight run, so an empty +// taskID would match (and let a caller rewrite) an arbitrary lightweight +// run instead of correctly finding nothing. +func (r *Repository) GetRoutineRunByLinkedTaskID( + ctx context.Context, taskID string, +) (*models.RoutineRun, error) { + if taskID == "" { + return nil, nil + } + var run models.RoutineRun + err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(` + SELECT * FROM office_routine_runs + WHERE linked_task_id = ? + ORDER BY created_at DESC LIMIT 1 + `), taskID).StructScan(&run) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &run, nil +} + +// GetTaskTerminalStatus reports the linked task outcome: "done" for +// COMPLETED, "failed" for FAILED, "cancelled" for CANCELLED or archived +// tasks, "missing" when the task row was deleted, and "" otherwise. +// Reads the shared tasks table directly because the routines gate must +// distinguish an active task from a terminal, archived, or missing task. +func (r *Repository) GetTaskTerminalStatus(ctx context.Context, taskID string) (string, error) { + var state string + var archivedAt sql.NullTime + err := r.ro.QueryRowxContext(ctx, r.ro.Rebind( + `SELECT COALESCE(state, ''), archived_at FROM tasks WHERE id = ?`), taskID).Scan(&state, &archivedAt) + if err == sql.ErrNoRows { + return "missing", nil + } + if err != nil { + return "", err + } + if archivedAt.Valid { + return "cancelled", nil + } + switch state { + case taskStateCompleted: + return "done", nil + case taskStateFailed: + return "failed", nil + case taskStateCancelled: + return "cancelled", nil + default: + return "", nil + } +} + +// TouchRoutineLastRun updates only last_run_at (+ updated_at) on a +// routine. Deliberately narrower than UpdateRoutine, whose whole-row +// snapshot write can revert concurrent edits to other columns (see +// UpdateRoutineConfigFields above) — a dispatch in flight should never +// clobber a config change made while it ran, or vice versa. +func (r *Repository) TouchRoutineLastRun(ctx context.Context, routineID string, at time.Time) error { + _, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE office_routines SET last_run_at = ?, updated_at = ? WHERE id = ? + `), at, time.Now().UTC(), routineID) + return err +} + // UpdateRunStatus updates a run's status and optionally its linked task. func (r *Repository) UpdateRunStatus( ctx context.Context, runID string, status models.RoutineRunStatus, linkedTaskID string, @@ -354,6 +429,31 @@ func (r *Repository) UpdateRunStatus( return err } +// UpdateRunStatusIfTaskCreated closes a run out with a terminal status, +// but only while it is still 'task_created' — the one status a routine +// run can gate its fingerprint from. The WHERE clause makes this an +// atomic claim: SyncRunStatus (TaskMoved-driven) and the concurrency +// gate's own inline check (applyConcurrencyPolicy) can both observe the +// same terminal task, and this is what lets exactly one of them win +// instead of a read-then-write race rewriting an already-closed run's +// status or completed_at. Returns whether this call was the one that +// closed it. +func (r *Repository) UpdateRunStatusIfTaskCreated( + ctx context.Context, runID string, status models.RoutineRunStatus, linkedTaskID string, +) (bool, error) { + now := time.Now().UTC() + res, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE office_routine_runs + SET status = ?, linked_task_id = ?, completed_at = ? + WHERE id = ? AND status = 'task_created' + `), status, linkedTaskID, now, runID) + if err != nil { + return false, err + } + rows, err := res.RowsAffected() + return rows > 0, err +} + // UpdateRunCoalesced marks a run as coalesced into another run. func (r *Repository) UpdateRunCoalesced(ctx context.Context, runID, coalescedIntoRunID string) error { now := time.Now().UTC() diff --git a/apps/backend/internal/office/repository/sqlite/routines_test.go b/apps/backend/internal/office/repository/sqlite/routines_test.go index d6bc4e53e2..4e0c3129c5 100644 --- a/apps/backend/internal/office/repository/sqlite/routines_test.go +++ b/apps/backend/internal/office/repository/sqlite/routines_test.go @@ -198,6 +198,7 @@ func TestRoutineRun_ActiveFingerprint(t *testing.T) { if none != nil { t.Error("expected nil for non-matching fingerprint") } + } func TestRoutineRun_Create(t *testing.T) { @@ -229,3 +230,135 @@ func TestRoutineRun_Create(t *testing.T) { t.Fatal("expected run ID to be set") } } + +// TestGetTaskTerminalStatus brings up just enough of the shared `tasks` +// table (owned by internal/task/repository/sqlite in the real app) for +// GetTaskTerminalStatus to query against it. office/repository/sqlite's +// own initSchema deliberately does not create this table (see +// TestInitSchema_AllTablesExist) — it shares the production DB's table +// rather than owning it. +func TestGetTaskTerminalStatus(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + if _, err := repo.ExecRaw(ctx, `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if _, err := repo.ExecRaw(ctx, `INSERT INTO tasks (id, state, archived_at) VALUES (?, ?, NULL), (?, ?, NULL), (?, ?, NULL), (?, ?, NULL), (?, ?, ?)`, + "task-done", "COMPLETED", "task-cancelled", "CANCELLED", "task-failed", "FAILED", "task-open", "IN_PROGRESS", "task-archived", "IN_PROGRESS", time.Now().UTC()); err != nil { + t.Fatalf("seed tasks: %v", err) + } + + cases := []struct { + taskID string + want string + }{ + {"task-done", "done"}, + {"task-cancelled", "cancelled"}, + {"task-failed", "failed"}, + {"task-open", ""}, + {"task-archived", "cancelled"}, + {"task-missing", "missing"}, + } + for _, c := range cases { + got, err := repo.GetTaskTerminalStatus(ctx, c.taskID) + if err != nil { + t.Fatalf("GetTaskTerminalStatus(%q): %v", c.taskID, err) + } + if got != c.want { + t.Errorf("GetTaskTerminalStatus(%q) = %q, want %q", c.taskID, got, c.want) + } + } +} + +func TestUpdateRunStatusIfTaskCreated(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Conditional Close", + TaskTemplate: "{}", + Status: "active", + ConcurrencyPolicy: "skip_if_active", + Variables: "{}", + } + if err := repo.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + run := &models.RoutineRun{ + RoutineID: routine.ID, + Source: "manual", + Status: "task_created", + TriggerPayload: "{}", + DispatchFingerprint: "fp-cond", + LinkedTaskID: "task-1", + } + if err := repo.CreateRoutineRun(ctx, run); err != nil { + t.Fatalf("create run: %v", err) + } + + closed, err := repo.UpdateRunStatusIfTaskCreated(ctx, run.ID, models.RoutineRunStatusDone, "task-1") + if err != nil { + t.Fatalf("first close: %v", err) + } + if !closed { + t.Fatal("expected first close to succeed from task_created") + } + + // A second attempt against the now-closed row must not win — this is + // the race SyncRunStatus and the concurrency gate's inline check can + // both hit, and it must not rewrite a "done" row to "cancelled" or + // move completed_at. + closedAgain, err := repo.UpdateRunStatusIfTaskCreated(ctx, run.ID, models.RoutineRunStatusCancelled, "task-1") + if err != nil { + t.Fatalf("second close: %v", err) + } + if closedAgain { + t.Fatal("second close against an already-closed run must report false") + } + + got, err := repo.GetRoutineRunByLinkedTaskID(ctx, "task-1") + if err != nil { + t.Fatalf("get by linked task: %v", err) + } + if got.Status != models.RoutineRunStatusDone { + t.Errorf("status = %q, want done (unchanged by the losing second close)", got.Status) + } +} + +func TestGetRoutineRunByLinkedTaskID_EmptyTaskIDGuarded(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Empty Guard", + TaskTemplate: "", + Status: "active", + ConcurrencyPolicy: "always_create", + Variables: "{}", + } + if err := repo.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + // Lightweight run: linked_task_id defaults to ''. + run := &models.RoutineRun{ + RoutineID: routine.ID, + Source: "manual", + Status: "done", + TriggerPayload: "{}", + DispatchFingerprint: "fp-empty", + } + if err := repo.CreateRoutineRun(ctx, run); err != nil { + t.Fatalf("create run: %v", err) + } + + got, err := repo.GetRoutineRunByLinkedTaskID(ctx, "") + if err != nil { + t.Fatalf("lookup with empty taskID: %v", err) + } + if got != nil { + t.Errorf("expected nil for empty taskID, got run %q (would have matched an arbitrary lightweight run)", got.ID) + } +} diff --git a/apps/backend/internal/office/repository/sqlite/wakeup_requests.go b/apps/backend/internal/office/repository/sqlite/wakeup_requests.go index 9f3c3d2aa9..5eac5f8086 100644 --- a/apps/backend/internal/office/repository/sqlite/wakeup_requests.go +++ b/apps/backend/internal/office/repository/sqlite/wakeup_requests.go @@ -10,14 +10,15 @@ import ( ) // Wakeup request status constants — kept in sync with the spec's -// "queued" | "claimed" | "coalesced" | "skipped" enum. Terminal states -// (claimed, coalesced, skipped) all stamp finished_at; only "queued" -// is visible to the dispatcher's claim path. +// "queued" | "claimed" | "coalesced" | "skipped" | "failed" enum. +// Terminal states all stamp finished_at; only "queued" is visible to the +// dispatcher's claim path. const ( WakeupStatusQueued = "queued" WakeupStatusClaimed = "claimed" WakeupStatusCoalesced = "coalesced" WakeupStatusSkipped = "skipped" + WakeupStatusFailed = "failed" ) // ErrWakeupIdempotencyConflict is returned by CreateWakeupRequest when @@ -295,6 +296,19 @@ func (r *Repository) MarkWakeupRequestSkipped( return err } +// MarkWakeupRequestFailed transitions a request to a terminal failed state +// when direct dispatch cannot claim it. This prevents a queued row from +// appearing healthy when no background poller can retry it. +func (r *Repository) MarkWakeupRequestFailed(ctx context.Context, id, reason string) error { + now := time.Now().UTC() + _, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE agent_wakeup_requests + SET status = ?, reason = ?, finished_at = ? + WHERE id = ? AND status = ? + `), WakeupStatusFailed, reason, now, id, WakeupStatusQueued) + return err +} + // isUniqueConstraintErr returns true when err looks like a SQLite // UNIQUE constraint violation. Driver-specific error inspection would // be cleaner but the go-sqlite3 driver's typed error doesn't surface diff --git a/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go b/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go index e82c943cc8..38c63c3417 100644 --- a/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go +++ b/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go @@ -147,6 +147,38 @@ func TestWakeupRequest_MarkSkipped(t *testing.T) { } } +func TestWakeupRequest_MarkFailed(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + _ = repo.CreateWakeupRequest(ctx, &sqlite.WakeupRequest{ + ID: "w-1", AgentProfileID: "agent-1", Source: "routine", + }) + if err := repo.MarkWakeupRequestFailed(ctx, "w-1", "dispatch failed"); err != nil { + t.Fatalf("mark failed: %v", err) + } + got, err := repo.GetWakeupRequest(ctx, "w-1") + if err != nil { + t.Fatalf("get: %v", err) + } + if got.Status != sqlite.WakeupStatusFailed { + t.Errorf("status: got %q", got.Status) + } + if got.Reason != "dispatch failed" { + t.Errorf("reason: got %q", got.Reason) + } + if !got.FinishedAt.Valid { + t.Error("finished_at should be set") + } + rows, err := repo.ListQueuedWakeupRequestsForAgent(ctx, "agent-1") + if err != nil { + t.Fatalf("list queued: %v", err) + } + if len(rows) != 0 { + t.Fatalf("failed request remains queued: %+v", rows) + } +} + // TestWakeupRequest_MarkCoalesced creates a runs row, then a wakeup // request, then marks the request coalesced into the run. Verifies the // status transition, coalesced_count bump, and that context_snapshot diff --git a/apps/backend/internal/office/routines/handler.go b/apps/backend/internal/office/routines/handler.go index 4c7a9ce54f..e6a591b8d7 100644 --- a/apps/backend/internal/office/routines/handler.go +++ b/apps/backend/internal/office/routines/handler.go @@ -252,7 +252,8 @@ func (h *Handler) fireWebhookTrigger(c *gin.Context) { return } - run, err := h.svc.DispatchRoutineRun(ctx, routine, trigger, "webhook", vars) + run, err := h.svc.DispatchRoutineRunWithIdempotencyKey( + ctx, routine, trigger, "webhook", vars, c.GetHeader("Idempotency-Key")) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return diff --git a/apps/backend/internal/office/routines/service.go b/apps/backend/internal/office/routines/service.go index 6c616bde7c..3879a42d4b 100644 --- a/apps/backend/internal/office/routines/service.go +++ b/apps/backend/internal/office/routines/service.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "encoding/json" + "errors" "fmt" "time" @@ -16,6 +17,16 @@ import ( taskservice "github.com/kandev/kandev/internal/task/service" ) +// ErrWakeupAlreadyRequested is the routines-package-local signal that a +// WakeupEnqueuer.CreateWakeupRequest call lost an idempotency race: a +// request for this fire already exists, enqueued by another caller (a +// cron tick and a manual fire landing in the same dedup bucket, for +// example). That is success by another route, not a failure — the +// concrete adapter translates the sqlite layer's +// ErrWakeupIdempotencyConflict into this sentinel so routines never +// imports office/repository/sqlite directly (see routineWakeupAdapter). +var ErrWakeupAlreadyRequested = errors.New("wakeup request already requested") + // Repository is the persistence interface required by RoutineService. type Repository interface { CreateRoutine(ctx context.Context, routine *Routine) error @@ -36,8 +47,23 @@ type Repository interface { ListRoutineRuns(ctx context.Context, routineID string, limit, offset int) ([]*RoutineRun, error) ListAllRuns(ctx context.Context, workspaceID string, limit int) ([]*RoutineRun, error) GetActiveRunForFingerprint(ctx context.Context, routineID, fingerprint string) (*RoutineRun, error) + GetRoutineRunByLinkedTaskID(ctx context.Context, taskID string) (*RoutineRun, error) UpdateRunStatus(ctx context.Context, runID string, status models.RoutineRunStatus, linkedTaskID string) error + // UpdateRunStatusIfTaskCreated closes out a run only while it is still + // task_created, returning whether this call was the one that closed + // it. Used by both the TaskMoved-driven SyncRunStatus and the + // concurrency gate's own inline check (applyConcurrencyPolicy), which + // can both observe the same terminal task; the conditional WHERE + // clause is what lets exactly one of them win instead of a + // read-then-write race between them. + UpdateRunStatusIfTaskCreated(ctx context.Context, runID string, status models.RoutineRunStatus, linkedTaskID string) (bool, error) UpdateRunCoalesced(ctx context.Context, runID, coalescedIntoRunID string) error + TouchRoutineLastRun(ctx context.Context, routineID string, at time.Time) error + // GetTaskTerminalStatus reads a task's real lifecycle state directly + // (not via the TaskMoved event — see applyConcurrencyPolicy) and + // reports "" when it is active, "done", "failed", "cancelled", or + // "missing" for the other outcomes. + GetTaskTerminalStatus(ctx context.Context, taskID string) (string, error) } // WakeupEnqueuer is the slim surface routines need to enqueue + dispatch @@ -50,6 +76,7 @@ type Repository interface { type WakeupEnqueuer interface { CreateWakeupRequest(ctx context.Context, req *WakeupRequest) error Dispatch(ctx context.Context, requestID string) error + FailWakeupRequest(ctx context.Context, requestID, reason string) error } // WakeupRequest mirrors *office/repository/sqlite.WakeupRequest with the @@ -466,7 +493,21 @@ func (s *RoutineService) DispatchRoutineRun( source string, provided map[string]string, ) (*RoutineRun, error) { - return s.dispatchRoutineRun(ctx, routine, trigger, source, provided, 0) + return s.dispatchRoutineRun(ctx, routine, trigger, source, provided, "", 0) +} + +// DispatchRoutineRunWithIdempotencyKey dispatches a fire with an explicit +// source request identity. Webhook callers can use this header to make +// retries idempotent without collapsing unrelated deliveries. +func (s *RoutineService) DispatchRoutineRunWithIdempotencyKey( + ctx context.Context, + routine *Routine, + trigger *RoutineTrigger, + source string, + provided map[string]string, + idempotencyKey string, +) (*RoutineRun, error) { + return s.dispatchRoutineRun(ctx, routine, trigger, source, provided, idempotencyKey, 0) } // DispatchRoutineRunWithMissed is the cron-tick entry point that @@ -484,7 +525,7 @@ func (s *RoutineService) DispatchRoutineRunWithMissed( provided map[string]string, missedTicks int, ) (*RoutineRun, error) { - return s.dispatchRoutineRun(ctx, routine, trigger, source, provided, missedTicks) + return s.dispatchRoutineRun(ctx, routine, trigger, source, provided, "", missedTicks) } func (s *RoutineService) dispatchRoutineRun( @@ -493,6 +534,7 @@ func (s *RoutineService) dispatchRoutineRun( trigger *RoutineTrigger, source string, provided map[string]string, + idempotencyKey string, missedTicks int, ) (*RoutineRun, error) { now := time.Now().UTC() @@ -531,10 +573,14 @@ func (s *RoutineService) dispatchRoutineRun( return run, nil } - if err := s.materialiseRoutineRun(ctx, routine, run, tmpl, title, description, vars, missedTicks); err != nil { + if err := s.materialiseRoutineRun(ctx, routine, run, tmpl, title, description, vars, source, idempotencyKey, missedTicks); err != nil { return run, err } + if err := s.repo.TouchRoutineLastRun(ctx, routine.ID, now); err != nil { + s.logger.Warn("touch routine last_run_at", + zap.String("routine", routine.Name), zap.Error(err)) + } routine.LastRunAt = &now s.logger.Info("routine run dispatched", zap.String("routine", routine.Name), @@ -545,8 +591,12 @@ func (s *RoutineService) dispatchRoutineRun( } // materialiseRoutineRun branches on tmpl.Title to choose the lightweight -// (taskless) or heavy (real task) path. Both paths transition the run -// to models.RoutineRunStatusTaskCreated; only the heavy path attaches a real task id. +// (taskless) or heavy (real task) path. The heavy path transitions the +// run to models.RoutineRunStatusTaskCreated and attaches the new task's +// id — that status stays until the linked task reaches a terminal step +// (see SyncRunStatus). The lightweight path has no task to wait on, so +// it resolves to a terminal status (done/failed) immediately; see +// materialiseLightweightRoutineRun. // missedTicks is forwarded to the lightweight wakeup payload; the heavy // path doesn't attribute it (the agent reads context from the task). func (s *RoutineService) materialiseRoutineRun( @@ -556,12 +606,14 @@ func (s *RoutineService) materialiseRoutineRun( tmpl taskTemplate, title, description string, vars map[string]string, + source string, + idempotencyKey string, missedTicks int, ) error { if tmpl.Title != "" && s.workflowEnsurer != nil && s.taskCreator != nil { return s.materialiseHeavyRoutineRun(ctx, routine, run, title, description) } - return s.materialiseLightweightRoutineRun(ctx, routine, run, vars, missedTicks) + return s.materialiseLightweightRoutineRun(ctx, routine, run, vars, source, idempotencyKey, missedTicks) } // materialiseHeavyRoutineRun creates a real task in the routine system @@ -594,9 +646,20 @@ func (s *RoutineService) materialiseHeavyRoutineRun( } // materialiseLightweightRoutineRun enqueues a wakeup-request for the -// routine assignee with source="routine". The wakeup dispatcher claims -// it, applies the per-routine concurrency policy, and creates a fresh -// taskless runs row. LinkedTaskID stays empty for lightweight. +// routine assignee with source="routine" and then terminates this run: +// a lightweight routine has no task to wait on, so its own lifecycle +// ends the moment the request is enqueued and handed to the wakeup +// dispatcher. From there the dispatcher owns everything downstream, +// including its own concurrency gate over the same routine policy +// (wakeup/dispatcher.go resolvePolicy/resolveRoutinePolicy) — so this +// run must not sit in an "active" status waiting for that outcome, or +// it permanently blocks the routine's next fire under skip_if_active / +// coalesce_if_active (see the office-routine-runs triage). +// +// Terminal status: `done` once the request is dispatched; `failed` when +// enqueue or dispatch fails. A failed request is marked terminal in the +// wakeup queue because no background poller retries direct dispatch. +// LinkedTaskID stays empty for lightweight runs. // // missedTicks > 0 surfaces in the wakeup payload when the cron tick // collapsed N missed fires into one (catch-up cap policy @@ -606,16 +669,14 @@ func (s *RoutineService) materialiseLightweightRoutineRun( routine *Routine, run *RoutineRun, vars map[string]string, + source string, + idempotencyKey string, missedTicks int, ) error { - run.Status = models.RoutineRunStatusTaskCreated - if err := s.repo.UpdateRunStatus(ctx, run.ID, run.Status, ""); err != nil { - return fmt.Errorf("update run status: %w", err) - } if s.wakeup == nil || routine.AssigneeAgentProfileID == "" { - return nil + return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusDone) } - idemKey := buildRoutineIdempotencyKey(routine.ID, run.TriggerID, run.StartedAt) + idemKey := buildRoutineIdempotencyKey(routine.ID, run.TriggerID, source, run.ID, idempotencyKey, run.StartedAt) payloadStr, _ := marshalRoutinePayload(routine.ID, vars, missedTicks) req := &WakeupRequest{ ID: uuid.New().String(), @@ -627,33 +688,66 @@ func (s *RoutineService) materialiseLightweightRoutineRun( RequestedAt: time.Now().UTC(), } if err := s.wakeup.CreateWakeupRequest(ctx, req); err != nil { - // Idempotency conflicts are quietly absorbed by the office repo - // returning a sentinel error; the caller treats it as a no-op. + if errors.Is(err, ErrWakeupAlreadyRequested) { + // A request for this explicit fire identity already exists. + // Another delivery completed the same request. + return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusDone) + } s.logger.Warn("create routine wakeup request", zap.String("routine", routine.Name), zap.Error(err)) - return nil + return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusFailed) } if err := s.wakeup.Dispatch(ctx, req.ID); err != nil { s.logger.Warn("dispatch routine wakeup request", zap.String("routine", routine.Name), zap.String("wakeup_id", req.ID), zap.Error(err)) + if failErr := s.wakeup.FailWakeupRequest(ctx, req.ID, "dispatch failed"); failErr != nil { + s.logger.Warn("mark failed routine wakeup request", + zap.String("routine", routine.Name), + zap.String("wakeup_id", req.ID), + zap.Error(failErr)) + } + if finalizeErr := s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusFailed); finalizeErr != nil { + return finalizeErr + } + return fmt.Errorf("dispatch routine wakeup request: %w", err) + } + return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusDone) +} + +// finalizeLightweightRun writes the lightweight run's terminal status. +// LinkedTaskID is always empty for lightweight runs. +func (s *RoutineService) finalizeLightweightRun( + ctx context.Context, run *RoutineRun, status models.RoutineRunStatus, +) error { + run.Status = status + if err := s.repo.UpdateRunStatus(ctx, run.ID, status, ""); err != nil { + return fmt.Errorf("update run status: %w", err) } return nil } // buildRoutineIdempotencyKey composes the source-level dedup key for a -// routine fire. Format: routine:::; -// the trigger segment is dropped when there is no trigger (manual fire). -// Mirrors the heartbeat key shape (heartbeat::). -func buildRoutineIdempotencyKey(routineID, triggerID string, startedAt *time.Time) string { +// routine fire. Cron fires use the trigger and minute bucket. Manual and +// webhook fires use a unique run identity unless the caller supplies an +// explicit request key, so distinct event deliveries never collide. +func buildRoutineIdempotencyKey( + routineID, triggerID, source, runID, explicitKey string, startedAt *time.Time, +) string { + if explicitKey != "" { + return fmt.Sprintf("routine:%s:%s:%s", routineID, source, explicitKey) + } + if source != shared.RoutineSourceCron { + return fmt.Sprintf("routine:%s:%s:%s", routineID, source, runID) + } now := time.Now().UTC() if startedAt != nil { now = *startedAt } minute := now.Unix() / 60 if triggerID == "" { - return fmt.Sprintf("routine:%s:%d", routineID, minute) + return fmt.Sprintf("routine:%s:%s:%d", routineID, source, minute) } return fmt.Sprintf("routine:%s:%s:%d", routineID, triggerID, minute) } @@ -679,6 +773,40 @@ func marshalRoutinePayload(routineID string, vars map[string]string, missedTicks return string(b), nil } +// selfHealIfTaskTerminal is the pull-based counterpart to SyncRunStatus: +// SyncRunStatus is driven by the TaskMoved event, but nothing in +// production populates the step names that event needs +// (office-routine-runs R1), so the gate cannot rely on that event alone +// ever clearing it. Reading the linked task's real state here means a +// heavy run's gate self-heals at the next fire even when that event +// never arrives. Returns active unchanged when it is still genuinely +// active, still lightweight (no linked task), or its state can't be +// determined. A lookup or close-out failure fails closed, leaving the task +// active until a later fire can retry reconciliation. Returns nil once the +// linked task is confirmed terminal, archived, or missing and the run closes. +func (s *RoutineService) selfHealIfTaskTerminal( + ctx context.Context, routine *Routine, active *RoutineRun, +) *RoutineRun { + if active.LinkedTaskID == "" { + return active + } + terminalStatus, err := s.repo.GetTaskTerminalStatus(ctx, active.LinkedTaskID) + if err != nil { + s.logger.Warn("check linked task terminal state", + zap.String("routine", routine.Name), zap.String("run_id", active.ID), zap.Error(err)) + return active + } + if terminalStatus == "" { + return active + } + if _, err := s.closeOutRun(ctx, active, terminalStatus); err != nil { + s.logger.Warn("close out stale active run", + zap.String("routine", routine.Name), zap.String("run_id", active.ID), zap.Error(err)) + return active + } + return nil +} + func (s *RoutineService) applyConcurrencyPolicy( ctx context.Context, routine *Routine, @@ -688,25 +816,31 @@ func (s *RoutineService) applyConcurrencyPolicy( if routine.ConcurrencyPolicy == models.ConcurrencyPolicyAlwaysCreate { return "", nil } - active, err := s.repo.GetActiveRunForFingerprint(ctx, routine.ID, fingerprint) - if err != nil { - return "", fmt.Errorf("check active run: %w", err) + for { + active, err := s.repo.GetActiveRunForFingerprint(ctx, routine.ID, fingerprint) + if err != nil { + return "", fmt.Errorf("check active run: %w", err) + } + if active == nil { + return "", nil + } + if repaired := s.selfHealIfTaskTerminal(ctx, routine, active); repaired == nil { + continue + } else { + active = repaired + } + switch routine.ConcurrencyPolicy { + case models.ConcurrencyPolicySkipIfActive: + _ = s.repo.UpdateRunStatus(ctx, run.ID, models.RoutineRunStatusSkipped, "") + run.Status = models.RoutineRunStatusSkipped + return models.RoutineRunStatusSkipped, nil + case models.ConcurrencyPolicyCoalesceIfActive: + _ = s.repo.UpdateRunCoalesced(ctx, run.ID, active.ID) + run.Status = models.RoutineRunStatusCoalesced + run.CoalescedIntoRunID = active.ID + return models.RoutineRunStatusCoalesced, nil + } } - if active == nil { - return "", nil - } - switch routine.ConcurrencyPolicy { - case models.ConcurrencyPolicySkipIfActive: - _ = s.repo.UpdateRunStatus(ctx, run.ID, models.RoutineRunStatusSkipped, "") - run.Status = models.RoutineRunStatusSkipped - return models.RoutineRunStatusSkipped, nil - case models.ConcurrencyPolicyCoalesceIfActive: - _ = s.repo.UpdateRunCoalesced(ctx, run.ID, active.ID) - run.Status = models.RoutineRunStatusCoalesced - run.CoalescedIntoRunID = active.ID - return models.RoutineRunStatusCoalesced, nil - } - return "", nil } // FireManual dispatches a routine run from a manual trigger. @@ -720,12 +854,51 @@ func (s *RoutineService) FireManual( return s.DispatchRoutineRun(ctx, routine, nil, "manual", variableValues) } -// SyncRunStatus updates a linked run when its task reaches a terminal state. +// SyncRunStatus closes out the heavy routine run linked to taskID when +// that task reaches a terminal step. This is the writer for the +// terminal side of D2 in the office-routine-runs triage: heavy runs +// stay in RoutineRunStatusTaskCreated (an active gate) until their task +// finishes, and this is what clears the gate. terminalStatus is +// "cancelled" for a task moved to the Cancelled step, "failed" for +// a failed task, and "done" for a completed task. A taskID with no +// linked run is not an error — most tasks aren't routine-created. An +// empty taskID is rejected outright: linked_task_id defaults to "" for +// every lightweight run, so an unguarded lookup would match (and +// rewrite) an arbitrary lightweight run instead of finding nothing. func (s *RoutineService) SyncRunStatus(ctx context.Context, taskID, terminalStatus string) error { - s.logger.Info("sync run status", - zap.String("task_id", taskID), - zap.String("status", terminalStatus)) - return nil + if taskID == "" { + return nil + } + run, err := s.repo.GetRoutineRunByLinkedTaskID(ctx, taskID) + if err != nil { + return fmt.Errorf("get routine run by linked task: %w", err) + } + if run == nil { + return nil + } + _, err = s.closeOutRun(ctx, run, terminalStatus) + return err +} + +// closeOutRun writes a run's terminal status, but only while it is still +// task_created (see Repository.UpdateRunStatusIfTaskCreated) — a replayed +// or racing terminal signal for an already-closed run must not move +// completed_at or flip a terminal result. terminalStatus maps to the +// matching routine-run terminal state. Returns whether this call was the +// one that closed the run. +func (s *RoutineService) closeOutRun(ctx context.Context, run *RoutineRun, terminalStatus string) (bool, error) { + status := models.RoutineRunStatusDone + switch terminalStatus { + case "cancelled": + status = models.RoutineRunStatusCancelled + case "failed", "missing": + status = models.RoutineRunStatusFailed + } + closed, err := s.repo.UpdateRunStatusIfTaskCreated(ctx, run.ID, status, run.LinkedTaskID) + if err != nil { + return false, fmt.Errorf("update run status: %w", err) + } + return closed, nil } // -- helpers -- @@ -757,6 +930,11 @@ func parseDeclaredDefaults(variablesJSON string) map[string]string { return defaults } +// computeFingerprint is the dedup key for "an identical unit of work is +// already in flight". Lightweight runs never sit in an active status +// (see materialiseLightweightRoutineRun), so the fingerprint only gates +// heavy runs in practice: it means "a task with this exact rendered +// title, description, and assignee is still open". func computeFingerprint(title, description, assignee string) string { h := sha256.Sum256([]byte(title + "|" + description + "|" + assignee)) return fmt.Sprintf("%x", h[:16]) diff --git a/apps/backend/internal/office/routines/service_reconciliation_test.go b/apps/backend/internal/office/routines/service_reconciliation_test.go new file mode 100644 index 0000000000..06128db6fe --- /dev/null +++ b/apps/backend/internal/office/routines/service_reconciliation_test.go @@ -0,0 +1,281 @@ +package routines_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/office/models" +) + +type dispatchErrorWakeupEnqueuer struct{ fakeWakeupEnqueuer } + +func (f *dispatchErrorWakeupEnqueuer) Dispatch(context.Context, string) error { + return errors.New("dispatcher unavailable") +} + +type sequenceTaskCreator struct { + ids []string + next int +} + +func (f *sequenceTaskCreator) CreateOfficeTaskInWorkflow( + context.Context, string, string, string, string, string, string, +) (string, error) { + if f.next >= len(f.ids) { + return "task-overflow", nil + } + id := f.ids[f.next] + f.next++ + return id, nil +} + +func TestDispatch_LightweightRoutine_SourceKeysAreUnique(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Manual keys", + TaskTemplate: "", + AssigneeAgentProfileID: "agent-1", + Status: "active", + ConcurrencyPolicy: "always_create", + } + if err := svc.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + enq := &fakeWakeupEnqueuer{} + svc.SetWakeupEnqueuer(enq) + + if _, err := svc.FireManual(ctx, routine.ID, map[string]string{"name": "same"}); err != nil { + t.Fatalf("first fire: %v", err) + } + if _, err := svc.FireManual(ctx, routine.ID, map[string]string{"name": "same"}); err != nil { + t.Fatalf("second fire: %v", err) + } + if len(enq.created) != 2 { + t.Fatalf("created wakeups = %d, want 2", len(enq.created)) + } + if enq.created[0].IdempotencyKey == enq.created[1].IdempotencyKey { + t.Fatalf("distinct manual fires reused idempotency key %q", enq.created[0].IdempotencyKey) + } +} + +func TestDispatch_LightweightRoutine_WebhookRequestKeyIsStable(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Webhook keys", + TaskTemplate: "", + AssigneeAgentProfileID: "agent-1", + Status: "active", + ConcurrencyPolicy: "always_create", + } + if err := svc.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + trigger := &models.RoutineTrigger{ID: "trigger-1", RoutineID: routine.ID, Kind: "webhook"} + enq := &fakeWakeupEnqueuer{} + svc.SetWakeupEnqueuer(enq) + + if _, err := svc.DispatchRoutineRunWithIdempotencyKey( + ctx, routine, trigger, "webhook", map[string]string{"branch": "main"}, "delivery-1"); err != nil { + t.Fatalf("first webhook: %v", err) + } + if _, err := svc.DispatchRoutineRunWithIdempotencyKey( + ctx, routine, trigger, "webhook", map[string]string{"branch": "main"}, "delivery-1"); err != nil { + t.Fatalf("retry webhook: %v", err) + } + if len(enq.created) != 2 { + t.Fatalf("created wakeups = %d, want 2", len(enq.created)) + } + if enq.created[0].IdempotencyKey != enq.created[1].IdempotencyKey { + t.Fatalf("retry changed idempotency key from %q to %q", enq.created[0].IdempotencyKey, enq.created[1].IdempotencyKey) + } +} + +func TestDispatch_LightweightRoutine_DispatchFailureFailsRun(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Dispatch failure", + TaskTemplate: "", + AssigneeAgentProfileID: "agent-1", + Status: "active", + ConcurrencyPolicy: "always_create", + } + if err := svc.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + enq := &dispatchErrorWakeupEnqueuer{} + svc.SetWakeupEnqueuer(enq) + + run, err := svc.FireManual(ctx, routine.ID, nil) + if err == nil { + t.Fatal("expected the dispatch error to reach the caller") + } + if run.Status != models.RoutineRunStatusFailed { + t.Errorf("status = %q, want failed", run.Status) + } + if len(enq.failed) != 1 { + t.Errorf("failed wakeups = %d, want 1", len(enq.failed)) + } +} + +func TestDispatch_HeavyRoutine_LongRunningTaskStillGates(t *testing.T) { + svc, db := newTestRoutineServiceWithDB(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + + routine := createTestRoutine(t, svc, "Stale Gate", "skip_if_active") + + run1, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("first run: %v", err) + } + if run1.Status != "task_created" { + t.Fatalf("first run status = %q, want task_created", run1.Status) + } + + if _, err := db.ExecContext(ctx, + `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if _, err := db.ExecContext(ctx, + `INSERT INTO tasks (id, state) VALUES (?, ?)`, run1.LinkedTaskID, "IN_PROGRESS"); err != nil { + t.Fatalf("insert linked task: %v", err) + } + + longAgo := time.Now().UTC().AddDate(0, -1, 0) + if _, err := db.ExecContext(ctx, + `UPDATE office_routine_runs SET created_at = ? WHERE id = ?`, longAgo, run1.ID, + ); err != nil { + t.Fatalf("backdate run: %v", err) + } + + run2, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second run: %v", err) + } + if run2.Status != models.RoutineRunStatusSkipped { + t.Errorf("second run status = %q, want skipped (live task must keep gating)", run2.Status) + } +} + +func TestDispatch_HeavyRoutine_RepairsNewestButStillBlocksOnOlder(t *testing.T) { + svc, db := newTestRoutineServiceWithDB(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + creator := &sequenceTaskCreator{ids: []string{"task-old", "task-new", "task-third"}} + svc.SetTaskCreator(creator) + + if _, err := db.ExecContext(ctx, + `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if _, err := db.ExecContext(ctx, + `INSERT INTO tasks (id, state) VALUES (?, ?), (?, ?)`, "task-old", "IN_PROGRESS", "task-new", "COMPLETED"); err != nil { + t.Fatalf("insert linked tasks: %v", err) + } + + routine := createTestRoutine(t, svc, "Multiple active rows", "always_create") + first, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("first fire: %v", err) + } + second, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second fire: %v", err) + } + if first.LinkedTaskID != "task-old" || second.LinkedTaskID != "task-new" { + t.Fatalf("linked tasks = %q, %q, want task-old, task-new", first.LinkedTaskID, second.LinkedTaskID) + } + if _, err := db.ExecContext(ctx, + `UPDATE office_routine_runs SET created_at = CASE id WHEN ? THEN ? WHEN ? THEN ? END WHERE id IN (?, ?)`, + first.ID, time.Now().UTC().Add(-2*time.Minute), second.ID, time.Now().UTC().Add(-time.Minute), first.ID, second.ID); err != nil { + t.Fatalf("order routine runs: %v", err) + } + routine.ConcurrencyPolicy = models.ConcurrencyPolicySkipIfActive + if err := svc.UpdateRoutine(ctx, routine); err != nil { + t.Fatalf("update policy: %v", err) + } + + third, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("third fire: %v", err) + } + if third.Status != models.RoutineRunStatusSkipped { + t.Fatalf("third run status = %q, want skipped while older task remains active", third.Status) + } + runs, err := svc.ListRoutineRuns(ctx, routine.ID, 10, 0) + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.ID == second.ID && run.Status != models.RoutineRunStatusDone { + t.Errorf("newest repaired run status = %q, want done", run.Status) + } + } +} + +func TestDispatch_HeavyRoutine_TerminalTaskStatesReleaseGate(t *testing.T) { + cases := []struct { + name string + state string + archived bool + wantOldStatus models.RoutineRunStatus + }{ + {name: "failed", state: "FAILED", wantOldStatus: models.RoutineRunStatusFailed}, + {name: "archived", state: "IN_PROGRESS", archived: true, wantOldStatus: models.RoutineRunStatusCancelled}, + {name: "missing", wantOldStatus: models.RoutineRunStatusFailed}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + svc, db := newTestRoutineServiceWithDB(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + routine := createTestRoutine(t, svc, "Terminal "+tc.name, "skip_if_active") + first, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("first fire: %v", err) + } + + if _, err := db.ExecContext(ctx, + `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if tc.name != "missing" { + var archivedAt any + if tc.archived { + archivedAt = time.Now().UTC() + } + if _, err := db.ExecContext(ctx, + `INSERT INTO tasks (id, state, archived_at) VALUES (?, ?, ?)`, first.LinkedTaskID, tc.state, archivedAt); err != nil { + t.Fatalf("insert linked task: %v", err) + } + } + + second, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second fire: %v", err) + } + if second.Status != models.RoutineRunStatusTaskCreated { + t.Fatalf("second run status = %q, want task_created", second.Status) + } + runs, err := svc.ListRoutineRuns(ctx, routine.ID, 10, 0) + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.ID == first.ID && run.Status != tc.wantOldStatus { + t.Errorf("old run status = %q, want %q", run.Status, tc.wantOldStatus) + } + } + }) + } +} diff --git a/apps/backend/internal/office/routines/service_test.go b/apps/backend/internal/office/routines/service_test.go index b42ba840ae..ccd2bac3b7 100644 --- a/apps/backend/internal/office/routines/service_test.go +++ b/apps/backend/internal/office/routines/service_test.go @@ -25,6 +25,14 @@ func (n *noopActivity) LogActivityWithRun(_ context.Context, _, _, _, _, _, _, _ // newTestRoutineService creates a RoutineService backed by in-memory SQLite. func newTestRoutineService(t *testing.T) *routines.RoutineService { + t.Helper() + svc, _ := newTestRoutineServiceWithDB(t) + return svc +} + +// newTestRoutineServiceWithDB is newTestRoutineService plus the backing +// *sqlx.DB, for tests that need to reach into shared task or routine tables. +func newTestRoutineServiceWithDB(t *testing.T) (*routines.RoutineService, *sqlx.DB) { t.Helper() db, err := sqlx.Open("sqlite3", ":memory:") if err != nil { @@ -38,7 +46,7 @@ func newTestRoutineService(t *testing.T) *routines.RoutineService { } log := logger.Default() - return routines.NewRoutineService(repo, log, &noopActivity{}) + return routines.NewRoutineService(repo, log, &noopActivity{}), db } // createTestRoutine creates a test routine in the DB. @@ -69,8 +77,12 @@ func TestFireManual_Basic(t *testing.T) { if err != nil { t.Fatalf("fire manual: %v", err) } - if run.Status != "task_created" { - t.Errorf("status = %q, want task_created", run.Status) + // No workflow ensurer / task creator wired on this service, so this + // routine (despite having a task_template) takes the lightweight + // path and terminates immediately at "done" — see D1 in the + // office-routine-runs triage. + if run.Status != "done" { + t.Errorf("status = %q, want done", run.Status) } if run.Source != "manual" { t.Errorf("source = %q, want manual", run.Source) @@ -80,13 +92,21 @@ func TestFireManual_Basic(t *testing.T) { } } +// TestDispatch_SkipIfActive exercises skip_if_active on the heavy path, +// where "active" is real: run1's task genuinely has not reached a +// terminal step yet (office-routine-runs D2). Wired lightweight (the +// pre-fix shape of this test), the gate can never engage — a +// lightweight run never sits in an active status (D1) — so the test +// would pass by asserting behaviour nothing exercises. func TestDispatch_SkipIfActive(t *testing.T) { svc := newTestRoutineService(t) ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) routine := createTestRoutine(t, svc, "Skip Test", "skip_if_active") - // First run should succeed. + // First run should succeed and create a task. run1, err := svc.FireManual(ctx, routine.ID, nil) if err != nil { t.Fatalf("first run: %v", err) @@ -95,7 +115,8 @@ func TestDispatch_SkipIfActive(t *testing.T) { t.Errorf("first run status = %q, want task_created", run1.Status) } - // Second run with same fingerprint should be skipped. + // Second run with the same fingerprint should be skipped: run1's + // task is still open. run2, err := svc.FireManual(ctx, routine.ID, nil) if err != nil { t.Fatalf("second run: %v", err) @@ -105,9 +126,14 @@ func TestDispatch_SkipIfActive(t *testing.T) { } } +// TestDispatch_CoalesceIfActive is the coalesce_if_active counterpart of +// TestDispatch_SkipIfActive; see its comment for why this needs the +// heavy path. func TestDispatch_CoalesceIfActive(t *testing.T) { svc := newTestRoutineService(t) ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) routine := createTestRoutine(t, svc, "Coalesce Test", "coalesce_if_active") @@ -131,6 +157,38 @@ func TestDispatch_CoalesceIfActive(t *testing.T) { } } +// TestDispatch_HeavyRoutine_SyncRunStatusUnblocksNextFire proves the +// gate closes and reopens rather than bricking after the first fire +// (the office-routine-runs symptom: 323 consecutive coalesces after one +// fire). Each cycle fires, expects a fresh task_created run (not +// skipped/coalesced against a prior cycle), then simulates the linked +// task reaching Done via SyncRunStatus — the seam finalizeDone calls in +// production. +func TestDispatch_HeavyRoutine_SyncRunStatusUnblocksNextFire(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + + routine := createTestRoutine(t, svc, "Sync Cycle", "coalesce_if_active") + + for i := 1; i <= 3; i++ { + run, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("fire %d: %v", i, err) + } + if run.Status != "task_created" { + t.Fatalf("fire %d status = %q, want task_created (previous cycle's gate should already be closed)", i, run.Status) + } + if run.LinkedTaskID == "" { + t.Fatalf("fire %d: expected a linked task", i) + } + if err := svc.SyncRunStatus(ctx, run.LinkedTaskID, "done"); err != nil { + t.Fatalf("sync run status after fire %d: %v", i, err) + } + } +} + func TestDispatch_AlwaysCreate(t *testing.T) { svc := newTestRoutineService(t) ctx := context.Background() @@ -147,14 +205,24 @@ func TestDispatch_AlwaysCreate(t *testing.T) { t.Fatalf("second run: %v", err) } - if run1.Status != "task_created" || run2.Status != "task_created" { - t.Errorf("both runs should be task_created, got %q and %q", run1.Status, run2.Status) + // No workflow ensurer / task creator wired: lightweight path, done + // on both fires (always_create never even consults the active-run + // gate, so this exercises only the "successful lightweight fire + // terminates" half of D1). + if run1.Status != "done" || run2.Status != "done" { + t.Errorf("both runs should be done, got %q and %q", run1.Status, run2.Status) } } +// TestDispatch_DifferentVarsNotSkipped is on the heavy path so the +// active-run gate is real: without it, a lightweight run is never +// "active" (D1) and the assertion would pass regardless of whether +// fingerprints actually differ. func TestDispatch_DifferentVarsNotSkipped(t *testing.T) { svc := newTestRoutineService(t) ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) routine := createTestRoutine(t, svc, "Diff Vars", "skip_if_active") @@ -166,7 +234,8 @@ func TestDispatch_DifferentVarsNotSkipped(t *testing.T) { t.Errorf("first run status = %q, want task_created", run1.Status) } - // Different variable -> different fingerprint -> not skipped. + // Different variable -> different fingerprint -> not skipped, even + // though run1's task is still open. run2, err := svc.FireManual(ctx, routine.ID, map[string]string{"name": "B"}) if err != nil { t.Fatalf("second run: %v", err) @@ -184,6 +253,7 @@ func TestDispatch_DifferentVarsNotSkipped(t *testing.T) { type fakeWakeupEnqueuer struct { created []*routines.WakeupRequest dispatched []string + failed []string } func (f *fakeWakeupEnqueuer) CreateWakeupRequest(_ context.Context, req *routines.WakeupRequest) error { @@ -196,6 +266,11 @@ func (f *fakeWakeupEnqueuer) Dispatch(_ context.Context, requestID string) error return nil } +func (f *fakeWakeupEnqueuer) FailWakeupRequest(_ context.Context, requestID, _ string) error { + f.failed = append(f.failed, requestID) + return nil +} + // fakeWorkflowEnsurer / fakeTaskCreator capture the heavy-path inputs // and return synthetic ids so the heavy-flow test can assert on them. type fakeWorkflowEnsurer struct { @@ -253,8 +328,11 @@ func TestDispatch_LightweightRoutine_EnqueuesWakeup(t *testing.T) { if run.LinkedTaskID != "" { t.Errorf("lightweight path must not link a task, got %q", run.LinkedTaskID) } - if run.Status != "task_created" { - t.Errorf("status = %q, want task_created (run row reused even for taskless)", run.Status) + // The run's own lifecycle ends once the wakeup-request is enqueued + // and dispatched (D1) — it does not wait in task_created for the + // wakeup dispatcher's own concurrency gate to resolve. + if run.Status != "done" { + t.Errorf("status = %q, want done", run.Status) } if len(enq.created) != 1 { t.Fatalf("expected 1 wakeup-request created, got %d", len(enq.created)) @@ -284,6 +362,48 @@ func TestDispatch_LightweightRoutine_EnqueuesWakeup(t *testing.T) { } } +// idempotencyConflictWakeupEnqueuer simulates a duplicate request identity +// already persisted by another caller. +type idempotencyConflictWakeupEnqueuer struct{} + +func (idempotencyConflictWakeupEnqueuer) CreateWakeupRequest(context.Context, *routines.WakeupRequest) error { + return routines.ErrWakeupAlreadyRequested +} + +func (idempotencyConflictWakeupEnqueuer) Dispatch(context.Context, string) error { return nil } + +func (idempotencyConflictWakeupEnqueuer) FailWakeupRequest(context.Context, string, string) error { + return nil +} + +// TestDispatch_LightweightRoutine_IdempotencyConflictIsNotFailed verifies +// that a duplicate request identity is recorded as done, not failed. +func TestDispatch_LightweightRoutine_IdempotencyConflictIsNotFailed(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Dedup Race", + TaskTemplate: "", + AssigneeAgentProfileID: "agent-1", + Status: "active", + ConcurrencyPolicy: "always_create", + } + if err := svc.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + svc.SetWakeupEnqueuer(idempotencyConflictWakeupEnqueuer{}) + + run, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("fire manual: %v", err) + } + if run.Status != "done" { + t.Errorf("status = %q, want done (idempotency conflict is success by another fire, not a failure)", run.Status) + } +} + // TestDispatch_HeavyRoutine_CreatesTaskInRoutineWorkflow verifies a // routine with a task_template materialises a real task pinned to the // routine workflow id. @@ -393,6 +513,12 @@ func TestDispatch_HeavyRoutine_FallsBackWithoutDeps(t *testing.T) { if run.LinkedTaskID != "" { t.Errorf("expected empty linked_task_id when deps missing, got %q", run.LinkedTaskID) } + // Falling back to lightweight must still resolve to a terminal + // status, not strand the run in task_created with no task to ever + // close it out. + if run.Status != "done" { + t.Errorf("status = %q, want done", run.Status) + } } func TestListRoutineRuns(t *testing.T) { @@ -412,3 +538,199 @@ func TestListRoutineRuns(t *testing.T) { t.Errorf("expected 3 runs, got %d", len(runs)) } } + +// TestDispatch_LightweightRoutine_SubsequentFiresMaterialise is the +// card's explicit regression requirement: a lightweight routine must +// fire every time, under both concurrency policies, not just once. Pre- +// fix, materialiseLightweightRoutineRun left every successful fire in +// task_created, which both policies treat as "still active" forever. +func TestDispatch_LightweightRoutine_SubsequentFiresMaterialise(t *testing.T) { + for _, policy := range []string{"skip_if_active", "coalesce_if_active"} { + t.Run(policy, func(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + + routine := &models.Routine{ + WorkspaceID: "ws-1", + Name: "Coordinator heartbeat", + TaskTemplate: "", // lightweight + AssigneeAgentProfileID: "agent-1", + Status: "active", + ConcurrencyPolicy: models.RoutineConcurrencyPolicy(policy), + } + if err := svc.CreateRoutine(ctx, routine); err != nil { + t.Fatalf("create routine: %v", err) + } + svc.SetWakeupEnqueuer(&fakeWakeupEnqueuer{}) + + for i := 1; i <= 3; i++ { + run, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("fire %d: %v", i, err) + } + if run.Status != "done" { + t.Fatalf("fire %d status = %q, want done (fire should materialise, not be gated by fire 1)", i, run.Status) + } + } + }) + } +} + +// TestDispatch_HeavyRoutine_GateSelfHealsWhenTaskTerminatesWithoutEvent is +// the R1 regression test: no production publisher of events.TaskMoved +// ever sets to_step_name (office-routine-runs review round 1), so +// SyncRunStatus is never reached by that event in production and cannot +// be the only way a heavy run's gate clears. This drives the actual +// production seam — FireManual -> dispatchRoutineRun -> +// applyConcurrencyPolicy — with no call to SyncRunStatus at all: the +// linked task's row is updated directly (as the real `tasks` table would +// be by an ordinary task-service move), and the gate must still notice +// and let the second fire through. +func TestDispatch_HeavyRoutine_GateSelfHealsWhenTaskTerminatesWithoutEvent(t *testing.T) { + for _, policy := range []string{"skip_if_active", "coalesce_if_active"} { + t.Run(policy, func(t *testing.T) { + svc, db := newTestRoutineServiceWithDB(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + + routine := createTestRoutine(t, svc, "Self Heal "+policy, policy) + + run1, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("first run: %v", err) + } + if run1.Status != "task_created" { + t.Fatalf("first run status = %q, want task_created", run1.Status) + } + if run1.LinkedTaskID == "" { + t.Fatalf("expected a linked task") + } + + // Mirror the shared `tasks` table's shape and mark the linked + // task COMPLETED directly — no TaskMoved event, no + // SyncRunStatus call. This is what the row looks like after a + // real task-service move in production. + if _, err := db.ExecContext(ctx, + `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if _, err := db.ExecContext(ctx, + `INSERT INTO tasks (id, state) VALUES (?, ?)`, run1.LinkedTaskID, "COMPLETED"); err != nil { + t.Fatalf("seed linked task: %v", err) + } + + run2, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second run: %v", err) + } + if run2.Status != "task_created" { + t.Fatalf("second run status = %q, want task_created (gate must self-heal once the linked task is terminal)", run2.Status) + } + }) + } +} + +// TestDispatch_HeavyRoutine_GateSelfHealsOnFailedTask covers the third +// terminal outcome GetTaskTerminalStatus recognizes: a linked task that +// reaches FAILED (not just COMPLETED/CANCELLED) must also release the +// gate, and the stale run must be closed out as "failed" rather than +// "done" so routine history reflects what actually happened. +func TestDispatch_HeavyRoutine_GateSelfHealsOnFailedTask(t *testing.T) { + for _, policy := range []string{"skip_if_active", "coalesce_if_active"} { + t.Run(policy, func(t *testing.T) { + svc, db := newTestRoutineServiceWithDB(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + + routine := createTestRoutine(t, svc, "Self Heal Failed "+policy, policy) + + run1, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("first run: %v", err) + } + if run1.Status != "task_created" { + t.Fatalf("first run status = %q, want task_created", run1.Status) + } + if run1.LinkedTaskID == "" { + t.Fatalf("expected a linked task") + } + + if _, err := db.ExecContext(ctx, + `CREATE TABLE tasks (id TEXT PRIMARY KEY, state TEXT, archived_at TIMESTAMP)`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + if _, err := db.ExecContext(ctx, + `INSERT INTO tasks (id, state) VALUES (?, ?)`, run1.LinkedTaskID, "FAILED"); err != nil { + t.Fatalf("seed linked task: %v", err) + } + + run2, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second run: %v", err) + } + if run2.Status != "task_created" { + t.Fatalf("second run status = %q, want task_created (gate must self-heal once the linked task fails)", run2.Status) + } + + var closedStatus string + if err := db.GetContext(ctx, &closedStatus, + `SELECT status FROM office_routine_runs WHERE id = ?`, run1.ID); err != nil { + t.Fatalf("read closed-out run status: %v", err) + } + if closedStatus != "failed" { + t.Errorf("run1 status after self-heal = %q, want failed", closedStatus) + } + }) + } +} + +// TestDispatch_LastRunAt verifies routines.last_run_at is populated by a +// materialised fire and left untouched by a gated one — the card's +// fourth symptom (empty last_run_at despite hundreds of runs). +func TestDispatch_LastRunAt(t *testing.T) { + svc := newTestRoutineService(t) + ctx := context.Background() + svc.SetWorkflowEnsurer(&fakeWorkflowEnsurer{}) + svc.SetTaskCreator(&fakeTaskCreator{}) + + routine := createTestRoutine(t, svc, "Last Run", "skip_if_active") + + before, err := svc.GetRoutine(ctx, routine.ID) + if err != nil { + t.Fatalf("get routine: %v", err) + } + if before.LastRunAt != nil { + t.Fatalf("last_run_at = %v before any fire, want nil", before.LastRunAt) + } + + if _, err := svc.FireManual(ctx, routine.ID, nil); err != nil { + t.Fatalf("first fire: %v", err) + } + afterFirst, err := svc.GetRoutine(ctx, routine.ID) + if err != nil { + t.Fatalf("get routine: %v", err) + } + if afterFirst.LastRunAt == nil { + t.Fatal("expected last_run_at to be set after a materialised fire") + } + firstStamp := *afterFirst.LastRunAt + + // Second fire is skipped (run1's task is still open) — last_run_at + // must not advance for a fire that didn't materialise. + run2, err := svc.FireManual(ctx, routine.ID, nil) + if err != nil { + t.Fatalf("second fire: %v", err) + } + if run2.Status != "skipped" { + t.Fatalf("second run status = %q, want skipped", run2.Status) + } + afterSecond, err := svc.GetRoutine(ctx, routine.ID) + if err != nil { + t.Fatalf("get routine: %v", err) + } + if !afterSecond.LastRunAt.Equal(firstStamp) { + t.Errorf("last_run_at changed from %v to %v after a skipped fire", firstStamp, afterSecond.LastRunAt) + } +} diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 0cc2656756..9f7d5308f3 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "go.uber.org/zap" @@ -1014,10 +1015,22 @@ func (s *Service) handleTaskMoved(ctx context.Context, event *bus.Event) error { return nil } -// finalizeDone resolves blockers and notifies parents when a task enters -// a terminal step. Both side-effects route through the engine via -// dispatchEngineTrigger (on_blocker_resolved / on_children_completed). +// finalizeDone resolves blockers, notifies parents, and closes out a +// linked routine run when a task enters a terminal step. The blocker +// and parent side-effects route through the engine via +// dispatchEngineTrigger (on_blocker_resolved / on_children_completed); +// the routine sync is a direct call since it's a simple status write, +// not an engine trigger. func (s *Service) finalizeDone(ctx context.Context, data *TaskMovedData) error { + if s.routineRunSyncer != nil { + terminal := "done" + if strings.EqualFold(data.ToStepName, "cancelled") { + terminal = "cancelled" + } + if err := s.routineRunSyncer.SyncRunStatus(ctx, data.TaskID, terminal); err != nil { + s.logger.Warn("sync routine run status", zap.Error(err)) + } + } if err := s.queueBlockersResolvedRuns(ctx, data.TaskID); err != nil { s.logger.Error("blocker resolution runs failed", zap.Error(err)) } diff --git a/apps/backend/internal/office/service/routine_run_sync_test.go b/apps/backend/internal/office/service/routine_run_sync_test.go new file mode 100644 index 0000000000..d34bed6378 --- /dev/null +++ b/apps/backend/internal/office/service/routine_run_sync_test.go @@ -0,0 +1,122 @@ +package service_test + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/events/bus" +) + +// fakeRoutineRunSyncer captures SyncRunStatus calls so tests can assert +// finalizeDone calls it with the right (taskID, terminalStatus) pair +// without pulling in the routines package. +type fakeRoutineRunSyncer struct { + calls []struct{ taskID, terminalStatus string } +} + +func (f *fakeRoutineRunSyncer) SyncRunStatus(_ context.Context, taskID, terminalStatus string) error { + f.calls = append(f.calls, struct{ taskID, terminalStatus string }{taskID, terminalStatus}) + return nil +} + +// TestFinalizeDone_SyncsRoutineRunOnDone verifies that a task moving +// into the Done step closes out its linked routine run via the wired +// RoutineRunSyncer (office-routine-runs D2) — the seam that lets a +// heavy routine's next fire through instead of gating on task_created +// forever. +func TestFinalizeDone_SyncsRoutineRunOnDone(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + syncer := &fakeRoutineRunSyncer{} + svc.SetRoutineRunSyncer(syncer) + + createTestAgent(t, svc, "ws-1", "worker-done") + taskID := createOfficeTask(t, svc, "ws-1", "worker-done") + + moveEvent := bus.NewEvent("task.moved", "test", map[string]string{ + "task_id": taskID, + "workspace_id": "ws-1", + "from_step_id": "step-1", + "to_step_id": "step-done", + "to_step_name": "Done", + "from_step_name": "In Progress", + "assignee_agent_profile_id": "worker-done", + "parent_id": "", + }) + if err := eb.Publish(ctx, "task.moved", moveEvent); err != nil { + t.Fatalf("publish task.moved: %v", err) + } + + if len(syncer.calls) != 1 { + t.Fatalf("expected 1 SyncRunStatus call, got %d", len(syncer.calls)) + } + if syncer.calls[0].taskID != taskID { + t.Errorf("taskID = %q, want %q", syncer.calls[0].taskID, taskID) + } + if syncer.calls[0].terminalStatus != "done" { + t.Errorf("terminalStatus = %q, want done", syncer.calls[0].terminalStatus) + } +} + +// TestFinalizeDone_SyncsRoutineRunOnCancelled mirrors the Done case for +// a task moved to Cancelled — categorizeStep groups both under +// stepCategoryDone, so finalizeDone must distinguish them itself when +// telling the routine run which terminal status to write. +func TestFinalizeDone_SyncsRoutineRunOnCancelled(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + syncer := &fakeRoutineRunSyncer{} + svc.SetRoutineRunSyncer(syncer) + + createTestAgent(t, svc, "ws-1", "worker-cancel") + taskID := createOfficeTask(t, svc, "ws-1", "worker-cancel") + + moveEvent := bus.NewEvent("task.moved", "test", map[string]string{ + "task_id": taskID, + "workspace_id": "ws-1", + "from_step_id": "step-1", + "to_step_id": "step-cancelled", + "to_step_name": "Cancelled", + "from_step_name": "In Progress", + "assignee_agent_profile_id": "worker-cancel", + "parent_id": "", + }) + if err := eb.Publish(ctx, "task.moved", moveEvent); err != nil { + t.Fatalf("publish task.moved: %v", err) + } + + if len(syncer.calls) != 1 { + t.Fatalf("expected 1 SyncRunStatus call, got %d", len(syncer.calls)) + } + if syncer.calls[0].terminalStatus != "cancelled" { + t.Errorf("terminalStatus = %q, want cancelled", syncer.calls[0].terminalStatus) + } +} + +// TestFinalizeDone_NoSyncerIsANoop guards the optional-wiring contract: +// a service without a RoutineRunSyncer wired (most test services, and +// any real deploy before this seam existed) must not panic or error on +// a task reaching Done. +func TestFinalizeDone_NoSyncerIsANoop(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "worker-nosync") + taskID := createOfficeTask(t, svc, "ws-1", "worker-nosync") + + moveEvent := bus.NewEvent("task.moved", "test", map[string]string{ + "task_id": taskID, + "workspace_id": "ws-1", + "from_step_id": "step-1", + "to_step_id": "step-done", + "to_step_name": "Done", + "from_step_name": "In Progress", + "assignee_agent_profile_id": "worker-nosync", + "parent_id": "", + }) + if err := eb.Publish(ctx, "task.moved", moveEvent); err != nil { + t.Fatalf("publish task.moved: %v", err) + } +} diff --git a/apps/backend/internal/office/service/service.go b/apps/backend/internal/office/service/service.go index 761cc7bc8a..ad62afffbe 100644 --- a/apps/backend/internal/office/service/service.go +++ b/apps/backend/internal/office/service/service.go @@ -281,8 +281,28 @@ type Service struct { // the budget pathways. Both CheckBudget and CheckPreExecutionBudget // delegate to it. budgetChecker BudgetEvaluator + + // routineRunSyncer closes out a heavy routine run when its linked + // task reaches a terminal step. Wired to the routines.RoutineService + // at startup; nil in tests that don't exercise routines. + routineRunSyncer RoutineRunSyncer } +// RoutineRunSyncer is the surface the office service needs from the +// routines feature to close out a routine run once its linked task +// finishes. Implemented by *routines.RoutineService.SyncRunStatus — +// declared here so tests can supply fakes without pulling the routines +// package. terminalStatus is "done" or "cancelled". +type RoutineRunSyncer interface { + SyncRunStatus(ctx context.Context, taskID, terminalStatus string) error +} + +// SetRoutineRunSyncer wires the routines.RoutineService (or a test fake) +// used to close out a heavy routine run when its linked task reaches a +// terminal step. Without this wired, a heavy routine's run stays in +// task_created until routines.activeRunMaxAge lets a later fire through. +func (s *Service) SetRoutineRunSyncer(r RoutineRunSyncer) { s.routineRunSyncer = r } + // BudgetEvaluator is the surface the office service needs from the // costs feature for budget evaluation. Implemented by // *costs.CostService.EvaluateBudget — declared here so tests can supply diff --git a/apps/backend/internal/office/wakeup/routine_e2e_test.go b/apps/backend/internal/office/wakeup/routine_e2e_test.go index 46ff66dde7..c9d575c9e1 100644 --- a/apps/backend/internal/office/wakeup/routine_e2e_test.go +++ b/apps/backend/internal/office/wakeup/routine_e2e_test.go @@ -242,3 +242,7 @@ func (a *routineE2EWakeupAdapter) CreateWakeupRequest( func (a *routineE2EWakeupAdapter) Dispatch(ctx context.Context, requestID string) error { return a.dispatcher.Dispatch(ctx, requestID) } + +func (a *routineE2EWakeupAdapter) FailWakeupRequest(ctx context.Context, requestID, reason string) error { + return a.repo.MarkWakeupRequestFailed(ctx, requestID, reason) +} diff --git a/docs/specs/office/system-design/scheduler-01.md b/docs/specs/office/system-design/scheduler-01.md index 1c9cb7d40c..117cdadd82 100644 --- a/docs/specs/office/system-design/scheduler-01.md +++ b/docs/specs/office/system-design/scheduler-01.md @@ -90,9 +90,12 @@ Coalescing happens entirely at the wakeup-request layer; the runs table is the e 1. **Source-level dedup via `idempotency_key`** (UNIQUE column). Format: - `routine:::` for cron routines. + - `routine::webhook:` for webhook retries that provide an `Idempotency-Key` header. + - `routine::webhook:` for webhook deliveries without an explicit request key. + - `routine::manual:` for manual fires. - `comment:` for comments. - `task_assigned::` for task creation/assignment events. - Duplicate inserts in the same window are rejected silently. Handles webhook re-delivery, event-bus replay, and restart recovery. + Duplicate inserts with the same source identity are rejected silently. Webhook callers use `Idempotency-Key` for redelivery. Manual and unkeyed webhook fires remain distinct. 2. **Claim-time merge.** The source first persists the wakeup-request with `status="queued"`. The dispatcher then looks for an in-flight run for the same agent (`queued` or `claimed`). If one exists **and it is still `queued`**: mark the persisted request with `status="coalesced"` and `run_id=`, merge its payload into the existing run's `context_snapshot`, increment the run's `coalesced_count`, and — if the existing run is periodic-classified (see "Idle wakeup skip" below) while the new request is event-classified — promote the run's `reason` to the new request's classification (monotonic: periodic -> event only, never the reverse). The promotion, request transition, count update, and payload merge use one transaction, so the scheduler cannot claim a partially updated run. The agent sees the merged context and the promoted reason when it actually runs. If the in-flight run is already **`claimed`** *and the promotion above would otherwise be required* (the run is periodic-classified and the new request is event-classified): it is not promoted or merged into — the scheduler has already read its `reason` into memory for the idle-skip decision and never re-reads the row, so mutating it afterward would race a decision already in flight. The new request instead creates its own fresh `runs` row and is marked `claimed` against it, exactly as if no in-flight run existed. A `claimed` run that needs no promotion (the existing run is already event-classified, or the new request is itself periodic-classified) is coalesced into exactly as before. If no in-flight run exists at all: create the corresponding `runs` row and mark the persisted request as `claimed` against it. @@ -123,7 +126,7 @@ Variables use `{{name}}` syntax in title/description with types `text`, `number` #### Routine runs -Each trigger firing creates a routine run record (`office_routine_runs`) with `routine_id`, `trigger_id`, `source` (`cron` | `webhook` | `manual`), `status` (`received` -> `task_created` | `skipped` | `coalesced` | `failed`), `trigger_payload` (resolved variable values), `linked_task_id` (heavy only), `coalesced_into_run_id`, `dispatch_fingerprint` (hash of resolved template + assignee), and lifecycle timestamps. +Each trigger firing creates a routine run record (`office_routine_runs`) with `routine_id`, `trigger_id`, `source` (`cron` | `webhook` | `manual`), `status` (`received` -> `task_created` | `done` | `skipped` | `coalesced` | `failed`, with heavy `task_created` runs later ending as `done`, `cancelled`, or `failed`), `trigger_payload` (resolved variable values), `linked_task_id` (heavy only), `coalesced_into_run_id`, `dispatch_fingerprint` (hash of resolved template + assignee), and lifecycle timestamps. #### Heavy vs lightweight routines @@ -137,7 +140,7 @@ Evaluated at dispatch by querying for an in-flight run for the same routine fing - `coalesce_if_active` (default): merge into the existing run. Mark `coalesced`. - `always_enqueue` / `always_create`: always proceed. -"Active" means the linked task / run is not in a terminal state. +"Active" means the linked task / run is not in a terminal state. A linked task is also inactive when it is archived or missing. The gate checks task state directly and does not release a live task because of its age. #### Catch-up policy diff --git a/docs/specs/office/system-design/scheduler-02.md b/docs/specs/office/system-design/scheduler-02.md index b874928725..96d2756f8d 100644 --- a/docs/specs/office/system-design/scheduler-02.md +++ b/docs/specs/office/system-design/scheduler-02.md @@ -239,7 +239,11 @@ Triggered manually via UI / API. ### Routine run ``` +received -> done (lightweight routine: wakeup dispatched) received -> task_created (heavy routine: real task created) +task_created -> done (linked task completed) +task_created -> cancelled (linked task cancelled or archived) +task_created -> failed (linked task failed or disappeared) received -> skipped (concurrency_policy=skip_if_active and active run exists) received -> coalesced (concurrency_policy=coalesce_if_active and active run exists) received -> failed (dispatch error) @@ -293,7 +297,7 @@ Survives a kandev process restart: all `agent_wakeup_requests` rows including `q Does NOT survive (reconstructed on next tick): in-memory claim leases - a `claimed` wakeup whose process died is picked up by the staleness/recovery path; the scheduler's claim query is the source of truth. The unstarted-task recovery sweep suppresses duplicates with its `NOT EXISTS` check on `runs`: any queued, claimed, or finished run of any age blocks redispatch, while failed and cancelled runs do not. -Retention: the idempotency lookup window is 24 hours, while persisted idempotency keys remain unique; summary cap 8 KB per row; routine run history retained for inspection (no automatic prune in scope here); catch-up cap (default 25) drops missed routine ticks beyond it (not recorded individually). +Retention: the idempotency lookup window is 24 hours, while persisted idempotency keys remain unique; summary cap 8 KB per row; routine run history retained for inspection (no automatic prune in scope here); catch-up cap (default 25) drops missed routine ticks beyond it (not recorded individually). A live linked task remains in the concurrency gate regardless of age. The scheduler reads all `queued` and unexpired-retry wakeup requests on boot and resumes processing them.