git.giacomolli.uk - onechat/blob - responses.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "context"
   7         "encoding/json"
   8         "errors"
   9         "fmt"
  10         "io"
  11         "net/http"
  12         "os"
  13         "path/filepath"
  14         "strings"
  15         "sync"
  16         "time"
  17 )
  18
  19 const (
  20         // chatgptURL is the Responses endpoint of the ChatGPT backend.
  21         chatgptURL = "https://chatgpt.com/backend-api/codex/responses"
  22
  23         // apiURL is the Responses endpoint of the OpenAI API.
  24         apiURL = "https://api.openai.com/v1/responses"
  25
  26         // quotaPrefix starts the names of the quota headers of the ChatGPT
  27         // backend, in canonical form.
  28         quotaPrefix = "X-Codex-"
  29
  30         // turnStateHeader is an x-codex-* header that holds an opaque state of
  31         // the backend, not quota.
  32         turnStateHeader = "X-Codex-Turn-State"
  33
  34         // errorBytes is the most bytes of an error response that go into the
  35         // error.
  36         errorBytes = 4096
  37 )
  38
  39 // authorizer adds the credentials of a model call to the headers of its
  40 // request.
  41 type authorizer interface {
  42         Authorize(ctx context.Context, header http.Header) error
  43 }
  44
  45 // apiKey authorizes calls to the OpenAI API with a key.
  46 type apiKey string
  47
  48 // Client sends model calls to a Responses endpoint for one agent. The ChatGPT
  49 // backend and the OpenAI API take the same request body and differ only in the
  50 // URL and the credentials. Every call sends the fixed cache key of the agent,
  51 // so the turns and the compactions of the agent share one prompt cache. The
  52 // ChatGPT backend caches only with the key in the session_id header, so a call
  53 // to it sends the key there and in the body. Every call is recorded in the
  54 // usage file.
  55 //
  56 // The request never holds max_output_tokens or metadata, because the ChatGPT
  57 // backend refuses them. It sets store to false, so the client sends the whole
  58 // input on each call, and it asks for the encrypted reasoning, so a later call
  59 // can send the reasoning items back.
  60 type Client struct {
  61         url      string
  62         auth     authorizer
  63         cacheKey string
  64         usage    *usageLog
  65         http     *http.Client
  66 }
  67
  68 // Request is one model call. Input holds the input items in order, such as
  69 // userMessage values or output items of an earlier response.
  70 type Request struct {
  71         Model        string
  72         Effort       string
  73         Instructions string
  74         Tools        []Tool
  75         Input        []any
  76 }
  77
  78 // Tool is the definition of a function tool.
  79 type Tool struct {
  80         Type        string `json:"type"`
  81         Name        string `json:"name"`
  82         Description string `json:"description"`
  83         Parameters  any    `json:"parameters"`
  84         Strict      bool   `json:"strict"`
  85 }
  86
  87 // Response is the result of a model call. Output holds the output items as the
  88 // endpoint sent them. Text is the text of the message items.
  89 type Response struct {
  90         Output []json.RawMessage
  91         Text   string
  92         Usage  Usage
  93 }
  94
  95 // Usage gives the token counts of a call. Cached is the part of the input
  96 // tokens that came from the prompt cache.
  97 type Usage struct {
  98         Input  int
  99         Cached int
 100         Output int
 101 }
 102
 103 // requestBody is the JSON body of a call.
 104 type requestBody struct {
 105         Model          string    `json:"model"`
 106         Instructions   string    `json:"instructions"`
 107         Input          []any     `json:"input"`
 108         Tools          []Tool    `json:"tools"`
 109         Reasoning      reasoning `json:"reasoning"`
 110         Store          bool      `json:"store"`
 111         Stream         bool      `json:"stream"`
 112         Include        []string  `json:"include"`
 113         PromptCacheKey string    `json:"prompt_cache_key"`
 114 }
 115
 116 type reasoning struct {
 117         Effort string `json:"effort"`
 118 }
 119
 120 // streamEvent holds the fields of the stream events that the client reads.
 121 type streamEvent struct {
 122         Type     string          `json:"type"`
 123         Item     json.RawMessage `json:"item"`
 124         Response *eventResponse  `json:"response"`
 125         Code     string          `json:"code"`
 126         Message  string          `json:"message"`
 127 }
 128
 129 type eventResponse struct {
 130         Output            []json.RawMessage `json:"output"`
 131         Usage             *wireUsage        `json:"usage"`
 132         Error             *wireError        `json:"error"`
 133         IncompleteDetails *struct {
 134                 Reason string `json:"reason"`
 135         } `json:"incomplete_details"`
 136 }
 137
 138 type wireUsage struct {
 139         InputTokens        int `json:"input_tokens"`
 140         OutputTokens       int `json:"output_tokens"`
 141         InputTokensDetails struct {
 142                 CachedTokens int `json:"cached_tokens"`
 143         } `json:"input_tokens_details"`
 144 }
 145
 146 type wireError struct {
 147         Code    string `json:"code"`
 148         Message string `json:"message"`
 149 }
 150
 151 // outputItem holds the fields of an output item that give its text.
 152 type outputItem struct {
 153         Type    string `json:"type"`
 154         Content []struct {
 155                 Type string `json:"type"`
 156                 Text string `json:"text"`
 157         } `json:"content"`
 158 }
 159
 160 // modelEndpoint returns the URL and the authorizer of the endpoint in config.
 161 // The ChatGPT backend uses the token file in dir. The OpenAI API uses the key in
 162 // OPENAI_API_KEY, and there is no fallback from one to the other. All clients
 163 // share the authorizer, so one refresh runs at a time.
 164 func modelEndpoint(config Config, dir string) (string, authorizer, error) {
 165         if config.Endpoint == endpointAPI {
 166                 key := os.Getenv("OPENAI_API_KEY")
 167                 if key == "" {
 168                         return "", nil, errors.New("endpoint api needs OPENAI_API_KEY")
 169                 }
 170
 171                 return apiURL, apiKey(key), nil
 172         }
 173
 174         return chatgptURL, newChatGPTAuth(filepath.Join(dir, tokenFileName), authIssuer), nil
 175 }
 176
 177 // newClient returns a client for the Responses endpoint at url. It records
 178 // each call in the usage file at usagePath.
 179 func newClient(url string, auth authorizer, cacheKey, usagePath string) *Client {
 180         return &Client{
 181                 url:      url,
 182                 auth:     auth,
 183                 cacheKey: cacheKey,
 184                 usage: &usageLog{
 185                         path: usagePath,
 186                 },
 187                 http: &http.Client{},
 188         }
 189 }
 190
 191 // Create sends the call and reads its event stream to the end.
 192 func (c *Client) Create(ctx context.Context, request Request) (Response, error) {
 193         body, err := json.Marshal(requestBody{
 194                 Model:        request.Model,
 195                 Instructions: request.Instructions,
 196                 Input:        request.Input,
 197                 Tools:        request.Tools,
 198                 Reasoning: reasoning{
 199                         Effort: request.Effort,
 200                 },
 201                 Store:          false,
 202                 Stream:         true,
 203                 Include:        []string{"reasoning.encrypted_content"},
 204                 PromptCacheKey: c.cacheKey,
 205         })
 206         if err != nil {
 207                 return Response{}, fmt.Errorf("encode request: %w", err)
 208         }
 209
 210         httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url, bytes.NewReader(body))
 211         if err != nil {
 212                 return Response{}, fmt.Errorf("create request: %w", err)
 213         }
 214
 215         httpRequest.Header.Set("Content-Type", "application/json")
 216         httpRequest.Header.Set("Accept", "text/event-stream")
 217         err = c.auth.Authorize(ctx, httpRequest.Header)
 218         if err != nil {
 219                 return Response{}, fmt.Errorf("authorize model call: %w", err)
 220         }
 221
 222         _, chatgpt := c.auth.(*chatgptAuth)
 223         if chatgpt {
 224                 httpRequest.Header.Set("session_id", c.cacheKey)
 225         }
 226
 227         record := usageRecord{
 228                 Date:  time.Now().Format(time.RFC3339Nano),
 229                 Model: request.Model,
 230         }
 231
 232         response, err := c.send(httpRequest, &record)
 233         if err != nil {
 234                 record.Error = err.Error()
 235         }
 236
 237         record.InputTokens = response.Usage.Input
 238         record.CachedTokens = response.Usage.Cached
 239         record.OutputTokens = response.Usage.Output
 240
 241         return response, errors.Join(err, c.usage.Write(record))
 242 }
 243
 244 // send makes the HTTP call and reads its stream. It puts the status and the
 245 // quota headers into record.
 246 func (c *Client) send(httpRequest *http.Request, record *usageRecord) (response Response, err error) {
 247         httpResponse, err := c.http.Do(httpRequest)
 248         if err != nil {
 249                 return Response{}, fmt.Errorf("send model call: %w", err)
 250         }
 251
 252         defer func() {
 253                 err = errors.Join(err, httpResponse.Body.Close())
 254         }()
 255
 256         record.Status = httpResponse.StatusCode
 257         record.Quota = quotaHeaders(httpResponse.Header)
 258         if httpResponse.StatusCode != http.StatusOK {
 259                 text, readErr := io.ReadAll(io.LimitReader(httpResponse.Body, errorBytes))
 260                 err = fmt.Errorf("model call: HTTP %d: %s", httpResponse.StatusCode, strings.TrimSpace(string(text)))
 261
 262                 return Response{}, errors.Join(err, readErr)
 263         }
 264
 265         return readStream(httpResponse.Body)
 266 }
 267
 268 // usageLog appends one JSON line for each model call to usage.jsonl. Calls can
 269 // run at once, so a mutex serializes the writes.
 270 type usageLog struct {
 271         mutex sync.Mutex
 272         path  string
 273 }
 274
 275 // usageRecord is one line of usage.jsonl. Status is the HTTP status, or 0 when
 276 // no response came. Quota holds the x-codex-* headers of the response, without
 277 // the turn state.
 278 type usageRecord struct {
 279         Date         string            `json:"date"`
 280         Model        string            `json:"model"`
 281         Status       int               `json:"status"`
 282         InputTokens  int               `json:"input_tokens"`
 283         CachedTokens int               `json:"cached_tokens"`
 284         OutputTokens int               `json:"output_tokens"`
 285         Quota        map[string]string `json:"quota,omitempty"`
 286         Error        string            `json:"error,omitempty"`
 287 }
 288
 289 // Write appends record to the file with one write.
 290 func (u *usageLog) Write(record usageRecord) (err error) {
 291         line, err := json.Marshal(record)
 292         if err != nil {
 293                 return fmt.Errorf("encode usage: %w", err)
 294         }
 295
 296         u.mutex.Lock()
 297         defer u.mutex.Unlock()
 298
 299         file, err := os.OpenFile(u.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
 300         if err != nil {
 301                 return fmt.Errorf("open usage file: %w", err)
 302         }
 303
 304         defer func() {
 305                 err = errors.Join(err, file.Close())
 306         }()
 307
 308         _, err = file.Write(append(line, '\n'))
 309         if err != nil {
 310                 return fmt.Errorf("write usage: %w", err)
 311         }
 312
 313         return nil
 314 }
 315
 316 // Authorize sets the key as the bearer token.
 317 func (k apiKey) Authorize(ctx context.Context, header http.Header) error {
 318         header.Set("Authorization", "Bearer "+string(k))
 319         return ctx.Err()
 320 }
 321
 322 // userMessage returns a user input item with one text part for each text.
 323 func userMessage(texts ...string) any {
 324         type part struct {
 325                 Type string `json:"type"`
 326                 Text string `json:"text"`
 327         }
 328
 329         content := make([]part, len(texts))
 330         for n, text := range texts {
 331                 content[n] = part{
 332                         Type: "input_text",
 333                         Text: text,
 334                 }
 335         }
 336
 337         return struct {
 338                 Role    string `json:"role"`
 339                 Content []part `json:"content"`
 340         }{
 341                 Role:    "user",
 342                 Content: content,
 343         }
 344 }
 345
 346 // quotaHeaders returns the x-codex-* headers, with lowercase names, without the
 347 // turn state.
 348 func quotaHeaders(header http.Header) map[string]string {
 349         var quota map[string]string
 350         for name, values := range header {
 351                 if !strings.HasPrefix(name, quotaPrefix) || name == turnStateHeader || len(values) == 0 {
 352                         continue
 353                 }
 354
 355                 if quota == nil {
 356                         quota = make(map[string]string)
 357                 }
 358
 359                 quota[strings.ToLower(name)] = values[0]
 360         }
 361
 362         return quota
 363 }
 364
 365 // readStream reads the server-sent events of a call until the response
 366 // completes. The output items come from the output_item.done events, or from
 367 // the completed response when the stream sent none.
 368 func readStream(stream io.Reader) (Response, error) {
 369         var response Response
 370         reader := bufio.NewReader(stream)
 371         var data bytes.Buffer
 372         for {
 373                 line, err := reader.ReadBytes('\n')
 374                 if err != nil && !errors.Is(err, io.EOF) {
 375                         return response, fmt.Errorf("read event stream: %w", err)
 376                 }
 377
 378                 line = bytes.TrimRight(line, "\r\n")
 379                 if len(line) > 0 {
 380                         value, found := bytes.CutPrefix(line, []byte("data:"))
 381                         if found {
 382                                 data.Write(bytes.TrimPrefix(value, []byte(" ")))
 383                                 data.WriteByte('\n')
 384                         }
 385                 } else if data.Len() > 0 {
 386                         done, eventErr := readEvent(&response, data.Bytes())
 387                         data.Reset()
 388                         if eventErr != nil || done {
 389                                 return response, eventErr
 390                         }
 391                 }
 392
 393                 if errors.Is(err, io.EOF) {
 394                         return response, errors.New("event stream ended before the response completed")
 395                 }
 396         }
 397 }
 398
 399 // readEvent adds one stream event to response, and tells if it ends the
 400 // stream.
 401 func readEvent(response *Response, data []byte) (bool, error) {
 402         var event streamEvent
 403         err := json.Unmarshal(data, &event)
 404         if err != nil {
 405                 return true, fmt.Errorf("decode stream event: %w", err)
 406         }
 407
 408         switch event.Type {
 409         case "response.output_item.done":
 410                 response.Output = append(response.Output, event.Item)
 411         case "response.completed":
 412                 if event.Response == nil {
 413                         return true, errors.New("completed event without a response")
 414                 }
 415
 416                 if len(response.Output) == 0 {
 417                         response.Output = event.Response.Output
 418                 }
 419
 420                 if event.Response.Usage != nil {
 421                         response.Usage = Usage{
 422                                 Input:  event.Response.Usage.InputTokens,
 423                                 Cached: event.Response.Usage.InputTokensDetails.CachedTokens,
 424                                 Output: event.Response.Usage.OutputTokens,
 425                         }
 426                 }
 427
 428                 response.Text, err = outputText(response.Output)
 429
 430                 return true, err
 431         case "response.incomplete":
 432                 reason := "unknown"
 433                 if event.Response != nil && event.Response.IncompleteDetails != nil {
 434                         reason = event.Response.IncompleteDetails.Reason
 435                 }
 436
 437                 return true, fmt.Errorf("response incomplete: %s", reason)
 438         case "response.failed":
 439                 message := "no details"
 440                 if event.Response != nil && event.Response.Error != nil {
 441                         message = event.Response.Error.Code + ": " + event.Response.Error.Message
 442                 }
 443
 444                 return true, fmt.Errorf("response failed: %s", message)
 445         case "error":
 446                 return true, fmt.Errorf("stream error: %s: %s", event.Code, event.Message)
 447         }
 448
 449         return false, nil
 450 }
 451
 452 // outputText returns the text of the message items of output.
 453 func outputText(output []json.RawMessage) (string, error) {
 454         var text strings.Builder
 455         for _, raw := range output {
 456                 var item outputItem
 457                 err := json.Unmarshal(raw, &item)
 458                 if err != nil {
 459                         return "", fmt.Errorf("decode output item: %w", err)
 460                 }
 461
 462                 if item.Type != "message" {
 463                         continue
 464                 }
 465
 466                 for _, part := range item.Content {
 467                         if part.Type == "output_text" {
 468                                 text.WriteString(part.Text)
 469                         }
 470                 }
 471         }
 472
 473         return text.String(), nil
 474 }