mirror of
https://github.com/coder/coder.git
synced 2026-10-31 19:36:41 +08:00
## 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.
205 lines
5.4 KiB
Go
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()
|
|
}
|