auth := newTestAuth(t, server)
writeTestTokens(t, auth, server.token(-time.Hour), "refresh-0")
- responses := newFakeResponses(t)
- client := newClient(responses.URL, auth, "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
- _, err := client.Create(t.Context(), testRequest())
+ responses := newModelFake(t)
+ client := newModelClient(responses.URL, auth, "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
+ _, err := client.Create(t.Context(), modelTestRequest())
require.NoError(t, err)
require.Equal(t, 1, server.refreshes)
dir := t.TempDir()
config := defaultConfig()
- t.Setenv("OPENAI_API_KEY", testKey)
+ t.Setenv("OPENAI_API_KEY", modelTestKey)
address, auth, err := modelEndpoint(config, dir)
require.NoError(t, err)
- require.Equal(t, chatgptURL, address)
+ require.Equal(t, modelChatGPTURL, address)
require.IsType(t, &chatgptAuth{}, auth)
require.Equal(t, filepath.Join(dir, tokenFileName), auth.(*chatgptAuth).path)
config.Endpoint = endpointAPI
address, auth, err = modelEndpoint(config, dir)
require.NoError(t, err)
- require.Equal(t, apiURL, address)
- require.Equal(t, apiKey(testKey), auth)
+ require.Equal(t, modelAPIURL, address)
+ require.Equal(t, ModelAPIKey(modelTestKey), auth)
t.Setenv("OPENAI_API_KEY", "")
_, _, err = modelEndpoint(config, dir)
// because the view measures the real sizes. A call that fails before any line
// fails the build, and the core tries it again at the next message.
type Compactor struct {
- client *Client
+ client *ModelClient
model ModelConfig
instructions func() (string, error)
- tools []Tool
+ tools []ModelTool
}
// newCompactor returns the compactor of an agent, which sends its calls through
// client with model. instructions gives the system prompt of the agent at each
// build, so a build sends the same prompt as the turns of that time.
-func newCompactor(client *Client, instructions func() (string, error), model ModelConfig) *Compactor {
+func newCompactor(client *ModelClient, instructions func() (string, error), model ModelConfig) *Compactor {
return &Compactor{
client: client,
model: model,
parts = append(parts, note)
}
- var response Response
- response, err = c.client.Create(ctx, Request{
+ var response ModelResponse
+ response, err = c.client.Create(ctx, ModelRequest{
Model: c.model.Model,
Effort: c.model.Effort,
Instructions: instructions,
Tools: c.tools,
- Input: []any{userMessage(parts...)},
+ Input: []any{modelUserMessage(parts...)},
Node: &nodeKey{task.Level, task.Index},
})
if err != nil {
)
func TestCompactorSendsThePromptToolsAndTask(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
compactor := newTestCompactor(t, server)
line, err := compactor.Summarize(t.Context(), Task{
func TestCompactorRetriesALongLineWithTheNewestNote(t *testing.T) {
long := []string{strings.Repeat("a", 600), strings.Repeat("b", 550), strings.Repeat("c", 400)}
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem(long[call.number])}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(long[call.number])}, 100, 0, 5))
}
compactor := newTestCompactor(t, server)
require.Equal(t, long[2], line)
require.Equal(t, 3, server.Calls())
- records := readUsage(t, compactor.client.usage.path)
+ records := modelReadUsage(t, compactor.client.usage.path)
require.Len(t, records, 3)
for n, record := range records {
require.Equal(t, &[2]int{0, 0}, record.Node)
func TestCompactorKeepsTheShortestOfFiveLines(t *testing.T) {
sizes := []int{700, 530, 600, 520, 560, 513}
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem(strings.Repeat("z", sizes[call.number]))}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(strings.Repeat("z", sizes[call.number]))}, 100, 0, 5))
}
line, err := newTestCompactor(t, server).Summarize(t.Context(), leafTask())
func TestCompactorFailsWithoutALine(t *testing.T) {
testCases := []struct {
name string
- reply func(http.ResponseWriter, fakeCall)
+ reply func(http.ResponseWriter, ModelFakeCall)
want string
}{
{
name: "failed call",
- reply: func(w http.ResponseWriter, call fakeCall) {
+ reply: func(w http.ResponseWriter, call ModelFakeCall) {
w.WriteHeader(http.StatusInternalServerError)
_, _ = io.WriteString(w, "down")
},
},
{
name: "tool call and no text",
- reply: func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{map[string]any{"type": "function_call", "name": "zoom", "arguments": "{}"}}, 100, 0, 5))
+ reply: func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{map[string]any{"type": "function_call", "name": "zoom", "arguments": "{}"}}, 100, 0, 5))
},
want: "no line",
},
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
server.reply = tc.reply
_, err := newTestCompactor(t, server).Summarize(t.Context(), leafTask())
}
func TestCompactorBuildsNodesInTheCore(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
w.WriteHeader(http.StatusServiceUnavailable)
return
}
- writeEvents(w, completedEvent([]any{messageItem("summary of the long output")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("summary of the long output")}, 100, 0, 5))
}
memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newTestCompactor(t, server))
require.Equal(t, 2, server.Calls())
}
-func newTestCompactor(t *testing.T, server *fakeResponses) *Compactor {
+func newTestCompactor(t *testing.T, server *ModelFake) *Compactor {
t.Helper()
- client := newClient(server.URL, apiKey(testKey), "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
+ client := newModelClient(server.URL, ModelAPIKey(modelTestKey), "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
return newCompactor(client, adaPrompt, ModelConfig{
Model: "gpt-test",
}
// inputParts returns the texts of the one user message of call.
-func inputParts(t *testing.T, call fakeCall) []string {
+func inputParts(t *testing.T, call ModelFakeCall) []string {
t.Helper()
var body struct {
return err
}
- client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
+ client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
if err != nil {
return err
)
func TestClientsSeeTheTurnAsItHappens(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
- writeEvents(w, completedEvent([]any{functionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`)}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`)}, 100, 0, 5))
return
}
- writeEvents(w, completedEvent([]any{messageItem("It said hi.")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 100, 0, 5))
}
socket := startTestDaemon(t, server)
}
func TestStopInterruptsTheTurn(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
arrived := make(chan struct{})
release := make(chan struct{})
t.Cleanup(func() {
close(release)
})
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
close(arrived)
<-release
}
// startTestDaemon serves a new agent on a socket in a temporary directory and
// returns the socket path. The commands of the agent run in a temporary
// directory.
-func startTestDaemon(t *testing.T, server *fakeResponses) string {
+func startTestDaemon(t *testing.T, server *ModelFake) string {
t.Helper()
return startTestDaemonWithBot(t, server, nil)
}
// startTestDaemonWithBot is startTestDaemon with bot, which can be nil.
-func startTestDaemonWithBot(t *testing.T, server *fakeResponses, bot *TelegramBot) string {
+func startTestDaemonWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string {
t.Helper()
dir := t.TempDir()
listener, err := listenSocket(socket)
require.NoError(t, err)
- client := newClient(server.URL, apiKey(testKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
+ client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
go func() {
)
const (
- // chatgptURL is the Responses endpoint of the ChatGPT backend.
- chatgptURL = "https://chatgpt.com/backend-api/codex/responses"
+ // modelChatGPTURL is the Responses endpoint of the ChatGPT backend.
+ modelChatGPTURL = "https://chatgpt.com/backend-api/codex/responses"
- // apiURL is the Responses endpoint of the OpenAI API.
- apiURL = "https://api.openai.com/v1/responses"
+ // modelAPIURL is the Responses endpoint of the OpenAI API.
+ modelAPIURL = "https://api.openai.com/v1/responses"
- // quotaPrefix starts the names of the quota headers of the ChatGPT
+ // modelQuotaPrefix starts the names of the quota headers of the ChatGPT
// backend, in canonical form.
- quotaPrefix = "X-Codex-"
+ modelQuotaPrefix = "X-Codex-"
- // turnStateHeader is an x-codex-* header that holds an opaque state of
- // the backend, not quota.
- turnStateHeader = "X-Codex-Turn-State"
+ // modelTurnStateHeader is an x-codex-* header that holds an opaque
+ // state of the backend, not quota.
+ modelTurnStateHeader = "X-Codex-Turn-State"
- // errorBytes is the most bytes of an error response that go into the
- // error.
- errorBytes = 4096
+ // modelErrorBytes is the most bytes of an error response that go into
+ // the error.
+ modelErrorBytes = 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 {
- Authorize(ctx context.Context, header http.Header) error
-}
-
-// apiKey authorizes calls to the OpenAI API with a key.
-type apiKey string
-
-// Client sends model calls to a Responses endpoint for one agent. The ChatGPT
-// backend and the OpenAI API take the same request body and differ only in the
-// URL and the credentials. Every call sends the fixed cache key of the agent,
-// so the turns and the compactions of the agent share one prompt cache. The
-// ChatGPT backend caches only with the key in the session_id header, so a call
-// to it sends the key there and in the body. Every call is recorded in the
-// usage file.
-//
-// The request never holds max_output_tokens or metadata, because the ChatGPT
-// backend refuses them. It sets store to false, so the client sends the whole
-// input on each call, and it asks for the encrypted reasoning, so a later call
-// can send the reasoning items back.
-type Client struct {
- url string
- auth authorizer
- 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. Turn names the request of a turn:
-// first or next. Node and Turn go only to the usage file, never to the
-// endpoint.
-type Request struct {
- Model string
- Effort string
- Instructions string
- Tools []Tool
- Input []any
- Node *nodeKey
- Turn string
-}
-
-// Tool is the definition of a function tool.
-type Tool struct {
- Type string `json:"type"`
- Name string `json:"name"`
- Description string `json:"description"`
- Parameters any `json:"parameters"`
- Strict bool `json:"strict"`
-}
-
-// Response is the result of a model call. Output holds the output items as the
-// endpoint sent them. Text is the text of the message items.
-type Response struct {
- Output []json.RawMessage
- Text string
- Usage Usage
-}
-
-// Usage gives the token counts of a call. Cached is the part of the input
-// tokens that came from the prompt cache.
-type Usage struct {
- Input int
- Cached int
- Output int
-}
-
-// requestBody is the JSON body of a call.
-type requestBody struct {
- Model string `json:"model"`
- Instructions string `json:"instructions"`
- Input []any `json:"input"`
- Tools []Tool `json:"tools"`
- Reasoning reasoning `json:"reasoning"`
- Store bool `json:"store"`
- Stream bool `json:"stream"`
- Include []string `json:"include"`
- PromptCacheKey string `json:"prompt_cache_key"`
-}
-
-type reasoning struct {
- Effort string `json:"effort"`
-}
-
-// streamEvent holds the fields of the stream events that the client reads.
-type streamEvent struct {
- Type string `json:"type"`
- Item json.RawMessage `json:"item"`
- Response *eventResponse `json:"response"`
- Code string `json:"code"`
- Message string `json:"message"`
-}
-
-type eventResponse struct {
- Output []json.RawMessage `json:"output"`
- Usage *wireUsage `json:"usage"`
- Error *wireError `json:"error"`
- IncompleteDetails *struct {
- Reason string `json:"reason"`
- } `json:"incomplete_details"`
-}
-
-type wireUsage struct {
- InputTokens int `json:"input_tokens"`
- OutputTokens int `json:"output_tokens"`
- InputTokensDetails struct {
- CachedTokens int `json:"cached_tokens"`
- } `json:"input_tokens_details"`
-}
-
-type wireError struct {
- Code string `json:"code"`
- Message string `json:"message"`
-}
-
-// outputItem holds the fields of an output item that give its text.
-type outputItem struct {
- Type string `json:"type"`
- Content []struct {
- Type string `json:"type"`
- Text string `json:"text"`
- } `json:"content"`
-}
+// errModelCallLimit tells that a client sent its limit of calls.
+var errModelCallLimit = errors.New("call limit reached")
// modelEndpoint returns the URL and the authorizer of the endpoint in config.
// The ChatGPT backend uses the token file in dir. The OpenAI API uses the key in
// OPENAI_API_KEY, and there is no fallback from one to the other. All clients
// share the authorizer, so one refresh runs at a time.
-func modelEndpoint(config Config, dir string) (string, authorizer, error) {
+func modelEndpoint(config Config, dir string) (string, ModelAuthorizer, error) {
if config.Endpoint == endpointAPI {
key := os.Getenv("OPENAI_API_KEY")
if key == "" {
return "", nil, errors.New("endpoint api needs OPENAI_API_KEY")
}
- return apiURL, apiKey(key), nil
+ return modelAPIURL, ModelAPIKey(key), nil
}
- return chatgptURL, newChatGPTAuth(filepath.Join(dir, tokenFileName), authIssuer), nil
+ return modelChatGPTURL, newChatGPTAuth(filepath.Join(dir, tokenFileName), authIssuer), nil
}
-// newClient returns a client for the Responses endpoint at url. It records
+// newModelClient returns a client for the Responses endpoint at url. It records
// each call in the usage file at usagePath.
-func newClient(url string, auth authorizer, cacheKey, usagePath string) *Client {
- return &Client{
+func newModelClient(url string, auth ModelAuthorizer, cacheKey, usagePath string) *ModelClient {
+ return &ModelClient{
url: url,
auth: auth,
cacheKey: cacheKey,
- usage: &usageLog{
+ usage: &ModelUsageLog{
path: usagePath,
},
http: &http.Client{},
}
}
+// modelUserMessage returns a user input item with one text part for each text.
+func modelUserMessage(texts ...string) any {
+ type part struct {
+ Type string `json:"type"`
+ Text string `json:"text"`
+ }
+
+ content := make([]part, len(texts))
+ for n, text := range texts {
+ content[n] = part{
+ Type: "input_text",
+ Text: text,
+ }
+ }
+
+ return struct {
+ Role string `json:"role"`
+ Content []part `json:"content"`
+ }{
+ Role: "user",
+ Content: content,
+ }
+}
+
+// modelOutputText returns the text of the message items of output.
+func modelOutputText(output []json.RawMessage) (string, error) {
+ var text strings.Builder
+ for _, raw := range output {
+ var item ModelOutputItem
+ err := json.Unmarshal(raw, &item)
+ if err != nil {
+ return "", fmt.Errorf("decode output item: %w", err)
+ }
+
+ if item.Type != "message" {
+ continue
+ }
+
+ for _, part := range item.Content {
+ if part.Type == "output_text" {
+ text.WriteString(part.Text)
+ }
+ }
+ }
+
+ return text.String(), nil
+}
+
+// ModelAuthorizer adds the credentials of a model call to the headers of its
+// request.
+type ModelAuthorizer interface {
+ Authorize(ctx context.Context, header http.Header) error
+}
+
+// ModelAPIKey authorizes calls to the OpenAI API with a key.
+type ModelAPIKey string
+
+// Authorize sets the key as the bearer token.
+func (k ModelAPIKey) Authorize(ctx context.Context, header http.Header) error {
+ header.Set("Authorization", "Bearer "+string(k))
+ return ctx.Err()
+}
+
+// ModelClient sends model calls to a Responses endpoint for one agent. The
+// ChatGPT backend and the OpenAI API take the same request body and differ only
+// in the URL and the credentials. Every call sends the fixed cache key of the
+// agent, so the turns and the compactions of the agent share one prompt cache.
+// The ChatGPT backend caches only with the key in the session_id header, so a
+// call to it sends the key there and in the body. Every call is recorded in the
+// usage file.
+//
+// The request never holds max_output_tokens or metadata, because the ChatGPT
+// backend refuses them. It sets store to false, so the client sends the whole
+// input on each call, and it asks for the encrypted reasoning, so a later call
+// can send the reasoning items back.
+type ModelClient struct {
+ url string
+ auth ModelAuthorizer
+ cacheKey string
+ usage *ModelUsageLog
+ http *http.Client
+ limit int64
+ calls atomic.Int64
+}
+
// LimitCalls makes the client refuse each call after the first limit calls.
-func (c *Client) LimitCalls(limit int64) {
+func (c *ModelClient) LimitCalls(limit int64) {
c.limit = limit
}
// LimitReached tells if the client refuses further calls.
-func (c *Client) LimitReached() bool {
+func (c *ModelClient) 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) {
+// limit it sends nothing and returns errModelCallLimit.
+func (c *ModelClient) Create(ctx context.Context, request ModelRequest) (ModelResponse, error) {
if c.calls.Add(1) > c.limit && c.limit > 0 {
- return Response{}, errCallLimit
+ return ModelResponse{}, errModelCallLimit
}
- body, err := json.Marshal(requestBody{
+ body, err := json.Marshal(ModelRequestBody{
Model: request.Model,
Instructions: request.Instructions,
Input: request.Input,
Tools: request.Tools,
- Reasoning: reasoning{
+ Reasoning: ModelReasoning{
Effort: request.Effort,
},
Store: false,
PromptCacheKey: c.cacheKey,
})
if err != nil {
- return Response{}, fmt.Errorf("encode request: %w", err)
+ return ModelResponse{}, fmt.Errorf("encode request: %w", err)
}
httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url, bytes.NewReader(body))
if err != nil {
- return Response{}, fmt.Errorf("create request: %w", err)
+ return ModelResponse{}, fmt.Errorf("create request: %w", err)
}
httpRequest.Header.Set("Content-Type", "application/json")
httpRequest.Header.Set("Accept", "text/event-stream")
err = c.auth.Authorize(ctx, httpRequest.Header)
if err != nil {
- return Response{}, fmt.Errorf("authorize model call: %w", err)
+ return ModelResponse{}, fmt.Errorf("authorize model call: %w", err)
}
_, chatgpt := c.auth.(*chatgptAuth)
}
start := time.Now()
- record := usageRecord{
+ record := ModelUsageRecord{
Date: start.Format(time.RFC3339Nano),
Model: request.Model,
}
// 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 *ModelClient) send(httpRequest *http.Request, record *ModelUsageRecord) (response ModelResponse, err error) {
httpResponse, err := c.http.Do(httpRequest)
if err != nil {
- return Response{}, fmt.Errorf("send model call: %w", err)
+ return ModelResponse{}, fmt.Errorf("send model call: %w", err)
}
defer func() {
}()
record.Status = httpResponse.StatusCode
- record.Quota = quotaHeaders(httpResponse.Header)
+ record.Quota = modelQuotaHeaders(httpResponse.Header)
if httpResponse.StatusCode != http.StatusOK {
- text, readErr := io.ReadAll(io.LimitReader(httpResponse.Body, errorBytes))
+ text, readErr := io.ReadAll(io.LimitReader(httpResponse.Body, modelErrorBytes))
err = fmt.Errorf("model call: HTTP %d: %s", httpResponse.StatusCode, strings.TrimSpace(string(text)))
- return Response{}, errors.Join(err, readErr)
+ return ModelResponse{}, errors.Join(err, readErr)
}
- return readStream(httpResponse.Body)
+ return modelReadStream(httpResponse.Body)
}
-// usageLog appends one JSON line for each model call to usage.jsonl. Calls can
-// run at once, so a mutex serializes the writes.
-type usageLog struct {
+// ModelUsageLog appends one JSON line for each model call to usage.jsonl. Calls
+// can run at once, so a mutex serializes the writes.
+type ModelUsageLog struct {
mutex sync.Mutex
path string
}
-// 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. 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"`
- OutputTokens int `json:"output_tokens"`
- TextBytes int `json:"text_bytes"`
- Quota map[string]string `json:"quota,omitempty"`
- Error string `json:"error,omitempty"`
-}
-
// Write appends record to the file with one write.
-func (u *usageLog) Write(record usageRecord) (err error) {
+func (u *ModelUsageLog) Write(record ModelUsageRecord) (err error) {
line, err := json.Marshal(record)
if err != nil {
return fmt.Errorf("encode usage: %w", err)
return nil
}
-// Authorize sets the key as the bearer token.
-func (k apiKey) Authorize(ctx context.Context, header http.Header) error {
- header.Set("Authorization", "Bearer "+string(k))
- return ctx.Err()
-}
-
-// userMessage returns a user input item with one text part for each text.
-func userMessage(texts ...string) any {
- type part struct {
+// ModelOutputItem holds the fields of an output item that give its text.
+type ModelOutputItem struct {
+ Type string `json:"type"`
+ Content []struct {
Type string `json:"type"`
Text string `json:"text"`
- }
+ } `json:"content"`
+}
- content := make([]part, len(texts))
- for n, text := range texts {
- content[n] = part{
- Type: "input_text",
- Text: text,
- }
- }
+// ModelRequest is one model call. Input holds the input items in order, such as
+// modelUserMessage values or output items of an earlier response. Node names
+// the node that a compaction builds, or is nil. Turn names the request of a
+// turn: first or next. Node and Turn go only to the usage file, never to the
+// endpoint.
+type ModelRequest struct {
+ Model string
+ Effort string
+ Instructions string
+ Tools []ModelTool
+ Input []any
+ Node *nodeKey
+ Turn string
+}
- return struct {
- Role string `json:"role"`
- Content []part `json:"content"`
- }{
- Role: "user",
- Content: content,
- }
+// ModelResponse is the result of a model call. Output holds the output items as
+// the endpoint sent them. Text is the text of the message items.
+type ModelResponse struct {
+ Output []json.RawMessage
+ Text string
+ Usage ModelUsage
+}
+
+// ModelRequestBody is the JSON body of a call.
+type ModelRequestBody struct {
+ Model string `json:"model"`
+ Instructions string `json:"instructions"`
+ Input []any `json:"input"`
+ Tools []ModelTool `json:"tools"`
+ Reasoning ModelReasoning `json:"reasoning"`
+ Store bool `json:"store"`
+ Stream bool `json:"stream"`
+ Include []string `json:"include"`
+ PromptCacheKey string `json:"prompt_cache_key"`
+}
+
+type ModelReasoning struct {
+ Effort string `json:"effort"`
}
-// quotaHeaders returns the x-codex-* headers, with lowercase names, without the
-// turn state.
-func quotaHeaders(header http.Header) map[string]string {
+// ModelUsageRecord 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. Turn names the request of a turn. Quota holds the
+// x-codex-* headers of the response, without the turn state.
+type ModelUsageRecord 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"`
+ OutputTokens int `json:"output_tokens"`
+ TextBytes int `json:"text_bytes"`
+ Quota map[string]string `json:"quota,omitempty"`
+ Error string `json:"error,omitempty"`
+}
+
+// ModelTool is the definition of a function tool.
+type ModelTool struct {
+ Type string `json:"type"`
+ Name string `json:"name"`
+ Description string `json:"description"`
+ Parameters any `json:"parameters"`
+ Strict bool `json:"strict"`
+}
+
+// ModelUsage gives the token counts of a call. Cached is the part of the input
+// tokens that came from the prompt cache.
+type ModelUsage struct {
+ Input int
+ Cached int
+ Output int
+}
+
+// ModelStreamEvent holds the fields of the stream events that the client reads.
+type ModelStreamEvent struct {
+ Type string `json:"type"`
+ Item json.RawMessage `json:"item"`
+ Response *ModelEventResponse `json:"response"`
+ Code string `json:"code"`
+ Message string `json:"message"`
+}
+
+type ModelEventResponse struct {
+ Output []json.RawMessage `json:"output"`
+ Usage *ModelWireUsage `json:"usage"`
+ Error *ModelWireError `json:"error"`
+ IncompleteDetails *struct {
+ Reason string `json:"reason"`
+ } `json:"incomplete_details"`
+}
+
+type ModelWireUsage struct {
+ InputTokens int `json:"input_tokens"`
+ OutputTokens int `json:"output_tokens"`
+ InputTokensDetails struct {
+ CachedTokens int `json:"cached_tokens"`
+ } `json:"input_tokens_details"`
+}
+
+type ModelWireError struct {
+ Code string `json:"code"`
+ Message string `json:"message"`
+}
+
+// modelQuotaHeaders returns the x-codex-* headers, with lowercase names,
+// without the turn state.
+func modelQuotaHeaders(header http.Header) map[string]string {
var quota map[string]string
for name, values := range header {
- if !strings.HasPrefix(name, quotaPrefix) || name == turnStateHeader || len(values) == 0 {
+ if !strings.HasPrefix(name, modelQuotaPrefix) || name == modelTurnStateHeader || len(values) == 0 {
continue
}
return quota
}
-// readStream reads the server-sent events of a call until the response
+// modelReadStream 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) {
- var response Response
+func modelReadStream(stream io.Reader) (ModelResponse, error) {
+ var response ModelResponse
reader := bufio.NewReader(stream)
var data bytes.Buffer
for {
data.WriteByte('\n')
}
} else if data.Len() > 0 {
- done, eventErr := readEvent(&response, data.Bytes())
+ done, eventErr := modelReadEvent(&response, data.Bytes())
data.Reset()
if eventErr != nil || done {
return response, eventErr
}
}
-// readEvent adds one stream event to response, and tells if it ends the
+// modelReadEvent adds one stream event to response, and tells if it ends the
// stream.
-func readEvent(response *Response, data []byte) (bool, error) {
- var event streamEvent
+func modelReadEvent(response *ModelResponse, data []byte) (bool, error) {
+ var event ModelStreamEvent
err := json.Unmarshal(data, &event)
if err != nil {
return true, fmt.Errorf("decode stream event: %w", err)
}
if event.Response.Usage != nil {
- response.Usage = Usage{
+ response.Usage = ModelUsage{
Input: event.Response.Usage.InputTokens,
Cached: event.Response.Usage.InputTokensDetails.CachedTokens,
Output: event.Response.Usage.OutputTokens,
}
}
- response.Text, err = outputText(response.Output)
+ response.Text, err = modelOutputText(response.Output)
return true, err
case "response.incomplete":
return false, nil
}
-
-// outputText returns the text of the message items of output.
-func outputText(output []json.RawMessage) (string, error) {
- var text strings.Builder
- for _, raw := range output {
- var item outputItem
- err := json.Unmarshal(raw, &item)
- if err != nil {
- return "", fmt.Errorf("decode output item: %w", err)
- }
-
- if item.Type != "message" {
- continue
- }
-
- for _, part := range item.Content {
- if part.Type == "output_text" {
- text.WriteString(part.Text)
- }
- }
- }
-
- return text.String(), nil
-}
"github.com/stretchr/testify/require"
)
-// testKey is a fake API key.
-const testKey = "sk-test-not-a-real-key"
+// modelTestKey is a fake API key.
+const modelTestKey = "sk-test-not-a-real-key"
func TestClientSendsTheFixedRequestFields(t *testing.T) {
- server := newFakeResponses(t)
- client := newTestClient(t, server)
+ server := newModelFake(t)
+ client := newTestModelClient(t, server)
- _, err := client.Create(t.Context(), Request{
+ _, err := client.Create(t.Context(), ModelRequest{
Model: "gpt-test",
Effort: "high",
Instructions: "You are a test.",
- Tools: []Tool{{
+ Tools: []ModelTool{{
Type: "function",
Name: "date",
Parameters: map[string]any{"type": "object"},
}},
- Input: []any{userMessage("<chat>\n</chat>\n", "hello")},
+ Input: []any{modelUserMessage("<chat>\n</chat>\n", "hello")},
})
require.NoError(t, err)
call := server.Call(0)
- require.Equal(t, "Bearer "+testKey, call.header.Get("Authorization"))
+ require.Equal(t, "Bearer "+modelTestKey, call.header.Get("Authorization"))
require.Equal(t, "text/event-stream", call.header.Get("Accept"))
require.Equal(t, "application/json", call.header.Get("Content-Type"))
testCases := []struct {
name string
- auth authorizer
+ auth ModelAuthorizer
want string
}{
{
},
{
name: "OpenAI API",
- auth: apiKey(testKey),
+ auth: ModelAPIKey(modelTestKey),
want: "",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- server := newFakeResponses(t)
- client := newClient(server.URL, tc.auth, "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
- _, err := client.Create(t.Context(), testRequest())
+ server := newModelFake(t)
+ client := newModelClient(server.URL, tc.auth, "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
+ _, err := client.Create(t.Context(), modelTestRequest())
require.NoError(t, err)
call := server.Call(0)
}
func TestClientReadsTheStream(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w,
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w,
map[string]any{"type": "response.created", "response": map[string]any{}},
map[string]any{"type": "response.output_item.done", "item": map[string]any{"type": "reasoning", "encrypted_content": "opaque"}},
map[string]any{"type": "response.output_text.delta", "delta": "Hel"},
- map[string]any{"type": "response.output_item.done", "item": messageItem("Hello")},
- map[string]any{"type": "response.output_item.done", "item": messageItem(" world")},
- completedEvent(nil, 1200, 1024, 30),
+ map[string]any{"type": "response.output_item.done", "item": modelMessageItem("Hello")},
+ map[string]any{"type": "response.output_item.done", "item": modelMessageItem(" world")},
+ modelCompletedEvent(nil, 1200, 1024, 30),
)
}
- response, err := newTestClient(t, server).Create(t.Context(), testRequest())
+ response, err := newTestModelClient(t, server).Create(t.Context(), modelTestRequest())
require.NoError(t, err)
require.Equal(t, "Hello world", response.Text)
require.Len(t, response.Output, 3)
require.JSONEq(t, `{"type": "reasoning", "encrypted_content": "opaque"}`, string(response.Output[0]))
- require.Equal(t, Usage{
+ require.Equal(t, ModelUsage{
Input: 1200,
Cached: 1024,
Output: 30,
}
func TestClientTakesTheOutputOfTheCompletedResponse(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem("only here")}, 10, 0, 2))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("only here")}, 10, 0, 2))
}
- response, err := newTestClient(t, server).Create(t.Context(), testRequest())
+ response, err := newTestModelClient(t, server).Create(t.Context(), modelTestRequest())
require.NoError(t, err)
require.Equal(t, "only here", response.Text)
}
func TestClientRecordsEachCall(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
w.Header().Set("X-Codex-Primary-Used-Percent", "12")
w.Header().Set("X-Codex-Plan-Type", "plus")
w.Header().Set("X-Other", "not quota")
return
}
- writeEvents(w, completedEvent([]any{messageItem("ok")}, 500, 400, 7))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("ok")}, 500, 400, 7))
}
- client := newTestClient(t, server)
- request := testRequest()
+ client := newTestModelClient(t, server)
+ request := modelTestRequest()
request.Node = &nodeKey{2, 5}
_, err := client.Create(t.Context(), request)
require.NoError(t, err)
require.NotContains(t, server.Call(0).body, "node")
require.NotContains(t, string(server.Call(0).raw), "[2,5]")
- _, err = client.Create(t.Context(), testRequest())
+ _, err = client.Create(t.Context(), modelTestRequest())
require.ErrorContains(t, err, "HTTP 429")
require.ErrorContains(t, err, "slow down")
- records := readUsage(t, client.usage.path)
+ records := modelReadUsage(t, client.usage.path)
require.Len(t, records, 2)
quota := map[string]string{
data, err := os.ReadFile(client.usage.path)
require.NoError(t, err)
- require.NotContains(t, string(data), testKey)
+ require.NotContains(t, string(data), modelTestKey)
info, err := os.Stat(client.usage.path)
require.NoError(t, err)
}
func TestClientRefusesCallsOverTheLimit(t *testing.T) {
- server := newFakeResponses(t)
- client := newTestClient(t, server)
+ server := newModelFake(t)
+ client := newTestModelClient(t, server)
client.LimitCalls(2)
for range 2 {
require.False(t, client.LimitReached())
- _, err := client.Create(t.Context(), testRequest())
+ _, err := client.Create(t.Context(), modelTestRequest())
require.NoError(t, err)
}
require.True(t, client.LimitReached())
- _, err := client.Create(t.Context(), testRequest())
- require.ErrorIs(t, err, errCallLimit)
+ _, err := client.Create(t.Context(), modelTestRequest())
+ require.ErrorIs(t, err, errModelCallLimit)
require.Equal(t, 2, server.Calls())
- require.Len(t, readUsage(t, client.usage.path), 2)
+ require.Len(t, modelReadUsage(t, client.usage.path), 2)
}
func TestClientFailsWhenTheResponseFails(t *testing.T) {
},
{
name: "stream ends early",
- events: []any{map[string]any{"type": "response.output_item.done", "item": messageItem("cut")}},
+ events: []any{map[string]any{"type": "response.output_item.done", "item": modelMessageItem("cut")}},
want: "ended before the response completed",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, tc.events...)
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, tc.events...)
}
- client := newTestClient(t, server)
- _, err := client.Create(t.Context(), testRequest())
+ client := newTestModelClient(t, server)
+ _, err := client.Create(t.Context(), modelTestRequest())
require.ErrorContains(t, err, tc.want)
- records := readUsage(t, client.usage.path)
+ records := modelReadUsage(t, client.usage.path)
require.Len(t, records, 1)
require.Contains(t, records[0].Error, tc.want)
})
}
}
-// fakeResponses is an httptest Responses endpoint. It keeps each call, and
-// reply writes the response. The default reply is one short message.
-type fakeResponses struct {
- *httptest.Server
- mutex sync.Mutex
- calls []fakeCall
- reply func(http.ResponseWriter, fakeCall)
-}
-
-// fakeCall is one call to a fake endpoint, numbered from 0.
-type fakeCall struct {
+// ModelFakeCall is one call to a fake endpoint, numbered from 0.
+type ModelFakeCall struct {
number int
header http.Header
raw []byte
body map[string]any
}
-func newFakeResponses(t *testing.T) *fakeResponses {
- t.Helper()
+// ModelFake is an httptest Responses endpoint. It keeps each call, and reply
+// writes the response. The default reply is one short message.
+type ModelFake struct {
+ *httptest.Server
+ mutex sync.Mutex
+ calls []ModelFakeCall
+ reply func(http.ResponseWriter, ModelFakeCall)
+}
- f := &fakeResponses{
- reply: func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem("a line")}, 100, 0, 5))
- },
- }
+// Call returns call number n.
+func (f *ModelFake) Call(n int) ModelFakeCall {
+ f.mutex.Lock()
+ defer f.mutex.Unlock()
- f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
- t.Cleanup(f.Close)
+ return f.calls[n]
+}
- return f
+// Calls returns the number of calls.
+func (f *ModelFake) Calls() int {
+ f.mutex.Lock()
+ defer f.mutex.Unlock()
+
+ return len(f.calls)
}
-func (f *fakeResponses) serve(w http.ResponseWriter, r *http.Request) {
+func (f *ModelFake) serve(w http.ResponseWriter, r *http.Request) {
raw, err := io.ReadAll(r.Body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
- call := fakeCall{
+ call := ModelFakeCall{
header: r.Header.Clone(),
raw: raw,
}
reply(w, call)
}
-// Call returns call number n.
-func (f *fakeResponses) Call(n int) fakeCall {
- f.mutex.Lock()
- defer f.mutex.Unlock()
+func newModelFake(t *testing.T) *ModelFake {
+ t.Helper()
- return f.calls[n]
-}
+ f := &ModelFake{
+ reply: func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("a line")}, 100, 0, 5))
+ },
+ }
-// Calls returns the number of calls.
-func (f *fakeResponses) Calls() int {
- f.mutex.Lock()
- defer f.mutex.Unlock()
+ f.Server = httptest.NewServer(http.HandlerFunc(f.serve))
+ t.Cleanup(f.Close)
- return len(f.calls)
+ return f
}
-func newTestClient(t *testing.T, server *fakeResponses) *Client {
+func newTestModelClient(t *testing.T, server *ModelFake) *ModelClient {
t.Helper()
- return newClient(server.URL, apiKey(testKey), "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
+ return newModelClient(server.URL, ModelAPIKey(modelTestKey), "agent-key", filepath.Join(t.TempDir(), "usage.jsonl"))
}
-func testRequest() Request {
- return Request{
+func modelTestRequest() ModelRequest {
+ return ModelRequest{
Model: "gpt-test",
Effort: "low",
- Input: []any{userMessage("hello")},
+ Input: []any{modelUserMessage("hello")},
}
}
-// writeEvents writes events as a server-sent event stream.
-func writeEvents(w http.ResponseWriter, events ...any) {
+// modelWriteEvents writes events as a server-sent event stream.
+func modelWriteEvents(w http.ResponseWriter, events ...any) {
w.Header().Set("Content-Type", "text/event-stream")
for _, event := range events {
data, err := json.Marshal(event)
}
}
-func messageItem(text string) map[string]any {
+func modelMessageItem(text string) map[string]any {
return map[string]any{
"type": "message",
"role": "assistant",
}
}
-func completedEvent(output []any, input, cached, out int) map[string]any {
+func modelCompletedEvent(output []any, input, cached, out int) map[string]any {
if output == nil {
output = []any{}
}
}
}
-func readUsage(t *testing.T, path string) []usageRecord {
+func modelReadUsage(t *testing.T, path string) []ModelUsageRecord {
t.Helper()
file, err := os.Open(path)
require.NoError(t, file.Close())
}()
- var records []usageRecord
+ var records []ModelUsageRecord
scanner := bufio.NewScanner(file)
for scanner.Scan() {
- var record usageRecord
+ var record ModelUsageRecord
require.NoError(t, json.Unmarshal(scanner.Bytes(), &record))
records = append(records, record)
}
return records
}
+
+func modelReasoningItem(id, encrypted string) map[string]any {
+ return map[string]any{
+ "type": "reasoning",
+ "id": id,
+ "encrypted_content": encrypted,
+ "summary": []any{},
+ }
+}
+
+func modelFunctionCallItem(id, callID, name, arguments string) map[string]any {
+ return map[string]any{
+ "type": "function_call",
+ "id": id,
+ "call_id": callID,
+ "name": name,
+ "arguments": arguments,
+ }
+}
}
usagePath := filepath.Join(root, "usage.jsonl")
- client := newClient(address, auth, "onechat-"+probeName, usagePath)
+ client := newModelClient(address, auth, "onechat-"+probeName, usagePath)
client.LimitCalls(probeCallLimit)
summarizer := slotSummarizer{
inner: newCompactor(client, probePrompt, config.Compactor),
// by the rules of probeStop.
type probeRun struct {
memory *Memory
- client *Client
+ client *ModelClient
usagePath string
}
// probeStop returns why a probe run must stop: a failed call, or a quota window
// whose used percent grew by more than probeQuotaLimit points since the first
// call. It returns "" when the run can go on.
-func probeStop(records []usageRecord) string {
+func probeStop(records []ModelUsageRecord) string {
for _, record := range records {
if record.Error != "" {
return "call failed: " + record.Error
// readUsageFile reads the records of usage.jsonl at path, in the order of the
// start of their calls. The file has them in the order of the end of their
// calls, which differs when calls run at once.
-func readUsageFile(path string) (records []usageRecord, err error) {
+func readUsageFile(path string) (records []ModelUsageRecord, err error) {
file, err := os.Open(path)
if errors.Is(err, os.ErrNotExist) {
return nil, nil
scanner := bufio.NewScanner(file)
scanner.Buffer(make([]byte, 64*1024), probeLineBytes)
for scanner.Scan() {
- var record usageRecord
+ var record ModelUsageRecord
err = json.Unmarshal(scanner.Bytes(), &record)
if err != nil {
return nil, fmt.Errorf("decode usage: %w", err)
return nil, fmt.Errorf("read usage file: %w", err)
}
- slices.SortStableFunc(records, func(a, b usageRecord) int {
+ slices.SortStableFunc(records, func(a, b ModelUsageRecord) int {
return callStart(a).Compare(callStart(b))
})
// callStart returns the start of the call of record, or the zero time when its
// date is not valid.
-func callStart(record usageRecord) time.Time {
+func callStart(record ModelUsageRecord) time.Time {
start, err := time.Parse(time.RFC3339Nano, record.Date)
if err != nil {
return time.Time{}
// usedPercents returns the used percent of each quota window of record, by
// header name.
-func usedPercents(record usageRecord) map[string]float64 {
+func usedPercents(record ModelUsageRecord) map[string]float64 {
used := make(map[string]float64)
for name, value := range record.Quota {
if !strings.HasSuffix(name, "-used-percent") {
}
// writeProbeReport writes the result of a probe run.
-func writeProbeReport(out io.Writer, appended int, stop string, records []usageRecord, sizes []int) error {
+func writeProbeReport(out io.Writer, appended int, stop string, records []ModelUsageRecord, sizes []int) error {
var input, cached, output, failed int
for _, record := range records {
input += record.InputTokens
// timeText describes the time per call and how many calls ran at once. A call
// runs from its date for its seconds.
-func timeText(records []usageRecord) string {
+func timeText(records []ModelUsageRecord) string {
type event struct {
at time.Time
delta int
}
// quotaText returns the quota headers of record, sorted by name.
-func quotaText(record usageRecord) string {
+func quotaText(record ModelUsageRecord) string {
names := slices.Sorted(maps.Keys(record.Quota))
parts := make([]string, len(names))
for n, name := range names {
}
func TestProbeStopRules(t *testing.T) {
- quota := func(primary, secondary string) usageRecord {
- return usageRecord{
+ quota := func(primary, secondary string) ModelUsageRecord {
+ return ModelUsageRecord{
Quota: map[string]string{
"x-codex-primary-used-percent": primary,
"x-codex-secondary-used-percent": secondary,
testCases := []struct {
name string
- records []usageRecord
+ records []ModelUsageRecord
want string
}{
{
},
{
name: "growth within the limit",
- records: []usageRecord{quota("10", "40"), quota("25", "41")},
+ records: []ModelUsageRecord{quota("10", "40"), quota("25", "41")},
want: "",
},
{
name: "growth over the limit",
- records: []usageRecord{quota("10", "40"), quota("12", "55.5")},
+ records: []ModelUsageRecord{quota("10", "40"), quota("12", "55.5")},
want: "x-codex-secondary-used-percent grew from 40 to 55.5",
},
{
name: "failed call",
- records: []usageRecord{quota("10", "40"), {
+ records: []ModelUsageRecord{quota("10", "40"), {
Status: 400,
Error: "model call: HTTP 400",
}},
}
func TestProbeRunStopsWhenTheQuotaGrows(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
w.Header().Set("X-Codex-Primary-Used-Percent", strconv.Itoa(10+call.number))
- writeEvents(w, completedEvent([]any{messageItem(strings.Repeat("s", 400))}, 1000, 800, 50))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(strings.Repeat("s", 400))}, 1000, 800, 50))
}
run := newTestProbeRun(t, server)
}
func TestProbeRunStopsAtTheCallLimit(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
run := newTestProbeRun(t, server)
run.client.LimitCalls(5)
gate := make(chan struct{})
var mutex sync.Mutex
running, most := 0, 0
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
mutex.Lock()
running++
most = max(most, running)
running--
mutex.Unlock()
- writeEvents(w, completedEvent([]any{messageItem("a line")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("a line")}, 100, 0, 5))
}
run := newTestProbeRun(t, server)
func TestReadUsageFileSortsByCallStart(t *testing.T) {
path := filepath.Join(t.TempDir(), "usage.jsonl")
- usage := usageLog{
+ usage := ModelUsageLog{
path: path,
}
for _, date := range []string{"2026-01-01T12:00:02Z", "2026-01-01T12:00:00.5Z", "2026-01-01T12:00:01.25Z"} {
- require.NoError(t, usage.Write(usageRecord{
+ require.NoError(t, usage.Write(ModelUsageRecord{
Date: date,
}))
}
func TestTimeTextCountsTheCallsAtOnce(t *testing.T) {
start := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
- record := func(offset, seconds float64) usageRecord {
- return usageRecord{
+ record := func(offset, seconds float64) ModelUsageRecord {
+ return ModelUsageRecord{
Date: start.Add(time.Duration(offset * float64(time.Second))).Format(time.RFC3339Nano),
Seconds: seconds,
}
}
- text := timeText([]usageRecord{record(0, 10), record(2, 4), record(3, 2), record(10, 10)})
+ text := timeText([]ModelUsageRecord{record(0, 10), record(2, 4), record(3, 2), record(10, 10)})
require.Equal(t, "time: 20 s in all, 12.0 calls per minute, 6.5 s per call (median 10.0, max 10.0), calls at once: at most 3, 1.3 on average", text)
}
return words[g.random.IntN(len(words))]
}
-func newTestProbeRun(t *testing.T, server *fakeResponses) *probeRun {
+func newTestProbeRun(t *testing.T, server *ModelFake) *probeRun {
t.Helper()
root := t.TempDir()
usagePath := filepath.Join(root, "usage.jsonl")
- client := newClient(server.URL, apiKey(testKey), "onechat-"+probeName, usagePath)
+ client := newModelClient(server.URL, ModelAPIKey(modelTestKey), "onechat-"+probeName, usagePath)
memory, err := openMemory(t.Context(), root, newTestClock().Now, newCompactor(client, probePrompt, defaultConfig().Compactor))
require.NoError(t, err)
// agentTools returns the function tools of every call: zoom, date, shell,
// read, write, and edit. A compaction sends them too and never calls them,
// because a change of the tools ends the shared prompt cache.
-func agentTools() []Tool {
- return []Tool{
+func agentTools() []ModelTool {
+ return []ModelTool{
{
Type: "function",
Name: "zoom",
)
func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem("Hello from the agent.")}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("Hello from the agent.")}, 100, 0, 5))
}
telegram := newTelegramFake(t)
func TestBotIgnoresEveryoneButTheOwner(t *testing.T) {
telegram := newTelegramFake(t)
bot, logs := telegramStartTestBot(t, telegram)
- socket := startTestDaemonWithBot(t, newFakeResponses(t), bot)
+ socket := startTestDaemonWithBot(t, newModelFake(t), bot)
terminal := startTestClient(t, socket)
terminal.WaitFor(t, "[connected to OneChat]")
func TestBotAnswersThatOnlyTextIsSupported(t *testing.T) {
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, newFakeResponses(t), bot)
+ startTestDaemonWithBot(t, newModelFake(t), bot)
telegram.Add(TelegramUpdate{
UpdateID: 1,
}
func TestStopFromTelegramInterruptsTheTurn(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
arrived := make(chan struct{})
release := make(chan struct{})
t.Cleanup(func() {
close(release)
})
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
close(arrived)
<-release
}
}
func TestBotStartsAtTheStoredOffset(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem("ok")}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("ok")}, 100, 0, 5))
}
telegram := newTelegramFake(t)
}
func TestBotTriesAgainAfterFailures(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem("ok")}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("ok")}, 100, 0, 5))
}
telegram := newTelegramFake(t)
func TestRepliesGoAsRichMessages(t *testing.T) {
reply := "**Done.** The table:\n\n| Name | Size |\n|:-----|-----:|\n| `a.go` | 10 |\n\n- one\n- two\n\n```go\nx := 1\n```"
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
- writeEvents(w, completedEvent([]any{messageItem(reply)}, 100, 0, 5))
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(reply)}, 100, 0, 5))
}
telegram := newTelegramFake(t)
telegram.Close()
bot, logs := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, newFakeResponses(t), bot)
+ startTestDaemonWithBot(t, newModelFake(t), bot)
require.Eventually(t, func() bool {
return strings.Count(logs.String(), "no valid answer") >= 2
"time"
)
-// modelClient sends model calls.
-type modelClient interface {
- Create(ctx context.Context, request Request) (Response, error)
+// TurnClient sends model calls.
+type TurnClient interface {
+ Create(ctx context.Context, request ModelRequest) (ModelResponse, error)
}
// Event tells the clients what the agent does as it happens. Kind is the kind
type loop struct {
agent *Agent
memory *Memory
- client modelClient
+ client TurnClient
tools *toolbox
layout *viewLayout
clock func() time.Time
// newLoop returns the loop of agent with its memory. Its calls go through
// client, and notify gets each event. The clock gives the date of a turn.
-func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time.Time, notify func(Event)) (*loop, error) {
+func newLoop(agent *Agent, memory *Memory, client TurnClient, clock func() time.Time, notify func(Event)) (*loop, error) {
layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName))
if err != nil {
return nil, err
input := turnInput(view, l.state(), messages)
for turn := "first"; ; turn = "next" {
- response, err := l.client.Create(ctx, Request{
+ response, err := l.client.Create(ctx, ModelRequest{
Model: l.agent.config.Model,
Effort: l.agent.config.Effort,
Instructions: instructions,
// input of the next request: input, the output items, the tool results, and
// the user messages that came in the meantime. After the final answer it
// returns no input.
-func (l *loop) handle(ctx context.Context, input []any, response Response) ([]any, error) {
+func (l *loop) handle(ctx context.Context, input []any, response ModelResponse) ([]any, error) {
var outputs []any
for _, raw := range response.Output {
item, err := inputItem(raw)
switch call.Type {
case "message":
- text, err := outputText([]json.RawMessage{raw})
+ text, err := modelOutputText([]json.RawMessage{raw})
if err != nil {
return nil, err
}
input = append(input, outputs...)
for _, message := range l.take() {
- input = append(input, userMessage(message.Text))
+ input = append(input, modelUserMessage(message.Text))
}
return input, nil
func turnInput(view []string, state string, messages []Message) []any {
input := make([]any, 0, len(view)+1+len(messages))
for _, text := range view {
- input = append(input, userMessage(text))
+ input = append(input, modelUserMessage(text))
}
- input = append(input, userMessage(state))
+ input = append(input, modelUserMessage(state))
for _, message := range messages {
- input = append(input, userMessage(message.Text))
+ input = append(input, modelUserMessage(message.Text))
}
return input
)
func TestTurnRunsToolsUntilTheFinalAnswer(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
- writeEvents(w, completedEvent([]any{
- reasoningItem("rs_1", "opaque thought"),
- functionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`),
+ modelWriteEvents(w, modelCompletedEvent([]any{
+ modelReasoningItem("rs_1", "opaque thought"),
+ modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`),
}, 100, 0, 5))
return
}
- writeEvents(w, completedEvent([]any{messageItem("It said hi.")}, 120, 100, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 120, 100, 5))
}
test := newTestLoop(t, server)
require.Equal(t, []string{"user", "busy", "tool", "echo", "reply", "idle"}, test.events.Kinds())
var turns []string
- for _, record := range readUsage(t, test.usagePath) {
+ for _, record := range modelReadUsage(t, test.usagePath) {
turns = append(turns, record.Turn)
}
}
func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
test := newTestLoop(t, server)
_, err := test.memory.Append("user", "an earlier message")
require.NoError(t, err)
}
func TestMessageDuringATurnJoinsTheNextRequest(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
arrived := make(chan struct{})
release := make(chan struct{})
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
close(arrived)
<-release
- writeEvents(w, completedEvent([]any{functionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
return
}
- writeEvents(w, completedEvent([]any{messageItem("done")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("done")}, 100, 0, 5))
}
test := newTestLoop(t, server)
}
func TestMessageDuringTheFinalRequestStartsANewTurn(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
arrived := make(chan struct{})
release := make(chan struct{})
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
close(arrived)
<-release
}
- writeEvents(w, completedEvent([]any{messageItem("answer")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5))
}
test := newTestLoop(t, server)
}
func TestTurnWaitsForTheSummaries(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
summarizer := newFakeSummarizer()
summarizer.gate = make(chan struct{})
test := newTestLoopWith(t, server, summarizer)
}
func TestInterruptEndsTheTurn(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
arrived := make(chan struct{})
release := make(chan struct{})
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
close(arrived)
<-release
}
- writeEvents(w, completedEvent([]any{messageItem("answer")}, 100, 0, 5))
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5))
}
test := newTestLoop(t, server)
}
func TestFailedCallEndsTheTurnWithoutARetry(t *testing.T) {
- server := newFakeResponses(t)
- server.reply = func(w http.ResponseWriter, call fakeCall) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
http.Error(w, "busy", http.StatusServiceUnavailable)
}
}
func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) {
- server := newFakeResponses(t)
+ server := newModelFake(t)
test := newTestLoop(t, server)
for n, text := range []string{"one", "two", "three"} {
_, err := test.loop.Submit(terminalSource, text)
}, 10*time.Second, time.Millisecond)
}
-func newTestLoop(t *testing.T, server *fakeResponses) *testLoop {
+func newTestLoop(t *testing.T, server *ModelFake) *testLoop {
t.Helper()
return newTestLoopWith(t, server, newFakeSummarizer())
// newTestLoopWith returns a running loop on a new agent whose builds use
// summarizer and whose commands run in a temporary directory.
-func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer) *testLoop {
+func newTestLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) *testLoop {
t.Helper()
agent, err := openAgent(t.TempDir())
usagePath: filepath.Join(t.TempDir(), "usage.jsonl"),
}
- client := newClient(server.URL, apiKey(testKey), agent.CacheKey(), test.usagePath)
+ client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), test.usagePath)
test.loop, err = newLoop(agent, memory, client, clock.Now, test.events.Add)
require.NoError(t, err)
ctx, cancel := context.WithCancel(t.Context())
}
// inputTexts returns the text of each user message of the input of call.
-func inputTexts(t *testing.T, call fakeCall) []string {
+func inputTexts(t *testing.T, call ModelFakeCall) []string {
t.Helper()
var body struct {
return texts
}
-
-func reasoningItem(id, encrypted string) map[string]any {
- return map[string]any{
- "type": "reasoning",
- "id": id,
- "encrypted_content": encrypted,
- "summary": []any{},
- }
-}
-
-func functionCallItem(id, callID, name, arguments string) map[string]any {
- return map[string]any{
- "type": "function_call",
- "id": id,
- "call_id": callID,
- "name": name,
- "arguments": arguments,
- }
-}
projects / onechat / commitdiff