fix(orchestrator): auto-start tasks created directly on a start step (#2967)
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"})
|
||||
})
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:<routine_id>`. 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
|
||||
|
||||
|
||||
@@ -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`.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user