fix(office): make auto-start work for Office tasks and surface failures on the kanban card (#2847)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
nova28
2026-08-21 01:12:31 +08:00
committed by GitHub
parent e9b6f45381
commit d5039f89c0
47 changed files with 1708 additions and 91 deletions
@@ -691,6 +691,7 @@ func mapKanbanTaskState(task taskdto.TaskDTO) map[string]any {
"queuedForStepId": nullString(task.QueuedForStepID),
"queuedAt": task.QueuedAt,
"interrupted": task.Interrupted,
"autoStartFailed": task.AutoStartFailed,
"statusSummary": task.StatusSummary,
"sessionCount": task.SessionCount,
"reviewStatus": nullString(string(task.ReviewStatus)),
@@ -31,6 +31,31 @@ func TestMapKanbanStateIncludesWIPAdmissionFields(t *testing.T) {
}
}
// TestMapKanbanTaskStateIncludesAutoStartFailed regression-tests Review round
// 2's MAJOR finding: mapKanbanTaskState is a camelCase whitelist that omitted
// auto_start_failed, so a task whose auto-start already failed rendered with
// no badge until the WS path (now fixed separately) delivered an update — the
// very first paint was wrong for a task that failed before this boot payload
// was ever built.
func TestMapKanbanTaskStateIncludesAutoStartFailed(t *testing.T) {
task := mapKanbanTaskState(taskdto.TaskDTO{
ID: "task-auto-start-failed",
WorkflowStepID: "step-review",
AutoStartFailed: true,
})
if task["autoStartFailed"] != true {
t.Fatalf("kanban task autoStartFailed = %#v, want true", task["autoStartFailed"])
}
cleared := mapKanbanTaskState(taskdto.TaskDTO{
ID: "task-auto-start-ok",
WorkflowStepID: "step-review",
})
if cleared["autoStartFailed"] != false {
t.Fatalf("kanban task autoStartFailed = %#v, want false for a task without the marker", cleared["autoStartFailed"])
}
}
func TestMapUserSettingsStateIncludesAzureDevOpsBrowsePreferences(t *testing.T) {
preferences := json.RawMessage(`{"workspace-1":{"mode":"board","filters":{"projectId":"project-2"},"board":{"teamId":"team-2","boardId":"board-2","focusedColumnId":"done"}}}`)
state := mapUserSettingsState(userdto.UserSettingsResponse{
@@ -220,7 +220,14 @@ func (si *SchedulerIntegration) processRun(ctx context.Context, run *models.Run)
}
// Staleness check.
if cancel, reason := si.evaluateRunStaleness(ctx, run); cancel {
cancel, reason, staleErr := si.evaluateRunStaleness(ctx, run)
if staleErr != nil {
si.logger.Warn("run staleness check failed; retrying run",
zap.String("run_id", runID), zap.Error(staleErr))
_ = si.svc.HandleRunFailure(ctx, run, staleErr)
return
}
if cancel {
si.cancelStaleRun(ctx, run, agent, reason)
return
}
@@ -54,6 +54,45 @@ func TestSchedulerIntegration_TickProcessesRun(t *testing.T) {
}
}
func TestSchedulerIntegration_CancelsRunForMovedWorkflowStep(t *testing.T) {
mock := &mockTaskStarter{}
svc := newTestService(t, service.ServiceOptions{TaskStarter: mock})
ctx := context.Background()
agent := makeAgent("worker-moved-step", models.AgentRoleWorker)
agent.ExecutorPreference = `{"type":"local_pc"}`
if err := svc.CreateAgentInstance(ctx, agent); err != nil {
t.Fatalf("create agent: %v", err)
}
svc.ExecSQL(t, `INSERT INTO tasks
(id, workspace_id, workflow_step_id, title, created_at, updated_at)
VALUES ('task-moved-step', 'ws-1', 'step-current', 'Moved task',
CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`)
// The run was queued while the task was on step-old. A later workflow
// move must prevent the scheduler from launching that stale step.
if err := svc.QueueRun(ctx, agent.ID, service.RunReasonTaskAssigned,
`{"task_id":"task-moved-step","workflow_step_id":"step-old"}`, ""); err != nil {
t.Fatalf("queue run: %v", err)
}
service.RunSchedulerTick(svc, ctx)
if mock.callCount() != 0 {
t.Fatalf("StartTask calls = %d, want 0 for stale workflow step", mock.callCount())
}
runs, err := svc.ListRuns(ctx, "ws-1")
if err != nil {
t.Fatalf("list runs: %v", err)
}
if len(runs) != 1 {
t.Fatalf("runs = %d, want 1", len(runs))
}
if runs[0].Status != service.RunStatusCancelled {
t.Fatalf("run status = %q, want %q", runs[0].Status, service.RunStatusCancelled)
}
}
func TestSchedulerIntegration_ResolvesExecutorFromTaskProject(t *testing.T) {
mock := &mockTaskStarter{}
svc := newTestService(t, service.ServiceOptions{TaskStarter: mock})
@@ -2,6 +2,7 @@ package service
import (
"context"
"fmt"
"time"
"go.uber.org/zap"
@@ -17,18 +18,30 @@ const staleRunThreshold = 2 * time.Hour
func (si *SchedulerIntegration) evaluateRunStaleness(
ctx context.Context,
run *models.Run,
) (bool, string) {
if run.RetryCount == 0 {
return false, ""
) (bool, string, error) {
if run.RetryCount > 0 && !run.RequestedAt.IsZero() &&
time.Since(run.RequestedAt) > staleRunThreshold {
return true, "execution_too_old", nil
}
if run.RequestedAt.IsZero() {
return false, ""
// Engine-emitted runs carry the workflow step that produced them. A task
// can move to another step while the run waits in the queue, so compare
// the queued step with the task's current step before launch. Legacy runs
// do not include workflow_step_id and retain the existing behaviour.
payload := ParseRunPayload(run.Payload)
expectedStepID := payload["workflow_step_id"]
taskID := payload["task_id"]
if expectedStepID == "" || taskID == "" {
return false, "", nil
}
if time.Since(run.RequestedAt) > staleRunThreshold {
return true, "execution_too_old"
currentStepID, err := si.svc.repo.GetTaskWorkflowStepID(ctx, taskID)
if err != nil {
return false, "", fmt.Errorf("resolve current workflow step for task %s: %w", taskID, err)
}
_ = ctx
return false, ""
if currentStepID != expectedStepID {
return true, "workflow_step_changed", nil
}
return false, "", nil
}
// cancelStaleRun marks the run cancelled, logs the event, and releases any checkout.
@@ -2,6 +2,7 @@ package orchestrator
import (
"context"
"errors"
"strings"
"sync/atomic"
"testing"
@@ -576,6 +577,89 @@ func TestAutoStart_NoTokenDoesNotBlock(t *testing.T) {
}
}
// TestAutoStart_FailedLaunchWithoutGuardDoesNotStampClaimed is the regression
// test for the bogus-claim-restore bug: restoreAutoStartClaim used to run
// unconditionally on any launch failure, so a task that never carried
// MetaKeyAutoStartGuard (and therefore never claimed MetaKeyAutoStartClaimed
// in the first place) still gained the token after a failed launch. A failed
// launch on a guardless task must leave auto_start_claimed absent, and must
// stamp auto_start_failed instead so the failure surfaces on the card.
func TestAutoStart_FailedLaunchWithoutGuardDoesNotStampClaimed(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
const taskID = "task-fail-no-guard"
const stepID = "step-review"
sg := newMockStepGetter()
sg.steps[stepID] = &wfmodels.WorkflowStep{
ID: stepID, WorkflowID: "wf1", Name: "Review", Position: 0,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
},
}
now := time.Now().UTC()
_ = repo.CreateWorkspace(ctx, &taskmodels.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})
_ = repo.CreateWorkflow(ctx, &taskmodels.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})
_ = repo.CreateTask(ctx, &taskmodels.Task{
ID: taskID, WorkspaceID: "ws1", WorkflowID: "wf1", WorkflowStepID: stepID,
Title: "T", Description: "D", State: v1.TaskStateInProgress,
Metadata: map[string]interface{}{taskmodels.MetaKeyAgentProfileID: testAgentProfileID},
CreatedAt: now, UpdatedAt: now,
})
taskRepo := newMockTaskRepo()
taskRepo.tasks[taskID] = &v1.Task{
ID: taskID,
State: v1.TaskStateInProgress,
Metadata: map[string]interface{}{
taskmodels.MetaKeyAgentProfileID: testAgentProfileID,
},
}
attempted := make(chan struct{}, 1)
agentMgr := &mockAgentManager{
repoForExecutionLookup: repo,
launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
attempted <- struct{}{}
return nil, errors.New("boom")
},
}
svc := createTestServiceWithScheduler(repo, sg, taskRepo, agentMgr)
svc.autoStartTaskForStep(ctx, taskID, stepID, "task.moved")
select {
case <-attempted:
case <-time.After(2 * time.Second):
t.Fatal("expected the launch attempt to fire")
}
deadline := time.After(2 * time.Second)
for {
task, err := repo.GetTask(ctx, taskID)
if err != nil {
t.Fatalf("GetTask: %v", err)
}
if task.Metadata[taskmodels.MetaKeyAutoStartFailed] != nil {
break
}
select {
case <-deadline:
t.Fatal("timed out waiting for auto_start_failed marker to be set")
case <-time.After(10 * time.Millisecond):
}
}
task, err := repo.GetTask(ctx, taskID)
if err != nil {
t.Fatalf("GetTask: %v", err)
}
if task.Metadata[taskmodels.MetaKeyAutoStartClaimed] != nil {
t.Error("a failed launch on a task without the guard must not stamp auto_start_claimed")
}
}
func TestReviewWatchLifecycle_BootReadyStaysOnReviewUntilTurnComplete(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
@@ -1022,6 +1022,7 @@ func (s *Service) updateTaskSessionStateWithHook(
// drop the red interruption icon. No-op when the marker is absent.
if nextState == models.TaskSessionStateStarting || nextState == models.TaskSessionStateRunning {
s.clearTaskInterruptedMarker(ctx, taskID)
s.clearTaskAutoStartFailedMarker(ctx, taskID)
}
if authoritativeUpdatedAt == nil {
s.logger.Warn("skipping session state_changed publish; could not read authoritative updated_at",
@@ -1769,6 +1770,7 @@ func (s *Service) setSessionStarting(
// updateTaskSessionStateWithHook, so clear the interruption marker here
// too (no-op when absent).
s.clearTaskInterruptedMarker(ctx, taskID)
s.clearTaskAutoStartFailedMarker(ctx, taskID)
if publishSession != nil {
s.publishTaskSessionStateChanged(ctx, taskID, session.ID, oldState, session.State, session.ErrorMessage, stateUpdatedAt, publishSession)
@@ -1805,6 +1807,36 @@ func (s *Service) clearTaskInterruptedMarker(ctx context.Context, taskID string)
s.publishTaskUpdated(ctx, task)
}
// clearTaskAutoStartFailedMarker removes the auto-start-failure marker from a
// task and republishes task.updated when it was actually present, so open
// clients drop the failure badge. Called from the same session-start funnel
// as clearTaskInterruptedMarker: a session entering STARTING/RUNNING means an
// agent did launch, so any earlier auto-start failure no longer applies.
// No-op when the marker is absent.
func (s *Service) clearTaskAutoStartFailedMarker(ctx context.Context, taskID string) {
if taskID == "" {
return
}
removed, err := s.repo.RemoveTaskMetadataKey(ctx, taskID, models.MetaKeyAutoStartFailed)
if err != nil {
s.logger.Warn("failed to clear auto-start-failed marker",
zap.String("task_id", taskID),
zap.Error(err))
return
}
if !removed {
return
}
task, err := s.repo.GetTask(ctx, taskID)
if err != nil || task == nil {
s.logger.Warn("failed to load task for auto-start-failed-clear publish",
zap.String("task_id", taskID),
zap.Error(err))
return
}
s.publishTaskUpdated(ctx, task)
}
func (s *Service) persistFullTaskSessionIfCurrent(
ctx context.Context,
session *models.TaskSession,
@@ -2644,6 +2644,72 @@ func TestSessionStartClearsInterruptedMarker(t *testing.T) {
})
}
func TestSessionStartClearsAutoStartFailedMarker(t *testing.T) {
ctx := context.Background()
t.Run("state hook transition to STARTING", func(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
require.NoError(t, repo.SetTaskMetadataKey(ctx, "t1", models.MetaKeyAutoStartFailed, true))
publisher := &recordingTaskUpdatedPublisher{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.SetTaskEventPublisher(publisher)
_, changed := svc.updateTaskSessionStateWithHook(
ctx, "t1", "s1", models.TaskSessionStateStarting, "", false, nil,
)
require.True(t, changed)
task, err := repo.GetTask(ctx, "t1")
require.NoError(t, err)
_, marked := task.Metadata[models.MetaKeyAutoStartFailed]
require.False(t, marked)
require.Equal(t, []string{"t1"}, publisher.updatedTaskIDs)
})
t.Run("launch path via setSessionStarting", func(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
require.NoError(t, repo.SetTaskMetadataKey(ctx, "t1", models.MetaKeyAutoStartFailed, true))
session, err := repo.GetTaskSession(ctx, "s1")
require.NoError(t, err)
session.State = models.TaskSessionStateStarting
session.ErrorMessage = ""
session.UpdatedAt = time.Now().UTC()
publisher := &recordingTaskUpdatedPublisher{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.SetTaskEventPublisher(publisher)
require.NoError(t, svc.setSessionStarting(ctx, "t1", session, models.TaskSessionStateRunning, true))
task, err := repo.GetTask(ctx, "t1")
require.NoError(t, err)
_, marked := task.Metadata[models.MetaKeyAutoStartFailed]
require.False(t, marked)
require.Equal(t, []string{"t1"}, publisher.updatedTaskIDs)
})
}
func TestAutoStartFailureDoesNotMarkTaskWithActiveSession(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
require.NoError(t, repo.UpdateTaskSessionState(
ctx, "s1", models.TaskSessionStateStarting, "",
))
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.setTaskAutoStartFailedMarker(ctx, "t1", "task.moved")
task, err := repo.GetTask(ctx, "t1")
require.NoError(t, err)
_, marked := task.Metadata[models.MetaKeyAutoStartFailed]
require.False(t, marked, "a late failure must not mark a task with active work")
}
// Nothing else ever retires a stored agent failure, so without this clear every
// path that re-derives task status from session metadata resurrects it and the
// error icon never goes away.
@@ -39,6 +39,10 @@ type taskMetadataKeySetter interface {
SetTaskMetadataKey(context.Context, string, string, interface{}) error
}
type taskMetadataKeySetterIfNoActiveSession interface {
SetTaskMetadataKeyIfNoActiveSession(context.Context, string, string, interface{}) (bool, error)
}
type lifecycleTaskMetadataLister interface {
ListTasksWithMetadataKey(context.Context, string) ([]*models.Task, error)
}
@@ -1071,6 +1075,11 @@ func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.T
return
}
if s.isOfficeTask(ctx, task.ID) {
s.autoStartOfficeTaskForLoadedStep(ctx, task, step, eventName, restoreQueuePromotion)
return
}
workflowAgentProfileID := s.resolveStepAgentProfile(ctx, step)
agentProfileID := workflowAgentProfileID
if agentProfileID == "" {
@@ -1097,7 +1106,8 @@ func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.T
// can fire. Only the first atomic claim of MetaKeyAutoStartClaimed wins;
// the loser skips launch because the winner will handle it.
// Absent guard = ordinary (non-watcher) auto-start; proceed as before.
if _, hasGuard := task.Metadata[models.MetaKeyAutoStartGuard]; hasGuard {
_, hasGuard := task.Metadata[models.MetaKeyAutoStartGuard]
if hasGuard {
if !s.claimAutoStart(asyncCtx, task.ID, eventName) {
s.logger.Debug(eventName+": auto-start claim lost; watcher path will launch",
zap.String("task_id", task.ID))
@@ -1114,14 +1124,103 @@ func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.T
s.logger.Error(eventName+": failed to auto-start task",
zap.String("task_id", task.ID),
zap.Error(err))
s.restoreAutoStartClaim(asyncCtx, task.ID, eventName)
if restoreQueuePromotion {
s.restoreTaskLifecycleToken(asyncCtx, task.ID, models.MetaKeyQueuePromotionPending, eventName)
}
s.handleAutoStartFailure(asyncCtx, task.ID, eventName, hasGuard, restoreQueuePromotion)
}
}()
}
// autoStartOfficeTaskForLoadedStep is the Office-aware counterpart of the
// kanban launch path above. An Office task has no session-based launch route
// (see validateOfficeRuntimeEnv) — StartTask can never succeed for it, it
// just logs "office runtime context is incomplete" and starts nothing. Queue
// a run through the engine's Office adapters instead, the same mechanism the
// scheduler's recovery sweep and the "assign task" flow already use to start
// Office work (internal/office/service/scheduler_recovery.go).
func (s *Service) autoStartOfficeTaskForLoadedStep(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep, eventName string, restoreQueuePromotion bool) {
s.logger.Info(eventName+": queueing office run (no session, auto-start step)",
zap.String("task_id", task.ID),
zap.String("to_step_id", step.ID))
// Async for the same reason as the kanban path above: the event bus
// delivers synchronously and blocking here would stall the HTTP handler
// that published the move.
go func() {
asyncCtx := context.WithoutCancel(ctx)
_, hasGuard := task.Metadata[models.MetaKeyAutoStartGuard]
if hasGuard {
if !s.claimAutoStart(asyncCtx, task.ID, eventName) {
s.logger.Debug(eventName+": auto-start claim lost; watcher path will launch",
zap.String("task_id", task.ID))
return
}
}
if err := s.queueOfficeAutoStartRun(asyncCtx, task, step); err != nil {
s.logger.Error(eventName+": failed to queue office auto-start run",
zap.String("task_id", task.ID),
zap.Error(err))
s.handleAutoStartFailure(asyncCtx, task.ID, eventName, hasGuard, restoreQueuePromotion)
}
}()
}
// officeAutoStartRunReason mirrors office/service.RunReasonTaskAssigned
// ("task_assigned"). Kept as a literal rather than importing
// internal/office/service: the two packages intentionally avoid importing
// each other (see runsServiceEngineAdapter in backendapp/main.go), so a
// shared reason string is duplicated rather than pulled across the boundary.
const officeAutoStartRunReason = "task_assigned"
// queueOfficeAutoStartRun makes auto_start_agent Office-aware: instead of
// StartTask (kanban-only), it resolves an agent the same way a "primary"
// queue_run target would — preferring the task's current runner participant,
// falling back to its assignee — and queues a run through engineRunQueue.
func (s *Service) queueOfficeAutoStartRun(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep) error {
if s.engineRunQueue == nil {
return fmt.Errorf("office run queue is not wired")
}
agentProfileID := task.AssigneeAgentProfileID
if s.enginePrimary != nil {
resolved, err := s.enginePrimary.PrimaryAgentProfileID(ctx, step.ID, task.ID)
if err != nil {
return fmt.Errorf("resolve primary agent for office auto-start on task %s: %w", task.ID, err)
}
if resolved != "" {
agentProfileID = resolved
}
}
if agentProfileID == "" {
return fmt.Errorf("no agent profile resolved for office auto-start on task %s", task.ID)
}
return s.engineRunQueue.QueueRun(ctx, engine.QueueRunRequest{
AgentProfileID: agentProfileID,
TaskID: task.ID,
WorkflowStepID: step.ID,
Reason: officeAutoStartRunReason,
IdempotencyKey: fmt.Sprintf("%s:%s:%s:%s", officeAutoStartRunReason, task.ID, agentProfileID, step.ID),
Payload: map[string]any{"task_id": task.ID},
})
}
// handleAutoStartFailure records a failed auto-start attempt. It restores the
// MetaKeyAutoStartClaimed token only when hasGuard is true: claimAutoStart is
// only ever called (and the token only ever taken) when the task carries
// MetaKeyAutoStartGuard, so restoring it unconditionally stamps
// auto_start_claimed onto tasks that never carried the guard, corrupting the
// invariant documented at MetaKeyAutoStartGuard. It also stamps
// MetaKeyAutoStartFailed so the failure surfaces on the task card instead of
// only in backend logs.
func (s *Service) handleAutoStartFailure(ctx context.Context, taskID, eventName string, hasGuard, restoreQueuePromotion bool) {
if hasGuard {
s.restoreAutoStartClaim(ctx, taskID, eventName)
}
if restoreQueuePromotion {
s.restoreTaskLifecycleToken(ctx, taskID, models.MetaKeyQueuePromotionPending, eventName)
}
s.setTaskAutoStartFailedMarker(ctx, taskID, eventName)
}
func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eventName string) bool {
if task.Metadata == nil {
return false
@@ -1247,6 +1346,37 @@ func (s *Service) restoreAutoStartClaim(ctx context.Context, taskID, eventName s
}
}
// setTaskAutoStartFailedMarker stamps MetaKeyAutoStartFailed on a task and
// republishes task.updated so the failure surfaces on the card. Cleared by
// clearTaskAutoStartFailedMarker when a session for the task next reaches
// STARTING/RUNNING.
func (s *Service) setTaskAutoStartFailedMarker(ctx context.Context, taskID, eventName string) {
setter, ok := s.repo.(taskMetadataKeySetterIfNoActiveSession)
if !ok {
s.logger.Warn(eventName+": repository cannot conditionally set auto-start-failed marker",
zap.String("task_id", taskID))
return
}
written, err := setter.SetTaskMetadataKeyIfNoActiveSession(ctx, taskID, models.MetaKeyAutoStartFailed, true)
if err != nil {
s.logger.Warn(eventName+": failed to set auto-start-failed marker",
zap.String("task_id", taskID), zap.Error(err))
return
}
if !written {
s.logger.Debug(eventName+": skipped auto-start-failed marker because task work is active",
zap.String("task_id", taskID))
return
}
task, err := s.repo.GetTask(ctx, taskID)
if err != nil || task == nil {
s.logger.Warn(eventName+": failed to load task for auto-start-failed publish",
zap.String("task_id", taskID), zap.Error(err))
return
}
s.publishTaskUpdated(ctx, task)
}
// handleTaskMovedWithSession handles the case where a task with an existing session
// is moved between steps. It processes on_exit for the source step and on_enter
// for the target step.
@@ -2,6 +2,7 @@ package orchestrator
import (
"context"
"errors"
"testing"
"testing/synctest"
"time"
@@ -10,10 +11,74 @@ import (
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
"github.com/kandev/kandev/internal/orchestrator/watcher"
"github.com/kandev/kandev/internal/task/models"
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/workflow/engine"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// fakeRunQueueAdapter is a minimal engine.RunQueueAdapter test double that
// captures QueueRun calls on a channel so tests can assert both that a call
// happened and what it carried, without a sleep-based race.
type fakeRunQueueAdapter struct {
calls chan engine.QueueRunRequest
err error
}
func (f *fakeRunQueueAdapter) QueueRun(_ context.Context, req engine.QueueRunRequest) error {
if f.calls != nil {
f.calls <- req
}
return f.err
}
// fakePrimaryAgentResolver is a minimal engine.PrimaryAgentResolver test
// double returning a fixed agent profile id (or error).
type fakePrimaryAgentResolver struct {
agentProfileID string
err error
}
func (f *fakePrimaryAgentResolver) PrimaryAgentProfileID(_ context.Context, _, _ string) (string, error) {
return f.agentProfileID, f.err
}
// failIfLaunched builds a mockAgentManager whose LaunchAgent fails the test
// if invoked — used by the Office auto-start tests below to prove the
// kanban StartTask path is never reached for an Office task.
func failIfLaunched(t *testing.T) *mockAgentManager {
t.Helper()
return &mockAgentManager{
launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
t.Error("LaunchAgent must not be called for an office task auto-start")
return nil, nil
},
}
}
// waitForAutoStartFailedMarker polls the task until MetaKeyAutoStartFailed is
// set, or fails the test after a timeout. Auto-start failure handling runs in
// a detached goroutine (see autoStartOfficeTaskForLoadedStep), so tests must
// observe it via the persisted side effect rather than a direct return value.
func waitForAutoStartFailedMarker(t *testing.T, repo *sqliterepo.Repository, ctx context.Context, taskID string) {
t.Helper()
deadline := time.After(2 * time.Second)
for {
task, err := repo.GetTask(ctx, taskID)
if err != nil {
t.Fatalf("GetTask: %v", err)
}
if task.Metadata[models.MetaKeyAutoStartFailed] != nil {
return
}
select {
case <-deadline:
t.Fatal("timed out waiting for auto_start_failed marker to be set")
case <-time.After(10 * time.Millisecond):
}
}
}
func TestHandleTaskMoved(t *testing.T) {
ctx := context.Background()
@@ -226,6 +291,209 @@ func TestHandleTaskMovedNoSession(t *testing.T) {
t.Fatal("timed out waiting for auto-start launch")
}
})
t.Run("office task queues a run instead of starting a session", func(t *testing.T) {
repo := setupTestRepo(t)
now := time.Now().UTC()
requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}))
// ProjectID non-empty is what makes IsFromOffice (a read-time SQL
// projection) evaluate true for this task.
requireNoError(t, repo.CreateTask(ctx, &models.Task{
ID: "t-office",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step1",
ProjectID: "proj1",
Title: "Office Task",
Description: "prompt",
State: v1.TaskStateCreated,
AssigneeAgentProfileID: "assignee-profile",
CreatedAt: now,
UpdatedAt: now,
}))
stepGetter := newMockStepGetter()
stepGetter.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Work", Position: 1,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{
{Type: wfmodels.OnEnterAutoStartAgent},
},
},
}
svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t))
queued := &fakeRunQueueAdapter{calls: make(chan engine.QueueRunRequest, 1)}
svc.engineRunQueue = queued
svc.enginePrimary = &fakePrimaryAgentResolver{agentProfileID: "resolved-primary"}
svc.handleTaskMovedNoSession(ctx, watcher.TaskMovedEventData{
TaskID: "t-office",
ToStepID: "step2",
})
select {
case req := <-queued.calls:
if req.AgentProfileID != "resolved-primary" {
t.Errorf("AgentProfileID = %q, want %q (from PrimaryAgentResolver, not task assignee)", req.AgentProfileID, "resolved-primary")
}
if req.TaskID != "t-office" {
t.Errorf("TaskID = %q, want t-office", req.TaskID)
}
if req.WorkflowStepID != "step2" {
t.Errorf("WorkflowStepID = %q, want step2", req.WorkflowStepID)
}
if req.Reason != officeAutoStartRunReason {
t.Errorf("Reason = %q, want %q", req.Reason, officeAutoStartRunReason)
}
if req.IdempotencyKey != "task_assigned:t-office:resolved-primary:step2" {
t.Errorf("IdempotencyKey = %q, want task_assigned:t-office:resolved-primary:step2", req.IdempotencyKey)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for QueueRun call")
}
})
t.Run("office task with unwired run queue sets auto_start_failed and does not panic", func(t *testing.T) {
repo := setupTestRepo(t)
now := time.Now().UTC()
requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateTask(ctx, &models.Task{
ID: "t-office-nil-adapter",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step1",
ProjectID: "proj1",
Title: "Office Task",
Description: "prompt",
State: v1.TaskStateCreated,
AssigneeAgentProfileID: "assignee-profile",
CreatedAt: now,
UpdatedAt: now,
}))
stepGetter := newMockStepGetter()
stepGetter.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Work", Position: 1,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{
{Type: wfmodels.OnEnterAutoStartAgent},
},
},
}
// engineRunQueue is left nil (never wired) — this is the exact defect
// scenario: no Office adapters wired, no way to launch, must fail
// visibly rather than silently.
svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t))
svc.handleTaskMovedNoSession(ctx, watcher.TaskMovedEventData{
TaskID: "t-office-nil-adapter",
ToStepID: "step2",
})
waitForAutoStartFailedMarker(t, repo, ctx, "t-office-nil-adapter")
task, err := repo.GetTask(ctx, "t-office-nil-adapter")
if err != nil {
t.Fatalf("GetTask: %v", err)
}
if task.Metadata[models.MetaKeyAutoStartClaimed] != nil {
t.Error("a task without the guard must not gain auto_start_claimed from a failed office queue attempt")
}
})
t.Run("office task with no resolvable agent sets auto_start_failed and does not panic", func(t *testing.T) {
repo := setupTestRepo(t)
now := time.Now().UTC()
requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}))
// No AssigneeAgentProfileID and no PrimaryAgentResolver result: the
// office branch has an adapter to call but nothing to call it with.
requireNoError(t, repo.CreateTask(ctx, &models.Task{
ID: "t-office-no-agent",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step1",
ProjectID: "proj1",
Title: "Office Task",
Description: "prompt",
State: v1.TaskStateCreated,
CreatedAt: now,
UpdatedAt: now,
}))
stepGetter := newMockStepGetter()
stepGetter.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Work", Position: 1,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{
{Type: wfmodels.OnEnterAutoStartAgent},
},
},
}
svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t))
queued := &fakeRunQueueAdapter{calls: make(chan engine.QueueRunRequest, 1)}
svc.engineRunQueue = queued
svc.enginePrimary = &fakePrimaryAgentResolver{agentProfileID: ""}
svc.handleTaskMovedNoSession(ctx, watcher.TaskMovedEventData{
TaskID: "t-office-no-agent",
ToStepID: "step2",
})
waitForAutoStartFailedMarker(t, repo, ctx, "t-office-no-agent")
select {
case <-queued.calls:
t.Error("QueueRun must not be called when no agent profile could be resolved")
default:
}
})
t.Run("office task does not fall back after primary resolver failure", func(t *testing.T) {
repo := setupTestRepo(t)
now := time.Now().UTC()
requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateTask(ctx, &models.Task{
ID: "t-office-primary-error",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step1",
ProjectID: "proj1",
Title: "Office Task",
Description: "prompt",
State: v1.TaskStateCreated,
AssigneeAgentProfileID: "stale-assignee",
CreatedAt: now,
UpdatedAt: now,
}))
stepGetter := newMockStepGetter()
stepGetter.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Work", Position: 1,
Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}},
}
svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t))
queued := &fakeRunQueueAdapter{calls: make(chan engine.QueueRunRequest, 1)}
svc.engineRunQueue = queued
svc.enginePrimary = &fakePrimaryAgentResolver{err: errors.New("runner lookup failed")}
svc.handleTaskMovedNoSession(ctx, watcher.TaskMovedEventData{
TaskID: "t-office-primary-error", ToStepID: "step2",
})
waitForAutoStartFailedMarker(t, repo, ctx, "t-office-primary-error")
select {
case req := <-queued.calls:
t.Fatalf("QueueRun called with %q after primary resolver failure", req.AgentProfileID)
default:
}
})
}
func TestHandleTaskMovedWithSession(t *testing.T) {
@@ -401,7 +401,18 @@ func (r *Repository) CoalesceRun(
ctx context.Context, agentInstanceID, reason string, windowSecs int, payload string,
) (bool, error) {
cutoff := time.Now().UTC().Add(-time.Duration(windowSecs) * time.Second)
res, err := r.db.ExecContext(ctx, r.db.Rebind(`
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" {
taskPredicate = fmt.Sprintf(" AND %s = ?", dialect.JSONExtract(r.db.DriverName(), "payload", "task_id"))
args = append(args, taskID)
}
query := fmt.Sprintf(`
UPDATE runs
SET coalesced_count = coalesced_count + 1, payload = ?
WHERE id = (
@@ -409,10 +420,12 @@ func (r *Repository) CoalesceRun(
WHERE agent_profile_id = ? AND reason = ? AND status = 'queued'
AND requested_at > ?
AND (idempotency_key IS NULL OR idempotency_key NOT LIKE ?)
%s
ORDER BY requested_at DESC
LIMIT 1
)
`), payload, agentInstanceID, reason, cutoff, commentkeys.TaskCommentPrefix+"%")
`, taskPredicate)
res, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...)
if err != nil {
return false, err
}
@@ -423,6 +436,15 @@ func (r *Repository) CoalesceRun(
return rows > 0, nil
}
func taskIDFromPayload(payload string) string {
var raw map[string]any
if err := json.Unmarshal([]byte(payload), &raw); err != nil {
return ""
}
taskID, _ := raw["task_id"].(string)
return taskID
}
// ClaimNextEligibleRun atomically claims the next queued run,
// skipping runs with a scheduled retry time in the future and
// agents that already have a claimed run. Agent status and cooldown
@@ -53,6 +53,24 @@ func TestCoalesceRun_MergesIntoMostRecentQueuedRun(t *testing.T) {
}
}
func TestCoalesceRun_DoesNotMergeDifferentTaskRuns(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
queued := queueRunAt(t, repo, "task-a", "a1", time.Now().UTC())
merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"task-b"}`)
if err != nil {
t.Fatalf("coalesce: %v", err)
}
if merged {
t.Fatal("coalesce = true for a different task, want false")
}
got := mustGetRun(t, repo, queued.ID)
checkInt(t, "coalesced_count", got.CoalescedCount, 1)
checkString(t, "payload", got.Payload, `{"task_id":"task-a"}`)
}
// TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow pins the counter
// arithmetic across several merges into the same run.
func TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow(t *testing.T) {
@@ -73,6 +73,48 @@ func TestQueueRun_InsertsRow(t *testing.T) {
}
}
func TestQueueRun_DoesNotCoalesceTaskAssignedAcrossTasks(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,
WorkflowStepID: "step-1",
Reason: "task_assigned",
IdempotencyKey: "task_assigned:" + taskID + ":agent-primary:step-1",
}); err != nil {
t.Fatalf("queue %s: %v", taskID, err)
}
}
var payloads []string
if err := repo.Reader().SelectContext(ctx, &payloads,
`SELECT payload FROM runs WHERE agent_profile_id = ? ORDER BY requested_at`,
"agent-primary"); err != nil {
t.Fatalf("list queued payloads: %v", err)
}
if len(payloads) != 2 {
t.Fatalf("queued task_assigned runs = %d, want 2", len(payloads))
}
seen := make(map[string]bool, len(payloads))
for _, raw := range payloads {
var payload map[string]any
if err := json.Unmarshal([]byte(raw), &payload); err != nil {
t.Fatalf("decode payload %q: %v", raw, err)
}
taskID, ok := payload["task_id"].(string)
if !ok {
t.Fatalf("payload task_id = %v, want string", payload["task_id"])
}
seen[taskID] = true
}
if !seen["task-a"] || !seen["task-b"] {
t.Fatalf("queued task IDs = %#v, want task-a and task-b", seen)
}
}
func TestQueueRun_UsesRequestAgentProfileIDAndAddsEnvelopePayload(t *testing.T) {
svc, _, repo := newTestServiceWithRepo(t)
ctx := context.Background()
+6
View File
@@ -207,6 +207,11 @@ type TaskDTO struct {
// died and has not been resumed since. Derived from the interrupted_at
// metadata key at DTO conversion time (see FromTaskWithSessionInfo).
Interrupted bool `json:"interrupted,omitempty"`
// AutoStartFailed reports that a workflow step's auto_start_agent on_enter
// action failed to launch a run for this task. Derived from the
// auto_start_failed metadata key at DTO conversion time (see
// FromTaskWithSessionInfo).
AutoStartFailed bool `json:"auto_start_failed,omitempty"`
// Dependency projection. Derived on every read from task_blockers plus each
// related task's own state — never persisted, because a stale copy would be
@@ -836,6 +841,7 @@ func FromTaskWithSessionInfo(
UpdatedAt: task.UpdatedAt,
Metadata: task.Metadata,
Interrupted: task.Metadata[models.MetaKeyInterruptedAt] != nil,
AutoStartFailed: task.Metadata[models.MetaKeyAutoStartFailed] != nil,
// Office extensions. AssigneeAgentProfileID is a read-time
// projection from workflow_step_participants (ADR 0005 Wave F);
// the repo's task SELECTs hydrate it via a correlated subquery.
+24 -15
View File
@@ -146,6 +146,14 @@ const (
// surfaces show the red interruption icon; the orchestrator removes the
// key when a session of the task next enters STARTING/RUNNING.
MetaKeyInterruptedAt = "interrupted_at"
// MetaKeyAutoStartFailed is set when a workflow step's auto_start_agent
// on_enter action fails to launch a run (kanban StartTask error, or an
// Office task queue-run error/unresolvable agent). Its presence makes the
// task DTO report `auto_start_failed: true` so the failure surfaces on
// the card instead of only in backend logs; the orchestrator removes the
// key when a session of the task next enters STARTING/RUNNING (mirrors
// MetaKeyInterruptedAt).
MetaKeyAutoStartFailed = "auto_start_failed"
// MetaKeyAgentTitlePending marks tasks created in prompt-first mode whose
// provisional title still needs the first eligible agent session to replace it.
MetaKeyAgentTitlePending = "agent_title_pending"
@@ -2160,21 +2168,22 @@ func (t *Task) ToAPI() *v1.Task {
}
result := &v1.Task{
ID: t.ID,
WorkspaceID: t.WorkspaceID,
WorkflowID: t.WorkflowID,
Title: t.Title,
Description: t.Description,
State: t.State,
Priority: t.Priority,
Repositories: repositories,
CreatedAt: t.CreatedAt,
UpdatedAt: t.UpdatedAt,
Metadata: t.Metadata,
Interrupted: t.Metadata[MetaKeyInterruptedAt] != nil,
IsEphemeral: t.IsEphemeral,
ParentID: t.ParentID,
Autopilot: t.Autopilot,
ID: t.ID,
WorkspaceID: t.WorkspaceID,
WorkflowID: t.WorkflowID,
Title: t.Title,
Description: t.Description,
State: t.State,
Priority: t.Priority,
Repositories: repositories,
CreatedAt: t.CreatedAt,
UpdatedAt: t.UpdatedAt,
Metadata: t.Metadata,
Interrupted: t.Metadata[MetaKeyInterruptedAt] != nil,
AutoStartFailed: t.Metadata[MetaKeyAutoStartFailed] != nil,
IsEphemeral: t.IsEphemeral,
ParentID: t.ParentID,
Autopilot: t.Autopilot,
}
if t.Identifier != "" {
result.Identifier = t.Identifier
@@ -842,6 +842,47 @@ func (r *Repository) SetTaskMetadataKey(ctx context.Context, taskID, key string,
return err
}
// SetTaskMetadataKeyIfNoActiveSession writes one metadata key only when the
// task has no session in STARTING or RUNNING. This prevents a delayed failure
// callback from re-adding a marker after a newer launch has already started.
func (r *Repository) SetTaskMetadataKeyIfNoActiveSession(
ctx context.Context, taskID, key string, value interface{},
) (bool, error) {
payload, err := json.Marshal(value)
if err != nil {
return false, err
}
var query string
if dialect.IsPostgres(r.db.DriverName()) {
query = `UPDATE tasks
SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ?
WHERE id = ?
AND NOT EXISTS (
SELECT 1 FROM task_sessions
WHERE task_id = ? AND state IN (?, ?)
)`
} else {
query = `UPDATE tasks
SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ?
WHERE id = ?
AND NOT EXISTS (
SELECT 1 FROM task_sessions
WHERE task_id = ? AND state IN (?, ?)
)`
}
path := key
if !dialect.IsPostgres(r.db.DriverName()) {
path = jsonPath(key)
}
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(),
taskID, taskID, models.TaskSessionStateStarting, models.TaskSessionStateRunning)
if err != nil {
return false, err
}
rows, err := result.RowsAffected()
return rows > 0, err
}
// SetTaskMetadataKeyIfNotArchived writes one metadata key atomically with the
// archived_at guard: the write only lands when the task row still has
// archived_at IS NULL, and reports whether it landed. Startup reconciliation
@@ -99,6 +99,50 @@ func runMetadataCASContract(t *testing.T, repo *Repository) {
}
}
func runMetadataNoActiveSessionContract(t *testing.T, repo *Repository) {
t.Helper()
ctx := context.Background()
if _, err := repo.db.ExecContext(ctx, repo.db.Rebind(`
INSERT INTO task_sessions (id, task_id, state, started_at, updated_at)
VALUES ('session-cas', ?, 'STARTING', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
`), casTaskID); err != nil {
t.Fatalf("insert starting session: %v", err)
}
written, err := repo.SetTaskMetadataKeyIfNoActiveSession(ctx, casTaskID, "auto_start_failed", true)
if err != nil {
t.Fatalf("SetTaskMetadataKeyIfNoActiveSession while starting: %v", err)
}
if written {
t.Fatal("active STARTING session must block the marker write")
}
if _, present := metadataValue(t, repo, "auto_start_failed"); present {
t.Fatal("blocked marker write changed metadata")
}
if _, err := repo.db.ExecContext(ctx, repo.db.Rebind(`UPDATE task_sessions SET state = 'RUNNING' WHERE id = 'session-cas'`)); err != nil {
t.Fatalf("update running session: %v", err)
}
written, err = repo.SetTaskMetadataKeyIfNoActiveSession(ctx, casTaskID, "auto_start_failed", true)
if err != nil {
t.Fatalf("SetTaskMetadataKeyIfNoActiveSession while running: %v", err)
}
if written {
t.Fatal("active RUNNING session must block the marker write")
}
if _, err := repo.db.ExecContext(ctx, repo.db.Rebind(`UPDATE task_sessions SET state = 'COMPLETED' WHERE id = 'session-cas'`)); err != nil {
t.Fatalf("complete session: %v", err)
}
written, err = repo.SetTaskMetadataKeyIfNoActiveSession(ctx, casTaskID, "auto_start_failed", true)
if err != nil {
t.Fatalf("SetTaskMetadataKeyIfNoActiveSession after completion: %v", err)
}
if !written {
t.Fatal("completed session must allow the marker write")
}
}
func newRepoForMetadataCASTests(t *testing.T) *Repository {
t.Helper()
dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "metadata-cas-test.db"))
@@ -121,6 +165,7 @@ func TestSetTaskMetadataKeyIfPresentSQLite(t *testing.T) {
"other_key": "keep me",
})
runMetadataCASContract(t, repo)
runMetadataNoActiveSessionContract(t, repo)
}
// The JSON patch and its presence predicate are written per dialect, so SQLite
@@ -134,6 +179,7 @@ func TestPostgresSetTaskMetadataKeyIfPresent(t *testing.T) {
"other_key": "keep me",
})
runMetadataCASContract(t, repo)
runMetadataNoActiveSessionContract(t, repo)
}
func newPostgresMetadataCASRepo(t *testing.T, db *sqlx.DB) *Repository {
@@ -385,6 +385,12 @@ func (s *Service) publishTaskEventNow(ctx context.Context, eventType string, tas
// Consumers that restore quick-chat tabs filter on origin, so it has to
// travel with the event and not just the HTTP DTO.
"origin": task.Origin,
// Sent as an explicit true/false (never omitted) so a clear reaches
// open clients too: preserveOmittedField on the frontend only pins the
// previous value when the key is absent from the payload, and an
// omitted key here would make clearTaskAutoStartFailedMarker's publish
// as invisible as the set it is meant to undo.
"auto_start_failed": task.Metadata[models.MetaKeyAutoStartFailed] != nil,
}
data["queued_for_step_id"] = task.QueuedForStepID
if task.QueuedAt != nil {
@@ -603,6 +603,46 @@ func TestPublishTaskUpdated_EmitsAutopilot(t *testing.T) {
}
}
// TestPublishTaskUpdated_EmitsAutoStartFailed regression-tests the WS gap
// found in Review round 2: setTaskAutoStartFailedMarker /
// clearTaskAutoStartFailedMarker both call PublishTaskUpdated expecting open
// clients to see the flip, but publishTaskEventNow hand-builds the payload
// map and never carried auto_start_failed, so the publishes were dead code
// and the badge only ever appeared after a reload (REST snapshot).
func TestPublishTaskUpdated_EmitsAutoStartFailed(t *testing.T) {
svc, eventBus, _ := createTestService(t)
svc.PublishTaskUpdated(context.Background(), &models.Task{
ID: "task-auto-start-failed", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1",
Metadata: map[string]interface{}{models.MetaKeyAutoStartFailed: true},
})
data := singlePublishedEventData(t, eventBus)
if got, ok := data["auto_start_failed"].(bool); !ok || !got {
t.Fatalf("auto_start_failed payload = %#v, want true", data["auto_start_failed"])
}
}
// TestPublishTaskUpdated_EmitsAutoStartFailedFalseWhenCleared proves the
// clear path also propagates: a task with no MetaKeyAutoStartFailed key
// publishes an explicit `false`, not an omitted field, so
// preserveOmittedField on the frontend does not pin the stale `true` from a
// previous failure.
func TestPublishTaskUpdated_EmitsAutoStartFailedFalseWhenCleared(t *testing.T) {
svc, eventBus, _ := createTestService(t)
svc.PublishTaskUpdated(context.Background(), &models.Task{
ID: "task-auto-start-cleared", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1",
})
data := singlePublishedEventData(t, eventBus)
got, ok := data["auto_start_failed"].(bool)
if !ok {
t.Fatalf("auto_start_failed payload missing or wrong type: %#v", data["auto_start_failed"])
}
if got {
t.Fatalf("auto_start_failed payload = true, want false for a task without the marker")
}
}
func TestPublishWorkspaceSourcesAdoptedPublishesSessionScopedPayload(t *testing.T) {
svc, eventBus, _ := createTestService(t)
eventBus.ClearEvents()
+10 -5
View File
@@ -152,11 +152,16 @@ type Task struct {
// when the backend died and has not been resumed since. Derived from the
// interrupted_at metadata key at DTO conversion time; the orchestrator
// clears it when a session of the task next enters STARTING/RUNNING.
Interrupted bool `json:"interrupted,omitempty"`
IsEphemeral bool `json:"is_ephemeral"` // Ephemeral tasks are not shown in kanban, used for quick chat
ParentID string `json:"parent_id,omitempty"` // FK to parent task for subtasks
Autopilot bool `json:"autopilot"`
Identifier string `json:"identifier,omitempty"`
Interrupted bool `json:"interrupted,omitempty"`
// AutoStartFailed reports that a workflow step's auto_start_agent on_enter
// action failed to launch a run for this task. Derived from the
// auto_start_failed metadata key at DTO conversion time; the orchestrator
// clears it when a session of the task next enters STARTING/RUNNING.
AutoStartFailed bool `json:"auto_start_failed,omitempty"`
IsEphemeral bool `json:"is_ephemeral"` // Ephemeral tasks are not shown in kanban, used for quick chat
ParentID string `json:"parent_id,omitempty"` // FK to parent task for subtasks
Autopilot bool `json:"autopilot"`
Identifier string `json:"identifier,omitempty"`
}
// TaskRepositoryInput for creating/updating task repositories
+14 -3
View File
@@ -316,16 +316,26 @@ export function renderTaskStatusIcon(
const showPermissionIcon = shouldUsePermissionTaskIcon(hasPendingPermission);
const needsMe = showQuestionIcon || showPermissionIcon;
const showInterrupted = !!task.interrupted;
const showAutoStartFailed = !!task.autoStartFailed;
const hasActivity =
task.foregroundActivity === "generating" || task.foregroundActivity === "background";
if (!showRunningSpinner && !needsMe && !hasActivity && !showInterrupted) {
if (!showRunningSpinner && !needsMe && !hasActivity && !showInterrupted && !showAutoStartFailed) {
return null;
}
// A "needs me" prompt (pending clarification / permission) must not be masked
// by the launch-spinner short-circuit — a mid-turn prompt can coincide with a
// coarse running state. Live foreground activity still wins, handled inside
// getTaskStateIcon.
if (showRunningSpinner && !needsMe && task.foregroundActivity !== "background") {
// getTaskStateIcon. A failed auto-start must not be masked either: startTask
// sets the task to SCHEDULING before the launch, so a launch failure before
// session creation leaves a session-less SCHEDULING/IN_PROGRESS task, which
// reads as showRunningSpinner=true — the exact shape the failure marker exists
// to surface.
if (
showRunningSpinner &&
!needsMe &&
!showAutoStartFailed &&
task.foregroundActivity !== "background"
) {
return <IconLoader2 className="h-4 w-4 text-blue-500 animate-spin" />;
}
return getTaskStateIcon(task.state, "h-4 w-4", {
@@ -333,6 +343,7 @@ export function renderTaskStatusIcon(
foregroundActivity: task.foregroundActivity,
hasPendingPermission,
interrupted: showInterrupted,
autoStartFailed: showAutoStartFailed,
});
}
@@ -9,6 +9,7 @@ import {
} from "@tabler/icons-react";
import { renderToStaticMarkup } from "react-dom/server";
import { renderSubagentCountChip, renderTaskStatusIcon } from "./kanban-card-content";
import { AutoStartFailedTaskIcon } from "@/lib/ui/state-icons";
import type { Task } from "./kanban-card";
function task(overrides: Partial<Task>): Task {
@@ -92,6 +93,60 @@ describe("renderTaskStatusIcon — waiting-for-input variants", () => {
});
});
describe("renderTaskStatusIcon — auto-start failed", () => {
it("shows the auto-start-failed triangle for a task whose on_enter launch failed", () => {
const node = renderTaskStatusIcon(
task({ state: "IN_PROGRESS", autoStartFailed: true }),
false,
false,
false,
);
expect(iconType(node)).toBe(AutoStartFailedTaskIcon);
});
// The real shape a failed kanban auto-start leaves behind: startTask sets the
// task to SCHEDULING before the launch, so a failure before session creation
// produces a session-less SCHEDULING/IN_PROGRESS task, which is exactly what
// shouldShowTaskRunningSpinner reads as "still launching" (showRunningSpinner
// true). The triangle must not be masked by the launch-spinner short-circuit
// the way needsMe already isn't.
it("shows the triangle over the launch spinner for a session-less SCHEDULING task", () => {
const node = renderTaskStatusIcon(
task({ state: "SCHEDULING", autoStartFailed: true }),
true,
false,
false,
);
expect(iconType(node)).toBe(AutoStartFailedTaskIcon);
expect(iconType(node)).not.toBe(IconLoader2);
});
it("shows the triangle over the launch spinner for a session-less IN_PROGRESS task", () => {
const node = renderTaskStatusIcon(
task({ state: "IN_PROGRESS", autoStartFailed: true }),
true,
false,
false,
);
expect(iconType(node)).toBe(AutoStartFailedTaskIcon);
expect(iconType(node)).not.toBe(IconLoader2);
});
it("keeps the terminal done check over a lingering auto-start-failed marker", () => {
const node = renderTaskStatusIcon(
task({ state: "COMPLETED", autoStartFailed: true }),
false,
false,
false,
);
expect(iconType(node)).toBe(IconCheck);
});
it("renders nothing for a resting task with the marker absent", () => {
expect(renderTaskStatusIcon(task({ state: "IN_PROGRESS" }), false, false, false)).toBeNull();
});
});
// active_subagent_count has been published end-to-end since the background-work
// liveness work, and reached the store with no component reading it — rendering
// it was an explicit non-goal of that spec. This is the follow-up.
+3
View File
@@ -69,6 +69,9 @@ export interface Task {
foregroundActivity?: ForegroundActivity | null;
/** True when the task's session was mid-turn when the backend died. */
interrupted?: boolean;
/** True when a workflow step's auto_start_agent on_enter action failed to
* launch a run for this task. */
autoStartFailed?: boolean;
/** Live subagents summed across this task's sessions; drives the count chip. */
activeSubagentCount?: number;
reviewStatus?: "pending" | "approved" | "changes_requested" | "rejected" | null;
@@ -1,5 +1,6 @@
import { cleanup, render } from "@testing-library/react";
import { afterEach, describe, expect, it, vi } from "vitest";
import { TooltipProvider } from "@kandev/ui/tooltip";
import { StateProvider } from "@/components/state-provider";
import type { Task } from "@/components/kanban-card";
import type { WorkflowStep } from "@/components/kanban-column";
@@ -70,6 +71,50 @@ describe("Graph2StepNode — task-level background-running affordance", () => {
});
});
describe("Graph2StepNode — auto-start-failed marker", () => {
function renderNodeWithTask(task: Task) {
return render(
<StateProvider>
<TooltipProvider>
<Graph2StepNode
step={STEP}
phase="current"
task={task}
hasPrev={false}
hasNext={false}
onMoveTask={() => undefined}
onPreviewTask={() => undefined}
/>
</TooltipProvider>
</StateProvider>,
);
}
it("shows the auto-start-failed triangle for a non-terminal task marked auto_start_failed", () => {
const task = {
id: "task-1",
title: "A task",
workflowStepId: "step-1",
state: "IN_PROGRESS",
autoStartFailed: true,
} as Task;
const { container } = renderNodeWithTask(task);
expect(container.querySelector('[data-testid="task-state-auto-start-failed"]')).not.toBeNull();
});
it("does not show the auto-start-failed triangle when the marker is absent", () => {
const task = {
id: "task-1",
title: "A task",
workflowStepId: "step-1",
state: "IN_PROGRESS",
autoStartFailed: false,
} as Task;
const { container } = renderNodeWithTask(task);
expect(container.querySelector('[data-testid="task-state-auto-start-failed"]')).toBeNull();
});
});
describe("Graph2StepNode — waiting-for-input variants", () => {
function renderWaitingNode(pendingAction: TaskPendingAction) {
const task = {
@@ -151,6 +151,7 @@ export function Graph2StepNode({
foregroundActivity: task.foregroundActivity,
hasPendingPermission: pendingInput.permission,
interrupted: task.interrupted,
autoStartFailed: task.autoStartFailed,
})}
</div>
<span className="text-[11px] font-medium text-foreground truncate">{step.title}</span>
+9
View File
@@ -499,6 +499,15 @@ export class ApiClient {
await this.request("PATCH", `/api/v1/tasks/${taskId}`, { title });
}
/** Replace a task's metadata via the same PATCH path a real orchestrator
* mutation would use, so the update travels over `task.updated` WS to any
* page that already has the task open, instead of only landing in the next
* HTTP snapshot. `UpdateTask` replaces metadata wholesale, so pass the full
* desired object. */
async updateTaskMetadata(taskId: string, metadata: Record<string, unknown>): Promise<void> {
await this.request("PATCH", `/api/v1/tasks/${taskId}`, { metadata });
}
async listAgents(): Promise<{ agents: Agent[]; total: number }> {
const response = await this.request<{ agents: Agent[]; total: number }>(
"GET",
@@ -0,0 +1,136 @@
import { test, expect } from "../../fixtures/test-base";
import { KanbanPage } from "../../pages/kanban-page";
import { watchWs } from "../../helpers/causal-waits";
// Auto-start-failed indicator: when a workflow step's `auto_start_agent`
// on_enter action cannot launch a run (for example an Office task moved into
// a step whose runtime context isn't ready), the orchestrator stamps the
// `auto_start_failed` metadata key and the kanban card shows a red alert icon
// instead of silently sitting there looking normal. Seeding `auto_start_failed`
// through the public task-metadata surface is the deterministic stand-in for
// a real failed launch; the icon itself must render from that marker alone.
//
// The fixture is left in a session-less IN_PROGRESS state on purpose: a real
// failed launch comes from startTask setting the task to SCHEDULING/IN_PROGRESS
// before session creation, then failing before a session ever attaches. That
// shape makes shouldShowTaskRunningSpinner report true, which previously
// short-circuited renderTaskStatusIcon straight to the launch spinner before
// the auto-start-failed marker ever got a chance to render.
test.describe("Kanban card — auto-start-failed icon", () => {
test("shows the red auto-start-failed icon only for marked, non-terminal tasks", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(90_000);
const failed = await apiClient.createTask(seedData.workspaceId, "Auto Start Failed Fixture", {
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
metadata: { auto_start_failed: "true" },
});
// Session-less IN_PROGRESS: the exact shape a launch failure before
// session creation leaves behind, and the one the running-spinner
// short-circuit previously masked the marker in.
await apiClient.updateTaskState(failed.id, "IN_PROGRESS");
const plain = await apiClient.createTask(seedData.workspaceId, "Plain Fixture", {
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
});
await apiClient.updateTaskState(plain.id, "REVIEW");
// A terminal marked task must keep its done affordance, never the red icon.
const terminal = await apiClient.createTask(seedData.workspaceId, "Terminal Fixture", {
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
metadata: { auto_start_failed: "true" },
});
await apiClient.updateTaskState(terminal.id, "COMPLETED");
const kanban = new KanbanPage(testPage);
await kanban.goto();
const failedCard = kanban.taskCardByTitle("Auto Start Failed Fixture");
await expect(failedCard).toBeVisible({ timeout: 20_000 });
await expect(failedCard.getByTestId("task-state-auto-start-failed")).toBeVisible({
timeout: 20_000,
});
const plainCard = kanban.taskCardByTitle("Plain Fixture");
await expect(plainCard).toBeVisible({ timeout: 20_000 });
await expect(plainCard.getByTestId("task-state-auto-start-failed")).toHaveCount(0);
const terminalCard = kanban.taskCardByTitle("Terminal Fixture");
await expect(terminalCard).toBeVisible({ timeout: 20_000 });
await expect(terminalCard.getByTestId("task-state-auto-start-failed")).toHaveCount(0);
// Reload: the marker must survive SSR hydration from the boot payload.
await testPage.reload();
await kanban.board.waitFor({ state: "visible" });
await expect(kanban.taskCardByTitle("Auto Start Failed Fixture")).toBeVisible({
timeout: 20_000,
});
await expect(
kanban
.taskCardByTitle("Auto Start Failed Fixture")
.getByTestId("task-state-auto-start-failed"),
).toBeVisible({ timeout: 20_000 });
await expect(
kanban.taskCardByTitle("Plain Fixture").getByTestId("task-state-auto-start-failed"),
).toHaveCount(0);
});
// Every other case in this file seeds `auto_start_failed` before the page
// ever loads, so it only ever exercises the HTTP snapshot / boot-payload
// producer. That left the WS producer (task.updated -> kanban.update store
// merge) unverified: it hand-builds its payload separately and had silently
// dropped the field, so a failure that happened while the board was already
// open never showed the badge until a reload. This test sets and clears the
// marker via a live PATCH while the board stays open, so only the WS path
// can make it pass.
test("shows and hides the auto-start-failed icon over a live WS update, without a reload", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(90_000);
const task = await apiClient.createTask(seedData.workspaceId, "Live WS Auto Start Fixture", {
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
});
await apiClient.updateTaskState(task.id, "IN_PROGRESS");
const kanban = new KanbanPage(testPage);
const ws = watchWs(testPage);
await kanban.goto();
const card = kanban.taskCardByTitle("Live WS Auto Start Fixture");
await expect(card).toBeVisible({ timeout: 20_000 });
await expect(card.getByTestId("task-state-auto-start-failed")).toHaveCount(0);
const markerSet = ws.waitForEvent("task.updated", {
where: (payload) => payload.task_id === task.id && payload.auto_start_failed === true,
});
await apiClient.updateTaskMetadata(task.id, { auto_start_failed: "true" });
await markerSet;
await expect(card.getByTestId("task-state-auto-start-failed")).toBeVisible({
timeout: 10_000,
});
const markerCleared = ws.waitForEvent("task.updated", {
where: (payload) => payload.task_id === task.id && payload.auto_start_failed === false,
});
await apiClient.updateTaskMetadata(task.id, {});
await markerCleared;
await expect(card.getByTestId("task-state-auto-start-failed")).toHaveCount(0);
});
});
@@ -8,6 +8,7 @@ type SnapshotTask = {
title: string;
position: number;
state: "IN_PROGRESS";
autoStartFailed?: boolean;
parentTaskId?: string;
statusSummary?: {
revision: number;
@@ -242,6 +243,58 @@ describe("useAllWorkflowSnapshots in-flight websocket tasks", () => {
});
});
describe("useAllWorkflowSnapshots auto-start marker races", () => {
it("keeps a newer live auto-start marker when an older snapshot finishes later", async () => {
resetState();
let resolveFetch: (value: unknown) => void = () => {};
mocks.fetchWorkflowSnapshot.mockReturnValueOnce(
new Promise((resolve) => {
resolveFetch = resolve;
}),
);
renderHook(() => useAllWorkflowSnapshots(WORKSPACE_ID));
await waitFor(() =>
expect(mocks.fetchWorkflowSnapshot).toHaveBeenCalledWith(WORKFLOW_ID, expect.anything()),
);
setLightweightSnapshot({
id: "task-with-live-auto-start-marker",
workflowStepId: STEP_ID,
title: "Live auto-start marker",
position: 0,
state: "IN_PROGRESS",
autoStartFailed: true,
});
resolveFetch({
steps: [{ id: STEP_ID, name: "Doing", color: null, position: 0 }],
tasks: [
{
id: "task-with-live-auto-start-marker",
workflow_step_id: STEP_ID,
title: "Live auto-start marker",
position: 0,
state: "IN_PROGRESS",
auto_start_failed: false,
},
],
});
await waitFor(() => expect(mocks.setWorkflowSnapshot).toHaveBeenCalled());
expect(mocks.setWorkflowSnapshot).toHaveBeenCalledWith(
WORKFLOW_ID,
expect.objectContaining({
tasks: [
expect.objectContaining({
id: "task-with-live-auto-start-marker",
autoStartFailed: true,
}),
],
}),
);
});
});
describe("useAllWorkflowSnapshots status invalidation races", () => {
it("keeps a live summary that advanced while an invalidating snapshot was in flight", async () => {
resetState();
@@ -37,6 +37,14 @@ function hasNewerLiveStatusSummary(
return existingRevision > (fetchStart.statusSummary?.revision ?? -1);
}
function hasNewerLiveAutoStartFailed(
existing: KanbanTask,
fetchStart: KanbanTask | undefined,
): boolean {
if (!fetchStart) return true;
return existing.autoStartFailed !== fetchStart.autoStartFailed;
}
function mergeFetchedTask(
mapped: KanbanTask,
existing: KanbanTask,
@@ -73,6 +81,11 @@ function mergeFetchedTask(
// Autopilot is immutable after creation. Keep the cached value when
// an older or partial snapshot does not include the field.
mapped.autopilot = mapped.autopilot ?? existing.autopilot;
// A task.updated event can set or clear this marker while the snapshot is
// in flight. Preserve that newer live value instead of rolling it back.
if (mapped.autoStartFailed === undefined || hasNewerLiveAutoStartFailed(existing, fetchStart)) {
mapped.autoStartFailed = existing.autoStartFailed;
}
return mapped;
}
+41 -1
View File
@@ -44,7 +44,16 @@ vi.mock("@/lib/api", () => ({
}));
vi.mock("@/lib/ssr/mapper", () => ({
snapshotToState: (snapshot: unknown) => ({ snapshot }),
snapshotToState: (snapshot: { tasks?: Array<{ id: string; auto_start_failed?: boolean }> }) => ({
kanban: {
workflowId: "wf-1",
steps: [],
tasks: (snapshot.tasks ?? []).map((task) => ({
id: task.id,
autoStartFailed: task.auto_start_failed,
})),
},
}),
}));
import { useWorkflowSnapshot } from "./use-workflow-snapshot";
@@ -175,3 +184,34 @@ describe("useWorkflowSnapshot — kanban.isLoading", () => {
expect(mockState.kanban.isLoading).toBe(false);
});
});
describe("useWorkflowSnapshot marker races", () => {
beforeEach(() => resetState());
it("keeps a newer live auto-start marker when an older snapshot finishes later", async () => {
let resolveFetch: (snapshot: { steps: unknown[]; tasks: unknown[] }) => void = () => {};
mockFetchWorkflowSnapshot.mockReturnValueOnce(
new Promise((resolve) => {
resolveFetch = resolve;
}),
);
renderHook(() => useWorkflowSnapshot("wf-1"));
await waitFor(() => expect(mockFetchWorkflowSnapshot).toHaveBeenCalled());
mockState.kanban.tasks = [{ id: "task-live-marker", autoStartFailed: true }];
resolveFetch({
steps: [],
tasks: [{ id: "task-live-marker", auto_start_failed: false }],
});
await waitFor(() => expect(mockHydrate).toHaveBeenCalled());
expect(mockHydrate).toHaveBeenCalledWith(
expect.objectContaining({
kanban: expect.objectContaining({
tasks: [expect.objectContaining({ id: "task-live-marker", autoStartFailed: true })],
}),
}),
);
});
});
+36 -1
View File
@@ -3,6 +3,31 @@ import { fetchWorkflowSnapshot } from "@/lib/api";
import { snapshotToState } from "@/lib/ssr/mapper";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import { isCurrentWorkspaceContext } from "@/lib/state/workspace-context";
import type { KanbanState } from "@/lib/state/slices/kanban/types";
type KanbanTask = KanbanState["tasks"][number];
function preserveLiveAutoStartFailed(
snapshotTasks: KanbanTask[],
fetchStartTasks: KanbanTask[],
currentTasks: KanbanTask[],
) {
const fetchStartByID = new Map(fetchStartTasks.map((task) => [task.id, task]));
const currentByID = new Map(currentTasks.map((task) => [task.id, task]));
return snapshotTasks.map((task) => {
const current = currentByID.get(task.id);
if (!current) return task;
const fetchStart = fetchStartByID.get(task.id);
if (
task.autoStartFailed === undefined ||
fetchStart === undefined ||
current.autoStartFailed !== fetchStart.autoStartFailed
) {
return { ...task, autoStartFailed: current.autoStartFailed };
}
return task;
});
}
export function useWorkflowSnapshot(workflowId: string | null) {
const store = useAppStoreApi();
@@ -36,7 +61,17 @@ export function useWorkflowSnapshot(workflowId: string | null) {
) {
return;
}
store.getState().hydrate(snapshotToState(snapshot));
const nextState = snapshotToState(snapshot);
if (nextState.kanban) {
const currentTasks = store.getState().kanban.tasks;
const fetchedAtStartTasks = existing.tasks;
nextState.kanban.tasks = preserveLiveAutoStartFailed(
nextState.kanban.tasks,
fetchedAtStartTasks,
currentTasks,
);
}
store.getState().hydrate(nextState);
})
.catch((error) => {
// Suppress superseded-fetch noise; retry happens on WS reconnect.
+4
View File
@@ -56,6 +56,9 @@ export type TaskLike = {
task_pending_action?: TaskPendingAction | null;
/** True when the task's session was mid-turn when the backend died. */
interrupted?: boolean;
/** True when a workflow step's auto_start_agent on_enter action failed to
* launch a run for this task. */
auto_start_failed?: boolean;
foreground_activity?: ForegroundActivity | null;
active_subagent_count?: number;
session_count?: number | null;
@@ -191,6 +194,7 @@ export function toKanbanTask(source: TaskLike): KanbanTask {
primarySessionPendingAction: pickPendingAction(source.primary_session_pending_action),
taskPendingAction: pickPendingAction(source.task_pending_action),
interrupted: source.interrupted,
autoStartFailed: source.auto_start_failed,
foregroundActivity: pickForegroundActivity(source.foreground_activity),
activeSubagentCount: source.active_subagent_count ?? undefined,
sessionCount: source.session_count ?? undefined,
+17
View File
@@ -47,6 +47,7 @@ function snapshotWithPendingAction(action: unknown): WorkflowSnapshot {
};
}
// eslint-disable-next-line max-lines-per-function -- test describe block, splitting hurts readability
describe("snapshotToState", () => {
it("keeps known primary session pending action values", () => {
const state = snapshotToState(snapshotWithPendingAction("permission"));
@@ -85,6 +86,22 @@ describe("snapshotToState", () => {
expect(state.kanban?.tasks[0]?.activeSubagentCount).toBe(expected);
});
it.each([
[true, true],
[false, false],
[undefined, false],
])(
"maps auto_start_failed %s so a page reload does not hide or resurrect the badge",
(wireValue, expected) => {
const snapshot = snapshotWithPendingAction(undefined);
snapshot.tasks[0].auto_start_failed = wireValue;
const state = snapshotToState(snapshot);
expect(state.kanban?.tasks[0]?.autoStartFailed).toBe(expected);
},
);
it("hydrates the task status summary into the initial kanban state", () => {
const snapshot = snapshotWithPendingAction(undefined);
snapshot.tasks[0].status_summary = {
+7
View File
@@ -10,6 +10,12 @@ import {
type KanbanTask = KanbanState["tasks"][number];
// Split out so the snapshot->task mapper (already at the complexity limit
// from its long list of `??` fallbacks) doesn't need to absorb one more.
function resolveAutoStartFailed(task: WorkflowSnapshot["tasks"][number]): boolean {
return task.auto_start_failed ?? false;
}
export function snapshotToState(snapshot: WorkflowSnapshot): Partial<AppState> {
// Handle empty snapshot (ephemeral tasks have no workflow)
if (!snapshot.workflow) {
@@ -61,6 +67,7 @@ export function snapshotToState(snapshot: WorkflowSnapshot): Partial<AppState> {
primarySessionPendingAction: pickPendingAction(task.primary_session_pending_action),
taskPendingAction: pickPendingAction(task.task_pending_action),
foregroundActivity: task.foreground_activity ?? undefined,
autoStartFailed: resolveAutoStartFailed(task),
activeSubagentCount: task.active_subagent_count ?? undefined,
sessionCount: task.session_count ?? undefined,
reviewStatus: task.review_status ?? undefined,
@@ -106,6 +106,9 @@ export type KanbanState = {
foregroundActivity?: ForegroundActivity | null;
/** True when the task's session was mid-turn when the backend died. */
interrupted?: boolean;
/** True when a workflow step's auto_start_agent on_enter action failed to
* launch a run for this task. */
autoStartFailed?: boolean;
/** Live subagents across this task's sessions; drives the board count chip. */
activeSubagentCount?: number;
sessionCount?: number | null;
+3
View File
@@ -378,6 +378,9 @@ export type Task = ActiveSubagentCountFields & {
/** True when the task's session was mid-turn when the backend died and has
* not been resumed since (startup reconciliation marker). */
interrupted?: boolean;
/** True when a workflow step's auto_start_agent on_enter action failed to
* launch a run for this task. */
auto_start_failed?: boolean;
/**
* Task-level MOST-ACTIVE-WINS activity across sessions. "generating" wins,
* then "background"; null/absent means none is known. The count is the
+94 -41
View File
@@ -167,6 +167,57 @@ describe("getTaskStateIcon — task-level activity tri-state", () => {
expect(iconType(getTaskStateIcon("COMPLETED", undefined))).toBe(IconCheck);
});
it("safe fallback: an in-progress task with a MISSING aggregate reads not-done, never a check", () => {
// safe default: a task whose turn is still
// open (coarse IN_PROGRESS) but whose task-level aggregate is unknown — e.g.
// the aggregate never reached this client, or the in-memory tracker reset on
// a backend restart — must fall back to the working spinner, never the done
// check. The coarse IN_PROGRESS reading is itself not-done, so a missing
// aggregate can only ever soften to working, never harden to done.
const missingUndefined = getTaskStateIcon("IN_PROGRESS", undefined);
const missingNull = getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: null });
expect(iconType(missingUndefined)).toBe(IconLoader2);
expect(iconType(missingUndefined)).not.toBe(IconCheck);
expect(iconType(missingNull)).toBe(IconLoader2);
expect(iconType(missingNull)).not.toBe(IconCheck);
});
it("distinguishes background from BOTH generating and done by icon SHAPE, not hue alone", () => {
// Icon TYPE (glyph) differs for all three, so the reading survives a
// grayscale/desaturated scan for color-vision-deficient operators
// The affordance remains distinguishable without color.
const generating = iconType(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "generating" }),
);
const background = iconType(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "background" }),
);
const done = iconType(getTaskStateIcon("COMPLETED", undefined, { foregroundActivity: null }));
expect(background).not.toBe(generating);
expect(background).not.toBe(done);
expect(generating).not.toBe(done);
});
it("also separates background from generating and done by HUE on the compact surfaces", () => {
// The dense board/list/graph surfaces get an extra hue separation on top of the
// shape difference so background reads apart from generating at a glance — its
// own violet, distinct from generating's blue and done's green.
const generating = iconClassName(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "generating" }),
);
const background = iconClassName(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "background" }),
);
const done = iconClassName(
getTaskStateIcon("COMPLETED", undefined, { foregroundActivity: null }),
);
expect(background).toContain("text-violet-500");
expect(background).not.toBe(generating);
expect(background).not.toBe(done);
});
});
describe("getTaskStateIcon — status markers", () => {
it("renders the accessible interrupted icon with the tooltip label", () => {
const { container } = render(
<TooltipProvider>
@@ -218,53 +269,55 @@ describe("getTaskStateIcon — task-level activity tri-state", () => {
).toBe(IconShieldQuestion);
});
it("safe fallback: an in-progress task with a MISSING aggregate reads not-done, never a check", () => {
// safe default: a task whose turn is still
// open (coarse IN_PROGRESS) but whose task-level aggregate is unknown — e.g.
// the aggregate never reached this client, or the in-memory tracker reset on
// a backend restart — must fall back to the working spinner, never the done
// check. The coarse IN_PROGRESS reading is itself not-done, so a missing
// aggregate can only ever soften to working, never harden to done.
const missingUndefined = getTaskStateIcon("IN_PROGRESS", undefined);
const missingNull = getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: null });
expect(iconType(missingUndefined)).toBe(IconLoader2);
expect(iconType(missingUndefined)).not.toBe(IconCheck);
expect(iconType(missingNull)).toBe(IconLoader2);
expect(iconType(missingNull)).not.toBe(IconCheck);
it("renders the accessible auto-start-failed icon with the tooltip label", () => {
const { container } = render(
<TooltipProvider>
{getTaskStateIcon("REVIEW", undefined, { autoStartFailed: true })}
</TooltipProvider>,
);
const icon = container.querySelector('[data-testid="task-state-auto-start-failed"]');
expect(icon).not.toBeNull();
expect(icon?.className).toContain("text-red-500");
expect(container.querySelector('[aria-label="Auto-start failed"]')).not.toBeNull();
// The icon itself is decorative; the label lives on the trigger.
expect(icon?.getAttribute("aria-hidden")).toBe("true");
});
it("distinguishes background from BOTH generating and done by icon SHAPE, not hue alone", () => {
// Icon TYPE (glyph) differs for all three, so the reading survives a
// grayscale/desaturated scan for color-vision-deficient operators
// The affordance remains distinguishable without color.
const generating = iconType(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "generating" }),
it("keeps terminal state icons over a lingering auto-start-failed marker", () => {
expect(iconType(getTaskStateIcon("COMPLETED", undefined, { autoStartFailed: true }))).toBe(
IconCheck,
);
const background = iconType(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "background" }),
expect(iconType(getTaskStateIcon("FAILED", undefined, { autoStartFailed: true }))).toBe(IconX);
expect(iconType(getTaskStateIcon("CANCELLED", undefined, { autoStartFailed: true }))).toBe(
IconX,
);
const done = iconType(getTaskStateIcon("COMPLETED", undefined, { foregroundActivity: null }));
expect(background).not.toBe(generating);
expect(background).not.toBe(done);
expect(generating).not.toBe(done);
});
it("also separates background from generating and done by HUE on the compact surfaces", () => {
// The dense board/list/graph surfaces get an extra hue separation on top of the
// shape difference so background reads apart from generating at a glance — its
// own violet, distinct from generating's blue and done's green.
const generating = iconClassName(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "generating" }),
);
const background = iconClassName(
getTaskStateIcon("IN_PROGRESS", undefined, { foregroundActivity: "background" }),
);
const done = iconClassName(
getTaskStateIcon("COMPLETED", undefined, { foregroundActivity: null }),
);
expect(background).toContain("text-violet-500");
expect(background).not.toBe(generating);
expect(background).not.toBe(done);
it("keeps active and pending affordances over the auto-start-failed marker", () => {
expect(
iconType(
getTaskStateIcon("REVIEW", undefined, {
foregroundActivity: "generating",
autoStartFailed: true,
}),
),
).toBe(IconLoader2);
expect(
iconType(
getTaskStateIcon("REVIEW", undefined, {
hasPendingClarification: true,
autoStartFailed: true,
}),
),
).toBe(IconMessageQuestion);
expect(
iconType(
getTaskStateIcon("REVIEW", undefined, {
hasPendingPermission: true,
autoStartFailed: true,
}),
),
).toBe(IconShieldQuestion);
});
});
+65 -8
View File
@@ -117,6 +117,16 @@ const TASK_INTERRUPTED_ICON: IconConfig = {
className: STYLE_ERROR,
};
// The task-level auto-start-failed affordance: a workflow step's
// auto_start_agent on_enter action ran but could not launch a run (kanban
// StartTask error, or an Office task with no queue adapter wired / no
// resolvable agent). A triangle — same error hue as the interrupted circle,
// but a distinct shape so the two failure causes never read as one marker.
const TASK_AUTO_START_FAILED_ICON: IconConfig = {
Icon: IconAlertTriangle,
className: STYLE_ERROR,
};
const DEFAULT_TASK_ICON: IconConfig = {
Icon: IconAlertCircle,
className: STYLE_MUTED,
@@ -202,6 +212,34 @@ export function InterruptedTaskIcon({ className }: { className?: string }) {
);
}
/**
* Shared red alert-triangle affordance for a task whose auto_start_agent
* on_enter action failed to launch a run. Carries the accessible "Auto-start
* failed" label and tooltip, so every surface that renders this state
* presents it consistently.
*/
export function AutoStartFailedTaskIcon({ className }: { className?: string }) {
const { t } = useTranslation();
return (
<Tooltip>
<TooltipTrigger asChild>
<span
aria-label={t("common:autoStartFailed")}
tabIndex={0}
className="flex shrink-0 rounded-sm focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-red-500 focus-visible:ring-offset-1"
>
<IconAlertTriangle
aria-hidden="true"
data-testid="task-state-auto-start-failed"
className={cn("text-red-500", className)}
/>
</span>
</TooltipTrigger>
<TooltipContent side="right">{t("common:autoStartFailed")}</TooltipContent>
</Tooltip>
);
}
/**
* Returns true when the kanban card should show the spinning loader. The task
* workflow state and the primary session's runtime state are decoupled — the
@@ -249,14 +287,33 @@ type TaskStateIconOptions = {
hasPendingPermission?: boolean;
/** True when the task's session was mid-turn when the backend died. */
interrupted?: boolean;
/** True when a workflow step's auto_start_agent action failed to launch a run. */
autoStartFailed?: boolean;
};
// Interrupted (startup reconciliation marker) and auto-start-failed
// (on_enter action that never launched a run) both replace the idle/done
// affordances but never override terminal states, which keep their own
// icons (done check, failure X, cancel pause). Interrupted takes precedence
// when both happen to be set.
function getMarkerIconOverride(
state: TaskState | undefined,
interrupted: boolean,
autoStartFailed: boolean,
): IconConfig | null {
if (TERMINAL_TASK_STATES.has(state)) return null;
if (interrupted) return TASK_INTERRUPTED_ICON;
if (autoStartFailed) return TASK_AUTO_START_FAILED_ICON;
return null;
}
function getTaskStateIconConfig(state?: TaskState, options: TaskStateIconOptions = {}): IconConfig {
const {
hasPendingClarification = false,
foregroundActivity,
hasPendingPermission = false,
interrupted = false,
autoStartFailed = false,
} = options;
if (shouldUsePermissionTaskIcon(hasPendingPermission)) {
return PENDING_PERMISSION_ICON;
@@ -270,12 +327,8 @@ function getTaskStateIconConfig(state?: TaskState, options: TaskStateIconOptions
if (foregroundActivity === "generating") return TASK_GENERATING_ICON;
if (foregroundActivity === "background") return TASK_BACKGROUND_ICON;
if (isWaitingForInputState(state)) return TASK_STATE_ICONS.WAITING_FOR_INPUT;
// Interrupted (startup reconciliation marker): replaces the idle/done
// affordances but never overrides terminal states, which keep their own
// icons (done check, failure X, cancel pause).
if (interrupted && !TERMINAL_TASK_STATES.has(state)) {
return TASK_INTERRUPTED_ICON;
}
const markerOverride = getMarkerIconOverride(state, interrupted, autoStartFailed);
if (markerOverride) return markerOverride;
if (!state) return DEFAULT_TASK_ICON;
return TASK_STATE_ICONS[state] ?? DEFAULT_TASK_ICON;
}
@@ -286,11 +339,15 @@ export function getTaskStateIcon(
options: TaskStateIconOptions = {},
) {
const config = getTaskStateIconConfig(state, options);
// The interrupted affordance carries its own tooltip and accessible label,
// so it must render through the shared component rather than a bare icon.
// The interrupted and auto-start-failed affordances carry their own
// tooltip and accessible label, so they must render through their shared
// component rather than a bare icon.
if (config === TASK_INTERRUPTED_ICON) {
return <InterruptedTaskIcon className={cn("h-4 w-4", className)} />;
}
if (config === TASK_AUTO_START_FAILED_ICON) {
return <AutoStartFailedTaskIcon className={cn("h-4 w-4", className)} />;
}
return <config.Icon className={cn("h-4 w-4", config.className, className)} />;
}
@@ -0,0 +1,159 @@
import { describe, it, expect } from "vitest";
import type { StoreApi } from "zustand";
import type { AppState } from "@/lib/state/store";
import { registerKanbanHandlers } from "./kanban";
// kanban.test.ts is already at the file-length limit, so this regression gets
// its own file rather than growing that one further.
//
// Review round 2's BLOCKER 2: autoStartFailed had only ever been wired into
// toKanbanTask. The kanban.update handler's two producer sites (the main
// kanban.tasks rebuild and the kanbanMulti snapshot merge) still dropped the
// field, so a badge set by a live task.updated -> kanban.update replay
// disappeared the next time the board resynced its step/task list.
const WORKFLOW_ID = "wf1";
const TASK_ID = "t1";
const STEP_ID = "s1";
const TASK_TITLE = "T1";
function makeStore(initial: Partial<AppState> = {}) {
let state = {
kanban: { workflowId: null, steps: [], tasks: [] },
kanbanMulti: { snapshots: {}, isLoading: false },
...initial,
} as unknown as AppState;
return {
getState: () => state,
setState: (updater: AppState | ((s: AppState) => AppState)) => {
state =
typeof updater === "function" ? (updater as (s: AppState) => AppState)(state) : updater;
},
subscribe: () => () => {},
destroy: () => {},
getInitialState: () => state,
} as unknown as StoreApi<AppState>;
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
function makeUpdateMessage(workflowId: string, tasks: unknown[], steps: unknown[] = []): any {
return {
id: "msg-1",
type: "notification",
action: "kanban.update",
payload: { workflowId, tasks, steps },
};
}
describe("kanban.update handler — autoStartFailed preservation", () => {
it("preserves autoStartFailed from existing tasks", () => {
const store = makeStore({
kanban: {
workflowId: WORKFLOW_ID,
steps: [],
tasks: [
{
id: TASK_ID,
workflowId: WORKFLOW_ID,
workflowStepId: STEP_ID,
title: TASK_TITLE,
position: 0,
autoStartFailed: true,
},
],
},
kanbanMulti: {
isLoading: false,
snapshots: {
[WORKFLOW_ID]: {
workflowId: WORKFLOW_ID,
workflowName: "WF1",
steps: [],
tasks: [
{
id: TASK_ID,
workflowId: WORKFLOW_ID,
workflowStepId: STEP_ID,
title: TASK_TITLE,
position: 0,
autoStartFailed: true,
},
],
},
},
},
} as Partial<AppState>);
const handler = registerKanbanHandlers(store)["kanban.update"]!;
handler(
makeUpdateMessage(WORKFLOW_ID, [
{ id: TASK_ID, workflowStepId: STEP_ID, title: TASK_TITLE, position: 0 },
]),
);
const task = store.getState().kanban.tasks.find((t) => t.id === TASK_ID);
expect(task?.autoStartFailed).toBe(true);
const snapshotTask = store
.getState()
.kanbanMulti.snapshots[WORKFLOW_ID]?.tasks.find((t) => t.id === TASK_ID);
expect(snapshotTask?.autoStartFailed).toBe(true);
});
it("does not restore stale snapshot value when a kanban.update run clears autoStartFailed", () => {
// A snapshot's own cached autoStartFailed (true) must not resurrect the
// badge once the primary kanban list has moved on without it (false):
// the kanbanMulti merge only falls back to the snapshot's value when the
// primary lookup is `undefined` (task absent), not when it resolves to an
// explicit false.
const store = makeStore({
kanban: {
workflowId: WORKFLOW_ID,
steps: [],
tasks: [
{
id: TASK_ID,
workflowId: WORKFLOW_ID,
workflowStepId: STEP_ID,
title: TASK_TITLE,
position: 0,
autoStartFailed: false,
},
],
},
kanbanMulti: {
isLoading: false,
snapshots: {
[WORKFLOW_ID]: {
workflowId: WORKFLOW_ID,
workflowName: "WF1",
steps: [],
tasks: [
{
id: TASK_ID,
workflowId: WORKFLOW_ID,
workflowStepId: STEP_ID,
title: TASK_TITLE,
position: 0,
autoStartFailed: true,
},
],
},
},
},
} as Partial<AppState>);
const handler = registerKanbanHandlers(store)["kanban.update"]!;
handler(
makeUpdateMessage(WORKFLOW_ID, [
{ id: TASK_ID, workflowStepId: STEP_ID, title: TASK_TITLE, position: 0 },
]),
);
const snapshotTask = store
.getState()
.kanbanMulti.snapshots[WORKFLOW_ID]?.tasks.find((t) => t.id === TASK_ID);
expect(snapshotTask?.autoStartFailed).toBe(false);
});
});
+3
View File
@@ -133,6 +133,7 @@ export function registerKanbanHandlers(store: StoreApi<AppState>): WsHandlers {
primarySessionPendingAction: existing?.primarySessionPendingAction,
taskPendingAction: existing?.taskPendingAction,
interrupted: existing?.interrupted,
autoStartFailed: existing?.autoStartFailed,
foregroundActivity: existing?.foregroundActivity,
...queueFields(task, existing),
...dependencyFields(task, existing),
@@ -177,6 +178,8 @@ export function registerKanbanHandlers(store: StoreApi<AppState>): WsHandlers {
? fallback?.foregroundActivity
: t.foregroundActivity,
interrupted: t.interrupted === undefined ? fallback?.interrupted : t.interrupted,
autoStartFailed:
t.autoStartFailed === undefined ? fallback?.autoStartFailed : t.autoStartFailed,
};
});
return {
+4
View File
@@ -130,6 +130,10 @@ function mergeTaskUpdate(
payloadKey: "interrupted",
taskField: "interrupted",
});
preserveOmittedField(existing, merged, payload, nextTask, {
payloadKey: "auto_start_failed",
taskField: "autoStartFailed",
});
if (
!hasPayloadField(payload, "active_subagent_count") &&
nextTask.activeSubagentCount === undefined
+1
View File
@@ -21,6 +21,7 @@
"autoApproveAllPermissions": "Auto-approve all permissions",
"automation": "Automation",
"automations": "Automations",
"autoStartFailed": "Auto-start failed",
"availableTemplates": "Available templates",
"back": "Back",
"backToTaskOverview": "Back to task overview",
+1
View File
@@ -21,6 +21,7 @@
"autoApproveAllPermissions": "Àũţō-àƥƥŕōvē àĺĺ ƥēŕḿĩśśĩōńś",
"automation": "Àũţōḿàţĩōń",
"automations": "Àũţōḿàţĩōńś",
"autoStartFailed": "Àũţō-śţàŕţ ƒàĩĺēď",
"availableTemplates": "Àvàĩĺàƀĺē ţēḿƥĺàţēś",
"back": "Ɓàćķ",
"backToTaskOverview": "Ɓàćķ ţō ţàśķ ōvēŕvĩēŵ",
+1
View File
@@ -20,6 +20,7 @@
"apply": "Aplicar",
"autoApproveAllPermissions": "Aprovar automaticamente todas as permissões",
"automations": "Automatizações",
"autoStartFailed": "Início automático falhou",
"availableTemplates": "Modelos disponíveis",
"back": "Voltar",
"backToTaskOverview": "Voltar à visão geral da tarefa",
+1
View File
@@ -21,6 +21,7 @@
"autoApproveAllPermissions": "自动批准所有权限",
"automation": "自动化",
"automations": "自动化",
"autoStartFailed": "自动启动失败",
"availableTemplates": "可用模板",
"back": "返回",
"backToTaskOverview": "返回任务概览",
+1
View File
@@ -21,6 +21,7 @@
"autoApproveAllPermissions": "自動批准所有權限",
"automation": "自動化",
"automations": "自動化",
"autoStartFailed": "自動啟動失敗",
"availableTemplates": "可用模板",
"back": "返回",
"backToTaskOverview": "返回任務概覽",
+1
View File
@@ -21,6 +21,7 @@
"autoApproveAllPermissions": "自動批准所有權限",
"automation": "自動化",
"automations": "自動化",
"autoStartFailed": "自動啟動失敗",
"availableTemplates": "可用模板",
"back": "返回",
"backToTaskOverview": "返回任務概覽",