feat(office): rate-limit agent-initiated task-assigned wakes per task (#3814)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-22 17:26:28 +08:00
committed by GitHub
parent 6e3fc6cc2a
commit fbf15bb5de
35 changed files with 3293 additions and 9 deletions
+1
View File
@@ -174,6 +174,7 @@ history and remains immutable.
- ADR 0015's step-completion-signal telemetry lives under `workflow_*`: `workflow_step_completion_signal_received_total` (`internal/workflow/signalmetrics/`), labelled by `source` and `agent_type`, counts accepted `step_complete_kandev` signals. Its separate `workflow_step_completion_signal_fallback_used_total` counter is labelled by `agent_type` and counts manual fallback uses. The fallback button does not have a production increment site yet, so the counter remains zero until that UI ships.
- Office stall detection (REQ-OFFICE-STALL-VISIBILITY) lives under `office_stall_*`: `office_stall_stranded_signal_total` (labelled by `gate`, which names which of the two watchdog gate sites saw it), `office_stall_decision_waiting_total`, and `office_stall_detector_skipped_total` (labelled by `reason`, so a detector that fails closed is visible rather than silent). Detection only: these counters never accompany a transition, a synthesized decision, or a queued run, because Office tasks are surfaced and never reclaimed. Also emitted as structured zap logs.
- Office session-termination suppression (REQ-OFFICE-SESSION-TERM) lives under `office_session_term_suppressed_total`, labelled by `reason` (one of `task_reassigned`, `participant_removed`, `participant_seat_claimed`) and `outcome` (`runner` or `seat` when the agent still holds that capacity, `read_failed` when the capacity read itself failed). It counts guarded terminations that left a session live, so a suppression is distinguishable from a path that never ran. Both dimensions are closed sets; no task, agent, step or session identifier is ever a label. Also emitted as structured zap logs.
- Office assignment-wake rate limiting (REQ-OFFICE-ASSIGN-RATE) lives under `office_assignment_rate_limit_total`, labelled by `reason` — one of `allowance_exhausted` (the wake was refused), `count_read_failed` (fail-open: the window count query errored), or `task_unattributed` (fail-open: the wake's task could not be determined). The label set is closed; no task, agent, or run identifier is ever a label. Also emitted as structured zap logs.
- Task-level stall detection (issue #3712) runs in the session reconciliation sweep (`internal/task/service/`, one-minute tick shared with the archived-session pass): an unarchived task holding a non-idle active session with no live execution behind it and no events/messages past `tasks.stallDetectionThreshold` (default 2h) emits a `task.stalled` event plus a `stalled_task` WARN, once per stall episode. At or beyond twice the threshold the sweep conditionally settles classified interrupted sessions to `WAITING_FOR_INPUT`, preserving the same conversation and abandoning only the observed turn without a completion event; the recovery runs only when every active session of the task is execution-less and past the grace window (an idle waiting sibling blocks it), and liveness is re-checked at the recovery boundary. A stall past the grace window can therefore emit `task.stalled` and a `session.state_changed` in the same sweep; recovery never depends on the detection event. The separate restart pass applies the same guarded settlement per stale `STARTING`/`RUNNING` session, including rows without an executor record.
### GitHub Operations
+1
View File
@@ -26,6 +26,7 @@
| `unread-divider.md` | shipped | Slack-style unread divider for background-running task sessions |
| `requirements/task-session-termination.md` + `system-design/task-session-termination-01.md` | shipped | Retained-capacity precondition on the three session-termination call sites (role removal, seat-claim displacement, reassignment): don't end a shared (task, agent) session while the agent still holds another capacity |
| `requirements/seat-claim-decision-guard.md` + `system-design/seat-claim-decision-guard-01.md` | shipped | `claimAutoSeat`'s `NOT EXISTS` guard against a roleless decision race; latent on every shipped decision path but the only defense on that path |
| `requirements/assignment-wake-rate-limit.md` + `system-design/assignment-wake-rate-limit-01.md` | shipped | Per-task rolling-window admission gate (5 admitted agent-initiated `task_assigned` wakes per 10-minute window) bounding self-reassignment run spam from an agent holding `can_assign_tasks` |
The two rows above are the only entries reflecting the newer `requirements/` + `system-design/` split; the other 15 predate that migration and this table has not been reconciled with the full current Office spec set. `docs/specs/office/README.md` no longer carries a tracked specification map (see ADR `2026-09-07-on-demand-document-catalogs`) — run `python3 scripts/list-docs.py specs --system office --format paths` for the authoritative, current list of anything not covered here.
@@ -0,0 +1,110 @@
package dashboard_test
import (
"context"
"encoding/json"
"testing"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/scheduler"
officeservice "github.com/kandev/kandev/internal/office/service"
)
// wireRealReactivity replaces the dashboard-side reactivity stub with the
// production adapter around a real *scheduler.SchedulerService backed by
// the same repository, so a dashboard-level mutation call reaches an
// actually-persisted runs row rather than a captured TaskReactivityChange.
func wireRealReactivity(t *testing.T, deps *testDeps) {
t.Helper()
log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
if err != nil {
t.Fatalf("logger: %v", err)
}
svc := officeservice.NewService(officeservice.ServiceOptions{Repo: deps.repo, Logger: log})
ss := scheduler.NewSchedulerService(deps.repo, log, svc)
deps.svc.SetReactivityApplier(scheduler.NewDashboardReactivityAdapter(ss))
}
// createIdleAgent registers an idle agent instance so the reactivity
// pipeline's guardAgentStatus check admits the wake instead of skipping it.
func createIdleAgent(t *testing.T, deps *testDeps, id, wsID string) {
t.Helper()
if err := deps.repo.CreateAgentInstance(context.Background(), &models.AgentInstance{
ID: id, WorkspaceID: wsID, Name: id, Status: models.AgentStatusIdle,
}); err != nil {
t.Fatalf("create agent %s: %v", id, err)
}
}
// latestTaskAssignedPayload reads back the most recently persisted
// task_assigned run row's JSON payload for the given assignee, decoded.
func latestTaskAssignedPayload(t *testing.T, deps *testDeps, assigneeID string) map[string]any {
t.Helper()
var payload string
err := deps.db.Get(&payload, `
SELECT payload FROM runs
WHERE reason = 'task_assigned' AND agent_profile_id = ?
ORDER BY requested_at DESC LIMIT 1
`, assigneeID)
if err != nil {
t.Fatalf("query persisted run payload for %s: %v", assigneeID, err)
}
var decoded map[string]any
if err := json.Unmarshal([]byte(payload), &decoded); err != nil {
t.Fatalf("decode payload %s: %v", payload, err)
}
return decoded
}
// TestSetTaskAssigneeAsAgent_PersistedRunPayloadCarriesActorType exercises
// runReactivityForAssigneeChange's actor-classification line
// (service_tasks.go's callerAgentID != "" branch) through a real
// SetTaskAssigneeAsAgent call, all the way to the persisted runs row.
// Existing coverage either hand-built the scheduler-side payload directly
// (assignment_rate_limit_inscope_producer_test.go) or stubbed the
// reactivity applier before any run was queued (session_termination_test.go)
// — neither proves this line's output ever reaches a stored payload.
func TestSetTaskAssigneeAsAgent_PersistedRunPayloadCarriesActorType(t *testing.T) {
deps := newTestDeps(t)
wireRealReactivity(t, deps)
deps.agents.names = map[string]string{"agent-actor": "agent-actor"}
deps.agents.roles = map[string]string{"agent-actor": string(models.AgentRoleWorker)}
insertTestTask(t, deps.db, "task-actor-classify", "ws-actor", "Actor classify", "todo", 2)
createIdleAgent(t, deps, "agent-assignee-by-agent", "ws-actor")
if err := deps.svc.SetTaskAssigneeAsAgent(
context.Background(), "agent-actor", "task-actor-classify", "agent-assignee-by-agent",
); err != nil {
t.Fatalf("set assignee (agent caller): %v", err)
}
got := latestTaskAssignedPayload(t, deps, "agent-assignee-by-agent")
if got["actor_type"] != "agent" {
t.Fatalf("persisted run payload actor_type = %v, want %q: %#v", got["actor_type"], "agent", got)
}
}
// TestSetTaskAssigneeAsAgent_EmptyCallerNeverPersistsAgentActor is the
// negative counterpart: an internal/admin caller (empty callerAgentID) must
// never persist actor_type "agent" in the run row the new assignee wakes on.
func TestSetTaskAssigneeAsAgent_EmptyCallerNeverPersistsAgentActor(t *testing.T) {
deps := newTestDeps(t)
wireRealReactivity(t, deps)
insertTestTask(t, deps.db, "task-actor-classify-2", "ws-actor-2", "Actor classify 2", "todo", 2)
createIdleAgent(t, deps, "agent-assignee-by-user", "ws-actor-2")
if err := deps.svc.SetTaskAssigneeAsAgent(
context.Background(), "", "task-actor-classify-2", "agent-assignee-by-user",
); err != nil {
t.Fatalf("set assignee (empty caller): %v", err)
}
got := latestTaskAssignedPayload(t, deps, "agent-assignee-by-user")
if got["actor_type"] == "agent" {
t.Fatalf("persisted run payload actor_type = %v, want anything but %q for an empty callerAgentID: %#v",
got["actor_type"], "agent", got)
}
}
@@ -253,15 +253,19 @@ func newTestDepsWithLogger(t *testing.T, log *logger.Logger) *testDeps {
}
// stubAgentReader returns nil/nil by default; tests that need agent
// resolution (e.g. pending_approvers name lookup) populate `names`.
// resolution (e.g. pending_approvers name lookup) populate `names`. Tests
// that need a caller to clear a permission check (e.g. can_assign_tasks)
// also populate `roles`, keyed the same way. `instances` backs
// ListAgentInstances for workspace-scoped listing tests.
type stubAgentReader struct {
names map[string]string
roles map[string]string
instances []*models.AgentInstance
}
func (s *stubAgentReader) GetAgentInstance(_ context.Context, id string) (*models.AgentInstance, error) {
if name, ok := s.names[id]; ok {
return &models.AgentInstance{ID: id, Name: name}, nil
return &models.AgentInstance{ID: id, Name: name, Role: models.AgentRole(s.roles[id])}, nil
}
return nil, nil
}
@@ -0,0 +1,60 @@
package onboarding
import (
"context"
"encoding/json"
"testing"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
// fakeOnboardingRunQueuer is a minimal shared.RunQueuer fake that captures
// the payload maybeCreateOnboardingTask enqueues, so the test can inspect
// it directly instead of inferring shape from a persisted row.
type fakeOnboardingRunQueuer struct {
reason, payload string
called bool
}
func (f *fakeOnboardingRunQueuer) QueueRun(
_ context.Context, _, reason, payload, _ string,
) (runsservice.QueueOutcome, error) {
f.called = true
f.reason = reason
f.payload = payload
return runsservice.QueueOutcomeQueued, nil
}
// TestMaybeCreateOnboardingTask_NeverCarriesAgentActorType pins
// AC-OFFICE-ASSIGN-RATE-001.12 for the onboarding task_assigned producer
// (documented in service.go as "a third task_assigned producer"): its
// payload is fmt.Sprintf(`{"task_id":%q}`, taskID), with no actor_type
// field at all, so it can never fall into the assignment-wake-rate-limit's
// scope.
func TestMaybeCreateOnboardingTask_NeverCarriesAgentActorType(t *testing.T) {
svc, _, _ := newTestOnboardingService(t)
svc.taskCreator = &mockTaskCreatorOnboarding{}
queuer := &fakeOnboardingRunQueuer{}
svc.runQueuer = queuer
taskID := svc.maybeCreateOnboardingTask(context.Background(), "ws-1", "agent-1", CompleteRequest{
TaskTitle: "Explore the codebase",
TaskDescription: "Create an engineering roadmap",
})
if taskID == "" {
t.Fatal("expected a created task ID")
}
if !queuer.called {
t.Fatal("expected QueueRun to be called")
}
if queuer.reason != runReasonTaskAssigned {
t.Fatalf("reason = %q, want %q", queuer.reason, runReasonTaskAssigned)
}
var payload map[string]any
if err := json.Unmarshal([]byte(queuer.payload), &payload); err != nil {
t.Fatalf("decode payload: %v", err)
}
if actorType, present := payload["actor_type"]; present && actorType == "agent" {
t.Fatalf("onboarding task_assigned payload carries actor_type=agent: %v", payload)
}
}
@@ -471,6 +471,7 @@ func (r *Repository) createRunTables() error {
finished_at TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_run_status_requested ON runs(status, requested_at);
CREATE INDEX IF NOT EXISTS idx_runs_assignment_rate_reason_requested ON runs(reason, requested_at);
CREATE UNIQUE INDEX IF NOT EXISTS idx_run_idempotency ON runs(idempotency_key) WHERE idempotency_key IS NOT NULL;
CREATE TABLE IF NOT EXISTS office_run_skills (
@@ -73,6 +73,9 @@ func (r *Repository) runMigrations() error {
if err := r.migrateRetentionIndexes(); err != nil {
return err
}
if err := r.migrateAssignmentWakeRateIndexes(); err != nil {
return err
}
if err := r.migrate.Err(); err != nil {
return err
}
@@ -192,6 +195,17 @@ func (r *Repository) migrateRetentionIndexes() error {
)
}
// migrateAssignmentWakeRateIndexes adds the portable prefix used by the
// rolling assignment-wake count. The count deliberately has no status
// predicate, so idx_run_status_requested cannot support its reason/time scan.
func (r *Repository) migrateAssignmentWakeRateIndexes() error {
return r.migrate.Apply(
"idx_runs_assignment_rate_reason_requested",
`CREATE INDEX IF NOT EXISTS idx_runs_assignment_rate_reason_requested
ON runs(reason, requested_at)`,
)
}
// migrateContinuationScope adds runs.continuation_scope for databases
// created before WO-16's claim-time scope persistence. Existing rows receive
// a scope from their stored context snapshot so queued or claimed taskless
@@ -9,10 +9,9 @@ import (
)
// TestPostgresRetentionIndexes_CreatedFreshAndReplaySafe is the PostgreSQL
// half of TestRetentionIndexes_CreatedFreshAndReplaySafe: the two
// expression indexes the retention sweep depends on must exist there too,
// with identical CREATE INDEX IF NOT EXISTS replay safety. Skips unless
// KANDEV_TEST_POSTGRES_DSN is set.
// half of TestRetentionIndexes_CreatedFreshAndReplaySafe. The run-history and
// assignment-wake indexes must exist there too, with identical CREATE INDEX IF
// NOT EXISTS replay safety. Skips unless KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
dsn := testutil.PostgresDSNFromEnv(t)
conn := testutil.OpenIsolatedPostgres(t, dsn)
@@ -28,6 +27,7 @@ func TestPostgresRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
assertPostgresIndexExists(t, conn, "idx_office_routine_runs_retention")
assertPostgresIndexExists(t, conn, "idx_runs_retention")
assertPostgresIndexExists(t, conn, "idx_office_agent_pause_recoveries_failed_run")
assertPostgresIndexExists(t, conn, "idx_runs_assignment_rate_reason_requested")
if _, err := sqlite.NewWithDB(conn, conn, nil); err != nil {
t.Fatalf("replay NewWithDB: %v", err)
@@ -35,6 +35,7 @@ func TestPostgresRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
assertPostgresIndexExists(t, conn, "idx_office_routine_runs_retention")
assertPostgresIndexExists(t, conn, "idx_runs_retention")
assertPostgresIndexExists(t, conn, "idx_office_agent_pause_recoveries_failed_run")
assertPostgresIndexExists(t, conn, "idx_runs_assignment_rate_reason_requested")
}
func assertPostgresIndexExists(t *testing.T, conn interface {
@@ -9,9 +9,9 @@ import (
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestRetentionIndexes_CreatedFreshAndReplaySafe proves the retention indexes
// exist after a fresh boot and that re-running schema init against the same
// database (the upgrade-path replay) is a no-op, not an error.
// TestRetentionIndexes_CreatedFreshAndReplaySafe proves the run-history and
// assignment-wake indexes exist after a fresh boot and that re-running schema
// init against the same database (the upgrade-path replay) is a no-op.
func TestRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
conn, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
@@ -26,6 +26,7 @@ func TestRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
assertIndexExists(t, conn, "idx_office_routine_runs_retention")
assertIndexExists(t, conn, "idx_runs_retention")
assertIndexExists(t, conn, "idx_office_agent_pause_recoveries_failed_run")
assertIndexExists(t, conn, "idx_runs_assignment_rate_reason_requested")
// Replay: schema init against the same, already-initialized database.
if _, err := sqlite.NewWithDB(conn, conn, nil); err != nil {
@@ -34,6 +35,7 @@ func TestRetentionIndexes_CreatedFreshAndReplaySafe(t *testing.T) {
assertIndexExists(t, conn, "idx_office_routine_runs_retention")
assertIndexExists(t, conn, "idx_runs_retention")
assertIndexExists(t, conn, "idx_office_agent_pause_recoveries_failed_run")
assertIndexExists(t, conn, "idx_runs_assignment_rate_reason_requested")
}
func assertIndexExists(t *testing.T, conn *sqlx.DB, name string) {
@@ -0,0 +1,105 @@
package scheduler
import (
"context"
"encoding/json"
"time"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
const assignmentWakeActorTypeAgent = "agent"
// assignmentWakePredicate is the result of classifying an incoming wake
// against the three-step predicate in the system design's "## The
// predicate": reason == task_assigned, actor_type == "agent", and the
// wake names a task. Evaluated in that order because a wake failing step
// 2 is out of scope and must never reach step 3's counted unattributed
// case.
type assignmentWakePredicate struct {
// inScope is true once steps 1 and 2 both pass: this is an
// agent-initiated assignment wake.
inScope bool
// taskAttributed is true once step 3 also passes: the wake names a
// task the allowance can be scoped to.
taskAttributed bool
taskID string
// actorID is the acting agent identifier, carried through for the
// Warn log fields REQ-OFFICE-ASSIGN-RATE-003 names. Empty when the
// payload carries none.
actorID string
}
// classifyAssignmentWake evaluates the predicate against the incoming
// (not-yet-persisted) wake. It reads actor_type and task_id out of the
// encoded payload in Go, with one json.Unmarshal — the incoming wake is
// not a row yet, so dialect.JSONExtract (a SQL fragment over a column)
// cannot apply to it; that only applies to the stored rows the window
// count scans. An unparseable payload carries no readable actor_type
// either, so it fails step 2 and is classified out of scope, exactly
// like a payload whose actor_type is merely absent or not "agent" —
// never counted as unattributed, which is reserved for a wake already
// shown to be in scope.
func classifyAssignmentWake(reason, payload string) assignmentWakePredicate {
if reason != RunReasonTaskAssigned {
return assignmentWakePredicate{}
}
var raw map[string]any
if err := json.Unmarshal([]byte(payload), &raw); err != nil {
return assignmentWakePredicate{}
}
actorType, _ := raw["actor_type"].(string)
if actorType != assignmentWakeActorTypeAgent {
return assignmentWakePredicate{}
}
actorID, _ := raw["actor_id"].(string)
pred := assignmentWakePredicate{inScope: true, actorID: actorID}
// A task cannot be determined when the identifier is absent, null,
// empty, or present but not a string; those shapes are treated
// identically.
v, present := raw["task_id"]
if !present || v == nil {
return pred
}
taskID, ok := v.(string)
if !ok || taskID == "" {
return pred
}
pred.taskID = taskID
pred.taskAttributed = true
return pred
}
// checkAssignmentWakeAllowance enforces REQ-OFFICE-ASSIGN-RATE-001..003.
// Returns true only when the wake must be refused — the caller must then
// insert no row and report QueueOutcomeRateLimited. Every other path
// (out of scope, unattributed, a failed count read, or room left in the
// allowance) returns false: admit the wake as queueRun otherwise would.
func (ss *SchedulerService) checkAssignmentWakeAllowance(
ctx context.Context, agentInstanceID, reason, payload string,
) bool {
pred := classifyAssignmentWake(reason, payload)
if !pred.inScope {
return false
}
if !pred.taskAttributed {
runsservice.ReportAssignmentRateLimitUnattributed(pred.actorID, AssignmentWakeAllowanceN)
return false
}
evaluationInstant := time.Now().UTC()
windowStart := evaluationInstant.Add(-AssignmentWakeAllowanceWindow)
count, err := ss.repo.CountAgentInitiatedAssignmentWakes(ctx, pred.taskID, reason, windowStart, evaluationInstant)
if err != nil {
runsservice.ReportAssignmentRateLimitCountReadFailed(
pred.taskID, agentInstanceID, pred.actorID, AssignmentWakeAllowanceN, err)
return false
}
if count < AssignmentWakeAllowanceN {
return false
}
runsservice.ReportAssignmentRateLimitRefused(
pred.taskID, agentInstanceID, pred.actorID, count, AssignmentWakeAllowanceN)
return true
}
@@ -0,0 +1,70 @@
package scheduler
import (
"context"
"sync"
"testing"
"time"
)
// TestCheckAssignmentWakeAllowance_ConcurrentEvaluationsBothAdmit covers
// AC-OFFICE-ASSIGN-RATE-002.1: with the allowance one wake short of
// exhausted, two genuinely concurrent evaluations — started together via a
// shared barrier channel, not a sequential call pair — must each observe
// room and must each admit, even though admitting both together pushes the
// task past N. The gate's count-then-insert shape offers no other outcome:
// serializing the check itself would be the "convenient" answer the spec's
// own overview warns against, and neither goroutine here inserts its row
// until after both have evaluated, so the count each one reads cannot
// reflect the other's admission.
func TestCheckAssignmentWakeAllowance_ConcurrentEvaluationsBothAdmit(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-concurrent"
for i := 0; i < AssignmentWakeAllowanceN-1; i++ {
createAssignmentWakeRun(t, repo, taskID, time.Now().UTC())
}
payload := `{"task_id":"` + taskID + `","actor_type":"agent"}`
start := make(chan struct{})
var wg sync.WaitGroup
refused := make([]bool, 2)
wg.Add(2)
for i := range refused {
i := i
go func() {
defer wg.Done()
<-start
refused[i] = ss.checkAssignmentWakeAllowance(ctx, "agent-1", RunReasonTaskAssigned, payload)
}()
}
close(start)
wg.Wait()
for i, r := range refused {
if r {
t.Fatalf("evaluation[%d] was refused with only %d of %d already admitted; AC-OFFICE-ASSIGN-RATE-002.1 forbids refusing before N are admitted", i, AssignmentWakeAllowanceN-1, AssignmentWakeAllowanceN)
}
}
// Each evaluation having found room, both now insert — mirroring what
// queueRun's CreateRun does immediately after a non-refusing check.
// This is the "over-admission bounded by the number of concurrent
// evaluations" the AC permits: two concurrent admits on top of the
// pre-seeded N-1 leaves the task one over its allowance, not unbounded.
createAssignmentWakeRun(t, repo, taskID, time.Now().UTC())
createAssignmentWakeRun(t, repo, taskID, time.Now().UTC())
evaluationInstant := time.Now().UTC()
count, err := repo.CountAgentInitiatedAssignmentWakes(
ctx, taskID, RunReasonTaskAssigned, evaluationInstant.Add(-AssignmentWakeAllowanceWindow), evaluationInstant)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != AssignmentWakeAllowanceN+1 {
t.Fatalf("count = %d, want %d (N-1 seeded plus the two concurrent admits)", count, AssignmentWakeAllowanceN+1)
}
}
@@ -0,0 +1,109 @@
package scheduler
import (
"context"
"expvar"
"path/filepath"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
"github.com/kandev/kandev/internal/common/logger"
officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
)
// newSplitConnTestRepo builds an office repository backed by two distinct
// SQLite connections to the same file-backed database — a writer and a
// reader, as production wiring uses. Unlike newReactivityTestRepo (one
// shared in-memory connection for both), this lets a test sever the
// reader alone and force a genuine read error, while the writer keeps
// working. WAL mode is required for two live connections against one
// SQLite file.
func newSplitConnTestRepo(t *testing.T) (repo *officesqlite.Repository, reader *sqlx.DB) {
t.Helper()
dbPath := filepath.Join(t.TempDir(), "test.db") + "?_journal_mode=WAL"
writer, err := sqlx.Open("sqlite3", dbPath)
if err != nil {
t.Fatalf("open writer: %v", err)
}
t.Cleanup(func() { _ = writer.Close() })
reader, err = sqlx.Open("sqlite3", dbPath)
if err != nil {
t.Fatalf("open reader: %v", err)
}
log := mustTestLogger(t)
repo, err = officesqlite.NewWithDB(writer, reader, log)
if err != nil {
t.Fatalf("new repo: %v", err)
}
return repo, reader
}
func mustTestLogger(t *testing.T) *logger.Logger {
t.Helper()
log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
if err != nil {
t.Fatalf("logger: %v", err)
}
return log
}
// assignmentRateLimitCounterValue reads one key out of the
// office_assignment_rate_limit_total expvar map, mirroring
// internal/runs/service's own counterValue test helper (unexported there,
// so duplicated here rather than exported for one cross-package test).
func assignmentRateLimitCounterValue(t *testing.T, key string) int64 {
t.Helper()
v := expvar.Get("office_assignment_rate_limit_total")
if v == nil {
t.Fatalf("expvar map %q not registered", "office_assignment_rate_limit_total")
}
m, ok := v.(*expvar.Map)
if !ok {
t.Fatalf("expvar %q is not a *expvar.Map", "office_assignment_rate_limit_total")
}
var got int64
m.Do(func(kv expvar.KeyValue) {
if kv.Key != key {
return
}
if iv, ok := kv.Value.(*expvar.Int); ok {
got = iv.Value()
}
})
return got
}
// TestCheckAssignmentWakeAllowance_RealCountReadFailureAdmits covers
// AC-OFFICE-ASSIGN-RATE-002.2 end to end through the production call
// path: with the reader connection actually severed (not a simulated
// error passed to the reporting helper directly), CountAgentInitiated
// AssignmentWakes returns a genuine error, and checkAssignmentWakeAllowance
// must still admit (return refused=false) and move the count_read_failed
// counter exactly once.
func TestCheckAssignmentWakeAllowance_RealCountReadFailureAdmits(t *testing.T) {
repo, reader := newSplitConnTestRepo(t)
ss := NewSchedulerService(repo, mustTestLogger(t), nil)
ctx := context.Background()
before := assignmentRateLimitCounterValue(t, "reason=count_read_failed")
if err := reader.Close(); err != nil {
t.Fatalf("close reader: %v", err)
}
refused := ss.checkAssignmentWakeAllowance(ctx, "agent-1", RunReasonTaskAssigned,
`{"task_id":"read-failure-task","actor_type":"agent"}`)
if refused {
t.Fatal("a genuine count-read failure must fail open (admit), not refuse")
}
after := assignmentRateLimitCounterValue(t, "reason=count_read_failed")
if after != before+1 {
t.Fatalf("count_read_failed counter = %d, want %d", after, before+1)
}
}
@@ -0,0 +1,92 @@
package scheduler
import (
"encoding/json"
"testing"
"github.com/kandev/kandev/internal/common/logger"
)
// TestReactToAssigneeChange_AgentActorPayloadCarriesActorTypeAgent is the
// positive counterpart to the 4 out-of-scope producer tests
// (internal/office/service, internal/office/onboarding,
// internal/orchestrator): reactToAssigneeChange is the only
// task_assigned producer whose payload-building type (RunContext) has an
// ActorType field, and this confirms the field actually reaches the
// JSON payload end to end through encodeRunContext — the exact shape
// classifyAssignmentWake's predicate looks for.
func TestReactToAssigneeChange_AgentActorPayloadCarriesActorTypeAgent(t *testing.T) {
ss := &SchedulerService{logger: logger.Default()}
task := &TaskSnapshot{
ID: "task-inscope",
WorkspaceID: "ws-1",
State: "TODO",
AssigneeAgentProfileID: "",
}
change := TaskMutation{
ActorID: "acting-agent-1",
ActorType: "agent",
}
var calls []recordedQueueCall
queue := func(agentID string, c RunContext) {
calls = append(calls, recordedQueueCall{agentID: agentID, ctx: c})
}
ss.reactToAssigneeChange(task, "new-assignee", change, queue, &ApplyTaskMutationResult{})
if len(calls) != 1 {
t.Fatalf("got %d queue calls, want 1", len(calls))
}
if calls[0].ctx.ActorType != "agent" {
t.Fatalf("RunContext.ActorType = %q, want %q", calls[0].ctx.ActorType, "agent")
}
payload, err := encodeRunContext(calls[0].ctx)
if err != nil {
t.Fatalf("encodeRunContext: %v", err)
}
var decoded map[string]any
if err := json.Unmarshal([]byte(payload), &decoded); err != nil {
t.Fatalf("decode payload: %v", err)
}
if decoded["actor_type"] != "agent" {
t.Fatalf("payload actor_type = %v, want \"agent\": %s", decoded["actor_type"], payload)
}
pred := classifyAssignmentWake(RunReasonTaskAssigned, payload)
if !pred.inScope || !pred.taskAttributed {
t.Fatalf("classifyAssignmentWake(%s) = %+v, want in-scope and attributed", payload, pred)
}
if pred.taskID != task.ID {
t.Fatalf("predicate taskID = %q, want %q", pred.taskID, task.ID)
}
}
// TestReactToAssigneeChange_UserActorPayloadNeverCarriesAgent pins the
// negative half: a user-initiated assignment produces a payload the
// predicate classifies out of scope.
func TestReactToAssigneeChange_UserActorPayloadNeverCarriesAgent(t *testing.T) {
ss := &SchedulerService{logger: logger.Default()}
task := &TaskSnapshot{ID: "task-inscope-2", WorkspaceID: "ws-1", State: "TODO"}
change := TaskMutation{ActorID: "", ActorType: "user"}
var calls []recordedQueueCall
queue := func(agentID string, c RunContext) {
calls = append(calls, recordedQueueCall{agentID: agentID, ctx: c})
}
ss.reactToAssigneeChange(task, "new-assignee", change, queue, &ApplyTaskMutationResult{})
if len(calls) != 1 {
t.Fatalf("got %d queue calls, want 1", len(calls))
}
payload, err := encodeRunContext(calls[0].ctx)
if err != nil {
t.Fatalf("encodeRunContext: %v", err)
}
pred := classifyAssignmentWake(RunReasonTaskAssigned, payload)
if pred.inScope {
t.Fatalf("classifyAssignmentWake(%s) = %+v, want out of scope for a user actor", payload, pred)
}
}
@@ -0,0 +1,422 @@
package scheduler
import (
"context"
"fmt"
"testing"
"time"
officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
// agentActorPayload and userActorPayload build task_assigned payloads for
// the two actor shapes the predicate distinguishes.
func agentActorPayload(taskID string) string {
return `{"task_id":"` + taskID + `","actor_type":"agent"}`
}
func userActorPayload(taskID string) string {
return `{"task_id":"` + taskID + `","actor_type":"user"}`
}
// admitNAgentWakes queues AssignmentWakeAllowanceN agent-actor
// task_assigned wakes for taskID through the real QueueRun path, one per
// distinct agent instance. CoalesceRun scopes its merge by
// agent_profile_id, so — unlike repeating the same agent — calls from N
// different agents within the 5s coalescing window never merge into each
// other, giving N genuinely separate admitted rows the way N different
// agents assigned to the same task would in production. Returns the
// agent IDs used, so a caller can issue one more call from any of them.
func admitNAgentWakes(t *testing.T, repo *officesqlite.Repository, ss *SchedulerService, agentPrefix, taskID string) []string {
t.Helper()
ctx := context.Background()
agentIDs := make([]string, AssignmentWakeAllowanceN)
for i := 0; i < AssignmentWakeAllowanceN; i++ {
agentID := fmt.Sprintf("%s-%d", agentPrefix, i)
createChildrenCompletedAgent(t, repo, agentID)
agentIDs[i] = agentID
outcome, err := ss.QueueRun(ctx, agentID, RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("admitNAgentWakes[%d]: %v", i, err)
}
if outcome != runsservice.QueueOutcomeQueued {
t.Fatalf("admitNAgentWakes[%d] outcome = %q, want queued", i, outcome)
}
}
return agentIDs
}
// TestQueueRun_AssignmentRateLimit_AdmitsNThenRefusesNPlus1 is the
// headline integration test for REQ-OFFICE-ASSIGN-RATE-001: the first N
// agent-initiated task_assigned wakes for a task insert a row each; the
// N+1th is refused with QueueOutcomeRateLimited and inserts nothing.
func TestQueueRun_AssignmentRateLimit_AdmitsNThenRefusesNPlus1(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-1"
admitNAgentWakes(t, repo, ss, "rl1-agent", taskID)
if got := runsCountForReason(t, ss, RunReasonTaskAssigned); got != AssignmentWakeAllowanceN {
t.Fatalf("runs inserted = %d, want %d", got, AssignmentWakeAllowanceN)
}
// A 6th, previously-unused agent still hits the task-scoped allowance.
createChildrenCompletedAgent(t, repo, "rl1-agent-extra")
outcome, err := ss.QueueRun(ctx, "rl1-agent-extra", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun[N+1]: %v", err)
}
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("QueueRun[N+1] outcome = %q, want rate_limited", outcome)
}
if got := runsCountForReason(t, ss, RunReasonTaskAssigned); got != AssignmentWakeAllowanceN {
t.Fatalf("runs inserted after refusal = %d, want %d (refusal must not insert)", got, AssignmentWakeAllowanceN)
}
}
// TestQueueRun_AssignmentRateLimit_UserActorAdmitsAfterExhaustion covers
// AC-OFFICE-ASSIGN-RATE-001.1: the gate applies only to agent-initiated
// wakes. A user-actor wake for the same task admits even after the
// allowance is exhausted.
func TestQueueRun_AssignmentRateLimit_UserActorAdmitsAfterExhaustion(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-2"
admitNAgentWakes(t, repo, ss, "rl2-agent", taskID)
// A brand new agent, never used above: reusing an exhausted agent's ID
// here would coalesce with that agent's own already-queued row
// (CoalesceRun's 5s window) rather than exercising the rate-limit gate.
createChildrenCompletedAgent(t, repo, "rl2-agent-new")
outcome, err := ss.QueueRun(ctx, "rl2-agent-new", RunReasonTaskAssigned, userActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun: %v", err)
}
if outcome != runsservice.QueueOutcomeQueued {
t.Fatalf("outcome = %q, want queued (user-actor wakes are never refused)", outcome)
}
}
// TestQueueRun_AssignmentRateLimit_NoActorTypeAdmitsAfterExhaustion covers
// the 4 out-of-scope producers' payload shape ({"task_id":...}, no
// actor_type): it must admit even after the allowance is exhausted, and
// must never move any of the three office_assignment_rate_limit_total
// reason counters — asserted directly here as a before/after delta, since
// the prior "proven indirectly" comment cited a different test that exits
// at the *reason* predicate step (a non-task_assigned RunReason) rather
// than exercising this actor-classification step at all.
func TestQueueRun_AssignmentRateLimit_NoActorTypeAdmitsAfterExhaustion(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-3"
admitNAgentWakes(t, repo, ss, "rl3-agent", taskID)
beforeExhausted := assignmentRateLimitCounterValue(t, "reason=allowance_exhausted")
beforeReadFailed := assignmentRateLimitCounterValue(t, "reason=count_read_failed")
beforeUnattributed := assignmentRateLimitCounterValue(t, "reason=task_unattributed")
createChildrenCompletedAgent(t, repo, "rl3-agent-new")
outcome, err := ss.QueueRun(ctx, "rl3-agent-new", RunReasonTaskAssigned, `{"task_id":"`+taskID+`"}`, "")
if err != nil {
t.Fatalf("QueueRun: %v", err)
}
if outcome != runsservice.QueueOutcomeQueued {
t.Fatalf("outcome = %q, want queued (no actor_type is out of scope)", outcome)
}
if got := assignmentRateLimitCounterValue(t, "reason=allowance_exhausted"); got != beforeExhausted {
t.Fatalf("allowance_exhausted counter = %d, want unchanged at %d (out-of-scope admit must not move it)", got, beforeExhausted)
}
if got := assignmentRateLimitCounterValue(t, "reason=count_read_failed"); got != beforeReadFailed {
t.Fatalf("count_read_failed counter = %d, want unchanged at %d (out-of-scope admit must not move it)", got, beforeReadFailed)
}
if got := assignmentRateLimitCounterValue(t, "reason=task_unattributed"); got != beforeUnattributed {
t.Fatalf("task_unattributed counter = %d, want unchanged at %d (out-of-scope admit must not move it)", got, beforeUnattributed)
}
}
// TestQueueRun_AssignmentRateLimit_SharedAcrossAgents covers
// AC-OFFICE-ASSIGN-RATE-001.3: the allowance is scoped to the task, not
// the acting agent — N wakes from N distinct agents against the same
// task exhaust one shared allowance, and a further wake from any agent
// (including one already used) is refused.
func TestQueueRun_AssignmentRateLimit_SharedAcrossAgents(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-4"
agentIDs := admitNAgentWakes(t, repo, ss, "rl4-agent", taskID)
// Past CoalesceRun's 5s window but inside the 10-minute rate-limit
// window, so reusing agentIDs[0] below reaches the gate instead of
// coalescing with that agent's own already-queued row.
ageRunsRequestedAt(t, ss, time.Minute)
outcome, err := ss.QueueRun(ctx, agentIDs[0], RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun[N+1] agentIDs[0]: %v", err)
}
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("outcome = %q, want rate_limited (allowance is task-scoped, shared across agents)", outcome)
}
createChildrenCompletedAgent(t, repo, "rl4-agent-new")
outcome, err = ss.QueueRun(ctx, "rl4-agent-new", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun[N+1] new agent: %v", err)
}
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("outcome = %q, want rate_limited for a brand new agent too", outcome)
}
}
// TestQueueRun_AssignmentRateLimit_RefusalsDoNotConsumeAllowance proves a
// refusal is not itself counted toward the allowance: repeated refused
// attempts never change the outcome or insert a row.
func TestQueueRun_AssignmentRateLimit_RefusalsDoNotConsumeAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-5"
agentIDs := admitNAgentWakes(t, repo, ss, "rl5-agent", taskID)
// Past CoalesceRun's 5s window but inside the 10-minute rate-limit
// window, so the repeated calls below reach the gate instead of
// coalescing with agentIDs[0]'s own already-queued row.
ageRunsRequestedAt(t, ss, time.Minute)
for i := 0; i < 3; i++ {
outcome, err := ss.QueueRun(ctx, agentIDs[0], RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun refusal[%d]: %v", i, err)
}
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("QueueRun refusal[%d] outcome = %q, want rate_limited", i, outcome)
}
}
if got := runsCountForReason(t, ss, RunReasonTaskAssigned); got != AssignmentWakeAllowanceN {
t.Fatalf("runs inserted = %d, want %d (repeated refusals must not insert)", got, AssignmentWakeAllowanceN)
}
}
// TestQueueRun_AssignmentRateLimit_CoalescedWakeDoesNotConsumeAllowance
// covers the ordering contract: coalescing happens before the rate-limit
// gate, so a coalesced wake (merged into an already-queued row) never
// reaches the gate and never inserts a second row.
func TestQueueRun_AssignmentRateLimit_CoalescedWakeDoesNotConsumeAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-1")
ctx := context.Background()
taskID := "rl-task-6"
first, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun first: %v", err)
}
if first != runsservice.QueueOutcomeQueued {
t.Fatalf("first outcome = %q, want queued", first)
}
for i := 0; i < AssignmentWakeAllowanceN+2; i++ {
outcome, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun coalesce[%d]: %v", i, err)
}
if outcome != runsservice.QueueOutcomeCoalesced {
t.Fatalf("QueueRun coalesce[%d] outcome = %q, want coalesced", i, outcome)
}
}
if got := runsCountForReason(t, ss, RunReasonTaskAssigned); got != 1 {
t.Fatalf("runs inserted = %d, want 1 (every repeat coalesced into the first row)", got)
}
}
// TestQueueRun_AssignmentRateLimit_DedupedWakeDoesNotConsumeAllowance
// covers the same ordering contract for the idempotency-key path: a
// windowed-duplicate wake is suppressed before the rate-limit gate runs.
func TestQueueRun_AssignmentRateLimit_DedupedWakeDoesNotConsumeAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-1")
ctx := context.Background()
taskID := "rl-task-7"
first, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "dedup-key-1")
if err != nil {
t.Fatalf("QueueRun first: %v", err)
}
if first != runsservice.QueueOutcomeQueued {
t.Fatalf("first outcome = %q, want queued", first)
}
for i := 0; i < AssignmentWakeAllowanceN+2; i++ {
outcome, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "dedup-key-1")
if err != nil {
t.Fatalf("QueueRun dedup[%d]: %v", i, err)
}
if outcome != runsservice.QueueOutcomeDeduped {
t.Fatalf("QueueRun dedup[%d] outcome = %q, want deduped", i, outcome)
}
}
if got := runsCountForReason(t, ss, RunReasonTaskAssigned); got != 1 {
t.Fatalf("runs inserted = %d, want 1 (every repeat deduped)", got)
}
}
// TestQueueRun_AssignmentRateLimit_KeylessRedeliveryConsumesAllowanceAgain
// covers AC-OFFICE-ASSIGN-RATE-002.5's second clause: a keyless wake has no
// dedup identity, so redelivering the same keyless occurrence consumes the
// allowance again rather than deduping — the opposite of
// TestQueueRun_AssignmentRateLimit_DedupedWakeDoesNotConsumeAllowance, which
// covers the keyed case.
func TestQueueRun_AssignmentRateLimit_KeylessRedeliveryConsumesAllowanceAgain(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-1")
ctx := context.Background()
taskID := "rl-task-11"
first, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun first: %v", err)
}
if first != runsservice.QueueOutcomeQueued {
t.Fatalf("first outcome = %q, want queued", first)
}
// Past CoalesceRun's 5s window but inside the 10-minute rate-limit
// window, so the redelivery below reaches the gate as a genuinely new
// wake instead of coalescing with the first row.
ageRunsRequestedAt(t, ss, time.Minute)
second, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun redelivery: %v", err)
}
if second != runsservice.QueueOutcomeQueued {
t.Fatalf("redelivery outcome = %q, want queued (keyless redelivery has no identity to dedup on)", second)
}
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, taskID, RunReasonTaskAssigned, time.Now().UTC().Add(-AssignmentWakeAllowanceWindow), time.Now().UTC())
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 2 {
t.Fatalf("count = %d, want 2 (both keyless deliveries consumed the allowance independently)", count)
}
}
// TestQueueRun_AssignmentRateLimit_OrderIndependenceAtBoundary covers
// AC-OFFICE-ASSIGN-RATE-001's window-count query having no ORDER BY: N
// admitted wakes sharing one requested_at timestamp (a real possibility
// since requested_at has millisecond-or-coarser resolution) must all
// count, regardless of any incidental row order.
func TestQueueRun_AssignmentRateLimit_OrderIndependenceAtBoundary(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-8"
// Insert N admitted wakes directly (all as agent-1, all sharing one
// requested_at timestamp), then confirm the (N+1)th QueueRun call from
// a different agent still refuses. createAssignmentWakeRun's rows
// don't go through CoalesceRun, so using the same agent for all of
// them here is fine; the follow-up call below uses a fresh agent so
// it reaches the gate instead of coalescing with agent-1's row.
shared := time.Now().UTC()
for i := 0; i < AssignmentWakeAllowanceN; i++ {
createAssignmentWakeRun(t, repo, taskID, shared)
}
createChildrenCompletedAgent(t, repo, "rl8-agent-new")
outcome, err := ss.QueueRun(ctx, "rl8-agent-new", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun: %v", err)
}
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("outcome = %q, want rate_limited", outcome)
}
}
// TestQueueRun_AssignmentRateLimit_MixedActorCoalesce_AgentOverwritesUser
// pins the "## Accepted consequence" mixed-actor coalesce behavior in one
// direction: a user-actor wake queued first, then coalesced with an
// agent-actor wake for the same task within the coalescing window, ends
// up counted as an agent-initiated admission even though only the
// coalesced (not a newly inserted) row exists — a bounded one-high
// miscount, not a bug to fix.
func TestQueueRun_AssignmentRateLimit_MixedActorCoalesce_AgentOverwritesUser(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-1")
ctx := context.Background()
taskID := "rl-task-9"
first, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, userActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun user: %v", err)
}
if first != runsservice.QueueOutcomeQueued {
t.Fatalf("first outcome = %q, want queued", first)
}
second, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun agent (coalesces): %v", err)
}
if second != runsservice.QueueOutcomeCoalesced {
t.Fatalf("second outcome = %q, want coalesced", second)
}
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, taskID, RunReasonTaskAssigned, time.Now().UTC().Add(-time.Minute), time.Now().UTC())
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("count = %d, want 1 (the coalesced row now reads actor_type=agent)", count)
}
}
// TestQueueRun_AssignmentRateLimit_MixedActorCoalesce_UserOverwritesAgent
// pins the same consequence in the other direction: an agent-actor wake
// queued first (an admitted, counted wake), then coalesced with a
// user-actor wake for the same task, ends up NOT counted — a bounded
// one-low miscount.
func TestQueueRun_AssignmentRateLimit_MixedActorCoalesce_UserOverwritesAgent(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-1")
ctx := context.Background()
taskID := "rl-task-10"
first, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun agent: %v", err)
}
if first != runsservice.QueueOutcomeQueued {
t.Fatalf("first outcome = %q, want queued", first)
}
second, err := ss.QueueRun(ctx, "agent-1", RunReasonTaskAssigned, userActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun user (coalesces): %v", err)
}
if second != runsservice.QueueOutcomeCoalesced {
t.Fatalf("second outcome = %q, want coalesced", second)
}
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, taskID, RunReasonTaskAssigned, time.Now().UTC().Add(-time.Minute), time.Now().UTC())
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 0 {
t.Fatalf("count = %d, want 0 (the coalesced row now reads actor_type=user)", count)
}
}
@@ -0,0 +1,200 @@
package scheduler
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/kandev/kandev/internal/office/models"
)
// TestClassifyAssignmentWake covers the three-step predicate in the
// system design's "## The predicate": reason == task_assigned,
// actor_type == "agent", the wake names a task. Table-driven per the
// design's non-symmetric failure-shape table.
func TestClassifyAssignmentWake(t *testing.T) {
tests := []struct {
name string
reason string
payload string
wantInScope bool
wantAttributed bool
wantTaskID string
wantActorID string
}{
{
name: "wrong reason is out of scope",
reason: RunReasonTaskComment,
payload: `{"task_id":"t1","actor_type":"agent"}`,
},
{
name: "actor_type user is out of scope",
reason: RunReasonTaskAssigned,
payload: `{"task_id":"t1","actor_type":"user"}`,
},
{
name: "actor_type absent is out of scope",
reason: RunReasonTaskAssigned,
payload: `{"task_id":"t1"}`,
},
{
name: "actor_type other value is out of scope",
reason: RunReasonTaskAssigned,
payload: `{"task_id":"t1","actor_type":"system"}`,
},
{
name: "unparseable payload is out of scope, never unattributed",
reason: RunReasonTaskAssigned,
payload: `not json`,
},
{
name: "both actor_type and task_id absent is out of scope on the actor step",
reason: RunReasonTaskAssigned,
payload: `{}`,
},
{
name: "task_id absent is in scope, unattributed",
reason: RunReasonTaskAssigned,
payload: `{"actor_type":"agent","actor_id":"a1"}`,
wantInScope: true,
wantActorID: "a1",
},
{
name: "task_id null is in scope, unattributed",
reason: RunReasonTaskAssigned,
payload: `{"task_id":null,"actor_type":"agent"}`,
wantInScope: true,
},
{
name: "task_id empty string is in scope, unattributed",
reason: RunReasonTaskAssigned,
payload: `{"task_id":"","actor_type":"agent"}`,
wantInScope: true,
},
{
name: "task_id non-string is in scope, unattributed",
reason: RunReasonTaskAssigned,
payload: `{"task_id":42,"actor_type":"agent"}`,
wantInScope: true,
},
{
name: "task_id valid string is in scope, attributed",
reason: RunReasonTaskAssigned,
payload: `{"task_id":"t1","actor_type":"agent","actor_id":"a1"}`,
wantInScope: true,
wantAttributed: true,
wantTaskID: "t1",
wantActorID: "a1",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got := classifyAssignmentWake(tc.reason, tc.payload)
if got.inScope != tc.wantInScope {
t.Fatalf("inScope = %v, want %v", got.inScope, tc.wantInScope)
}
if got.taskAttributed != tc.wantAttributed {
t.Fatalf("taskAttributed = %v, want %v", got.taskAttributed, tc.wantAttributed)
}
if got.taskID != tc.wantTaskID {
t.Fatalf("taskID = %q, want %q", got.taskID, tc.wantTaskID)
}
if got.actorID != tc.wantActorID {
t.Fatalf("actorID = %q, want %q", got.actorID, tc.wantActorID)
}
})
}
}
// TestCheckAssignmentWakeAllowance_OutOfScopeNeverTouchesDB proves an
// out-of-scope wake short-circuits before any repo call by using a
// scheduler whose repo is nil — a DB touch would panic.
func TestCheckAssignmentWakeAllowance_OutOfScopeNeverTouchesDB(t *testing.T) {
ss := &SchedulerService{}
refused := ss.checkAssignmentWakeAllowance(context.Background(), "agent-1", RunReasonTaskComment, `{"task_id":"t1","actor_type":"agent"}`)
if refused {
t.Fatal("out-of-scope wake must never be refused")
}
}
// TestCheckAssignmentWakeAllowance_AdmitsUnderAllowance covers the silent
// admit path: an in-scope wake with room left in the allowance is
// admitted with no counter movement.
func TestCheckAssignmentWakeAllowance_AdmitsUnderAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newReactivityTestScheduler(t, repo)
ctx := context.Background()
for i := 0; i < AssignmentWakeAllowanceN-1; i++ {
createAssignmentWakeRun(t, repo, "under-allowance-task", time.Now().UTC())
}
refused := ss.checkAssignmentWakeAllowance(ctx, "agent-1", RunReasonTaskAssigned,
`{"task_id":"under-allowance-task","actor_type":"agent"}`)
if refused {
t.Fatal("wake under the allowance must be admitted")
}
}
// TestCheckAssignmentWakeAllowance_RefusesAtAllowance covers the refusal
// path: N admitted wakes already exist in the window, so the N+1th must
// be refused.
func TestCheckAssignmentWakeAllowance_RefusesAtAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newReactivityTestScheduler(t, repo)
ctx := context.Background()
for i := 0; i < AssignmentWakeAllowanceN; i++ {
createAssignmentWakeRun(t, repo, "at-allowance-task", time.Now().UTC())
}
refused := ss.checkAssignmentWakeAllowance(ctx, "agent-1", RunReasonTaskAssigned,
`{"task_id":"at-allowance-task","actor_type":"agent"}`)
if !refused {
t.Fatal("wake at the allowance must be refused")
}
}
// TestCheckAssignmentWakeAllowance_TaskUnattributedAdmits covers
// AC-OFFICE-ASSIGN-RATE-002.3: an otherwise in-scope wake whose task
// cannot be determined is admitted, not refused.
func TestCheckAssignmentWakeAllowance_TaskUnattributedAdmits(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newReactivityTestScheduler(t, repo)
ctx := context.Background()
refused := ss.checkAssignmentWakeAllowance(ctx, "agent-1", RunReasonTaskAssigned,
`{"actor_type":"agent"}`)
if refused {
t.Fatal("a wake whose task cannot be determined must fail open (admit)")
}
}
// createAssignmentWakeRun inserts an admitted agent-initiated
// task_assigned run for taskID, backdated to requestedAt, using the
// established repo pattern (SetRunRequestedAtForTest) instead of
// time.Sleep.
func createAssignmentWakeRun(t *testing.T, repo interface {
CreateRun(ctx context.Context, r *models.Run) error
SetRunRequestedAtForTest(ctx context.Context, runID string, ts time.Time) error
}, taskID string, requestedAt time.Time) {
t.Helper()
ctx := context.Background()
r := &models.Run{
ID: uuid.New().String(),
AgentProfileID: "agent-1",
Reason: RunReasonTaskAssigned,
Payload: `{"task_id":"` + taskID + `","actor_type":"agent"}`,
Status: RunStatusQueued,
CoalescedCount: 1,
}
if err := repo.CreateRun(ctx, r); err != nil {
t.Fatalf("create run: %v", err)
}
if err := repo.SetRunRequestedAtForTest(ctx, r.ID, requestedAt); err != nil {
t.Fatalf("set requested_at: %v", err)
}
}
@@ -0,0 +1,61 @@
package scheduler
import (
"context"
"testing"
"time"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
// TestQueueRun_AssignmentRateLimit_AdmitsAgainAfterFullWindowElapses covers
// the allowance's actual recovery: every other test in this package that
// calls ageRunsRequestedAt only ages runs past CoalesceRun's 5-second
// window, never past the 10-minute AssignmentWakeAllowanceWindow itself,
// so nothing previously proved the count query's requested_at cutoff
// actually lets the allowance refill once the exhausting wakes age out.
func TestQueueRun_AssignmentRateLimit_AdmitsAgainAfterFullWindowElapses(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
ctx := context.Background()
taskID := "rl-task-window-recovery"
agentIDs := admitNAgentWakes(t, repo, ss, "rlw-agent", taskID)
// Confirm exhaustion before aging, so a later admit can only be
// explained by the window recovering, not by never having been
// exhausted in the first place.
createChildrenCompletedAgent(t, repo, "rlw-agent-probe")
probeOutcome, err := ss.QueueRun(ctx, "rlw-agent-probe", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun probe: %v", err)
}
if probeOutcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("probe outcome = %q, want rate_limited before aging", probeOutcome)
}
// Age every persisted run past the full allowance window (not just
// CoalesceRun's 5s window), so the window-count query sees none of
// the exhausting wakes anymore.
ageRunsRequestedAt(t, ss, AssignmentWakeAllowanceWindow+time.Minute)
for i, agentID := range agentIDs {
outcome, err := ss.QueueRun(ctx, agentID, RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun after window recovery [%d]: %v", i, err)
}
if outcome != runsservice.QueueOutcomeQueued {
t.Fatalf("QueueRun after window recovery [%d] outcome = %q, want queued (the aged-out wakes must no longer count)", i, outcome)
}
}
// The N freshly-admitted wakes above re-exhaust the allowance.
createChildrenCompletedAgent(t, repo, "rlw-agent-probe-2")
finalOutcome, err := ss.QueueRun(ctx, "rlw-agent-probe-2", RunReasonTaskAssigned, agentActorPayload(taskID), "")
if err != nil {
t.Fatalf("QueueRun final probe: %v", err)
}
if finalOutcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("final probe outcome = %q, want rate_limited (re-exhausted after recovery)", finalOutcome)
}
}
@@ -3,6 +3,7 @@ package scheduler
import (
"context"
"testing"
"time"
)
// TestApplyTaskMutation_SameAgentRepeatAssignment_QueuesWithoutInterrupt
@@ -97,6 +98,109 @@ func TestApplyTaskMutation_DifferentAgentReassignment_InterruptsPreviousAssignee
}
}
// TestApplyTaskMutation_DifferentAgentReassignment_RateLimitedStillInterrupts
// pins AC-OFFICE-ASSIGN-RATE-001.7 through the real dispatcher, not just
// through checkAssignmentWakeAllowance or QueueRun directly: when the
// task's agent-initiated assignment allowance is already exhausted, the
// reassignment wake itself is refused (absent from res.Runs), but the
// mutation is not aborted — the previous assignee's session interrupt
// (res.InterruptSessionID, set in reactToAssigneeChange before queue() is
// ever called) still fires, and ApplyTaskMutation returns no error.
func TestApplyTaskMutation_DifferentAgentReassignment_RateLimitedStillInterrupts(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-old")
createChildrenCompletedAgent(t, repo, "agent-new")
taskID := "task-reassign-rate-limited"
for i := 0; i < AssignmentWakeAllowanceN; i++ {
createAssignmentWakeRun(t, repo, taskID, time.Now().UTC())
}
task := &TaskSnapshot{
ID: taskID,
WorkspaceID: "ws-1",
AssigneeAgentProfileID: "agent-old",
}
newAssignee := "agent-new"
gen := int64(1)
change := TaskMutation{
NewAssigneeID: &newAssignee,
AssignmentGeneration: &gen,
ActorID: "agent-old",
ActorType: "agent",
}
res, err := ss.ApplyTaskMutation(context.Background(), task, change)
if err != nil {
t.Fatalf("ApplyTaskMutation: %v", err)
}
if res.InterruptSessionID != task.ID {
t.Fatalf("InterruptSessionID = %q, want %q (a refusal must not abort the mutation's other effects)",
res.InterruptSessionID, task.ID)
}
for _, r := range res.Runs {
if r.AgentID == "agent-new" && r.Reason == RunReasonTaskAssigned {
t.Fatalf("expected the rate-limited reassignment wake to be refused (absent from res.Runs), got %+v", res.Runs)
}
}
}
// TestApplyTaskMutation_Unassignment_NeitherRefusesNorConsumesAllowance is
// AC-OFFICE-ASSIGN-RATE-001.9: reactToAssigneeChange returns before ever
// calling queue() when NewAssigneeID is "", so an unassignment on a task
// whose allowance is already exhausted must not be refused (there is no
// wake to refuse) and must not move the rate-limit counter (there is no
// wake to count). The previous assignee's session interrupt still fires —
// unassignment is a real handoff away, not a no-op.
func TestApplyTaskMutation_Unassignment_NeitherRefusesNorConsumesAllowance(t *testing.T) {
repo := newReactivityTestRepo(t)
ss := newChildrenCompletedTestScheduler(t, repo)
createChildrenCompletedAgent(t, repo, "agent-old")
taskID := "task-unassign-rate-limit"
for i := 0; i < AssignmentWakeAllowanceN; i++ {
createAssignmentWakeRun(t, repo, taskID, time.Now().UTC())
}
before := assignmentRateLimitCounterValue(t, "reason=allowance_exhausted")
task := &TaskSnapshot{
ID: taskID,
WorkspaceID: "ws-1",
AssigneeAgentProfileID: "agent-old",
}
newAssignee := ""
gen := int64(1)
change := TaskMutation{
NewAssigneeID: &newAssignee,
AssignmentGeneration: &gen,
ActorID: "agent-old",
ActorType: "agent",
}
res, err := ss.ApplyTaskMutation(context.Background(), task, change)
if err != nil {
t.Fatalf("ApplyTaskMutation: %v", err)
}
if res.InterruptSessionID != task.ID {
t.Fatalf("InterruptSessionID = %q, want %q (unassigning the current assignee still interrupts their session)",
res.InterruptSessionID, task.ID)
}
for _, r := range res.Runs {
if r.Reason == RunReasonTaskAssigned {
t.Fatalf("unassignment must never attempt a task_assigned wake, got %+v", res.Runs)
}
}
after := assignmentRateLimitCounterValue(t, "reason=allowance_exhausted")
if after != before {
t.Fatalf("allowance_exhausted counter = %d, want unchanged at %d (unassignment must not be refused or counted)", after, before)
}
}
// TestApplyTaskMutation_StatusChangeAndCommentDispatch drives
// ApplyTaskMutation's NewStatus and Comment branches together (an unblock
// plus an attached comment), covering the dispatcher itself rather than only
@@ -153,6 +153,16 @@ const CoalesceWindowSeconds = 5
// IdempotencyWindowHours is the deduplication window.
const IdempotencyWindowHours = 24
// AssignmentWakeAllowanceN and AssignmentWakeAllowanceWindow are the fixed
// N and W of REQ-OFFICE-ASSIGN-RATE-001: at most N agent-initiated
// assignment wakes admitted per task within any rolling window of duration
// W. Fixed values of this capability rather than operator-configurable;
// referenced by the tests rather than restated.
const (
AssignmentWakeAllowanceN = 5
AssignmentWakeAllowanceWindow = 10 * time.Minute
)
// TaskStarter launches agent sessions on behalf of the office scheduler.
// Implemented by the orchestrator; the scheduler depends only on this interface.
type TaskStarter interface {
@@ -357,6 +367,18 @@ func (ss *SchedulerService) queueRun(
}
}
// Assignment wake rate limit (REQ-OFFICE-ASSIGN-RATE-001..003).
// Unconditional — unlike CoalesceRun above, this gate is not skipped
// when waveKey is set: the only reason in this codebase that sets a
// WaveKey is RunReasonTaskChildrenCompleted, never a task_assigned
// wake, but the gate does not rely on that coincidence to stay
// correct. It sits here, after the recent-duplicate lookup and the
// coalescing attempt and before CreateRun, so "admitted" means
// exactly "a row was inserted".
if refused := ss.checkAssignmentWakeAllowance(ctx, agentInstanceID, reason, payload); refused {
return runsservice.QueueOutcomeRateLimited, nil
}
var idemKeyPtr *string
if idempotencyKey != "" {
idemKeyPtr = &idempotencyKey
@@ -0,0 +1,51 @@
package service_test
// TestProducer_EventSubscriber_TaskAssignedNeverCarriesAgentActorType and
// its siblings in internal/office/onboarding, internal/office/service
// (scheduler_recovery), and internal/orchestrator pin
// AC-OFFICE-ASSIGN-RATE-001.12: only one of the five task_assigned
// producers — the mutation-reactivity path in internal/office/scheduler —
// can ever carry actor_type: "agent", because RunContext is the only
// payload-building type with that field. This file's test covers
// event_subscribers.go's queueTaskAssignedRun; scheduler_recovery_test.go
// covers recoverUnstartedTasks in this same package.
import (
"context"
"encoding/json"
"testing"
)
func TestProducer_EventSubscriber_TaskAssignedNeverCarriesAgentActorType(t *testing.T) {
svc, eb := newTestServiceWithBus(t)
ctx := context.Background()
createTestAgent(t, svc, "ws-1", "worker-scope")
insertTestTask(t, svc, "task-scope", "ws-1")
svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-scope")
publishTaskAssigned(t, ctx, eb, "task-scope", "worker-scope", gen(0))
runs, err := svc.ListRuns(ctx, "ws-1")
if err != nil {
t.Fatalf("list runs: %v", err)
}
found := false
for _, run := range runs {
if run.AgentProfileID != "worker-scope" || run.Reason != "task_assigned" {
continue
}
found = true
var payload map[string]any
if err := json.Unmarshal([]byte(run.Payload), &payload); err != nil {
t.Fatalf("decode run payload: %v", err)
}
if actorType, present := payload["actor_type"]; present && actorType == "agent" {
t.Fatalf("event-subscriber task_assigned payload carries actor_type=agent, "+
"which would put it in the assignment-wake-rate-limit's scope: %v", payload)
}
}
if !found {
t.Fatal("expected a task_assigned run for worker-scope, found none")
}
}
@@ -2,6 +2,7 @@ package service_test
import (
"context"
"encoding/json"
"testing"
"github.com/kandev/kandev/internal/office/service"
@@ -60,3 +61,42 @@ func TestOfficeRecoveryHandlerActivatesAfterOfficeProjectAdoption(t *testing.T)
t.Fatalf("recovery runs = %#v, want one task_assigned run", runs)
}
}
// TestOfficeRecoveryHandler_TaskAssignedNeverCarriesAgentActorType pins
// AC-OFFICE-ASSIGN-RATE-001.12 for recoverUnstartedTasks: its
// task_assigned payload is a bare {"task_id":...} with no actor_type
// field, so it can never fall into the assignment-wake-rate-limit's
// scope.
func TestOfficeRecoveryHandler_TaskAssignedNeverCarriesAgentActorType(t *testing.T) {
svc := newTestService(t)
ctx := context.Background()
createTestAgent(t, svc, "ws-1", "worker-office")
insertTestTask(t, svc, "task-office-project", "ws-1")
svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "task-office-project")
setTestTaskAssignee(t, svc, "task-office-project", "worker-office")
svc.ExecSQL(t, `
INSERT INTO office_projects (id, workspace_id, name, created_at, updated_at)
VALUES ('office-project', 'ws-1', 'Office Project', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
`)
handler := service.NewOfficeRecoveryHandler(service.NewSchedulerIntegration(svc, 0))
if err := handler.Tick(ctx); err != nil {
t.Fatalf("recovery tick: %v", err)
}
runs, err := svc.ListRuns(ctx, "ws-1")
if err != nil {
t.Fatalf("list runs: %v", err)
}
if len(runs) != 1 {
t.Fatalf("recovery runs = %#v, want one task_assigned run", runs)
}
var payload map[string]any
if err := json.Unmarshal([]byte(runs[0].Payload), &payload); err != nil {
t.Fatalf("decode run payload: %v", err)
}
if actorType, present := payload["actor_type"]; present && actorType == "agent" {
t.Fatalf("recovery task_assigned payload carries actor_type=agent: %v", payload)
}
}
@@ -0,0 +1,74 @@
package orchestrator
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/orchestrator/watcher"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/workflow/engine"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// TestQueueOfficeAutoStartRun_RequestCarriesNoActorTypeField pins
// AC-OFFICE-ASSIGN-RATE-001.12 for queueOfficeAutoStartRun: unlike the
// other three out-of-scope producers (which build a raw JSON payload that
// simply omits actor_type), this one is a structural guarantee —
// engine.QueueRunRequest has no ActorType field at all, so no payload it
// builds could carry actor_type: "agent" even by accident. Exercises the
// real handleTaskMovedNoSession -> queueOfficeAutoStartRun path and
// inspects the request the engine.RunQueueAdapter actually received.
func TestQueueOfficeAutoStartRun_RequestCarriesNoActorTypeField(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
now := time.Now().UTC()
requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}))
requireNoError(t, repo.CreateTask(ctx, &models.Task{
ID: "t-producer-scope",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step1",
ProjectID: "proj1",
Title: "Office Task",
Description: "prompt",
State: v1.TaskStateCreated,
AssigneeAgentProfileID: "assignee-profile",
CreatedAt: now,
UpdatedAt: now,
}))
stepGetter := newMockStepGetter()
stepGetter.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Work", Position: 1,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
},
}
svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), failIfLaunched(t))
calls := make(chan engine.QueueRunRequest, 1)
svc.engineRunQueue = &fakeRunQueueAdapter{outcome: engine.QueueOutcomeQueued, calls: calls}
svc.enginePrimary = &fakePrimaryAgentResolver{agentProfileID: "resolved-primary"}
svc.handleTaskMovedNoSession(ctx, watcher.TaskMovedEventData{
TaskID: "t-producer-scope",
ToStepID: "step2",
})
select {
case req := <-calls:
if req.Reason != officeAutoStartRunReason {
t.Fatalf("reason = %q, want %q", req.Reason, officeAutoStartRunReason)
}
if actorType, present := req.Payload["actor_type"]; present {
t.Fatalf("office auto-start payload carries an actor_type key (%v) — "+
"QueueRunRequest has no such field, so this must never happen", actorType)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for QueueRun to be called")
}
}
@@ -0,0 +1,129 @@
package sqlite_test
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/office/models"
)
// TestPostgresCountAgentInitiatedAssignmentWakes_BasicCount is the
// Postgres twin of TestCountAgentInitiatedAssignmentWakes_BasicCount.
// CountAgentInitiatedAssignmentWakes is built from dialect.JSONExtract,
// which emits payload::jsonb->>'task_id' on Postgres versus
// json_extract(payload, '$.task_id') on SQLite — a passing SQLite
// assertion is not evidence the Postgres fragment even parses. Skips
// unless KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresCountAgentInitiatedAssignmentWakes_BasicCount(t *testing.T) {
repo := newTestRepoPostgres(t)
ctx := context.Background()
now := time.Now().UTC()
windowStart := now.Add(-10 * time.Minute)
inWindow := func(id, payload string) {
r := mustCreateRun(t, repo, &models.Run{
ID: id, AgentProfileID: "a1", Reason: "task_assigned",
Payload: payload, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, r.ID, now)
}
inWindow("pg-cnt-agent-1", `{"task_id":"pg-t1","actor_type":"agent"}`)
inWindow("pg-cnt-agent-2", `{"task_id":"pg-t1","actor_type":"agent"}`)
inWindow("pg-cnt-user", `{"task_id":"pg-t1","actor_type":"user"}`)
inWindow("pg-cnt-other-task", `{"task_id":"pg-t2","actor_type":"agent"}`)
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "pg-t1", "task_assigned", windowStart, now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 2 {
t.Fatalf("count = %d, want 2", count)
}
}
// TestPostgresCountAgentInitiatedAssignmentWakes_GuardsNonStringTaskID is
// the Postgres twin of the non-string task_id guard test: Postgres's ->>
// operator converts a stored JSON number to text before comparison, so a
// naive text-only predicate would let a string-valued taskID match a
// stored {"task_id":42} and silently join another task's allowance. This
// is the exact dialect coercion dialect.JSONTypeIsString exists to guard
// against.
func TestPostgresCountAgentInitiatedAssignmentWakes_GuardsNonStringTaskID(t *testing.T) {
repo := newTestRepoPostgres(t)
ctx := context.Background()
now := time.Now().UTC()
r := mustCreateRun(t, repo, &models.Run{
ID: "pg-guard-numeric", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":42,"actor_type":"agent"}`, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, r.ID, now)
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "42", "task_assigned", now.Add(-time.Minute), now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 0 {
t.Fatalf("count = %d, want 0 (a numeric stored task_id must not match a string taskID)", count)
}
}
// TestPostgresCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenExclusiveStart
// is the Postgres twin of the window-boundary test.
func TestPostgresCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenExclusiveStart(t *testing.T) {
repo := newTestRepoPostgres(t)
ctx := context.Background()
windowStart := time.Now().UTC()
atBoundary := mustCreateRun(t, repo, &models.Run{
ID: "pg-win-at-boundary", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"pg-t3","actor_type":"agent"}`, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, atBoundary.ID, windowStart)
insideWindow := mustCreateRun(t, repo, &models.Run{
ID: "pg-win-inside", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"pg-t3","actor_type":"agent"}`, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, insideWindow.ID, windowStart.Add(time.Millisecond))
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "pg-t3", "task_assigned", windowStart, windowStart.Add(time.Millisecond))
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("count = %d, want 1 (only the row strictly after windowStart)", count)
}
}
// TestPostgresCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenInclusiveEnd
// is the Postgres twin of the window's upper-bound test: a row at the
// evaluation instant counts, a row strictly after it does not.
func TestPostgresCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenInclusiveEnd(t *testing.T) {
repo := newTestRepoPostgres(t)
ctx := context.Background()
evaluationInstant := time.Now().UTC()
windowStart := evaluationInstant.Add(-10 * time.Minute)
atBoundary := mustCreateRun(t, repo, &models.Run{
ID: "pg-win-end-at-boundary", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"pg-t4","actor_type":"agent"}`, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, atBoundary.ID, evaluationInstant)
future := mustCreateRun(t, repo, &models.Run{
ID: "pg-win-end-future", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"pg-t4","actor_type":"agent"}`, Status: "queued", CoalescedCount: 1,
})
setRequestedAt(t, repo, future.ID, evaluationInstant.Add(time.Millisecond))
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "pg-t4", "task_assigned", windowStart, evaluationInstant)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("count = %d, want 1 (the at-boundary row counts, the future row does not)", count)
}
}
@@ -0,0 +1,196 @@
package sqlite_test
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/office/models"
)
// TestCountAgentInitiatedAssignmentWakes_BasicCount covers
// AC-OFFICE-ASSIGN-RATE-001.2: only rows matching reason, actor_type
// "agent", the target task, and the window count.
func TestCountAgentInitiatedAssignmentWakes_BasicCount(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
now := time.Now().UTC()
windowStart := now.Add(-10 * time.Minute)
inWindow := func(id, payload string) {
r := mustCreateRun(t, repo, &models.Run{
ID: id, AgentProfileID: "a1", Reason: "task_assigned",
Payload: payload, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, r.ID, now)
}
inWindow("cnt-agent-1", `{"task_id":"t1","actor_type":"agent"}`)
inWindow("cnt-agent-2", `{"task_id":"t1","actor_type":"agent"}`)
inWindow("cnt-user", `{"task_id":"t1","actor_type":"user"}`)
inWindow("cnt-other-task", `{"task_id":"t2","actor_type":"agent"}`)
other := mustCreateRun(t, repo, &models.Run{
ID: "cnt-other-reason", AgentProfileID: "a1", Reason: "task_comment",
Payload: `{"task_id":"t1","actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, other.ID, now)
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "t1", "task_assigned", windowStart, now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 2 {
t.Fatalf("count = %d, want 2", count)
}
}
// TestCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenExclusiveStart
// covers AC-OFFICE-ASSIGN-RATE-001.11: a row whose requested_at is exactly
// windowStart is one window old and must not count; a row one instant
// later must.
func TestCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenExclusiveStart(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
windowStart := time.Now().UTC()
atBoundary := mustCreateRun(t, repo, &models.Run{
ID: "win-at-boundary", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t1","actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, atBoundary.ID, windowStart)
insideWindow := mustCreateRun(t, repo, &models.Run{
ID: "win-inside", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t1","actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, insideWindow.ID, windowStart.Add(time.Millisecond))
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "t1", "task_assigned", windowStart, windowStart.Add(time.Millisecond))
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("count = %d, want 1 (only the row strictly after windowStart)", count)
}
}
// TestCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenInclusiveEnd covers
// AC-OFFICE-ASSIGN-RATE-001.11's other edge: a row whose requested_at is
// exactly the evaluation instant is inside the window and must count; a row
// strictly after the evaluation instant (a future timestamp, reachable under
// clock skew) is outside it and must not.
func TestCountAgentInitiatedAssignmentWakes_WindowIsHalfOpenInclusiveEnd(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
evaluationInstant := time.Now().UTC()
windowStart := evaluationInstant.Add(-10 * time.Minute)
atBoundary := mustCreateRun(t, repo, &models.Run{
ID: "win-end-at-boundary", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t-end","actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, atBoundary.ID, evaluationInstant)
future := mustCreateRun(t, repo, &models.Run{
ID: "win-end-future", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t-end","actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, future.ID, evaluationInstant.Add(time.Millisecond))
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "t-end", "task_assigned", windowStart, evaluationInstant)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("count = %d, want 1 (the at-boundary row counts, the future row does not)", count)
}
}
// TestCountAgentInitiatedAssignmentWakes_GuardsNonStringTaskID covers the
// dialect.JSONTypeIsString guard the design's prose directs (the SQL
// sketch omits it): a stored numeric task_id must not textually match an
// incoming string task_id and silently join another task's allowance.
func TestCountAgentInitiatedAssignmentWakes_GuardsNonStringTaskID(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
now := time.Now().UTC()
r := mustCreateRun(t, repo, &models.Run{
ID: "guard-numeric", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":42,"actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, r.ID, now)
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "42", "task_assigned", now.Add(-time.Minute), now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 0 {
t.Fatalf("count = %d, want 0 (a numeric stored task_id must not match a string taskID)", count)
}
}
// TestCountAgentInitiatedAssignmentWakes_NoStatusFilter covers
// AC-OFFICE-ASSIGN-RATE-001.4's deliberate omission: "admitted" means "a
// runs row was inserted", full stop, so a run that has since completed
// still holds its allowance slot until the window passes.
func TestCountAgentInitiatedAssignmentWakes_NoStatusFilter(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
now := time.Now().UTC()
for _, tc := range []struct {
id string
status models.RunStatus
}{
{"status-queued", models.RunStatusQueued},
{"status-claimed", models.RunStatusClaimed},
{"status-finished", models.RunStatusFinished},
{"status-failed", models.RunStatusFailed},
} {
r := mustCreateRun(t, repo, &models.Run{
ID: tc.id, AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t-status","actor_type":"agent"}`,
Status: tc.status, CoalescedCount: 1,
})
setRequestedAt(t, repo, r.ID, now)
}
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "t-status", "task_assigned", now.Add(-time.Minute), now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 4 {
t.Fatalf("count = %d, want 4 (every status counts)", count)
}
}
// TestCountAgentInitiatedAssignmentWakes_TasklessAndAbsentActorDoNotMatch
// pins that a taskless or actor-less stored row cannot accidentally
// satisfy the count query for a real task/actor pair.
func TestCountAgentInitiatedAssignmentWakes_TasklessAndAbsentActorDoNotMatch(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
now := time.Now().UTC()
taskless := mustCreateRun(t, repo, &models.Run{
ID: "taskless", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"actor_type":"agent"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, taskless.ID, now)
noActor := mustCreateRun(t, repo, &models.Run{
ID: "no-actor", AgentProfileID: "a1", Reason: "task_assigned",
Payload: `{"task_id":"t1"}`, Status: models.RunStatusQueued, CoalescedCount: 1,
})
setRequestedAt(t, repo, noActor.ID, now)
count, err := repo.CountAgentInitiatedAssignmentWakes(ctx, "t1", "task_assigned", now.Add(-time.Minute), now)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 0 {
t.Fatalf("count = %d, want 0", count)
}
}
@@ -611,6 +611,45 @@ func taskIDFromPayload(payload string) (taskID string, invalidTaskID bool) {
return s, false
}
// CountAgentInitiatedAssignmentWakes counts runs for taskID with the given
// reason whose stored payload carries actor_type "agent" and whose
// requested_at falls in the half-open interval (windowStart,
// evaluationInstant], exclusive of its old edge and inclusive of its new
// one. The inclusive upper bound keeps a future-dated row (clock skew
// across writers under Postgres) from counting indefinitely instead of
// aging out with the window it actually belongs to. Deliberately carries
// no status filter: "admitted" is defined as "a runs row was inserted",
// so a run that has since completed still holds its allowance slot until
// the window passes. taskID must already be known non-empty; callers
// with an unattributable task must not reach this method.
func (r *Repository) CountAgentInitiatedAssignmentWakes(
ctx context.Context, taskID, reason string, windowStart, evaluationInstant time.Time,
) (int, error) {
driver := r.ro.DriverName()
taskExtract := dialect.JSONExtract(driver, "payload", "task_id")
actorExtract := dialect.JSONExtract(driver, "payload", "actor_type")
// Guard the stored task_id's JSON type the same way CoalesceRun does:
// Postgres's ->> converts a stored JSON number to text before
// comparison, so an untyped payload with e.g. {"task_id":42} could
// otherwise textually match taskID and join another task's allowance.
query := fmt.Sprintf(`
SELECT COUNT(*) FROM runs
WHERE reason = ?
AND %s
AND %s = ?
AND %s = ?
AND requested_at > ?
AND requested_at <= ?
`, dialect.JSONTypeIsString(driver, "payload", "task_id"), taskExtract, actorExtract)
var count int
err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(query),
reason, taskID, "agent", windowStart, evaluationInstant).Scan(&count)
if err != nil {
return 0, err
}
return count, nil
}
// ClaimNextEligibleRun atomically claims the next queued run,
// skipping runs with a scheduled retry time in the future and
// agents that already have a claimed run. Agent status and cooldown
@@ -0,0 +1,129 @@
package service_test
import (
"errors"
"testing"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"go.uber.org/zap/zaptest/observer"
"github.com/kandev/kandev/internal/common/logger"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
// withObservedWarnLogs swaps the package-level default logger (the only
// logger dedup.go's reporting functions use — dedupLogger() always calls
// logger.Default()) for one backed by a zaptest/observer core, and restores
// the original on cleanup. Not safe for t.Parallel(): the swap is process-
// global for the duration of the test.
func withObservedWarnLogs(t *testing.T) *observer.ObservedLogs {
t.Helper()
core, logs := observer.New(zapcore.WarnLevel)
observed, err := logger.NewFromZap(zap.New(core))
if err != nil {
t.Fatalf("NewFromZap: %v", err)
}
original := logger.Default()
logger.SetDefault(observed)
t.Cleanup(func() { logger.SetDefault(original) })
return logs
}
// TestReportAssignmentRateLimitRefused_LogFields covers
// AC-OFFICE-ASSIGN-RATE-003.3's full field set: task id, assignee profile
// id, acting agent id, observed count and allowance are all present —
// every argument this function receives is logged.
func TestReportAssignmentRateLimitRefused_LogFields(t *testing.T) {
logs := withObservedWarnLogs(t)
runsservice.ReportAssignmentRateLimitRefused("task-log-1", "assignee-log-1", "actor-log-1", 5, 5)
entries := logs.FilterMessage("assignment wake refused (rate limit)").All()
if len(entries) != 1 {
t.Fatalf("warning count = %d, want 1", len(entries))
}
fields := entries[0].ContextMap()
if fields["task_id"] != "task-log-1" ||
fields["assignee_agent_profile_id"] != "assignee-log-1" ||
fields["acting_agent_id"] != "actor-log-1" ||
fields["observed_count"] != int64(5) ||
fields["allowance"] != int64(5) {
t.Fatalf("fields = %#v, want task_id/assignee_agent_profile_id/acting_agent_id/observed_count/allowance all present", fields)
}
}
// TestReportAssignmentRateLimitCountReadFailed_LogFields covers
// AC-OFFICE-ASSIGN-RATE-002.2: task id, assignee profile id, acting agent
// id, allowance and the error are present, but observed_count is
// deliberately absent — a failed count read never produced one.
func TestReportAssignmentRateLimitCountReadFailed_LogFields(t *testing.T) {
logs := withObservedWarnLogs(t)
runsservice.ReportAssignmentRateLimitCountReadFailed(
"task-log-2", "assignee-log-2", "actor-log-2", 5, errors.New("db unavailable"))
entries := logs.FilterMessage("assignment wake admitted (rate limit count read failed)").All()
if len(entries) != 1 {
t.Fatalf("warning count = %d, want 1", len(entries))
}
fields := entries[0].ContextMap()
if fields["task_id"] != "task-log-2" ||
fields["assignee_agent_profile_id"] != "assignee-log-2" ||
fields["acting_agent_id"] != "actor-log-2" ||
fields["allowance"] != int64(5) {
t.Fatalf("fields = %#v, want task_id/assignee_agent_profile_id/acting_agent_id/allowance all present", fields)
}
if errVal, ok := fields["error"]; !ok || errVal == "" {
t.Fatalf("fields = %#v, want a non-empty error field", fields)
}
if _, present := fields["observed_count"]; present {
t.Fatalf("fields = %#v, want observed_count absent — a failed count read never produced one", fields)
}
}
// TestReportAssignmentRateLimitUnattributed_LogFields covers
// AC-OFFICE-ASSIGN-RATE-002.3 with a non-empty acting agent id: reason,
// allowance and acting_agent_id are present, but task_id is deliberately
// absent — the wake's task could not be determined at all.
func TestReportAssignmentRateLimitUnattributed_LogFields(t *testing.T) {
logs := withObservedWarnLogs(t)
runsservice.ReportAssignmentRateLimitUnattributed("actor-log-3", 5)
entries := logs.FilterMessage("assignment wake admitted (task unattributed)").All()
if len(entries) != 1 {
t.Fatalf("warning count = %d, want 1", len(entries))
}
fields := entries[0].ContextMap()
if fields["reason"] != "task_unattributed" || fields["allowance"] != int64(5) || fields["acting_agent_id"] != "actor-log-3" {
t.Fatalf("fields = %#v, want reason/allowance/acting_agent_id all present", fields)
}
if _, present := fields["task_id"]; present {
t.Fatalf("fields = %#v, want task_id absent — the wake's task could not be determined", fields)
}
}
// TestReportAssignmentRateLimitUnattributed_LogFields_EmptyActorID covers
// the same function's other deliberate omission: an empty acting agent id
// is dropped from the fields entirely rather than logged as "".
func TestReportAssignmentRateLimitUnattributed_LogFields_EmptyActorID(t *testing.T) {
logs := withObservedWarnLogs(t)
runsservice.ReportAssignmentRateLimitUnattributed("", 5)
entries := logs.FilterMessage("assignment wake admitted (task unattributed)").All()
if len(entries) != 1 {
t.Fatalf("warning count = %d, want 1", len(entries))
}
fields := entries[0].ContextMap()
if fields["reason"] != "task_unattributed" || fields["allowance"] != int64(5) {
t.Fatalf("fields = %#v, want reason/allowance present", fields)
}
if _, present := fields["acting_agent_id"]; present {
t.Fatalf("fields = %#v, want acting_agent_id absent for an empty caller", fields)
}
if _, present := fields["task_id"]; present {
t.Fatalf("fields = %#v, want task_id absent — the wake's task could not be determined", fields)
}
}
@@ -0,0 +1,119 @@
package service_test
import (
"errors"
"expvar"
"strings"
"testing"
runsservice "github.com/kandev/kandev/internal/runs/service"
)
// counterValue returns the current value of one key in a *expvar.Map, or 0
// if the key has never been touched. Used instead of counterHasLabel's
// presence check because office_assignment_rate_limit_total's label set is
// the fixed three-value reason (no per-test-unique label to key off), so
// these tests read the delta across a call instead.
func counterValue(t *testing.T, mapName, key string) int64 {
t.Helper()
v := expvar.Get(mapName)
if v == nil {
t.Fatalf("expvar map %q not registered", mapName)
}
m, ok := v.(*expvar.Map)
if !ok {
t.Fatalf("expvar %q is not a *expvar.Map", mapName)
}
var got int64
m.Do(func(kv expvar.KeyValue) {
if kv.Key != key {
return
}
if iv, ok := kv.Value.(*expvar.Int); ok {
got = iv.Value()
}
})
return got
}
const assignmentRateLimitMap = "office_assignment_rate_limit_total"
// TestReportAssignmentRateLimitRefused covers AC-OFFICE-ASSIGN-RATE-003.1's
// allowance_exhausted counter and the new QueueOutcomeRateLimited value.
func TestReportAssignmentRateLimitRefused(t *testing.T) {
before := counterValue(t, assignmentRateLimitMap, "reason=allowance_exhausted")
outcome := runsservice.ReportAssignmentRateLimitRefused("task-1", "agent-assignee", "agent-actor", 5, 5)
if outcome != runsservice.QueueOutcomeRateLimited {
t.Fatalf("outcome = %q, want rate_limited", outcome)
}
after := counterValue(t, assignmentRateLimitMap, "reason=allowance_exhausted")
if after != before+1 {
t.Fatalf("counter = %d, want %d", after, before+1)
}
}
// TestReportAssignmentRateLimitCountReadFailed covers
// AC-OFFICE-ASSIGN-RATE-002.2's degraded-admission counter: the wake is
// still admitted, so this function has no return value to assert — only
// the counter move.
func TestReportAssignmentRateLimitCountReadFailed(t *testing.T) {
before := counterValue(t, assignmentRateLimitMap, "reason=count_read_failed")
runsservice.ReportAssignmentRateLimitCountReadFailed("task-1", "agent-assignee", "agent-actor", 5, errors.New("db unavailable"))
after := counterValue(t, assignmentRateLimitMap, "reason=count_read_failed")
if after != before+1 {
t.Fatalf("counter = %d, want %d", after, before+1)
}
}
// TestReportAssignmentRateLimitUnattributed covers
// AC-OFFICE-ASSIGN-RATE-002.3's degraded-admission counter for an
// in-scope wake whose task could not be determined.
func TestReportAssignmentRateLimitUnattributed(t *testing.T) {
before := counterValue(t, assignmentRateLimitMap, "reason=task_unattributed")
runsservice.ReportAssignmentRateLimitUnattributed("agent-actor", 5)
after := counterValue(t, assignmentRateLimitMap, "reason=task_unattributed")
if after != before+1 {
t.Fatalf("counter = %d, want %d", after, before+1)
}
}
// TestReportAssignmentRateLimitUnattributed_EmptyActorID covers the same
// path with no acting agent identifier, which the function must accept
// without panicking (the conditional log field append).
func TestReportAssignmentRateLimitUnattributed_EmptyActorID(t *testing.T) {
before := counterValue(t, assignmentRateLimitMap, "reason=task_unattributed")
runsservice.ReportAssignmentRateLimitUnattributed("", 5)
after := counterValue(t, assignmentRateLimitMap, "reason=task_unattributed")
if after != before+1 {
t.Fatalf("counter = %d, want %d", after, before+1)
}
}
// TestAssignmentRateLimitCounter_LabelHasOnlyReason covers
// AC-OFFICE-ASSIGN-RATE-003.2: the counter's key must carry only the
// closed reason label, never a task, agent, or run identifier.
func TestAssignmentRateLimitCounter_LabelHasOnlyReason(t *testing.T) {
runsservice.ReportAssignmentRateLimitRefused("task-should-not-appear", "assignee-should-not-appear", "actor-should-not-appear", 5, 5)
v := expvar.Get(assignmentRateLimitMap)
m, ok := v.(*expvar.Map)
if !ok {
t.Fatalf("expvar %q is not a *expvar.Map", assignmentRateLimitMap)
}
m.Do(func(kv expvar.KeyValue) {
if !strings.Contains(kv.Key, "allowance_exhausted") {
return
}
if kv.Key != "reason=allowance_exhausted" {
t.Fatalf("key = %q, want exactly \"reason=allowance_exhausted\" with no identifier labels", kv.Key)
}
})
}
@@ -114,3 +114,65 @@ func ReportKeylessEnqueue(reason string, cause KeylessCause, detail string) {
zap.String("cause", string(cause)),
zap.String("detail", detail))
}
// Reason labels for office_assignment_rate_limit_total — a closed
// three-value set.
const (
assignmentRateLimitReasonAllowanceExhausted = "allowance_exhausted"
assignmentRateLimitReasonCountReadFailed = "count_read_failed"
assignmentRateLimitReasonTaskUnattributed = "task_unattributed"
)
// ReportAssignmentRateLimitRefused records a wake refused because its
// task's REQ-OFFICE-ASSIGN-RATE-001 allowance was already exhausted:
// counts reason="allowance_exhausted", logs at Warn with the task,
// assignee, acting-agent, observed-count and allowance fields, and
// returns QueueOutcomeRateLimited. The caller must not insert a row for
// this wake.
func ReportAssignmentRateLimitRefused(
taskID, assigneeAgentProfileID, actingAgentID string, observedCount, allowance int,
) QueueOutcome {
incAssignmentRateLimit(assignmentRateLimitReasonAllowanceExhausted)
dedupLogger().Warn("assignment wake refused (rate limit)",
zap.String("task_id", taskID),
zap.String("assignee_agent_profile_id", assigneeAgentProfileID),
zap.String("acting_agent_id", actingAgentID),
zap.Int("observed_count", observedCount),
zap.Int("allowance", allowance))
return QueueOutcomeRateLimited
}
// ReportAssignmentRateLimitCountReadFailed records a degraded admission
// caused by a failed window-count read: counts reason="count_read_failed"
// and logs at Warn without an observed count, which the failed read did
// not produce. The wake is still admitted — this reports the
// degradation, it does not decide it.
func ReportAssignmentRateLimitCountReadFailed(
taskID, assigneeAgentProfileID, actingAgentID string, allowance int, err error,
) {
incAssignmentRateLimit(assignmentRateLimitReasonCountReadFailed)
dedupLogger().Warn("assignment wake admitted (rate limit count read failed)",
zap.String("task_id", taskID),
zap.String("assignee_agent_profile_id", assigneeAgentProfileID),
zap.String("acting_agent_id", actingAgentID),
zap.Int("allowance", allowance),
zap.Error(err))
}
// ReportAssignmentRateLimitUnattributed records a degraded admission
// caused by an otherwise in-scope wake whose task could not be
// determined: counts reason="task_unattributed" and logs at Warn
// without a task identifier, which is by definition the
// value that could not be determined. actingAgentID is logged only when
// non-empty. The wake is still admitted.
func ReportAssignmentRateLimitUnattributed(actingAgentID string, allowance int) {
incAssignmentRateLimit(assignmentRateLimitReasonTaskUnattributed)
fields := []zap.Field{
zap.String("reason", assignmentRateLimitReasonTaskUnattributed),
zap.Int("allowance", allowance),
}
if actingAgentID != "" {
fields = append(fields, zap.String("acting_agent_id", actingAgentID))
}
dedupLogger().Warn("assignment wake admitted (task unattributed)", fields...)
}
@@ -45,6 +45,11 @@ var metricReasons = map[string]struct{}{
var (
runDedupTotal = expvar.NewMap("office_run_dedup_total")
runDedupKeylessTotal = expvar.NewMap("office_run_dedup_keyless_total")
// assignmentRateLimitTotal counts REQ-OFFICE-ASSIGN-RATE-003 refusals
// and degraded admissions by a closed three-value "reason" label
// (allowance_exhausted, count_read_failed, task_unattributed) — never a
// task, agent, workspace, session or run identifier.
assignmentRateLimitTotal = expvar.NewMap("office_assignment_rate_limit_total")
)
// ParentWakeDedupedTotal counts a task_children_completed insert rejected by
@@ -82,6 +87,14 @@ func incRunDedupKeyless(reason string, cause KeylessCause) {
runDedupKeylessTotal.Add(metricLabel("reason", metricReason(reason), "cause", string(cause)), 1)
}
// incAssignmentRateLimit increments office_assignment_rate_limit_total for
// one of the three fixed reason values dedup.go's Report* helpers pass —
// never run through metricReason, since this label set is closed by this
// capability itself rather than driven by agent-supplied input.
func incAssignmentRateLimit(reason string) {
assignmentRateLimitTotal.Add(metricLabel("reason", reason), 1)
}
func metricReason(reason string) string {
if _, ok := metricReasons[reason]; ok {
return reason
@@ -20,3 +20,15 @@ func TestQueueOutcomeNone_MatchesEngineDeclaration(t *testing.T) {
runsservice.QueueOutcomeNone, engine.QueueOutcomeNone)
}
}
// TestQueueOutcomeRateLimited_MatchesEngineDeclaration is
// TestQueueOutcomeNone_MatchesEngineDeclaration's twin for the value added
// by REQ-OFFICE-ASSIGN-RATE-001: both packages declare QueueOutcome and
// its members independently, so the engine adapter cannot report a refused
// assignment wake unless the two stay in sync.
func TestQueueOutcomeRateLimited_MatchesEngineDeclaration(t *testing.T) {
if string(runsservice.QueueOutcomeRateLimited) != string(engine.QueueOutcomeRateLimited) {
t.Fatalf("runsservice.QueueOutcomeRateLimited = %q, engine.QueueOutcomeRateLimited = %q; both MUST match",
runsservice.QueueOutcomeRateLimited, engine.QueueOutcomeRateLimited)
}
}
@@ -50,6 +50,14 @@ const (
// queued row for the same agent, reason, and task bucket within the
// coalescing window, so nothing new was inserted.
QueueOutcomeCoalesced QueueOutcome = "coalesced"
// QueueOutcomeRateLimited means an agent-initiated assignment wake was
// refused because its task's REQ-OFFICE-ASSIGN-RATE-001 allowance was
// already exhausted: no row was inserted and none was merged. Distinct
// from QueueOutcomeNone, which means no enqueue was attempted or the
// attempt errored — a caller must not infer a rate-limit refusal from
// an absence. Declared identically in internal/workflow/engine/adapters.go
// — both declarations MUST match.
QueueOutcomeRateLimited QueueOutcome = "rate_limited"
)
// QueueRunRequest carries everything the queue needs to insert a row.
@@ -40,6 +40,11 @@ const (
// outcome. Declared identically in internal/runs/service — both
// declarations MUST match.
QueueOutcomeNone QueueOutcome = ""
// QueueOutcomeRateLimited means an agent-initiated assignment wake was
// refused because its task's REQ-OFFICE-ASSIGN-RATE-001 allowance was
// already exhausted: no row was inserted and none was merged. Declared
// identically in internal/runs/service — both declarations MUST match.
QueueOutcomeRateLimited QueueOutcome = "rate_limited"
)
// QueueRunRequest is the typed payload the engine hands to RunQueueAdapter.
@@ -0,0 +1,57 @@
---
created: 2026-09-17
status: done
requirements:
- REQ-OFFICE-ASSIGN-RATE-001
- REQ-OFFICE-ASSIGN-RATE-002
- REQ-OFFICE-ASSIGN-RATE-003
system_design:
- ../../specs/office/system-design/assignment-wake-rate-limit-01.md
---
# Implementation Plan: Office Assignment Wake Rate Limit
## Overview
Bound the number of agent-initiated `task_assigned` wakes one task can produce
in a rolling window. Removing the same-agent equality gate in
`ApplyTaskMutation` let an agent holding `can_assign_tasks` reassign a task to
itself repeatedly, spaced past the five-second coalescing window, minting an
unbounded stream of genuinely distinct wakes. Neither existing bound catches
this: the dedup key is generation-aware by design, and `task_assigned` is on the
attended-provenance allowlist, so the daily spend ceiling never applies.
## Scope
- One admission gate in `SchedulerService.queueRun`, after the recent-duplicate
lookup and coalescing, before `CreateRun`. It binds the task-mutation
producer, which is the only `task_assigned` producer routing through that
function and the only one carrying an actor.
- A task-scoped window count derived from existing `runs` rows; no new table,
column, migration, or background job.
- A new `QueueOutcome` value, declared in both places the enum lives.
- One `expvar` counter labelled by a closed three-value reason set, plus a Warn
log at the decision site.
## Out of scope
- Dedup key identity, the coalescing window, and the 24-hour lookback.
- The event-subscriber producer of assignment wakes; a per-task allowance
cannot bind it.
- The provenance classifier and the budget admission gates.
- Rate-limiting the assignment write itself; a refused wake still persists the
assignee and bumps the generation.
- Any frontend. No screen renders the allowance, the counters, or a refusal.
## Implementation Wave
- [ ] [task-01-assignment-wake-rate-limit](task-01-assignment-wake-rate-limit.md)
## Verification
```bash
cd apps/backend && go test ./internal/office/scheduler ./internal/office/service \
./internal/runs/service ./internal/workflow/engine
python3 scripts/lint-spec-files.py --all
python3 scripts/list-docs.py validate
```
@@ -0,0 +1,145 @@
---
id: "01-assignment-wake-rate-limit"
title: "Bound agent-initiated assignment wakes per task"
status: done
wave: 1
depends_on: []
plan: "plan.md"
requirements:
- REQ-OFFICE-ASSIGN-RATE-001
- REQ-OFFICE-ASSIGN-RATE-002
- REQ-OFFICE-ASSIGN-RATE-003
acceptance_criteria:
- AC-OFFICE-ASSIGN-RATE-001.1
- AC-OFFICE-ASSIGN-RATE-001.2
- AC-OFFICE-ASSIGN-RATE-001.3
- AC-OFFICE-ASSIGN-RATE-001.4
- AC-OFFICE-ASSIGN-RATE-001.5
- AC-OFFICE-ASSIGN-RATE-001.6
- AC-OFFICE-ASSIGN-RATE-001.7
- AC-OFFICE-ASSIGN-RATE-001.8
- AC-OFFICE-ASSIGN-RATE-001.9
- AC-OFFICE-ASSIGN-RATE-001.10
- AC-OFFICE-ASSIGN-RATE-001.11
- AC-OFFICE-ASSIGN-RATE-001.12
- AC-OFFICE-ASSIGN-RATE-001.13
- AC-OFFICE-ASSIGN-RATE-002.1
- AC-OFFICE-ASSIGN-RATE-002.2
- AC-OFFICE-ASSIGN-RATE-002.3
- AC-OFFICE-ASSIGN-RATE-002.4
- AC-OFFICE-ASSIGN-RATE-002.5
- AC-OFFICE-ASSIGN-RATE-002.6
- AC-OFFICE-ASSIGN-RATE-003.1
- AC-OFFICE-ASSIGN-RATE-003.2
- AC-OFFICE-ASSIGN-RATE-003.3
- AC-OFFICE-ASSIGN-RATE-003.4
- AC-OFFICE-ASSIGN-RATE-003.5
- AC-OFFICE-ASSIGN-RATE-003.6
system_design:
- ../../specs/office/system-design/assignment-wake-rate-limit-01.md
---
# Task 01: Bound agent-initiated assignment wakes per task
## Summary
Add one admission gate to the Office run queue that refuses an agent-initiated
`task_assigned` wake once a task has already produced `N` of them inside a
rolling window `W`. The gate is the last check before insertion, so "admitted"
means "a row was written" and the count needs no state of its own.
## Scope
- A window-count repository read over the **stored** `runs` rows, filtered by
`reason`, the payload's `task_id` and `actor_type`, and `requested_at`, using
`dialect.JSONExtract` and the string-type guard `CoalesceRun` already needs.
- The **incoming** wake's `actor_type` and `task_id` read in Go, by unmarshalling
the payload `queueRun` was handed, the way `taskIDFromPayload` does.
`dialect.JSONExtract` is SQL over a column and does not apply to a wake that is
not a row yet; the two reads are separate.
- The gate in `SchedulerService.queueRun`, positioned after `CheckIdempotencyKey`
and `CoalesceRun` and before `CreateRun`. That position binds the
task-mutation producer only (`AC-OFFICE-ASSIGN-RATE-001.12`); no other
`task_assigned` producer routes through this function.
- `RunContext.ActorType` reaching the payload for the mutation producer's
agent-actor wakes, pinned by a test
(`AC-OFFICE-ASSIGN-RATE-001.13`). A wake carrying no `actor_type` is out of
scope: admit, no allowance consumed, no counter moved.
- `N = 5` and `W = 10 * time.Minute` declared once beside
`CoalesceWindowSeconds` and `IdempotencyWindowHours`.
- A new `QueueOutcome` value added to **both** declarations
(`internal/runs/service` and `internal/workflow/engine/adapters.go`).
- A reporting helper beside `ReportWindowedDedup` / `ReportDurableDedup`, a new
`expvar.Map` beside `office_run_dedup_total`, and a Warn log carrying task id,
assignee profile id, acting agent id, observed count and allowance.
## Exclusions
- No change to dedup key construction, the coalescing predicate, the 24-hour
lookback, provenance classification, or budget admission.
- No change to `SetTaskAssigneeAsAgent`'s permission check or to the mutation
itself: a refused wake still persists the assignee, bumps
`assignment_generation`, publishes the task update, and interrupts a displaced
assignee's session.
- No bound on the other four `task_assigned` producers: the task-lifecycle event
subscriber (`service/event_subscribers.go`), onboarding, the recovery sweep,
and the orchestrator's workflow auto-start. None carries an actor, and a
per-task allowance cannot bind a producer that wakes a task just created,
onboarded or swept. Do not "fix" this by giving one of them an actor.
- No ordering or tiebreak logic. `AC-OFFICE-ASSIGN-RATE-001.10` makes the
decision order-independent; no `ORDER BY` belongs in the window count.
- No frontend, no migration, no new configuration surface.
## Implementation acceptance conditions
1. `N` agent-actor assignment wakes for one task admit and the next refuses,
writing no row and returning the new outcome; a `user`-actor wake and an
empty-`callerAgentID` call are never refused, and neither is a wake carrying
no `actor_type` at all.
2. A wake merged by coalescing or suppressed by the idempotency lookup neither
consumes the allowance nor can be refused by it; a refused wake does not
consume it either.
3. A failing count read and an unattributable task both **admit**, each
incrementing the counter under its own `reason` label, with no identifier
used as a label. Unattributable covers an absent, null or empty `task_id` and
a present-but-non-string one, identically; an unparseable payload is *not*
unattributable — it fails the actor step first and leaves scope with no
counter moved.
4. A test enumerates the `task_assigned` producers and asserts the actor type
each emits, pinning `AC-OFFICE-ASSIGN-RATE-001.12`'s invariant that no other
producer sets `actor_type` `agent`.
5. A mixed-actor coalesce is covered in both directions and asserts the count the
gate then observes, which is the accepted off-by-one in the requirement's
`## Accepted consequence` rather than an exact count.
## Verification
```bash
cd apps/backend && go test ./internal/office/scheduler ./internal/office/service \
./internal/runs/service ./internal/workflow/engine
cd apps/backend && make lint
```
## Likely files
- `apps/backend/internal/office/scheduler/run.go`
- `apps/backend/internal/office/scheduler/reactivity.go`
- `apps/backend/internal/runs/service/dedup.go`
- `apps/backend/internal/runs/service/metrics_vars.go`
- `apps/backend/internal/workflow/engine/adapters.go`
- `apps/backend/internal/runs/repository/**` (window-count query, SQLite and
PostgreSQL twins)
## Risks
- The `QueueOutcome` enum is declared twice and must stay in sync; adding the
value to one declaration only breaks the engine adapter's reporting.
- A hand-written `json_extract` diverges between SQLite and PostgreSQL. Use
`dialect.JSONExtract` and guard the JSON string type, or a numeric `task_id`
in an older payload silently joins another task's allowance.
- Tests must inject time (`synctest`), never `time.Sleep`.
- `CoalesceRun` overwrites the surviving row's payload and does not match on
`actor_type`, so a mixed-actor merge relabels the actor the count reads. This
is an accepted consequence, not a bug to fix here: do not add `actor_type` to
the coalescing predicate, and do not write a test that asserts an exact count
across such a merge.
@@ -0,0 +1,354 @@
---
status: draft
system: office
created: 2026-09-17
owners:
- kandev
---
# Office: Assignment Wake Rate Limit Requirements
## Overview
An Office agent holding `can_assign_tasks` — the default `worker` role
permission — can call the dashboard's assignee mutator repeatedly with itself as
the target. Each call commits a new `tasks.assignment_generation`, so each
produces a genuinely new wake occurrence with a distinct dedup key. Spaced past
the coalescing window, the agent mints an unbounded stream of never-suppressed
`task_assigned` runs against its own task.
Two mechanisms look like they should bound this and do not.
**Deduplication does not**, by design. `AC-OFFICE-RUN-DEDUP-001.3` *requires* a
second assignment to the same agent to insert a run: a key must suppress a
redelivery and must not suppress a repeat. Every wake in this stream is a
repeat.
**The spend ceiling does not**, and this is the load-bearing finding.
`ClassifyRunProvenance` (`AC-OFFICE-BUDGET-007.1`) is a total function of
`reason` alone, and `task_assigned` sits on its attended allowlist because a
*person* assigning a task is plausibly waiting on it. Attended runs are exempt
from operator policies and from the default ceiling, and a reason-only
classifier cannot tell an agent actor from a human one, so these runs never
reach it. The spend is unbounded, not loosely bounded.
This document owns the bound: a per-task allowance on agent-initiated
assignment wakes, a separate guard rather than a change to coalescing.
## Terminology
- **Agent-initiated assignment wake:** a `task_assigned` wake for a mutation
whose actor type is `agent`; defined normatively by
`AC-OFFICE-ASSIGN-RATE-001.1`.
- **Allowance:** the admitted agent-initiated assignment wakes permitted for one
task in one rolling window.
- **Admitted:** a wake that inserted a `runs` row. One suppressed by dedup or
merged by coalescing is neither admitted nor refused; an earlier gate decided
it.
- **Refused:** a wake this capability declined, inserting no row and merging into
none.
## Prior art
### Our own recorded reasoning (wiki)
**Searched:** QMD collection `wiki` (564 documents, re-indexed 2026-09-17),
semantic query with reranking off. A semantic result, not a degraded keyword
scan.
[[agent-budget-governance]] states the position this capability departs from: an
operator-initiated agent is bounded by the operator's attention, a cron- or
webhook-triggered one by nothing except a budget. Its companion
[[event-triggered-agent-activation]] frames the axis as *who starts it*.
Our case is a third kind neither page enumerates: **agent-initiated** activation
that the classifier deciding whether a budget applies cannot tell from
operator-initiated. It has an actor, so Kandev reads it as attended and bounded
by an attention nobody is paying. That is why the budget is not the bound here.
Two of its anti-patterns are adopted: an org cap without a per-principal cap
(Kandev's default ceiling is exactly an org cap, this adds a narrower per-task
one, and the missing aggregate is named under `## Out of scope`), and treating a
cap as cost control when it is also the blast radius of an agent's autonomy
(queue rows, scheduler churn and session launches are blast radius no spend
ceiling touches). We depart from its window shape, where the corpus converges on
`calendar_month_utc`, because a calendar window lets an attacker spend a full
allowance either side of the boundary.
### What other products shipped (saas-kb)
Recorded in the system design's `## Prior art considered`, beside the choices it
informs: a count cap rather than a cooldown, and why the near-universal
per-principal concurrency cap is not this axis.
### Prior art inside this repository
Three precedents shape the criteria below: Office automations'
`max_concurrent_runs` cap, the workspace pause gate's refused admission, and
`checkIdleSkip`'s open failure, which `AC-OFFICE-ASSIGN-RATE-002.2` follows and
the pause gate deliberately does not. Call sites and the contrast are in the
system design.
## Requirements
### REQ-OFFICE-ASSIGN-RATE-001: A per-task allowance bounds agent-initiated assignment wakes
**Intent:** A repeat assignment must still wake the agent, so the bound cannot
suppress the class. It bounds how often the class may fire, scoped so a human
operator is never throttled.
**User story:** As an Office operator, I want an agent that can assign tasks to
be unable to mint an endless stream of runs against a task, so that an
unattended workspace cannot spend without limit while nobody is watching.
#### Acceptance criteria
- **AC-OFFICE-ASSIGN-RATE-001.1:** An *agent-initiated assignment wake* is a
`task_assigned` wake produced for a task mutation whose actor type is exactly
`agent`. A wake carrying any other actor type — `user`, an unrecognised value,
or no actor type at all — is not one, and no criterion here applies to it: it
shall be admitted without consuming an allowance, never refused, and shall
move none of the counters in `REQ-OFFICE-ASSIGN-RATE-003`. The internal and administrative caller that
supplies an empty acting agent identifier is classified `user`, so it is
never limited. An absent actor type is a normal classification outcome, not a
degraded one: the out-of-scope producers `AC-OFFICE-ASSIGN-RATE-001.12` names
carry none.
- **AC-OFFICE-ASSIGN-RATE-001.2:** The system shall admit at most `N`
agent-initiated assignment wakes for one task within any rolling window of
duration `W` in serial evaluation. Concurrent evaluation may exceed `N`;
see AC-OFFICE-ASSIGN-RATE-002.1.
- **AC-OFFICE-ASSIGN-RATE-001.3:** The allowance shall be scoped to the task
alone. Wakes for one task count against one allowance whichever agent is being
assigned, so alternating assignment between two agents shall not yield two
allowances.
- **AC-OFFICE-ASSIGN-RATE-001.4:** Only an admitted wake shall consume the
allowance. A refused wake shall not consume it: each admitted wake frees its
share of the allowance one window after its own request instant, however many
refusals fell in between, subject to `## Accepted consequence`'s coalescing
relabel.
- **AC-OFFICE-ASSIGN-RATE-001.5:** The allowance shall be evaluated after the
recent-duplicate lookup and after the coalescing attempt, and before the run
row is inserted. A wake that an earlier gate already suppressed shall be
neither refused by this one nor counted against the allowance, subject to
`## Accepted consequence`'s coalescing relabel.
- **AC-OFFICE-ASSIGN-RATE-001.6:** When the allowance is exhausted, the system
shall refuse the wake: insert no run row, merge into no existing run, and
report neither a queued nor a coalesced nor a deduplicated outcome for it.
- **AC-OFFICE-ASSIGN-RATE-001.7:** A refusal shall not abort or roll back the
mutation. The assignee change stays persisted, the assignment generation stays
bumped, and the mutation's other effects — the previous assignee's session
interrupt, that assignee's office session row transition, and the task-updated
publication — shall still occur. This mutation produces no
comment wake and no mention wake, so none is required here and a regression
test for this criterion shall not assert one.
- **AC-OFFICE-ASSIGN-RATE-001.8:** Within the producer scope fixed by
`AC-OFFICE-ASSIGN-RATE-001.12`, the allowance shall apply whether or not the
wake carries a dedup key. A wake enqueued keyless because its generation could
not be resolved, per `REQ-OFFICE-RUN-DEDUP-003`, is still bounded, so the
keyless path is not an unbounded bypass of this allowance.
- **AC-OFFICE-ASSIGN-RATE-001.9:** An unassignment — a mutation whose new
assignee identifier is empty — produces no assignment wake, and shall
therefore be neither refused nor counted against the allowance.
- **AC-OFFICE-ASSIGN-RATE-001.10:** The admission decision shall be a count of
the wakes falling in the window and shall not depend on the order in which
they fall within it. No ordering rule and no tiebreak between two wakes
sharing one persisted request instant is part of this contract; window
membership is fixed by `AC-OFFICE-ASSIGN-RATE-001.11` and the limit by
`AC-OFFICE-ASSIGN-RATE-001.2`. A test that must identify *which* wake was the
`N`th shall arrange distinct persisted request instants rather than rely on a
tiebreak.
- **AC-OFFICE-ASSIGN-RATE-001.11:** The window shall be the half-open interval
from the evaluation instant less `W`, exclusive, to the evaluation instant,
inclusive. A wake whose request instant is exactly one window old falls
outside the window and shall not count.
- **AC-OFFICE-ASSIGN-RATE-001.12:** The boundary is payload shape, not a count
of producers: the gate binds exactly those `task_assigned` wakes that reach it
carrying actor type `agent`, and every criterion here is scoped to that shape.
One producer does so today, the task-mutation reactivity path, the only one
carrying the mutation's actor. No other producer of a
`task_assigned` wake shall set actor type `agent`; that invariant shall be
pinned by a test enumerating the producers and asserting the actor type each
emits. Producers carrying no actor are out of scope per `## Out of scope`.
- **AC-OFFICE-ASSIGN-RATE-001.13:** A wake produced by the in-scope producer for
a mutation whose actor is an agent shall carry the actor type `agent` on the
wake itself, so the gate classifies it without re-reading the mutation.
Losing that field would take every such wake out of scope under
`AC-OFFICE-ASSIGN-RATE-001.1` and silently stop the allowance binding.
### REQ-OFFICE-ASSIGN-RATE-002: The allowance degrades safely and predictably
**Intent:** A cost guard that turns a transient database error into dropped
legitimate wakes trades a bounded cost problem for an unbounded liveness one.
Failure direction, race behaviour and constants must be stated, not discovered.
**User story:** As a Kandev maintainer, I want the allowance's failure and race
behaviour fixed in the contract, so that an implementation cannot pick the
convenient answer.
#### Acceptance criteria
- **AC-OFFICE-ASSIGN-RATE-002.1:** When two agent-initiated assignment wakes
for one task are evaluated concurrently, each may observe an allowance with
room and both may be admitted. The system shall permit over-admission bounded
by the number of concurrent evaluations, and shall never refuse a wake before
`N` have been admitted in the window.
- **AC-OFFICE-ASSIGN-RATE-002.2:** When the read deriving the current count
fails, the system shall admit the wake and shall count and log that degraded
admission under its own distinct reason. It shall not refuse the wake. The
degraded admission shall be logged at warning level carrying the task
identifier, the assignee agent profile identifier, the acting agent
identifier, the allowance, and the underlying read failure; it shall not carry
an observed count, which the failed read did not produce. This is
deliberately the opposite of the workspace pause gate, whose closed failure is
correct because a kill switch failing open is not a kill switch. The precedent
is `checkIdleSkip`.
- **AC-OFFICE-ASSIGN-RATE-002.3:** When a wake's task cannot be determined, the
system shall admit it and count it under a distinct unattributed reason, so a
payload-shape change cannot silently disable the gate by making every wake
unattributable. This case arises only for a wake already in scope
under `AC-OFFICE-ASSIGN-RATE-001.1`: a wake out of scope by actor type is
admitted there and never reaches this criterion, so a wake carrying neither an
actor type nor a task is not counted here. A task cannot be determined when
the identifier is absent, null, empty, or present but not a string; those
shapes are treated identically.
That admission shall be logged at warning level carrying the reason, the
allowance, and the acting agent identifier when the wake carries one; it shall
not carry a task identifier, which is by definition the value that could not
be determined.
- **AC-OFFICE-ASSIGN-RATE-002.4:** The gate shall hold no state of its own: the
count shall be derived from already-persisted run rows. A restart, a crash
between evaluation and insert, or a deleted run row therefore changes the
observed count without leaving the allowance inconsistent; no reconciliation
step is required.
- **AC-OFFICE-ASSIGN-RATE-002.5:** Re-delivering one assignment occurrence that
carries a dedup key shall not consume the allowance twice: redelivery is
decided by the dedup identity, which `AC-OFFICE-ASSIGN-RATE-001.5` places
before this gate. A keyless wake has no such identity, so re-delivering one
does consume the allowance again — intended, because
`AC-OFFICE-ASSIGN-RATE-001.8` bounds that path precisely where nothing
upstream suppresses it. A test for this criterion shall use a keyed wake.
- **AC-OFFICE-ASSIGN-RATE-002.6:** `N` shall be 5 and `W` shall be 10 minutes,
fixed values of this capability rather than operator-configurable, stated in
one place the implementation and its tests share. `N` shall be at least 1; 0,
which would refuse every agent-initiated assignment wake, is not permitted.
### REQ-OFFICE-ASSIGN-RATE-003: A refusal is never silent
**Intent:** A wake that vanishes without trace is indistinguishable from one
never produced, and an operator debugging a stalled task cannot tell. The
sibling deduplication capability set the floor at operator telemetry.
**User story:** As an Office operator, I want a refused assignment wake to be
visible in telemetry and logs, so that an agent not waking up is diagnosable
instead of mysterious.
#### Acceptance criteria
- **AC-OFFICE-ASSIGN-RATE-003.1:** Every refusal and every degraded admission
shall increment a counter labelled by reason, drawn from a closed set of
three values: allowance exhausted, count read failed, and task
unattributed.
- **AC-OFFICE-ASSIGN-RATE-003.2:** No task, agent, workspace, session or run
identifier shall ever be a counter label, so cardinality cannot grow with
traffic.
- **AC-OFFICE-ASSIGN-RATE-003.3:** A refusal shall be logged at warning level
carrying the task identifier, the assignee agent profile identifier, the
acting agent identifier, the observed count and the allowance.
- **AC-OFFICE-ASSIGN-RATE-003.4:** An admitted wake this capability did not
alter shall move none of these counters; ordinary admitted volume is already
visible on the runs queue.
- **AC-OFFICE-ASSIGN-RATE-003.5:** The outcome reported to the producer for a
refused wake shall be distinguishable from queued, coalesced and deduplicated
outcomes, and shall not be the outcome meaning no enqueue was attempted. A
caller shall not infer a rate-limit refusal from an absence.
- **AC-OFFICE-ASSIGN-RATE-003.6:** The counter shall be published on the same
operator telemetry surface as the run-deduplication counters.
## Out of scope
- **The five-second coalescing window, and which reasons coalesce.** Held
unchanged by repeated human decision on the run-dedup-generation card. This
capability is a separate guard placed after it; a coalesced wake neither
consumes the allowance nor can be refused by it.
- **The 24-hour recent-duplicate lookback and the durable unique index.**
Unchanged. Both decide *identity*, which this capability never touches.
- **What makes a wake occurrence distinct.** `REQ-OFFICE-RUN-DEDUP-001` through
`-004` own dedup key identity. This capability refuses wakes whose identity is
genuinely new; it never redefines identity.
- **A per-acting-agent allowance.** [[agent-budget-governance]] and the vendor
corpus both argue for a per-principal cap. Not adopted: it would throttle a
coordinator legitimately assigning many *different* tasks in a burst, the
behaviour Office's coordinator role exists to perform. Sizing one needs a
legitimate-coordinator rate measured from real workspace data, which does not
exist yet.
- **A workspace-aggregate launch budget, and a causation depth cap.** Card
`7dbc0b94` owns work-in-progress limits, depth caps and self-trigger
suppression. This is the per-task half only: across `T` tasks an agent retains
`5T` admitted wakes per window, and only an aggregate budget bounds that. The
two compose; neither substitutes for the other.
- **Reclassifying an agent-initiated `task_assigned` run as unattended.** The
adjacent fix, deliberately not taken here: `AC-OFFICE-BUDGET-007.1` makes
provenance a total function of `reason` and `-007.2` pins the attended
allowlist to a literal set including `task_assigned`, so an actor-aware
classifier changes `budget-run-provenance.md`'s contract and belongs to its
owner. It is incomplete alone in any case — a daily ceiling permits a full
day's allowance to burn in minutes and bounds no queue rows, session launches
or scheduler churn.
- **Rate-limiting the assignment mutation itself.** The assignee mutator keeps
accepting every call its permission check allows and the assignment generation
keeps bumping; only the wake is refused. Refusing the write would make the
board un-editable by agents.
- **A persisted record of a refused wake.** Office automations record a
`skipped` row, the better surface in the abstract. Not adopted: `runs` has no
such status and adding one to queue infrastructure shared by every workflow
style is larger than this guard warrants. `REQ-OFFICE-RUN-DEDUP-004` set the
floor for a high-frequency suppression outcome at operator telemetry and logs,
which `REQ-OFFICE-ASSIGN-RATE-003` meets.
- **Notifying anybody.** No inbox item, approval, activity entry, task-timeline
row or escalation. The counter added here is the evidence needed to size such
a surface later.
- **Operator tunability of `N` and `W`.** They ship as fixed values, like the
coalescing window. A settings surface, its API and its frontend are separate
work.
- **A dedicated runtime feature toggle.** Office is already off in the prod
profile, so this ships behind the flag gating the surface it guards. A second
toggle inside a disabled feature buys no rollout safety.
- **Any frontend.** No screen renders the allowance, the counters, or a
refusal.
- **Every other producer of a `task_assigned` wake.** Besides the in-scope
mutation path the repository has four: the task-lifecycle event subscriber,
onboarding, the unstarted-task recovery sweep, and the orchestrator's workflow
auto-start action. Two reasons, not convenience. None carries
an actor, so none is agent-initiated under `AC-OFFICE-ASSIGN-RATE-001.1`,
which `AC-OFFICE-ASSIGN-RATE-001.12` makes a standing invariant rather than an
observation. And a *per-task* allowance structurally cannot bind them: each
wakes a task just created, onboarded or swept, where a fresh or long-idle task
brings a fresh allowance, and the subscriber's update path is a redelivery
already owned by the dedup identity. Bounding them needs the aggregate scope
named above.
- **Existing run rows.** No migration, backfill or reinterpretation. Already
persisted rows count toward a window exactly as new ones do.
## Accepted consequence
When the allowance is exhausted and the refused wake targets a *different* agent
from the current assignee, that agent is left assigned and not woken until a
later occurrence or a user-initiated assignment wakes it. Accepted, not
mitigated: it needs five agent-driven reassignments of one task inside ten
minutes; user-initiated assignment is never limited, so an operator can recover
the task by hand; and the refusal is counted and logged under
`REQ-OFFICE-ASSIGN-RATE-003`. Exempting cross-agent
reassignment would reopen the exploit: alternating assignment between two agents
was already unbounded.
A coalescing merge can also relabel the actor type the count reads, because
`CoalesceRun` does not match on actor type and overwrites the surviving row's
payload. Per mixed-actor merge the observed count may run one high or one low.
Accepted: both available fixes are excluded, actor-aware coalescing by
`## Out of scope` and state of this gate's own by
`AC-OFFICE-ASSIGN-RATE-002.4`. The mechanism is in the design's
`## Counting without new state`.
## System design
[Assignment wake rate limit](../system-design/assignment-wake-rate-limit-01.md).
@@ -0,0 +1,472 @@
---
status: draft
system: office
requirements:
- REQ-OFFICE-ASSIGN-RATE-001
- REQ-OFFICE-ASSIGN-RATE-002
- REQ-OFFICE-ASSIGN-RATE-003
---
# Office: Assignment Wake Rate Limit System Design
## Purpose and boundaries
This design places one additional admission gate in the Office run queue: an
allowance on agent-initiated `task_assigned` wakes, scoped per task and to one
producer — the task-mutation reactivity path, per
`AC-OFFICE-ASSIGN-RATE-001.12`. It adds no table, no column, no migration, and
no background job. The count is derived from rows the queue already writes.
It does not touch dedup key identity
(`docs/specs/office/requirements/run-dedup-generation.md`), the five-second
coalescing predicate, the 24-hour recent-duplicate lookback, the attended and
unattended provenance classifier, or the pre-launch budget gates.
## Requirement mapping
| Requirement | Mechanism |
|---|---|
| `REQ-OFFICE-ASSIGN-RATE-001` | A gate in `SchedulerService.queueRun`, evaluated after the idempotency and coalescing checks and before `CreateRun`, backed by a task-scoped window count over `runs`. Its position fixes the producer scope: see below. |
| `REQ-OFFICE-ASSIGN-RATE-002` | The gate is a pure read plus the insert it guards; it fails open, admits under a race, and derives its count from persisted rows only. |
| `REQ-OFFICE-ASSIGN-RATE-003` | A new `expvar.Map` beside the run-dedup counters, plus a Warn log at the decision site, plus a new `QueueOutcome` value. |
## Where the gate sits
`apps/backend/internal/office/scheduler/run.go`'s `queueRun` is where the
mutation producer's wakes converge — `reactToAssigneeChange` reaches it through
`QueueRunCtx` — and it already owns the other two admission decisions.
It is **not** a choke point every Office producer reaches, and nothing here may
assume it is. The repository has five `task_assigned` producers, listed by
inventory rather than inferred from the two this capability happens to discuss:
| Producer | Route to an insert | Actor on its payload |
|---|---|---|
| `scheduler/reactivity.go`'s `reactToAssigneeChange` — **in scope** | `QueueRunCtx` → `SchedulerService.queueRun` | `agent` or `user`, from the mutation |
| `service/event_subscribers.go`'s `queueTaskAssignedRun` | `Service.QueueRun` | none |
| `office/onboarding/service.go` (its own comment calls it "a third `task_assigned` producer") | `Service.QueueRun` | none |
| `service/scheduler_recovery.go`'s `recoverUnstartedTasks` sweep | `Service.QueueRun` | none |
| `orchestrator/event_handlers_workflow.go`'s `queueOfficeAutoStartRun`, the `auto_start_agent` step action (`officeAutoStartRunReason`) | `engineRunQueue.QueueRun` → `runsservice.QueueRun` | none |
`Service.QueueRun` (`service/run.go`) delegates to `runsservice.QueueRun` when a
runs service is wired and otherwise re-implements the idempotency, coalescing
and insert steps inline in `queueRunInline`; neither route, and neither does the
orchestrator's, passes through `SchedulerService.queueRun`. So a gate placed
here binds the mutation producer and only the mutation producer.
Two independent facts agree today — one producer reaches this function, and one
producer puts an actor on the payload — and `AC-OFFICE-ASSIGN-RATE-001.12`
deliberately makes the **second** the contract. Call position is an
implementation fact a refactor can move without anyone noticing; payload shape
is checkable from the payload alone, and that criterion's enumeration test is
what keeps the two from drifting apart silently. A future producer that reaches
`queueRun` without an actor is out of scope by the same rule that covers the
four below, and one that sets `actor_type: agent` violates a written invariant
instead of quietly joining the gate.
Excluding the other four is a scope decision, not an accepted hole: a *per-task*
allowance could not bind any of them. The subscriber's live path is
`handleTaskCreated`, which fires once per task creation, where a fresh task
identifier always brings a fresh allowance; its `handleTaskUpdated` is
documented in its own doc comment as a redelivery and defensive path, and today
reaches no wake at all, because neither `task.updated` publisher puts
`assignee_agent_profile_id` in the payload and `queueTaskAssignedRun` returns
early without one. Onboarding wakes a task it has just created. The recovery
sweep wakes a task that was never started, and is capped at `maxRecoveryPerTick`
rows per tick anyway. The orchestrator's is workflow-driven: it fires on a step
transition running `auto_start_agent`, and its one reference to the participant
slate is how it *resolves* which agent to wake, not a second producer. None of
the four is an agent minting fresh occurrences against one task in a loop, which
is the shape this capability bounds.
The gate belongs in `queueRun` rather than in `reactToAssigneeChange` or in
`DashboardService.SetTaskAssigneeAsAgent`:
- In `reactToAssigneeChange` it would bind the same single producer, but
*upstream* of the recent-duplicate lookup and the coalescing attempt. A wake
later suppressed or merged would already have consumed the allowance,
contradicting `AC-OFFICE-ASSIGN-RATE-001.5` and `.4`. Position, not producer
coverage, is what rules it out.
- In `SetTaskAssigneeAsAgent` it would have to refuse the *mutation*, which
`REQ-OFFICE-ASSIGN-RATE-001.7` and the `## Out of scope` entry on
rate-limiting the write both forbid.
The resulting order inside `queueRun`, satisfying
`AC-OFFICE-ASSIGN-RATE-001.5`:
1. `guardAgentStatus`
2. `checkPauseGate`
3. `CheckIdempotencyKey` — windowed dedup (unchanged)
4. `CoalesceRun` — five-second merge (unchanged)
5. **the assignment allowance** (new)
6. `CreateRun`
Placing it at 5 is what makes "admitted" mean exactly "a row was inserted".
A wake suppressed at 3 or merged at 4 never reaches the gate, so it neither
consumes the allowance nor can be refused by it — `AC-OFFICE-ASSIGN-RATE-001.5`
and the coalescing exclusion in `## Out of scope` are the same fact.
## The predicate
The gate applies only when all three hold, per
`AC-OFFICE-ASSIGN-RATE-001.1`, `.2` and `.9`, and in this order:
1. `reason == RunReasonTaskAssigned`
2. the wake's `actor_type` is `agent`
3. the wake names a task
The order is contract, not style. A wake failing 2 is out of scope and admitted
silently; a wake that passes 2 but fails 3 is the counted unattributed case of
`AC-OFFICE-ASSIGN-RATE-002.3`. Evaluating 3 first would count a wake that
carries neither field against the unattributed reason, which is the one the
event-subscriber producer's payload would land on if it ever reached here.
`actor_type` and `task_id` are already carried on `RunContext` and already
serialised into the payload (`run.go`'s `RunContext.ActorType`, `.TaskID`).
`queueRun` receives the encoded payload rather than the struct, so it reads both
back out of it **in Go**, with one `json.Unmarshal` of the incoming payload —
exactly what `CoalesceRun` does through `taskIDFromPayload` before it builds any
SQL.
The two reads are not interchangeable and the build must not collapse them.
`dialect.JSONExtract(driver, col, path)` returns a SQL fragment over a *column*,
so it applies to the **stored** rows the window count scans
(`## Counting without new state`) and cannot apply to the **incoming** wake,
which is not a row yet. Incoming: Go. Stored: SQL.
Parsing the incoming payload has three failure shapes, which `CoalesceRun`
deliberately keeps apart. The evaluation order above decides where each one
lands, and the mapping is deliberately not symmetric:
| Incoming payload | Where it lands |
|---|---|
| `task_id` a non-empty JSON string | step 3 passes, the gate applies |
| `task_id` absent, JSON null, or `""` | step 3 fails: `AC-OFFICE-ASSIGN-RATE-002.3`, admit and count unattributed |
| `task_id` present but not a string | step 3 fails: `AC-OFFICE-ASSIGN-RATE-002.3`, admit and count unattributed |
| the payload does not parse at all | no readable `actor_type` either, so it fails **step 2** first: out of scope under `AC-OFFICE-ASSIGN-RATE-001.1`, admitted, no counter moved |
The last row is the one to be careful about, and it is part of why the order is
contract. A payload this gate cannot parse is not an unattributable *in-scope*
wake, it is a wake never shown to be in scope at all, and
`AC-OFFICE-ASSIGN-RATE-002.3` says in terms that a wake out of scope by actor
type never reaches it. Counting it as unattributed would count wakes never shown
to be agent-initiated, which `AC-OFFICE-ASSIGN-RATE-001.1` exempts by
definition.
The two task-shaped failures do share one bucket, unlike `CoalesceRun`, which
must hold `invalidTaskID` apart from taskless because it *merges* rows and a
wrong merge overwrites another launch's payload. This gate only counts and both
shapes admit, so splitting them would buy nothing and would add a counter value
`AC-OFFICE-ASSIGN-RATE-003.1`'s closed set does not have.
Step 2 needs no equivalent table: an `actor_type` that is absent, not a string,
or any string other than `agent` simply is not `agent`, which
`AC-OFFICE-ASSIGN-RATE-001.1` already covers as "any other actor type ... or no
actor type at all" — out of scope, admitted, no counter moved.
`actor_type` can be *absent* rather than merely different: the out-of-scope
event-subscriber producer builds its payload as `{"task_id": …}` and carries no
actor at all. Absent therefore has to classify the same way `"user"` does — out
of scope, admitted, no allowance consumed, no counter moved, per
`AC-OFFICE-ASSIGN-RATE-001.1` — and not as a degraded case with its own reason
value. A counter that increments on every Office task creation is not a
regression signal, it is background noise that hides one. What guards against
the in-scope producer actually losing the field is
`AC-OFFICE-ASSIGN-RATE-001.13`'s payload contract, pinned by a test, rather than
a runtime counter that could not tell that loss apart from ordinary traffic.
Because `actor_type` is `"user"` whenever the acting agent identifier is empty
(`runReactivityForAssigneeChange`'s `userSentinel` default), the
internal and administrative caller that passes `callerAgentID == ""` classifies
as `user` and is never limited. `AC-OFFICE-ASSIGN-RATE-001.1` states that as a
contract rather than leaving it as a consequence of the sentinel's value.
## Counting without new state
`AC-OFFICE-ASSIGN-RATE-002.4` requires the gate to hold no state of its own.
One repository read satisfies it:
```text
SELECT COUNT(*) FROM runs
WHERE reason = 'task_assigned'
AND <json_extract payload task_id> = :taskID
AND <json_extract payload actor_type> = 'agent'
AND requested_at > :windowStart
AND requested_at <= :evaluationInstant
```
`windowStart` is `evaluationInstant - W`. The lower comparison is strictly
greater-than and the upper comparison is inclusive, which is
`AC-OFFICE-ASSIGN-RATE-001.11`'s half-open interval. `requested_at` is written
by `queueRun` itself on every row, so it is the persisted request instant the
window is measured against.
No `ORDER BY` appears here or anywhere else in this capability:
`AC-OFFICE-ASSIGN-RATE-001.10` makes the decision explicitly order-independent,
which is what a count needs and all it needs. That is not a simplification of a
richer rule — the richer rule was not implementable. The assignment generation
is not orderable from a run row: `runs` has no `assignment_generation` column,
`RunContext` carries no such field, and the generation survives only inside the
string `dedupkeys.AssignmentKey` formats, which a keyless wake does not have at
all. Adding a payload field to carry it would *not* break coalescing —
`CoalesceRun` matches on `agent_profile_id`, `reason`, status, the window, the
comment-key exclusion and `task_id` alone, then overwrites the row's payload
rather than comparing payloads for equality — but the surviving coalesced row
would then carry the *last* merged wake's generation instead of the first, which
makes a payload-borne generation an unreliable ordering key in precisely the
case an ordering would be asked for.
Two portability notes carried over from `CoalesceRun`, which solved the same
problem in the same table:
- Use `dialect.JSONExtract` / `dialect.JSONTypeIsString`, never a literal
`json_extract`. The runs repository supports SQLite and PostgreSQL, and
`runs_queue_postgres_test.go` exists precisely because a hand-written
`json_extract` diverges between them.
- Guard the string type. `CoalesceRun` learned that `{"task_id":42}` textually
matches an incoming `{"task_id":"42"}`; the same guard applies here or a
numeric `task_id` in an old payload silently joins another task's allowance.
Deriving the count from `runs` has a deliberate consequence worth stating: the
history-retention capability deleting old run rows lowers an observed count.
That cannot raise the allowance above `N` inside a live window, because
retention operates on rows far older than ten minutes.
### The actor label on a stored row is not stable
The count reads `actor_type` off the **stored** payload, and `CoalesceRun`
rewrites exactly that. It matches on `agent_profile_id`, `reason`, status, the
window, the comment-key exclusion and `task_id` — never on `actor_type` — and
then writes the incoming wake's payload over the surviving row while leaving
that row's `requested_at` untouched. A queued row therefore carries the **last**
merged wake's actor and the **first** wake's request instant.
Reachable, not theoretical. `runReactivityForAssigneeChange` stamps `agent` or
`user` off `callerAgentID`, and one endpoint — `dashboard/handler.go`'s assignee
route — produces both, so two wakes for one (agent, task) inside five seconds
can carry different actors and merge. Per mixed-actor merge the window count
runs one high (an agent wake relabels a queued user row, so a row
`AC-OFFICE-ASSIGN-RATE-001.1` exempts starts counting) or one low (a user wake
relabels a queued agent row, returning a slot).
Neither direction is an agent-only capability: the loosening one needs a human to
assign the same task within five seconds of the agent's own wake, and the
tightening one is conservative and bounded at one row per merge.
Accepted rather than fixed, and recorded in the requirement's
`## Accepted consequence` so the build meets it as a stated consequence rather
than as a surprise. Both available fixes are closed: matching `actor_type` inside
`CoalesceRun` is the coalescing change the requirement's `## Out of scope` holds
unchanged by repeated human decision, and a column or cache of this gate's own is
what `AC-OFFICE-ASSIGN-RATE-002.4` forbids. What the build owes here is a test,
not a mechanism — `## Test plan` carries the mixed-actor coalesce in both
directions, so the behaviour is pinned at its measured value instead of assumed
away.
## Constants
`N = 5` and `W = 10 * time.Minute`, declared once beside
`CoalesceWindowSeconds` and `IdempotencyWindowHours` in
`scheduler/run.go`, and referenced by the tests rather than restated —
`AC-OFFICE-ASSIGN-RATE-002.6`. The duplicate `RunReason*` constant blocks in
`office/scheduler` and `office/service` are a known wart that
`run-dedup-generation.md` already excluded from its scope; do not add a second
copy of these two.
## Failure direction, and why it differs from its neighbour
Three gates in this one function now fail three different ways, so the choice
has to be explicit rather than inferred:
| Gate | On its own read failing | Why |
|---|---|---|
| `checkPauseGate` | **closed** (`ErrPauseGateUnavailable`) | a kill switch that fails open is not a kill switch |
| `CheckIdempotencyKey` | **error** returned to the caller | a duplicate admitted twice is a correctness defect |
| assignment allowance | **open**, counted and logged | a closed failure drops a legitimate wake, trading a bounded cost problem for an unbounded liveness one |
`AC-OFFICE-ASSIGN-RATE-002.2` fixes the third. The in-repo precedent is
`checkIdleSkip` (`service/scheduler_integration.go`), which fails open on a
database error and is documented as doing so in
`apps/backend/internal/office/AGENTS.md`. The degraded admission is counted
under its own reason so that a gate silently admitting everything because its
read is broken is visible, rather than indistinguishable from a workspace that
simply never exhausts its allowance.
`AC-OFFICE-ASSIGN-RATE-002.3`'s unattributed case is the same reasoning applied
to a payload-shape change: if `task_id` stops resolving, every wake becomes
unattributable and the gate stops binding. Counting that case separately turns
a silent regression into a visible one.
## Concurrency
The count read and the subsequent `CreateRun` are not in one transaction, and
`AC-OFFICE-ASSIGN-RATE-002.1` deliberately does not require them to be. Two
concurrent agent-initiated assignment wakes for one task can both read `N-1`
and both insert, yielding `N+1` admitted in the window.
That is accepted rather than fixed. Making it exact would need either a
serialising lock on the task or a conditional insert, both of which put a
contended write on the hot path of a queue shared by every workflow style, to
buy exactness in a bound whose purpose is to turn "unbounded" into "about five
per ten minutes". The asymmetry in `AC-OFFICE-ASSIGN-RATE-002.1` is the part
that matters and is testable: over-admission is bounded by the number of
concurrent evaluations, and under-admission — refusing before `N` are admitted
— is never permitted.
## Observability
A new `expvar.Map`, published beside `office_run_dedup_total` and
`office_run_dedup_keyless_total` in `internal/runs/service/metrics_vars.go`,
counting refusals and degraded admissions by a closed three-value `reason`
label: allowance exhausted, count read failed, task unattributed. Labels follow
the existing `metricLabel("reason", …)` `k=v;k=v` convention.
`AC-OFFICE-ASSIGN-RATE-003.2`'s no-identifier rule is what keeps the map
bounded, matching the reasoning behind `metricReasons`' `custom` fallback: a
process-global `expvar.Map` must not be able to grow once per distinct input.
`AC-OFFICE-ASSIGN-RATE-003.4` keeps admitted volume off these counters
entirely, so the map's absolute values read as "how often did the guard act",
not as traffic.
The Warn log's fields are fixed by the criteria rather than left to the
implementation: `AC-OFFICE-ASSIGN-RATE-003.3` for a refusal,
`AC-OFFICE-ASSIGN-RATE-002.2` for a failed count read (no observed count — the
read did not produce one), and `AC-OFFICE-ASSIGN-RATE-002.3` for an
unattributable task (no task identifier, for the same reason). The
no-identifier rule above governs counter *labels* only; these are log fields.
The reporting helper belongs beside `ReportWindowedDedup` and
`ReportDurableDedup` in `internal/runs/service/dedup.go` and follows their
shape: increment, log, return an outcome.
## The outcome value
`AC-OFFICE-ASSIGN-RATE-003.5` needs an outcome distinct from
`QueueOutcomeQueued`, `QueueOutcomeCoalesced`, `QueueOutcomeDeduped` and
`QueueOutcomeNone`. Reusing `QueueOutcomeNone` would conflate a deliberate
refusal with "no enqueue was attempted, or the attempt errored", which is what
its own doc comment says it means.
So a new value — and one trap: `QueueOutcome` and `QueueOutcomeNone` are
declared **twice**, in `internal/runs/service` and in
`internal/workflow/engine/adapters.go`, with a comment on each stating the two
declarations must match. A new member has to be added to both or the engine
adapter cannot report it.
Callers treat the new value the way they treat `QueueOutcomeCoalesced`: not an
error, and not a queued run. `ApplyTaskMutation`'s `queue` closure already
appends to `res.Runs` only on `QueueOutcomeQueued`, so a refused wake correctly
does not appear in `ApplyTaskMutationResult.Runs` with no change to that
branch.
## What the mutation still does
`AC-OFFICE-ASSIGN-RATE-001.7` is mostly a statement that nothing needs to
change: the gate lives downstream of the mutation, and
`runReactivityForAssigneeChange` already ignores a non-queued outcome. So a
refusal leaves intact the `UpdateTaskAssignee` write and its
`assignment_generation` bump, the `publishTaskUpdated` broadcast, the
`res.InterruptSessionID` hard-cancel of a displaced assignee's session, and the
prev-assignee office session row transition. The criterion exists so a future
refactor that moves the gate upstream has to break a written contract rather
than a coincidence.
It does **not** include a comment or mention wake. `ApplyTaskMutation` calls
`reactToComment` only when the mutation carries a comment, and the assignee
mutator builds its `TaskReactivityChange` without one, so no such wake exists on
this path to preserve. `AC-OFFICE-ASSIGN-RATE-001.7` says so explicitly, because
a regression test that asserted one would fail against correct code.
## Prior art considered
**Searched:** `search_saas_docs` (`category: "ai_sdlc"`) on runaway-loop and
self-trigger rate limiting, then the LangChain deepagents and Cloudflare Agents
SDK documents. Each item is a vendor claim, not evidence the approach works.
- **Cloudflare Agents SDK** passes a `source` to `onStateChanged` and documents
the best practice as *"Avoid infinite loops: be careful not to trigger state
updates in response to your own updates"*. Same insight: gate on *who caused
it*. **We cannot adopt its conclusion**, which is to suppress self-caused
reactions outright — `AC-OFFICE-RUN-DEDUP-001.3` requires a same-agent repeat
assignment to wake the agent. We keep the provenance distinction and bound the
rate instead of suppressing the class.
- **LangChain deepagents** classifies *"Excessive calls (runaway loops)"* as
system-handled, with the remedy *"cap model and tool calls per run"* — a
**count cap** over a scope, not a minimum interval. We follow that shape,
because a cooldown long enough to matter also breaks a legitimate
assign/hand-back/reassign burst.
- **Concurrency caps are near-universal** and per-principal (Multica, Coder,
Kiro). All bound *in-flight* work; none bounds the *arrival rate* of
self-caused triggers, the axis here.
**In-repo precedents.** The requirement's `## Prior art` names three; their call
sites live here. `internal/automation` caps a firing with `MaxConcurrentRuns` and
records a `RunStatusSkipped` run rather than retrying — the better surface in the
abstract, excluded here only because `runs` has no such status. The workspace
pause gate (`checkPauseGateForAgent`) is the established refused-admission
pattern for `runs`: typed error, no row written. `checkIdleSkip`
(`service/scheduler_integration.go`) fails open on a database error, the
precedent `AC-OFFICE-ASSIGN-RATE-002.2` follows and the deliberate opposite of
the pause gate's choice; the table in `## Failure direction, and why it differs
from its neighbour` is that contrast.
## Test plan
Backend, table-driven, in `apps/backend/internal/office/scheduler`:
- `N` agent-actor assignment wakes for one task admit; the `N+1`th refuses,
inserts no row, and returns the new outcome (`001.2`, `001.6`).
- A `user`-actor wake admits after the allowance is exhausted, and an
empty-`callerAgentID` call classifies as `user` (`001.1`).
- A wake carrying *no* `actor_type` admits after the allowance is exhausted,
consumes no allowance, and moves no counter (`001.1`).
- Alternating assignment between two agents shares one allowance (`001.3`).
- A refused wake does not consume the allowance: after `N` admits and several
refusals, exactly `N` wakes are admitted in the next window (`001.4`).
- A wake inside the coalescing window merges and does not consume the
allowance; a wake suppressed by the idempotency lookup likewise (`001.5`).
- A keyless wake from the in-scope producer is still bounded (`001.8`).
- The in-scope producer's payload carries `actor_type` `agent` for an
agent-actor mutation (`001.13`). This is the test that makes losing the field
a failure rather than a silent unbinding of the gate.
- A wake from the event-subscriber producer is unaffected by an exhausted
allowance, fixing the producer scope as a test rather than a comment
(`001.12`).
- Two admitted wakes sharing one persisted request instant both count, and the
decision at `N` is the same whichever of them was enqueued first — the gate
reads a count, not a sequence (`001.10`).
- An unassignment neither refuses nor consumes (`001.9`).
- A row whose `requested_at` is exactly `W` old does not count (`001.11`).
- A failing count read admits and increments the degraded counter, and logs
the fields `002.2` names without an observed count (`002.2`); an
unattributable task likewise, without a task identifier (`002.3`).
- A wake carrying neither an `actor_type` nor a task moves no counter at all,
pinning the predicate's evaluation order (`001.1`, `002.3`).
- A mixed-actor coalesce in both directions: an agent-actor wake merging into a
queued user-actor row, and a user-actor wake merging into a queued agent-actor
row. Assert the count the gate then observes — the accepted off-by-one of the
requirement's `## Accepted consequence`, not an assumed exact count — so a
later change to `CoalesceRun` breaks a test rather than silently moving the
allowance (`001.4`, `001.5`).
- Re-delivering a *keyed* wake does not consume the allowance twice;
re-delivering a *keyless* one does, which is what `002.5` states and what
`001.8` bounds. Two cases, not one.
- The `task_assigned` producers are enumerated and the actor type each one emits
is asserted, so a new producer setting `actor_type` `agent` fails this test
instead of joining the gate unannounced (`001.12`).
- An absent, null or empty `task_id` and a present-but-non-string `task_id` each
take the unattributed path; an unparseable payload instead leaves scope at the
actor step, admitted with no counter moved, so the two are distinguishable in
the counters (`002.3`, `001.1`).
- Refusal moves the counter with the expected `reason` label and no identifier
label, and logs the five fields `003.3` names (`003.1`, `003.2`, `003.3`); an
ordinary admit moves none (`003.4`).
Time is injected rather than slept — `synctest` per the repo's test discipline,
never `time.Sleep`. PostgreSQL twins of the window-count query belong beside
the existing `runs_queue_postgres_test.go` cases, for the same reason those
exist.
No E2E coverage: `REQ-OFFICE-ASSIGN-RATE-001`'s `## Out of scope` excludes any
frontend, so no user-visible surface changes and there is nothing for
Playwright to drive.