fix: count queue entries on marker wins and require newer status events

ibetitsmike · Aug 18, 2026 · Aug 5, 2026 · Aug 5, 2026 · Aug 5, 2026 · Aug 5, 2026
commit da90da4cc55983152d1df272ab72fe1bb0c67a55
diff --git a/coderd/database/querier.go b/coderd/database/querier.go
diff --git a/coderd/database/queries.sql.go b/coderd/database/queries.sql.go
diff --git a/coderd/database/queries/chats.sql b/coderd/database/queries/chats.sql
@@ -2327,13 +2327,9 @@ WHERE chat_id = @chat_id::uuid
 -- Missing ownership is worker_id IS NULL. Inconsistent ownership is
 -- runner_id IS NULL while worker_id is set. Stale ownership is no
 -- heartbeat row for (chat_id, runner_id), or one older than
--- @stale_seconds by database time. Ordering: interrupting first (stop
--- requests skip the queue), then requires_action (bypasses capacity
--- admission), then running chats interleaved across the root/subagent
--- pools, FIFO inside each pool. Interleaving keeps each pool's oldest
--- candidate near the front so a deep backlog in one full pool cannot
--- starve the other, letting the worker end a pass once a batch makes
--- no progress. @exclude_ids advances past refusals.
+-- Interrupting and requires_action bypass admission. Running chats are FIFO
+-- in interleaved root/subagent pools so one full pool cannot starve the
+-- other. @stale_seconds uses DB time; @exclude_ids skips refusals.
 SELECT
     chats_expanded.*,
     chat_heartbeats.heartbeat_at AS current_heartbeat_at,

diff --git a/coderd/x/chatd/ARCHITECTURE.md b/coderd/x/chatd/ARCHITECTURE.md
@@ -959,7 +959,7 @@ Admission decisions are made per pool: root chats (`parent_chat_id IS NULL`) and
 
 When a runner observes its chat leaving the counted statuses, it publishes a bare ownership hint so all workers immediately re-run acquisition and admit queued chats; the acquisition ticker (default 30s) is the fallback for missed nudges, entitlement changes, and heartbeat-staleness releases. Admission clears `capacity_queued_at` in the acquiring transaction and the worker publishes `capacity_change` so open chat pages clear the banner without a reload. `UpdateChatExecutionState`, `UpdateChatStatus`, archive, and auto-archive clear stray markers whenever a chat leaves the counted statuses.
 
-The bypass rule for licensed deployments: uncapped only while `codersdk.FeatureAgentRuntimeHours` is enabled and recorded usage (`Actual`) is below the allocation (`Limit`). A nil `Actual` or `Limit` fails open, because capping paying customers on a missing usage reading is worse than bounded over-admission. Entitlements do not yet populate `Actual` for this feature (runtime-hour usage accrues externally via usage events), so today an enabled runtime-hours license is always uncapped; the exhaustion comparison activates when usage wiring lands. Queued chats admit automatically after entitlement changes via the acquisition ticker.
+The bypass rule for licensed deployments: uncapped only while `codersdk.FeatureAgentRuntimeHours` is enabled and recorded usage (`Actual`) is below the allocation (`Limit`). A nil `Actual` or `Limit` fails open, because capping paying customers on a missing usage reading is worse than bounded over-admission. Runtime-hour license decoding leaves `Actual` unset, so enabled licenses take this fail-open path. Queued chats admit automatically after entitlement changes via the acquisition ticker.
 
 ## Auto-archive loop
 

diff --git a/coderd/x/chatd/agentadmission.go b/coderd/x/chatd/agentadmission.go
@@ -17,6 +17,10 @@ type AgentAdmission interface {
 	// Admit reports whether the worker may acquire the chat. Refused chats
 	// remain unowned and are retried from the capacity queue.
 	Admit(ctx context.Context, store database.Store, chat database.Chat) (bool, error)
+	// RecordQueued observes a chat entering the capacity queue. Workers call
+	// it only after their queue-marker write wins, so replicas refusing the
+	// same chat concurrently record one queue entry.
+	RecordQueued()
 }
 
 // AgentAdmissionOptions configures an AgentAdmissionFactory.

