From 85d91472082b3ccfb8075181763a79165434c305 Mon Sep 17 00:00:00 2001 From: Henry Date: Wed, 2 Sep 2026 11:47:42 +0000 Subject: [PATCH] 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 Co-authored-by: Carlos Florencio --- apps/backend/internal/office/AGENTS.md | 2 +- .../repository/sqlite/wakeup_requests.go | 57 ++++ .../repository/sqlite/wakeup_requests_test.go | 125 +++++++++ .../wo46_idle_skip_routine_dispatch_test.go | 19 +- .../internal/office/shared/runreasons.go | 27 +- .../internal/office/shared/runreasons_test.go | 48 ++++ .../internal/office/wakeup/dispatcher.go | 60 ++++- .../office/wakeup/dispatcher_race_test.go | 122 +++++++++ .../internal/office/wakeup/dispatcher_test.go | 249 ++++++++++++++++++ .../office/wakeup/routine_e2e_test.go | 6 +- .../office/system-design/scheduler-01.md | 5 +- 11 files changed, 693 insertions(+), 27 deletions(-) create mode 100644 apps/backend/internal/office/shared/runreasons_test.go create mode 100644 apps/backend/internal/office/wakeup/dispatcher_race_test.go diff --git a/apps/backend/internal/office/AGENTS.md b/apps/backend/internal/office/AGENTS.md index ee7557472d..277488e01d 100644 --- a/apps/backend/internal/office/AGENTS.md +++ b/apps/backend/internal/office/AGENTS.md @@ -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:` and `agent:`; 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. diff --git a/apps/backend/internal/office/repository/sqlite/wakeup_requests.go b/apps/backend/internal/office/repository/sqlite/wakeup_requests.go index b01de2f6d0..9f3c3d2aa9 100644 --- a/apps/backend/internal/office/repository/sqlite/wakeup_requests.go +++ b/apps/backend/internal/office/repository/sqlite/wakeup_requests.go @@ -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. diff --git a/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go b/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go index f6caa3b59b..e82c943cc8 100644 --- a/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go +++ b/apps/backend/internal/office/repository/sqlite/wakeup_requests_test.go @@ -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)) } diff --git a/apps/backend/internal/office/service/wo46_idle_skip_routine_dispatch_test.go b/apps/backend/internal/office/service/wo46_idle_skip_routine_dispatch_test.go index 88ac67b29f..2268840c0e 100644 --- a/apps/backend/internal/office/service/wo46_idle_skip_routine_dispatch_test.go +++ b/apps/backend/internal/office/service/wo46_idle_skip_routine_dispatch_test.go @@ -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) } diff --git a/apps/backend/internal/office/shared/runreasons.go b/apps/backend/internal/office/shared/runreasons.go index 3e2b354036..5c1cd8c025 100644 --- a/apps/backend/internal/office/shared/runreasons.go +++ b/apps/backend/internal/office/shared/runreasons.go @@ -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 } diff --git a/apps/backend/internal/office/shared/runreasons_test.go b/apps/backend/internal/office/shared/runreasons_test.go new file mode 100644 index 0000000000..ea8c7cfb01 --- /dev/null +++ b/apps/backend/internal/office/shared/runreasons_test.go @@ -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) + } +} diff --git a/apps/backend/internal/office/wakeup/dispatcher.go b/apps/backend/internal/office/wakeup/dispatcher.go index f6815d8092..cd3a9bd6ee 100644 --- a/apps/backend/internal/office/wakeup/dispatcher.go +++ b/apps/backend/internal/office/wakeup/dispatcher.go @@ -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 = "{}" diff --git a/apps/backend/internal/office/wakeup/dispatcher_race_test.go b/apps/backend/internal/office/wakeup/dispatcher_race_test.go new file mode 100644 index 0000000000..2373b861d2 --- /dev/null +++ b/apps/backend/internal/office/wakeup/dispatcher_race_test.go @@ -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) + } +} diff --git a/apps/backend/internal/office/wakeup/dispatcher_test.go b/apps/backend/internal/office/wakeup/dispatcher_test.go index 7df93fcdba..72d2ff6038 100644 --- a/apps/backend/internal/office/wakeup/dispatcher_test.go +++ b/apps/backend/internal/office/wakeup/dispatcher_test.go @@ -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{ diff --git a/apps/backend/internal/office/wakeup/routine_e2e_test.go b/apps/backend/internal/office/wakeup/routine_e2e_test.go index 9feb98464a..46ff66dde7 100644 --- a/apps/backend/internal/office/wakeup/routine_e2e_test.go +++ b/apps/backend/internal/office/wakeup/routine_e2e_test.go @@ -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:") 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) diff --git a/docs/specs/office/system-design/scheduler-01.md b/docs/specs/office/system-design/scheduler-01.md index 1ba239883b..eb1d8af3d4 100644 --- a/docs/specs/office/system-design/scheduler-01.md +++ b/docs/specs/office/system-design/scheduler-01.md @@ -89,7 +89,7 @@ Coalescing happens entirely at the wakeup-request layer; the runs table is the e - `task_assigned::` 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=`, 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=`, 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: