401 lines
12 KiB
Go
401 lines
12 KiB
Go
// Package generation coordinates model attempts, backend permits, and bounded
|
|
// built-in HTTP retries for one generation request.
|
|
package generation
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math/rand/v2"
|
|
"time"
|
|
|
|
"go.maximumdirect.net/promptkit/internal/capacity"
|
|
"go.maximumdirect.net/promptkit/internal/domain"
|
|
"go.maximumdirect.net/promptkit/internal/llm"
|
|
"go.maximumdirect.net/promptkit/internal/observation"
|
|
)
|
|
|
|
const (
|
|
initialBackoff = 500 * time.Millisecond
|
|
maximumBackoff = 8 * time.Second
|
|
)
|
|
|
|
// Coordinator implements the internal model client using the existing backend
|
|
// capacity manager. Exactly one of builtIn and injected is selected.
|
|
type Coordinator struct {
|
|
manager *capacity.Manager
|
|
builtIn *llm.OpenAICompatibleClient
|
|
injected llm.Client
|
|
now func() time.Time
|
|
jitter func(time.Duration) time.Duration
|
|
wait func(context.Context, time.Duration) error
|
|
}
|
|
|
|
// NewBuiltIn selects the built-in HTTP retry path.
|
|
func NewBuiltIn(manager *capacity.Manager, client *llm.OpenAICompatibleClient) *Coordinator {
|
|
return &Coordinator{manager: manager, builtIn: client, now: time.Now, jitter: randomJitter, wait: waitContext}
|
|
}
|
|
|
|
// NewInjected selects one opaque client dispatch with no automatic retry.
|
|
func NewInjected(manager *capacity.Manager, client llm.Client) *Coordinator {
|
|
return &Coordinator{manager: manager, injected: client, now: time.Now, jitter: randomJitter, wait: waitContext}
|
|
}
|
|
|
|
// ExhaustedError retains the final eligible status failure and the number of
|
|
// attempts actually dispatched.
|
|
type ExhaustedError struct {
|
|
attempts uint64
|
|
cause error
|
|
}
|
|
|
|
func (e *ExhaustedError) Error() string {
|
|
if e == nil {
|
|
return "transient model retries exhausted"
|
|
}
|
|
return fmt.Sprintf("transient model retries exhausted after %d attempts", e.attempts)
|
|
}
|
|
|
|
func (e *ExhaustedError) Unwrap() error {
|
|
if e == nil {
|
|
return nil
|
|
}
|
|
return e.cause
|
|
}
|
|
|
|
func (e *ExhaustedError) Attempts() uint64 {
|
|
if e == nil {
|
|
return 0
|
|
}
|
|
return e.attempts
|
|
}
|
|
|
|
// Generate executes either one opaque client call or a frozen built-in HTTP
|
|
// sequence with one permit per actual dispatch.
|
|
func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) (*domain.GenerateResponse, error) {
|
|
response, err, _ := c.generate(ctx, req, false)
|
|
return response, err
|
|
}
|
|
|
|
// GenerateWithEvidence runs the built-in retry path and returns its factual
|
|
// attempt evidence alongside any response retained before cancellation.
|
|
func (c *Coordinator) GenerateWithEvidence(
|
|
ctx context.Context,
|
|
req domain.GenerateRequest,
|
|
) (*domain.GenerateResponse, error, GenerationEvidence) {
|
|
return c.generate(ctx, req, true)
|
|
}
|
|
|
|
// PreparationError identifies a built-in request preparation failure.
|
|
type PreparationError struct{ Cause error }
|
|
|
|
func (e *PreparationError) Error() string { return e.Cause.Error() }
|
|
func (e *PreparationError) Unwrap() error { return e.Cause }
|
|
|
|
func (c *Coordinator) generate(
|
|
ctx context.Context,
|
|
req domain.GenerateRequest,
|
|
captureEvidence bool,
|
|
) (*domain.GenerateResponse, error, GenerationEvidence) {
|
|
if c.injected != nil {
|
|
return c.generateInjected(ctx, req, captureEvidence)
|
|
}
|
|
prepared, err := c.builtIn.PrepareGeneration(req)
|
|
if err != nil {
|
|
return nil, &PreparationError{Cause: err}, GenerationEvidence{PreparationFailed: true}
|
|
}
|
|
|
|
evidence := GenerationEvidence{}
|
|
if captureEvidence {
|
|
evidence.InitialEndpoint = prepared.InitialEndpoint()
|
|
}
|
|
var usage observation.UsageAccumulator
|
|
finish := func(response *domain.GenerateResponse, err error) (*domain.GenerateResponse, error, GenerationEvidence) {
|
|
if captureEvidence {
|
|
evidence.Usage = usage.Observed()
|
|
}
|
|
return response, err, evidence
|
|
}
|
|
var attempts, retriesUsed uint64
|
|
recorder := observation.FromContext(ctx)
|
|
generation := req.Generation
|
|
if generation == 0 {
|
|
generation = 1
|
|
}
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
return finish(nil, err)
|
|
}
|
|
ordinal := attempts + 1
|
|
result := c.execute(ctx, req.Target.BackendID, prepared, captureEvidence)
|
|
if result.Dispatched {
|
|
attempts++
|
|
}
|
|
recordAttempt(result, false, captureEvidence, &evidence, &usage, recorder, generation, ordinal)
|
|
if result.Outcome.Err == nil {
|
|
if err := ctx.Err(); err != nil {
|
|
if captureEvidence {
|
|
return finish(result.Outcome.Response, err)
|
|
}
|
|
return finish(nil, err)
|
|
}
|
|
return finish(result.Outcome.Response, nil)
|
|
}
|
|
if !result.Dispatched || !eligibleStatus(result.Outcome) {
|
|
if captureEvidence {
|
|
return finish(result.Outcome.Response, result.Outcome.Err)
|
|
}
|
|
return finish(nil, result.Outcome.Err)
|
|
}
|
|
if retriesUsed == req.TransientRetries {
|
|
if retriesUsed == 0 {
|
|
return finish(nil, result.Outcome.Err)
|
|
}
|
|
return finish(nil, &ExhaustedError{attempts: attempts, cause: result.Outcome.Err})
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return finish(nil, err)
|
|
}
|
|
retryOrdinal := retriesUsed + 1
|
|
delay := retryDelay(c.backoff(retryOrdinal), result.Outcome.RetryNotBefore, c.now())
|
|
waitStart := c.now()
|
|
waitErr := c.wait(ctx, delay)
|
|
waitDuration := elapsed(waitStart, c.now())
|
|
if captureEvidence {
|
|
evidence.RetryDelays = append(evidence.RetryDelays, waitDuration)
|
|
}
|
|
if recorder != nil {
|
|
recorder.Delay(observation.DelayData{Generation: generation, Attempt: attempts,
|
|
Elapsed: waitDuration, Failure: ClassifyFailure(waitErr)})
|
|
}
|
|
if waitErr != nil {
|
|
return finish(nil, waitErr)
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return finish(nil, err)
|
|
}
|
|
retriesUsed++
|
|
}
|
|
}
|
|
|
|
func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateRequest, captureEvidence bool) (*domain.GenerateResponse, error, GenerationEvidence) {
|
|
evidence := GenerationEvidence{}
|
|
var usage observation.UsageAccumulator
|
|
generation := req.Generation
|
|
if generation == 0 {
|
|
generation = 1
|
|
}
|
|
result := c.executeInjected(ctx, req)
|
|
recordAttempt(result, true, captureEvidence, &evidence, &usage, observation.FromContext(ctx), generation, 1)
|
|
if captureEvidence {
|
|
evidence.Usage = usage.Observed()
|
|
}
|
|
if result.Outcome.Err != nil {
|
|
return nil, result.Outcome.Err, evidence
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
if captureEvidence {
|
|
return result.Outcome.Response, err, evidence
|
|
}
|
|
return nil, err, evidence
|
|
}
|
|
return result.Outcome.Response, nil, evidence
|
|
}
|
|
|
|
func (c *Coordinator) executeInjected(ctx context.Context, req domain.GenerateRequest) attemptResult {
|
|
result := attemptResult{StartedAt: c.now()}
|
|
release, err := c.manager.Acquire(ctx, req.Target.BackendID)
|
|
result.CapacityWait = c.capacityWait(req.Target.BackendID, result.StartedAt)
|
|
if err != nil {
|
|
result.Outcome.Err = err
|
|
return result
|
|
}
|
|
defer release()
|
|
if err := ctx.Err(); err != nil {
|
|
result.Outcome.Err = err
|
|
return result
|
|
}
|
|
start := c.now()
|
|
result.Dispatched = true
|
|
requested := req.StructuredOutput != nil
|
|
result.Outcome.StructuredOutputRequested = &requested
|
|
result.Outcome.Response, result.Outcome.Err = c.injected.Generate(ctx, req)
|
|
result.RequestDuration = elapsed(start, c.now())
|
|
if result.Outcome.Err == nil && result.Outcome.Response == nil {
|
|
result.Outcome.Err = llm.ErrMalformedResponse
|
|
}
|
|
if result.Outcome.Err == nil {
|
|
result.Outcome.Usage = result.Outcome.Response.Usage
|
|
result.Outcome.UsagePresence = result.Outcome.Response.UsagePresence
|
|
}
|
|
return result
|
|
}
|
|
|
|
type attemptResult struct {
|
|
Outcome llm.AttemptOutcome
|
|
Dispatched bool
|
|
CapacityWait time.Duration
|
|
RequestDuration time.Duration
|
|
StartedAt time.Time
|
|
}
|
|
|
|
// recordAttempt runs after the attempt has released its permit, including when
|
|
// the recorder calls application code. Injected clients expose no HTTP facts.
|
|
func recordAttempt(result attemptResult, injected, capture bool, evidence *GenerationEvidence, usage *observation.UsageAccumulator, recorder *observation.Recorder, generation, ordinal uint64) {
|
|
outcome := result.Outcome
|
|
failure := ClassifyFailure(outcome.Err)
|
|
if capture {
|
|
attempt := AttemptEvidence{
|
|
StartedAt: result.StartedAt.UTC(), Dispatched: result.Dispatched,
|
|
CapacityWait: result.CapacityWait, RequestDuration: result.RequestDuration,
|
|
Failure: failure,
|
|
}
|
|
if injected {
|
|
attempt.ResponseReceived = outcome.Response != nil
|
|
attempt.GenerationCompleted = outcome.Err == nil && outcome.Response != nil
|
|
if result.Dispatched {
|
|
attempt.CapacityWait = 0
|
|
}
|
|
} else {
|
|
attempt.ResponseReceived = outcome.Evidence.ResponseReceived
|
|
attempt.HTTP = cloneHTTPEvidence(outcome.Evidence)
|
|
attempt.ResponseDecoded = outcome.ResponseDecoded
|
|
attempt.GenerationCompleted = outcome.GenerationCompleted
|
|
}
|
|
if outcome.StructuredOutputRequested != nil {
|
|
requested := *outcome.StructuredOutputRequested
|
|
attempt.StructuredOutputRequested = &requested
|
|
}
|
|
if result.Dispatched {
|
|
usage.Add(outcome.Usage, outcome.UsagePresence)
|
|
var single observation.UsageAccumulator
|
|
single.Add(outcome.Usage, outcome.UsagePresence)
|
|
attempt.Usage = single.Observed()
|
|
}
|
|
evidence.Attempts = append(evidence.Attempts, attempt)
|
|
}
|
|
if recorder != nil {
|
|
recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: ordinal,
|
|
Dispatched: result.Dispatched, CapacityWait: result.CapacityWait, GenerationDuration: result.RequestDuration,
|
|
Outcome: attemptOutcome(outcome.Err), Failure: failure, StatusCode: outcome.StatusCode,
|
|
Usage: outcome.Usage, UsagePresence: outcome.UsagePresence})
|
|
}
|
|
}
|
|
|
|
func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *llm.PreparedGeneration, captureEvidence bool) attemptResult {
|
|
result := attemptResult{StartedAt: c.now()}
|
|
release, err := c.manager.Acquire(ctx, backendID)
|
|
result.CapacityWait = c.capacityWait(backendID, result.StartedAt)
|
|
if err != nil {
|
|
result.Outcome.Err = err
|
|
return result
|
|
}
|
|
defer release()
|
|
if err := ctx.Err(); err != nil {
|
|
result.Outcome.Err = err
|
|
return result
|
|
}
|
|
dispatchStart := c.now()
|
|
if captureEvidence {
|
|
result.Outcome = prepared.ExecuteWithEvidence(ctx)
|
|
result.Dispatched = result.Outcome.Evidence.DispatchInvoked
|
|
} else {
|
|
result.Outcome = prepared.Execute(ctx)
|
|
result.Dispatched = true
|
|
}
|
|
result.RequestDuration = elapsed(dispatchStart, c.now())
|
|
return result
|
|
}
|
|
|
|
func (c *Coordinator) capacityWait(backendID string, start time.Time) time.Duration {
|
|
if !c.manager.Limited(backendID) {
|
|
return 0
|
|
}
|
|
return elapsed(start, c.now())
|
|
}
|
|
|
|
func elapsed(start, end time.Time) time.Duration {
|
|
if d := end.Sub(start); d > 0 {
|
|
return d
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func attemptOutcome(err error) domain.ExecutionOutcome {
|
|
if err == nil {
|
|
return domain.ExecutionOutcomeSuccess
|
|
}
|
|
return domain.ExecutionOutcomeError
|
|
}
|
|
|
|
// ClassifyFailure maps an execution error to the shared content-free category.
|
|
func ClassifyFailure(err error) domain.ExecutionFailureKind {
|
|
if err == nil {
|
|
return domain.ExecutionFailureNone
|
|
}
|
|
if errors.Is(err, context.Canceled) {
|
|
return domain.ExecutionFailureCanceled
|
|
}
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
return domain.ExecutionFailureDeadline
|
|
}
|
|
if errors.Is(err, llm.ErrUnexpectedStatus) {
|
|
return domain.ExecutionFailureProviderStatus
|
|
}
|
|
if errors.Is(err, llm.ErrTransport) {
|
|
return domain.ExecutionFailureTransport
|
|
}
|
|
if errors.Is(err, llm.ErrMalformedResponse) {
|
|
return domain.ExecutionFailureMalformedResponse
|
|
}
|
|
if errors.Is(err, llm.ErrInvalidRequest) {
|
|
return domain.ExecutionFailureInvalidRequest
|
|
}
|
|
return domain.ExecutionFailureGeneration
|
|
}
|
|
|
|
func eligibleStatus(outcome llm.AttemptOutcome) bool {
|
|
if !errors.Is(outcome.Err, llm.ErrUnexpectedStatus) {
|
|
return false
|
|
}
|
|
switch outcome.StatusCode {
|
|
case 429, 500, 502, 503, 504:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (c *Coordinator) backoff(ordinal uint64) time.Duration {
|
|
base := initialBackoff
|
|
for step := uint64(1); step < ordinal && base < maximumBackoff; step++ {
|
|
base *= 2
|
|
}
|
|
half := base / 2
|
|
return half + c.jitter(half)
|
|
}
|
|
|
|
func retryDelay(local time.Duration, notBefore, now time.Time) time.Duration {
|
|
if providerDelay := notBefore.Sub(now); providerDelay > local {
|
|
return providerDelay
|
|
}
|
|
return local
|
|
}
|
|
|
|
func randomJitter(max time.Duration) time.Duration {
|
|
return time.Duration(rand.Int64N(int64(max) + 1))
|
|
}
|
|
|
|
func waitContext(ctx context.Context, delay time.Duration) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
timer := time.NewTimer(delay)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
return ctx.Err()
|
|
}
|
|
}
|