| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | |
| 7 | "reasonix/internal/event" |
| 8 | "reasonix/internal/provider" |
| 9 | ) |
| 10 | |
| 11 | // runSamplingAttempt records the exact HTTP-attempt delta for one replay of a |
| 12 | // frozen provider request. Providers outside SendWithRetry retain their own |
| 13 | // RequestCount, while pre-wire failures do not invent a billable request. |
| 14 | func (a *Agent) runSamplingAttempt(ctx context.Context, turn int, sink event.Sink, frozen *samplingRequest, attemptID string) streamedTurn { |
| 15 | before := provider.RequestAttemptCount(ctx) |
| 16 | result := a.streamWithFrozen(ctx, turn, event.WithMessageIdentity(sink, attemptID, attemptID), frozen, attemptID) |
| 17 | result.messageID = attemptID |
| 18 | if result.err == nil && isEmptyStreamResult(result.text, result.reasoning, result.calls, result.responsesItems, result.serverSearch) { |
| 19 | result.err = fmt.Errorf("%w: model returned a completed response with no content", provider.ErrEmptyResponse) |
| 20 | } |
| 21 | delta := max(provider.RequestAttemptCount(ctx)-before, 0) |
| 22 | result.usage = estimateFailedAttemptUsage(result.usage, *frozen, result, delta) |
| 23 | if result.usage != nil { |
| 24 | if delta > 0 { |
| 25 | result.usage.RequestCount = delta |
| 26 | } |
| 27 | } else if delta > 0 { |
| 28 | result.usage = &provider.Usage{RequestCount: delta, Unknown: true} |
| 29 | } |
| 30 | return result |
| 31 | } |
| 32 | |
| 33 | func (a *Agent) samplingAttemptSinks() (*deferredStreamSink, event.Sink) { |
| 34 | // Buffer when missing reasoning can reject or replace the attempt. Protocols |
| 35 | // that adopt an empty fallback without retry must keep streaming live because |
| 36 | // their first response always wins. |
| 37 | replaySensitive := provider.RequiresToolCallReasoning(a.svc.prov) || provider.RequiresReasoningRoundTrip(a.svc.prov) |
| 38 | if replaySensitive && !provider.AllowsEmptyReasoningFallback(a.svc.prov) { |
| 39 | streamSink := newReasoningAwareStreamSink(a.svc.sink) |
| 40 | return streamSink, streamSink |
| 41 | } |
| 42 | return nil, a.svc.sink |
| 43 | } |
| 44 | |
| 45 | // recordSamplingAttempt keeps the latest single-request shape separate from |
| 46 | // the multi-attempt billable aggregate used for final accounting. |
| 47 | func (a *Agent) recordSamplingAttempt(billable *provider.Usage, result streamedTurn) (*provider.Usage, streamedTurn) { |
| 48 | billable = mergeSamplingUsage(billable, result.usage) |
| 49 | a.storeLatestRequestUsage(result.usage) |
| 50 | latest := result |
| 51 | latest.usage = finalizeSamplingUsage(billable, result.usage) |
| 52 | return billable, latest |
| 53 | } |
| 54 |