diff --git a/apps/backend/internal/backendapp/adapters_office.go b/apps/backend/internal/backendapp/adapters_office.go index 4b9921e510..b07b6c09a0 100644 --- a/apps/backend/internal/backendapp/adapters_office.go +++ b/apps/backend/internal/backendapp/adapters_office.go @@ -186,12 +186,27 @@ func (a *taskCreatorAdapter) createOfficeTask( } // CreateOfficeTaskInWorkflow creates an office task pinned to a specific -// workflow id, bypassing the workspace's default office_workflow_id. Used -// for the standing coordination task that lives on the dedicated -// Coordination workflow. +// workflow id, bypassing the workspace's default office_workflow_id. Its +// only production caller (as of WO-36) is the routines dispatcher, which +// pins a materialized heavy-routine run to the dedicated Routine workflow. func (a *taskCreatorAdapter) CreateOfficeTaskInWorkflow( ctx context.Context, workspaceID, projectID, assigneeAgentID, workflowID, title, description string, ) (string, error) { + metadata := map[string]interface{}{ + // The Routine workflow's start step carries on_enter: + // auto_start_agent, but a materialized run lands directly on that + // step rather than transitioning into it, so nothing would + // otherwise evaluate on_enter for it. This opts the task into + // handleTaskCreated's create-time on_enter evaluation. + models.MetaKeyAutoStartOnCreate: true, + } + if assigneeAgentID != "" { + // The Routine workflow's start step pins no agent (routine.yml), so + // the kanban auto-start path's fallback read of + // task.Metadata[MetaKeyAgentProfileID] is what lets a materialized + // heavy-routine task actually launch with the routine's assignee. + metadata[models.MetaKeyAgentProfileID] = assigneeAgentID + } result, err := a.taskSvc.CreateTask(ctx, &taskservice.CreateTaskRequest{ //nolint:exhaustruct WorkspaceID: workspaceID, WorkflowID: workflowID, @@ -199,6 +214,7 @@ func (a *taskCreatorAdapter) CreateOfficeTaskInWorkflow( Description: description, ProjectID: projectID, AssigneeAgentProfileID: assigneeAgentID, + Metadata: metadata, Origin: models.TaskOriginOnboarding, }) if err != nil { diff --git a/apps/backend/internal/backendapp/adapters_office_test.go b/apps/backend/internal/backendapp/adapters_office_test.go index 9da4269d9c..63b8001599 100644 --- a/apps/backend/internal/backendapp/adapters_office_test.go +++ b/apps/backend/internal/backendapp/adapters_office_test.go @@ -185,6 +185,64 @@ func TestCreateWorkspaceKanbanBootstrapCreatesUsableSteps(t *testing.T) { } } +// TestCreateOfficeTaskInWorkflowCarriesAssigneeIntoLaunchMetadata covers +// WO-36's routine-launch-inputs gap: a materialized heavy routine run lands +// on the Routine workflow's start step, which pins no agent (routine.yml). +// The kanban auto-start path's only fallback for the agent to launch with is +// task.Metadata[MetaKeyAgentProfileID] — AssigneeAgentProfileID alone (a +// separate DB column) never reaches it. Without this, the routine's task +// gets created but the orchestrator has no agent profile to start. +// +// It also covers WO-36 round 2: handleTaskCreated only evaluates a step's +// on_enter for a freshly created task when the task carries the positive +// MetaKeyAutoStartOnCreate opt-in (see event_handlers_workflow.go). This +// method is that opt-in's only production setter, so both cases here — +// with and without an assignee — must always stamp it true. +func TestCreateOfficeTaskInWorkflowCarriesAssigneeIntoLaunchMetadata(t *testing.T) { + adapter, taskSvc := newOfficeTaskAdapterHarness(t) + ctx := context.Background() + + workflows, err := taskSvc.ListWorkflows(ctx, "ws-1", true) + if err != nil || len(workflows) == 0 { + t.Fatalf("ListWorkflows: %v (len=%d)", err, len(workflows)) + } + workflowID := workflows[0].ID + + taskID, err := adapter.CreateOfficeTaskInWorkflow( + ctx, "ws-1", "", "routine-assignee", workflowID, "Routine run", "Materialized run", + ) + if err != nil { + t.Fatalf("CreateOfficeTaskInWorkflow: %v", err) + } + task, err := taskSvc.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got, _ := task.Metadata[models.MetaKeyAgentProfileID].(string); got != "routine-assignee" { + t.Errorf("task.Metadata[MetaKeyAgentProfileID] = %q, want routine-assignee", got) + } + if !models.HasAutoStartOnCreateIntent(task.Metadata) { + t.Errorf("task.Metadata[MetaKeyAutoStartOnCreate] not set, want true") + } + + noAssigneeTaskID, err := adapter.CreateOfficeTaskInWorkflow( + ctx, "ws-1", "", "", workflowID, "Unassigned run", "Materialized run", + ) + if err != nil { + t.Fatalf("CreateOfficeTaskInWorkflow (no assignee): %v", err) + } + noAssigneeTask, err := taskSvc.GetTask(ctx, noAssigneeTaskID) + if err != nil { + t.Fatalf("GetTask (no assignee): %v", err) + } + if _, ok := noAssigneeTask.Metadata[models.MetaKeyAgentProfileID]; ok { + t.Errorf("task.Metadata[MetaKeyAgentProfileID] set for an unassigned routine, want absent") + } + if !models.HasAutoStartOnCreateIntent(noAssigneeTask.Metadata) { + t.Errorf("task.Metadata[MetaKeyAutoStartOnCreate] not set for an unassigned routine, want true") + } +} + func newOfficeTaskAdapterHarness(t *testing.T) (*taskCreatorAdapter, *taskservice.Service) { t.Helper() dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "office-adapter.db")) diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 5103a5f1d0..3f04ec2075 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -49,6 +49,8 @@ type mockStepGetter struct { workflowMetaErr error // optional error from GetWorkflowMeta workflowMetaDelay time.Duration // optional sleep before returning meta workflowMetaMu sync.Mutex // guards workflowMetaCalls for concurrent tests + getStepCalls int // GetStep invocations, guarded by getStepMu + getStepMu sync.Mutex } func newMockStepGetter() *mockStepGetter { @@ -59,12 +61,26 @@ func newMockStepGetter() *mockStepGetter { } func (m *mockStepGetter) GetStep(_ context.Context, stepID string) (*wfmodels.WorkflowStep, error) { + m.getStepMu.Lock() + m.getStepCalls++ + m.getStepMu.Unlock() if s, ok := m.steps[stepID]; ok { return s, nil } return nil, nil } +// GetStepCalls reports how many times GetStep has been invoked. autoStartTaskForStep +// calls GetStep synchronously, before any launch work is handed off to a detached +// goroutine (see autoStartTaskForLoadedStep) — so a zero count is a race-free way for +// a test to prove autoStartTaskForStep was never entered at all, without waiting on +// or racing against async launch/DB-teardown timing. +func (m *mockStepGetter) GetStepCalls() int { + m.getStepMu.Lock() + defer m.getStepMu.Unlock() + return m.getStepCalls +} + func (m *mockStepGetter) GetNextStepByPosition(_ context.Context, workflowID string, currentPosition int) (*wfmodels.WorkflowStep, error) { var best *wfmodels.WorkflowStep for _, s := range m.steps { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index a22c8ba933..c1f6b0dc53 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -794,6 +794,63 @@ func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.Task s.autoStartTaskForLoadedStep(ctx, task, targetStep, "task.queue_promoted", true, data.StepTransitionID) } +// handleTaskCreated lets a task that is created directly onto a step whose +// on_enter carries auto_start_agent actually launch. Every other trigger +// that reaches autoStartTaskForStep represents a task ENTERING a step via a +// transition (task.moved, task.queue_promoted, dependency resolution); a +// freshly created task enters its start step too, but nothing evaluated +// on_enter for it before this handler existed. +// +// The guard is a POSITIVE opt-in (models.HasAutoStartOnCreateIntent), not an +// absence check. An earlier version launched unless the task carried +// MetaKeyDeferredLaunch, treating every other absence as "no launch opinion, +// please auto-start" — but REST/MCP/WS creates without start_agent or +// prepare_session never set that key either, so a task explicitly created +// without a launch request got auto-started anyway. Requiring an explicit +// MetaKeyAutoStartOnCreate marker means only a producer that actually wants +// create-time on_enter evaluation gets it (today: CreateOfficeTaskInWorkflow, +// for materialized heavy-routine runs); every other producer's silence is +// left exactly as before. +// +// Office tasks (task.IsFromOffice) are excluded even when opted in: the +// office subscriber's own task.created handler already queues a run for +// every office task using a different idempotency key than +// autoStartOfficeTaskForLoadedStep's, so both firing would double-queue the +// same task. No current opt-in producer creates office tasks — heavy +// routines are never IsFromOffice — but the exclusion is explicit rather +// than incidental on that non-overlap. +// +// No MetaKeyDeferredLaunch check is needed here: repository.CreateTask +// unconditionally clears that key (models.DropWIPDeferredLaunch) for any +// task admitted directly onto a step, which is every task this handler +// fires for — a fresh task lands here precisely because it wasn't queued. +// A task that IS queued (QueuedForStepID != "") keeps the key, but +// autoStartTaskForStep's own leading check already returns before doing +// anything with it. Either way the key can never change this handler's +// outcome, so checking it here would be dead code. +// +// The opt-in itself is claimed via claimTaskEventMetadata before launching, +// the same one-shot-token pattern used for MetaKeyQueuePromotionPending: a +// duplicate task.created delivery for the same task would otherwise see the +// key still present and launch a second time. Only the delivery that wins +// the atomic RemoveTaskMetadataKey proceeds. +func (s *Service) handleTaskCreated(ctx context.Context, data watcher.TaskEventData) { + task, err := s.repo.GetTask(ctx, data.TaskID) + if err != nil || task == nil { + if err != nil { + s.logger.Warn("task.created: failed to load task", zap.String("task_id", data.TaskID), zap.Error(err)) + } + return + } + if task.IsFromOffice || !models.HasAutoStartOnCreateIntent(task.Metadata) { + return + } + if !s.claimTaskEventMetadata(ctx, task, models.MetaKeyAutoStartOnCreate) { + return + } + s.autoStartTaskForStep(ctx, task.ID, task.WorkflowStepID, events.TaskCreated, data.StepTransitionID) +} + func (s *Service) claimTaskEventMetadata(ctx context.Context, task *models.Task, key string) bool { if task == nil || task.Metadata == nil { return false diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_created_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_created_test.go new file mode 100644 index 0000000000..e02bda726a --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_created_test.go @@ -0,0 +1,326 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestHandleTaskCreated covers WO-36: a task created directly onto a step +// whose on_enter carries auto_start_agent (e.g. a materialized heavy +// routine run landing on the Routine workflow's start step) never got its +// on_enter evaluated, because every other autoStartTaskForStep caller +// represents a transition INTO a step (task.moved, task.queue_promoted, +// dependency resolution) — creation was never wired in. +// +// Subtests that expect NO launch attempt assert on stepGetter.GetStepCalls() +// rather than on the mock agent manager. autoStartTaskForStep calls GetStep +// synchronously before any launch work is handed off to a detached goroutine +// (see autoStartTaskForLoadedStep), so a zero call count is race-free proof +// that autoStartTaskForStep was never entered — unlike waiting on the agent +// manager, which races the test's own DB teardown against that goroutine. +func TestHandleTaskCreated(t *testing.T) { + ctx := context.Background() + + t.Run("launches a session for a task created directly on an auto-start step", 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})) + + // Mirrors what CreateOfficeTaskInWorkflow now stamps for a heavy + // routine run: the positive create-time opt-in, plus the routine's + // assignee carried as a task-level fallback since the Routine + // workflow's start step pins no agent. + metadata := map[string]interface{}{ + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t1", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Routine run", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Routine Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks["t1"] = &v1.Task{ + ID: "t1", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + } + launched := make(chan string, 1) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t1"}) + + select { + case got := <-launched: + if got != "routine-assignee" { + t.Fatalf("AgentProfileID = %q, want routine-assignee", got) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for auto-start launch on task creation") + } + }) + + t.Run("skips a queued task even when it carries the create-time opt-in", 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})) + + // QueuedForStepID marks a task as WIP-blocked: it isn't really + // entering step1 yet, it's waiting for capacity. autoStartTaskForStep + // bails on this before doing anything else, so a task.created firing + // for a queued task (e.g. via CreateTaskWithWorkflowStepAdmission + // overflow placement) must not attempt a launch even though it + // carries the create-time opt-in. + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t2", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + QueuedForStepID: "step1", + QueuedAt: &now, + Title: "Queued create", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: map[string]interface{}{models.MetaKeyAutoStartOnCreate: true}, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t)) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t2"}) + + if calls := stepGetter.GetStepCalls(); calls != 0 { + t.Fatalf("GetStepCalls() = %d, want 0 (autoStartTaskForStep must bail on QueuedForStepID before loading the step)", calls) + } + }) + + t.Run("skips a task created on an auto-start step without the create-time opt-in (R-1)", 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})) + + // An ordinary task with no launch opinion at all: no + // MetaKeyAutoStartOnCreate. A prior version of this guard launched + // unless the task carried MetaKeyDeferredLaunch, treating every + // other absence as "no launch opinion, please auto-start" — which + // incorrectly launched every producer that simply never set it + // (REST/MCP/WS creates without start_agent or prepare_session, + // CreateChildTask, etc.). Landing on an auto-start step must not be + // enough on its own. + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t3", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Ordinary create", + Description: "prompt", + State: v1.TaskStateCreated, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t)) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t3"}) + + if calls := stepGetter.GetStepCalls(); calls != 0 { + t.Fatalf("GetStepCalls() = %d, want 0 (handleTaskCreated must not call autoStartTaskForStep without the create-time opt-in)", calls) + } + }) + + t.Run("does not launch an office task even when it carries the create-time opt-in (R-2)", func(t *testing.T) { + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ + ID: "ws1", Name: "Test", OfficeWorkflowID: "wf1", CreatedAt: now, UpdatedAt: now, + })) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "Office WF", CreatedAt: now, UpdatedAt: now})) + + // task.IsFromOffice is a read-time projection: true because this + // task's workflow matches the workspace's OfficeWorkflowID. The + // office subscriber's own task.created handler already queues a run + // for every office task; if handleTaskCreated also launched one here + // (even though this task, hypothetically, carried the opt-in), the + // two would race to double-queue the same task with different + // idempotency keys. handleTaskCreated must defer to the office + // subscriber entirely, regardless of the opt-in marker. + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t4", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Office task", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: map[string]interface{}{models.MetaKeyAutoStartOnCreate: true}, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t)) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t4"}) + + if calls := stepGetter.GetStepCalls(); calls != 0 { + t.Fatalf("GetStepCalls() = %d, want 0 (handleTaskCreated must never call autoStartTaskForStep for an office task)", calls) + } + }) + + t.Run("claims the create-time opt-in so a duplicate task.created delivery cannot double-launch", 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})) + + metadata := map[string]interface{}{ + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t5", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Routine run", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Routine Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks["t5"] = &v1.Task{ + ID: "t5", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + } + launched := make(chan string, 2) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + // First delivery: the opt-in is present, so it is claimed (removed) + // synchronously — before the (async) launch even starts — and the + // launch proceeds. + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t5"}) + + reloaded, err := repo.GetTask(ctx, "t5") + requireNoError(t, err) + if models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("MetaKeyAutoStartOnCreate still present on the task after the first delivery; a duplicate task.created delivery would double-launch") + } + + select { + case <-launched: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the first delivery's launch") + } + + // Second, duplicate delivery of the same event (e.g. a redelivered + // task.created): the opt-in was already claimed by the first call, + // so HasAutoStartOnCreateIntent is now false and this must return + // before attempting a second launch. + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t5"}) + select { + case got := <-launched: + t.Fatalf("unexpected second launch for agent profile %q", got) + case <-time.After(100 * time.Millisecond): + } + }) + + t.Run("skips when the task cannot be found", func(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + // Should not panic — task not found is logged and returns. + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "nonexistent"}) + }) +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 8ca50590df..1881c24f94 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1275,6 +1275,7 @@ func NewService( OnContextWindowUpdated: s.handleContextWindowUpdated, OnTaskMoved: s.handleTaskMoved, OnTaskQueuePromoted: s.handleTaskQueuePromoted, + OnTaskCreated: s.handleTaskCreated, } s.watcher = watcher.NewWatcher(eventBus, handlers, cfg.QueueGroup, log) diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index d0ce51f411..97e8d619f1 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -189,6 +189,17 @@ const ( // that produced the launch-only runtime seed. A seed is valid only for this // profile, even if task profile selection changes before the first launch. MetaKeyInitialSessionRuntimeConfigProfileID = "initial_session_runtime_config_profile_id" + // MetaKeyAutoStartOnCreate is a positive opt-in a task creator stamps when + // it wants task.created to evaluate the destination step's on_enter + // actions immediately, as if creation were itself a transition into that + // step. Absence is the default and preserves existing behavior for every + // other producer (REST/MCP/WS create with or without start_agent / + // prepare_session, CreateChildTask, etc.) — those already have their own + // launch decision, and task.created must not second-guess it. Set today + // only by CreateOfficeTaskInWorkflow for materialized heavy-routine runs, + // whose Routine workflow start step has no other transition to carry it + // into an auto_start_agent evaluation. + MetaKeyAutoStartOnCreate = "auto_start_on_create" ) // IsAgentTitlePending reports whether task metadata contains the durable @@ -212,6 +223,15 @@ func IsAgentTitleOwner(metadata map[string]interface{}, sessionID string) bool { return sessionID != "" && IsAgentTitlePending(metadata) && AgentTitleOwnerSessionID(metadata) == sessionID } +// HasAutoStartOnCreateIntent reports whether task metadata carries the +// positive MetaKeyAutoStartOnCreate opt-in. Only an explicit true value +// counts — absence (the default for nearly every task producer) must never +// be read as "please auto-start me". +func HasAutoStartOnCreateIntent(metadata map[string]interface{}) bool { + intent, ok := metadata[MetaKeyAutoStartOnCreate].(bool) + return ok && intent +} + // TaskSession.Metadata key that records how the session came into existence. // workflow_switch means the session profile was selected by workflow routing // rather than direct user selection. diff --git a/docs/specs/office/system-design/scheduler-01.md b/docs/specs/office/system-design/scheduler-01.md index 4c2cb8df80..1ba239883b 100644 --- a/docs/specs/office/system-design/scheduler-01.md +++ b/docs/specs/office/system-design/scheduler-01.md @@ -123,7 +123,7 @@ Each trigger firing creates a routine run record (`office_routine_runs`) with `r #### Heavy vs lightweight routines - **Lightweight** (`task_template` empty): fire produces a taskless agent run. Continuation summary keyed by `routine:`. Use case: "check upstream PRs" without a trackable artifact. -- **Heavy** (`task_template` set): fire creates a fresh task in the `routine` workflow (a single auto-completing `in_progress -> done` step, system-flagged via `SystemWorkflowTemplateIDs` so heavy routine tasks inherit the hide-by-default UX), then a normal task-bound run. Use case: "daily review" where output should be a trackable item. +- **Heavy** (`task_template` set): fire creates a task in the system `routine` workflow (one hidden `in_progress -> done` step). Its `task.created` event evaluates `auto_start_agent` and starts a task-bound run. Use case: "daily review" with trackable output. #### Concurrency policy diff --git a/docs/specs/office/system-design/scheduler-02.md b/docs/specs/office/system-design/scheduler-02.md index 762b4b9d86..3abf85b730 100644 --- a/docs/specs/office/system-design/scheduler-02.md +++ b/docs/specs/office/system-design/scheduler-02.md @@ -316,7 +316,7 @@ The scheduler reads all `queued` and unexpired-retry wakeup requests on boot and - **GIVEN** a parent task with subtasks [Spec (requires_approval, assigned to planner), Build (blocked_by Spec, assigned to developer)], **WHEN** the planner completes the Spec subtask, **THEN** Spec moves to `in_review`. **WHEN** the user approves, **THEN** Spec moves to `done`, Build's blocker resolves, and a `task_blockers_resolved` wakeup is queued for the developer agent. The developer starts working on Build automatically. -- **GIVEN** a heavy routine "Daily Dep Update" with cron `0 9 * * *` in UTC and assignee "Frontend Worker", **WHEN** the clock reaches 09:00 UTC, **THEN** a task titled "Daily Dep Update - 2026-04-25" is created on the `routine` workflow, assigned to Frontend Worker, and a `task_assigned` wakeup is queued. +- **GIVEN** a heavy routine "Daily Dep Update" with cron `0 9 * * *` and assignee "Frontend Worker", **WHEN** 09:00 UTC arrives, **THEN** its task is created on the `routine` workflow and `task.created` starts the normal task-bound session. - **GIVEN** a routine with `concurrency_policy=skip_if_active` and an active task from a previous run, **WHEN** the routine fires again, **THEN** no new task is created and the routine run is recorded with status `skipped`.