fix(office): make idle-skip gate actually skip periodic routine wakeups (#2973)
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
This commit is contained in:
@@ -394,7 +394,7 @@ func (s *RoutineService) processCronTrigger(ctx context.Context, trigger *Routin
|
||||
// runCount-1 as missed_ticks ("you missed N since the last fire").
|
||||
missedForPayload = runCount - 1
|
||||
}
|
||||
_, err = s.DispatchRoutineRunWithMissed(ctx, routine, trigger, "cron", nil, missedForPayload)
|
||||
_, err = s.DispatchRoutineRunWithMissed(ctx, routine, trigger, shared.RoutineSourceCron, nil, missedForPayload)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -621,7 +621,7 @@ func (s *RoutineService) materialiseLightweightRoutineRun(
|
||||
ID: uuid.New().String(),
|
||||
AgentProfileID: routine.AssigneeAgentProfileID,
|
||||
Source: "routine",
|
||||
Reason: "routine_dispatch",
|
||||
Reason: shared.RoutineDispatchReason(run.Source),
|
||||
Payload: payloadStr,
|
||||
IdempotencyKey: idemKey,
|
||||
RequestedAt: time.Now().UTC(),
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/repository/sqlite"
|
||||
"github.com/kandev/kandev/internal/office/routines"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
taskservice "github.com/kandev/kandev/internal/task/service"
|
||||
)
|
||||
|
||||
@@ -265,8 +266,12 @@ func TestDispatch_LightweightRoutine_EnqueuesWakeup(t *testing.T) {
|
||||
if got.Source != "routine" {
|
||||
t.Errorf("source = %q, want routine", got.Source)
|
||||
}
|
||||
if got.Reason != "routine_dispatch" {
|
||||
t.Errorf("reason = %q, want routine_dispatch", got.Reason)
|
||||
// FireManual is event/user-triggered, not periodic (WO-46 Review round
|
||||
// 1, Blocker 1): it must get RunReasonRoutineDispatchEvent, never the
|
||||
// cron-only RunReasonRoutineDispatch the idle-skip gate treats as a
|
||||
// skippable periodic wake.
|
||||
if got.Reason != shared.RunReasonRoutineDispatchEvent {
|
||||
t.Errorf("reason = %q, want %q", got.Reason, shared.RunReasonRoutineDispatchEvent)
|
||||
}
|
||||
if !strings.Contains(got.Payload, `"routine_id":"`+routine.ID+`"`) {
|
||||
t.Errorf("payload missing routine_id, got %q", got.Payload)
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
package routines_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/routines"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
)
|
||||
|
||||
// newLightweightTestRoutine builds an active, taskless routine assigned
|
||||
// to agent-1. Shared setup for both halves of the WO-46 anti-drift lock
|
||||
// below.
|
||||
func newLightweightTestRoutine(t *testing.T, svc *routines.RoutineService) *models.Routine {
|
||||
t.Helper()
|
||||
routine := &models.Routine{
|
||||
WorkspaceID: "ws-1",
|
||||
Name: "Lightweight",
|
||||
TaskTemplate: "", // lightweight
|
||||
AssigneeAgentProfileID: "agent-1",
|
||||
Status: "active",
|
||||
ConcurrencyPolicy: "always_create",
|
||||
}
|
||||
if err := svc.CreateRoutine(context.Background(), routine); err != nil {
|
||||
t.Fatalf("create routine: %v", err)
|
||||
}
|
||||
return routine
|
||||
}
|
||||
|
||||
// TestDispatch_LightweightRoutine_ManualFireIsNotPeriodicTasklessWake is
|
||||
// half of the WO-46 anti-drift lock (Review round 1, Blocker 2). A manual
|
||||
// "Fire now" is event/user-triggered, not periodic — per
|
||||
// docs/specs/office/scheduler.md ("Event-triggered wakeups always
|
||||
// proceed - the skip applies only to periodic wakes") the idle-skip gate
|
||||
// must never swallow it. Drives the reason through the real writer
|
||||
// (materialiseLightweightRoutineRun via FireManual) so a future change
|
||||
// that makes FireManual periodic-tagged again fails here first — this is
|
||||
// the case Review round 1 found: the original version of this test
|
||||
// asserted the opposite of what production behavior should be.
|
||||
func TestDispatch_LightweightRoutine_ManualFireIsNotPeriodicTasklessWake(t *testing.T) {
|
||||
svc := newTestRoutineService(t)
|
||||
ctx := context.Background()
|
||||
|
||||
routine := newLightweightTestRoutine(t, svc)
|
||||
|
||||
enq := &fakeWakeupEnqueuer{}
|
||||
svc.SetWakeupEnqueuer(enq)
|
||||
|
||||
if _, err := svc.FireManual(ctx, routine.ID, map[string]string{"name": "alpha"}); err != nil {
|
||||
t.Fatalf("fire manual: %v", err)
|
||||
}
|
||||
if len(enq.created) != 1 {
|
||||
t.Fatalf("expected 1 wakeup-request created, got %d", len(enq.created))
|
||||
}
|
||||
|
||||
got := enq.created[0].Reason
|
||||
if shared.IsPeriodicTasklessWake(got) {
|
||||
t.Errorf("shared.IsPeriodicTasklessWake(%q) = true, want false — "+
|
||||
"a manual Fire-now would be silently swallowed by the idle-skip "+
|
||||
"gate for any SkipIdleRuns assignee with no actionable tasks", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatch_LightweightRoutine_CronFireIsPeriodicTasklessWake is the
|
||||
// other half of the WO-46 anti-drift lock. A cron-driven fire is the
|
||||
// only routine trigger that represents a periodic, unattended wake —
|
||||
// the class the idle-skip gate exists to catch. Drives the reason
|
||||
// through the real cron path (TickScheduledTriggers →
|
||||
// materialiseLightweightRoutineRun) rather than asserting against a
|
||||
// copied literal, so the writer and shared.IsPeriodicTasklessWake
|
||||
// cannot silently diverge again the way RunReasonHeartbeat did.
|
||||
func TestDispatch_LightweightRoutine_CronFireIsPeriodicTasklessWake(t *testing.T) {
|
||||
svc := newTestRoutineService(t)
|
||||
ctx := context.Background()
|
||||
|
||||
routine := newLightweightTestRoutine(t, svc)
|
||||
if err := svc.CreateRoutineTrigger(ctx, &models.RoutineTrigger{
|
||||
RoutineID: routine.ID,
|
||||
Kind: "cron",
|
||||
CronExpression: "* * * * *",
|
||||
Timezone: "UTC",
|
||||
Enabled: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("create trigger: %v", err)
|
||||
}
|
||||
|
||||
enq := &fakeWakeupEnqueuer{}
|
||||
svc.SetWakeupEnqueuer(enq)
|
||||
|
||||
// The trigger's next_run_at is the next full minute after creation;
|
||||
// ticking 2 minutes out guarantees it is due.
|
||||
if err := svc.TickScheduledTriggers(ctx, time.Now().UTC().Add(2*time.Minute)); err != nil {
|
||||
t.Fatalf("tick scheduled triggers: %v", err)
|
||||
}
|
||||
if len(enq.created) != 1 {
|
||||
t.Fatalf("expected 1 wakeup-request created, got %d", len(enq.created))
|
||||
}
|
||||
|
||||
got := enq.created[0].Reason
|
||||
if !shared.IsPeriodicTasklessWake(got) {
|
||||
t.Errorf("shared.IsPeriodicTasklessWake(%q) = false, want true — "+
|
||||
"the idle-skip gate would silently stop recognizing production "+
|
||||
"cron-triggered routine-dispatch runs", got)
|
||||
}
|
||||
}
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/office/repository/sqlite"
|
||||
"github.com/kandev/kandev/internal/office/routing"
|
||||
"github.com/kandev/kandev/internal/office/service"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
)
|
||||
|
||||
// ErrRoutingNotSupported is returned by TaskStarter.StartTaskWithRoute
|
||||
@@ -61,9 +62,12 @@ const (
|
||||
RunReasonTaskChildrenCompleted = "task_children_completed"
|
||||
RunReasonApprovalResolved = "approval_resolved"
|
||||
RunReasonRoutineTrigger = "routine_trigger"
|
||||
RunReasonHeartbeat = "heartbeat"
|
||||
RunReasonBudgetAlert = "budget_alert"
|
||||
RunReasonAgentError = "agent_error"
|
||||
// RunReasonHeartbeat aliases shared.RunReasonHeartbeat so this package's
|
||||
// local constant and the shared idle-skip classifier cannot drift apart
|
||||
// the way the un-aliased pair did before WO-46 (Review round 1, S2).
|
||||
RunReasonHeartbeat = shared.RunReasonHeartbeat
|
||||
RunReasonBudgetAlert = "budget_alert"
|
||||
RunReasonAgentError = "agent_error"
|
||||
|
||||
// Reactivity-pipeline reasons.
|
||||
RunReasonTaskUnblocked = "task_unblocked" // status: blocked → not blocked
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/events"
|
||||
"github.com/kandev/kandev/internal/events/bus"
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
"github.com/kandev/kandev/internal/runs/commentkeys"
|
||||
runsservice "github.com/kandev/kandev/internal/runs/service"
|
||||
)
|
||||
@@ -24,9 +25,12 @@ const (
|
||||
RunReasonTaskChildrenCompleted = "task_children_completed"
|
||||
RunReasonApprovalResolved = "approval_resolved"
|
||||
RunReasonRoutineTrigger = "routine_trigger"
|
||||
RunReasonHeartbeat = "heartbeat"
|
||||
RunReasonBudgetAlert = "budget_alert"
|
||||
RunReasonAgentError = "agent_error"
|
||||
// RunReasonHeartbeat aliases shared.RunReasonHeartbeat so this package's
|
||||
// local constant and the shared idle-skip classifier cannot drift apart
|
||||
// the way the un-aliased pair did before WO-46 (Review round 1, S2).
|
||||
RunReasonHeartbeat = shared.RunReasonHeartbeat
|
||||
RunReasonBudgetAlert = "budget_alert"
|
||||
RunReasonAgentError = "agent_error"
|
||||
)
|
||||
|
||||
// These reasons were persisted by earlier workflow templates. Keep them
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
officeruntime "github.com/kandev/kandev/internal/office/runtime"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
@@ -766,12 +767,12 @@ func isAgentActive(status models.AgentStatus) bool {
|
||||
}
|
||||
|
||||
// checkIdleSkip returns true if the run should be skipped because the agent
|
||||
// is configured to skip idle heartbeats and has no actionable tasks assigned.
|
||||
// Returns false (do not skip) on any DB error to fail open.
|
||||
// is configured to skip idle periodic wakes and has no actionable tasks
|
||||
// assigned. Returns false (do not skip) on any DB error to fail open.
|
||||
func (si *SchedulerIntegration) checkIdleSkip(
|
||||
ctx context.Context, run *models.Run, agent *models.AgentInstance,
|
||||
) bool {
|
||||
if run.Reason != RunReasonHeartbeat {
|
||||
if !shared.IsPeriodicTasklessWake(run.Reason) {
|
||||
return false
|
||||
}
|
||||
if !agent.SkipIdleRuns {
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
package service_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/service"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
)
|
||||
|
||||
// TestIdleSkip_RoutineDispatchNoTasks_Skipped is the WO-46 regression test.
|
||||
// Production routine dispatch (internal/office/routines) queues runs with
|
||||
// reason "routine_dispatch", not RunReasonHeartbeat — checkIdleSkip must
|
||||
// recognize that reason too, or the idle-skip gate is unreachable in
|
||||
// production even though the existing RunReasonHeartbeat-driven tests pass.
|
||||
func TestIdleSkip_RoutineDispatchNoTasks_Skipped(t *testing.T) {
|
||||
mock := &mockTaskStarter{}
|
||||
svc := newTestService(t, service.ServiceOptions{TaskStarter: mock})
|
||||
ctx := context.Background()
|
||||
|
||||
agent := &models.AgentInstance{
|
||||
WorkspaceID: "ws-1",
|
||||
Name: "idle-worker-routine-dispatch",
|
||||
Role: models.AgentRoleWorker,
|
||||
Status: models.AgentStatusIdle,
|
||||
ExecutorPreference: `{"type":"worktree"}`,
|
||||
}
|
||||
if err := svc.CreateAgentInstance(ctx, agent); err != nil {
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
// Worker defaults to skip_idle_runs=true, no tasks assigned.
|
||||
|
||||
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatch, `{}`, ""); err != nil {
|
||||
t.Fatalf("queue: %v", err)
|
||||
}
|
||||
|
||||
service.RunSchedulerTick(svc, ctx)
|
||||
|
||||
// mock.callCount() == 0 is not asserted here: a taskless run never
|
||||
// reaches StartTask regardless of whether the idle-skip gate fired
|
||||
// correctly, so it can't discriminate a working gate from a broken one
|
||||
// (WO-46 Review round 1, S1). The activity-entry and queue-drained
|
||||
// checks below are what actually prove the skip happened.
|
||||
|
||||
// Queue should be empty (run was finished).
|
||||
next, err := svc.ClaimNextRun(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("claim after tick: %v", err)
|
||||
}
|
||||
if next != nil {
|
||||
t.Error("expected queue to be empty after idle skip")
|
||||
}
|
||||
|
||||
// Activity log should have a run_idle_skipped entry.
|
||||
entries, err := svc.ListActivity(ctx, "ws-1", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list activity: %v", err)
|
||||
}
|
||||
found := false
|
||||
for _, e := range entries {
|
||||
if e.Action == "run_idle_skipped" {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Error("expected run_idle_skipped activity entry for reason=routine_dispatch")
|
||||
}
|
||||
}
|
||||
|
||||
// TestIdleSkip_RoutineDispatch_CoordinatorNotSkipped is the complementary
|
||||
// case to TestIdleSkip_RoutineDispatchNoTasks_Skipped (PR #2973 review
|
||||
// round 1, github-actions suggestion): a coordinator (CEO role) defaults to
|
||||
// SkipIdleRuns=false because its heartbeat purpose is self-directed and
|
||||
// does not require a directly assigned task, so a routine_dispatch fire for
|
||||
// one must never be idle-skipped. checkIdleSkip's guard order happens to
|
||||
// check the reason before SkipIdleRuns, so this passes today, but nothing
|
||||
// pinned the coordinator side of the contract — a future reorder of the
|
||||
// guards in checkIdleSkip would silently break it without this test.
|
||||
func TestIdleSkip_RoutineDispatch_CoordinatorNotSkipped(t *testing.T) {
|
||||
mock := &mockTaskStarter{}
|
||||
svc := newTestService(t, service.ServiceOptions{TaskStarter: mock})
|
||||
ctx := context.Background()
|
||||
|
||||
agent := &models.AgentInstance{
|
||||
WorkspaceID: "ws-2",
|
||||
Name: "coordinator-no-skip",
|
||||
Role: models.AgentRoleCEO,
|
||||
Status: models.AgentStatusIdle,
|
||||
ExecutorPreference: `{"type":"worktree"}`,
|
||||
}
|
||||
if err := svc.CreateAgentInstance(ctx, agent); err != nil {
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
if agent.SkipIdleRuns {
|
||||
t.Fatalf("CEO role should default to SkipIdleRuns=false")
|
||||
}
|
||||
|
||||
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatch, `{}`, ""); err != nil {
|
||||
t.Fatalf("queue: %v", err)
|
||||
}
|
||||
|
||||
service.RunSchedulerTick(svc, ctx)
|
||||
|
||||
entries, err := svc.ListActivity(ctx, "ws-2", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list activity: %v", err)
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.Action == "run_idle_skipped" {
|
||||
t.Error("coordinator run must not be idle-skipped when SkipIdleRuns=false")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package shared
|
||||
|
||||
// Run reason constants shared between routine writers (internal/office/routines)
|
||||
// and scheduler readers (internal/office/service) so the two cannot silently
|
||||
// drift out of sync with each other.
|
||||
const (
|
||||
// RunReasonRoutineDispatch is set by RoutineService when it materializes
|
||||
// a lightweight (taskless) routine wakeup triggered by its cron
|
||||
// schedule — the only routine trigger that represents a periodic,
|
||||
// unattended wake. See RoutineDispatchReason.
|
||||
RunReasonRoutineDispatch = "routine_dispatch"
|
||||
|
||||
// RunReasonRoutineDispatchEvent is set by RoutineService when a
|
||||
// lightweight routine fires from a manual "Fire now" or an inbound
|
||||
// webhook. Both are event/user-triggered, not periodic — per
|
||||
// docs/specs/office/scheduler.md ("Event-triggered wakeups always
|
||||
// proceed - the skip applies only to periodic wakes") a run carrying
|
||||
// this reason must never be treated as a skippable idle wake.
|
||||
RunReasonRoutineDispatchEvent = "routine_dispatch_event"
|
||||
|
||||
// RunReasonHeartbeat is retired: the agent-level heartbeat cron was
|
||||
// replaced by the coordinator-heartbeat routine, so no production writer
|
||||
// sets it any more. Kept so any pre-retirement run row still queued is
|
||||
// still recognized as a periodic taskless wake.
|
||||
RunReasonHeartbeat = "heartbeat"
|
||||
|
||||
// RoutineSourceCron identifies a routine fire triggered by its cron
|
||||
// schedule (RoutineRun.Source). Defined here — rather than as a
|
||||
// literal at the producer call site — so RoutineDispatchReason and
|
||||
// its one production caller (routines.processCronTrigger) share a
|
||||
// single source of truth and cannot drift apart the way
|
||||
// RunReasonHeartbeat did.
|
||||
RoutineSourceCron = "cron"
|
||||
)
|
||||
|
||||
// IsPeriodicTasklessWake reports whether reason represents a periodic,
|
||||
// taskless wake — the class of run the idle-skip gate is allowed to skip
|
||||
// when the agent has no actionable tasks assigned.
|
||||
func IsPeriodicTasklessWake(reason string) bool {
|
||||
switch reason {
|
||||
case RunReasonRoutineDispatch, RunReasonHeartbeat:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// RoutineDispatchReason returns the run-reason value a lightweight
|
||||
// routine wakeup should carry for the given RoutineRun.Source. Only a
|
||||
// cron-driven fire is periodic; a manual "Fire now" or an inbound
|
||||
// webhook fire is event/user-triggered and must always proceed even
|
||||
// when the assignee has SkipIdleRuns set — see IsPeriodicTasklessWake.
|
||||
func RoutineDispatchReason(source string) string {
|
||||
if source == RoutineSourceCron {
|
||||
return RunReasonRoutineDispatch
|
||||
}
|
||||
return RunReasonRoutineDispatchEvent
|
||||
}
|
||||
Reference in New Issue
Block a user