fix(office): stop idle-skip from swallowing manual wakeups queued behind a cron run (#3281)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-02 11:47:42 +00:00
committed by GitHub
parent a1a3b06033
commit 85d9147208
11 changed files with 693 additions and 27 deletions
+1 -1
View File
@@ -29,7 +29,7 @@
## Traps
- **A run is not a task start.** The `runs` table carries provider routing (`resolved_provider_id`, `logical_provider_order`, `requested_tier`, plus the `office_run_route_attempts` ledger), the per-reason `assembled_prompt`, the `AppendRunEvent` timeline, and retry/backoff/coalescing (`retry_count`, `earliest_retry_at`, `coalesced_count`, `idempotency_key` with unique index `idx_run_idempotency`). `Service.QueueRun` (`service/run.go`) and `SchedulerService.QueueRun` (`scheduler/run.go`) are the entry points that populate all of it; `TaskStarter` (the `StartTask` seam office calls, `service/service.go`) runs only *after*. Calling `StartTask` directly, or any ordinary kanban start path, skips all of the above. See `scheduler.md`.
- **`heartbeat` as a wake source is retired** (`scheduler.md`: routines are "the only mechanism for periodic wakes"; `wakeup/payloads.go` states `"heartbeat"` is intentionally absent from the `Source` enum) — but three literal `"heartbeat"` constants still exist in code (`scheduler/run.go`'s and `service/run.go`'s duplicate `RunReasonHeartbeat`, `routing/types.go`'s `WakeReasonHeartbeat`, still listed in `AllWakeReasons` and used to seed a `TierPerReason` policy in `onboarding/service.go`). `checkIdleSkip` (`service/scheduler_integration.go`) only applies when `run.Reason == RunReasonHeartbeat`, while the lightweight routine path sets `Reason: "routine_dispatch"` (`routines/service.go`, copied onto the run by `wakeup/dispatcher.go`). Check the reason string before assuming an idle-skip gate fires for a given run.
- **`heartbeat` as a wake source is retired** (`scheduler.md`: routines are "the only mechanism for periodic wakes"; `wakeup/payloads.go` states `"heartbeat"` is intentionally absent from the `Source` enum) — but three literal `"heartbeat"` constants still exist in code (`scheduler/run.go`'s and `service/run.go`'s duplicate `RunReasonHeartbeat`, `routing/types.go`'s `WakeReasonHeartbeat`, still listed in `AllWakeReasons` and used to seed a `TierPerReason` policy in `onboarding/service.go`). `checkIdleSkip` (`service/scheduler_integration.go`) delegates classification to `shared.IsPeriodicTasklessWake(run.Reason)`, true only for `RunReasonRoutineDispatchCron` and `RunReasonHeartbeat`. The lightweight routine path (`routines/service.go`) sets the reason via `shared.RoutineDispatchReason(source)` — cron gets `RunReasonRoutineDispatchCron` (skippable), manual/webhook gets `RunReasonRoutineDispatchEvent` (never skippable). The legacy `RunReasonRoutineDispatch` ("routine_dispatch") literal, written for all three triggers before the cron-only constant existed, is treated as non-skippable since a persisted run row can't say which trigger produced it. The wakeup dispatcher's claim-time coalesce (`wakeup/dispatcher.go`'s `coalesceIntoInflightRun`) promotes an in-flight `queued` run's reason to the incoming request's (or its `Source` when `Reason` is empty — same fallback as `createFreshRun`) when the merge would otherwise let a periodic-classified run swallow an event-triggered one — promotion is monotonic (periodic → event only). A **`claimed`** inflight run is never promoted in place: `processRun` (`service/scheduler_integration.go`) evaluates `checkIdleSkip` against the `*models.Run` it captured at claim time and never re-reads `Reason`, so mutating the row afterward would race a decision already in flight. An event request that would otherwise coalesce into a claimed periodic run gets its own fresh run instead. Check the reason string (and whether it could have been promoted by a coalesce, or split into a second run because the inflight one was already claimed) before assuming an idle-skip gate fires for a given run.
- **Continuation summary scope must come from `summaryScopeForRun`** (`service/event_subscribers.go`), never a literal string. Its sole caller, `refreshContinuationSummary`, upserts under the same value it reads, so reader and writer can't drift. Two scope forms exist: `routine:<id>` and `agent:<id>`; the legacy `"heartbeat"` scope is retired.
- **Participant slate: one builder, not office's own.** The AC-50 slate construction lives in `internal/workflow/engine` — `(*Engine).requiredSeats` / `requiredSeatsForWorkflow` (`workflow/engine/quorum.go`) — and `ResolveParticipantRole` (same file) reuses it so role resolution and slate membership can never disagree about who participates. Office reaches it through `engine_dispatcher.Dispatcher.ResolveParticipantRole`, which forwards verbatim. Do not add an office-local "who are this task's reviewers" query; that is a fourth answer to a question that already has exactly one.
- **Lightweight vs. heavy routines fail differently.** A routine with an empty `task_template` (`routines/service.go`; `CreateDefaultCoordinatorRoutine` sets `TaskTemplate: ""`, cron `*/5 * * * *`) is lightweight. Its **idle skip** — firing with no actionable task, launching nothing — is specified behavior (`scheduler.md`, telemetry `wakeup_idle_skipped`), not a bug; don't "fix" it. `checkIdleSkip` fails **open** on a DB error.
@@ -185,6 +185,63 @@ func (r *Repository) MarkWakeupRequestCoalesced(
return r.mergeWakeupPayloadIntoRunSnapshot(ctx, id, intoRunID)
}
// PromoteRunAndCoalesceWakeupIfQueued atomically promotes a queued run,
// attaches the wakeup request, increments the coalesced count, and merges the
// request payload. The first run update holds the writer lock until commit, so
// the scheduler cannot claim the run with only part of the coalesced state.
// It returns false when the scheduler claimed the run before this transaction.
func (r *Repository) PromoteRunAndCoalesceWakeupIfQueued(
ctx context.Context, requestID, runID, reason string,
) (bool, error) {
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return false, err
}
defer func() { _ = tx.Rollback() }()
res, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE runs SET reason = ? WHERE id = ? AND status = 'queued'
`), reason, runID)
if err != nil {
return false, err
}
updated, err := res.RowsAffected()
if err != nil {
return false, err
}
if updated == 0 {
return false, nil
}
now := time.Now().UTC()
if _, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE agent_wakeup_requests
SET status = ?, run_id = ?, claimed_at = ?, finished_at = ?
WHERE id = ?
`), WakeupStatusCoalesced, runID, now, now, requestID); err != nil {
return false, err
}
if _, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE runs SET coalesced_count = coalesced_count + 1 WHERE id = ?
`), runID); err != nil {
return false, err
}
if _, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE runs
SET context_snapshot = json_patch(
COALESCE(NULLIF(context_snapshot, ''), '{}'),
(SELECT payload FROM agent_wakeup_requests WHERE id = ?)
)
WHERE id = ?
`), requestID, runID); err != nil {
return false, err
}
if err := tx.Commit(); err != nil {
return false, err
}
return true, nil
}
// bumpRunCoalescedCount increments coalesced_count on the runs row by 1.
// Used by MarkWakeupRequestCoalesced when a wakeup-request lands on top
// of an in-flight run for the same agent.
@@ -11,6 +11,7 @@ import (
officemodels "github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
"github.com/kandev/kandev/internal/office/shared"
)
func TestWakeupRequest_CreateAndGet(t *testing.T) {
@@ -207,6 +208,130 @@ func TestWakeupRequest_MarkCoalesced(t *testing.T) {
}
}
func TestWakeupRequest_PromoteAndCoalesceCommitsAllState(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
run := &officemodels.Run{
ID: "run-cron",
AgentProfileID: "agent-1",
Reason: shared.RunReasonRoutineDispatchCron,
Payload: "{}",
Status: "queued",
CoalescedCount: 1,
ContextSnapshot: `{"existing":"value"}`,
}
if err := repo.CreateRun(ctx, run); err != nil {
t.Fatalf("create run: %v", err)
}
if err := repo.CreateWakeupRequest(ctx, &sqlite.WakeupRequest{
ID: "w-event", AgentProfileID: "agent-1", Source: "routine",
Reason: shared.RunReasonRoutineDispatchEvent,
Payload: `{"routine_id":"r-1"}`,
}); err != nil {
t.Fatalf("create wakeup: %v", err)
}
coalesced, err := repo.PromoteRunAndCoalesceWakeupIfQueued(
ctx, "w-event", run.ID, shared.RunReasonRoutineDispatchEvent,
)
if err != nil {
t.Fatalf("promote and coalesce: %v", err)
}
if !coalesced {
t.Fatal("expected queued run to accept atomic coalesce")
}
gotReq, err := repo.GetWakeupRequest(ctx, "w-event")
if err != nil {
t.Fatalf("get wakeup: %v", err)
}
if gotReq.Status != sqlite.WakeupStatusCoalesced || gotReq.RunID != run.ID {
t.Errorf("wakeup state = (%q, %q), want (coalesced, %q)", gotReq.Status, gotReq.RunID, run.ID)
}
gotRun, err := repo.GetRunByID(ctx, run.ID)
if err != nil {
t.Fatalf("get run: %v", err)
}
if gotRun.Reason != shared.RunReasonRoutineDispatchEvent {
t.Errorf("reason = %q, want %q", gotRun.Reason, shared.RunReasonRoutineDispatchEvent)
}
if gotRun.CoalescedCount != 2 {
t.Errorf("coalesced_count = %d, want 2", gotRun.CoalescedCount)
}
if !contains(gotRun.ContextSnapshot, `"routine_id":"r-1"`) {
t.Errorf("context_snapshot = %q, want routine payload", gotRun.ContextSnapshot)
}
}
func TestWakeupRequest_PromoteAndCoalesceClaimedRunDoesNothing(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
run := &officemodels.Run{
ID: "run-cron", AgentProfileID: "agent-1",
Reason: shared.RunReasonRoutineDispatchCron, Payload: "{}",
Status: "claimed", CoalescedCount: 1, ContextSnapshot: "{}",
}
if err := repo.CreateRun(ctx, run); err != nil {
t.Fatalf("create run: %v", err)
}
if err := repo.CreateWakeupRequest(ctx, &sqlite.WakeupRequest{
ID: "w-event", AgentProfileID: "agent-1", Source: "routine",
Payload: `{"routine_id":"r-1"}`,
}); err != nil {
t.Fatalf("create wakeup: %v", err)
}
coalesced, err := repo.PromoteRunAndCoalesceWakeupIfQueued(
ctx, "w-event", run.ID, shared.RunReasonRoutineDispatchEvent,
)
if err != nil {
t.Fatalf("promote and coalesce: %v", err)
}
if coalesced {
t.Fatal("claimed run must not accept atomic coalesce")
}
gotReq, _ := repo.GetWakeupRequest(ctx, "w-event")
if gotReq.Status != sqlite.WakeupStatusQueued {
t.Errorf("wakeup status = %q, want queued", gotReq.Status)
}
gotRun, _ := repo.GetRunByID(ctx, run.ID)
if gotRun.Reason != shared.RunReasonRoutineDispatchCron || gotRun.CoalescedCount != 1 {
t.Errorf("run state changed: reason=%q count=%d", gotRun.Reason, gotRun.CoalescedCount)
}
}
func TestWakeupRequest_PromoteAndCoalesceRollsBackOnMergeError(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
run := &officemodels.Run{
ID: "run-cron", AgentProfileID: "agent-1",
Reason: shared.RunReasonRoutineDispatchCron, Payload: "{}",
Status: "queued", CoalescedCount: 1, ContextSnapshot: "{}",
}
if err := repo.CreateRun(ctx, run); err != nil {
t.Fatalf("create run: %v", err)
}
if err := repo.CreateWakeupRequest(ctx, &sqlite.WakeupRequest{
ID: "w-event", AgentProfileID: "agent-1", Source: "routine", Payload: "{invalid-json",
}); err != nil {
t.Fatalf("create wakeup: %v", err)
}
if _, err := repo.PromoteRunAndCoalesceWakeupIfQueued(
ctx, "w-event", run.ID, shared.RunReasonRoutineDispatchEvent,
); err == nil {
t.Fatal("expected malformed payload to fail the atomic coalesce")
}
gotReq, _ := repo.GetWakeupRequest(ctx, "w-event")
if gotReq.Status != sqlite.WakeupStatusQueued || gotReq.RunID != "" {
t.Errorf("wakeup transaction was not rolled back: status=%q run_id=%q", gotReq.Status, gotReq.RunID)
}
gotRun, _ := repo.GetRunByID(ctx, run.ID)
if gotRun.Reason != shared.RunReasonRoutineDispatchCron || gotRun.CoalescedCount != 1 {
t.Errorf("run transaction was not rolled back: reason=%q count=%d", gotRun.Reason, gotRun.CoalescedCount)
}
}
func contains(s, sub string) bool {
return len(s) >= len(sub) && (s == sub || (len(sub) > 0 && stringIndexOf(s, sub) >= 0))
}
@@ -10,10 +10,11 @@ import (
)
// 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.
// Production cron-driven routine dispatch (internal/office/routines) queues
// runs with reason "routine_dispatch_cron", 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})
@@ -31,7 +32,7 @@ func TestIdleSkip_RoutineDispatchNoTasks_Skipped(t *testing.T) {
}
// Worker defaults to skip_idle_runs=true, no tasks assigned.
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatch, `{}`, ""); err != nil {
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatchCron, `{}`, ""); err != nil {
t.Fatalf("queue: %v", err)
}
@@ -65,7 +66,7 @@ func TestIdleSkip_RoutineDispatchNoTasks_Skipped(t *testing.T) {
}
}
if !found {
t.Error("expected run_idle_skipped activity entry for reason=routine_dispatch")
t.Error("expected run_idle_skipped activity entry for reason=routine_dispatch_cron")
}
}
@@ -73,8 +74,8 @@ func TestIdleSkip_RoutineDispatchNoTasks_Skipped(t *testing.T) {
// 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
// does not require a directly assigned task, so a routine_dispatch_cron 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.
@@ -97,7 +98,7 @@ func TestIdleSkip_RoutineDispatch_CoordinatorNotSkipped(t *testing.T) {
t.Fatalf("CEO role should default to SkipIdleRuns=false")
}
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatch, `{}`, ""); err != nil {
if err := svc.QueueRun(ctx, agent.ID, shared.RunReasonRoutineDispatchCron, `{}`, ""); err != nil {
t.Fatalf("queue: %v", err)
}
@@ -4,12 +4,21 @@ package shared
// 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 is the legacy literal written by every
// pre-cron-constant build for a lightweight routine dispatch,
// regardless of trigger (cron, manual, or webhook alike). Because
// the source cannot be recovered from an already-persisted run row,
// it is treated as non-skippable rather than guessed — see
// IsPeriodicTasklessWake. Still the value new code must recognize
// on any run row queued before an upgrade; do not delete.
RunReasonRoutineDispatch = "routine_dispatch"
// RunReasonRoutineDispatchCron 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.
RunReasonRoutineDispatchCron = "routine_dispatch_cron"
// 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
@@ -35,10 +44,14 @@ const (
// 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.
// when the agent has no actionable tasks assigned. The legacy
// RunReasonRoutineDispatch literal is deliberately excluded: it was
// written for cron, manual, and webhook fires alike before
// RunReasonRoutineDispatchCron existed, so its trigger cannot be
// recovered from the run row and it defaults to non-skippable.
func IsPeriodicTasklessWake(reason string) bool {
switch reason {
case RunReasonRoutineDispatch, RunReasonHeartbeat:
case RunReasonRoutineDispatchCron, RunReasonHeartbeat:
return true
default:
return false
@@ -52,7 +65,7 @@ func IsPeriodicTasklessWake(reason string) bool {
// when the assignee has SkipIdleRuns set — see IsPeriodicTasklessWake.
func RoutineDispatchReason(source string) string {
if source == RoutineSourceCron {
return RunReasonRoutineDispatch
return RunReasonRoutineDispatchCron
}
return RunReasonRoutineDispatchEvent
}
@@ -0,0 +1,48 @@
package shared_test
import (
"testing"
"github.com/kandev/kandev/internal/office/shared"
)
// TestIsPeriodicTasklessWake_LegacyRoutineDispatchIsNonSkippable pins the
// WO-46.1 Finding 2 fix: the legacy "routine_dispatch" literal was written
// for cron, manual, and webhook fires alike before
// RunReasonRoutineDispatchCron existed, so a persisted run row cannot say
// which trigger produced it. Treating it as periodic would risk silently
// skipping a pre-upgrade manual/webhook fire — see
// docs/specs/office/scheduler.md ("Event-triggered wakeups always proceed").
func TestIsPeriodicTasklessWake_LegacyRoutineDispatchIsNonSkippable(t *testing.T) {
if shared.IsPeriodicTasklessWake(shared.RunReasonRoutineDispatch) {
t.Errorf("IsPeriodicTasklessWake(%q) = true, want false — the legacy "+
"literal is ambiguous and must default to non-skippable",
shared.RunReasonRoutineDispatch)
}
}
// TestIsPeriodicTasklessWake_CronConstantIsSkippable pins the new
// cron-only constant as the sole "definitely periodic" routine-dispatch
// value going forward.
func TestIsPeriodicTasklessWake_CronConstantIsSkippable(t *testing.T) {
if !shared.IsPeriodicTasklessWake(shared.RunReasonRoutineDispatchCron) {
t.Errorf("IsPeriodicTasklessWake(%q) = false, want true",
shared.RunReasonRoutineDispatchCron)
}
}
// TestRoutineDispatchReason_SourceMapping pins RoutineDispatchReason's
// source→reason mapping so RunReasonRoutineDispatchCron and
// RunReasonRoutineDispatchEvent cannot silently drift apart from
// RoutineSourceCron / non-cron classification.
func TestRoutineDispatchReason_SourceMapping(t *testing.T) {
if got := shared.RoutineDispatchReason(shared.RoutineSourceCron); got != shared.RunReasonRoutineDispatchCron {
t.Errorf("RoutineDispatchReason(cron) = %q, want %q", got, shared.RunReasonRoutineDispatchCron)
}
if got := shared.RoutineDispatchReason("manual"); got != shared.RunReasonRoutineDispatchEvent {
t.Errorf("RoutineDispatchReason(manual) = %q, want %q", got, shared.RunReasonRoutineDispatchEvent)
}
if got := shared.RoutineDispatchReason("webhook"); got != shared.RunReasonRoutineDispatchEvent {
t.Errorf("RoutineDispatchReason(webhook) = %q, want %q", got, shared.RunReasonRoutineDispatchEvent)
}
}
@@ -13,6 +13,7 @@ import (
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/office/models"
officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
"github.com/kandev/kandev/internal/office/shared"
)
// Concurrency policy values stored on office_routines.concurrency_policy.
@@ -141,16 +142,68 @@ func (d *Dispatcher) Dispatch(ctx context.Context, requestID string) error {
case PolicySkipIfActive:
return d.repo.MarkWakeupRequestSkipped(ctx, req.ID, "policy:skip_if_active")
case PolicyCoalesceIfActive:
return d.repo.MarkWakeupRequestCoalesced(ctx, req.ID, inflight.ID)
return d.coalesceIntoInflightRun(ctx, req, inflight)
}
// Unknown policies fall back to coalesce — the safest "do something"
// behaviour. Surface a warning so misconfigured rows are visible.
d.log.Warn("unknown wakeup concurrency policy; coalescing",
zap.String("policy", policy),
zap.String("agent_id", req.AgentProfileID))
return d.coalesceIntoInflightRun(ctx, req, inflight)
}
// coalesceIntoInflightRun merges req into inflight, promoting the run's
// reason first when the merge would otherwise let a periodic
// classification survive on top of an event/user-triggered request —
// see docs/specs/office/scheduler.md ("Event-triggered wakeups always
// proceed"). Promotion is monotonic (periodic → event only): a
// periodic request coalescing into an already event-classified run
// never demotes it back to periodic.
//
// The inflight run's Status field reflects a read taken before this
// call, not the row's current state: the scheduler can claim it
// concurrently. processRun evaluates checkIdleSkip against the
// *models.Run it captured at claim time and never re-reads Reason, so
// promoting a row the scheduler already claimed would race a decision
// already in flight and could still let the event trigger be silently
// idle-skipped. PromoteRunAndCoalesceWakeupIfQueued makes the run
// status update, request transition, count increment, and payload merge
// one transaction. Its conditional UPDATE only lands while the row is
// still 'queued'. When it does not land (claimed out from under us),
// route the event to its own fresh run instead of trusting the stale
// in-memory status.
func (d *Dispatcher) coalesceIntoInflightRun(
ctx context.Context, req *officesqlite.WakeupRequest, inflight *models.Run,
) error {
reason := effectiveReason(req)
if reason != "" &&
shared.IsPeriodicTasklessWake(inflight.Reason) &&
!shared.IsPeriodicTasklessWake(reason) {
promoted, err := d.repo.PromoteRunAndCoalesceWakeupIfQueued(
ctx, req.ID, inflight.ID, reason,
)
if err != nil {
return fmt.Errorf("promote run and coalesce wakeup into %s: %w", inflight.ID, err)
}
if !promoted {
return d.createFreshRun(ctx, req)
}
return nil
}
return d.repo.MarkWakeupRequestCoalesced(ctx, req.ID, inflight.ID)
}
// effectiveReason returns the reason a run derived from req should carry:
// req.Reason when set, else req.Source — mirroring createFreshRun's
// fallback so a request is classified the same way whether it lands on
// a fresh run or merges into an in-flight one.
func effectiveReason(req *officesqlite.WakeupRequest) string {
if req.Reason != "" {
return req.Reason
}
return req.Source
}
// resolvePolicy returns the concurrency policy for a wakeup-request.
//
// For source="routine" it reads office_routines.concurrency_policy via
@@ -223,10 +276,7 @@ func normaliseRoutinePolicy(p string) string {
func (d *Dispatcher) createFreshRun(
ctx context.Context, req *officesqlite.WakeupRequest,
) error {
reason := req.Reason
if reason == "" {
reason = req.Source
}
reason := effectiveReason(req)
payload := req.Payload
if payload == "" {
payload = "{}"
@@ -0,0 +1,122 @@
package wakeup
import (
"context"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
"github.com/kandev/kandev/internal/common/logger"
officemodels "github.com/kandev/kandev/internal/office/models"
officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
"github.com/kandev/kandev/internal/office/shared"
)
// TestCoalesceIntoInflightRun_ClaimedBetweenReadAndPromote pins the
// TOCTOU this package's atomic promotion and coalesce transaction exists
// to close: coalesceIntoInflightRun must consult the affected-row count
// from the conditional UPDATE, not the in-memory Status captured by the
// earlier FindInflightRunForAgent read.
//
// Reverting coalesceIntoInflightRun to a stale in-memory check (`if
// inflight.Status != "queued" { createFreshRun }` followed by an
// unconditional reason write) passes every other dispatcher test,
// because they all seed the run already `claimed` or already `queued`
// with no interleaving — the in-memory status and the row's current
// status always agree. This test forces them to disagree: it captures
// `inflight` while the row is still queued, then claims the row (as
// the scheduler would, concurrently) before invoking the unexported
// coalesce path with the now-stale copy.
func TestCoalesceIntoInflightRun_ClaimedBetweenReadAndPromote(t *testing.T) {
db, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
if _, _, err := settingsstore.Provide(db, db, nil); err != nil {
t.Fatalf("settings store: %v", err)
}
repo, err := officesqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("repo: %v", err)
}
agent := &officemodels.AgentInstance{
ID: "agent-1",
WorkspaceID: "ws-1",
Name: "ceo",
AgentDisplayName: "CEO",
Role: officemodels.AgentRoleCEO,
Status: officemodels.AgentStatusIdle,
}
if err := repo.CreateAgentInstance(context.Background(), agent); err != nil {
t.Fatalf("seed agent: %v", err)
}
run := &officemodels.Run{
ID: "run-cron",
AgentProfileID: agent.ID,
Reason: shared.RunReasonRoutineDispatchCron,
Payload: "{}",
Status: officemodels.RunStatus("queued"),
CoalescedCount: 1,
ContextSnapshot: `{"prior":"snapshot"}`,
}
if err := repo.CreateRun(context.Background(), run); err != nil {
t.Fatalf("seed run: %v", err)
}
req := &officesqlite.WakeupRequest{
ID: "w-event", AgentProfileID: agent.ID, Source: SourceRoutine,
Payload: `{"routine_id":"r-1"}`, Reason: shared.RunReasonRoutineDispatchEvent,
}
if err := repo.CreateWakeupRequest(context.Background(), req); err != nil {
t.Fatalf("seed wakeup: %v", err)
}
// Dispatcher's real Dispatch() reads the in-flight run here, while
// the row is still queued. Capture that same stale snapshot.
inflight, err := repo.FindInflightRunForAgent(context.Background(), agent.ID)
if err != nil {
t.Fatalf("find inflight: %v", err)
}
if inflight.Status != officemodels.RunStatus("queued") {
t.Fatalf("precondition: inflight.Status = %q, want queued", inflight.Status)
}
// The scheduler races in and claims the row before the dispatcher's
// promotion write lands.
claimed, err := repo.ClaimNextEligibleRun(context.Background())
if err != nil {
t.Fatalf("claim: %v", err)
}
if claimed.ID != run.ID {
t.Fatalf("claim: got %q, want %q", claimed.ID, run.ID)
}
d := NewDispatcher(repo, repo, logger.Default())
if err := d.coalesceIntoInflightRun(context.Background(), req, inflight); err != nil {
t.Fatalf("coalesceIntoInflightRun: %v", err)
}
gotReq, err := repo.GetWakeupRequest(context.Background(), req.ID)
if err != nil {
t.Fatalf("get wakeup request: %v", err)
}
if gotReq.Status != officesqlite.WakeupStatusClaimed {
t.Errorf("status: got %q, want claimed (own run, not coalesced into the claimed one)", gotReq.Status)
}
if gotReq.RunID == "" || gotReq.RunID == run.ID {
t.Errorf("expected a fresh run distinct from %q, got %q", run.ID, gotReq.RunID)
}
claimedRun, err := repo.GetRunByID(context.Background(), run.ID)
if err != nil {
t.Fatalf("get claimed run: %v", err)
}
if claimedRun.Reason != shared.RunReasonRoutineDispatchCron {
t.Errorf("claimed run.Reason = %q, want unchanged %q — promoting it here would race "+
"the idle-skip decision the scheduler already started", claimedRun.Reason, shared.RunReasonRoutineDispatchCron)
}
}
@@ -14,6 +14,7 @@ import (
"github.com/kandev/kandev/internal/common/logger"
officemodels "github.com/kandev/kandev/internal/office/models"
officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
"github.com/kandev/kandev/internal/office/shared"
"github.com/kandev/kandev/internal/office/wakeup"
)
@@ -94,6 +95,36 @@ func (h *testHarness) seedRun(id, status string) {
}
}
// seedRunWithReason is seedRun with an explicit reason, for tests that
// exercise the idle-skip reason-classification promotion logic.
func (h *testHarness) seedRunWithReason(id, status, reason string) {
h.t.Helper()
run := &officemodels.Run{
ID: id,
AgentProfileID: h.agentID,
Reason: reason,
Payload: "{}",
Status: officemodels.RunStatus(status),
CoalescedCount: 1,
ContextSnapshot: `{"prior":"snapshot"}`,
}
if err := h.repo.CreateRun(context.Background(), run); err != nil {
h.t.Fatalf("seed run: %v", err)
}
}
// seedWakeupWithReason is seedWakeup with an explicit reason, mirroring
// how routines.RoutineService sets Reason via shared.RoutineDispatchReason
// before enqueuing.
func (h *testHarness) seedWakeupWithReason(id, source, payload, reason string) {
h.t.Helper()
if err := h.repo.CreateWakeupRequest(context.Background(), &officesqlite.WakeupRequest{
ID: id, AgentProfileID: h.agentID, Source: source, Payload: payload, Reason: reason,
}); err != nil {
h.t.Fatalf("seed wakeup: %v", err)
}
}
func TestDispatch_NoInflight_CreatesFreshRun(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedWakeup("w-1", wakeup.SourceSelf, `{"reason":"test"}`)
@@ -147,6 +178,186 @@ func TestDispatch_CoalesceIntoQueuedRun(t *testing.T) {
}
}
// TestDispatch_EventCoalescingIntoQueuedCronRun_PromotesReason is the
// WO-46.1 Finding 1 regression test. A manual/webhook wakeup-request
// (event-classified) that coalesces into a still-queued cron run must
// promote the run's reason to the event classification — otherwise the
// run stays periodic-class and checkIdleSkip can silently discard the
// event trigger. See docs/specs/office/scheduler.md ("Event-triggered
// wakeups always proceed").
func TestDispatch_EventCoalescingIntoQueuedCronRun_PromotesReason(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRunWithReason("run-cron", "queued", shared.RunReasonRoutineDispatchCron)
h.seedWakeupWithReason("w-event", wakeup.SourceRoutine, `{"routine_id":"r-1"}`, shared.RunReasonRoutineDispatchEvent)
if err := h.dispatcher.Dispatch(context.Background(), "w-event"); err != nil {
t.Fatalf("dispatch: %v", err)
}
gotReq, err := h.repo.GetWakeupRequest(context.Background(), "w-event")
if err != nil {
t.Fatalf("get wakeup request: %v", err)
}
if gotReq.Status != officesqlite.WakeupStatusCoalesced {
t.Errorf("status: got %q, want coalesced", gotReq.Status)
}
if gotReq.RunID != "run-cron" {
t.Errorf("run_id: got %q, want run-cron", gotReq.RunID)
}
run, err := h.repo.GetRunByID(context.Background(), "run-cron")
if err != nil {
t.Fatalf("get run: %v", err)
}
if run.Reason != shared.RunReasonRoutineDispatchEvent {
t.Errorf("run.Reason = %q, want promoted to %q", run.Reason, shared.RunReasonRoutineDispatchEvent)
}
if shared.IsPeriodicTasklessWake(run.Reason) {
t.Error("merged run must not classify as a periodic taskless wake — " +
"the idle-skip gate could silently swallow the event trigger")
}
if run.CoalescedCount != 2 {
t.Errorf("coalesced_count: got %d want 2", run.CoalescedCount)
}
if !strings.Contains(run.ContextSnapshot, `"routine_id":"r-1"`) {
t.Errorf("expected merged context snapshot, got %q", run.ContextSnapshot)
}
}
// TestDispatch_CronCoalescingIntoQueuedEventRun_DoesNotDemote is the
// monotonic half of the Finding 1 fix: a cron request merging into an
// already event-classified in-flight run must never demote the run
// back to periodic.
func TestDispatch_CronCoalescingIntoQueuedEventRun_DoesNotDemote(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRunWithReason("run-event", "queued", shared.RunReasonRoutineDispatchEvent)
h.seedWakeupWithReason("w-cron", wakeup.SourceRoutine, `{"routine_id":"r-1"}`, shared.RunReasonRoutineDispatchCron)
if err := h.dispatcher.Dispatch(context.Background(), "w-cron"); err != nil {
t.Fatalf("dispatch: %v", err)
}
run, err := h.repo.GetRunByID(context.Background(), "run-event")
if err != nil {
t.Fatalf("get run: %v", err)
}
if run.Reason != shared.RunReasonRoutineDispatchEvent {
t.Errorf("run.Reason = %q, want unchanged %q (no demotion)", run.Reason, shared.RunReasonRoutineDispatchEvent)
}
}
// TestDispatch_PeriodicCoalescingIntoQueuedPeriodicRun_DoesNotPromote
// pins the other half of the monotonicity rule: a periodic request
// merging into an already periodic-classified run must not be
// "promoted" either, even though both reasons are periodic-classified
// (heartbeat vs. routine_dispatch_cron). The condition guarding
// promotion short-circuits on inflight.Reason's classification before
// ever checking the new request's, so this case is the only one that
// exercises the `!IsPeriodicTasklessWake(reason)` clause at all.
func TestDispatch_PeriodicCoalescingIntoQueuedPeriodicRun_DoesNotPromote(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRunWithReason("run-heartbeat", "queued", shared.RunReasonHeartbeat)
h.seedWakeupWithReason("w-cron", wakeup.SourceRoutine, `{"routine_id":"r-1"}`, shared.RunReasonRoutineDispatchCron)
if err := h.dispatcher.Dispatch(context.Background(), "w-cron"); err != nil {
t.Fatalf("dispatch: %v", err)
}
run, err := h.repo.GetRunByID(context.Background(), "run-heartbeat")
if err != nil {
t.Fatalf("get run: %v", err)
}
if run.Reason != shared.RunReasonHeartbeat {
t.Errorf("run.Reason = %q, want unchanged %q (periodic->periodic is not a promotion)", run.Reason, shared.RunReasonHeartbeat)
}
}
// TestDispatch_EventCoalescingIntoClaimedCronRun_CreatesFreshRun is the
// WO-46.1 Review-round-1 R1-F1 regression test. processRun evaluates
// checkIdleSkip against the *models.Run it captured at claim time and
// never re-reads Reason, so promoting a claimed run's persisted reason
// (as the queued case does) would race a decision already in flight —
// the scheduler could still idle-skip on the stale periodic reason
// while the wakeup-request is already marked coalesced, silently
// discarding the event trigger. The dispatcher must route the event
// to its own fresh run instead of coalescing into the claimed one.
func TestDispatch_EventCoalescingIntoClaimedCronRun_CreatesFreshRun(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRunWithReason("run-cron", "claimed", shared.RunReasonRoutineDispatchCron)
h.seedWakeupWithReason("w-event", wakeup.SourceRoutine, `{"routine_id":"r-1"}`, shared.RunReasonRoutineDispatchEvent)
if err := h.dispatcher.Dispatch(context.Background(), "w-event"); err != nil {
t.Fatalf("dispatch: %v", err)
}
gotReq, err := h.repo.GetWakeupRequest(context.Background(), "w-event")
if err != nil {
t.Fatalf("get wakeup request: %v", err)
}
if gotReq.Status != officesqlite.WakeupStatusClaimed {
t.Errorf("status: got %q, want claimed (own run, not coalesced)", gotReq.Status)
}
if gotReq.RunID == "" || gotReq.RunID == "run-cron" {
t.Errorf("expected a fresh run distinct from the claimed one, got %q", gotReq.RunID)
}
freshRun, err := h.repo.GetRunByID(context.Background(), gotReq.RunID)
if err != nil {
t.Fatalf("get fresh run: %v", err)
}
if freshRun.Reason != shared.RunReasonRoutineDispatchEvent {
t.Errorf("fresh run.Reason = %q, want %q", freshRun.Reason, shared.RunReasonRoutineDispatchEvent)
}
// The claimed run's reason must stay untouched — the scheduler may
// already be mid-decision against it, so the fix must not mutate it.
claimedRun, err := h.repo.GetRunByID(context.Background(), "run-cron")
if err != nil {
t.Fatalf("get claimed run: %v", err)
}
if claimedRun.Reason != shared.RunReasonRoutineDispatchCron {
t.Errorf("claimed run.Reason = %q, want unchanged %q", claimedRun.Reason, shared.RunReasonRoutineDispatchCron)
}
}
// TestDispatch_EmptyReasonFallsBackToSourceForPromotion is the WO-46.1
// Review-round-1 R1-F2 regression test. coalesceIntoInflightRun must
// use the same reason-or-source fallback as createFreshRun, so an event
// request with an empty Reason still promotes a periodic queued run
// instead of silently leaving it periodic-classified.
func TestDispatch_EmptyReasonFallsBackToSourceForPromotion(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRunWithReason("run-cron", "queued", shared.RunReasonRoutineDispatchCron)
h.seedWakeup("w-1", wakeup.SourceComment, `{"task_id":"t-1","comment_id":"c-1"}`)
if err := h.dispatcher.Dispatch(context.Background(), "w-1"); err != nil {
t.Fatalf("dispatch: %v", err)
}
gotReq, err := h.repo.GetWakeupRequest(context.Background(), "w-1")
if err != nil {
t.Fatalf("get wakeup request: %v", err)
}
if gotReq.Status != officesqlite.WakeupStatusCoalesced {
t.Errorf("status: got %q, want coalesced", gotReq.Status)
}
if gotReq.RunID != "run-cron" {
t.Errorf("run_id: got %q, want run-cron", gotReq.RunID)
}
run, err := h.repo.GetRunByID(context.Background(), "run-cron")
if err != nil {
t.Fatalf("get run: %v", err)
}
if run.Reason != wakeup.SourceComment {
t.Errorf("run.Reason = %q, want promoted to source %q", run.Reason, wakeup.SourceComment)
}
if shared.IsPeriodicTasklessWake(run.Reason) {
t.Error("merged run must not classify as a periodic taskless wake")
}
if run.CoalescedCount != 2 {
t.Errorf("coalesced_count: got %d want 2", run.CoalescedCount)
}
if !strings.Contains(run.ContextSnapshot, `"task_id":"t-1"`) {
t.Errorf("expected merged context snapshot, got %q", run.ContextSnapshot)
}
}
func TestDispatch_CoalesceIntoClaimedRun(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.seedRun("run-running", "claimed")
@@ -256,6 +467,44 @@ func TestDispatch_Routine_NoLookupFallsBackToCoalesce(t *testing.T) {
}
}
// TestDispatch_UnknownPolicy_FallsBackToCoalesceAndPromotes exercises an
// unrecognized routine policy value. normaliseRoutinePolicy's own
// default maps it to PolicyCoalesceIfActive before Dispatch's switch
// ever sees it, so this runs the same `case PolicyCoalesceIfActive`
// path as an explicit coalesce_if_active policy — not dispatcher.go's
// switch default, which unknown *routine* policies never reach. Using
// a periodic in-flight run + event-classified request also forces that
// path through the same reason-promotion logic as an explicit
// PolicyCoalesceIfActive, so the branch cannot silently regress to a
// bare MarkWakeupRequestCoalesced call without losing this assertion.
func TestDispatch_UnknownPolicy_FallsBackToCoalesceAndPromotes(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
h.dispatcher.SetRoutineLookup(&fakeRoutineLookup{policy: "not_a_real_policy"})
h.seedRunWithReason("run-cron", "queued", shared.RunReasonRoutineDispatchCron)
h.seedWakeupWithReason("w-event", wakeup.SourceRoutine, `{"routine_id":"r-1"}`, shared.RunReasonRoutineDispatchEvent)
if err := h.dispatcher.Dispatch(context.Background(), "w-event"); err != nil {
t.Fatalf("dispatch: %v", err)
}
gotReq, err := h.repo.GetWakeupRequest(context.Background(), "w-event")
if err != nil {
t.Fatalf("get wakeup request: %v", err)
}
if gotReq.Status != officesqlite.WakeupStatusCoalesced {
t.Errorf("status = %q, want coalesced (unknown policy falls back to coalesce)", gotReq.Status)
}
if gotReq.RunID != "run-cron" {
t.Errorf("run_id: %q, want run-cron", gotReq.RunID)
}
run, err := h.repo.GetRunByID(context.Background(), "run-cron")
if err != nil {
t.Fatalf("get run: %v", err)
}
if run.Reason != shared.RunReasonRoutineDispatchEvent {
t.Errorf("run.Reason = %q, want promoted to %q", run.Reason, shared.RunReasonRoutineDispatchEvent)
}
}
func TestCreateWakeupRequest_IdempotencyConflictReturnsSentinel(t *testing.T) {
h := newHarness(t, wakeup.PolicyCoalesceIfActive)
first := &officesqlite.WakeupRequest{
@@ -30,7 +30,7 @@ import (
// 2. The routines service materialises the lightweight routine into
// an agent_wakeup_requests row with source=routine.
// 3. The wakeup dispatcher claims the row and creates a fresh runs
// row (taskless) tagged reason="routine_dispatch" with the
// row (taskless) tagged reason="routine_dispatch_cron" with the
// routine_id mirrored into the run's context_snapshot.
// 4. A second tick five minutes later produces a second fresh run
// and the (agent, "routine:<id>") summary upsert path is reachable
@@ -64,8 +64,8 @@ func TestRoutine_EndToEnd_CoordinatorHeartbeatFire(t *testing.T) {
}
run := requireInflightRun(t, repo, agentID)
if run.Reason != "routine_dispatch" {
t.Errorf("run.reason: got %q want routine_dispatch", run.Reason)
if run.Reason != "routine_dispatch_cron" {
t.Errorf("run.reason: got %q want routine_dispatch_cron", run.Reason)
}
if !strings.Contains(run.ContextSnapshot, `"routine_id":"`+routine.ID+`"`) {
t.Errorf("expected routine_id in context_snapshot, got %q", run.ContextSnapshot)
@@ -89,7 +89,7 @@ Coalescing happens entirely at the wakeup-request layer; the runs table is the e
- `task_assigned:<task_id>:<agent_instance_id>` for task creation/assignment events.
Duplicate inserts in the same window are rejected silently. Handles webhook re-delivery, event-bus replay, and restart recovery.
2. **Claim-time merge.** When the dispatcher processes a wakeup-request, it looks for an in-flight run for the same agent (`queued` -> `scheduled-retry` -> `running`, in that order). If one exists: insert the new request with `status="coalesced"`, `run_id=<existing>`, merge the new request's payload into the existing run's `context_snapshot`, and increment `coalesced_count` on the in-flight wakeup-request. The agent sees the merged context when it actually runs. If none exists: insert the wakeup-request with `status="queued"` and create the corresponding `runs` row.
2. **Claim-time merge.** The source first persists the wakeup-request with `status="queued"`. The dispatcher then looks for an in-flight run for the same agent (`queued` or `claimed`). If one exists **and it is still `queued`**: mark the persisted request with `status="coalesced"` and `run_id=<existing>`, merge its payload into the existing run's `context_snapshot`, increment the run's `coalesced_count`, and — if the existing run is periodic-classified (see "Idle wakeup skip" below) while the new request is event-classified — promote the run's `reason` to the new request's classification (monotonic: periodic -> event only, never the reverse). The promotion, request transition, count update, and payload merge use one transaction, so the scheduler cannot claim a partially updated run. The agent sees the merged context and the promoted reason when it actually runs. If the in-flight run is already **`claimed`** *and the promotion above would otherwise be required* (the run is periodic-classified and the new request is event-classified): it is not promoted or merged into — the scheduler has already read its `reason` into memory for the idle-skip decision and never re-reads the row, so mutating it afterward would race a decision already in flight. The new request instead creates its own fresh `runs` row and is marked `claimed` against it, exactly as if no in-flight run existed. A `claimed` run that needs no promotion (the existing run is already event-classified, or the new request is itself periodic-classified) is coalesced into exactly as before. If no in-flight run exists at all: create the corresponding `runs` row and mark the persisted request as `claimed` against it.
3. **`runs.idempotency_key`** is kept as a defensive secondary key. Rarely tripped now (the wakeup-request layer handles the common case), but useful for the rare "two cron processes fired the same tick from different leaders" scenario during a leadership change.
@@ -189,7 +189,8 @@ Coordinator agents default to `false` because their heartbeat purpose is self-di
| Source | Skippable? |
|--------|-----------|
| `routine` (lightweight, heartbeat-style) | Yes |
| `routine`, cron-triggered (lightweight, heartbeat-style) | Yes |
| `routine`, manual "Fire now" or webhook-triggered | No |
| `task_assigned`, `comment`, `task_blockers_resolved`, `task_children_completed`, `approval_resolved`, `agent_error`, `budget_alert`, `self`, `user` | No |
Skipped wakeups are not silently discarded: