diff --git a/apps/backend/internal/backendapp/cron.go b/apps/backend/internal/backendapp/cron.go index 5c9075fd8f..7eefc7260f 100644 --- a/apps/backend/internal/backendapp/cron.go +++ b/apps/backend/internal/backendapp/cron.go @@ -18,7 +18,7 @@ import ( ) // startCronScheduler builds and starts the Phase 5 shared cron loop. -// All three handlers (heartbeat, budget, routines) ride a single +// All handlers (heartbeat, budget, routines, Office recovery) ride a single // goroutine so the backend has one cron driver for all // task-model-unification timers. // @@ -31,16 +31,18 @@ func startCronScheduler( repos *Repositories, dispatcher *officeenginedispatcher.Dispatcher, routineSvc *officeroutines.RoutineService, + officeRecovery schedulercron.Handler, log *logger.Logger, -) { +) *schedulercron.Loop { heartbeat := buildHeartbeatHandler(repos, dispatcher, log) budget := buildBudgetHandler(repos, dispatcher, log) routines := schedulercron.NewRoutinesHandler(routineSvc, nil, log) loop := schedulercron.NewLoop(schedulercron.DefaultTickInterval, log, - heartbeat, budget, routines) - go loop.Start(ctx) + heartbeat, budget, routines, officeRecovery) + loop.Start(ctx) log.Info("phase 5 cron loop started", zap.Duration("interval", schedulercron.DefaultTickInterval)) + return loop } func buildHeartbeatHandler( diff --git a/apps/backend/internal/backendapp/helpers.go b/apps/backend/internal/backendapp/helpers.go index f3fa2d6769..f82603c266 100644 --- a/apps/backend/internal/backendapp/helpers.go +++ b/apps/backend/internal/backendapp/helpers.go @@ -1519,11 +1519,12 @@ func externalMCPAuthMiddleware(authSvc *auth.Service) gin.HandlerFunc { func runGracefulShutdown( server *http.Server, listeners *serverListeners, + scheduling *schedulingRuntime, orchestratorSvc *orchestrator.Service, lifecycleMgr *lifecycle.Manager, runCleanups func(), log *logger.Logger, -) { +) []error { start := time.Now() var shutdownErrs []error log.Info("Graceful shutdown started", @@ -1533,7 +1534,9 @@ func runGracefulShutdown( // Stop the background bind-retry loop before Shutdown so no new listener // can be created after the server begins closing its listeners. - listeners.Stop() + if listeners != nil { + listeners.Stop() + } shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), httpShutdownTimeout) defer shutdownCancel() @@ -1543,9 +1546,16 @@ func runGracefulShutdown( log.Error("HTTP server shutdown error", zap.Error(err)) } - if err := orchestratorSvc.Stop(); err != nil { + if err := scheduling.Stop(); err != nil { shutdownErrs = append(shutdownErrs, err) - log.Error("Orchestrator stop error", zap.Error(err)) + log.Error("Scheduler stop error", zap.Error(err)) + } + + if orchestratorSvc != nil { + if err := orchestratorSvc.Stop(); err != nil { + shutdownErrs = append(shutdownErrs, err) + log.Error("Orchestrator stop error", zap.Error(err)) + } } if err := stopLifecycleManager(lifecycleMgr, log); err != nil { @@ -1565,6 +1575,7 @@ func runGracefulShutdown( zap.Duration("duration", time.Since(start)), zap.Int("error_count", len(shutdownErrs))) _ = log.Sync() + return shutdownErrs } // stopLifecycleManager gracefully stops all agents and the lifecycle manager. diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index d8ed8bc0fb..77dab320e5 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -95,6 +95,7 @@ import ( // Runs queue (Phase 3 of task-model-unification) runsscheduler "github.com/kandev/kandev/internal/runs/scheduler" runsservice "github.com/kandev/kandev/internal/runs/service" + schedulercron "github.com/kandev/kandev/internal/scheduler/cron" // Workflow engine adapters (Phase 3.2 of task-model-unification) officeengineadapters "github.com/kandev/kandev/internal/office/engine_adapters" @@ -787,11 +788,16 @@ func startGatewayAndServe( } // ============================================ - // ORCHESTRATE CONFIG LOADER + WAKEUP SCHEDULER + // OFFICE FEATURES + GLOBAL RUN SCHEDULING // ============================================ - if ok := initOfficeServices(ctx, cfg, log, repos, services, orchestratorSvc, eventBus, agentctlBinaryPath, addCleanup, lifecycleMgr, agentRegistry); !ok { + runProcessorSvc, ok := initOfficeServices(ctx, cfg, log, repos, services, orchestratorSvc, eventBus, agentctlBinaryPath, addCleanup, lifecycleMgr, agentRegistry) + if !ok { return false } + scheduling := startSchedulingRuntime( + ctx, repos, services, eventBus, orchestratorSvc, runProcessorSvc, log, + ) + addCleanup(scheduling.Stop) // Wire subscription usage provider into the office agents service so the // /agents/:id/utilization endpoint can fetch live utilization data. @@ -882,7 +888,7 @@ func startGatewayAndServe( // and any subsequent /health call will land on a wired route. go waitListenerThenMarkReady(listeners.probeAddr(), log) - awaitShutdown(server, listeners, orchestratorSvc, lifecycleMgr, runCleanups, log) + awaitShutdown(server, listeners, scheduling, orchestratorSvc, lifecycleMgr, runCleanups, log) return true } @@ -944,9 +950,10 @@ func waitListenerThenMarkReady(listenAddr string, log *logger.Logger) { } } -// initOfficeServices constructs the office service with all dependencies, then -// sets up the reconciler, event subscribers, scheduler, and garbage collector. -// Returns false if a fatal error prevents startup. +// initOfficeServices constructs the run processor service for every backend and +// adds Office-only services, reconciliation, and subscribers when the feature +// is enabled. Global run and cron scheduling starts separately so this feature +// gate cannot disable workflow-generic queue_run dispatch. func initOfficeServices( ctx context.Context, cfg *config.Config, @@ -959,45 +966,24 @@ func initOfficeServices( addCleanup func(func() error), lifecycleMgr *lifecycle.Manager, agentRegistry *registry.Registry, -) bool { - // Feature gate: when features.office is off (the production default), - // skip every Office construction step. Downstream call sites - // (helpers.go route registration, cron scheduler, skill backfill, - // usage-provider wiring) already nil-check services.Office / - // services.OfficeSvcs, so leaving them nil is safe. - // See docs/decisions/0007-runtime-feature-flags.md. - if !cfg.Features.Office { - log.Info("Office feature disabled (features.office=false); skipping initialization") - return true - } - +) (*officeservice.Service, bool) { configBasePath := cfg.ResolvedHomeDir() - cfgLoader, cfgWriter := initOfficeConfigLoader(configBasePath, log, addCleanup) - - apiPort := cfg.Server.Port - if apiPort == 0 { - apiPort = ports.Backend + var cfgLoader *configloader.ConfigLoader + var cfgWriter *configloader.FileWriter + if cfg.Features.Office { + cfgLoader, cfgWriter = initOfficeConfigLoader(configBasePath, log, addCleanup) } - taskStarter := newOfficeTaskStarter(orchestratorSvc) + runProcessorSvc := newRunProcessorService( + cfg, repos, services, orchestratorSvc, eventBus, + agentctlBinaryPath, cfgLoader, cfgWriter, log, + ) + if !cfg.Features.Office { + log.Info("Office feature disabled; Office services skipped while global run scheduling remains enabled") + return runProcessorSvc, true + } - // Construct the office service with all dependencies at once so the - // compiler catches missing fields rather than failing at runtime. - services.Office = officeservice.NewService(officeservice.ServiceOptions{ - Repo: repos.Office, - Logger: log, - CfgLoader: cfgLoader, - CfgWriter: cfgWriter, - WorkspaceCreator: &taskWorkspaceCreatorAdapter{taskSvc: services.Task}, - TaskWorkspace: services.Task, - TaskCreator: &taskCreatorAdapter{taskSvc: services.Task}, - TaskPRs: &taskPRListerAdapter{gh: services.GitHub}, - APIBaseURL: fmt.Sprintf("http://localhost:%d/api/v1", apiPort), - TaskStarter: taskStarter, - TaskCanceller: orchestratorSvc, - AgentctlBinaryPath: agentctlBinaryPath, - EventBus: eventBus, - }) + services.Office = runProcessorSvc log.Info("Office service constructed with all dependencies") // office-costs Wave B: lazy models.dev pricing lookup. The Client @@ -1046,14 +1032,46 @@ func initOfficeServices( // receive defaults; curated lists are left alone. backfillAgentDefaultSkills(ctx, services, log) - // Register event subscribers and start scheduler + // Register Office-only event subscribers. Global scheduling starts after + // this initializer returns, regardless of the feature flag. if err := services.Office.RegisterEventSubscribers(eventBus); err != nil { log.Error("Failed to register office event subscribers", zap.Error(err)) - return false + return nil, false } - startOfficeSchedulersAndGC(ctx, cfg, repos, services, eventBus, orchestratorSvc, log) - return true + return runProcessorSvc, true +} + +func newRunProcessorService( + cfg *config.Config, + repos *Repositories, + services *Services, + orchestratorSvc *orchestrator.Service, + eventBus bus.EventBus, + agentctlBinaryPath string, + cfgLoader *configloader.ConfigLoader, + cfgWriter *configloader.FileWriter, + log *logger.Logger, +) *officeservice.Service { + apiPort := cfg.Server.Port + if apiPort == 0 { + apiPort = ports.Backend + } + return officeservice.NewService(officeservice.ServiceOptions{ + Repo: repos.Office, + Logger: log, + CfgLoader: cfgLoader, + CfgWriter: cfgWriter, + WorkspaceCreator: &taskWorkspaceCreatorAdapter{taskSvc: services.Task}, + TaskWorkspace: services.Task, + TaskCreator: &taskCreatorAdapter{taskSvc: services.Task}, + TaskPRs: &taskPRListerAdapter{gh: services.GitHub}, + APIBaseURL: fmt.Sprintf("http://localhost:%d/api/v1", apiPort), + TaskStarter: newOfficeTaskStarter(orchestratorSvc), + TaskCanceller: orchestratorSvc, + AgentctlBinaryPath: agentctlBinaryPath, + EventBus: eventBus, + }) } // wireOfficeSvcsDependencies wires inter-service dependencies into the @@ -1203,21 +1221,21 @@ func (a *schedulerTaskStarterAdapter) StartTaskWithRoute( }) } -// startOfficeSchedulersAndGC wires the runs service, workflow engine dispatcher, -// runs scheduler, cron scheduler, and GC sweep. Extracted to keep -// initOfficeServices within funlen limits. -func startOfficeSchedulersAndGC( +// startSchedulingRuntime wires the backend-wide runs service, workflow engine +// dispatcher, runs scheduler, and shared cron loop. Office recovery is attached +// only when Office feature services were initialized. +func startSchedulingRuntime( ctx context.Context, - cfg *config.Config, repos *Repositories, services *Services, eventBus bus.EventBus, orchestratorSvc *orchestrator.Service, + runProcessorSvc *officeservice.Service, log *logger.Logger, -) { - log.Info("Office scheduler wired to orchestrator StartTask") +) *schedulingRuntime { + log.Info("Global run processor wired to orchestrator StartTask") orchScheduler := officeservice.NewSchedulerIntegration( - services.Office, runsscheduler.TickIntervalFromEnv(), + runProcessorSvc, runsscheduler.TickIntervalFromEnv(), ) // Office task-handoffs prompt enrichment. The HandoffService is // constructed alongside the HTTP routes (helpers.go); we stash the @@ -1229,12 +1247,12 @@ func startOfficeSchedulersAndGC( runsSvc := runsservice.New( repos.Office.RunsRepository(), eventBus, log, nil, ) - services.Office.SetRunsService(runsSvc) + runProcessorSvc.SetRunsService(runsSvc) // Phase 4 (ADR-0004): wire the workflow engine's dependencies and a // dispatcher so office event subscribers route through the engine // unconditionally. engineDispatcher := wireWorkflowEngineForOffice( - orchestratorSvc, services.Office, services.Task, services.Workflow, repos, runsSvc, log, + orchestratorSvc, runProcessorSvc, services.Task, services.Workflow, repos, runsSvc, log, ) if services.OfficeSvcs != nil { services.OfficeSvcs.Dashboard.SetWorkflowEngineDispatcher(engineDispatcher) @@ -1245,7 +1263,7 @@ func startOfficeSchedulersAndGC( orchScheduler, runsSvc.SubscribeSignal(), runsscheduler.TickIntervalFromEnv(), log, ) - go runScheduler.Start(ctx) + runScheduler.Start(ctx) log.Info("Runs scheduler started", zap.Duration("tick", runsscheduler.TickIntervalFromEnv())) // Phase 5 (ADR-0004): start the shared cron loop. The routines handler @@ -1255,7 +1273,14 @@ func startOfficeSchedulersAndGC( if services.OfficeSvcs != nil { officeRoutines = services.OfficeSvcs.Routines } - startCronScheduler(ctx, repos, engineDispatcher, officeRoutines, log) + var officeRecovery schedulercron.Handler + if services.Office != nil { + officeRecovery = officeservice.NewOfficeRecoveryHandler(orchScheduler) + } + cronLoop := startCronScheduler( + ctx, repos, engineDispatcher, officeRoutines, officeRecovery, log, + ) + return &schedulingRuntime{runs: runScheduler, cron: cronLoop} } // wireWorkflowEngineForOffice composes the Phase 2 (ADR-0004) @@ -1823,6 +1848,7 @@ func buildHTTPServer( func awaitShutdown( server *http.Server, listeners *serverListeners, + scheduling *schedulingRuntime, orchestratorSvc *orchestrator.Service, lifecycleMgr *lifecycle.Manager, runCleanups func(), @@ -1849,5 +1875,5 @@ func awaitShutdown( log.Info("Received shutdown signal", zap.String("signal", sig.String()), zap.Int("pid", os.Getpid())) - runGracefulShutdown(server, listeners, orchestratorSvc, lifecycleMgr, runCleanups, log) + runGracefulShutdown(server, listeners, scheduling, orchestratorSvc, lifecycleMgr, runCleanups, log) } diff --git a/apps/backend/internal/backendapp/scheduler_wiring_test.go b/apps/backend/internal/backendapp/scheduler_wiring_test.go new file mode 100644 index 0000000000..0f9a33b29a --- /dev/null +++ b/apps/backend/internal/backendapp/scheduler_wiring_test.go @@ -0,0 +1,58 @@ +package backendapp + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" +) + +func TestGlobalSchedulingStartsOutsideOfficeFeatureGate(t *testing.T) { + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, "main.go", nil, 0) + if err != nil { + t.Fatalf("parse main.go: %v", err) + } + + var callsInOfficeInit, callsElsewhere int + var foundOfficeInit bool + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok { + continue + } + count := countNamedCalls(fn, "startSchedulingRuntime") + if fn.Name.Name == "initOfficeServices" { + foundOfficeInit = true + callsInOfficeInit += count + continue + } + callsElsewhere += count + } + + if !foundOfficeInit { + t.Fatal("initOfficeServices not found in main.go; re-point this guard at the Office-gated function") + } + if callsInOfficeInit > 0 { + t.Errorf("startSchedulingRuntime is called inside the Office-gated initializer %d time(s)", callsInOfficeInit) + } + if callsElsewhere == 0 { + t.Error("startSchedulingRuntime is not called outside the Office feature gate") + } +} + +func countNamedCalls(fn *ast.FuncDecl, name string) int { + count := 0 + ast.Inspect(fn, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + ident, ok := call.Fun.(*ast.Ident) + if ok && ident.Name == name { + count++ + } + return true + }) + return count +} diff --git a/apps/backend/internal/backendapp/shutdown_test.go b/apps/backend/internal/backendapp/shutdown_test.go new file mode 100644 index 0000000000..50b151b25e --- /dev/null +++ b/apps/backend/internal/backendapp/shutdown_test.go @@ -0,0 +1,85 @@ +package backendapp + +import ( + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +type blockingShutdownStopper struct { + entered chan struct{} + release chan struct{} + err error +} + +func (s *blockingShutdownStopper) Stop() error { + close(s.entered) + <-s.release + return s.err +} + +func TestRunGracefulShutdownStopsSchedulersBeforeDatabaseCleanup(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + defer server.Close() + + stopErr := errors.New("scheduler stop failed") + stopper := &blockingShutdownStopper{ + entered: make(chan struct{}), + release: make(chan struct{}), + err: stopErr, + } + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(stopper.release) }) } + t.Cleanup(release) + scheduling := &schedulingRuntime{runs: stopper} + dbClosed := make(chan struct{}) + cleanup := func() { + close(dbClosed) + } + + done := make(chan []error, 1) + go func() { + done <- runGracefulShutdown( + server.Config, + nil, + scheduling, + nil, + nil, + cleanup, + logger.Default(), + ) + }() + + select { + case <-stopper.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("scheduler stop was not called") + } + select { + case <-dbClosed: + t.Fatal("database cleanup ran before scheduler stop completed") + case <-time.After(50 * time.Millisecond): + } + + release() + var shutdownErrs []error + select { + case shutdownErrs = <-done: + case <-time.After(500 * time.Millisecond): + t.Fatal("graceful shutdown did not finish after scheduler stop completed") + } + select { + case <-dbClosed: + case <-time.After(100 * time.Millisecond): + t.Fatal("database cleanup did not run") + } + + if !errors.Is(errors.Join(shutdownErrs...), stopErr) { + t.Fatalf("shutdown errors = %v, want scheduler stop error", shutdownErrs) + } +} diff --git a/apps/backend/internal/backendapp/types.go b/apps/backend/internal/backendapp/types.go index 38f83b94bc..313bcb073b 100644 --- a/apps/backend/internal/backendapp/types.go +++ b/apps/backend/internal/backendapp/types.go @@ -1,6 +1,8 @@ package backendapp import ( + "errors" + settingsstore "github.com/kandev/kandev/internal/agent/settings/store" analyticsrepository "github.com/kandev/kandev/internal/analytics/repository" authservice "github.com/kandev/kandev/internal/auth" @@ -109,3 +111,33 @@ type Services struct { // Mode() == ModeDisabled and the middleware injects the synthetic identity. Auth *authservice.Service } + +type schedulerStopper interface { + Stop() error +} + +// schedulingRuntime owns the backend-wide queue and cron loops. Keeping the +// handles together gives startup-failure cleanup and signal-driven shutdown +// the same idempotent stop path. +type schedulingRuntime struct { + runs schedulerStopper + cron schedulerStopper +} + +func (s *schedulingRuntime) Stop() error { + if s == nil { + return nil + } + var errs []error + if s.cron != nil { + if err := s.cron.Stop(); err != nil { + errs = append(errs, err) + } + } + if s.runs != nil { + if err := s.runs.Stop(); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} diff --git a/apps/backend/internal/office/dashboard/handler_test.go b/apps/backend/internal/office/dashboard/handler_test.go index 5803c93d34..cfb8a3051d 100644 --- a/apps/backend/internal/office/dashboard/handler_test.go +++ b/apps/backend/internal/office/dashboard/handler_test.go @@ -149,6 +149,13 @@ func newTestDeps(t *testing.T) *testDeps { if err != nil { t.Fatalf("create tasks table: %v", err) } + _, err = db.Exec(`CREATE TABLE IF NOT EXISTS workspaces ( + id TEXT PRIMARY KEY, + office_workflow_id TEXT DEFAULT '' + )`) + if err != nil { + t.Fatalf("create workspaces table: %v", err) + } // workflows is referenced by FKs on workflow_steps (workflow store // schema). Create it before NewWithDB so the workflow repo's diff --git a/apps/backend/internal/office/infra/gc_test.go b/apps/backend/internal/office/infra/gc_test.go index 57c4aad674..e691a74d00 100644 --- a/apps/backend/internal/office/infra/gc_test.go +++ b/apps/backend/internal/office/infra/gc_test.go @@ -88,6 +88,12 @@ func newTestGC( if err != nil { t.Fatalf("create tasks table: %v", err) } + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS workspaces ( + id TEXT PRIMARY KEY, + office_workflow_id TEXT DEFAULT '' + )`); err != nil { + t.Fatalf("create workspaces table: %v", err) + } // ADR 0005 Wave F: GetTaskExecutionFields and other office reads // use a correlated subquery to workflow_step_participants / diff --git a/apps/backend/internal/office/repository/sqlite/tasks.go b/apps/backend/internal/office/repository/sqlite/tasks.go index dfdf436ee5..f32831cfec 100644 --- a/apps/backend/internal/office/repository/sqlite/tasks.go +++ b/apps/backend/internal/office/repository/sqlite/tasks.go @@ -59,6 +59,7 @@ type TaskExecutionFields struct { AssigneeAgentProfileID string `db:"assignee_agent_profile_id"` State string `db:"state"` WorkspaceID string `db:"workspace_id"` + IsFromOffice bool `db:"is_from_office"` } // GetTaskExecutionFields returns the execution-related fields for a task. @@ -68,7 +69,8 @@ func (r *Repository) GetTaskExecutionFields(ctx context.Context, taskID string) SELECT tasks.id, `+RunnerProjection("tasks")+` as assignee_agent_profile_id, COALESCE(tasks.state, '') as state, - COALESCE(tasks.workspace_id, '') as workspace_id + COALESCE(tasks.workspace_id, '') as workspace_id, + `+taskrepo.IsFromOfficePredicate("tasks")+` AS is_from_office FROM tasks WHERE tasks.id = ? `), taskID).StructScan(&fields) if err == sql.ErrNoRows { @@ -812,6 +814,23 @@ type UnstartedTaskRow struct { WorkspaceID string `db:"workspace_id"` } +// HasOfficeAdoption reports whether persisted Office configuration exists +// anywhere in the backend. It is intentionally cheap because the shared +// cron loop checks it before every recovery pass. +func (r *Repository) HasOfficeAdoption(ctx context.Context) (bool, error) { + var adopted bool + err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(` + SELECT EXISTS ( + SELECT 1 FROM workspaces + WHERE COALESCE(office_workflow_id, '') != '' + ) OR EXISTS ( + SELECT 1 FROM office_projects + WHERE COALESCE(workspace_id, '') != '' + ) + `)).Scan(&adopted) + return adopted, err +} + // ListUnstartedTasks returns TODO tasks with an assignee, not archived, // within the lookback window, that have no active run (queued/claimed/finished). // CountTasksByWorkspace returns the number of non-archived, non-ephemeral tasks @@ -837,6 +856,7 @@ func (r *Repository) ListUnstartedTasks( t.workspace_id FROM tasks t WHERE t.state = 'TODO' + AND `+taskrepo.IsFromOfficePredicate("t")+` AND `+RunnerProjection("t")+` != '' AND t.archived_at IS NULL AND t.created_at >= datetime('now', '-' || ? || ' hours') diff --git a/apps/backend/internal/office/repository/sqlite/tasks_test.go b/apps/backend/internal/office/repository/sqlite/tasks_test.go index 74979d8193..4f3c80a27f 100644 --- a/apps/backend/internal/office/repository/sqlite/tasks_test.go +++ b/apps/backend/internal/office/repository/sqlite/tasks_test.go @@ -19,6 +19,15 @@ func newSearchTestRepo(t *testing.T) *sqlite.Repository { repo := newTestRepo(t) ctx := context.Background() + if _, err := repo.ExecRaw(ctx, ` + CREATE TABLE IF NOT EXISTS workspaces ( + id TEXT PRIMARY KEY, + office_workflow_id TEXT DEFAULT '' + ) + `); err != nil { + t.Fatalf("create workspaces table: %v", err) + } + _, err := repo.ExecRaw(ctx, ` CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, @@ -125,6 +134,61 @@ func TestGetTaskExecutionFields_FallsBackToLatestTaskRunner(t *testing.T) { } } +func TestListUnstartedTasks_OnlyReturnsOfficeTasks(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workspaces (id, office_workflow_id) VALUES + ('ws-office', 'wf-office'), + ('ws-kanban', '') + `); err != nil { + t.Fatalf("insert workspaces: %v", err) + } + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_steps (id, agent_profile_id) VALUES + ('step-office', 'runner-office'), + ('step-kanban', 'runner-kanban') + `); err != nil { + t.Fatalf("insert workflow steps: %v", err) + } + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO tasks + (id, workspace_id, workflow_id, workflow_step_id, project_id, state, title, created_at, updated_at) + VALUES + ('task-office-workflow', 'ws-office', 'wf-office', 'step-office', '', 'TODO', 'Office workflow', datetime('now'), datetime('now')), + ('task-office-project', 'ws-kanban', 'wf-kanban', 'step-kanban', 'project-1', 'TODO', 'Office project', datetime('now'), datetime('now')), + ('task-kanban', 'ws-kanban', 'wf-kanban', 'step-kanban', '', 'TODO', 'Kanban task', datetime('now'), datetime('now')) + `); err != nil { + t.Fatalf("insert tasks: %v", err) + } + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants + (id, step_id, task_id, role, agent_profile_id, decision_required, position) + VALUES + ('runner-office-workflow', 'step-office', 'task-office-workflow', 'runner', 'runner-office', 0, 0), + ('runner-office-project', 'step-kanban', 'task-office-project', 'runner', 'runner-kanban', 0, 0), + ('runner-kanban', 'step-kanban', 'task-kanban', 'runner', 'runner-kanban', 0, 0) + `); err != nil { + t.Fatalf("insert runners: %v", err) + } + + rows, err := repo.ListUnstartedTasks(ctx, 24, 10) + if err != nil { + t.Fatalf("ListUnstartedTasks: %v", err) + } + got := make(map[string]bool, len(rows)) + for _, row := range rows { + got[row.ID] = true + } + if !got["task-office-workflow"] || !got["task-office-project"] { + t.Fatalf("office tasks = %#v, want workflow and project tasks", got) + } + if got["task-kanban"] { + t.Fatalf("ordinary Kanban task was recovered: %#v", got) + } +} + func TestSearchTasks_MatchesTitle(t *testing.T) { repo := newSearchTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/office/service/base_test.go b/apps/backend/internal/office/service/base_test.go index 514a4004d3..11e2718889 100644 --- a/apps/backend/internal/office/service/base_test.go +++ b/apps/backend/internal/office/service/base_test.go @@ -33,8 +33,8 @@ func initSharedAgentProfilesSchema(t *testing.T, db *sqlx.DB) { } // newTestService creates a Service for testing. Callers may supply a -// ServiceOptions to inject mocks or override defaults; Repo, Logger, -// CfgLoader and CfgWriter are always set from the test harness. +// ServiceOptions to inject mocks or override defaults; Repo, CfgLoader and +// CfgWriter are always set from the test harness. func newTestService(t *testing.T, overrides ...service.ServiceOptions) *service.Service { t.Helper() db, err := sqlx.Open("sqlite3", ":memory:") @@ -72,6 +72,12 @@ func newTestService(t *testing.T, overrides ...service.ServiceOptions) *service. if err != nil { t.Fatalf("create tasks table: %v", err) } + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS workspaces ( + id TEXT PRIMARY KEY, + office_workflow_id TEXT DEFAULT '' + )`); err != nil { + t.Fatalf("create workspaces table: %v", err) + } if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS workflow_steps ( id TEXT PRIMARY KEY, agent_profile_id TEXT NOT NULL DEFAULT '' @@ -138,6 +144,9 @@ func newTestService(t *testing.T, overrides ...service.ServiceOptions) *service. // applyServiceOverrides merges non-zero fields from o into opts. func applyServiceOverrides(opts *service.ServiceOptions, o service.ServiceOptions) { + if o.Logger != nil { + opts.Logger = o.Logger + } if o.CfgLoader != nil { opts.CfgLoader = o.CfgLoader } diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 6d963ed909..2419e5788a 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -674,11 +674,17 @@ func (s *Service) queueTaskAssignedRun( if taskID == "" { return nil } - if agentProfileID == "" && fallbackToStoredRunner { - fields, err := s.repo.GetTaskExecutionFields(ctx, taskID) - if err != nil || fields == nil { + fields, err := s.repo.GetTaskExecutionFields(ctx, taskID) + if err != nil { + if errors.Is(err, sqlite.ErrTaskNotFound) { return nil } + return fmt.Errorf("get task execution fields for assignment: %w", err) + } + if fields == nil || !fields.IsFromOffice { + return nil + } + if agentProfileID == "" && fallbackToStoredRunner { agentProfileID = fields.AssigneeAgentProfileID } if agentProfileID == "" { diff --git a/apps/backend/internal/office/service/event_subscribers_test.go b/apps/backend/internal/office/service/event_subscribers_test.go index b09f57362b..20f8d52683 100644 --- a/apps/backend/internal/office/service/event_subscribers_test.go +++ b/apps/backend/internal/office/service/event_subscribers_test.go @@ -8,6 +8,10 @@ import ( "sync/atomic" "testing" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" @@ -73,9 +77,15 @@ func (d *queueRunDispatcher) HandleTrigger( // fakeDispatcher / nil via svc.SetWorkflowEngineDispatcher. func newTestServiceWithBus(t *testing.T) (*service.Service, bus.EventBus) { t.Helper() - svc := newTestService(t) + return newTestServiceWithBusLogger(t, logger.Default()) +} + +func newTestServiceWithBusLogger( + t *testing.T, log *logger.Logger, +) (*service.Service, bus.EventBus) { + t.Helper() + svc := newTestService(t, service.ServiceOptions{Logger: log}) svc.SetSyncHandlers(true) - log := logger.Default() eb := bus.NewMemoryEventBus(log) if err := svc.RegisterEventSubscribers(eb); err != nil { t.Fatalf("register subscribers: %v", err) @@ -227,6 +237,7 @@ func TestTaskCreated_WakesAssigneeFromStoredRunner(t *testing.T) { createTestAgent(t, svc, "ws-1", "worker-created") insertTestTask(t, svc, "task-created-assigned", "ws-1") + svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-created-assigned") setTestTaskAssignee(t, svc, "task-created-assigned", "worker-created") event := bus.NewEvent(events.TaskCreated, "test", map[string]string{ @@ -286,6 +297,7 @@ func TestTaskAssigned_ReassignmentUsesAgentScopedIdempotency(t *testing.T) { createTestAgent(t, svc, "ws-1", "worker-old") createTestAgent(t, svc, "ws-1", "worker-new") insertTestTask(t, svc, "task-reassigned", "ws-1") + svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-reassigned") for _, agentID := range []string{"worker-old", "worker-new"} { event := bus.NewEvent(events.TaskUpdated, "test", map[string]string{ @@ -313,6 +325,54 @@ func TestTaskAssigned_ReassignmentUsesAgentScopedIdempotency(t *testing.T) { } } +func TestTaskAssigned_KanbanRunnerIsIgnored(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "worker-kanban") + insertTestTask(t, svc, "task-kanban-assigned", "ws-1") + setTestTaskAssignee(t, svc, "task-kanban-assigned", "worker-kanban") + + event := bus.NewEvent(events.TaskUpdated, "test", map[string]string{ + "task_id": "task-kanban-assigned", + "assignee_agent_profile_id": "worker-kanban", + }) + if err := eb.Publish(ctx, events.TaskUpdated, event); err != nil { + t.Fatalf("publish task updated event: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.Reason == service.RunReasonTaskAssigned { + t.Fatalf("Kanban assignment queued task_assigned run: %#v", run) + } + } +} + +func TestTaskAssigned_PropagatesTaskLookupFailure(t *testing.T) { + core, logs := observer.New(zapcore.ErrorLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("create observer logger: %v", err) + } + svc, eb := newTestServiceWithBusLogger(t, log) + svc.ExecSQL(t, `DROP TABLE tasks`) + + event := bus.NewEvent(events.TaskUpdated, "test", map[string]string{ + "task_id": "task-with-broken-lookup", + "assignee_agent_profile_id": "worker-1", + }) + if err := eb.Publish(context.Background(), events.TaskUpdated, event); err != nil { + t.Fatalf("publish task update: %v", err) + } + if logs.FilterMessage("Event handler error").Len() == 0 { + t.Fatal("task lookup failure was not surfaced by the event bus") + } +} + func TestCommentCreated_SkipsSelfComment(t *testing.T) { svc, eb := newTestServiceWithBus(t) ctx := context.Background() diff --git a/apps/backend/internal/office/service/scheduler_integration.go b/apps/backend/internal/office/service/scheduler_integration.go index 1d37ce033d..9b2edea8f2 100644 --- a/apps/backend/internal/office/service/scheduler_integration.go +++ b/apps/backend/internal/office/service/scheduler_integration.go @@ -74,7 +74,7 @@ func NewSchedulerIntegration(svc *Service, tickInterval time.Duration) *Schedule return &SchedulerIntegration{ svc: svc, tickInterval: tickInterval, - logger: svc.logger.WithFields(zap.String("component", "office-scheduler")), + logger: svc.logger.WithFields(zap.String("component", "runs-processor")), } } @@ -114,6 +114,9 @@ func (si *SchedulerIntegration) tick(ctx context.Context) { for i := 0; i < maxRunsPerTick; i++ { run, err := si.svc.ClaimNextRun(ctx) if err != nil { + if ctx.Err() != nil { + return + } si.logger.Error("failed to claim run", zap.Error(err)) return } @@ -123,7 +126,6 @@ func (si *SchedulerIntegration) tick(ctx context.Context) { si.processRun(ctx, run) } - si.recoverUnstartedTasks(ctx) si.recoverStaleClaimedRuns(ctx) } @@ -140,6 +142,9 @@ func (si *SchedulerIntegration) liftParkedRoutingRuns(ctx context.Context) { } lifted, err := rd.LiftParkedRuns(ctx, time.Now().UTC()) if err != nil { + if ctx.Err() != nil { + return + } si.logger.Warn("lift parked runs failed", zap.Error(err)) return } @@ -151,6 +156,9 @@ func (si *SchedulerIntegration) liftParkedRoutingRuns(ctx context.Context) { func (si *SchedulerIntegration) recoverStaleClaimedRuns(ctx context.Context) { count, err := si.svc.repo.RecoverStale(ctx, time.Now().UTC().Add(-staleClaimedRunAge)) if err != nil { + if ctx.Err() != nil { + return + } si.logger.Error("failed to recover stale claimed runs", zap.Error(err)) return } diff --git a/apps/backend/internal/office/service/scheduler_recovery.go b/apps/backend/internal/office/service/scheduler_recovery.go index 06f43cd5fb..3f8a73b24c 100644 --- a/apps/backend/internal/office/service/scheduler_recovery.go +++ b/apps/backend/internal/office/service/scheduler_recovery.go @@ -2,33 +2,79 @@ package service import ( "context" + "fmt" "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" ) // maxRecoveryPerTick caps the number of unstarted tasks recovered in one tick. const maxRecoveryPerTick = 5 -// recoverUnstartedTasks queues a task_assigned run for TODO tasks that were -// never picked up by the scheduler, guarded by the recovery lookback window. -func (si *SchedulerIntegration) recoverUnstartedTasks(ctx context.Context) { +// OfficeRecoveryHandler drives Office-only unstarted-task maintenance from +// the shared cron loop. Generic queued-run dispatch remains owned by the +// runs scheduler. +type OfficeRecoveryHandler struct { + scheduler *SchedulerIntegration + logger *logger.Logger +} + +// NewOfficeRecoveryHandler constructs the Office maintenance handler for a +// scheduler integration. +func NewOfficeRecoveryHandler(si *SchedulerIntegration) *OfficeRecoveryHandler { + return &OfficeRecoveryHandler{ + scheduler: si, + logger: si.svc.logger.WithFields(zap.String("component", "office-recovery")), + } +} + +// Name implements scheduler/cron.Handler. +func (h *OfficeRecoveryHandler) Name() string { return "office_recovery" } + +// Tick implements scheduler/cron.Handler. The adoption check prevents an +// Office-enabled but Kanban-only installation from scanning task rows. +func (h *OfficeRecoveryHandler) Tick(ctx context.Context) error { + adopted, err := h.scheduler.svc.repo.HasOfficeAdoption(ctx) + if err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("check Office adoption: %w", err) + } + if !adopted { + return nil + } + h.scheduler.recoverUnstartedTasks(ctx, h.logger) + return nil +} + +// recoverUnstartedTasks queues a task_assigned run for authoritative Office +// TODO tasks that were never picked up, guarded by the lookback window. +func (si *SchedulerIntegration) recoverUnstartedTasks(ctx context.Context, log *logger.Logger) { lookbackHours := si.svc.GetRecoveryLookbackHours() tasks, err := si.svc.repo.ListUnstartedTasks(ctx, lookbackHours, maxRecoveryPerTick) if err != nil { - si.logger.Error("recovery sweep: list unstarted tasks failed", zap.Error(err)) + if ctx.Err() != nil { + return + } + log.Error("recovery sweep: list unstarted tasks failed", zap.Error(err)) return } for _, t := range tasks { - si.logger.Info("recovery sweep: re-queueing unstarted task", + log.Info("recovery sweep: re-queueing unstarted task", zap.String("task_id", t.ID), zap.String("agent_profile_id", t.AssigneeAgentProfileID)) payload := mustJSON(map[string]string{"task_id": t.ID}) if err := si.svc.QueueRun(ctx, t.AssigneeAgentProfileID, RunReasonTaskAssigned, payload, ""); err != nil { - si.logger.Error("recovery sweep: queue run failed", + if ctx.Err() != nil { + return + } + log.Error("recovery sweep: queue run failed", zap.String("task_id", t.ID), zap.Error(err)) } } diff --git a/apps/backend/internal/office/service/scheduler_recovery_test.go b/apps/backend/internal/office/service/scheduler_recovery_test.go new file mode 100644 index 0000000000..0549205c3c --- /dev/null +++ b/apps/backend/internal/office/service/scheduler_recovery_test.go @@ -0,0 +1,62 @@ +package service_test + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/office/service" +) + +func TestOfficeRecoveryHandler_SkipsWithoutOfficeAdoption(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "worker-office") + insertTestTask(t, svc, "task-office-project", "ws-1") + svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-office-project") + setTestTaskAssignee(t, svc, "task-office-project", "worker-office") + + handler := service.NewOfficeRecoveryHandler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("recovery tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("recovery ran without Office adoption: %#v", runs) + } +} + +func TestOfficeRecoveryHandlerActivatesAfterOfficeProjectAdoption(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "worker-office") + insertTestTask(t, svc, "task-office-project", "ws-1") + svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-office-project") + setTestTaskAssignee(t, svc, "task-office-project", "worker-office") + + handler := service.NewOfficeRecoveryHandler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("initial recovery tick: %v", err) + } + + svc.ExecSQL(t, ` + INSERT INTO office_projects (id, workspace_id, name, created_at, updated_at) + VALUES ('office-project', 'ws-1', 'Office Project', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + `) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("activated recovery tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 1 || runs[0].Reason != service.RunReasonTaskAssigned { + t.Fatalf("recovery runs = %#v, want one task_assigned run", runs) + } +} diff --git a/apps/backend/internal/runs/scheduler/goleak_test.go b/apps/backend/internal/runs/scheduler/goleak_test.go new file mode 100644 index 0000000000..f4754e9e81 --- /dev/null +++ b/apps/backend/internal/runs/scheduler/goleak_test.go @@ -0,0 +1,11 @@ +package scheduler + +import ( + "testing" + + "go.uber.org/goleak" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} diff --git a/apps/backend/internal/runs/scheduler/scheduler.go b/apps/backend/internal/runs/scheduler/scheduler.go index 3d50d55a00..1f443fd4c0 100644 --- a/apps/backend/internal/runs/scheduler/scheduler.go +++ b/apps/backend/internal/runs/scheduler/scheduler.go @@ -19,6 +19,7 @@ import ( "context" "os" "strconv" + "sync" "time" "go.uber.org/zap" @@ -74,6 +75,11 @@ type Scheduler struct { signal <-chan struct{} tickInterval time.Duration log *logger.Logger + + mu sync.Mutex + cancel context.CancelFunc + wg sync.WaitGroup + started bool } // New constructs a Scheduler. The signal channel comes from the runs @@ -96,11 +102,47 @@ func New( } } -// Start runs the tick + signal loop until the context is cancelled. -// Call it from a background goroutine. A single goroutine handles -// both wake-up sources so claim ordering is deterministic and -// per-agent serialisation is unaffected by the new signal path. +// Start launches the tick + signal loop. Calling Start more than once +// without Stop is a no-op. A single owned goroutine handles both wake-up +// sources so claim ordering is deterministic and per-agent serialisation is +// unaffected by the signal path. func (s *Scheduler) Start(ctx context.Context) { + s.mu.Lock() + if s.started { + s.mu.Unlock() + return + } + s.started = true + loopCtx, cancel := context.WithCancel(ctx) + s.cancel = cancel + s.wg.Add(1) + s.mu.Unlock() + + go s.loop(loopCtx) +} + +// Stop cancels the scheduler loop and waits for the active processor tick to +// return. It is safe to call repeatedly and concurrently with Start. +func (s *Scheduler) Stop() error { + s.mu.Lock() + defer s.mu.Unlock() + if !s.started { + return nil + } + + if s.cancel != nil { + s.cancel() + } + s.wg.Wait() + + s.started = false + s.cancel = nil + s.log.Info("runs scheduler stopped") + return nil +} + +func (s *Scheduler) loop(ctx context.Context) { + defer s.wg.Done() s.log.Info("runs scheduler starting", zap.Duration("tick_interval", s.tickInterval), zap.Bool("signal_enabled", s.signal != nil)) @@ -118,8 +160,14 @@ func (s *Scheduler) Start(ctx context.Context) { s.log.Info("runs scheduler stopping") return case <-ticker.C: + if ctx.Err() != nil { + return + } s.processor.Tick(ctx) case <-s.signal: + if ctx.Err() != nil { + return + } now := time.Now() if now.Sub(lastSignal) < SignalCoalesceWindow { continue diff --git a/apps/backend/internal/runs/scheduler/scheduler_concurrency_test.go b/apps/backend/internal/runs/scheduler/scheduler_concurrency_test.go new file mode 100644 index 0000000000..611c3b7deb --- /dev/null +++ b/apps/backend/internal/runs/scheduler/scheduler_concurrency_test.go @@ -0,0 +1,88 @@ +package scheduler + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +type cancelBlockingProcessor struct { + entered chan struct{} + release chan struct{} +} + +func (p *cancelBlockingProcessor) Tick(ctx context.Context) { + close(p.entered) + <-ctx.Done() + <-p.release +} + +func TestScheduler_ConcurrentStopUsesOneLifecycleTransition(t *testing.T) { + signalCh := make(chan struct{}, 1) + processor := &cancelBlockingProcessor{ + entered: make(chan struct{}), + release: make(chan struct{}), + } + scheduler := New(processor, signalCh, time.Hour, logger.Default()) + scheduler.Start(context.Background()) + + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(processor.release) }) } + t.Cleanup(func() { + release() + _ = scheduler.Stop() + }) + + signalCh <- struct{}{} + select { + case <-processor.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("scheduler did not enter processor tick") + } + + firstCancel := make(chan struct{}) + secondCancel := make(chan struct{}) + var cancelCalls atomic.Int32 + scheduler.mu.Lock() + originalCancel := scheduler.cancel + scheduler.cancel = func() { + switch cancelCalls.Add(1) { + case 1: + close(firstCancel) + case 2: + close(secondCancel) + } + originalCancel() + } + stopDone := make(chan error, 2) + go func() { stopDone <- scheduler.Stop() }() + go func() { stopDone <- scheduler.Stop() }() + scheduler.mu.Unlock() + + select { + case <-firstCancel: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not cancel the scheduler") + } + select { + case <-secondCancel: + t.Fatal("concurrent Stop started a second lifecycle transition") + case <-time.After(100 * time.Millisecond): + } + + release() + for range 2 { + select { + case err := <-stopDone: + if err != nil { + t.Fatalf("Stop returned error: %v", err) + } + case <-time.After(100 * time.Millisecond): + t.Fatal("concurrent Stop did not return after the tick drained") + } + } +} diff --git a/apps/backend/internal/runs/scheduler/scheduler_test.go b/apps/backend/internal/runs/scheduler/scheduler_test.go index e5c850474f..b2f5afa6dd 100644 --- a/apps/backend/internal/runs/scheduler/scheduler_test.go +++ b/apps/backend/internal/runs/scheduler/scheduler_test.go @@ -2,6 +2,7 @@ package scheduler_test import ( "context" + "sync" "sync/atomic" "testing" "time" @@ -23,6 +24,23 @@ func (f *fakeProcessor) Tick(_ context.Context) { f.last.Store(time.Now().UnixNano()) } +type blockingProcessor struct { + calls atomic.Int64 + entered chan struct{} + second chan struct{} + release chan struct{} +} + +func (p *blockingProcessor) Tick(_ context.Context) { + switch p.calls.Add(1) { + case 1: + close(p.entered) + case 2: + close(p.second) + } + <-p.release +} + func newTestLogger(t *testing.T) *logger.Logger { t.Helper() log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"}) @@ -42,6 +60,7 @@ func TestScheduler_TickFiresProcessor(t *testing.T) { proc := &fakeProcessor{} s := scheduler.New(proc, nil, 20*time.Millisecond, newTestLogger(t)) go s.Start(ctx) + t.Cleanup(func() { _ = s.Stop() }) deadline := time.Now().Add(500 * time.Millisecond) for time.Now().Before(deadline) { @@ -67,6 +86,7 @@ func TestScheduler_SignalDrivesClaimUnder100ms(t *testing.T) { // the signal path, not the safety-net timer. s := scheduler.New(proc, signalCh, 5*time.Second, newTestLogger(t)) go s.Start(ctx) + t.Cleanup(func() { _ = s.Stop() }) // Give the goroutine a moment to enter the select. time.Sleep(10 * time.Millisecond) @@ -97,6 +117,7 @@ func TestScheduler_StopsOnContextCancel(t *testing.T) { proc := &fakeProcessor{} s := scheduler.New(proc, nil, 20*time.Millisecond, newTestLogger(t)) go s.Start(ctx) + t.Cleanup(func() { _ = s.Stop() }) time.Sleep(80 * time.Millisecond) cancel() @@ -110,3 +131,104 @@ func TestScheduler_StopsOnContextCancel(t *testing.T) { t.Errorf("scheduler kept ticking after cancel: pre=%d post=%d", pre, post) } } + +func TestScheduler_StartIsIdempotentAndStopWaitsForActiveTick(t *testing.T) { + ctx := context.Background() + signalCh := make(chan struct{}, 2) + proc := &blockingProcessor{ + entered: make(chan struct{}), + second: make(chan struct{}), + release: make(chan struct{}), + } + s := scheduler.New(proc, signalCh, time.Hour, newTestLogger(t)) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(proc.release) }) } + t.Cleanup(func() { + release() + _ = s.Stop() + }) + go s.Start(ctx) + go s.Start(ctx) + + signalCh <- struct{}{} + select { + case <-proc.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("scheduler did not start the processor") + } + + signalCh <- struct{}{} + select { + case <-proc.second: + t.Fatal("duplicate Start launched a second scheduler loop") + case <-time.After(50 * time.Millisecond): + } + + stopDone := make(chan struct{}) + go func() { + if err := s.Stop(); err != nil { + t.Errorf("Stop returned error: %v", err) + } + close(stopDone) + }() + select { + case <-stopDone: + t.Fatal("Stop returned while a processor tick was still active") + case <-time.After(50 * time.Millisecond): + } + + release() + select { + case <-stopDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not join the processor loop") + } + if err := s.Stop(); err != nil { + t.Fatalf("Stop returned error: %v", err) + } +} + +func TestScheduler_ParentCancellationDrainsActiveTick(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + signalCh := make(chan struct{}, 1) + proc := &blockingProcessor{ + entered: make(chan struct{}), + second: make(chan struct{}), + release: make(chan struct{}), + } + s := scheduler.New(proc, signalCh, time.Hour, newTestLogger(t)) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(proc.release) }) } + t.Cleanup(func() { + release() + _ = s.Stop() + }) + go s.Start(ctx) + signalCh <- struct{}{} + select { + case <-proc.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("scheduler did not start the processor") + } + + cancel() + stopDone := make(chan struct{}) + go func() { + if err := s.Stop(); err != nil { + t.Errorf("Stop returned error: %v", err) + } + close(stopDone) + }() + select { + case <-stopDone: + t.Fatal("parent cancellation did not wait for the active tick") + case <-time.After(50 * time.Millisecond): + } + + release() + select { + case <-stopDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not finish after the active tick drained") + } +} diff --git a/apps/backend/internal/scheduler/cron/cron.go b/apps/backend/internal/scheduler/cron/cron.go index 16e90b2a0f..db84c4d082 100644 --- a/apps/backend/internal/scheduler/cron/cron.go +++ b/apps/backend/internal/scheduler/cron/cron.go @@ -60,6 +60,11 @@ type Loop struct { interval time.Duration handlers []Handler log *logger.Logger + + mu sync.Mutex + cancel context.CancelFunc + wg sync.WaitGroup + started bool } // NewLoop constructs a Loop. A non-positive interval falls back to @@ -70,21 +75,64 @@ func NewLoop(interval time.Duration, log *logger.Logger, handlers ...Handler) *L if interval <= 0 { interval = DefaultTickInterval } + configured := make([]Handler, 0, len(handlers)) + for _, handler := range handlers { + if handler != nil { + configured = append(configured, handler) + } + } return &Loop{ interval: interval, - handlers: handlers, + handlers: configured, log: log.WithFields(zap.String("component", "scheduler-cron")), } } -// Start runs the loop until ctx is cancelled. Call from a background -// goroutine. The first tick fires after interval — handlers must not -// rely on a "tick at t=0" semantic. +// Start launches the loop. Calling Start more than once without Stop is a +// no-op. The first tick fires after interval — handlers must not rely on a +// "tick at t=0" semantic. func (l *Loop) Start(ctx context.Context) { if len(l.handlers) == 0 { l.log.Info("cron loop start: no handlers, exiting") return } + + l.mu.Lock() + if l.started { + l.mu.Unlock() + return + } + l.started = true + loopCtx, cancel := context.WithCancel(ctx) + l.cancel = cancel + l.wg.Add(1) + l.mu.Unlock() + + go l.loop(loopCtx) +} + +// Stop cancels the loop and waits for all active handlers to return. It is +// safe to call repeatedly and concurrently with Start. +func (l *Loop) Stop() error { + l.mu.Lock() + defer l.mu.Unlock() + if !l.started { + return nil + } + + if l.cancel != nil { + l.cancel() + } + l.wg.Wait() + + l.started = false + l.cancel = nil + l.log.Info("cron loop stopped") + return nil +} + +func (l *Loop) loop(ctx context.Context) { + defer l.wg.Done() names := make([]string, 0, len(l.handlers)) for _, h := range l.handlers { names = append(names, h.Name()) @@ -102,6 +150,9 @@ func (l *Loop) Start(ctx context.Context) { l.log.Info("cron loop stopping") return case <-ticker.C: + if ctx.Err() != nil { + return + } l.fanOut(ctx) } } @@ -124,7 +175,7 @@ func (l *Loop) fanOut(ctx context.Context) { zap.Any("recover", r)) } }() - if err := h.Tick(ctx); err != nil { + if err := h.Tick(ctx); err != nil && ctx.Err() == nil { l.log.Error("cron handler tick failed", zap.String("handler", h.Name()), zap.Error(err)) diff --git a/apps/backend/internal/scheduler/cron/cron_test.go b/apps/backend/internal/scheduler/cron/cron_test.go index 6ce6f80609..19a2be2d36 100644 --- a/apps/backend/internal/scheduler/cron/cron_test.go +++ b/apps/backend/internal/scheduler/cron/cron_test.go @@ -3,6 +3,7 @@ package cron import ( "context" "errors" + "sync" "sync/atomic" "testing" "time" @@ -22,6 +23,25 @@ func (h *recordHandler) Tick(_ context.Context) error { return h.err } +type blockingHandler struct { + calls atomic.Int32 + entered chan struct{} + second chan struct{} + release chan struct{} +} + +func (h *blockingHandler) Name() string { return "blocking" } +func (h *blockingHandler) Tick(_ context.Context) error { + switch h.calls.Add(1) { + case 1: + close(h.entered) + case 2: + close(h.second) + } + <-h.release + return nil +} + func TestNewLoop_FallsBackToDefaultInterval(t *testing.T) { l := NewLoop(0, logger.Default()) if l.interval != DefaultTickInterval { @@ -29,6 +49,14 @@ func TestNewLoop_FallsBackToDefaultInterval(t *testing.T) { } } +func TestNewLoop_IgnoresNilHandlers(t *testing.T) { + handler := &recordHandler{name: "configured"} + l := NewLoop(time.Second, logger.Default(), handler, nil) + if len(l.handlers) != 1 || l.handlers[0] != handler { + t.Fatalf("handlers = %#v, want only configured handler", l.handlers) + } +} + func TestLoop_FiresEachHandlerPerTick(t *testing.T) { a := &recordHandler{name: "a"} b := &recordHandler{name: "b"} @@ -40,6 +68,7 @@ func TestLoop_FiresEachHandlerPerTick(t *testing.T) { l.Start(ctx) close(done) }() + t.Cleanup(func() { _ = l.Stop() }) // Wait for at least two ticks of each handler. deadline := time.After(500 * time.Millisecond) @@ -60,6 +89,7 @@ func TestLoop_ErrorFromHandlerDoesNotStopLoop(t *testing.T) { l := NewLoop(15*time.Millisecond, logger.Default(), bad, good) ctx, cancel := context.WithCancel(context.Background()) go l.Start(ctx) + t.Cleanup(func() { _ = l.Stop() }) defer cancel() deadline := time.After(500 * time.Millisecond) @@ -84,6 +114,7 @@ func TestLoop_PanicInHandlerIsRecovered(t *testing.T) { l := NewLoop(15*time.Millisecond, logger.Default(), panicker, good) ctx, cancel := context.WithCancel(context.Background()) go l.Start(ctx) + t.Cleanup(func() { _ = l.Stop() }) defer cancel() deadline := time.After(500 * time.Millisecond) @@ -96,6 +127,178 @@ func TestLoop_PanicInHandlerIsRecovered(t *testing.T) { } } +func TestLoop_StartIsIdempotentAndStopWaitsForFanOut(t *testing.T) { + h := &blockingHandler{ + entered: make(chan struct{}), + second: make(chan struct{}), + release: make(chan struct{}), + } + l := NewLoop(time.Millisecond, logger.Default(), h) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(h.release) }) } + t.Cleanup(func() { + release() + _ = l.Stop() + }) + ctx := context.Background() + go l.Start(ctx) + go l.Start(ctx) + + select { + case <-h.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("cron loop did not start the handler") + } + select { + case <-h.second: + t.Fatal("duplicate Start launched a second cron loop") + case <-time.After(50 * time.Millisecond): + } + + stopDone := make(chan struct{}) + go func() { + if err := l.Stop(); err != nil { + t.Errorf("Stop returned error: %v", err) + } + close(stopDone) + }() + select { + case <-stopDone: + t.Fatal("Stop returned while handler fan-out was still active") + case <-time.After(50 * time.Millisecond): + } + + release() + select { + case <-stopDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not join cron handler fan-out") + } + if err := l.Stop(); err != nil { + t.Fatalf("Stop returned error: %v", err) + } +} + +func TestLoop_ParentCancellationDrainsFanOut(t *testing.T) { + h := &blockingHandler{ + entered: make(chan struct{}), + second: make(chan struct{}), + release: make(chan struct{}), + } + ctx, cancel := context.WithCancel(context.Background()) + l := NewLoop(time.Millisecond, logger.Default(), h) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(h.release) }) } + t.Cleanup(func() { + release() + _ = l.Stop() + }) + go l.Start(ctx) + + select { + case <-h.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("cron loop did not start the handler") + } + cancel() + stopDone := make(chan struct{}) + go func() { + if err := l.Stop(); err != nil { + t.Errorf("Stop returned error: %v", err) + } + close(stopDone) + }() + select { + case <-stopDone: + t.Fatal("parent cancellation did not wait for handler fan-out") + case <-time.After(50 * time.Millisecond): + } + + release() + select { + case <-stopDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not finish after handler fan-out drained") + } +} + +type cancelBlockingHandler struct { + entered chan struct{} + release chan struct{} +} + +func (h *cancelBlockingHandler) Name() string { return "cancel_blocking" } +func (h *cancelBlockingHandler) Tick(ctx context.Context) error { + close(h.entered) + <-ctx.Done() + <-h.release + return nil +} + +func TestLoop_ConcurrentStopUsesOneLifecycleTransition(t *testing.T) { + handler := &cancelBlockingHandler{ + entered: make(chan struct{}), + release: make(chan struct{}), + } + loop := NewLoop(time.Millisecond, logger.Default(), handler) + loop.Start(context.Background()) + + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(handler.release) }) } + t.Cleanup(func() { + release() + _ = loop.Stop() + }) + + select { + case <-handler.entered: + case <-time.After(100 * time.Millisecond): + t.Fatal("cron loop did not enter handler tick") + } + + firstCancel := make(chan struct{}) + secondCancel := make(chan struct{}) + var cancelCalls atomic.Int32 + loop.mu.Lock() + originalCancel := loop.cancel + loop.cancel = func() { + switch cancelCalls.Add(1) { + case 1: + close(firstCancel) + case 2: + close(secondCancel) + } + originalCancel() + } + stopDone := make(chan error, 2) + go func() { stopDone <- loop.Stop() }() + go func() { stopDone <- loop.Stop() }() + loop.mu.Unlock() + + select { + case <-firstCancel: + case <-time.After(100 * time.Millisecond): + t.Fatal("Stop did not cancel the cron loop") + } + select { + case <-secondCancel: + t.Fatal("concurrent Stop started a second lifecycle transition") + case <-time.After(100 * time.Millisecond): + } + + release() + for range 2 { + select { + case err := <-stopDone: + if err != nil { + t.Fatalf("Stop returned error: %v", err) + } + case <-time.After(100 * time.Millisecond): + t.Fatal("concurrent Stop did not return after fan-out drained") + } + } +} + // handlerFunc adapts a function pair to the Handler interface so tests // can express tiny throwaway handlers without ceremony. type handlerFunc struct { diff --git a/apps/backend/internal/scheduler/cron/goleak_test.go b/apps/backend/internal/scheduler/cron/goleak_test.go new file mode 100644 index 0000000000..0eb2235e07 --- /dev/null +++ b/apps/backend/internal/scheduler/cron/goleak_test.go @@ -0,0 +1,11 @@ +package cron + +import ( + "testing" + + "go.uber.org/goleak" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 3043c2b1c8..06e57798a0 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -117,6 +117,14 @@ func isFromOfficeProjection(alias string) string { )` } +// IsFromOfficePredicate returns the shared SQL predicate for authoritative +// Office-task identity. Callers that query tasks outside this repository must +// use this expression so project-linked tasks and canonical Office-workflow +// tasks are classified consistently with models.Task.IsFromOffice. +func IsFromOfficePredicate(alias string) string { + return isFromOfficeProjection(alias) +} + // excludeConfigModePredicate delegates to the shared dialect helper (also // used by internal/analytics/repository/sqlite's CodeStats read, so both // cover the same office config-mode exclusion). diff --git a/docs/decisions/2026-08-01-global-run-scheduler-ownership.md b/docs/decisions/2026-08-01-global-run-scheduler-ownership.md new file mode 100644 index 0000000000..81009d0d6b --- /dev/null +++ b/docs/decisions/2026-08-01-global-run-scheduler-ownership.md @@ -0,0 +1,92 @@ +# ADR-2026-08-01-global-run-scheduler-ownership: Separate Global Run Dispatch from Office Maintenance + +**Status:** accepted +**Date:** 2026-08-01 +**Area:** backend, workflow + +## Context + +Task-model unification made `runs` a shared queue for workflow-emitted work, +but its loop is still constructed inside Office startup and its processor still +mixes generic queue draining with Office unstarted-task recovery. The loops are +started as untracked goroutines using the application context, whose +cancellation cleanup currently runs after repository cleanup; a periodic tick +can therefore query SQLite after it has closed. Enabling Office also makes the +recovery sweep inspect any assigned `TODO` task, including ordinary Kanban +tasks, even when no workspace has adopted Office. + +## Decision + +Kandev owns one backend-wide scheduler for the shared `runs` queue. The +scheduler is not replicated per workspace and processes explicit queued work +from every workflow style. + +Generic run dispatch and Office maintenance are separate responsibilities: + +- Queue draining, missed-signal recovery, scheduled retries, and stale-claim + recovery belong to the shared runs subsystem. +- Assignment alone is not an autonomy signal for ordinary Kanban tasks. + Office assignment subscribers and unstarted-task recovery use the same + authoritative Office-task predicate as `Task.IsFromOffice`: a project-linked + task or a task in `workspaces.office_workflow_id`. +- Explicit workflow `queue_run` actions remain generic and are processed for + any workflow style. +- Office unstarted-task recovery is removed from the five-second shared queue + tick and driven as Office maintenance only when relevant Office + configuration exists. Routines, budgets, and heartbeat handlers retain + their own configuration checks. + +Every long-running scheduler owns its goroutine with idempotent `Start` and +`Stop` semantics. `Stop` cancels and waits for the active tick and handler +fan-out. Graceful shutdown quiesces these schedulers before stopping downstream +orchestrator/runtime services and before closing repositories or the database. +Shutdown completion is logged only after all owned loops have joined. + +This decision refines ADR-0004's generic-runs-queue direction; it does not +introduce an Office workspace process or use workflow style as a generic +dispatch discriminator. + +## Consequences + +- Multiple Office workspaces share one fair-by-request-time queue and do not + multiply ticker goroutines. +- Kanban tasks retain user/workflow-controlled launch behavior unless their + workflow explicitly emits `queue_run`. +- An Office-enabled installation with no Office adoption avoids repeated + Office task scans, though the shared queue's lightweight safety loop remains + available. +- Scheduler lifecycles become testable and shutdown can deterministically + prove that no database access follows pool closure. +- The existing transitional `office/service.SchedulerIntegration` must be + decomposed enough to move Office recovery out of its generic tick; a full + migration of all prompt/routing policy out of the Office package is not + required by this decision. +- Dynamically adding the first Office workflow requires no new per-workspace + scheduler: persisted configuration is observed by the next Office + maintenance pass. + +## Alternatives Considered + +### One scheduler per Office workspace + +Rejected because `runs` is a shared queue, per-agent serialization is global, +and workspace-local loops would duplicate timers and complicate fair claiming, +dynamic workspace deletion, and shutdown. + +### Start the existing combined scheduler only when an Office workspace exists + +Rejected because explicit `queue_run` is workflow-generic and because the +first Office workflow can be created while the backend is running. A startup +count would either miss new work until restart or require a lifecycle +controller solely to recreate a queue consumer. + +### Keep the combined global tick and accept idle scans + +Rejected because it conflates assignment with autonomy, causes needless +Office database work in Kanban-only installations, and leaves the shutdown +race unresolved. + +### Cancel the application context earlier without joining loops + +Rejected because cancellation alone does not prove that an in-flight tick has +returned before SQLite closes. Owned goroutines and a join are required. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index ed740f7335..ac9fa7eca5 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -95,3 +95,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 2026-07-31-isolate-manual-pr-review-content | [Isolate Manual PR Review Content](2026-07-31-isolate-manual-pr-review-content.md) | accepted | infra, workflow, security | 2026-07-31 | | 2026-07-31-system-service-user-continuity | [Preserve System Service Identity Across Reinstallation](2026-07-31-system-service-user-continuity.md) | accepted | backend, cli, security, operations | 2026-07-31 | | 2026-08-01-validate-mcp-tool-arguments | [Validate MCP Tool Arguments at the Shared Server Boundary](2026-08-01-validate-mcp-tool-arguments.md) | accepted | backend, protocol | 2026-08-01 | +| 2026-08-01-global-run-scheduler-ownership | [Separate Global Run Dispatch from Office Maintenance](2026-08-01-global-run-scheduler-ownership.md) | accepted | backend, workflow | 2026-08-01 | diff --git a/docs/plans/run-scheduler-lifecycle/plan.md b/docs/plans/run-scheduler-lifecycle/plan.md new file mode 100644 index 0000000000..a40944eb4b --- /dev/null +++ b/docs/plans/run-scheduler-lifecycle/plan.md @@ -0,0 +1,144 @@ +--- +spec: docs/specs/tasks/run-scheduling.md +created: 2026-08-01 +status: done +--- + +# Implementation Plan: Run Scheduler Lifecycle and Office Scoping + +## Overview + +The confirmed shutdown defect is caused by context cancellation running after +repository cleanup while the runs and cron loops are untracked. The change +first gives both loops owned, joinable lifecycles; then makes Office task +identity reusable and scopes autonomous assignment/recovery; then removes +Office recovery from the five-second queue tick; finally wires the owned loops +into graceful shutdown before SQLite cleanup. No schema, API, or frontend +change is required. + +--- + +## Backend + +### Owned scheduler lifecycles + +- Update `internal/runs/scheduler.Scheduler` to own its derived context, + goroutine, mutex, cancellation function, and wait group. `Start` and `Stop` + become idempotent; `Stop` waits for an in-progress `RunProcessor.Tick`. +- Apply the same lifecycle to `internal/scheduler/cron.Loop`. Its `Stop` waits + for `fanOut`, which already waits for every handler. +- Replace sleep-based shutdown assertions with channel synchronization and + `testing/synctest` where practical. + +### Authoritative Office-task scoping + +- Export the existing task-repository Office predicate so the task projection + and Office repository queries cannot drift. +- Extend `office/repository/sqlite.TaskExecutionFields` with + `IsFromOffice` and use the shared predicate in its query. +- Restrict `ListUnstartedTasks` to authoritative Office tasks while preserving + its lookback, runner, archive, and prior-run conditions. +- Make task-created/task-updated Office subscribers skip non-Office tasks even + when a runner is present. Explicit workflow-engine `queue_run` actions remain + unchanged and generic. + +### Separate Office maintenance from queue draining + +- Keep queued-run claim/dispatch, routing-unpark, and stale-claim recovery in + the shared run processor. +- Remove `recoverUnstartedTasks` from + `office/service.SchedulerIntegration.tick` and expose it through an Office + recovery handler driven by the shared cron loop. +- Add a cheap repository capability check for authoritative Office adoption so + the recovery handler skips the task scan when no workspace has an Office + workflow/project. Keep handler activation data-derived so adding the first + Office workflow works without restart or another scheduler goroutine. +- Give the processor/recovery loggers distinct `runs-processor` and + `office-recovery` component names so operational logs reflect ownership. + +### Graceful shutdown composition + +- Return an owned scheduling runtime from backend startup rather than + launching fire-and-forget goroutines. +- Thread its stop operation into `awaitShutdown` / `runGracefulShutdown` and + invoke it after HTTP intake closes but before `orchestratorSvc.Stop`, agent + runtime teardown, and `runCleanups`. +- Preserve the root application-context cleanup as a final fallback for other + components, but do not rely on its cleanup-stack position to stop database + users. +- Include scheduling stop failures in the graceful-shutdown error count and + emit the completion log only after both loops have joined. + +--- + +## Tests + +- **What:** duplicate `Start`/`Stop`, parent cancellation, and Stop waiting for + an active tick. + **File:** `apps/backend/internal/runs/scheduler/scheduler_test.go`. + **How:** channel-controlled fake processor plus `testing/synctest` for timer + behavior. +- **What:** cron Stop waits for all handler fan-out and does not tick after + return. + **File:** `apps/backend/internal/scheduler/cron/cron_test.go`. + **How:** blocking handlers coordinated by channels. +- **What:** Kanban tasks with runners are excluded while project-linked and + canonical Office-workflow tasks are recoverable across multiple workspaces. + **Files:** `apps/backend/internal/office/repository/sqlite/tasks_test.go`, + `apps/backend/internal/office/service/event_subscribers_test.go`, and a + focused Office recovery test beside `scheduler_recovery.go`. + **How:** real in-memory SQLite repositories and synchronous event handlers. +- **What:** no Office configuration skips the Office task scan, while adding + the first Office workflow activates the next recovery pass without restart. + **File:** `apps/backend/internal/office/service/scheduler_recovery_test.go`. + **How:** real repository with two maintenance passes before and after + persisted Office-project adoption. +- **What:** graceful shutdown joins database-using schedulers before the pool + cleanup and reports stop errors. + **File:** `apps/backend/internal/backendapp/helpers_test.go` or a focused + `shutdown_test.go` in the same package. + **How:** ordered fakes and a blocking scheduler; assert no operation occurs + after the fake database-close marker. + +Exact task-level commands are recorded in the task files below. + +--- + +## Implementation Waves And Parallel Candidates + +Wave 1 (parallel candidates; user authorization required): + +- [x] [task-01-owned-loop-lifecycles](task-01-owned-loop-lifecycles.md) +- [x] [task-02-office-task-scoping](task-02-office-task-scoping.md) + +Wave 2: + +- [x] [task-03-office-maintenance-separation](task-03-office-maintenance-separation.md) + +Wave 3: + +- [x] [task-04-shutdown-composition](task-04-shutdown-composition.md) + +The default execution order is sequential in the primary conversation. These +waves do not authorize subagents. + +## Risks + +- The Office scheduler spec predates the unified queue; implementation must + preserve current `runs` behavior rather than reintroducing the retired + `agent_wakeup_requests` design. +- Task identity must reuse the exact `Task.IsFromOffice` predicate. A subtly + different repository filter could either wake Kanban tasks or strand Office + project tasks. +- Stop must not hold a mutex while waiting for a goroutine that needs the same + mutex to finish. +- Moving recovery to the 30-second cron cadence intentionally increases the + worst-case repair latency for a missed assignment event while leaving normal + signal-driven queue latency unchanged. + +## Out of scope + +- Frontend or E2E changes; this has no user-interface contract change. +- Full migration of prompt assembly, routing, and launch policy out of + `internal/office/service`. +- Dynamic generic safety-timer/backoff redesign. diff --git a/docs/plans/run-scheduler-lifecycle/task-01-owned-loop-lifecycles.md b/docs/plans/run-scheduler-lifecycle/task-01-owned-loop-lifecycles.md new file mode 100644 index 0000000000..58ebfd5010 --- /dev/null +++ b/docs/plans/run-scheduler-lifecycle/task-01-owned-loop-lifecycles.md @@ -0,0 +1,68 @@ +--- +id: "01-owned-loop-lifecycles" +title: "Own scheduler loop lifecycles" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/tasks/run-scheduling.md" +--- + +# Task 01: Own scheduler loop lifecycles + +## Acceptance + +- `runs/scheduler.Scheduler` and `scheduler/cron.Loop` each start at most one + owned goroutine and expose idempotent stop behavior. +- Stop cancels future ticks and blocks until the currently executing processor + tick or cron handler fan-out has returned. +- Parent-context cancellation drains through the same wait-group path without + producing post-stop calls. + +## Verification + +```bash +cd apps/backend && go test ./internal/runs/scheduler ./internal/scheduler/cron +``` + +## Files likely touched + +- `apps/backend/internal/runs/scheduler/scheduler.go` +- `apps/backend/internal/runs/scheduler/scheduler_test.go` +- `apps/backend/internal/scheduler/cron/cron.go` +- `apps/backend/internal/scheduler/cron/cron_test.go` + +## Dependencies + +None. + +## Parallelism + +`parallel-safe` relative to Task 02 only: the file sets are disjoint and no +schema or shared contract file changes. + +## Inputs + +- Spec: scheduler lifecycle and shutdown scenarios. +- Plan: Owned scheduler lifecycles. +- Pattern: `internal/integrations/healthpoll.Poller` Start/Stop ownership. + +## Risks + +- Avoid waiting while holding the lifecycle mutex. +- Preserve the under-100ms signal-driven dispatch test. + +## Output contract + +Report the lifecycle behavior, files changed, test command/result, blockers, +remaining risks, and update this task plus `plan.md` status in the same +conversation. + +## Result + +- `Scheduler` and `cron.Loop` now own their loop goroutines and expose + idempotent `Start`/`Stop` methods; `Stop` waits for active processor ticks + and cron handler fan-out. +- Added channel-controlled tests for duplicate starts, active-work draining, + and parent-context cancellation. +- Verification: `cd apps/backend && go test -race ./internal/runs/scheduler ./internal/scheduler/cron -count=1` passed. diff --git a/docs/plans/run-scheduler-lifecycle/task-02-office-task-scoping.md b/docs/plans/run-scheduler-lifecycle/task-02-office-task-scoping.md new file mode 100644 index 0000000000..5a2ff88855 --- /dev/null +++ b/docs/plans/run-scheduler-lifecycle/task-02-office-task-scoping.md @@ -0,0 +1,75 @@ +--- +id: "02-office-task-scoping" +title: "Scope autonomous assignment to Office tasks" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/tasks/run-scheduling.md" +--- + +# Task 02: Scope autonomous assignment to Office tasks + +## Acceptance + +- Task and Office repositories share one authoritative Office-task SQL + predicate matching `Task.IsFromOffice`. +- Office task-created/task-updated subscribers never queue `task_assigned` for + a Kanban task solely because it has a runner. +- Recovery selects canonical Office-workflow and project-linked tasks across + multiple workspaces, while excluding ordinary Kanban tasks. + +## Verification + +```bash +cd apps/backend && go test ./internal/task/repository/sqlite ./internal/office/repository/sqlite ./internal/office/service +``` + +## Files likely touched + +- `apps/backend/internal/task/repository/sqlite/task.go` +- `apps/backend/internal/task/repository/sqlite/is_from_office_test.go` +- `apps/backend/internal/office/repository/sqlite/tasks.go` +- `apps/backend/internal/office/repository/sqlite/tasks_test.go` +- `apps/backend/internal/office/service/event_subscribers.go` +- `apps/backend/internal/office/service/event_subscribers_test.go` +- `apps/backend/internal/office/service/scheduler_recovery_test.go` + +## Dependencies + +None. + +## Parallelism + +`parallel-safe` relative to Task 01 only: the file sets are disjoint and no +schema, package configuration, or generated contract changes. + +## Inputs + +- Spec: autonomous assignment and Office identity scenarios. +- Plan: Authoritative Office-task scoping. +- Existing projection: `task/repository/sqlite.isFromOfficeProjection`. + +## Risks + +- Project-linked Office tasks may use a non-canonical workflow and must remain + included. +- Explicit engine `queue_run` actions for custom workflows must remain + unaffected. + +## Output contract + +Report the scoping behavior, files changed, test command/result, blockers, +remaining risks, and update this task plus `plan.md` status in the same +conversation. + +## Result + +- Exported the task repository's Office identity predicate and reused it in + Office execution-field and recovery queries. +- Task-created/task-updated assignment subscribers now ignore ordinary Kanban + tasks even when they have a runner; project-linked and canonical + Office-workflow tasks remain eligible. +- Added SQLite and event-subscriber coverage for both Office identity forms, + Kanban exclusion, and multi-workspace recovery selection. +- Verification: `cd apps/backend && go test -race ./internal/task/repository/sqlite ./internal/office/repository/sqlite ./internal/office/service -count=1` passed. diff --git a/docs/plans/run-scheduler-lifecycle/task-03-office-maintenance-separation.md b/docs/plans/run-scheduler-lifecycle/task-03-office-maintenance-separation.md new file mode 100644 index 0000000000..fb43b2ac2a --- /dev/null +++ b/docs/plans/run-scheduler-lifecycle/task-03-office-maintenance-separation.md @@ -0,0 +1,71 @@ +--- +id: "03-office-maintenance-separation" +title: "Separate Office recovery maintenance" +status: done +wave: 2 +depends_on: ["01-owned-loop-lifecycles", "02-office-task-scoping"] +plan: "plan.md" +spec: "../../specs/tasks/run-scheduling.md" +--- + +# Task 03: Separate Office recovery maintenance + +## Acceptance + +- The five-second run processor tick drains queued runs and performs generic + routing/stale-claim recovery without running the Office unstarted-task scan. +- Office unstarted-task recovery runs through the shared cron loop and skips + its task query when no authoritative Office workspace/project exists. +- Adding the first Office workflow/project is observed on the next maintenance + pass without restarting Kandev or creating a per-workspace goroutine. + +## Verification + +```bash +cd apps/backend && go test ./internal/office/service ./internal/scheduler/cron ./internal/backendapp +``` + +## Files likely touched + +- `apps/backend/internal/office/service/scheduler_integration.go` +- `apps/backend/internal/office/service/scheduler_recovery.go` +- `apps/backend/internal/office/service/scheduler_recovery_test.go` +- `apps/backend/internal/office/repository/sqlite/tasks.go` +- `apps/backend/internal/backendapp/cron.go` +- `apps/backend/internal/backendapp/main.go` + +## Dependencies + +- Task 01 provides the owned cron lifecycle. +- Task 02 provides authoritative Office-task identity and recovery filtering. + +## Parallelism + +`sequential`; it shares Office scheduler and backend wiring with its +dependencies and Task 04. + +## Inputs + +- Spec: Office-disabled/no-adoption and recovery scenarios. +- Plan: Separate Office maintenance from queue draining. +- Existing cron handler pattern: `internal/scheduler/cron.Handler`. + +## Risks + +- Normal queue signals must retain their existing sub-100ms path. +- Recovery moves from a five-second to a thirty-second maximum cadence; only + missed-event repair latency changes. + +## Output contract + +Report the maintenance split, files changed, test command/result, blockers, +remaining risks, and update this task plus `plan.md` status in the same +conversation. + +## Result + +- Removed unstarted-task recovery from the five-second runs processor tick. +- Added `OfficeRecoveryHandler` to the shared cron loop with a persisted + Office-adoption check, so Kanban-only installations skip the task scan and + newly adopted Office projects activate without restart. +- Verification: `cd apps/backend && go test -race ./internal/office/service ./internal/scheduler/cron ./internal/backendapp -count=1` passed. diff --git a/docs/plans/run-scheduler-lifecycle/task-04-shutdown-composition.md b/docs/plans/run-scheduler-lifecycle/task-04-shutdown-composition.md new file mode 100644 index 0000000000..b2be36db0d --- /dev/null +++ b/docs/plans/run-scheduler-lifecycle/task-04-shutdown-composition.md @@ -0,0 +1,77 @@ +--- +id: "04-shutdown-composition" +title: "Stop schedulers before database cleanup" +status: done +wave: 3 +depends_on: ["01-owned-loop-lifecycles", "03-office-maintenance-separation"] +plan: "plan.md" +spec: "../../specs/tasks/run-scheduling.md" +--- + +# Task 04: Stop schedulers before database cleanup + +## Acceptance + +- Backend startup retains owned handles for the runs scheduler and cron loop; + no scheduler is launched as an untracked `go Start(ctx)` call. +- Graceful shutdown stops and joins both loops before orchestrator/runtime + teardown and before repository/database cleanup. +- Regression coverage proves an in-progress database-backed tick finishes + before the close marker, scheduler stop errors affect `error_count`, and no + scheduler log can appear after the completion log. + +## Verification + +```bash +cd apps/backend && go test ./internal/backendapp ./internal/runs/scheduler ./internal/scheduler/cron ./internal/office/service +``` + +## Files likely touched + +- `apps/backend/internal/backendapp/main.go` +- `apps/backend/internal/backendapp/cron.go` +- `apps/backend/internal/backendapp/helpers.go` +- `apps/backend/internal/backendapp/types.go` +- `apps/backend/internal/backendapp/helpers_test.go` +- `apps/backend/internal/backendapp/shutdown_test.go` + +## Dependencies + +- Task 01 provides owned Stop-and-wait loops. +- Task 03 provides the final run/cron composition returned from startup. + +## Parallelism + +`sequential`; this is the integration task and owns shared backend startup and +shutdown composition. + +## Inputs + +- Spec: failure modes and shutdown scenarios. +- Plan: Graceful shutdown composition. +- Confirmed root cause: cleanup callbacks run in reverse registration order, + while root context cancellation was registered before database cleanup and + therefore executed after it. + +## Risks + +- Stop schedulers before stopping the orchestrator they dispatch into. +- Preserve startup-failure cleanup through the existing root-context fallback. +- Do not close the logger before final scheduler-stop diagnostics are emitted. + +## Output contract + +Report the shutdown ordering, files changed, test command/result, blockers, +remaining risks, and update this task plus `plan.md` status in the same +conversation. + +## Result + +- Backend startup now retains an owned scheduling runtime for the runs and + cron loops, with an idempotent cleanup fallback for startup failures. +- Graceful shutdown stops and joins cron/run scheduling before the + orchestrator, lifecycle manager, and database cleanup; scheduler stop errors + are included in `error_count`. +- Added a blocking shutdown regression proving database cleanup cannot run + before scheduler stop completes. +- Verification: `cd apps/backend && go test -race ./internal/backendapp ./internal/runs/scheduler ./internal/scheduler/cron ./internal/office/service -count=1` passed. diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index e6af89b8b8..f24be77b94 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -54,6 +54,7 @@ Kandev's task model: documents, execution stages, labels, blocker escalation, su | [labels](tasks/labels.md) | shipped | | [title-length-limit](tasks/title-length-limit.md) | complete | | [model-unification](tasks/model-unification.md) | draft | +| [run-scheduling](tasks/run-scheduling.md) | building | | [without-repositories](tasks/without-repositories.md) | draft | | [attach-workspace-sources](tasks/attach-workspace-sources.md) | building | | [subtask-checklist](tasks/subtask-checklist.md) | shipped | diff --git a/docs/specs/office/scheduler.md b/docs/specs/office/scheduler.md index 36855d2e8c..e778d618f8 100644 --- a/docs/specs/office/scheduler.md +++ b/docs/specs/office/scheduler.md @@ -10,10 +10,23 @@ owner: cfl Kandev's base task scheduler is reactive: tasks enter the queue only when a user explicitly starts them or sends a prompt. Office adds autonomous agent operation, which requires the system to wake agents on its own when events happen (assignments, comments, blocker resolutions, approvals), on a schedule (routines), and on heartbeat ticks (periodic coordinator checks). Without an autonomous wakeup pipeline, every interaction needs a human to initiate it, and the cost / reliability story (idle skips, rate-limit retries, staleness, recovery) has nowhere to live. -The office scheduler is the single seam for all autonomous wakeups: a SQLite-persisted wakeup queue with coalescing and idempotency, a claim/dispatch loop that creates one-shot agent runs, a routines table that drives all periodic and webhook-triggered wakes, plus the reliability primitives (rate-limit retry, idle skip, staleness check, recovery sweep) that keep the system honest at long uptime. +Office supplies autonomous run producers and Office-specific maintenance. The persisted `runs` queue and its single backend-wide consumer are shared workflow infrastructure, not one scheduler per Office workspace. The shared ownership, scoping, and shutdown contract is defined by [queued run scheduling](../tasks/run-scheduling.md) and [ADR-2026-08-01-global-run-scheduler-ownership](../../decisions/2026-08-01-global-run-scheduler-ownership.md). ## What +### Shared scheduler boundary + +- One shared runs scheduler processes persisted work for every workspace. +- Office event subscribers and unstarted-task recovery act only on tasks whose + project/workflow identity makes `Task.IsFromOffice` true. A runner on an + ordinary Kanban task is not sufficient. +- Explicit workflow `queue_run` actions remain available to every workflow + style and are not filtered by Office identity. +- Office recovery is maintenance, not part of every five-second queue drain. + When no workspace has adopted Office, it skips the task scan. +- The shared runs scheduler and cron loop stop and join before database cleanup + during graceful shutdown. + ### Wakeup queue A SQLite-persisted queue of "wake this agent up" requests. Every periodic, event-driven, and reactive trigger flows through this queue before becoming an agent run. Each request: @@ -32,7 +45,7 @@ A SQLite-persisted queue of "wake this agent up" requests. Every periodic, event | `agent_error` | A sub-agent's session failed (escalation to coordinator) | `{agent_profile_id, run_id, error}` | | `self` | Agent self-wake via tool call | `{reason, payload?}` | | `user` | User mention / explicit wake from the UI | `{user_id, context?}` | -| `task_assigned` | Task's `assignee_agent_instance_id` is set or changed, including a newly-created task/subtask that already has a runner | `{task_id}` | +| `task_assigned` | An authoritative Office task's `assignee_agent_instance_id` is set or changed, including a newly-created task/subtask that already has a runner | `{task_id}` | | `task_blockers_resolved` | All blocking tasks of an assigned task reach `done` | `{task_id, resolved_blocker_ids}` | | `task_children_completed` | All child tasks of an assigned task reach terminal state | `{task_id}` | | `approval_resolved` | An approval requested by this agent is approved/rejected | `{approval_id, status, decision_note}` | @@ -226,9 +239,9 @@ Log fields gain `source: "rate_limit_parsed"` vs `source: "backoff"`, plus `pars **Default backoff for non-rate-limit retries**: 4 attempts at `[2m, 10m, 30m, 2h]` with 25% jitter. After `MaxRetryCount` (4) failures, `escalateFailure` is called - the wakeup is marked `failed`, an `agent.error` inbox item is created, and the coordinator receives an `agent_error` wakeup. -### Recovery sweep (unstarted tasks) +### Recovery sweep (unstarted Office tasks) -The scheduler tick is extended with a recovery sweep that runs once per tick after the wakeup drain. It finds assigned `TODO` tasks with no prior queued or running wakeup and dispatches them as `task_assigned` wakeups. +Office maintenance performs a recovery sweep separately from the shared queue-drain tick. It finds authoritative Office `TODO` tasks created inside the workspace recovery lookback window and dispatches them as `task_assigned` runs only when no queued, claimed, or finished run exists for the task. The task-creation timestamp is bounded by the lookback; matching run rows are not, so a task that already started is never reclassified as unstarted merely because its prior run is old. Failed and cancelled rows do not block recovery. Assignment on an ordinary Kanban task does not imply autonomy. Selection: @@ -236,12 +249,18 @@ Selection: SELECT t.id FROM tasks t WHERE t.state = 'TODO' AND t.assignee_agent_instance_id IS NOT NULL + AND ( + COALESCE(t.project_id, '') != '' + OR t.workflow_id = ( + SELECT w.office_workflow_id FROM workspaces w WHERE w.id = t.workspace_id + ) + ) AND t.archived_at IS NULL AND t.created_at >= NOW() - INTERVAL ' hours' AND NOT EXISTS ( - SELECT 1 FROM wakeup_requests w - WHERE w.payload->>'task_id' = t.id - AND w.status IN ('queued', 'claimed', 'finished') + SELECT 1 FROM runs r + WHERE r.payload->>'task_id' = t.id + AND r.status IN ('queued', 'claimed', 'finished') ) ``` @@ -382,7 +401,7 @@ Before launching an agent session, the scheduler checks all applicable budget po ### Integration with existing scheduler -The base `scheduler.Scheduler` and `queue.TaskQueue` continue to handle user-initiated task execution unchanged. The wakeup queue is a parallel path: same scheduler tick loop, different queue, different processing logic. Both paths converge at the lifecycle manager - the same `LaunchAgent()` / `StartAgentProcess()` calls regardless of whether the session was user-initiated or wakeup-initiated. +User-initiated task execution continues through the orchestrator directly. Engine-emitted work uses the single shared `runs` scheduler across all workspaces. Office contributes event-driven producers and maintenance handlers to that shared path; it does not own a per-workspace scheduler. Both paths converge at the agent runtime. ## Data model @@ -625,6 +644,7 @@ A formal per-route authorization model (workspace membership, admin role, RBAC o |---|---| | SQLite write failure during enqueue | Source caller surfaces error; idempotency-key collision is silent (treated as already-enqueued). | | Claim query returns 0 rows | Another process won the claim or no eligible wakeup; tick exits cleanly. | +| Office is enabled with no Office workflow/project | Office recovery skips its task scan; ordinary Kanban assignments are not launched. | | Agent paused / stopped at claim | Wakeup marked `finished` with no action; not retried. | | Task referenced by payload is missing | Staleness check cancels with `task_not_found`; activity logged. | | Task assignee changed before claim | Cancelled with `assignee_changed`. | @@ -640,12 +660,13 @@ A formal per-route authorization model (workspace membership, admin role, RBAC o | Coordinator's pre-installed routine fails to install at onboarding | Logged + warned; coordinator's agent detail UI shows "no scheduled wake-ups" empty state. User can install one manually. | | Routine cron tick missed (scheduler down) | `enqueue_missed_with_cap` (default): fire missed ticks up to cap=25 with "missed N ticks" in next prompt context. `skip_missed`: fire current tick only. | | Webhook signature verification fails | Trigger rejected with 401; no routine run created. | +| Graceful shutdown begins | Queue and cron loops cancel and join before repositories and SQLite close; no scheduler logs `database is closed`. | ## Persistence guarantees Survives a kandev process restart: all `agent_wakeup_requests` rows including `queued`, `claimed` (re-claimed on restart), and `scheduled_retry_at`; all `office_routines`, `office_routine_triggers` (with `next_run_at` advanced), `office_routine_runs` history; `agent_continuation_summaries` (last-good); `runs` rows including `result_json`, `assembled_prompt`, `summary_injected` snapshots. -Does NOT survive (reconstructed on next tick): in-memory claim leases - a `claimed` wakeup whose process died is picked up by the staleness/recovery path; the scheduler's claim query is the source of truth. The recovery sweep's idempotency is via the `NOT EXISTS` check on `wakeup_requests` plus the dispatched wakeup's `idempotency_key`. +Does NOT survive (reconstructed on next tick): in-memory claim leases - a `claimed` wakeup whose process died is picked up by the staleness/recovery path; the scheduler's claim query is the source of truth. The unstarted-task recovery sweep suppresses duplicates with its `NOT EXISTS` check on `runs`: any queued, claimed, or finished run of any age blocks redispatch, while failed and cancelled runs do not. Retention: idempotency-key dedup window 24 hours; summary cap 8 KB per row; routine run history retained for inspection (no automatic prune in scope here); catch-up cap (default 25) drops missed routine ticks beyond it (not recorded individually). @@ -653,7 +674,10 @@ The scheduler reads all `queued` and unexpired-retry wakeup requests on boot and ## Scenarios -- **GIVEN** a task is assigned to a worker agent instance, **WHEN** the assignment is saved, **THEN** a `task_assigned` wakeup is queued for that agent. The scheduler claims it, creates a session, and the agent starts working on the task. +- **GIVEN** an authoritative Office task is assigned to a worker agent instance, **WHEN** the assignment is saved, **THEN** a `task_assigned` wakeup is queued for that agent. The scheduler claims it, creates a session, and the agent starts working on the task. +- **GIVEN** Office mode is enabled but a workspace contains only Kanban workflows, **WHEN** a Kanban task with a runner remains in `TODO`, **THEN** Office subscribers and recovery do not queue a `task_assigned` run for it. + +- **GIVEN** Kandev is shutting down while the shared runs scheduler or cron loop is active, **WHEN** graceful shutdown reaches database cleanup, **THEN** those loops have already stopped and joined and no `sql: database is closed` scheduler error is emitted. - **GIVEN** a worker agent is currently running a session (at capacity), **WHEN** a `comment` wakeup arrives for the same agent, **THEN** the wakeup stays in `queued` status. When the current session completes, the next scheduler tick picks up the wakeup and the agent processes the comment. @@ -689,7 +713,7 @@ The scheduler reads all `queued` and unexpired-retry wakeup requests on boot and - **GIVEN** a wakeup for task T fails and a retry is scheduled 10 minutes out, **WHEN** task T is reassigned (or cancelled) before the retry fires, **THEN** the retry is cancelled at promotion time with reason `retry_stale_assignee` (or `retry_task_cancelled`), execution locks are cleared, and a `wakeup_retry_cancelled` activity entry is logged. A PATCH reassign on the API cancels pending retries for the previous assignee immediately. -- **GIVEN** a task in `TODO` state assigned to agent A has no queued or finished wakeup and was created within the lookback window, **WHEN** the recovery sweep runs, **THEN** a `task_assigned` wakeup is dispatched and a `recovery_dispatch` activity entry is logged. Tasks that already have a queued/finished wakeup, or that fall outside `recovery_lookback_hours`, are skipped. +- **GIVEN** an authoritative Office task in `TODO` state assigned to agent A has no queued, claimed, or finished run and was created within the lookback window, **WHEN** the recovery sweep runs, **THEN** a `task_assigned` run is dispatched and a `recovery_dispatch` activity entry is logged. Ordinary Kanban tasks, tasks with a queued, claimed, or finished run of any age, and tasks created outside `recovery_lookback_hours` are skipped. - **GIVEN** a wakeup for a task that has reached `DONE` state, **WHEN** the staleness check runs at claim time, **THEN** the wakeup is cancelled with reason `task_terminal` and the agent is not launched. diff --git a/docs/specs/tasks/model-unification.md b/docs/specs/tasks/model-unification.md index 992615ba10..db131eb8e0 100644 --- a/docs/specs/tasks/model-unification.md +++ b/docs/specs/tasks/model-unification.md @@ -178,6 +178,15 @@ User-initiated launches (clicking Start on a kanban task) bypass the queue and call `runtime.Launch` directly — those don't need coalescing and shouldn't pay the scheduler's tick latency. +The queue has one backend-wide consumer across all workspaces; there +is no scheduler per Office workspace. Assignment alone does not make +an ordinary Kanban task autonomous. Office assignment recovery uses +the authoritative Office-task identity, while a workflow of any style +can opt into queued automation explicitly with `queue_run`. Shared +scheduler lifecycle, Office maintenance activation, and shutdown +ordering are specified in [queued run scheduling](run-scheduling.md) +and [ADR-2026-08-01-global-run-scheduler-ownership](../../decisions/2026-08-01-global-run-scheduler-ownership.md). + `run_events` (the audit log per run) keeps its current shape and event-streaming behaviour. WS event subjects (`office.run.queued`, `office.run.processed`, `office.run.event_appended.`) keep @@ -296,6 +305,12 @@ preserved. runs through the linear steps as today. No regression. No new triggers fire because the kanban template doesn't configure them. +- **GIVEN** Office mode is enabled but every workspace contains only + ordinary Kanban workflows, **WHEN** scheduler maintenance runs, + **THEN** no Kanban assignment is recovered as autonomous Office + work. Explicit user Start and configured Kanban workflow actions + retain their existing behavior. + - **GIVEN** a workspace that adopts the Office Default workflow, **WHEN** the user creates a task assigned to the CEO, **THEN** the task lands in the `Work` step. The CEO runs `on_enter`. When the diff --git a/docs/specs/tasks/run-scheduling.md b/docs/specs/tasks/run-scheduling.md new file mode 100644 index 0000000000..261c701e98 --- /dev/null +++ b/docs/specs/tasks/run-scheduling.md @@ -0,0 +1,148 @@ +--- +status: building +created: 2026-08-01 +owner: cfl +--- + +# Queued run scheduling + +## Why + +Workflow actions and Office automation can enqueue agent work without a user +clicking Start. Users need that work to be dispatched reliably across multiple +workspaces without making ordinary Kanban assignment autonomous, wasting +resources on Office-only scans when Office is unused, or producing database +errors during a normal shutdown. + +## What + +- Kandev has one backend-wide scheduler for the persisted `runs` queue. It + serves every workspace and never creates one scheduler goroutine per + workspace. +- Runs are claimed globally in request order while preserving the existing + per-agent serialization rule. A busy workspace does not create another + scheduler or bypass agent capacity checks. +- An explicit `queue_run` workflow action may enqueue work from any workflow + style. User-initiated Kanban launches continue to bypass `runs` and launch + through the interactive orchestrator path. +- Merely assigning a runner to an ordinary Kanban task does not make that task + autonomous. Office assignment subscribers and unstarted-task recovery act + only on authoritative Office tasks: tasks linked to an Office project or + tasks whose workflow is the workspace's canonical `office_workflow_id`. +- Office-only maintenance is separate from queue draining. Unstarted-task + recovery, Office routines, and Office budget work run only when their + corresponding Office configuration exists. A deployment with Office enabled + but no configured Office workflow performs no five-second Office task + recovery scan. +- The queue scheduler remains event-driven: an in-process queue signal wakes it + immediately, while a periodic safety pass recovers persisted work after a + missed signal or process restart. +- Every scheduler or cron loop owns the goroutine it starts. `Stop` is + idempotent, cancels future ticks, and waits for an in-progress tick and any + handler fan-out to return. +- Graceful shutdown stops external intake, quiesces and joins queue/cron + schedulers, stops the orchestrator and agent runtime, and only then closes + event-bus, repository, and database resources. +- A normal signal-driven shutdown emits no `sql: database is closed` scheduler + errors and does not log scheduler activity after `Graceful shutdown + complete`. + +Decision: [ADR-2026-08-01-global-run-scheduler-ownership](../../decisions/2026-08-01-global-run-scheduler-ownership.md). + +Implementation plan: [run-scheduler-lifecycle](../../plans/run-scheduler-lifecycle/plan.md). + +## Data model + +No schema change is required. + +- `runs` remains the durable, backend-wide queue and run record. +- `workspaces.office_workflow_id` plus a non-empty task `project_id` remain the + authoritative persisted Office-task identity used by `Task.IsFromOffice`. +- Office routines, budget policies, and related configuration remain + workspace-scoped in their existing tables. + +## State machine + +### Scheduler lifecycle + +```text +stopped -> running -> stopping -> stopped +``` + +- `Start` moves `stopped -> running` and owns exactly one loop goroutine. +- A duplicate `Start` while running is a no-op. +- `Stop` moves `running -> stopping`, cancels the loop, waits for the active + tick to drain, then moves to `stopped`. +- A duplicate `Stop` is a no-op. +- Parent-context cancellation follows the same drain path as `Stop`. + +### Persisted run lifecycle + +The existing `queued`, `claimed`, and terminal run states are unchanged. +Process shutdown does not create a new run state and does not mark queued work +failed. A claimed run interrupted by process exit is recovered through the +existing stale-claim policy after restart. + +## Failure modes + +| Condition | Required behavior | +|---|---| +| Queue signal is missed | The periodic safety pass eventually claims the persisted row. | +| No eligible run exists | The scheduler returns to waiting without an error. | +| Office is enabled but no Office workflow/configuration exists | Generic queue recovery remains available; Office-only recovery and cron work do not scan ordinary Kanban tasks. | +| A Kanban task has a runner but no explicit `queue_run` action | Office subscribers and recovery ignore it; the user or Kanban workflow controls launch timing. | +| Shutdown begins while the scheduler is idle | `Stop` cancels the wait and joins before the database closes. | +| Shutdown begins during a tick | Shutdown cancels future ticks and waits for the active tick and cron handler fan-out to return before repository cleanup. The backend does not impose an internal join deadline. | +| A handler does not return after cancellation | Graceful shutdown remains in progress and repository cleanup does not begin. The launcher or process supervisor may enforce its outer grace period and terminate the process, but the backend never closes SQLite underneath live scheduler work. | + +## Persistence guarantees + +- Queued and scheduled-retry runs survive restart. +- Shutdown never deletes or terminally fails a queued run merely because the + process is exiting. +- Scheduler ownership state, contexts, timers, and in-memory queue signals do + not survive restart; persisted rows are the recovery source of truth. +- Office activation state is derived from persisted workspace/workflow and + Office configuration rather than an in-memory workspace list. + +## Scenarios + +- **GIVEN** two workspaces have queued runs for different agents, **WHEN** the + global scheduler drains the queue, **THEN** it processes both workspaces in + request order while allowing at most one claimed run per agent. +- **GIVEN** Office mode is enabled and no workspace has an Office workflow, + Office project, active Office routine, or Office budget policy, **WHEN** the + backend remains idle, **THEN** no Office unstarted-task recovery query runs + on the five-second queue cadence and no task is launched. +- **GIVEN** an ordinary Kanban `TODO` task has a step runner, **WHEN** Office + assignment subscribers and recovery execute, **THEN** no `task_assigned` run + is created for that task. +- **GIVEN** an authoritative Office `TODO` task has a runner, was created inside + the recovery lookback window, and has no prior queued, claimed, or finished + run, **WHEN** Office recovery executes, **THEN** exactly one idempotent + `task_assigned` run is queued. +- **GIVEN** a custom non-Office workflow explicitly executes `queue_run`, + **WHEN** the row is persisted and signalled, **THEN** the global scheduler + processes it even though the task is not an Office task. +- **GIVEN** the backend receives SIGINT while all schedulers are idle, **WHEN** + graceful shutdown completes, **THEN** the database closes after every + scheduler has stopped and the log contains no `database is closed` error. +- **GIVEN** a scheduler tick or cron handler is blocked inside database-backed + work, **WHEN** graceful shutdown begins, **THEN** shutdown waits for that work + to return before closing the database and prints no scheduler-stopping log + after the completion message. +- **GIVEN** queued runs exist when the process exits, **WHEN** Kandev starts + again, **THEN** the periodic safety pass resumes them without requiring an + Office workspace to own a separate scheduler. + +## Out of scope + +- Multiple active backend processes or distributed leader election. +- Per-workspace scheduler goroutines. +- Changing global FIFO ordering or adding workspace fairness/quotas. +- Renaming the existing `office.run.*` WebSocket subjects. +- Changing retry counts, retry backoff, or agent cooldown policy. +- Adding a user-facing scheduler status page or new feature toggle. +- Replacing the five-second generic safety cadence with dynamic retry timers; + this change removes Office-only work from that cadence but preserves its + existing queue-recovery contract.