- revive
- wrapcheck
- nlreturn
+ - wsl_v5
settings:
+ wsl_v5:
+ # go-code: Keep an operation with its immediate error check.
+ enable:
+ - err
+ - after-block
+ - after-decl
+ - after-defer
+ - after-expr
+ - after-go
+ # go-code: Keep a short declaration with its immediate use.
+ allow-first-in-block: true
+ allow-whole-block: true
+ cuddle-max-statements: 2
+ disable:
+ # go-code: Keep a short declaration with its immediate assignment use.
+ - assign
+ # go-code: Keep a short declaration, its operation, and its error check
+ # together.
+ - if
+ # go-code: A short declaration can feed the next loop setup.
+ - for
+ # go-code: A short declaration can feed the next switch setup.
+ - switch
+ # go-code: Keep a declared cleanup with its immediate defer.
+ - defer
+ # go-code: Keep a short declaration with its immediate append use.
+ - append
+ # go-code: Keep a short declaration with its immediate worker setup.
+ - go
nlreturn:
# go-code: Keep a short declaration with its immediate use.
block-size: 2
decoder := json.NewDecoder(bytes.NewReader(data))
decoder.DisallowUnknownFields()
+
err = decoder.Decode(&config)
if err == nil && !errors.Is(decoder.Decode(&struct{}{}), io.EOF) {
err = errors.New("data after the JSON object")
require.Equal(t, prompt(agent.config.Name, filepath.Join(dir, agentInstructionsFileName), ""), text)
require.NoError(t, os.WriteFile(filepath.Join(dir, agentInstructionsFileName), []byte("I am Lena.\n"), 0o600))
+
text, err = agent.Prompt()
require.NoError(t, err)
require.Equal(t, prompt(agent.config.Name, filepath.Join(dir, agentInstructionsFileName), "I am Lena."), text)
// Done removes a built node from the queue.
func (q *BuildQueue) Done(key NodeKey) {
delete(q.running, key)
+
if key.level > 0 {
delete(q.queued, key)
+
return
}
// Fail keeps a node in the queue, but it cannot start until Retry.
func (q *BuildQueue) Fail(key NodeKey) {
delete(q.running, key)
+
if key.level > 0 {
q.retry = append(q.retry, key)
return
// Retry lets the failed nodes start again.
func (q *BuildQueue) Retry() {
clear(q.failed)
+
for _, key := range q.retry {
heap.Push(&q.merges, key)
}
queue.AddMerge(NodeKey{1, 1})
var order []NodeKey
+
for {
key, ready := queue.Next()
if !ready {
require.True(t, ready)
queue.Fail(key)
+
_, ready = queue.Next()
require.False(t, ready)
queue.Retry()
+
key, ready = queue.Next()
require.True(t, ready)
require.Equal(t, NodeKey{1, 0}, key)
queue.Done(key)
queue.AddMerge(NodeKey{1, 0})
+
_, ready = queue.Next()
require.True(t, ready)
}
scanner := bufio.NewScanner(file)
scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
+
for scanner.Scan() {
var record ModelUsageRecord
+
err = json.Unmarshal(scanner.Bytes(), &record)
if err != nil {
return nil, fmt.Errorf("decode usage: %w", err)
// buildWriteReport writes the result of a build run.
func buildWriteReport(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
cached += record.CachedTokens
}
var lines []string
+
if stop != "" {
lines = append(lines, "stopped early: "+stop)
}
}
var events []event
+
seconds := make([]float64, 0, len(records))
total := 0.0
for _, record := range records {
}
wall := events[len(events)-1].at.Sub(events[0].at).Seconds()
+
slices.Sort(seconds)
return fmt.Sprintf("time: %.0f s in all, %.1f calls per minute, %.1f s per call (median %.1f, max %.1f), calls at once: at most %d, %.1f on average", wall, 60*float64(len(seconds))/max(wall, 1e-9), total/float64(len(seconds)), seconds[len(seconds)/2], seconds[len(seconds)-1], most, total/max(wall, 1e-9))
require.Greater(t, len(records), int(buildQuotaLimit))
var report strings.Builder
+
require.NoError(t, buildWriteReport(&report, appended, stop, records, sizes))
require.Contains(t, report.String(), "stopped early: ")
require.Contains(t, report.String(), "cached (80.0%)")
func TestBuildRunWaitsOnlyBeforeUserMessages(t *testing.T) {
gate := make(chan struct{})
+
var mutex sync.Mutex
+
running, most := 0, 0
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
mutex.Lock()
+
running++
most = max(most, running)
mutex.Unlock()
<-gate
mutex.Lock()
+
running--
mutex.Unlock()
}
var text string
+
input, text, err = c.handle(ctx, input, response)
if err != nil {
return "", err
// the messages of take. After the final answer it returns no input, and the
// replies of the answer as its text.
func (c *Chain) handle(ctx context.Context, input []any, response ModelResponse) ([]any, string, error) {
- var outputs []any
- var replies []string
+ var (
+ outputs []any
+ replies []string
+ )
+
for _, raw := range response.Output {
item, err := chainInputItem(raw)
if err != nil {
input = append(input, item)
var call ChainFunctionCall
+
err = json.Unmarshal(raw, &call)
if err != nil {
return nil, "", fmt.Errorf("decode output item: %w", err)
// id.
func chainInputItem(raw json.RawMessage) (map[string]any, error) {
var item map[string]any
+
err := json.Unmarshal(raw, &item)
if err != nil {
return nil, fmt.Errorf("decode output item: %w", err)
require.Equal(t, 2, server.Calls())
var kinds []string
+
for id := range memory.Len() {
message, err := memory.Message(id)
require.NoError(t, err)
defer a.mutex.Unlock()
var code ChatGPTDeviceCode
+
err := a.postJSON(ctx, "/api/accounts/deviceauth/usercode", map[string]string{
"client_id": chatgptClientID,
}, &code)
}
var tokens ChatGPTTokenFile
+
err = json.Unmarshal(data, &tokens)
if err == nil && (tokens.AccessToken == "" || tokens.RefreshToken == "" || tokens.AccountID == "") {
err = errors.New("a token or the account ID is missing")
// It keeps the old refresh token and ID token when the answer has none.
func (a *ChatGPTAuth) refresh(ctx context.Context, tokens ChatGPTTokenFile) (ChatGPTTokenFile, error) {
var response ChatGPTTokenResponse
+
err := a.postJSON(ctx, "/oauth/token", map[string]string{
"client_id": chatgptClientID,
"grant_type": "refresh_token",
for {
var grant ChatGPTDeviceGrant
+
status, err := a.post(ctx, "/api/accounts/deviceauth/token", "application/json", body, &grant)
if err == nil && (grant.AuthorizationCode == "" || grant.CodeVerifier == "") {
err = errors.New("device grant has no authorization code or code verifier")
select {
case <-ctx.Done():
timer.Stop()
+
return ChatGPTDeviceGrant{}, fmt.Errorf("wait for approval: %w", ctx.Err())
case <-timer.C:
}
}
var response ChatGPTTokenResponse
+
_, err := a.post(ctx, "/oauth/token", "application/x-www-form-urlencoded", []byte(form.Encode()), &response)
if err != nil {
return ChatGPTTokenFile{}, fmt.Errorf("exchange code for tokens: %w", err)
}
request.Header.Set("Content-Type", contentType)
+
response, err := a.http.Do(request)
if err != nil {
return 0, fmt.Errorf("send request: %w", err)
// none.
func chatgptAccountID(token string) string {
var claims map[string]json.RawMessage
+
if !chatgptDecodeClaims(token, &claims) {
return ""
}
func (f *ChatGPTFake) userCode(w http.ResponseWriter, r *http.Request) {
var body map[string]string
+
err := json.NewDecoder(r.Body).Decode(&body)
if err != nil || body["client_id"] != chatgptClientID {
http.Error(w, "bad client", http.StatusBadRequest)
+
return
}
defer f.mutex.Unlock()
var body map[string]string
+
err := json.NewDecoder(r.Body).Decode(&body)
if err != nil || body["device_auth_id"] != "device-1" || body["user_code"] != "CODE-1234" {
http.Error(w, "bad device", http.StatusBadRequest)
+
return
}
f.polls++
if f.deny {
http.Error(w, "denied", http.StatusBadRequest)
+
return
}
if f.polls <= f.pending {
http.Error(w, "pending", http.StatusForbidden)
+
return
}
err := r.ParseForm()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
+
return
}
}
var body map[string]string
+
err := json.NewDecoder(r.Body).Decode(&body)
if err != nil || body["grant_type"] != "refresh_token" || body["client_id"] != chatgptClientID {
http.Error(w, "bad refresh", http.StatusBadRequest)
+
return
}
f.refreshed = body["refresh_token"]
if f.failRefresh {
http.Error(w, `{"error": "invalid_grant", "token": "`+body["refresh_token"]+`"}`, http.StatusUnauthorized)
+
return
}
mux.HandleFunc("POST /api/accounts/deviceauth/usercode", f.userCode)
mux.HandleFunc("POST /api/accounts/deviceauth/token", f.deviceToken)
mux.HandleFunc("POST /oauth/token", f.oauthToken)
+
f.Server = httptest.NewServer(mux)
t.Cleanup(f.Close)
func chatgptWriteJSON(w http.ResponseWriter, value any) {
w.Header().Set("Content-Type", "application/json")
+
err := json.NewEncoder(w).Encode(value)
if err != nil {
panic(fmt.Sprint("encode answer: ", err))
auth := newTestChatGPTAuth(t, server)
var out strings.Builder
+
err := auth.Login(t.Context(), &out)
require.NoError(t, err)
chatgptWriteTestTokens(t, auth, server.token(time.Second), "refresh-0")
var group sync.WaitGroup
+
errs := make(chan error, 8)
for range 8 {
group.Go(func() {
group.Wait()
close(errs)
+
for err := range errs {
require.NoError(t, err)
}
server.failRefresh = true
chatgptWriteTestTokens(t, auth, server.token(-time.Minute), "refresh-0")
+
before, err := os.ReadFile(auth.path)
require.NoError(t, err)
require.NoError(t, err)
require.Equal(t, 1, server.refreshes)
+
call := responses.Call(0)
require.Equal(t, "Bearer "+server.access[1], call.header.Get("Authorization"))
require.Equal(t, chatgptTestAccount, call.header.Get("ChatGPT-Account-Id"))
config := defaultConfig()
t.Setenv("OPENAI_API_KEY", modelTestKey)
+
address, auth, err := modelEndpoint(config, dir)
require.NoError(t, err)
require.Equal(t, modelChatGPTURL, address)
require.Equal(t, ModelAPIKey(modelTestKey), auth)
t.Setenv("OPENAI_API_KEY", "")
+
_, _, err = modelEndpoint(config, dir)
require.ErrorContains(t, err, "OPENAI_API_KEY")
}
require.NoError(t, err)
var tokens ChatGPTTokenFile
+
require.NoError(t, json.Unmarshal(data, &tokens))
return tokens
}
var first []any
+
for _, message := range compactorTexts(task.Context, text) {
first = append(first, modelUserMessage(message))
}
var lines []string
+
note := ""
for len(lines) < compactorTries {
input := first
}
var response ModelResponse
+
response, err = c.client.Create(ctx, ModelRequest{
Model: c.model.Model,
Effort: c.model.Effort,
whole := len(lines) - len(lines)%compactorBlock
var texts []string
+
for n := 0; n < whole; n += compactorBlock {
texts = append(texts, viewMessage(n == 0, lines[n:n+compactorBlock]))
}
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)
require.Equal(t, len(long[n]), record.TextBytes)
first := compactorInputTexts(t, server.Call(0))
require.Len(t, first, 2)
+
for n, want := range []string{long[0], long[1]} {
call := server.Call(n + 1)
texts := compactorInputTexts(t, call)
name: "failed call",
reply: func(w http.ResponseWriter, _ ModelFakeCall) {
w.WriteHeader(http.StatusInternalServerError)
+
_, _ = io.WriteString(w, "down")
},
want: "HTTP 500",
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
w.WriteHeader(http.StatusServiceUnavailable)
+
return
}
require.Equal(t, "user", message.Role)
require.Len(t, message.Content, 1)
require.Equal(t, "input_text", message.Content[0].Type)
+
texts[n] = message.Content[0].Text
}
decoder := json.NewDecoder(bytes.NewReader(data))
decoder.DisallowUnknownFields()
+
err = decoder.Decode(&config)
if err == nil && !errors.Is(decoder.Decode(&struct{}{}), io.EOF) {
err = errors.New("data after the JSON object")
}()
whatsappSetDeviceProps()
+
logger := daemonLogger(out, "onechat: ")
whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)
if err != nil {
logger.Printf("ready: socket %s, model %s (%s), workdir %s, %s, %d schedules", listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir, linked, schedules.Len())
var telegram *TelegramBot
+
if hasBot {
telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))
+
logger.Print("Telegram bot started")
}
}
h.mutex.Lock()
+
h.clients[client] = true
h.mutex.Unlock()
h.workers.Add(2)
+
go h.write(client)
go h.read(ctx, client, l)
scanner := bufio.NewScanner(client.conn)
scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
+
for ctx.Err() == nil && scanner.Scan() {
var request SocketRequest
+
err := json.Unmarshal(scanner.Bytes(), &request)
if err == nil {
err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
delete(h.clients, client)
client.once.Do(func() {
close(client.events)
+
_ = client.conn.Close()
})
}
// CloseAll drops every client and waits for their readers and writers.
func (h *DaemonHub) CloseAll() {
h.mutex.Lock()
+
for client := range h.clients {
h.remove(client)
}
+
h.mutex.Unlock()
h.workers.Wait()
}
var workers sync.WaitGroup
+
workers.Add(2)
+
go func() {
defer workers.Done()
}()
var whatsappErr error
+
go func() {
defer workers.Done()
whatsappErr = whatsapp.Run(ctx, l, notify)
+
cancel()
}()
var schedulesErr error
+
workers.Add(1)
+
go func() {
defer workers.Done()
schedulesErr = schedules.Run(ctx, l)
+
cancel()
}()
var botErr error
+
if bot != nil {
workers.Add(1)
+
go func() {
defer workers.Done()
botErr = bot.Run(ctx, l, whatsapp)
+
cancel()
}()
}
}
cancel()
+
err = errors.Join(err, listener.Close())
+
hub.CloseAll()
workers.Wait()
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`)}, 100, 0, 5))
+
return
}
second.WaitFor(t, "[connected to OneChat]")
first.Type("Run echo hi.")
+
for _, client := range []*TerminalTestClient{first, second} {
client.WaitFor(t, "[done]")
require.Equal(t, "[connected to OneChat]\n[working]\n[tool] shell {\"command\":\"echo hi\"}\n[result] hi\nIt said hi.\n[done]\n", client.Output())
t.Setenv("JOURNAL_STREAM", tc.journal)
var out strings.Builder
+
daemonLogger(&out, "onechat: ").Print("the agent is ready")
require.Regexp(t, tc.line, out.String())
})
dir := os.Getenv("ONECHAT_DIR")
if dir == "" {
var err error
+
dir, err = defaultDataDir()
if err != nil {
return "", err
reader, open := j.readers[place.day]
if !open {
var err error
+
reader, err = os.Open(filepath.Join(j.dir, journalDayName(place.day)+".jsonl"))
if err != nil {
return nil, fmt.Errorf("open day file: %w", err)
err := j.closeWriter()
for day, reader := range j.readers {
err = errors.Join(err, reader.Close())
+
delete(j.readers, day)
}
}
slices.Sort(paths)
+
for n, path := range paths {
day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
if err != nil {
}
var err error
+
now := l.clock()
if date == "" {
date = now.Format(time.RFC3339Nano)
}
var message Message
+
err = json.Unmarshal(line, &message)
if err != nil {
return Message{}, fmt.Errorf("decode message %d: %w", id, err)
return l.journal.Scan(ctx, start, func(line []byte, place Position) error {
var message Message
+
err := json.Unmarshal(line, &message)
if err != nil {
return fmt.Errorf("decode log message: %w", err)
require.NoError(t, err)
var written []Message
+
for _, kind := range []string{"user", "reply", "tool", "echo", "work", "note"} {
message, err := log.Append(kind, kind+" text ✓")
require.NoError(t, err)
}()
require.Equal(t, len(written), log.Len())
+
for _, message := range written {
read, err := log.Message(message.ID)
require.NoError(t, err)
first := `{"i":0,"kind":"user","text":"x","size":1,"date":"2026-01-01T12:00:00Z"}` + "\n" + `{"i":1,"kin`
second := `{"i":1,"kind":"user","text":"y","size":1,"date":"2026-01-02T12:00:00Z"}` + "\n"
+
require.NoError(t, os.WriteFile(filepath.Join(root, "main", "2026-01-01.jsonl"), []byte(first), 0o600))
require.NoError(t, os.WriteFile(filepath.Join(root, "main", "2026-01-02.jsonl"), []byte(second), 0o600))
if err != nil {
return err
}
+
// A signal is the normal stop of the daemon and the client, so their
// result does not include ctx.Err().
case len(args) == 1 && args[0] == "daemon":
defer m.mutex.Unlock()
var err error
+
if m.syncer != nil {
err = m.syncer.Close()
}
// children reads the two children of the node key.
func (m *Memory) children(key NodeKey) (Node, Node, error) {
var children [2]Node
+
for n := range children {
child, built, err := m.tree.Node(key.level-1, 2*key.index+n)
if err != nil {
func (m *Memory) Wait(ctx context.Context) error {
for {
m.mutex.Lock()
+
running := m.queue.Running()
changed := m.changed
m.mutex.Unlock()
func (m *Memory) Settle(ctx context.Context) error {
for {
m.mutex.Lock()
+
pending := m.chat.Pending()
running := m.queue.Running()
changed := m.changed
}
var file MemoryViewFile
+
err = json.Unmarshal(data, &file)
if err != nil {
return fmt.Errorf("decode view: %w", err)
leaf, free := freeLeaf(message)
if !free {
m.queue.AddLeaf(id)
+
continue
}
}
m.workers.Add(1)
+
go m.run(key, task)
}
}
node, free := freeMerge(left, right)
if !free {
m.queue.AddMerge(parent)
+
return nil
}
}
close(m.changed)
+
m.changed = make(chan struct{})
m.dispatch()
}
}
var sizes []int
+
for _, key := range keys {
var node Node
memory.chat.high = 200
memory.chat.low = 100
+
memory.cview.high = 80
memory.cview.low = 40
for n := range 100 {
memory.chat.high = 200
memory.chat.low = 100
+
memory.cview.high = 80
memory.cview.low = 40
for n := range 64 {
}
memory.mutex.Lock()
+
running := slices.Collect(maps.Keys(memory.queue.running))
memory.mutex.Unlock()
slices.SortFunc(running, func(a, b NodeKey) int {
func TestMemorySettleFailsWhileABuildFailed(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
+
logs := &SyncBuffer{}
memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, log.New(logs, "", 0))
require.NoError(t, err)
require.Equal(t, "summary failed, retry at next message: node (0, 0): fake failure\n", logs.String())
summarizer.SetFailing(false)
+
_, err = memory.Append("user", "retry")
require.NoError(t, err)
require.NoError(t, memory.Settle(t.Context()))
func TestMemoryStartsALeafAfterFewerThanEightUnbuilt(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
+
memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
}
require.NoError(t, memory.Wait(t.Context()))
+
for _, task := range summarizer.Tasks() {
require.Less(t, task.Index, 8)
}
clock := newTestClock()
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
+
memory, err := openMemory(t.Context(), root, clock.Now, summarizer, discardLog())
require.NoError(t, err)
func TestMemoryKeepsALongMessageOutOfTheView(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
+
memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
// 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)
// 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()
}
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 ModelResponse{}, fmt.Errorf("authorize model call: %w", err)
// 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, modelQuotaPrefix) || name == modelTurnStateHeader || len(values) == 0 {
continue
// the completed response when the stream sent none.
func modelReadStream(stream io.Reader) (ModelResponse, error) {
var response ModelResponse
+
reader := bufio.NewReader(stream)
+
var data bytes.Buffer
+
for {
line, err := reader.ReadBytes('\n')
if err != nil && !errors.Is(err, io.EOF) {
} else if data.Len() > 0 {
done, eventErr := modelReadEvent(&response, data.Bytes())
data.Reset()
+
if eventErr != nil || done {
return response, eventErr
}
// stream.
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)
raw, err := io.ReadAll(r.Body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
+
return
}
err = json.Unmarshal(raw, &call.body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
+
return
}
f.mutex.Lock()
+
call.number = len(f.calls)
f.calls = append(f.calls, call)
reply := f.reply
// 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)
if err != nil {
}()
var records []ModelUsageRecord
+
scanner := bufio.NewScanner(file)
for scanner.Scan() {
var record ModelUsageRecord
+
require.NoError(t, json.Unmarshal(scanner.Bytes(), &record))
+
records = append(records, record)
}
w.Header().Set("X-Codex-Plan-Type", "plus")
w.Header().Set("X-Other", "not quota")
w.Header().Set("X-Codex-Turn-State", "opaque state")
+
if call.number == 1 {
w.WriteHeader(http.StatusTooManyRequests)
+
_, _ = io.WriteString(w, `{"detail": "slow down"}`)
return
"x-codex-primary-used-percent": "12",
"x-codex-plan-type": "plus",
}
+
require.Equal(t, "gpt-test", records[0].Model)
require.Equal(t, http.StatusOK, records[0].Status)
require.Equal(t, 500, records[0].InputTokens)
for range 2 {
require.False(t, client.LimitReached())
+
_, err := client.Create(t.Context(), modelTestRequest())
require.NoError(t, err)
}
require.True(t, client.LimitReached())
+
_, err := client.Create(t.Context(), modelTestRequest())
require.ErrorIs(t, err, errModelCallLimit)
require.Equal(t, 2, server.Calls())
usagePath := filepath.Join(root, "usage.jsonl")
client := newModelClient(address, auth, "onechat-"+probeName, usagePath)
client.LimitCalls(buildCallLimit)
+
summarizer := ProbeSlotSummarizer{
inner: newCompactor(client, probePrompt, config.Compactor),
slots: make(chan struct{}, builds),
scanner := bufio.NewScanner(file)
scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
+
for len(entries) < count && scanner.Scan() {
var entry DatedMessage
+
err = json.Unmarshal(scanner.Bytes(), &entry)
if err != nil {
return nil, fmt.Errorf("probe log line %d: %w", len(entries)+1, err)
// fill joins lines of write until the text has about size bytes.
func (g *ProbeWriter) fill(write func() string, size int) string {
var text strings.Builder
+
for text.Len() < size {
if text.Len() > 0 {
text.WriteByte(' ')
// toolCall returns a tool name and its JSON input of about size bytes.
func (g *ProbeWriter) toolCall(size int) string {
name := g.pick([]string{"shell", "read", "write", "edit"})
+
var input map[string]string
+
switch name {
case "shell":
input = map[string]string{"command": "grep -n '" + g.pick(probeThings) + "' " + g.pick(probeFiles)}
counts[message.Kind]++
require.True(t, logKinds[message.Kind], message.Kind)
require.NotEmpty(t, message.Text)
+
for _, char := range message.Text {
require.True(t, char == '\n' || (char >= ' ' && char <= '~'), "character %q", char)
}
}
var data strings.Builder
+
for _, message := range messages {
line, err := json.Marshal(message)
require.NoError(t, err)
}
var group sync.WaitGroup
+
for n := range 6 {
group.Go(func() {
_, err := summarizer.Summarize(t.Context(), Task{
for _, tool := range tools {
schema := tool.Parameters.(map[string]any)
properties := schema["properties"].(map[string]any)
+
require.True(t, tool.Strict, tool.Name)
require.Equal(t, false, schema["additionalProperties"], tool.Name)
require.ElementsMatch(t, slices.Collect(maps.Keys(properties)), schema["required"], tool.Name)
}
client.LimitCalls(buildCallLimit)
+
run := BuildRun{
memory: memory,
client: client,
}
var work SubagentWorkFile
+
err = json.Unmarshal(data, &work)
if err != nil {
return fmt.Errorf("decode running subagents: %w", err)
}
slices.Sort(paths)
+
var entries []DatedMessage
+
for _, path := range paths {
var day uint32
err = journalScanFile(ctx, path, day, 0, false, func(line []byte, _ Position) error {
var message Message
+
err = json.Unmarshal(line, &message)
if err != nil {
return fmt.Errorf("decode message: %w", err)
}
var date time.Time
+
date, err = time.Parse(time.RFC3339Nano, entry.Date)
if err != nil {
return nil, fmt.Errorf("parse merge date: %w", err)
}
case entry.Type().IsRegular():
var data []byte
+
data, err = os.ReadFile(path)
if err != nil {
return fmt.Errorf("read agent file: %w", err)
before := rebuildTestSnapshot(t, agent.dir)
server := newModelFake(t)
client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(t.TempDir(), "usage.jsonl"))
+
var report strings.Builder
+
added := []DatedMessage{
{
Kind: "work",
}
rebuildTestLog(t, agent.dir, old)
+
before := rebuildTestSnapshot(t, agent.dir)
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
}()
batches := 0
+
low := viewChatHigh
long := make([]bool, messages)
for id := range messages {
runtime.GC()
var stats runtime.MemStats
+
runtime.ReadMemStats(&stats)
t.Logf("heap in use: %d KB", stats.HeapInuse/1024)
}
}
require.Equal(t, head+strings.ReplaceAll(node.Text, "\n", " "), rendered[n+1])
+
if line.level == 0 && long[line.index] {
require.True(t, strings.HasPrefix(node.Text, "summary "), "message %d", line.index)
}
return err
}
- var timer *time.Timer
- var timeout <-chan time.Time
+ var (
+ timer *time.Timer
+ timeout <-chan time.Time
+ )
+
wait, found := s.Wait(s.clock())
if found {
timer = time.NewTimer(min(wait, s.check))
// text of the schedule.
func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
now := s.clock()
+
var turns []Message
+
for _, schedule := range s.Due(now) {
message, err := l.Work(scheduleSource, schedule.Message(now), "")
if err != nil {
// Due returns the schedules whose time is not after now, by due time.
func (s *Schedules) Due(now time.Time) []Schedule {
var due []Schedule
+
for _, schedule := range s.sorted() {
if !schedule.Due.After(now) {
due = append(due, schedule)
// sorted returns a copy of the schedules by due time, then by ID.
func (s *Schedules) sorted() []Schedule {
s.mutex.Lock()
+
schedules := slices.Clone(s.file.Schedules)
s.mutex.Unlock()
}
var sets [5]uint64
+
for index, field := range fields {
set, err := scheduleCronField(field, scheduleCronFields[index].low, scheduleCronFields[index].high)
if err != nil {
// one day are not always in order. The first time is the
// smallest one.
var next time.Time
+
for hour := range 24 {
if c.hours&(1<<hour) == 0 {
continue
// and the minute fields, also across midnight.
func (c ScheduleCron) gap() int {
var times []int
+
for hour := range 24 {
for minute := range 60 {
if c.hours&(1<<hour) != 0 && c.minutes&(1<<minute) != 0 {
// high.
func scheduleCronField(field string, low, high int) (uint64, error) {
var set uint64
+
for item := range strings.SplitSeq(field, ",") {
base, stepText, stepped := strings.Cut(item, "/")
step := 1
if stepped {
var err error
+
step, err = scheduleCronNumber(stepText, 1, high)
if err != nil {
return 0, fmt.Errorf("the step of %q: %w", item, err)
case base == "*":
case ranged:
var err error
+
first, err = scheduleCronNumber(startText, low, high)
if err == nil {
last, err = scheduleCronNumber(endText, low, high)
return 0, fmt.Errorf("%q has a step after a single value: write a range N-M/S or */S", item)
default:
var err error
+
first, err = scheduleCronNumber(base, low, high)
if err != nil {
return 0, err
now := start.Add(3*time.Hour + 30*time.Minute)
due := schedules.Due(now)
require.Len(t, due, 2)
+
for _, schedule := range due {
require.NoError(t, schedules.Done(schedule, now))
}
defer s.mutex.Unlock()
var err error
+
for name, subagent := range s.open {
err = errors.Join(err, subagent.close())
+
delete(s.open, name)
}
}
r.mutex.Lock()
+
closed := r.closed
r.mutex.Unlock()
// its log with a new run, with the view of now and the effort of its note.
func (r *SubagentRuns) Tell(ctx context.Context, name, message string) (string, error) {
r.mutex.Lock()
+
run, found := r.running[name]
if found {
defer r.mutex.Unlock()
// Close ends every run and waits for them. No subagent starts after Close.
func (r *SubagentRuns) Close() {
r.mutex.Lock()
+
r.closed = true
for _, run := range r.running {
run.cancel()
}
+
r.mutex.Unlock()
r.workers.Wait()
ctx, run.cancel = context.WithCancel(context.WithoutCancel(ctx))
r.mutex.Lock()
+
_, found := r.running[run.subagent.name]
if r.closed || found {
r.mutex.Unlock()
}
var file SubagentWorkFile
+
err = json.Unmarshal(data, &file)
if err != nil {
return nil, fmt.Errorf("decode running subagents: %w", err)
defer r.workers.Done()
defer run.cancel()
- var replies []string
- var failure error
+ var (
+ replies []string
+ failure error
+ )
+
stopped := false
status := subagentFinished
for more := true; more; {
reply, err := r.chain(ctx, run)
r.mutex.Lock()
+
stopped = run.stopped
switch {
case err != nil && stopped:
more = err == nil && ctx.Err() == nil && len(run.pending) > 0
run.pending = nil
+
var saveErr error
+
if !more {
delete(r.running, run.subagent.name)
+
if !r.closed {
saveErr = r.save()
}
}
+
r.mutex.Unlock()
if saveErr != nil {
func TestSpawnRunsASubagentInTheBackground(t *testing.T) {
server := newModelFake(t)
+
var test *TurnTestLoop
+
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
last := subagentTestLastItem(call)
switch {
name := test.events.Last("subagent")
require.True(t, subagentNameValid(name), name)
+
work := filepath.Join(test.loop.runs.subagents.dir, name, "work")
require.DirExists(t, work)
require.Equal(t, "subagent "+name+" started\nsubagent "+name+" done after 0s\n", test.logs.String())
var first, second *ModelFakeCall
+
for n := range server.Calls() {
call := server.Call(n)
if !subagentTestCall(call) {
require.Equal(t, instructions, first.body["instructions"])
var tools []string
+
for _, tool := range first.body["tools"].([]any) {
tools = append(tools, tool.(map[string]any)["name"].(string))
}
require.Equal(t, "0|note: subagent\n\n1|user: Count the files.\n\n2|tool: shell {\"command\":\"pwd\"}\n\n3|echo: "+work+"\n\n\n4|reply: There are 3 files.", log)
var agents []string
+
for _, record := range modelReadUsage(t, test.usagePath) {
agents = append(agents, record.Agent)
}
require.NoError(t, err)
<-arrived
+
report := subagentReport("GreenOtter", []string{"It is done."})
test.loop.Report("GreenOtter", report, subagentFinished)
close(release)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if subagentTestCall(call) {
w.WriteHeader(http.StatusBadRequest)
+
_, _ = io.WriteString(w, subagentTestFullContext)
return
require.NoError(t, err)
<-arrived
+
name := test.events.Last("subagent")
output, err := runs.Tell(t.Context(), name, "Use the second disk.")
require.NoError(t, err)
require.NoError(t, err)
<-arrived
+
name := test.events.Last("subagent")
_, err = runs.Tell(t.Context(), name, "Count again.")
require.NoError(t, err)
})
var restarted atomic.Bool
+
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
switch {
case subagentTestCall(call) && !restarted.Load():
arrived <- struct{}{}
+
<-release
case subagentTestCall(call):
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("Built.")}, 100, 0, 5))
require.NoError(t, err)
<-arrived
+
name := first.Last("subagent")
+
stop()
require.Zero(t, first.Count("work"))
require.JSONEq(t, `{"running": ["`+name+`"]}`, string(data))
restarted.Store(true)
+
second := &TurnEventRecorder{}
loop, memory, stop := subagentTestStart(t, agent, server, second)
defer stop()
second.WaitFor(t, "idle", 1)
+
note := "The daemon restarted, cutting " + name + "; nothing was lost: go on."
log := subagentTestChat(t, memory)
require.Equal(t, []string{
t.Helper()
var lines []string
+
for id := range memory.Len() {
message, err := memory.Message(id)
require.NoError(t, err)
// another item.
func subagentTestText(item map[string]any) string {
content, _ := item["content"].([]any)
+
var text strings.Builder
+
for _, part := range content {
value, _ := part.(map[string]any)["text"].(string)
text.WriteString(value)
}
require.True(t, subagentNameValid("GreenOtter"))
+
for _, name := range []string{"", "Green", "Greenotter", "greenOtter", "Green-Otter", "GreenOtterX", "OtterGreen", "../GreenOtter"} {
require.False(t, subagentNameValid(name), name)
}
func (f *FakeSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
f.mutex.Lock()
+
f.tasks = append(f.tasks, task)
f.active++
f.peak = max(f.peak, f.active)
defer func() {
f.mutex.Lock()
+
f.active--
f.mutex.Unlock()
}()
// Flush returns when every write before the call is durable.
func (s *Syncer) Flush(ctx context.Context) error {
s.mutex.Lock()
+
target := s.written
s.mutex.Unlock()
for {
s.mutex.Lock()
+
reached := s.durable >= target
failure := s.failure
done := s.done
func (s *Syncer) Close() error {
err := s.Flush(context.Background())
s.workers.Wait()
+
if s.file != nil {
err = errors.Join(err, s.file.Close())
s.file = nil
s.running = true
s.workers.Add(1)
+
go s.run(s.written)
}
}
close(s.done)
+
s.done = make(chan struct{})
s.start()
}
// This write comes during the first sync, so that sync does not cover
// it.
s.Written()
+
flushed := make(chan error)
go func() {
flushed <- s.Flush(t.Context())
gate.Release()
gate.WaitStarted(t, 2)
+
select {
case err := <-flushed:
t.Fatalf("flush returned before the second sync: %v", err)
s := newSyncer(gate.Sync)
s.Written()
gate.WaitStarted(t, 1)
+
for range 10 {
s.Written()
}
memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
require.GreaterOrEqual(t, memory.log.Len(), acknowledged)
+
for id := range memory.log.Len() {
message, err := memory.log.Message(id)
require.NoError(t, err)
func (g *SyncerGatedSync) Sync() error {
g.mutex.Lock()
+
g.count++
g.mutex.Unlock()
g.started <- struct{}{}
+
<-g.release
return nil
// Release lets the running sync end.
func (g *SyncerGatedSync) Release() {
<-g.started
+
g.release <- struct{}{}
}
}
require.NoError(t, command.Process.Kill())
+
for scanner.Scan() {
_, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
if err == nil {
}
var config TelegramConfig
+
decoder := json.NewDecoder(bytes.NewReader(data))
decoder.DisallowUnknownFields()
+
err = decoder.Decode(&config)
if err == nil && !errors.Is(decoder.Decode(&struct{}{}), io.EOF) {
err = errors.New("data after the JSON object")
}
b.mutex.Lock()
+
b.queue = append(b.queue, event)
b.mutex.Unlock()
defer cancel()
var sender sync.WaitGroup
+
sender.Add(1)
+
go func() {
defer sender.Done()
}()
err := b.poll(ctx, l, whatsapp)
+
cancel()
sender.Wait()
switch event.Kind {
case "busy":
busy = true
+
b.typing(ctx)
ticker.Reset(telegramTypingInterval)
case "idle":
}
outage.End()
+
wait = b.minWait
for _, update := range updates {
if ctx.Err() != nil {
message := update.Message
if message == nil || message.Chat.Type != "private" || message.Chat.ID != b.owner || message.From == nil || message.From.ID != b.owner {
b.logger.Print("message ignored: not from the owner")
+
return nil
}
}
var answer *TelegramError
+
if len(part.entities) > 0 && errors.As(err, &answer) && answer.Status == http.StatusBadRequest {
b.logger.Printf("reply sent as plain text: %v", err)
+
part.entities = nil
continue
if !retry {
b.logger.Printf("reply lost: %v", err)
+
return
}
if telegramRejected(err) {
b.logger.Printf("reply sent with simple formatting: %v", err)
+
for _, part := range telegramReplyParts(telegramUnescapeTags(markdown)) {
b.deliver(ctx, part)
}
if !retry {
b.logger.Printf("reply lost: %v", err)
+
return
}
}
o.logger.Printf("%s after %s", o.back, durationText(time.Since(o.start)))
+
o.start = time.Time{}
}
}
var file TelegramOffsetFile
+
err = json.Unmarshal(data, &file)
if err != nil {
return 0, fmt.Errorf("%s: %w", path, err)
// never cuts a character.
func telegramSplitMessage(text string, limit int) []string {
var parts []string
+
for {
end := telegramPrefixEnd(text, limit)
if end == len(text) {
// not cut an entity of telegramEscapeTags. It never cuts a character.
func telegramSplitRich(markdown string, limit int) []string {
var parts []string
+
for len(markdown) > limit {
cut, fence, opening := telegramRichCut(markdown, limit)
if fence == "" {
// so each cut makes the rest shorter.
func telegramRichCut(markdown string, limit int) (int, string, string) {
blank, line, code := 0, 0, 0
+
var fence, opening, inFence, inOpening string
+
for start := 0; start < limit; {
end := strings.IndexByte(markdown[start:], '\n')
if end < 0 || start+end+1 > limit {
// An entity that crosses a cut becomes one entity in each part.
func telegramSplitEntities(text string, entities []TelegramEntity, limit int) []TelegramPart {
var parts []TelegramPart
+
start := 0
for _, part := range telegramSplitMessage(text, limit) {
end := start + telegramUTF16Units(part)
+
var kept []TelegramEntity
+
for _, entity := range entities {
from := max(entity.Offset, start)
to := min(entity.Offset+entity.Length, end)
// the Bot API asks for, or 0.
func telegramRetryable(err error) (bool, time.Duration) {
var network *TelegramNetworkError
+
if errors.As(err, &network) {
return true, 0
}
var answer *TelegramError
+
if errors.As(err, &answer) {
return answer.Status == http.StatusTooManyRequests || answer.Status >= 500, answer.RetryAfter
}
// tell this, so it is not a refusal.
func telegramRejected(err error) bool {
var answer *TelegramError
+
return errors.As(err, &answer) && answer.Refused && (answer.Status == http.StatusBadRequest || answer.Status == http.StatusNotFound)
}
// messages are asked for.
func (b *TelegramAPI) GetUpdates(ctx context.Context, offset int64) ([]TelegramUpdate, error) {
var updates []TelegramUpdate
+
err := b.call(ctx, telegramPollTimeout, "getUpdates", map[string]any{
"offset": offset,
"timeout": telegramPollSeconds,
}
request.Header.Set("Content-Type", "application/json")
+
response, err := b.http.Do(request)
if err != nil {
return b.networkError("no answer", err, timeout)
}
var answer TelegramAnswer
+
err = json.Unmarshal(data, &answer)
if err != nil && response.StatusCode != http.StatusOK {
return &TelegramError{
// cause is kept.
func (b *TelegramAPI) redact(err error) string {
var urlError *url.Error
+
if errors.As(err, &urlError) {
err = urlError.Err
}
api.sendTimeout = time.Hour
ctx, cancel := context.WithCancel(t.Context())
time.AfterFunc(50*time.Millisecond, cancel)
+
start = time.Now()
err = tc.send(ctx, api)
require.Error(t, err)
name: "rate limit waits as asked",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusTooManyRequests)
+
_, _ = io.WriteString(w, `{"ok":false,"error_code":429,"description":"Too Many Requests: retry after 3","parameters":{"retry_after":3}}`)
},
retry: true,
name: "server error page is retried",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusBadGateway)
+
_, _ = io.WriteString(w, "<html>bad gateway</html>")
},
retry: true,
name: "token echoed in a description is hidden",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusUnauthorized)
+
_, _ = io.WriteString(w, `{"ok":false,"error_code":401,"description":"Unauthorized: bot`+telegramTestToken+`"}`)
},
retry: false,
name: "bad request is a refusal",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusBadRequest)
+
_, _ = io.WriteString(w, `{"ok":false,"error_code":400,"description":"Bad Request: can't parse"}`)
},
retry: false,
name: "unknown method is a refusal",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusNotFound)
+
_, _ = io.WriteString(w, `{"ok":false,"error_code":404,"description":"Not Found"}`)
},
retry: false,
name: "not found page of a proxy is not a refusal",
reply: func(w http.ResponseWriter) {
w.WriteHeader(http.StatusNotFound)
+
_, _ = io.WriteString(w, "<html>not found</html>")
},
retry: false,
// Add makes update available to getUpdates.
func (f *TelegramFake) Add(update TelegramUpdate) {
f.mutex.Lock()
+
f.updates = append(f.updates, update)
f.mutex.Unlock()
// sendRichMessage calls that succeeded, in order.
func (f *TelegramFake) Sent() []string {
var texts []string
+
for _, call := range f.Calls() {
switch {
case !call.ok:
method, found := strings.CutPrefix(r.URL.Path, "/bot"+telegramTestToken+"/")
if !found {
http.Error(w, `{"ok":false,"error_code":404,"description":"Not Found"}`, http.StatusNotFound)
+
return
}
var params map[string]any
+
err := json.NewDecoder(r.Body).Decode(¶ms)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
+
return
}
}
f.mutex.Lock()
+
status, description := 0, ""
if f.fail != nil {
status, description = f.fail(call)
if stall {
f.record(call)
+
select {
case <-r.Context().Done():
case <-f.stopped:
if status != 0 {
f.record(call)
w.WriteHeader(status)
+
wait := 0
if status == http.StatusTooManyRequests {
wait = 1
call.ok = true
f.record(call)
+
if method != "getUpdates" {
_, _ = io.WriteString(w, `{"ok":true,"result":true}`)
return
defer f.mutex.Unlock()
var updates []TelegramUpdate
+
kept := f.updates[:0]
for _, update := range f.updates {
if update.UpdateID >= offset {
// change the rest of the reply.
func telegramMarkdownEntities(markdown string) (string, []TelegramEntity) {
var out TelegramMarkdownText
+
lines := strings.Split(markdown, "\n")
for index := 0; index < len(lines); index++ {
if index > 0 {
// add appends text.
func (m *TelegramMarkdownText) add(text string) {
m.text.WriteString(text)
+
m.units += telegramUTF16Units(text)
}
for index < len(line) {
if line[index] == '\\' && index+1 < len(line) && strings.IndexByte(telegramMarkdownPunctuation, line[index+1]) >= 0 {
m.add(line[plain:index])
+
plain = index + 1
index += 2
}
m.add(line[plain:index])
+
start := m.units
if span.kind == "code" {
m.add(span.content)
// close runs to the end, as in Markdown.
func telegramRewriteProse(markdown string, rewrite func(string) string) string {
var out strings.Builder
+
lines := strings.Split(markdown, "\n")
for index := 0; index < len(lines); index++ {
if index > 0 {
_, fence := telegramCodeFence(line)
if fence == "" {
telegramRewriteLineProse(&out, line, rewrite)
+
continue
}
}
out.WriteString(strings.Join(lines[index:end+1], "\n"))
+
index = end
}
out.WriteString(rewrite(line[prose:index]))
out.WriteString(line[index:end])
+
index = end
prose = end
telegramWaitForSent(t, telegram, telegramTextOnly)
var sends []TelegramFakeCall
+
for _, call := range telegram.Calls() {
if call.method == "sendMessage" || call.method == "sendRichMessage" {
sends = append(sends, call)
})
var methods []string
+
for _, call := range telegram.Calls() {
methods = append(methods, call.method)
}
})
var fallback, rich []string
+
for _, call := range telegram.Calls() {
switch {
case !call.ok:
})
telegramWaitForSent(t, telegram, "ok")
+
lines := strings.Split(strings.TrimSuffix(logs.String(), "\n"), "\n")
require.Len(t, lines, 2)
require.Equal(t, "receive failed: 502 Bad Gateway (retry in 1ms)", lines[0])
}
var dialer net.Dialer
+
conn, err := dialer.DialContext(ctx, "unix", path)
if err != nil {
return fmt.Errorf(`connect to the daemon (start it with "onechat daemon"): %w`, err)
}()
var err error
+
select {
case err = <-sent:
err = errors.Join(err, conn.Close())
+
<-shown
case err = <-shown:
err = errors.Join(terminalCmpError(err, errTerminalDaemonClosed), conn.Close())
case <-ctx.Done():
err = conn.Close()
+
<-shown
}
encoder := json.NewEncoder(conn)
scanner := bufio.NewScanner(in)
scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
+
for ctx.Err() == nil && scanner.Scan() {
line := scanner.Text()
request := SocketRequest{
func terminalShowEvents(conn net.Conn, out io.Writer) error {
scanner := bufio.NewScanner(conn)
scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
+
for scanner.Scan() {
var event Event
+
err := json.Unmarshal(scanner.Bytes(), &event)
if err != nil {
return fmt.Errorf("decode event: %w", err)
// run runs one tool and returns its output, or its error as text.
func (b *Toolbox) run(ctx context.Context, name, arguments string) string {
- var output string
- var err error
+ var (
+ output string
+ err error
+ )
+
switch name {
case "zoom":
var args ToolZoomArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case err != nil:
}
case "date":
var args ToolDateArguments
+
err = toolDecodeArguments(arguments, &args)
if err == nil {
output, err = b.date(args.ID)
}
case "shell":
var args ToolShellArguments
+
err = toolDecodeArguments(arguments, &args)
if err == nil {
output = b.shell(ctx, args.Command)
}
case "read":
var args ToolPathArguments
+
err = toolDecodeArguments(arguments, &args)
if err == nil {
output, err = b.read(args.Path)
}
case "write":
var args ToolWriteArguments
+
err = toolDecodeArguments(arguments, &args)
if err == nil {
output, err = b.write(args.Path, args.Text)
}
case "edit":
var args ToolEditArguments
+
err = toolDecodeArguments(arguments, &args)
if err == nil {
output, err = b.edit(args.Path, args.Old, args.New)
}
case "spawn":
var args ToolSpawnArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.runs == nil:
}
case "tell":
var args ToolTellArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.runs == nil:
}
case "whatsapp_find":
var args ToolWhatsAppFindArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.whatsapp == nil:
}
case "whatsapp_send":
var args ToolWhatsAppSendArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.whatsapp == nil:
}
case "schedule_set":
var args ToolScheduleSetArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.schedules == nil:
}
case "schedule_remove":
var args ToolScheduleRemoveArguments
+
err = toolDecodeArguments(arguments, &args)
switch {
case b.schedules == nil:
}
level := bits.TrailingZeros(uint(n)) - 1
+
var lines []string
+
for _, index := range []int{2 * id / n, 2*id/n + 1} {
node, built, err := b.chat.Node(level, index)
if err != nil {
err := cmd.Run()
text := output.String()
+
var exit *exec.ExitError
+
switch {
case ctx.Err() != nil:
text += "\n[interrupted]"
func toolDecodeArguments(arguments string, args any) error {
decoder := json.NewDecoder(strings.NewReader(arguments))
decoder.DisallowUnknownFields()
+
err := decoder.Decode(args)
if err != nil {
return fmt.Errorf("arguments: %w", err)
start := time.Now()
output := box.shell(ctx, "echo started; sleep 30 & sleep 30")
+
require.Less(t, time.Since(start), 10*time.Second)
require.Equal(t, "started\n\n[interrupted]", output)
}
for level, index := range t.levels {
err = errors.Join(err, index.Close())
+
delete(t.levels, level)
}
}
var node Node
+
err = json.Unmarshal(line, &node)
if err != nil {
return Node{}, false, fmt.Errorf("decode node (%d, %d): %w", level, index, err)
func (t *Tree) open(ctx context.Context, total int) error {
var err error
+
t.order, err = openRecords(filepath.Join(t.dir, "tree.idx"), treeRecordBytes)
if err != nil {
return err
return t.journal.Scan(ctx, start, func(line []byte, place Position) error {
var node Node
+
err := json.Unmarshal(line, &node)
if err != nil {
return fmt.Errorf("decode tree node: %w", err)
}()
require.Equal(t, len(nodes), tree.Count())
+
for _, node := range nodes {
require.Equal(t, node, treeReadNode(t, tree, node.Level, node.Index))
}
}()
require.Equal(t, len(nodes), tree.Count())
+
for _, node := range nodes {
require.Equal(t, node, treeReadNode(t, tree, node.Level, node.Index))
}
}
l.mutex.Lock()
+
l.pending = append(l.pending, message)
l.mutex.Unlock()
}
l.mutex.Lock()
+
l.pending = append(l.pending, messages...)
l.mutex.Unlock()
// fail stops the loop with err.
func (l *TurnLoop) fail(err error) {
l.mutex.Lock()
+
l.failure = errors.Join(l.failure, err)
l.mutex.Unlock()
defer cancel()
l.mutex.Lock()
+
l.cancel = cancel
l.mutex.Unlock()
defer func() {
l.mutex.Lock()
+
l.cancel = nil
l.mutex.Unlock()
}()
require.Equal(t, []string{"user", "busy", "tool", "echo", "reply", "idle"}, test.events.Kinds())
var turns []string
+
for _, record := range modelReadUsage(t, test.usagePath) {
turns = append(turns, record.Turn)
}
require.NoError(t, err)
<-arrived
+
_, err = test.loop.Submit(terminalSource, "second, during the turn")
require.NoError(t, err)
close(release)
test.events.WaitFor(t, "idle", 1)
+
input := server.Call(1).body["input"].([]any)
require.Equal(t, map[string]any{
"role": "user",
require.NoError(t, err)
<-arrived
+
_, err = test.loop.Submit(terminalSource, "second")
require.NoError(t, err)
close(release)
first := server.Call(1).body["input"].([]any)
second := server.Call(2).body["input"].([]any)
+
require.Len(t, first, 4)
require.Len(t, second, 5)
require.Equal(t, first[:2], second[:2])
t.Helper()
var lines []string
+
for id := range l.memory.Len() {
message, err := l.memory.Message(id)
require.NoError(t, err)
require.NoError(t, json.Unmarshal(call.raw, &body))
var texts []string
+
for _, item := range body.Input {
if item.Role != "user" {
continue
}
var text strings.Builder
+
for _, part := range item.Content {
text.WriteString(part.Text)
}
whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), clock)
test.loop, err = newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, clock.Now), client, clock.Now, log.New(test.logs, "", 0), test.events.Add)
require.NoError(t, err)
+
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
go func() {
}
var text strings.Builder
+
v.renderLine(&text, line)
+
lines = append(lines, text.String())
}
// sees a pending mark or a part of a message.
func (v *View) Context(end int) string {
var text strings.Builder
+
text.WriteString("<chat>\n")
+
for _, line := range v.lines {
_, built := v.nodes[line]
if !built || nodeEnd(line.level, line.index) > end {
size += v.lineSize(parent) - v.lineSize(left) - v.lineSize(right)
delete(v.nodes, left)
delete(v.nodes, right)
+
v.lines[best] = parent
v.lines = append(v.lines[:best+1], v.lines[best+2:]...)
merged = true
// earlier messages up to the first one whose text differs, and puts the rest
// of the lines into one new message.
func (l *ViewLayout) Split(lines []string) ([]string, error) {
- var texts []string
- var kept []ViewLayoutMessage
+ var (
+ texts []string
+ kept []ViewLayoutMessage
+ )
+
start := 0
for _, message := range l.Messages {
if start+message.Lines > len(lines) {
func TestViewMergesInTheOrderOfTheRollbackPush(t *testing.T) {
tree := ViewBuiltTree{}
+
view := newView(0, 0)
+
var states *ViewPushEntry
+
for tick := range 20001 {
states = viewPush(tick, states)
starts := states.starts()
func TestViewMergesInOneBatchFromHighToLow(t *testing.T) {
tree := ViewBuiltTree{}
+
view := newView(10, 5)
sizes := make([]int, 0, 20)
for range 20 {
_, err := view.Append(tree)
require.NoError(t, err)
+
sizes = append(sizes, view.Size())
}
// starts returns the states of the list, oldest first.
func (e *ViewPushEntry) starts() []int {
var starts []int
+
for entry := e; entry != nil; entry = entry.older {
starts = append(starts, entry.state)
}
}
var err error
+
w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow)
if err != nil {
return nil, err
if w.device.ID != nil {
w.linked = true
w.inbox.waitForCatchup()
+
w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
}
// would be lost.
func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) error {
w.mutex.Lock()
+
w.notify = notify
linked := w.linked
w.retrying = linked
defer timer.Stop()
var retry <-chan time.Time
+
delay := whatsappRetryDelay
connect := func() {
retry = nil
if w.connect(ctx) {
timer.Reset(delay)
+
retry = timer.C
delay = min(2*delay, whatsappRetryLimit)
}
stopRetry := func() {
timer.Stop()
+
retry = nil
+
w.mutex.Lock()
+
w.retrying = false
w.mutex.Unlock()
}
var pairing sync.WaitGroup
+
cancel := func() {}
defer func() {
cancel()
connect()
case <-w.logouts:
stopRetry()
+
err = w.loggedOut(ctx, l)
if err != nil {
return err
case request := <-w.requests:
cancel()
pairing.Wait()
+
cancel = func() {}
switch {
case request.unlink:
stopRetry()
+
err = w.unlink(ctx, l, request.answer)
if err != nil {
return err
switch {
case len(fields) == 1 && fields[0] == "status":
answer(w.Status())
+
return
case len(fields) == 1 && fields[0] == "unlink":
request.unlink = true
phone, valid := whatsappPhone(strings.Join(fields[1:], ""))
if !valid {
answer("PHONE must be the number of your WhatsApp account with its country code: 7 to 15 digits, such as +15550001111.")
+
return
}
request.phone = phone
default:
answer(whatsappUsage)
+
return
}
// authentication.
func (w *WhatsApp) connect(ctx context.Context) bool {
w.mutex.Lock()
+
if ctx.Err() != nil || !w.linked || !w.retrying || w.closed {
w.mutex.Unlock()
+
return false
}
w.mutex.Unlock()
err := client.ConnectContext(ctx)
+
w.mutex.Lock()
+
if err == nil || ctx.Err() != nil || !w.retrying || !w.linked {
w.retrying = false
w.mutex.Unlock()
first := w.startOutage()
cause := whatsappRedact(w.own, err.Error())
w.mutex.Unlock()
+
if first {
w.logger.Printf("first connect failed, retrying: %s", cause)
}
// close ends the connection and the timer of the notice, and stops all notices.
func (w *WhatsApp) close() {
w.mutex.Lock()
+
w.closed = true
if w.timer != nil {
w.timer.Stop()
}
w.logger.Print("logged out by the phone, run /whatsapp pair PHONE")
+
err := w.inbox.deliver(ctx, l, w.link(), true)
if err != nil {
return err
// file.
func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text string)) error {
w.mutex.Lock()
+
linked := w.linked
client := w.client
w.mutex.Unlock()
if !linked {
answer("WhatsApp is not linked.")
+
return nil
}
}
var logoutErr error
+
if client != nil {
ctx, cancel := context.WithTimeout(ctx, whatsappLogoutTime)
defer cancel()
err = w.removeSession()
if err != nil {
answer("The unlink failed: " + w.redact(err.Error()))
+
return nil
}
w.logger.Print("unlinked")
+
err = l.note("WhatsApp is unlinked.")
if err != nil {
w.logger.Printf("chat note failed: %v", err)
if client == nil || logoutErr != nil {
answer("WhatsApp is unlinked here, but the phone did not get the logout. In WhatsApp on your phone, open Linked devices and remove the device.")
+
return nil
}
// its files.
func (w *WhatsApp) removeSession() error {
w.mutex.Lock()
+
client := w.client
w.client = nil
w.linked = false
}
var err error
+
if container != nil {
err = container.Close()
}
device, err := w.newDevice(ctx)
if err != nil {
answer("The pairing failed: " + err.Error())
+
return
}
if err != nil {
answer("The connection to WhatsApp failed: " + err.Error())
+
return
}
case item, open := <-qr:
if !open || item.Event != whatsmeow.QRChannelEventCode {
answer("WhatsApp did not start the pairing. Try again later.")
+
return
}
case <-ctx.Done():
code, err := client.PairPhone(ctx, phone, true, whatsmeow.PairClientChrome, "Chrome (Linux)")
if err != nil {
answer("WhatsApp refused the pairing: " + err.Error() + ". Check the phone number and run /whatsapp pair PHONE again.")
+
return
}
for {
var item whatsmeow.QRChannelItem
+
open := true
select {
case item, open = <-qr:
switch {
case !open || item.Event == whatsmeow.QRChannelTimeout.Event:
answer(whatsappPairExpired)
+
return
case item.Event == whatsmeow.QRChannelEventCode:
continue
case item.Event == whatsmeow.QRChannelSuccess.Event:
linked = true
+
w.linkedBy(client)
w.logger.Print("linked")
return
default:
answer("The pairing failed: " + whatsappItemError(item) + ". Run /whatsapp pair PHONE again.")
+
return
}
}
return true
case *events.JoinedGroup:
w.groups.set(event.JID.String(), event.Name)
+
return true
case *events.OfflineSyncCompleted:
w.inbox.endCatchup()
+
return true
case *events.Connected:
w.inbox.startCatchup()
}
var notice, line string
+
w.mutex.Lock()
+
switch event := event.(type) {
case *events.PairSuccess:
w.own = whatsappOwnIDs(event.ID.User, event.LID.User)
// disconnection with the number disconnections ended.
func (w *WhatsApp) disconnectedLong(disconnections int) {
w.mutex.Lock()
+
long := !w.closed && w.linked && !w.connected && !w.noticed && w.disconnections == disconnections
if long {
w.noticed = true
// broadcast gives text as a notice to all clients.
func (w *WhatsApp) broadcast(text string) {
w.mutex.Lock()
+
notify := w.notify
closed := w.closed
w.mutex.Unlock()
// name OneChat and a desktop icon only when the platform type is known.
func whatsappSetDeviceProps() {
store.SetOSInfo("OneChat", [3]uint32{0, 1, 0})
+
store.DeviceProps.PlatformType = waCompanionReg.DeviceProps_DESKTOP.Enum()
}
// whatsappOwnIDs returns the own number and the own LID that are not empty.
func whatsappOwnIDs(ids ...string) []string {
var own []string
+
for _, id := range ids {
if id != "" {
own = append(own, id)
// ConnectContext keeps ctx and tells the test that the daemon connected.
func (f *WhatsAppFake) ConnectContext(ctx context.Context) error {
f.mutex.Lock()
+
f.connections = append(f.connections, ctx)
f.connectTimes = append(f.connectTimes, time.Now())
f.mutex.Unlock()
// Emit gives event to the handler of the daemon and returns its answer.
func (f *WhatsAppFake) Emit(event any) bool {
f.mutex.Lock()
+
handle := f.handle
f.mutex.Unlock()
}, 10*time.Second, time.Millisecond)
var texts []string
+
for id := range memory.Len() {
message, err := memory.Message(id)
require.NoError(t, err)
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
whatsapp.state.quiet = 50 * time.Millisecond
contacts := whatsapp.device.Contacts
require.NoError(t, contacts.PutContactName(t.Context(), whatsappTestAna, "Ana", "Ana Souza"))
require.NoError(t, contacts.PutContactName(t.Context(), types.NewJID("15550004444", types.DefaultUserServer), "João", "João Silva"))
+
_, _, err := contacts.PutBusinessName(t.Context(), types.NewJID("15550005555", types.DefaultUserServer), "Corner Bakery")
require.NoError(t, err)
require.NoError(t, contacts.PutContactName(t.Context(), whatsappTestOwn, "Me", "Me Souza"))
}
var matches []WhatsAppMatch
+
for jid, contact := range contacts {
names := []string{contact.FullName, contact.FirstName, contact.BusinessName, contact.PushName}
shown := slices.IndexFunc(names, func(name string) bool {
})
var lines []string
+
for _, match := range matches[:min(len(matches), whatsappFindMatches)] {
lines = append(lines, match.id+" | "+match.name+" | "+match.kind)
}
func TestAgentFindsAWhatsAppContact(t *testing.T) {
whatsapp, fake := whatsappStartTestLinked(t)
whatsappTestContacts(t, whatsapp)
+
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "whatsapp_find", `{"name":"ana"}`)}, 100, 0, 5))
+
return
}
// again, at most once in whatsappGroupsRefresh.
func (g *WhatsAppGroups) name(ctx context.Context, link WhatsAppLink, id string) string {
g.mutex.Lock()
+
name, found := g.names[id]
fresh := time.Since(g.triedAt) < whatsappGroupsRefresh
g.mutex.Unlock()
g.read(ctx, link)
g.mutex.Lock()
+
name, found = g.names[id]
g.mutex.Unlock()
}
}
var state WhatsAppStateFile
+
err = json.Unmarshal(data, &state)
if err != nil {
return nil, "", fmt.Errorf("decode pending import names: %w", err)
}
var err error
+
plan.added, err = rebuildMerge(nil, plan.added)
if err != nil {
return plan, fmt.Errorf("order import batches: %w", err)
}()
var file *zip.File
+
for _, entry := range archive.File {
if filepath.Base(entry.Name) == "_chat.txt" && !entry.FileInfo().IsDir() {
if file != nil {
scanner := bufio.NewScanner(io.LimitReader(reader, whatsappImportBytes+1))
scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
+
for scanner.Scan() {
line := strings.ReplaceAll(scanner.Text(), "\u200e", "")
match := whatsappImportStart.FindStringSubmatch(line)
}})
before := rebuildTestSnapshot(t, agent.dir)
+
var report strings.Builder
+
require.NoError(t, whatsappImport(t.Context(), []string{"-n", whatsappImportSample(t)}, &report))
require.Equal(t, before, rebuildTestSnapshot(t, agent.dir))
agent := newRebuildTestAgent(t)
dir := filepath.Dir(agent.dir)
t.Setenv("ONECHAT_DIR", dir)
+
path := whatsappImportSample(t)
plan, err := whatsappImportPlan(t.Context(), []string{path}, nil, nil, "", time.UTC)
require.NoError(t, err)
require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte("not model settings"), 0o600))
before := rebuildTestSnapshot(t, agent.dir)
+
var report strings.Builder
+
require.NoError(t, whatsappImport(t.Context(), []string{path}, &report))
require.Contains(t, report.String(), "batches: 0")
require.Contains(t, report.String(), "estimated calls: 0")
func TestWhatsAppImportReadsOnlyLocalSessionNames(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
session := &WhatsApp{
dir: filepath.Join(dir, whatsappDirName),
}
// first. With all, it closes and logs all batches also during a catch-up.
func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppLink, all bool) error {
i.mutex.Lock()
+
catchingUp := i.catchingUp
caughtUp := i.caughtUp
i.caughtUp = false
i.mutex.Unlock()
var err error
+
switch {
case all || caughtUp:
err = i.state.CloseAll()
}
var turns []Message
+
for _, batch := range i.state.Closed() {
if batch.Chat.Group && batch.Chat.Name == "" {
batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
// The catch-up holds the batch open until its end, so the two messages
// form one batch also on a slow machine.
require.True(t, fake.Emit(&events.Connected{}))
+
at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "Are you at home?", at)))
require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestOwn, "m2", "Yes", at)))
text := "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n" + tc.line
if tc.turn {
text += "\n" + whatsappBatchEmptyReply
+
client.WaitFor(t, "[done]")
require.Equal(t, 1, server.Calls())
} else {
<-fake.connects
require.True(t, fake.Emit(&events.Connected{}))
+
at := time.Date(2026, 1, 1, 8, 0, 0, 0, time.UTC)
for index, event := range []*events.Message{
whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "Good morning", at),
time.Sleep(100 * time.Millisecond)
require.Equal(t, 1, server.Calls())
+
for _, text := range []string{"Good morning", "Call me", "Are you there?"} {
require.Contains(t, string(server.Call(0).raw), text)
}
func TestWhatsAppStateOfTheStateFileWaitForTheCatchup(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
before, _, _ := whatsappStartTest(t, dir, newTestClock())
_, err := before.state.Add(whatsappTestChat, WhatsAppMessage{
ID: "m1",
func TestWhatsAppSendThatFails(t *testing.T) {
whatsapp, fake := whatsappStartTestLinked(t)
fake.sendErr = errors.New("server error for " + whatsappTestNumber)
+
daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
whatsappTestConnect(t, fake)
require.EqualError(t, err, "WhatsApp is disconnected, so nothing was sent: try again later")
require.True(t, fake.Emit(&events.Connected{}))
+
answer, err := whatsapp.Send(t.Context(), "keys", whatsappTestAna.String(), "hi")
require.NoError(t, err)
require.Equal(t, "Sent to +15550002222 at 12:05.", answer)
func TestAgentSendsAWhatsAppMessage(t *testing.T) {
whatsapp, fake := whatsappStartTestLinked(t)
whatsappTestContacts(t, whatsapp)
+
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
if call.number == 0 {
modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "whatsapp_send", `{"id":"keys","chat":"15550002222@s.whatsapp.net","text":"The keys are under the mat."}`)}, 100, 0, 5))
+
return
}
defer s.mutex.Unlock()
var closed []WhatsAppBatch
+
for _, batch := range s.file.Batches {
if batch.Closed {
closed = append(closed, batch)
defer s.mutex.Unlock()
var first time.Time
+
for _, batch := range s.file.Batches {
deadline := s.deadline(batch)
if !batch.Closed && (first.IsZero() || deadline.Before(first)) {
}
require.NoError(t, batches.CloseAll())
+
var sizes []int
+
for _, batch := range batches.Closed() {
sizes = append(sizes, len(batch.Messages))
}
again := whatsappTestState(t, path)
require.NoError(t, again.CloseAll())
+
closed := again.Closed()
require.Len(t, closed, 1)
require.Equal(t, []string{"hi a", "hi b"}, []string{closed[0].Messages[0].Text, closed[0].Messages[1].Text})
func TestWhatsAppStatusFollowsTheConnection(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
clock := newTestClock()
whatsapp, fake, _ := whatsappStartTest(t, dir, clock)
daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
require.Equal(t, "WhatsApp is linked and connected since 2026-01-01 12:00.", whatsapp.Status())
clock.now = clock.now.Add(20 * time.Minute)
+
require.True(t, fake.Emit(&events.Disconnected{}))
+
clock.now = clock.now.Add(time.Minute)
+
require.True(t, fake.Emit(&events.Disconnected{}))
require.Equal(t, "WhatsApp is linked and disconnected since 2026-01-01 12:20. The last connection started at 2026-01-01 12:00.", whatsapp.Status())
}
require.True(t, test.fake.Emit(&events.Disconnected{}))
delays := []time.Duration{2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 32 * time.Second, 64 * time.Second, 128 * time.Second, 256 * time.Second, 5 * time.Minute, 5 * time.Minute}
+
var outage time.Duration
+
for index, delay := range delays {
time.Sleep(delay - time.Nanosecond)
synctest.Wait()
time.Sleep(time.Nanosecond)
synctest.Wait()
require.Len(t, test.fake.Connections(), index+2)
+
outage += delay
}
require.Equal(t, "WhatsApp is linked already. Run /whatsapp unlink first.", <-answers)
synctest.Wait()
test.fake.mutex.Lock()
+
test.fake.connectErr = nil
test.fake.mutex.Unlock()
time.Sleep(whatsappRetryDelay)
require.False(t, test.whatsapp.isLinked())
test.fake.mutex.Lock()
+
test.fake.connectErr = nil
test.fake.mutex.Unlock()
test.whatsapp.Command("pair 15550001111", func(_ string) {
})
+
items := <-test.fake.pairings
items <- whatsmeow.QRChannelItem{Event: whatsmeow.QRChannelEventCode}
+
synctest.Wait()
require.True(t, test.fake.Emit(&events.PairSuccess{
ID: types.NewADJID(whatsappTestNumber, 0, 1),
LID: types.NewADJID(whatsappTestLID, 0, 1),
}))
+
items <- whatsmeow.QRChannelSuccess
+
synctest.Wait()
require.True(t, test.whatsapp.isLinked())
time.Sleep(10 * time.Minute)
test.fake.connectErr = errors.New("network unavailable")
test.start(t)
synctest.Wait()
+
failure := errors.New("state write failed")
test.whatsapp.inbox.failures <- failure
+
require.ErrorIs(t, <-test.done, failure)
time.Sleep(10 * time.Minute)
synctest.Wait()
test.fake.connectErr = errors.New("network unavailable")
test.start(t)
synctest.Wait()
+
since := time.Now().Format(whatsappTimeLayout)
+
time.Sleep(whatsappNoticeDelay)
synctest.Wait()
require.Equal(t, 1, test.events.Count("notice"))
synctest.Wait()
require.Equal(t, 1, test.events.Count("notice"))
test.fake.mutex.Lock()
+
test.fake.connectErr = nil
test.fake.mutex.Unlock()
time.Sleep(whatsappRetryLimit)
func TestWhatsAppStartingRetriesDoNotDuplicateLibraryRetries(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, _, _ := whatsappStartTest(t, dir, newTestClock())
client, ok := whatsappNewClient(whatsapp.device, whatsapp.handle).(*whatsmeow.Client)
require.True(t, ok)
func TestWhatsAppLogsEachChangeOfTheConnectionOnce(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
clock := newTestClock()
whatsapp, fake, logs := whatsappStartTest(t, dir, clock)
daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
<-fake.connects
require.True(t, fake.Emit(&events.Connected{}))
+
clock.now = clock.now.Add(time.Minute)
+
require.True(t, fake.Emit(&events.Disconnected{}))
+
clock.now = clock.now.Add(time.Minute)
+
require.True(t, fake.Emit(&events.Disconnected{}))
+
clock.now = clock.now.Add(25 * time.Second)
+
require.True(t, fake.Emit(&events.Connected{}))
require.Equal(t, []string{
t.Run(tc.name, func(t *testing.T) {
dir := t.TempDir()
tc.session(t, dir)
+
whatsapp, _, _ := whatsappStartTest(t, dir, newTestClock())
require.Equal(t, "WhatsApp is not linked. Run /whatsapp pair PHONE to link it.", whatsapp.Status())
})
other.WaitFor(t, "[connected to OneChat]")
asking.Type("/whatsapp pair +1 555-000-1111")
+
items := <-fake.pairings
items <- whatsmeow.QRChannelItem{
Event: whatsmeow.QRChannelEventCode,
ID: types.NewADJID(whatsappTestNumber, 0, 1),
LID: types.NewADJID(whatsappTestLID, 0, 1),
}))
+
items <- whatsmeow.QRChannelSuccess
+
asking.WaitFor(t, "WhatsApp is linked to +15550001111.")
// The library connects again after the link with the context of the
client.Type("/whatsapp pair 15550001111")
<-fake.pairings
client.Type("/whatsapp pair 15550002222")
+
items := <-fake.pairings
items <- whatsmeow.QRChannelItem{
Event: whatsmeow.QRChannelEventCode,
func TestWhatsAppPairWhenLinked(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, _, _ := whatsappStartTest(t, dir, newTestClock())
socket, _ := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
client := terminalStartTestClient(t, socket)
t.Run(tc.name, func(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
fake.logoutErr = tc.logoutErr
socket, memory := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
client := terminalStartTestClient(t, socket)
+
<-fake.connects
client.Type("/whatsapp unlink")
func TestWhatsAppLogoutFromThePhone(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
<-fake.connects
require.True(t, fake.Emit(&events.LoggedOut{}))
+
notice := "WhatsApp was unlinked from the phone. Run /whatsapp pair PHONE to link it again."
client.WaitFor(t, notice)
telegramWaitForSent(t, telegram, notice)
func TestWhatsAppLongDisconnectionGivesOneNotice(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
clock := newTestClock()
whatsapp, fake, _ := whatsappStartTest(t, dir, clock)
whatsapp.noticeDelay = 300 * time.Millisecond
require.Equal(t, "[connected to OneChat]\n", client.Output())
clock.now = clock.now.Add(time.Hour)
+
require.True(t, fake.Emit(&events.Disconnected{}))
require.True(t, fake.Emit(&events.Disconnected{}))
client.WaitFor(t, "WhatsApp is disconnected since 2026-01-01 13:00. OneChat tries again.\n")
func TestWhatsAppDisconnectionWithNoRetryGivesANoticeAtOnce(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
socket, _ := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
client := terminalStartTestClient(t, socket)
func TestWhatsAppLogsNoOwnNumber(t *testing.T) {
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock())
fake.connectErr = errors.New("no route for " + whatsappTestNumber + "@s.whatsapp.net and " + whatsappTestLID + "@lid")
+
daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
<-fake.connects
dir := t.TempDir()
whatsappTestSession(t, dir, true)
+
whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock())
whatsapp.clock = time.Now
agent, err := openAgent(t.TempDir())
require.NoError(t, err)
+
memory, err := openMemory(t.Context(), agent.dir, time.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
t.Cleanup(func() {
projects / onechat / commitdiff