diff --git a/coderd/x/agenthooks/dispatch/dispatcher.go b/coderd/x/agenthooks/dispatch/dispatcher.go index db7e1566e1..b0c58e0ac5 100644 --- a/coderd/x/agenthooks/dispatch/dispatcher.go +++ b/coderd/x/agenthooks/dispatch/dispatcher.go @@ -28,11 +28,22 @@ import ( const ( maxConcurrentDispatches = 256 - maxResponseBodyBytes = 1_048_576 - maxModelContextBytes = 16_384 - capacityWaitLimit = 250 * time.Millisecond - retryBackoff = 250 * time.Millisecond - clockSkewLeeway = 30 * time.Second + // Keep capacity reachable by work that a chat already admitted. + maxAdmissionDispatches = 192 + maxResponseBodyBytes = 1_048_576 + maxModelContextBytes = 16_384 + capacityWaitLimit = 250 * time.Millisecond + retryBackoff = 250 * time.Millisecond + clockSkewLeeway = 30 * time.Second +) + +// CapacityClass selects which share of dispatch capacity an event draws from. +type CapacityClass int + +const ( + CapacityClassUnset CapacityClass = iota + CapacityClassAdmission + CapacityClassGeneration ) // Result classifies the terminal outcome of a dispatch attempt. @@ -53,7 +64,8 @@ const ( type Event struct { Type agenthooks.EventType agenthooks.ChatRef - Data any + Data any + Capacity CapacityClass } // Error preserves the attempt ID and failure class. @@ -90,6 +102,7 @@ type Dispatcher struct { deploymentID string userAgent string semaphore chan struct{} + admission chan struct{} metrics *metrics } @@ -167,6 +180,7 @@ func New( deploymentID: deploymentID, userAgent: "coderd-agenthooks/" + coderVersion, semaphore: make(chan struct{}, maxConcurrentDispatches), + admission: make(chan struct{}, maxAdmissionDispatches), metrics: newMetrics(reg), } } @@ -187,25 +201,21 @@ func (d *Dispatcher) Dispatch(ctx context.Context, event Event) (agenthooks.Resp return agenthooks.Response{}, uuid.Nil, xerrors.Errorf("chat hook URL rejected: %w", d.hookURLErr) } + switch event.Capacity { + case CapacityClassAdmission, CapacityClassGeneration: + default: + return agenthooks.Response{}, uuid.Nil, xerrors.Errorf("dispatch event %q has no capacity class", event.Type) + } + startedAt := time.Now() dispatchID := uuid.New() - wait := max(min(d.timeout, capacityWaitLimit), 0) - capacityTimer := time.NewTimer(wait) - defer capacityTimer.Stop() + capacityDeadline := startedAt.Add(max(min(d.timeout, capacityWaitLimit), 0)) - // The capacity wait runs against its own timer rather than a dispatch - // deadline, so a timeout shorter than capacityWaitLimit cannot make the - // over-capacity and caller-cancellation cases race. - select { - case d.semaphore <- struct{}{}: - defer func() { <-d.semaphore }() - case <-ctx.Done(): - outcome := dispatchOutcome{result: ResultTimeout, err: ctx.Err()} - return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, outcome) - case <-capacityTimer.C: - outcome := dispatchOutcome{result: ResultOverCapacity, err: context.DeadlineExceeded} - return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, outcome) + release, refused, ok := d.acquireCapacity(ctx, event.Capacity, capacityDeadline) + if !ok { + return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, refused) } + defer release() // Both post attempts share whatever remains of the configured timeout so // that waiting for capacity cannot extend the dispatch past it. @@ -219,6 +229,55 @@ func (d *Dispatcher) Dispatch(ctx context.Context, event Event) (agenthooks.Resp return outcome.response, dispatchID, nil } +// Admission acquires its gate before the shared pool so queued admission +// dispatches cannot occupy the reserved capacity. +func (d *Dispatcher) acquireCapacity( + ctx context.Context, + capacity CapacityClass, + deadline time.Time, +) (release func(), outcome dispatchOutcome, ok bool) { + if capacity == CapacityClassAdmission { + releaseAdmission, refused, admitted := acquire(ctx, d.admission, deadline) + if !admitted { + return nil, refused, false + } + releaseShared, refused, acquired := acquire(ctx, d.semaphore, deadline) + if !acquired { + releaseAdmission() + return nil, refused, false + } + return func() { + releaseShared() + releaseAdmission() + }, dispatchOutcome{}, true + } + return acquire(ctx, d.semaphore, deadline) +} + +// Check the deadline before select because select randomly chooses among ready +// cases, including a free slot and an expired timer. +func acquire( + ctx context.Context, + pool chan struct{}, + deadline time.Time, +) (release func(), outcome dispatchOutcome, ok bool) { + overCapacity := dispatchOutcome{result: ResultOverCapacity, err: context.DeadlineExceeded} + remaining := time.Until(deadline) + if remaining <= 0 { + return nil, overCapacity, false + } + timer := time.NewTimer(remaining) + defer timer.Stop() + select { + case pool <- struct{}{}: + return func() { <-pool }, dispatchOutcome{}, true + case <-ctx.Done(): + return nil, dispatchOutcome{result: ResultTimeout, err: ctx.Err()}, false + case <-timer.C: + return nil, overCapacity, false + } +} + func (d *Dispatcher) finish( ctx context.Context, event Event, diff --git a/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go b/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go index 320fd33b14..04915f7e40 100644 --- a/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go +++ b/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go @@ -671,7 +671,8 @@ func newTestEvent(t *testing.T, eventType agenthooks.EventType, data any) Event ChatID: uuid.New(), OwnerID: uuid.New(), }, - Data: data, + Data: data, + Capacity: CapacityClassGeneration, } } @@ -699,3 +700,111 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } + +func TestDispatcherCapacityClassRequired(t *testing.T) { + t.Parallel() + + event := newTestEvent(t, agenthooks.EventStop, agenthooks.StopData{}) + event.Capacity = CapacityClassUnset + dispatcher := newTestDispatcher(t, nil, "https://unused.test", time.Second) + + _, dispatchID, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitShort), event) + require.ErrorContains(t, err, "no capacity class") + require.Equal(t, uuid.Nil, dispatchID) +} + +func TestDispatcherAdmissionReserve(t *testing.T) { + t.Parallel() + + fill := func(t *testing.T, pool chan struct{}, count int) { + t.Helper() + for range count { + pool <- struct{}{} + } + t.Cleanup(func() { + for range count { + <-pool + } + }) + } + + t.Run("AdmissionReleasesBothPools", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := w.Write([]byte(`{}`)) + assert.NoError(t, err) + })) + t.Cleanup(server.Close) + + dispatcher := New( + testutil.Logger(t), server.Client(), server.URL, testSecret, testutil.WaitShort, + testDeploymentID, testVersion, prometheus.NewRegistry(), + ) + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + require.NoError(t, err) + require.Empty(t, dispatcher.admission) + require.Empty(t, dispatcher.semaphore) + }) + + t.Run("SaturatedAdmissionRefusesAdmission", func(t *testing.T) { + t.Parallel() + + dispatcher := newTestDispatcher(t, nil, "https://unused.test", 10*time.Millisecond) + fill(t, dispatcher.admission, maxAdmissionDispatches) + + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + assertDispatchErrorClass(t, err, ResultOverCapacity) + require.Empty(t, dispatcher.semaphore, "a refused admission must not hold a shared slot") + }) + + t.Run("ExpiredDeadlineRefusesFreeSlot", func(t *testing.T) { + t.Parallel() + + pool := make(chan struct{}, 1) + release, outcome, ok := acquire(testutil.Context(t, testutil.WaitShort), pool, time.Now().Add(-time.Millisecond)) + require.False(t, ok) + require.Nil(t, release) + require.Equal(t, ResultOverCapacity, outcome.result) + require.Empty(t, pool) + }) + + t.Run("RefusedSharedAcquireReleasesAdmission", func(t *testing.T) { + t.Parallel() + + dispatcher := newTestDispatcher(t, nil, "https://unused.test", 10*time.Millisecond) + fill(t, dispatcher.semaphore, maxConcurrentDispatches) + + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + assertDispatchErrorClass(t, err, ResultOverCapacity) + require.Empty(t, dispatcher.admission, "an admission refused by the shared pool must release its gate token") + }) + + t.Run("SaturatedAdmissionStillServesGeneration", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := w.Write([]byte(`{}`)) + assert.NoError(t, err) + })) + t.Cleanup(server.Close) + + dispatcher := New( + testutil.Logger(t), server.Client(), server.URL, testSecret, testutil.WaitShort, + testDeploymentID, testVersion, prometheus.NewRegistry(), + ) + fill(t, dispatcher.admission, maxAdmissionDispatches) + fill(t, dispatcher.semaphore, maxAdmissionDispatches) + + event := newTestEvent(t, agenthooks.EventStop, agenthooks.StopData{}) + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + require.NoError(t, err) + }) +} diff --git a/coderd/x/chatd/ARCHITECTURE.md b/coderd/x/chatd/ARCHITECTURE.md index f2bfab782a..26c553af8c 100644 --- a/coderd/x/chatd/ARCHITECTURE.md +++ b/coderd/x/chatd/ARCHITECTURE.md @@ -915,6 +915,8 @@ The consumer can observe activity, add model-only or user-visible context, repla Lifecycle hooks fail closed. If the consumer cannot be reached or returns an invalid response, Coder stops the triggering operation rather than continuing without the consumer's decision. Affected chats can enter an error state until the consumer recovers or hooks are disabled. +Concurrent dispatches are capped per replica, and each dispatch declares whether it admits new work into a chat or belongs to work a chat already admitted. Admission can hold only part of the cap, so a burst of new submissions cannot consume the capacity that already-admitted work depends on. The caller declares this, because the event type does not determine it: a subagent spawn submits a prompt from inside a running turn, and editing a message starts a session at admission time. + Coder stores no hook-specific dispatch or decision state. Delivery is best-effort and can duplicate, and a failed dispatch is never redelivered, so the consumer owns durable policy state, audit records, and deduplication based on stable event identifiers. # Stream loop diff --git a/coderd/x/chatd/chatd.go b/coderd/x/chatd/chatd.go index ccc12ad80c..3cfb17a4bf 100644 --- a/coderd/x/chatd/chatd.go +++ b/coderd/x/chatd/chatd.go @@ -1334,7 +1334,7 @@ func (p *Server) CreateChat(ctx context.Context, opts CreateOptions) (database.C OwnerID: opts.OwnerID, WorkspaceID: opts.WorkspaceID, TurnID: &turnID, - }, promptMessage, agenthooks.EventUserPromptSubmit) + }, promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return database.Chat{}, chathooks.UserPromptDenial(err) } @@ -1487,7 +1487,7 @@ func (p *Server) SendMessage( if err != nil { return SendMessageResult{}, err } - promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit) + promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return SendMessageResult{}, p.handleUserPromptDispatchError(ctx, opts.ChatID, chathooks.UserPromptDenial(err)) } @@ -1791,7 +1791,7 @@ func (p *Server) EditMessage( if _, err := validateModelConfigOverride(ctx, p.db, opts.ModelConfigID); err != nil { return EditMessageResult{}, err } - sessionStartHookResult, err = p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), chathooks.Message{Source: chathooks.SessionStartSourceClear}, agenthooks.EventSessionStart) + sessionStartHookResult, err = p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), chathooks.Message{Source: chathooks.SessionStartSourceClear}, agenthooks.EventSessionStart, dispatch.CapacityClassAdmission) if err != nil { return EditMessageResult{}, p.handleAPIDispatchError(ctx, opts.ChatID, agenthooks.EventSessionStart, err) } @@ -1799,7 +1799,7 @@ func (p *Server) EditMessage( if err != nil { return EditMessageResult{}, err } - promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit) + promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return EditMessageResult{}, p.handleUserPromptDispatchError(ctx, opts.ChatID, chathooks.UserPromptDenial(err)) } @@ -2170,7 +2170,7 @@ func (p *Server) SubmitToolResults( return err } for _, result := range opts.Results { - response, err := p.hooks.Trigger(ctx, chathooks.ChatFor(state.chat, nil), chathooks.DynamicPostToolUseMessage(result, state.toolNames[result.ToolCallID]), agenthooks.EventPostToolUse) + response, err := p.hooks.Trigger(ctx, chathooks.ChatFor(state.chat, nil), chathooks.DynamicPostToolUseMessage(result, state.toolNames[result.ToolCallID]), agenthooks.EventPostToolUse, dispatch.CapacityClassGeneration) if err != nil { // Leave pending calls intact so the client can resubmit after recovery. return chathooks.GenerationDispatchError(agenthooks.EventPostToolUse, err) diff --git a/coderd/x/chatd/chathooks/hooks_internal_test.go b/coderd/x/chatd/chathooks/hooks_internal_test.go index 58a8dab9ad..f30b475129 100644 --- a/coderd/x/chatd/chathooks/hooks_internal_test.go +++ b/coderd/x/chatd/chathooks/hooks_internal_test.go @@ -65,9 +65,9 @@ func TestSessionStartDispatchSources(t *testing.T) { turnID := uuid.New() ctx := testutil.Context(t, testutil.WaitLong) - _, err := trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource(nil)}, agenthooks.EventSessionStart) + _, err := trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource(nil)}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) require.NoError(t, err) - _, err = trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource([]database.ChatMessage{{Role: database.ChatMessageRoleAssistant}})}, agenthooks.EventSessionStart) + _, err = trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource([]database.ChatMessage{{Role: database.ChatMessageRoleAssistant}})}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) require.NoError(t, err) startup := <-receivedCh @@ -120,7 +120,7 @@ func TestHookTriggerDisabled(t *testing.T) { t.Run(name, func(t *testing.T) { t.Parallel() require.False(t, trigger.Enabled()) - result, err := trigger.Trigger(t.Context(), Chat{ID: uuid.New()}, Message{}, agenthooks.EventStop) + result, err := trigger.Trigger(t.Context(), Chat{ID: uuid.New()}, Message{}, agenthooks.EventStop, dispatch.CapacityClassGeneration) require.NoError(t, err) require.Empty(t, result.GetModelContext()) require.Empty(t, result.GetUserMessage()) @@ -145,7 +145,7 @@ func TestHookTriggerDeny(t *testing.T) { ToolUseID: "call_1", ToolName: "execute", ToolInput: json.RawMessage(`{}`), - }, agenthooks.EventPreToolUse) + }, agenthooks.EventPreToolUse, dispatch.CapacityClassGeneration) require.Nil(t, result) var denied *deniedError require.ErrorAs(t, err, &denied) @@ -174,7 +174,7 @@ func TestHookTriggerEventPayloads(t *testing.T) { ctx := testutil.Context(t, testutil.WaitShort) dispatchEvent := func(t *testing.T, msg Message, event agenthooks.EventType) agenthooks.Request { t.Helper() - _, err := trigger.Trigger(ctx, chat, msg, event) + _, err := trigger.Trigger(ctx, chat, msg, event, dispatch.CapacityClassGeneration) require.NoError(t, err) request := <-requests require.Equal(t, event, request.Type) @@ -215,7 +215,7 @@ func TestHookTriggerEventPayloads(t *testing.T) { dispatchEvent(t, Message{}, event) } - _, err := trigger.Trigger(ctx, chat, Message{}, agenthooks.EventType("bogus")) + _, err := trigger.Trigger(ctx, chat, Message{}, agenthooks.EventType("bogus"), dispatch.CapacityClassGeneration) require.ErrorContains(t, err, "unsupported hook event") } diff --git a/coderd/x/chatd/chathooks/tooluse.go b/coderd/x/chatd/chathooks/tooluse.go index 09dd15e2d4..daa19202a8 100644 --- a/coderd/x/chatd/chathooks/tooluse.go +++ b/coderd/x/chatd/chathooks/tooluse.go @@ -8,6 +8,7 @@ import ( "charm.land/fantasy" "golang.org/x/xerrors" + "github.com/coder/coder/v2/coderd/x/agenthooks/dispatch" "github.com/coder/coder/v2/codersdk" "github.com/coder/coder/v2/codersdk/x/agenthooks" ) @@ -59,7 +60,7 @@ func (t *Trigger) PreflightPendingToolCalls( ToolUseID: toolCall.ToolCallID, ToolName: toolCall.ToolName, ToolInput: json.RawMessage(toolCall.Input), - }, agenthooks.EventPreToolUse) + }, agenthooks.EventPreToolUse, dispatch.CapacityClassGeneration) if err != nil { denied, ok := errors.AsType[*deniedError](err) if !ok { @@ -139,7 +140,7 @@ func (t *Trigger) PostToolUseResults( } continue } - result, err := t.Trigger(ctx, chat, msg, agenthooks.EventPostToolUse) + result, err := t.Trigger(ctx, chat, msg, agenthooks.EventPostToolUse, dispatch.CapacityClassGeneration) if err != nil { if firstErr == nil { firstErr = err diff --git a/coderd/x/chatd/chathooks/trigger.go b/coderd/x/chatd/chathooks/trigger.go index 3af38fa65b..6b705fd4dc 100644 --- a/coderd/x/chatd/chathooks/trigger.go +++ b/coderd/x/chatd/chathooks/trigger.go @@ -142,6 +142,7 @@ func (t *Trigger) Trigger( chat Chat, msg Message, event agenthooks.EventType, + capacity dispatch.CapacityClass, ) (*Result, error) { if !t.Enabled() { return emptyResult, nil @@ -166,9 +167,10 @@ func (t *Trigger) Trigger( return nil, xerrors.Errorf("unsupported hook event %q", event) } response, _, err := t.dispatcher.Dispatch(ctx, dispatch.Event{ - Type: event, - ChatRef: chat.ref(), - Data: data, + Type: event, + ChatRef: chat.ref(), + Data: data, + Capacity: capacity, }) if err != nil { return nil, err diff --git a/coderd/x/chatd/generation.go b/coderd/x/chatd/generation.go index 8a440187a0..23179195b9 100644 --- a/coderd/x/chatd/generation.go +++ b/coderd/x/chatd/generation.go @@ -14,6 +14,7 @@ import ( "cdr.dev/slog/v3" "github.com/coder/coder/v2/coderd/database" + "github.com/coder/coder/v2/coderd/x/agenthooks/dispatch" "github.com/coder/coder/v2/coderd/x/chatd/chatdebug" "github.com/coder/coder/v2/coderd/x/chatd/chaterror" "github.com/coder/coder/v2/coderd/x/chatd/chathooks" @@ -399,7 +400,7 @@ func (s *taskStarter) startGenerationSession( // Re-arm the claim until its response is applied so a replacement task // can replay session_start effects. defer func() { complete(completed) }() - response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{Source: chathooks.SessionStartSource(messages)}, agenthooks.EventSessionStart) + response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{Source: chathooks.SessionStartSource(messages)}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) if err != nil { return sessionStartResult{}, true, chathooks.GenerationDispatchError(agenthooks.EventSessionStart, err) } @@ -940,7 +941,7 @@ func (s *taskStarter) generateCompaction( overrideModel.modelConfig, ) } - preResult, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPreCompact) + preResult, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPreCompact, dispatch.CapacityClassGeneration) if err != nil { return chathooks.GenerationDispatchError(agenthooks.EventPreCompact, err) } @@ -987,7 +988,7 @@ func (s *taskStarter) generateCompaction( // Hook effects and fail-closed errors must commit atomically with // compaction; a separate commit races the runner and can be dropped // on crash. - postResult, postDispatchErr := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPostCompact) + postResult, postDispatchErr := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPostCompact, dispatch.CapacityClassGeneration) var postCommitErr error if postDispatchErr != nil { postCommitErr = chathooks.GenerationDispatchError(agenthooks.EventPostCompact, postDispatchErr) @@ -1349,7 +1350,7 @@ func (s *taskStarter) finishGenerationTurn( if err != nil { return normalizeTaskTransitionError(err, "load stop hook state") } - response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventStop) + response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventStop, dispatch.CapacityClassGeneration) if err != nil { return s.finishGenerationError(ctx, machine, input, chathooks.GenerationDispatchError(agenthooks.EventStop, err), fence) } diff --git a/coderd/x/chatd/subagent.go b/coderd/x/chatd/subagent.go index 8a99b07acd..a9fccfcc17 100644 --- a/coderd/x/chatd/subagent.go +++ b/coderd/x/chatd/subagent.go @@ -1313,7 +1313,7 @@ func (p *Server) createChildSubagentChatWithOptions( ParentChatID: uuid.NullUUID{UUID: parent.ID, Valid: true}, RootChatID: uuid.NullUUID{UUID: rootChatID, Valid: true}, TurnID: &mintedTurnID, - }, promptMessage, agenthooks.EventUserPromptSubmit) + }, promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassGeneration) if err != nil { return database.Chat{}, chathooks.UserPromptDenial(err) } diff --git a/docs/admin/setup/chat-lifecycle-hooks.md b/docs/admin/setup/chat-lifecycle-hooks.md index e33cea7991..5be326cb4e 100644 --- a/docs/admin/setup/chat-lifecycle-hooks.md +++ b/docs/admin/setup/chat-lifecycle-hooks.md @@ -167,6 +167,13 @@ Coder checks admission before dispatching, but concurrent requests can still fai The consumer then observes an event for a request that Coder rejects, and the rejected request doesn't persist a prompt or tool result. Treat events as attempt notifications rather than proof of a committed operation, and key idempotent tool-event processing on `tool_use_id`. +Each `coderd` replica runs at most 256 dispatches at once and waits up to 250 ms for a free slot; a dispatch that waits out that limit fails as over capacity. +The limit is per replica rather than deployment-wide, so size the consumer for 256 concurrent requests per replica. +Slow consumer responses hold slots for longer, so a slow consumer turns a burst of chat activity into over-capacity failures. +Prompt admission (creating, sending, or editing a message) can hold at most 192 of a replica's slots, so at least 64 stay reachable only by dispatches for work a chat already admitted. +That bound stops a burst of new submissions from consuming every slot, but it doesn't make the remaining slots sufficient: a saturated dispatcher can still fail a dispatch for a running chat and leave that chat in the error state. +Watch `coderd_chatd_hook_dispatches_total{result="over_capacity"}` to see whether the consumer's latency is turning normal traffic into rejections. + Delivery is best-effort and can duplicate. Coder never queues a failed dispatch for redelivery, so plan for duplicates without assuming every event arrives. Coder retries one connection failure per dispatch with the same JWT, so use `dispatch_id` to recognize a repeated HTTP attempt and return the same response.