diff --git a/coderd/x/chatd/agentadmission_internal_test.go b/coderd/x/chatd/agentadmission_internal_test.go
@@ -18,6 +18,9 @@ import (
 type fakeAdmission struct {
 	mu      sync.Mutex
 	refused map[uuid.UUID]bool
+
+	admitCalls    int
+	queuedRecords int
 }
 
 func newFakeAdmission() *fakeAdmission {
@@ -39,9 +42,28 @@ func (f *fakeAdmission) allow(chatID uuid.UUID) {
 func (f *fakeAdmission) Admit(_ context.Context, _ database.Store, chat database.Chat) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	f.admitCalls++
 	return !f.refused[chat.ID], nil
 }
 
+func (f *fakeAdmission) RecordQueued() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.queuedRecords++
+}
+
+func (f *fakeAdmission) admitCallCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.admitCalls
+}
+
+func (f *fakeAdmission) queuedRecordCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.queuedRecords
+}
+
 func capacityQueuedAt(ctx context.Context, t *testing.T, db database.Store, chatID uuid.UUID) *database.Chat {
 	t.Helper()
 	chat, err := db.GetChatByID(ctx, chatID)
@@ -100,6 +122,32 @@ func TestWorker_AdmissionRefusalQueuesChat(t *testing.T) {
 	require.Equal(t, 0, chatOwnershipMessages(t, recording, chat.ID))
 }
 
+func TestWorker_RefusalRecordsQueueEntryOnce(t *testing.T) {
+	t.Parallel()
+	f := newWorkerTestFixture(t)
+	starter := newRecordingTaskStarter()
+	admission := newFakeAdmission()
+	opts := testOptions(t, f, starter)
+	opts.AgentAdmission = admission
+
+	chat := f.createRunningChat(t)
+	admission.refuse(chat.ID)
+	worker := startWorker(t, opts)
+
+	ctx := testutil.Context(t, testutil.WaitLong)
+	require.Eventually(t, func() bool {
+		return capacityQueuedAt(ctx, t, f.db, chat.ID).CapacityQueuedAt.Valid
+	}, testutil.WaitLong, testutil.IntervalFast)
+	require.Equal(t, 1, admission.queuedRecordCount())
+
+	admitCallsAfterMark := admission.admitCallCount()
+	worker.Wake()
+	require.Eventually(t, func() bool {
+		return admission.admitCallCount() > admitCallsAfterMark
+	}, testutil.WaitLong, testutil.IntervalFast)
+	require.Equal(t, 1, admission.queuedRecordCount())
+}
+
 func TestWorker_AdmissionAdmitClearsQueueMark(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
@@ -159,8 +207,6 @@ func TestWorker_InterruptingSortsBeforeCapacityQueue(t *testing.T) {
 	require.Equal(t, queued.ID, rows[2].ID)
 }
 
-// rootRefusingAdmission simulates a full root pool with free subagent
-// capacity: running root chats refuse, everything else admits.
 type rootRefusingAdmission struct {
 	mu    sync.Mutex
 	calls int
@@ -176,6 +222,8 @@ func (a *rootRefusingAdmission) Admit(_ context.Context, _ database.Store, chat
 	return chat.ParentChatID.Valid, nil
 }
 
+func (*rootRefusingAdmission) RecordQueued() {}
+
 func (a *rootRefusingAdmission) callCount() int {
 	a.mu.Lock()
 	defer a.mu.Unlock()
@@ -190,9 +238,8 @@ func TestWorker_FullPoolDoesNotStarveOtherPool(t *testing.T) {
 	opts.AgentAdmission = &rootRefusingAdmission{}
 	ctx := testutil.Context(t, testutil.WaitLong)
 
-	// A pre-marked root backlog deeper than two batches; the subagent is
-	// created last, so an ordering that ignores pools buries it behind a
-	// batch of pure re-skips and the pass ends before reaching it.
+	// The root backlog is deep enough to hide the later subagent without
+	// pool-interleaved ordering.
 	roots := make([]database.Chat, 0, 2*int(opts.AcquisitionBatchSize)+5)
 	for range cap(roots) {
 		chat := f.createRunningChat(t)
@@ -212,11 +259,6 @@ func TestWorker_FullPoolDoesNotStarveOtherPool(t *testing.T) {
 	require.Equal(t, sub.ID, call.input.ChatID)
 }
 
-// A configured batch size of 1 would only ever surface the
-// tie-break-favored root pool: with two already-marked queued roots the
-// pass refuses one, re-skips the other without progress, and ends
-// before examining the subagent pool. The floor of 2 keeps both pool
-// heads in every batch.
 func TestWorker_BatchSizeOneCannotHideAPool(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
@@ -243,10 +285,6 @@ func TestWorker_BatchSizeOneCannotHideAPool(t *testing.T) {
 	require.Equal(t, sub.ID, call.input.ChatID)
 }
 
-// After the first refusal proves a pool full, the rest of the pass must
-// queue-mark that pool's chats without opening refusal transactions.
-// All-marked is the pass-completion signal, so the count is stable when
-// read.
 func TestWorker_FullPoolSkipsRefusalsAfterFirst(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
@@ -274,10 +312,6 @@ func TestWorker_FullPoolSkipsRefusalsAfterFirst(t *testing.T) {
 		"a full pool must be skipped after one refusal, not re-refused per chat")
 }
 
-// A persisted queue marker (for example left by an enterprise
-// deployment) must clear on acquisition even when no admission hook is
-// configured, or the chat page keeps reporting a generating agent as
-// queued.
 func TestWorker_AcquisitionWithoutAdmissionClearsQueueMark(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
@@ -336,6 +370,8 @@ func (runningRefusingAdmission) Admit(_ context.Context, _ database.Store, chat
 	return chat.Status != database.ChatStatusRunning, nil
 }
 
+func (runningRefusingAdmission) RecordQueued() {}
+
 func TestWorker_InterruptClaimsCapacityQueuedChat(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
@@ -358,9 +394,6 @@ func TestWorker_InterruptClaimsCapacityQueuedChat(t *testing.T) {
 	require.Equal(t, chat.ID, call.input.ChatID)
 }
 
-// One acquisition pass must page past a full pool's backlog (via
-// @exclude_ids) and queue-mark every skipped chat, not just the first
-// batch.
 func TestWorker_AdmissionPassReachesChatsBeyondRefusedBatch(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)

diff --git a/coderd/x/chatd/options.go b/coderd/x/chatd/options.go
@@ -231,11 +231,8 @@ func (o chatWorkerOptions) withDefaults() (chatWorkerOptions, error) {
 	if o.AcquisitionBatchSize <= 0 {
 		o.AcquisitionBatchSize = defaultAcquisitionBatchSize
 	}
-	// The acquisition query interleaves the two capacity pools, so any
-	// batch of two or more contains each pool's oldest candidate and an
-	// all-skipped batch proves nothing admittable remains. A batch of
-	// one sees only the tie-break-favored pool and would end the pass
-	// with the other pool unexamined.
+	// A batch size of one can end the pass on the tie-break-favored pool
+	// before the other pool's head is examined.
 	if o.AcquisitionBatchSize < 2 {
 		o.AcquisitionBatchSize = 2
 	}

diff --git a/coderd/x/chatd/runner.go b/coderd/x/chatd/runner.go
@@ -178,9 +178,8 @@ func (r *runner) processState(state runnerStateUpdate) {
 	r.acceptState(state)
 }
 
-// occupiesCapacitySlot reports whether a chat in this state counts
-// against its concurrency pool. The zero runnerStateUpdate does not, so
-// callers need no hasAcceptedState guard.
+// The zero value is not capacity-counted, so callers need no
+// hasAcceptedState guard.
 func occupiesCapacitySlot(state runnerStateUpdate) bool {
 	return !state.Archived &&
 		(state.Status == database.ChatStatusRunning || state.Status == database.ChatStatusInterrupting)

diff --git a/coderd/x/chatd/worker.go b/coderd/x/chatd/worker.go
@@ -191,10 +191,8 @@ func (w *chatWorker) acquireOnce(ctx context.Context, workerID uuid.UUID, manage
 	// Capacity-refused chats remain candidates, so each batch must exclude
 	// prior attempts or a full pool would hide all later candidates.
 	excludeIDs := []uuid.UUID{}
-	// One refusal proves a pool full for the rest of the pass: later
-	// running chats in that pool are queue-marked without an acquisition
-	// attempt. Keyed by parent_chat_id presence (false = root pool,
-	// true = subagent pool).
+	// One refusal marks the pool full for the rest of the pass. The key is
+	// parent_chat_id presence: false for roots, true for subagents.
 	refusedPools := map[bool]bool{}
 	for {
 		rows, err := w.opts.Store.GetChatWorkerAcquisitionCandidates(ctx, database.GetChatWorkerAcquisitionCandidatesParams{
@@ -234,9 +232,7 @@ func (w *chatWorker) acquireOnce(ctx context.Context, workerID uuid.UUID, manage
 				w.opts.Logger.Warn(ctx, "chatworker acquisition candidate failed", slogError(err))
 			}
 		}
-		// A batch of nothing but re-skipped queued chats means every
-		// admittable candidate was already tried: the query sorts
-		// bypassing statuses and each pool's oldest chats first.
+		// An all-skipped batch means the available pool heads were tried.
 		if !progressed || len(rows) < int(w.opts.AcquisitionBatchSize) {
 			return
 		}
@@ -310,10 +306,8 @@ func (w *chatWorker) acquireCandidate(
 				return errCapacityRefused
 			}
 		}
-		// Clear independently of the admission hook: a persisted marker
-		// must not survive acquisition on a deployment that later runs
-		// without admission (for example an AGPL build over an
-		// enterprise database).
+		// Clear markers left by deployments with admission enabled, even when
+		// this worker has no admission hook.
 		if chat.CapacityQueuedAt.Valid {
 			if _, err := store.ClearChatCapacityQueued(ctx, chat.ID); err != nil {
 				return xerrors.Errorf("clear capacity queue marker: %w", err)
@@ -361,6 +355,9 @@ func (w *chatWorker) markCapacityQueued(ctx context.Context, chatID uuid.UUID) {
 	if marked == 0 {
 		return
 	}
+	if w.opts.AgentAdmission != nil {
+		w.opts.AgentAdmission.RecordQueued()
+	}
 	w.publishCapacityChange(ctx, chatID)
 }
 

diff --git a/enterprise/coderd/x/chatd/agentadmission.go b/enterprise/coderd/x/chatd/agentadmission.go
@@ -160,21 +160,15 @@ func (a *admission) Admit(ctx context.Context, store database.Store, chat databa
 		used, capacity = counts.SubagentCount, a.subagentCapacity
 	}
 	if used >= capacity {
-		if !chat.CapacityQueuedAt.Valid {
-			a.queueTotal.Inc()
-		}
 		return false, nil
 	}
 	a.observeAdmission(chat)
 	return true, nil
 }
 
 // uncapped returns true for an enabled entitlement with remaining hours.
-// Missing limit or usage data also fails open to avoid capping on incomplete
-// entitlement data. Entitlements do not yet populate Actual for this feature
-// (usage accrues externally via usage events), so today every enabled
-// runtime-hours license is uncapped; the exhaustion branch activates when
-// usage wiring lands.
+// Missing limit or usage data fails open to avoid capping on incomplete
+// entitlement data; runtime-hour license decoding leaves Actual unset.
 func (a *admission) uncapped() bool {
 	f, ok := a.entitlements.Feature(codersdk.FeatureAgentRuntimeHours)
 	if !ok || !f.Enabled {
@@ -186,6 +180,10 @@ func (a *admission) uncapped() bool {
 	return *f.Actual < *f.Limit
 }
 
+func (a *admission) RecordQueued() {
+	a.queueTotal.Inc()
+}
+
 func (a *admission) observeAdmission(chat database.Chat) {
 	if !chat.CapacityQueuedAt.Valid {
 		return

diff --git a/enterprise/coderd/x/chatd/agentadmission_internal_test.go b/enterprise/coderd/x/chatd/agentadmission_internal_test.go
@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/xerrors"
 
@@ -118,6 +119,28 @@ func TestAdmission_RootPoolCap(t *testing.T) {
 	require.True(t, admitted)
 }
 
+func TestAdmission_QueueCounterCountsMarkerWinsOnly(t *testing.T) {
+	t.Parallel()
+	f := newAdmissionFixture(t)
+	ctx := testutil.Context(t, testutil.WaitLong)
+	a := testAdmission(f, nil)
+
+	f.occupiedRoot(t)
+	f.occupiedRoot(t)
+
+	chat := f.chat(t, database.Chat{})
+	for range 2 {
+		admitted, err := a.Admit(ctx, f.db, chat)
+		require.NoError(t, err)
+		require.False(t, admitted)
+	}
+	require.Zero(t, promtestutil.ToFloat64(a.queueTotal),
+		"refusals must not count queue entries: concurrent replicas can refuse the same chat before one marker write wins")
+
+	a.RecordQueued()
+	require.Equal(t, float64(1), promtestutil.ToFloat64(a.queueTotal))
+}
+
 func TestAdmission_SubagentPoolCap(t *testing.T) {
 	t.Parallel()
 	f := newAdmissionFixture(t)