"path/filepath"
"strings"
"sync"
+ "sync/atomic"
"time"
)
errorBytes = 4096
)
+// errCallLimit tells that a client sent its limit of calls.
+var errCallLimit = errors.New("call limit reached")
+
// authorizer adds the credentials of a model call to the headers of its
// request.
type authorizer interface {
cacheKey string
usage *usageLog
http *http.Client
+ limit int64
+ calls atomic.Int64
}
// Request is one model call. Input holds the input items in order, such as
// userMessage values or output items of an earlier response. Node names the
-// node that a compaction builds, or is nil. It goes only to the usage file,
-// never to the endpoint.
+// node that a compaction builds, or is nil. Turn names the request of a turn:
+// first, next, or warmup. Node and Turn go only to the usage file, never to the
+// endpoint. A call with StopAtOutput ends at the first output item, after the
+// endpoint has read the whole input, and gives no output and no usage.
type Request struct {
Model string
Effort string
Tools []Tool
Input []any
Node *nodeKey
+ Turn string
+ StopAtOutput bool
}
// Tool is the definition of a function tool.
}
}
-// Create sends the call and reads its event stream to the end.
+// LimitCalls makes the client refuse each call after the first limit calls.
+func (c *Client) LimitCalls(limit int64) {
+ c.limit = limit
+}
+
+// LimitReached tells if the client refuses further calls.
+func (c *Client) LimitReached() bool {
+ return c.limit > 0 && c.calls.Load() >= c.limit
+}
+
+// Create sends the call and reads its event stream to the end. Over the call
+// limit it sends nothing and returns errCallLimit.
func (c *Client) Create(ctx context.Context, request Request) (Response, error) {
+ if c.calls.Add(1) > c.limit && c.limit > 0 {
+ return Response{}, errCallLimit
+ }
+
body, err := json.Marshal(requestBody{
Model: request.Model,
Instructions: request.Instructions,
record.Node = &[2]int{request.Node.level, request.Node.index}
}
- response, err := c.send(httpRequest, &record)
+ record.Turn = request.Turn
+ response, err := c.send(httpRequest, &record, request.StopAtOutput)
if err != nil {
record.Error = err.Error()
}
// send makes the HTTP call and reads its stream. It puts the status and the
// quota headers into record.
-func (c *Client) send(httpRequest *http.Request, record *usageRecord) (response Response, err error) {
+func (c *Client) send(httpRequest *http.Request, record *usageRecord, stopAtOutput bool) (response Response, err error) {
httpResponse, err := c.http.Do(httpRequest)
if err != nil {
return Response{}, fmt.Errorf("send model call: %w", err)
return Response{}, errors.Join(err, readErr)
}
- return readStream(httpResponse.Body)
+ return readStream(httpResponse.Body, stopAtOutput)
}
// usageLog appends one JSON line for each model call to usage.jsonl. Calls can
// usageRecord is one line of usage.jsonl. Date is the start of the call, and
// Seconds its length. Node is the node [l, i] of a compaction. Status is the
// HTTP status, or 0 when no response came. TextBytes is the size of the text of
-// the reply. Quota holds the x-codex-* headers of the response, without the
+// the reply. Turn names the request of a turn. Quota holds the x-codex-* headers of the response, without the
// turn state.
type usageRecord struct {
Date string `json:"date"`
Seconds float64 `json:"seconds"`
Model string `json:"model"`
Node *[2]int `json:"node,omitempty"`
+ Turn string `json:"turn,omitempty"`
Status int `json:"status"`
InputTokens int `json:"input_tokens"`
CachedTokens int `json:"cached_tokens"`
// readStream reads the server-sent events of a call until the response
// completes. The output items come from the output_item.done events, or from
-// the completed response when the stream sent none.
-func readStream(stream io.Reader) (Response, error) {
+// the completed response when the stream sent none. With stopAtOutput it ends
+// at the first output item.
+func readStream(stream io.Reader, stopAtOutput bool) (Response, error) {
var response Response
reader := bufio.NewReader(stream)
var data bytes.Buffer
data.WriteByte('\n')
}
} else if data.Len() > 0 {
- done, eventErr := readEvent(&response, data.Bytes())
+ done, eventErr := readEvent(&response, data.Bytes(), stopAtOutput)
data.Reset()
if eventErr != nil || done {
return response, eventErr
// readEvent adds one stream event to response, and tells if it ends the
// stream.
-func readEvent(response *Response, data []byte) (bool, error) {
+func readEvent(response *Response, data []byte, stopAtOutput bool) (bool, error) {
var event streamEvent
err := json.Unmarshal(data, &event)
if err != nil {
}
switch event.Type {
+ case "response.output_item.added":
+ return stopAtOutput, nil
case "response.output_item.done":
response.Output = append(response.Output, event.Item)
case "response.completed":
projects / onechat / blobdiff