fix: prevent scheduler database access during shutdown (#2135)

This commit is contained in:
Carlos Florêncio
2026-08-01 20:11:04 +01:00
committed by GitHub
parent 1d70a0df6d
commit 1e353fbbbc
35 changed files with 1862 additions and 102 deletions
+6 -4
View File
@@ -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(
+15 -4
View File
@@ -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.
+84 -58
View File
@@ -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)
}
@@ -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
}
@@ -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)
}
}
+32
View File
@@ -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...)
}
@@ -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
@@ -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 /
@@ -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')
@@ -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()
@@ -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
}
@@ -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 == "" {
@@ -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()
@@ -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
}
@@ -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))
}
}
@@ -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)
}
}
@@ -0,0 +1,11 @@
package scheduler
import (
"testing"
"go.uber.org/goleak"
)
func TestMain(m *testing.M) {
goleak.VerifyTestMain(m)
}
@@ -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
@@ -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")
}
}
}
@@ -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")
}
}
+56 -5
View File
@@ -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))
@@ -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 {
@@ -0,0 +1,11 @@
package cron
import (
"testing"
"go.uber.org/goleak"
)
func TestMain(m *testing.M) {
goleak.VerifyTestMain(m)
}
@@ -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).
@@ -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.
+1
View File
@@ -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 |
+144
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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.
@@ -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.
+1
View File
@@ -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 |
+35 -11
View File
@@ -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 '<lookback_hours> 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.
+15
View File
@@ -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.<run_id>`) 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
+148
View File
@@ -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.