Files
coder/coderd/x/chatd/chathooks/trigger.go
T
Michael Suchacz fc24c27dfd fix: reserve chat hook dispatch capacity for running turns (#27656)
## Context

Follow-up fix from live UAT of the merged chat lifecycle hooks stack
(#27430). Its companion UAT fix (#27655) has merged, so this targets
`main` directly.

## Why?

UAT measured a burst of 1,500 concurrent chat creations against a
consumer with 1.2s latency. 255 were admitted and 1,245 got `502
hook_dispatch_failed (over_capacity)`, which is correct fail-closed
behavior. The collateral wasn't: the same burst failed 24 `stop`
dispatches, parking chats that had already been admitted and had already
executed tools. One 256-slot semaphore served every event, so new-work
admission could take every slot and kill turns in flight.

Callers now classify each dispatch as admission or generation, and
admission draws from a 192-slot gate held *before* the shared pool. At
least 64 shared slots stay reachable only by dispatches for work a chat
already admitted. The dispatcher is per `coderd` replica, so these
limits are per replica, not deployment-wide, and the docs say so.

**The caller classifies, not the event type.** Event type isn't a
reliable proxy in either direction: a subagent spawn dispatches
`user_prompt_submit` from inside a running turn, and the edit path
dispatches `session_start` at admission time. `CapacityClassUnset` is
rejected in `Dispatch`, so a new call site fails closed rather than
silently inheriting a share.

**Acquisition order is load-bearing.** Admission takes its own gate
first. Taking a shared slot first would let admissions queued on the
gate occupy the very capacity the reserve protects. `acquireCapacity` is
the only path that takes either pool, so the order can't be bypassed.

## What this does not guarantee

Nothing bounds how many turns generate concurrently, so the 192/64 split
is a judgement call, not a derived ceiling. This stops an *admission*
burst from consuming every slot; it does not make the remainder
sufficient. A large enough generation load can still exhaust the reserve
and error a running chat. The docs say so explicitly rather than
promising a guarantee the code doesn't deliver.

Generation can now take all 256 slots, so generation traffic starves
admission harder than before. That's the intended priority: rejecting a
new prompt is recoverable, ending a turn that already ran tools is not.

## Testing

Red-green proved both new tests. Removing the release-on-failure path
fails `RefusedSharedAcquireReleasesAdmission` deterministically;
removing the expired-deadline check fails
`ExpiredDeadlineRefusesFreeSlot` in 18/30 runs.

That deadline check fixes a real race found in review. `acquire`
previously shared one `time.Timer` across both acquires. Because
`select` picks a ready case at random, an admission dispatch could take
a slot after its capacity deadline had passed. Measured over 300 trials:
135 late acquisitions, worst overshoot 2.1ms. `acquire` now takes an
absolute deadline and refuses an expired one before selecting, which
measures 0/300.

Go: `coderd/x/agenthooks/...` and `coderd/x/chatd/...`, plus `-race
-count=3` on the dispatcher.

> Mux opened this PR on Mike's behalf.
2026-08-03 19:04:54 +02:00

205 lines
5.4 KiB
Go

// Package chathooks integrates chat lifecycle hooks into chatd: it
// builds event envelopes, dispatches them, and converts consumer
// responses into transcript effects and permission decisions.
package chathooks
import (
"context"
"encoding/json"
"strings"
"github.com/google/uuid"
"golang.org/x/xerrors"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/x/agenthooks/dispatch"
"github.com/coder/coder/v2/coderd/x/chatd/chatprompt"
"github.com/coder/coder/v2/codersdk"
"github.com/coder/coder/v2/codersdk/x/agenthooks"
)
const (
SessionStartSourceStartup = "startup"
SessionStartSourceResume = "resume"
SessionStartSourceClear = "clear"
)
func SessionStartSource(messages []database.ChatMessage) string {
for _, message := range messages {
if message.Role == database.ChatMessageRoleAssistant {
return SessionStartSourceResume
}
}
return SessionStartSourceStartup
}
// Trigger is the only component that talks to the hook dispatcher.
// Every lifecycle event flows through trigger, which builds the wire
// envelope, dispatches, and normalizes the outcome.
type Trigger struct {
dispatcher *dispatch.Dispatcher
}
func NewTrigger(dispatcher *dispatch.Dispatcher) *Trigger {
return &Trigger{dispatcher: dispatcher}
}
func (t *Trigger) Enabled() bool {
return t != nil && t.dispatcher.Enabled()
}
// Chat identifies the chat and turn an event belongs to. Admission
// events for chats that do not exist yet (create, subagent spawn) fill
// the fields directly instead of loading a row.
type Chat struct {
ID uuid.UUID
OwnerID uuid.UUID
WorkspaceID uuid.NullUUID
ParentChatID uuid.NullUUID
RootChatID uuid.NullUUID
TurnID *uuid.UUID
}
func ChatFor(chat database.Chat, turnID *uuid.UUID) Chat {
return Chat{
ID: chat.ID,
OwnerID: chat.OwnerID,
WorkspaceID: chat.WorkspaceID,
ParentChatID: chat.ParentChatID,
RootChatID: chat.RootChatID,
TurnID: turnID,
}
}
func (c Chat) ref() agenthooks.ChatRef {
ref := agenthooks.ChatRef{
ChatID: c.ID,
OwnerID: c.OwnerID,
TurnID: c.TurnID,
}
if c.WorkspaceID.Valid {
ref.WorkspaceID = &c.WorkspaceID.UUID
}
if c.ParentChatID.Valid {
ref.ParentChatID = &c.ParentChatID.UUID
}
if c.RootChatID.Valid {
ref.RootChatID = &c.RootChatID.UUID
}
return ref
}
type Message struct {
Source string
Prompt string
Parts json.RawMessage
ToolUseID string
ToolName string
ToolInput json.RawMessage
ToolResponse json.RawMessage
ToolError string
}
func UserPromptMessage(parts []codersdk.ChatMessagePart) (Message, error) {
encoded, err := chatprompt.MarshalParts(parts)
if err != nil {
return Message{}, xerrors.Errorf("marshal prompt parts for hook: %w", err)
}
return Message{
Prompt: textFromParts(parts),
Parts: encoded.RawMessage,
}, nil
}
// Result is a consumer response normalized for callers: a non-empty
// InputOverride means the permission decision was allow with a
// replacement input (the wire contract rejects allow without one).
// Denials surface as *deniedError instead.
type Result struct {
InputOverride json.RawMessage
ModelContext string
UserMessage string
}
var emptyResult = &Result{}
func (r *Result) GetModelContext() string {
if r == nil {
return ""
}
return r.ModelContext
}
func (r *Result) GetUserMessage() string {
if r == nil {
return ""
}
return r.UserMessage
}
func (t *Trigger) Trigger(
ctx context.Context,
chat Chat,
msg Message,
event agenthooks.EventType,
capacity dispatch.CapacityClass,
) (*Result, error) {
if !t.Enabled() {
return emptyResult, nil
}
var data any
switch event {
case agenthooks.EventSessionStart:
data = agenthooks.SessionStartData{Source: msg.Source}
case agenthooks.EventUserPromptSubmit:
data = agenthooks.UserPromptSubmitData{Prompt: msg.Prompt, Parts: msg.Parts}
case agenthooks.EventPreToolUse:
data = agenthooks.PreToolUseData{ToolUseID: msg.ToolUseID, ToolName: msg.ToolName, ToolInput: msg.ToolInput}
case agenthooks.EventPostToolUse:
data = agenthooks.PostToolUseData{ToolUseID: msg.ToolUseID, ToolName: msg.ToolName, ToolResponse: msg.ToolResponse, ToolError: msg.ToolError}
case agenthooks.EventPreCompact:
data = agenthooks.PreCompactData{}
case agenthooks.EventPostCompact:
data = agenthooks.PostCompactData{}
case agenthooks.EventStop:
data = agenthooks.StopData{}
default:
return nil, xerrors.Errorf("unsupported hook event %q", event)
}
response, _, err := t.dispatcher.Dispatch(ctx, dispatch.Event{
Type: event,
ChatRef: chat.ref(),
Data: data,
Capacity: capacity,
})
if err != nil {
return nil, err
}
if response.Permission != nil && response.Permission.Decision == agenthooks.PermissionDeny {
return nil, &deniedError{
Event: event,
Reason: response.Permission.Reason,
ModelContext: response.ModelContext,
UserMessage: response.UserMessage,
}
}
result := &Result{
ModelContext: response.ModelContext,
UserMessage: response.UserMessage,
}
if response.Permission != nil && response.Permission.Decision == agenthooks.PermissionAllow {
result.InputOverride = response.Permission.InputOverride
}
return result, nil
}
func textFromParts(parts []codersdk.ChatMessagePart) string {
var builder strings.Builder
for _, part := range parts {
if part.Type == codersdk.ChatMessagePartTypeText {
_, _ = builder.WriteString(part.Text)
}
}
return builder.String()
}