From 522d7fd117285e4e4ce3ea5ec9b5a1722521b681 Mon Sep 17 00:00:00 2001 From: "Anderson A. Giacomolli" Date: Mon, 9 Nov 2026 18:14:59 -0300 Subject: [PATCH 1/1] Check spacing between Go statement blocks Wsl v5 checks empty lines around blocks and multiline declarations and calls. Spacing keeps immediate error checks and short declarations with their uses. Checks that split those groups are off because they conflict with the Go rule. The cleanup changes only spacing and groups of variable declarations. --- .golangci.yml | 30 +++++++++++++++++++++++ agent.go | 1 + agent_test.go | 1 + build_queue.go | 4 ++++ build_queue_test.go | 4 ++++ build_run.go | 6 +++++ build_run_test.go | 5 ++++ chain.go | 10 ++++++-- chain_test.go | 1 + chatgpt.go | 8 +++++++ chatgpt_fake_helper_test.go | 12 ++++++++++ chatgpt_test.go | 8 +++++++ compactor.go | 4 ++++ compactor_test.go | 5 ++++ config.go | 1 + daemon.go | 22 +++++++++++++++++ daemon_test.go | 3 +++ datadir.go | 1 + journal.go | 3 +++ log.go | 3 +++ log_test.go | 3 +++ main.go | 1 + memory.go | 10 ++++++++ memory_test.go | 9 +++++++ model.go | 10 ++++++++ model_fake_helper_test.go | 7 ++++++ model_test.go | 5 ++++ probe.go | 3 +++ probe_log_helper_test.go | 3 +++ probe_test.go | 3 +++ prompt_test.go | 1 + rebuild.go | 7 ++++++ rebuild_test.go | 3 +++ replay_test.go | 3 +++ schedule.go | 11 +++++++-- schedule_cron.go | 7 ++++++ schedule_test.go | 1 + subagent.go | 2 ++ subagent_run.go | 18 ++++++++++++-- subagent_run_test.go | 19 +++++++++++++++ subagent_test.go | 1 + summarizer_helper_test.go | 2 ++ syncer.go | 5 ++++ syncer_test.go | 8 +++++++ telegram.go | 23 ++++++++++++++++++ telegram_api.go | 7 ++++++ telegram_api_test.go | 7 ++++++ telegram_fake_helper_test.go | 10 ++++++++ telegram_markdown.go | 8 +++++++ telegram_test.go | 4 ++++ terminal.go | 7 ++++++ tools.go | 24 +++++++++++++++++-- tools_test.go | 1 + tree.go | 4 ++++ tree_test.go | 2 ++ turn.go | 5 ++++ turn_test.go | 9 +++++++ view.go | 5 ++++ view_layout.go | 7 ++++-- view_test.go | 6 +++++ whatsapp.go | 46 ++++++++++++++++++++++++++++++++++++ whatsapp_fake_helper_test.go | 5 ++++ whatsapp_find.go | 2 ++ whatsapp_find_test.go | 2 ++ whatsapp_groups.go | 2 ++ whatsapp_import.go | 4 ++++ whatsapp_import_test.go | 6 +++++ whatsapp_inbox.go | 3 +++ whatsapp_inbox_test.go | 5 ++++ whatsapp_send_test.go | 4 ++++ whatsapp_state.go | 2 ++ whatsapp_state_test.go | 3 +++ whatsapp_test.go | 43 +++++++++++++++++++++++++++++++++ 73 files changed, 520 insertions(+), 10 deletions(-) diff --git a/.golangci.yml b/.golangci.yml index d5f20b3..2a4cdaf 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -9,7 +9,37 @@ linters: - 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 diff --git a/agent.go b/agent.go index 3fc9546..0d97cdf 100644 --- a/agent.go +++ b/agent.go @@ -50,6 +50,7 @@ func openAgent(dir string) (*Agent, error) { 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") diff --git a/agent_test.go b/agent_test.go index a9418ae..be74b40 100644 --- a/agent_test.go +++ b/agent_test.go @@ -103,6 +103,7 @@ func TestAgentPromptReadsTheInstructionsEachTime(t *testing.T) { 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) diff --git a/build_queue.go b/build_queue.go index 53e804d..7db24be 100644 --- a/build_queue.go +++ b/build_queue.go @@ -82,8 +82,10 @@ func (q *BuildQueue) Next() (NodeKey, bool) { // 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 } @@ -96,6 +98,7 @@ func (q *BuildQueue) Done(key NodeKey) { // 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 @@ -107,6 +110,7 @@ func (q *BuildQueue) Fail(key NodeKey) { // Retry lets the failed nodes start again. func (q *BuildQueue) Retry() { clear(q.failed) + for _, key := range q.retry { heap.Push(&q.merges, key) } diff --git a/build_queue_test.go b/build_queue_test.go index 078e06a..34a4571 100644 --- a/build_queue_test.go +++ b/build_queue_test.go @@ -14,6 +14,7 @@ func TestQueueStartsInMessageOrder(t *testing.T) { queue.AddMerge(NodeKey{1, 1}) var order []NodeKey + for { key, ready := queue.Next() if !ready { @@ -34,16 +35,19 @@ func TestQueueRetriesAFailedMergeAtTheNextMessage(t *testing.T) { 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) } diff --git a/build_run.go b/build_run.go index 9c6b767..da75679 100644 --- a/build_run.go +++ b/build_run.go @@ -151,8 +151,10 @@ func buildReadUsage(path string) (records []ModelUsageRecord, err error) { 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) @@ -176,6 +178,7 @@ func buildReadUsage(path string) (records []ModelUsageRecord, err error) { // 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 @@ -186,6 +189,7 @@ func buildWriteReport(out io.Writer, appended int, stop string, records []ModelU } var lines []string + if stop != "" { lines = append(lines, "stopped early: "+stop) } @@ -288,6 +292,7 @@ func buildTimeText(records []ModelUsageRecord) string { } var events []event + seconds := make([]float64, 0, len(records)) total := 0.0 for _, record := range records { @@ -324,6 +329,7 @@ func buildTimeText(records []ModelUsageRecord) string { } 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)) diff --git a/build_run_test.go b/build_run_test.go index aef8922..fc8b4b7 100644 --- a/build_run_test.go +++ b/build_run_test.go @@ -82,6 +82,7 @@ func TestBuildRunStopsWhenTheQuotaGrows(t *testing.T) { 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%)") @@ -111,11 +112,14 @@ func TestBuildRunStopsAtTheCallLimit(t *testing.T) { 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() @@ -123,6 +127,7 @@ func TestBuildRunWaitsOnlyBeforeUserMessages(t *testing.T) { <-gate mutex.Lock() + running-- mutex.Unlock() diff --git a/chain.go b/chain.go index 0211787..c4a4302 100644 --- a/chain.go +++ b/chain.go @@ -47,6 +47,7 @@ func (c *Chain) Run(ctx context.Context, input []any) (string, error) { } var text string + input, text, err = c.handle(ctx, input, response) if err != nil { return "", err @@ -63,8 +64,11 @@ func (c *Chain) Run(ctx context.Context, input []any) (string, error) { // 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 { @@ -74,6 +78,7 @@ func (c *Chain) handle(ctx context.Context, input []any, response ModelResponse) input = append(input, item) var call ChainFunctionCall + err = json.Unmarshal(raw, &call) if err != nil { return nil, "", fmt.Errorf("decode output item: %w", err) @@ -144,6 +149,7 @@ type ChainFunctionCallOutput struct { // 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) diff --git a/chain_test.go b/chain_test.go index 5517620..e329c27 100644 --- a/chain_test.go +++ b/chain_test.go @@ -56,6 +56,7 @@ func TestChainReturnsTheRepliesOfTheFinalAnswer(t *testing.T) { require.Equal(t, 2, server.Calls()) var kinds []string + for id := range memory.Len() { message, err := memory.Message(id) require.NoError(t, err) diff --git a/chatgpt.go b/chatgpt.go index 7a5afc2..8849ada 100644 --- a/chatgpt.go +++ b/chatgpt.go @@ -131,6 +131,7 @@ func (a *ChatGPTAuth) Login(ctx context.Context, out io.Writer) error { defer a.mutex.Unlock() var code ChatGPTDeviceCode + err := a.postJSON(ctx, "/api/accounts/deviceauth/usercode", map[string]string{ "client_id": chatgptClientID, }, &code) @@ -183,6 +184,7 @@ func (a *ChatGPTAuth) read() (ChatGPTTokenFile, error) { } 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") @@ -199,6 +201,7 @@ func (a *ChatGPTAuth) read() (ChatGPTTokenFile, error) { // 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", @@ -255,6 +258,7 @@ func (a *ChatGPTAuth) poll(ctx context.Context, deviceID, userCode string, inter 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") @@ -272,6 +276,7 @@ func (a *ChatGPTAuth) poll(ctx context.Context, deviceID, userCode string, inter select { case <-ctx.Done(): timer.Stop() + return ChatGPTDeviceGrant{}, fmt.Errorf("wait for approval: %w", ctx.Err()) case <-timer.C: } @@ -289,6 +294,7 @@ func (a *ChatGPTAuth) exchange(ctx context.Context, grant ChatGPTDeviceGrant) (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) @@ -337,6 +343,7 @@ func (a *ChatGPTAuth) post(ctx context.Context, path, contentType string, body [ } request.Header.Set("Content-Type", contentType) + response, err := a.http.Do(request) if err != nil { return 0, fmt.Errorf("send request: %w", err) @@ -417,6 +424,7 @@ func chatgptPollInterval(raw json.RawMessage) time.Duration { // none. func chatgptAccountID(token string) string { var claims map[string]json.RawMessage + if !chatgptDecodeClaims(token, &claims) { return "" } diff --git a/chatgpt_fake_helper_test.go b/chatgpt_fake_helper_test.go index dfcf17d..ddd4c67 100644 --- a/chatgpt_fake_helper_test.go +++ b/chatgpt_fake_helper_test.go @@ -67,9 +67,11 @@ func (f *ChatGPTFake) newToken(ttl time.Duration) string { 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 } @@ -85,20 +87,24 @@ func (f *ChatGPTFake) deviceToken(w http.ResponseWriter, r *http.Request) { 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 } @@ -117,6 +123,7 @@ func (f *ChatGPTFake) oauthToken(w http.ResponseWriter, r *http.Request) { err := r.ParseForm() if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) + return } @@ -131,9 +138,11 @@ func (f *ChatGPTFake) oauthToken(w http.ResponseWriter, r *http.Request) { } 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 } @@ -141,6 +150,7 @@ func (f *ChatGPTFake) oauthToken(w http.ResponseWriter, r *http.Request) { f.refreshed = body["refresh_token"] if f.failRefresh { http.Error(w, `{"error": "invalid_grant", "token": "`+body["refresh_token"]+`"}`, http.StatusUnauthorized) + return } @@ -164,6 +174,7 @@ func newChatGPTFake(t *testing.T) *ChatGPTFake { 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) @@ -190,6 +201,7 @@ func chatgptWriteTestTokens(t *testing.T, auth *ChatGPTAuth, access, refresh str 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)) diff --git a/chatgpt_test.go b/chatgpt_test.go index a644daf..cb144fb 100644 --- a/chatgpt_test.go +++ b/chatgpt_test.go @@ -20,6 +20,7 @@ func TestLoginWritesTheTokenFile(t *testing.T) { auth := newTestChatGPTAuth(t, server) var out strings.Builder + err := auth.Login(t.Context(), &out) require.NoError(t, err) @@ -120,6 +121,7 @@ func TestAuthorizeRunsOneRefreshAtATime(t *testing.T) { chatgptWriteTestTokens(t, auth, server.token(time.Second), "refresh-0") var group sync.WaitGroup + errs := make(chan error, 8) for range 8 { group.Go(func() { @@ -129,6 +131,7 @@ func TestAuthorizeRunsOneRefreshAtATime(t *testing.T) { group.Wait() close(errs) + for err := range errs { require.NoError(t, err) } @@ -145,6 +148,7 @@ func TestAuthorizeNeedsALogin(t *testing.T) { server.failRefresh = true chatgptWriteTestTokens(t, auth, server.token(-time.Minute), "refresh-0") + before, err := os.ReadFile(auth.path) require.NoError(t, err) @@ -168,6 +172,7 @@ func TestForcedRefreshLetsTheNextCallWork(t *testing.T) { 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")) @@ -179,6 +184,7 @@ func TestModelEndpointFollowsTheConfig(t *testing.T) { config := defaultConfig() t.Setenv("OPENAI_API_KEY", modelTestKey) + address, auth, err := modelEndpoint(config, dir) require.NoError(t, err) require.Equal(t, modelChatGPTURL, address) @@ -192,6 +198,7 @@ func TestModelEndpointFollowsTheConfig(t *testing.T) { require.Equal(t, ModelAPIKey(modelTestKey), auth) t.Setenv("OPENAI_API_KEY", "") + _, _, err = modelEndpoint(config, dir) require.ErrorContains(t, err, "OPENAI_API_KEY") } @@ -203,6 +210,7 @@ func chatgptReadTokenFile(t *testing.T, path string) ChatGPTTokenFile { require.NoError(t, err) var tokens ChatGPTTokenFile + require.NoError(t, json.Unmarshal(data, &tokens)) return tokens diff --git a/compactor.go b/compactor.go index efb183f..be17c15 100644 --- a/compactor.go +++ b/compactor.go @@ -69,11 +69,13 @@ func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) { } 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 @@ -82,6 +84,7 @@ func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) { } var response ModelResponse + response, err = c.client.Create(ctx, ModelRequest{ Model: c.model.Model, Effort: c.model.Effort, @@ -157,6 +160,7 @@ func compactorTexts(view, task string) []string { 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])) } diff --git a/compactor_test.go b/compactor_test.go index f4c51f4..8c83594 100644 --- a/compactor_test.go +++ b/compactor_test.go @@ -167,6 +167,7 @@ func TestCompactorRetriesALongLineWithTheNewestNote(t *testing.T) { 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) @@ -174,6 +175,7 @@ func TestCompactorRetriesALongLineWithTheNewestNote(t *testing.T) { 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) @@ -211,6 +213,7 @@ func TestCompactorFailsWithoutALine(t *testing.T) { name: "failed call", reply: func(w http.ResponseWriter, _ ModelFakeCall) { w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, "down") }, want: "HTTP 500", @@ -249,6 +252,7 @@ func TestCompactorBuildsNodesInTheCore(t *testing.T) { server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if call.number == 0 { w.WriteHeader(http.StatusServiceUnavailable) + return } @@ -310,6 +314,7 @@ func compactorInputTexts(t *testing.T, call ModelFakeCall) []string { 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 } diff --git a/config.go b/config.go index 6e2da84..3b5172d 100644 --- a/config.go +++ b/config.go @@ -35,6 +35,7 @@ func loadConfig(dir string) (Config, error) { 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") diff --git a/daemon.go b/daemon.go index a289338..5ec220e 100644 --- a/daemon.go +++ b/daemon.go @@ -65,6 +65,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { }() whatsappSetDeviceProps() + logger := daemonLogger(out, "onechat: ") whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient) if err != nil { @@ -118,8 +119,10 @@ func daemon(ctx context.Context, out io.Writer) (err error) { 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") } @@ -154,10 +157,12 @@ func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLo } h.mutex.Lock() + h.clients[client] = true h.mutex.Unlock() h.workers.Add(2) + go h.write(client) go h.read(ctx, client, l) @@ -194,8 +199,10 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) 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) { @@ -251,6 +258,7 @@ func (h *DaemonHub) remove(client *DaemonClient) { delete(h.clients, client) client.once.Do(func() { close(client.events) + _ = client.conn.Close() }) } @@ -268,9 +276,11 @@ func (h *DaemonHub) Broadcast(event Event) { // 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() @@ -340,7 +350,9 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w } var workers sync.WaitGroup + workers.Add(2) + go func() { defer workers.Done() @@ -348,29 +360,37 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w }() 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() }() } @@ -381,7 +401,9 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w } cancel() + err = errors.Join(err, listener.Close()) + hub.CloseAll() workers.Wait() diff --git a/daemon_test.go b/daemon_test.go index 6a8cd3e..bfb4efb 100644 --- a/daemon_test.go +++ b/daemon_test.go @@ -16,6 +16,7 @@ func TestClientsSeeTheTurnAsItHappens(t *testing.T) { 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 } @@ -29,6 +30,7 @@ func TestClientsSeeTheTurnAsItHappens(t *testing.T) { 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()) @@ -121,6 +123,7 @@ func TestDaemonLogHasATimeStampOnlyOutsideTheJournal(t *testing.T) { 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()) }) diff --git a/datadir.go b/datadir.go index a16ee34..f191e5c 100644 --- a/datadir.go +++ b/datadir.go @@ -14,6 +14,7 @@ func dataDir() (string, error) { dir := os.Getenv("ONECHAT_DIR") if dir == "" { var err error + dir, err = defaultDataDir() if err != nil { return "", err diff --git a/journal.go b/journal.go index c12f7d0..bb1897c 100644 --- a/journal.go +++ b/journal.go @@ -81,6 +81,7 @@ func (j *Journal) Read(place Position) ([]byte, error) { 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) @@ -111,6 +112,7 @@ func (j *Journal) Close() error { err := j.closeWriter() for day, reader := range j.readers { err = errors.Join(err, reader.Close()) + delete(j.readers, day) } @@ -128,6 +130,7 @@ func (j *Journal) Scan(ctx context.Context, place Position, add func([]byte, Pos } slices.Sort(paths) + for n, path := range paths { day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl")) if err != nil { diff --git a/log.go b/log.go index 75080f9..6d0d478 100644 --- a/log.go +++ b/log.go @@ -111,6 +111,7 @@ func (l *Log) AppendDated(kind, text, date string) (Message, error) { } var err error + now := l.clock() if date == "" { date = now.Format(time.RFC3339Nano) @@ -176,6 +177,7 @@ func (l *Log) Message(id int) (Message, error) { } var message Message + err = json.Unmarshal(line, &message) if err != nil { return Message{}, fmt.Errorf("decode message %d: %w", id, err) @@ -206,6 +208,7 @@ func (l *Log) catchUp(ctx context.Context) error { 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) diff --git a/log_test.go b/log_test.go index e8e07ac..5b30426 100644 --- a/log_test.go +++ b/log_test.go @@ -16,6 +16,7 @@ func TestLogReopenLoadsTheSameMessages(t *testing.T) { 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) @@ -35,6 +36,7 @@ func TestLogReopenLoadsTheSameMessages(t *testing.T) { }() require.Equal(t, len(written), log.Len()) + for _, message := range written { read, err := log.Message(message.ID) require.NoError(t, err) @@ -186,6 +188,7 @@ func TestLogRejectsAnIncompleteLineBeforeTheLastDay(t *testing.T) { 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)) diff --git a/main.go b/main.go index e25e7f3..9c1c8b9 100644 --- a/main.go +++ b/main.go @@ -49,6 +49,7 @@ func run() error { 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": diff --git a/memory.go b/memory.go index bcc1095..7cbcce6 100644 --- a/memory.go +++ b/memory.go @@ -125,6 +125,7 @@ func (m *Memory) Close() error { defer m.mutex.Unlock() var err error + if m.syncer != nil { err = m.syncer.Close() } @@ -174,6 +175,7 @@ func (m *Memory) AppendDated(kind, text, date string) (Message, error) { // 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 { @@ -194,6 +196,7 @@ func (m *Memory) children(key NodeKey) (Node, Node, error) { func (m *Memory) Wait(ctx context.Context) error { for { m.mutex.Lock() + running := m.queue.Running() changed := m.changed m.mutex.Unlock() @@ -217,6 +220,7 @@ func (m *Memory) Wait(ctx context.Context) error { func (m *Memory) Settle(ctx context.Context) error { for { m.mutex.Lock() + pending := m.chat.Pending() running := m.queue.Running() changed := m.changed @@ -295,6 +299,7 @@ func (m *Memory) load() error { } var file MemoryViewFile + err = json.Unmarshal(data, &file) if err != nil { return fmt.Errorf("decode view: %w", err) @@ -342,6 +347,7 @@ func (m *Memory) restoreQueue() error { leaf, free := freeLeaf(message) if !free { m.queue.AddLeaf(id) + continue } @@ -410,6 +416,7 @@ func (m *Memory) dispatch() { } m.workers.Add(1) + go m.run(key, task) } } @@ -508,6 +515,7 @@ func (m *Memory) settle(key NodeKey) error { node, free := freeMerge(left, right) if !free { m.queue.AddMerge(parent) + return nil } @@ -583,6 +591,7 @@ func (m *Memory) run(key NodeKey, task Task) { } close(m.changed) + m.changed = make(chan struct{}) m.dispatch() } @@ -659,6 +668,7 @@ func (m *Memory) buildModelLines() ([]int, error) { } var sizes []int + for _, key := range keys { var node Node diff --git a/memory_test.go b/memory_test.go index 3930435..fe050c4 100644 --- a/memory_test.go +++ b/memory_test.go @@ -23,6 +23,7 @@ func TestMemoryRestartLoadsTheSameView(t *testing.T) { memory.chat.high = 200 memory.chat.low = 100 + memory.cview.high = 80 memory.cview.low = 40 for n := range 100 { @@ -89,6 +90,7 @@ func TestMemoryKeepsTheCompactionViewCoarser(t *testing.T) { memory.chat.high = 200 memory.chat.low = 100 + memory.cview.high = 80 memory.cview.low = 40 for n := range 64 { @@ -124,6 +126,7 @@ func TestMemoryRunsAtMostEightBuilds(t *testing.T) { } memory.mutex.Lock() + running := slices.Collect(maps.Keys(memory.queue.running)) memory.mutex.Unlock() slices.SortFunc(running, func(a, b NodeKey) int { @@ -179,6 +182,7 @@ func TestMemorySettleWaitsForThePendingLines(t *testing.T) { 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) @@ -193,6 +197,7 @@ func TestMemorySettleFailsWhileABuildFailed(t *testing.T) { 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())) @@ -220,6 +225,7 @@ func TestMemorySettleEndsWithItsContext(t *testing.T) { 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) @@ -233,6 +239,7 @@ func TestMemoryStartsALeafAfterFewerThanEightUnbuilt(t *testing.T) { } require.NoError(t, memory.Wait(t.Context())) + for _, task := range summarizer.Tasks() { require.Less(t, task.Index, 8) } @@ -278,6 +285,7 @@ func TestMemoryRestartRestoresTheQueue(t *testing.T) { clock := newTestClock() summarizer := newFakeSummarizer() summarizer.SetFailing(true) + memory, err := openMemory(t.Context(), root, clock.Now, summarizer, discardLog()) require.NoError(t, err) @@ -320,6 +328,7 @@ func TestMemoryBuildsFreeParents(t *testing.T) { 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) diff --git a/model.go b/model.go index 7c85643..507fbb1 100644 --- a/model.go +++ b/model.go @@ -98,8 +98,10 @@ func modelUserMessage(texts ...string) any { // 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) @@ -131,6 +133,7 @@ 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() } @@ -214,6 +217,7 @@ func (c *ModelClient) Create(ctx context.Context, request ModelRequest) (ModelRe 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) @@ -434,6 +438,7 @@ type ModelWireError struct { // 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 @@ -454,8 +459,11 @@ func modelQuotaHeaders(header http.Header) map[string]string { // 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) { @@ -472,6 +480,7 @@ func modelReadStream(stream io.Reader) (ModelResponse, error) { } else if data.Len() > 0 { done, eventErr := modelReadEvent(&response, data.Bytes()) data.Reset() + if eventErr != nil || done { return response, eventErr } @@ -487,6 +496,7 @@ func modelReadStream(stream io.Reader) (ModelResponse, error) { // 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) diff --git a/model_fake_helper_test.go b/model_fake_helper_test.go index 08c4732..9873fb6 100644 --- a/model_fake_helper_test.go +++ b/model_fake_helper_test.go @@ -54,6 +54,7 @@ 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 } @@ -65,10 +66,12 @@ func (f *ModelFake) serve(w http.ResponseWriter, r *http.Request) { 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 @@ -103,6 +106,7 @@ func modelTestRequest() ModelRequest { // 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 { @@ -155,10 +159,13 @@ func modelReadUsage(t *testing.T, path string) []ModelUsageRecord { }() 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) } diff --git a/model_test.go b/model_test.go index e932107..90d0e93 100644 --- a/model_test.go +++ b/model_test.go @@ -145,8 +145,10 @@ func TestClientRecordsEachCall(t *testing.T) { 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 @@ -176,6 +178,7 @@ func TestClientRecordsEachCall(t *testing.T) { "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) @@ -213,11 +216,13 @@ func TestClientRefusesCallsOverTheLimit(t *testing.T) { 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()) diff --git a/probe.go b/probe.go index 69d81d1..f5553ce 100644 --- a/probe.go +++ b/probe.go @@ -98,6 +98,7 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) { 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), @@ -180,8 +181,10 @@ func probeReadLog(path string, count int) (entries []DatedMessage, err error) { 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) diff --git a/probe_log_helper_test.go b/probe_log_helper_test.go index ddce81d..0097dda 100644 --- a/probe_log_helper_test.go +++ b/probe_log_helper_test.go @@ -38,6 +38,7 @@ func (g *ProbeWriter) message() (string, string) { // 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(' ') @@ -82,7 +83,9 @@ func (g *ProbeWriter) pick(words []string) string { // 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)} diff --git a/probe_test.go b/probe_test.go index a412106..298e057 100644 --- a/probe_test.go +++ b/probe_test.go @@ -26,6 +26,7 @@ func TestProbeLogHasTheSyntheticMix(t *testing.T) { 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) } @@ -51,6 +52,7 @@ func TestProbeLogHasTheSyntheticMix(t *testing.T) { } var data strings.Builder + for _, message := range messages { line, err := json.Marshal(message) require.NoError(t, err) @@ -81,6 +83,7 @@ func TestSlotSummarizerLimitsTheBuildsAtOnce(t *testing.T) { } var group sync.WaitGroup + for n := range 6 { group.Go(func() { _, err := summarizer.Summarize(t.Context(), Task{ diff --git a/prompt_test.go b/prompt_test.go index 4b0ee95..ba02b23 100644 --- a/prompt_test.go +++ b/prompt_test.go @@ -134,6 +134,7 @@ func TestAgentToolsAreStrictAndStable(t *testing.T) { 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) diff --git a/rebuild.go b/rebuild.go index 8522d70..52c0861 100644 --- a/rebuild.go +++ b/rebuild.go @@ -56,6 +56,7 @@ func rebuild(ctx context.Context, agent *Agent, added []DatedMessage, client *Mo } client.LimitCalls(buildCallLimit) + run := BuildRun{ memory: memory, client: client, @@ -136,6 +137,7 @@ func rebuildIdle(root string) error { } var work SubagentWorkFile + err = json.Unmarshal(data, &work) if err != nil { return fmt.Errorf("decode running subagents: %w", err) @@ -157,7 +159,9 @@ func rebuildReadLog(ctx context.Context, root string) ([]DatedMessage, error) { } slices.Sort(paths) + var entries []DatedMessage + for _, path := range paths { var day uint32 @@ -168,6 +172,7 @@ func rebuildReadLog(ctx context.Context, root string) ([]DatedMessage, error) { 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) @@ -215,6 +220,7 @@ func rebuildMerge(old, added []DatedMessage) ([]DatedMessage, error) { } var date time.Time + date, err = time.Parse(time.RFC3339Nano, entry.Date) if err != nil { return nil, fmt.Errorf("parse merge date: %w", err) @@ -263,6 +269,7 @@ func rebuildCopy(source, target string) error { } case entry.Type().IsRegular(): var data []byte + data, err = os.ReadFile(path) if err != nil { return fmt.Errorf("read agent file: %w", err) diff --git a/rebuild_test.go b/rebuild_test.go index e3757ea..8d3c9d4 100644 --- a/rebuild_test.go +++ b/rebuild_test.go @@ -47,7 +47,9 @@ func TestRebuildMergesDatesAndKeepsAgentFiles(t *testing.T) { 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", @@ -169,6 +171,7 @@ func TestRebuildKeepsOriginalWhenAnInvisibleMergeFails(t *testing.T) { } rebuildTestLog(t, agent.dir, old) + before := rebuildTestSnapshot(t, agent.dir) server := newModelFake(t) server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { diff --git a/replay_test.go b/replay_test.go index 547b5e4..e6b8daa 100644 --- a/replay_test.go +++ b/replay_test.go @@ -46,6 +46,7 @@ func TestReplaySyntheticLog(t *testing.T) { }() batches := 0 + low := viewChatHigh long := make([]bool, messages) for id := range messages { @@ -81,6 +82,7 @@ func TestReplaySyntheticLog(t *testing.T) { runtime.GC() var stats runtime.MemStats + runtime.ReadMemStats(&stats) t.Logf("heap in use: %d KB", stats.HeapInuse/1024) } @@ -210,6 +212,7 @@ func replayRequireOnlySummaries(t *testing.T, memory *Memory, long []bool, full } 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) } diff --git a/schedule.go b/schedule.go index 3b4a4ee..d2f53e0 100644 --- a/schedule.go +++ b/schedule.go @@ -115,8 +115,11 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { 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)) @@ -253,7 +256,9 @@ func (s *Schedules) Remove(id int) (string, error) { // 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 { @@ -293,6 +298,7 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { // 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) @@ -363,6 +369,7 @@ func (s *Schedules) change() { // 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() diff --git a/schedule_cron.go b/schedule_cron.go index 78ec90b..91fe8ce 100644 --- a/schedule_cron.go +++ b/schedule_cron.go @@ -70,6 +70,7 @@ func scheduleParseCron(expression string) (ScheduleCron, error) { } var sets [5]uint64 + for index, field := range fields { set, err := scheduleCronField(field, scheduleCronFields[index].low, scheduleCronFields[index].high) if err != nil { @@ -146,6 +147,7 @@ func (c ScheduleCron) Next(after time.Time) (time.Time, bool) { // 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< 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 { diff --git a/subagent_run_test.go b/subagent_run_test.go index b76c18d..f637dcf 100644 --- a/subagent_run_test.go +++ b/subagent_run_test.go @@ -21,7 +21,9 @@ const subagentTestFullContext = `{"error": {"message": "Your input exceeds the c func TestSpawnRunsASubagentInTheBackground(t *testing.T) { server := newModelFake(t) + var test *TurnTestLoop + server.reply = func(w http.ResponseWriter, call ModelFakeCall) { last := subagentTestLastItem(call) switch { @@ -47,6 +49,7 @@ func TestSpawnRunsASubagentInTheBackground(t *testing.T) { 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) @@ -69,6 +72,7 @@ func TestSpawnRunsASubagentInTheBackground(t *testing.T) { 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) { @@ -97,6 +101,7 @@ func TestSpawnRunsASubagentInTheBackground(t *testing.T) { 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)) } @@ -112,6 +117,7 @@ func TestSpawnRunsASubagentInTheBackground(t *testing.T) { 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) } @@ -140,6 +146,7 @@ func TestReportJoinsTheNextRequestOfARunningTurn(t *testing.T) { require.NoError(t, err) <-arrived + report := subagentReport("GreenOtter", []string{"It is done."}) test.loop.Report("GreenOtter", report, subagentFinished) close(release) @@ -170,6 +177,7 @@ func TestFullContextGivesAFailedReport(t *testing.T) { server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if subagentTestCall(call) { w.WriteHeader(http.StatusBadRequest) + _, _ = io.WriteString(w, subagentTestFullContext) return @@ -260,6 +268,7 @@ func TestTellReachesARunningSubagentBetweenItsToolCalls(t *testing.T) { 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) @@ -294,6 +303,7 @@ func TestTellAfterTheFinalAnswerStartsAnotherChain(t *testing.T) { require.NoError(t, err) <-arrived + name := test.events.Last("subagent") _, err = runs.Tell(t.Context(), name, "Count again.") require.NoError(t, err) @@ -428,10 +438,12 @@ func TestRestartResumesTheCutSubagents(t *testing.T) { }) 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)) @@ -449,7 +461,9 @@ func TestRestartResumesTheCutSubagents(t *testing.T) { require.NoError(t, err) <-arrived + name := first.Last("subagent") + stop() require.Zero(t, first.Count("work")) @@ -458,11 +472,13 @@ func TestRestartResumesTheCutSubagents(t *testing.T) { 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{ @@ -543,6 +559,7 @@ func subagentTestChat(t *testing.T, memory *Memory) []string { t.Helper() var lines []string + for id := range memory.Len() { message, err := memory.Message(id) require.NoError(t, err) @@ -571,7 +588,9 @@ func subagentTestLastItem(call ModelFakeCall) map[string]any { // 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) diff --git a/subagent_test.go b/subagent_test.go index 4fab1b1..54fa715 100644 --- a/subagent_test.go +++ b/subagent_test.go @@ -13,6 +13,7 @@ func TestSubagentNamesAreAnAdjectiveAndANoun(t *testing.T) { } require.True(t, subagentNameValid("GreenOtter")) + for _, name := range []string{"", "Green", "Greenotter", "greenOtter", "Green-Otter", "GreenOtterX", "OtterGreen", "../GreenOtter"} { require.False(t, subagentNameValid(name), name) } diff --git a/summarizer_helper_test.go b/summarizer_helper_test.go index 94c8154..e78a69e 100644 --- a/summarizer_helper_test.go +++ b/summarizer_helper_test.go @@ -23,6 +23,7 @@ type FakeSummarizer struct { 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) @@ -31,6 +32,7 @@ func (f *FakeSummarizer) Summarize(ctx context.Context, task Task) (string, erro defer func() { f.mutex.Lock() + f.active-- f.mutex.Unlock() }() diff --git a/syncer.go b/syncer.go index 1549461..3da6d7a 100644 --- a/syncer.go +++ b/syncer.go @@ -61,11 +61,13 @@ func (s *Syncer) Written() { // 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 @@ -92,6 +94,7 @@ func (s *Syncer) Flush(ctx context.Context) error { 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 @@ -109,6 +112,7 @@ func (s *Syncer) start() { s.running = true s.workers.Add(1) + go s.run(s.written) } @@ -130,6 +134,7 @@ func (s *Syncer) run(target uint64) { } close(s.done) + s.done = make(chan struct{}) s.start() } diff --git a/syncer_test.go b/syncer_test.go index efa174a..16f3345 100644 --- a/syncer_test.go +++ b/syncer_test.go @@ -28,6 +28,7 @@ func TestFlushWaitsForASyncThatStartsAfterTheWrite(t *testing.T) { // 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()) @@ -35,6 +36,7 @@ func TestFlushWaitsForASyncThatStartsAfterTheWrite(t *testing.T) { gate.Release() gate.WaitStarted(t, 2) + select { case err := <-flushed: t.Fatalf("flush returned before the second sync: %v", err) @@ -52,6 +54,7 @@ func TestOneSyncCoversTheWritesDuringTheSyncBeforeIt(t *testing.T) { s := newSyncer(gate.Sync) s.Written() gate.WaitStarted(t, 1) + for range 10 { s.Written() } @@ -103,6 +106,7 @@ func TestKilledWriterLosesNoLoggedMessage(t *testing.T) { 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) @@ -146,10 +150,12 @@ type SyncerGatedSync struct { func (g *SyncerGatedSync) Sync() error { g.mutex.Lock() + g.count++ g.mutex.Unlock() g.started <- struct{}{} + <-g.release return nil @@ -167,6 +173,7 @@ func (g *SyncerGatedSync) WaitStarted(t *testing.T, n int) { // Release lets the running sync end. func (g *SyncerGatedSync) Release() { <-g.started + g.release <- struct{}{} } @@ -206,6 +213,7 @@ func syncerRunKilledWriter(t *testing.T, root string, count int) int { } require.NoError(t, command.Process.Kill()) + for scanner.Scan() { _, err = strconv.Atoi(strings.TrimSpace(scanner.Text())) if err == nil { diff --git a/telegram.go b/telegram.go index af1b91e..2a9345a 100644 --- a/telegram.go +++ b/telegram.go @@ -73,8 +73,10 @@ func loadTelegramConfig(dir string) (TelegramConfig, bool, error) { } 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") @@ -180,6 +182,7 @@ func (b *TelegramBot) Notify(event Event) { } b.mutex.Lock() + b.queue = append(b.queue, event) b.mutex.Unlock() @@ -197,7 +200,9 @@ func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) defer cancel() var sender sync.WaitGroup + sender.Add(1) + go func() { defer sender.Done() @@ -205,6 +210,7 @@ func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) }() err := b.poll(ctx, l, whatsapp) + cancel() sender.Wait() @@ -237,6 +243,7 @@ func (b *TelegramBot) send(ctx context.Context) { switch event.Kind { case "busy": busy = true + b.typing(ctx) ticker.Reset(telegramTypingInterval) case "idle": @@ -281,6 +288,7 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) } outage.End() + wait = b.minWait for _, update := range updates { if ctx.Err() != nil { @@ -372,6 +380,7 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp 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 } @@ -429,8 +438,10 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { } 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 @@ -438,6 +449,7 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { if !retry { b.logger.Printf("reply lost: %v", err) + return } @@ -468,6 +480,7 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { if telegramRejected(err) { b.logger.Printf("reply sent with simple formatting: %v", err) + for _, part := range telegramReplyParts(telegramUnescapeTags(markdown)) { b.deliver(ctx, part) } @@ -477,6 +490,7 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { if !retry { b.logger.Printf("reply lost: %v", err) + return } @@ -529,6 +543,7 @@ func (o *TelegramOutage) End() { } o.logger.Printf("%s after %s", o.back, durationText(time.Since(o.start))) + o.start = time.Time{} } @@ -557,6 +572,7 @@ func telegramLoadOffset(path string) (int64, error) { } var file TelegramOffsetFile + err = json.Unmarshal(data, &file) if err != nil { return 0, fmt.Errorf("%s: %w", path, err) @@ -588,6 +604,7 @@ func telegramSaveOffset(path string, offset int64) error { // never cuts a character. func telegramSplitMessage(text string, limit int) []string { var parts []string + for { end := telegramPrefixEnd(text, limit) if end == len(text) { @@ -637,6 +654,7 @@ func telegramText(event Event) string { // 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 == "" { @@ -692,7 +710,9 @@ func telegramPrefixEnd(text string, limit int) int { // 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 { @@ -751,10 +771,13 @@ func telegramRichCut(markdown string, limit int) (int, string, string) { // 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) diff --git a/telegram_api.go b/telegram_api.go index 7b9457f..07d2bf1 100644 --- a/telegram_api.go +++ b/telegram_api.go @@ -52,11 +52,13 @@ func newTelegramAPI(base, token string) *TelegramAPI { // 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 } @@ -70,6 +72,7 @@ func telegramRetryable(err error) (bool, time.Duration) { // 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) } @@ -89,6 +92,7 @@ type TelegramAPI struct { // 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, @@ -155,6 +159,7 @@ func (b *TelegramAPI) call(ctx context.Context, timeout time.Duration, method st } request.Header.Set("Content-Type", "application/json") + response, err := b.http.Do(request) if err != nil { return b.networkError("no answer", err, timeout) @@ -167,6 +172,7 @@ func (b *TelegramAPI) call(ctx context.Context, timeout time.Duration, method st } var answer TelegramAnswer + err = json.Unmarshal(data, &answer) if err != nil && response.StatusCode != http.StatusOK { return &TelegramError{ @@ -221,6 +227,7 @@ func (b *TelegramAPI) networkError(what string, err error, timeout time.Duration // cause is kept. func (b *TelegramAPI) redact(err error) string { var urlError *url.Error + if errors.As(err, &urlError) { err = urlError.Err } diff --git a/telegram_api_test.go b/telegram_api_test.go index 43e952e..98aa232 100644 --- a/telegram_api_test.go +++ b/telegram_api_test.go @@ -123,6 +123,7 @@ func TestSendCallsTimeOutOnAStalledConnection(t *testing.T) { 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) @@ -155,6 +156,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { 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, @@ -165,6 +167,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { name: "server error page is retried", reply: func(w http.ResponseWriter) { w.WriteHeader(http.StatusBadGateway) + _, _ = io.WriteString(w, "bad gateway") }, retry: true, @@ -174,6 +177,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { 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, @@ -183,6 +187,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { 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, @@ -193,6 +198,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { 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, @@ -203,6 +209,7 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { name: "not found page of a proxy is not a refusal", reply: func(w http.ResponseWriter) { w.WriteHeader(http.StatusNotFound) + _, _ = io.WriteString(w, "not found") }, retry: false, diff --git a/telegram_fake_helper_test.go b/telegram_fake_helper_test.go index a70a199..3056aa1 100644 --- a/telegram_fake_helper_test.go +++ b/telegram_fake_helper_test.go @@ -50,6 +50,7 @@ type TelegramFake struct { // Add makes update available to getUpdates. func (f *TelegramFake) Add(update TelegramUpdate) { f.mutex.Lock() + f.updates = append(f.updates, update) f.mutex.Unlock() @@ -71,6 +72,7 @@ func (f *TelegramFake) Calls() []TelegramFakeCall { // sendRichMessage calls that succeeded, in order. func (f *TelegramFake) Sent() []string { var texts []string + for _, call := range f.Calls() { switch { case !call.ok: @@ -88,13 +90,16 @@ func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { 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 } @@ -104,6 +109,7 @@ func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { } f.mutex.Lock() + status, description := 0, "" if f.fail != nil { status, description = f.fail(call) @@ -114,6 +120,7 @@ func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { if stall { f.record(call) + select { case <-r.Context().Done(): case <-f.stopped: @@ -125,6 +132,7 @@ func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { if status != 0 { f.record(call) w.WriteHeader(status) + wait := 0 if status == http.StatusTooManyRequests { wait = 1 @@ -137,6 +145,7 @@ func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { call.ok = true f.record(call) + if method != "getUpdates" { _, _ = io.WriteString(w, `{"ok":true,"result":true}`) return @@ -175,6 +184,7 @@ func (f *TelegramFake) from(offset int64) []TelegramUpdate { defer f.mutex.Unlock() var updates []TelegramUpdate + kept := f.updates[:0] for _, update := range f.updates { if update.UpdateID >= offset { diff --git a/telegram_markdown.go b/telegram_markdown.go index 8738bcd..712cf28 100644 --- a/telegram_markdown.go +++ b/telegram_markdown.go @@ -31,6 +31,7 @@ var ( // 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 { @@ -144,6 +145,7 @@ type TelegramMarkdownText struct { // add appends text. func (m *TelegramMarkdownText) add(text string) { m.text.WriteString(text) + m.units += telegramUTF16Units(text) } @@ -164,6 +166,7 @@ func (m *TelegramMarkdownText) inline(line string) { 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 @@ -177,6 +180,7 @@ func (m *TelegramMarkdownText) inline(line string) { } m.add(line[plain:index]) + start := m.units if span.kind == "code" { m.add(span.content) @@ -256,6 +260,7 @@ func telegramHeadingText(line string) string { // 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 { @@ -266,6 +271,7 @@ func telegramRewriteProse(markdown string, rewrite func(string) string) string { _, fence := telegramCodeFence(line) if fence == "" { telegramRewriteLineProse(&out, line, rewrite) + continue } @@ -275,6 +281,7 @@ func telegramRewriteProse(markdown string, rewrite func(string) string) string { } out.WriteString(strings.Join(lines[index:end+1], "\n")) + index = end } @@ -302,6 +309,7 @@ func telegramRewriteLineProse(out *strings.Builder, line string, rewrite func(st out.WriteString(rewrite(line[prose:index])) out.WriteString(line[index:end]) + index = end prose = end diff --git a/telegram_test.go b/telegram_test.go index e360045..f1a76b6 100644 --- a/telegram_test.go +++ b/telegram_test.go @@ -322,6 +322,7 @@ func TestRepliesGoAsRichMessages(t *testing.T) { telegramWaitForSent(t, telegram, telegramTextOnly) var sends []TelegramFakeCall + for _, call := range telegram.Calls() { if call.method == "sendMessage" || call.method == "sendRichMessage" { sends = append(sends, call) @@ -433,6 +434,7 @@ func TestRichMessageFailures(t *testing.T) { }) var methods []string + for _, call := range telegram.Calls() { methods = append(methods, call.method) } @@ -551,6 +553,7 @@ func TestRefusedRichPartGoesWithTheTextAsWritten(t *testing.T) { }) var fallback, rich []string + for _, call := range telegram.Calls() { switch { case !call.ok: @@ -640,6 +643,7 @@ func TestBotLogsAnOutageOnlyAtItsStartAndEnd(t *testing.T) { }) 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]) diff --git a/terminal.go b/terminal.go index eff009c..4f1c378 100644 --- a/terminal.go +++ b/terminal.go @@ -37,6 +37,7 @@ func terminal(ctx context.Context, in io.Reader, out io.Writer) error { } 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) @@ -60,14 +61,17 @@ func terminalSession(ctx context.Context, conn net.Conn, in io.Reader, out io.Wr }() 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 } @@ -83,6 +87,7 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error { 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{ @@ -130,8 +135,10 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error { 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) diff --git a/tools.go b/tools.go index 1cb098a..4c00350 100644 --- a/tools.go +++ b/tools.go @@ -94,11 +94,15 @@ func (b *Toolbox) Call(ctx context.Context, name, arguments string) (string, 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: @@ -109,36 +113,42 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } 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: @@ -148,6 +158,7 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } case "tell": var args ToolTellArguments + err = toolDecodeArguments(arguments, &args) switch { case b.runs == nil: @@ -157,6 +168,7 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } case "whatsapp_find": var args ToolWhatsAppFindArguments + err = toolDecodeArguments(arguments, &args) switch { case b.whatsapp == nil: @@ -166,6 +178,7 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } case "whatsapp_send": var args ToolWhatsAppSendArguments + err = toolDecodeArguments(arguments, &args) switch { case b.whatsapp == nil: @@ -175,6 +188,7 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } case "schedule_set": var args ToolScheduleSetArguments + err = toolDecodeArguments(arguments, &args) switch { case b.schedules == nil: @@ -192,6 +206,7 @@ func (b *Toolbox) run(ctx context.Context, name, arguments string) string { } case "schedule_remove": var args ToolScheduleRemoveArguments + err = toolDecodeArguments(arguments, &args) switch { case b.schedules == nil: @@ -241,7 +256,9 @@ func (b *Toolbox) zoom(id, n int) (string, error) { } 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 { @@ -298,7 +315,9 @@ func (b *Toolbox) shell(ctx context.Context, command string) string { err := cmd.Run() text := output.String() + var exit *exec.ExitError + switch { case ctx.Err() != nil: text += "\n[interrupted]" @@ -530,6 +549,7 @@ func toolClip(text string) string { 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) diff --git a/tools_test.go b/tools_test.go index ed58438..487a5fc 100644 --- a/tools_test.go +++ b/tools_test.go @@ -93,6 +93,7 @@ func TestShellInterruptKillsTheCommandAndItsChildren(t *testing.T) { 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) } diff --git a/tree.go b/tree.go index 83ecda3..69558c5 100644 --- a/tree.go +++ b/tree.go @@ -124,6 +124,7 @@ func (t *Tree) Close() error { for level, index := range t.levels { err = errors.Join(err, index.Close()) + delete(t.levels, level) } @@ -178,6 +179,7 @@ func (t *Tree) Node(level, index int) (Node, bool, error) { } var node Node + err = json.Unmarshal(line, &node) if err != nil { return Node{}, false, fmt.Errorf("decode node (%d, %d): %w", level, index, err) @@ -212,6 +214,7 @@ func (t *Tree) Keys() ([]NodeKey, error) { 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 @@ -245,6 +248,7 @@ func (t *Tree) open(ctx context.Context, total int) error { 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) diff --git a/tree_test.go b/tree_test.go index f2cec17..d927a9e 100644 --- a/tree_test.go +++ b/tree_test.go @@ -90,6 +90,7 @@ func TestTreeReopenLoadsTheSameNodes(t *testing.T) { }() require.Equal(t, len(nodes), tree.Count()) + for _, node := range nodes { require.Equal(t, node, treeReadNode(t, tree, node.Level, node.Index)) } @@ -125,6 +126,7 @@ func TestTreeBuildsAMissingIndexAgain(t *testing.T) { }() require.Equal(t, len(nodes), tree.Count()) + for _, node := range nodes { require.Equal(t, node, treeReadNode(t, tree, node.Level, node.Index)) } diff --git a/turn.go b/turn.go index 40645b1..4288604 100644 --- a/turn.go +++ b/turn.go @@ -112,6 +112,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -167,6 +168,7 @@ func (l *TurnLoop) Wake(messages ...Message) { } l.mutex.Lock() + l.pending = append(l.pending, messages...) l.mutex.Unlock() @@ -285,6 +287,7 @@ func (l *TurnLoop) waiting() bool { // 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() @@ -313,11 +316,13 @@ func (l *TurnLoop) turn(ctx context.Context) error { defer cancel() l.mutex.Lock() + l.cancel = cancel l.mutex.Unlock() defer func() { l.mutex.Lock() + l.cancel = nil l.mutex.Unlock() }() diff --git a/turn_test.go b/turn_test.go index 5d93e08..8173906 100644 --- a/turn_test.go +++ b/turn_test.go @@ -61,6 +61,7 @@ func TestTurnRunsToolsUntilTheFinalAnswer(t *testing.T) { 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) } @@ -106,11 +107,13 @@ func TestMessageDuringATurnJoinsTheNextRequest(t *testing.T) { 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", @@ -140,6 +143,7 @@ func TestMessageDuringTheFinalRequestStartsANewTurn(t *testing.T) { require.NoError(t, err) <-arrived + _, err = test.loop.Submit(terminalSource, "second") require.NoError(t, err) close(release) @@ -237,6 +241,7 @@ func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) { 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]) @@ -264,6 +269,7 @@ func (l *TurnTestLoop) Log(t *testing.T) []string { t.Helper() var lines []string + for id := range l.memory.Len() { message, err := l.memory.Message(id) require.NoError(t, err) @@ -360,12 +366,14 @@ func turnInputTexts(t *testing.T, call ModelFakeCall) []string { 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) } @@ -401,6 +409,7 @@ func newTestTurnLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) 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() { diff --git a/view.go b/view.go index 2124d3d..2357198 100644 --- a/view.go +++ b/view.go @@ -197,7 +197,9 @@ func (v *View) LinesWithout(skip map[int]bool) []string { } var text strings.Builder + v.renderLine(&text, line) + lines = append(lines, text.String()) } @@ -209,7 +211,9 @@ func (v *View) LinesWithout(skip map[int]bool) []string { // sees a pending mark or a part of a message. func (v *View) Context(end int) string { var text strings.Builder + text.WriteString("\n") + for _, line := range v.lines { _, built := v.nodes[line] if !built || nodeEnd(line.level, line.index) > end { @@ -261,6 +265,7 @@ func (v *View) fit(tree ViewNodeReader) (bool, error) { 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 diff --git a/view_layout.go b/view_layout.go index 0d4e756..3db2f01 100644 --- a/view_layout.go +++ b/view_layout.go @@ -59,8 +59,11 @@ type ViewLayout struct { // 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) { diff --git a/view_test.go b/view_test.go index f3d4e2a..46f76da 100644 --- a/view_test.go +++ b/view_test.go @@ -10,8 +10,11 @@ import ( 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() @@ -31,11 +34,13 @@ func TestViewMergesInTheOrderOfTheRollbackPush(t *testing.T) { 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()) } @@ -164,6 +169,7 @@ type ViewPushEntry struct { // 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) } diff --git a/whatsapp.go b/whatsapp.go index b076708..6e1e1e6 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -116,6 +116,7 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge } var err error + w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow) if err != nil { return nil, err @@ -144,6 +145,7 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge if w.device.ID != nil { w.linked = true w.inbox.waitForCatchup() + w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User) } @@ -215,6 +217,7 @@ type WhatsApp struct { // 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 @@ -225,11 +228,13 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err 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) } @@ -237,13 +242,17 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err stopRetry := func() { timer.Stop() + retry = nil + w.mutex.Lock() + w.retrying = false w.mutex.Unlock() } var pairing sync.WaitGroup + cancel := func() {} defer func() { cancel() @@ -272,6 +281,7 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err connect() case <-w.logouts: stopRetry() + err = w.loggedOut(ctx, l) if err != nil { return err @@ -279,11 +289,13 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) 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 @@ -315,6 +327,7 @@ func (w *WhatsApp) Command(args string, answer func(text string)) { switch { case len(fields) == 1 && fields[0] == "status": answer(w.Status()) + return case len(fields) == 1 && fields[0] == "unlink": request.unlink = true @@ -322,12 +335,14 @@ func (w *WhatsApp) Command(args string, answer func(text string)) { 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 } @@ -442,8 +457,10 @@ func (w *WhatsApp) link() WhatsAppLink { // 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 } @@ -455,7 +472,9 @@ func (w *WhatsApp) connect(ctx context.Context) bool { 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() @@ -466,6 +485,7 @@ func (w *WhatsApp) connect(ctx context.Context) bool { first := w.startOutage() cause := whatsappRedact(w.own, err.Error()) w.mutex.Unlock() + if first { w.logger.Printf("first connect failed, retrying: %s", cause) } @@ -494,6 +514,7 @@ func (w *WhatsApp) startOutage() bool { // 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() @@ -518,6 +539,7 @@ func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error { } 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 @@ -545,12 +567,14 @@ func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error { // 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 } @@ -560,6 +584,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str } var logoutErr error + if client != nil { ctx, cancel := context.WithTimeout(ctx, whatsappLogoutTime) defer cancel() @@ -570,10 +595,12 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str 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) @@ -581,6 +608,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str 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 } @@ -593,6 +621,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str // its files. func (w *WhatsApp) removeSession() error { w.mutex.Lock() + client := w.client w.client = nil w.linked = false @@ -617,6 +646,7 @@ func (w *WhatsApp) removeSession() error { } var err error + if container != nil { err = container.Close() } @@ -646,6 +676,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, device, err := w.newDevice(ctx) if err != nil { answer("The pairing failed: " + err.Error()) + return } @@ -664,6 +695,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, if err != nil { answer("The connection to WhatsApp failed: " + err.Error()) + return } @@ -671,6 +703,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, case item, open := <-qr: if !open || item.Event != whatsmeow.QRChannelEventCode { answer("WhatsApp did not start the pairing. Try again later.") + return } case <-ctx.Done(): @@ -686,6 +719,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, 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 } @@ -694,6 +728,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, for { var item whatsmeow.QRChannelItem + open := true select { case item, open = <-qr: @@ -708,11 +743,13 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, 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") @@ -726,6 +763,7 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, return default: answer("The pairing failed: " + whatsappItemError(item) + ". Run /whatsapp pair PHONE again.") + return } } @@ -798,16 +836,20 @@ func (w *WhatsApp) handle(event any) bool { 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) @@ -876,6 +918,7 @@ func (w *WhatsApp) handle(event any) bool { // 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 @@ -892,6 +935,7 @@ func (w *WhatsApp) disconnectedLong(disconnections int) { // 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() @@ -936,6 +980,7 @@ type WhatsAppRequest struct { // 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() } @@ -999,6 +1044,7 @@ func whatsappContactName(ctx context.Context, device *store.Device, jid types.JI // 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) diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index 53fc0e8..1625fe3 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -93,6 +93,7 @@ type WhatsAppFake struct { // 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() @@ -223,6 +224,7 @@ func (f *WhatsAppFake) Logouts() int { // 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() @@ -338,6 +340,7 @@ func whatsappTestLogTexts(t *testing.T, memory *Memory) []string { }, 10*time.Second, time.Millisecond) var texts []string + for id := range memory.Len() { message, err := memory.Message(id) require.NoError(t, err) @@ -355,6 +358,7 @@ func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) { dir := t.TempDir() whatsappTestSession(t, dir, true) + whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock()) whatsapp.state.quiet = 50 * time.Millisecond @@ -379,6 +383,7 @@ func whatsappTestContacts(t *testing.T, whatsapp *WhatsApp) { 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")) diff --git a/whatsapp_find.go b/whatsapp_find.go index 545e01c..b825a3e 100644 --- a/whatsapp_find.go +++ b/whatsapp_find.go @@ -61,6 +61,7 @@ func whatsappFind(ctx context.Context, link WhatsAppLink, groups *WhatsAppGroups } 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 { @@ -119,6 +120,7 @@ func whatsappFindAnswer(name string, matches []WhatsAppMatch, groups bool) strin }) var lines []string + for _, match := range matches[:min(len(matches), whatsappFindMatches)] { lines = append(lines, match.id+" | "+match.name+" | "+match.kind) } diff --git a/whatsapp_find_test.go b/whatsapp_find_test.go index b0d20be..ada1627 100644 --- a/whatsapp_find_test.go +++ b/whatsapp_find_test.go @@ -99,10 +99,12 @@ func TestWhatsAppFindWithoutALink(t *testing.T) { 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 } diff --git a/whatsapp_groups.go b/whatsapp_groups.go index c8d066a..56676e5 100644 --- a/whatsapp_groups.go +++ b/whatsapp_groups.go @@ -49,6 +49,7 @@ type WhatsAppGroups struct { // 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() @@ -57,6 +58,7 @@ func (g *WhatsAppGroups) name(ctx context.Context, link WhatsAppLink, id string) g.read(ctx, link) g.mutex.Lock() + name, found = g.names[id] g.mutex.Unlock() } diff --git a/whatsapp_import.go b/whatsapp_import.go index ef2bcba..81eb808 100644 --- a/whatsapp_import.go +++ b/whatsapp_import.go @@ -296,6 +296,7 @@ func whatsappImportNames(ctx context.Context, dir string) (names map[string]What } var state WhatsAppStateFile + err = json.Unmarshal(data, &state) if err != nil { return nil, "", fmt.Errorf("decode pending import names: %w", err) @@ -451,6 +452,7 @@ func whatsappImportPlan(ctx context.Context, paths []string, old []DatedMessage, } var err error + plan.added, err = rebuildMerge(nil, plan.added) if err != nil { return plan, fmt.Errorf("order import batches: %w", err) @@ -485,6 +487,7 @@ func whatsappImportRead(path, owner string, location *time.Location) (chat Whats }() var file *zip.File + for _, entry := range archive.File { if filepath.Base(entry.Name) == "_chat.txt" && !entry.FileInfo().IsDir() { if file != nil { @@ -517,6 +520,7 @@ func whatsappImportRead(path, owner string, location *time.Location) (chat Whats 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) diff --git a/whatsapp_import_test.go b/whatsapp_import_test.go index 5f7d21c..d735f4c 100644 --- a/whatsapp_import_test.go +++ b/whatsapp_import_test.go @@ -242,7 +242,9 @@ func TestWhatsAppImportPlanCommandDoesNotChangeAgentOrCallModel(t *testing.T) { }}) 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)) @@ -266,6 +268,7 @@ func TestWhatsAppImportDoesNotRebuildWhenAllDaysAreHeld(t *testing.T) { 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) @@ -274,7 +277,9 @@ func TestWhatsAppImportDoesNotRebuildWhenAllDaysAreHeld(t *testing.T) { 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") @@ -284,6 +289,7 @@ func TestWhatsAppImportDoesNotRebuildWhenAllDaysAreHeld(t *testing.T) { func TestWhatsAppImportReadsOnlyLocalSessionNames(t *testing.T) { dir := t.TempDir() whatsappTestSession(t, dir, true) + session := &WhatsApp{ dir: filepath.Join(dir, whatsappDirName), } diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go index d76e43c..c7fbf51 100644 --- a/whatsapp_inbox.go +++ b/whatsapp_inbox.go @@ -89,12 +89,14 @@ func (i *WhatsAppInbox) due() <-chan time.Time { // 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() @@ -109,6 +111,7 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL } 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) diff --git a/whatsapp_inbox_test.go b/whatsapp_inbox_test.go index 6264d19..c3937df 100644 --- a/whatsapp_inbox_test.go +++ b/whatsapp_inbox_test.go @@ -31,6 +31,7 @@ func TestWhatsAppPrivateBatchStartsATurnWithAnEmptyReply(t *testing.T) { // 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))) @@ -118,6 +119,7 @@ func TestWhatsAppGroupTurns(t *testing.T) { 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 { @@ -212,6 +214,7 @@ func TestWhatsAppCatchupGivesOneTurn(t *testing.T) { <-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), @@ -235,6 +238,7 @@ func TestWhatsAppCatchupGivesOneTurn(t *testing.T) { 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) } @@ -257,6 +261,7 @@ func TestWhatsAppCatchupEndsAfterItsLimit(t *testing.T) { 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", diff --git a/whatsapp_send_test.go b/whatsapp_send_test.go index 51651b2..469912b 100644 --- a/whatsapp_send_test.go +++ b/whatsapp_send_test.go @@ -70,6 +70,7 @@ func TestWhatsAppSendToAGroup(t *testing.T) { 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) @@ -170,6 +171,7 @@ func TestWhatsAppSendWhileDisconnected(t *testing.T) { 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) @@ -186,10 +188,12 @@ func TestWhatsAppSendWithoutALink(t *testing.T) { 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 } diff --git a/whatsapp_state.go b/whatsapp_state.go index b8929f0..c09423d 100644 --- a/whatsapp_state.go +++ b/whatsapp_state.go @@ -170,6 +170,7 @@ func (s *WhatsAppState) Closed() []WhatsAppBatch { defer s.mutex.Unlock() var closed []WhatsAppBatch + for _, batch := range s.file.Batches { if batch.Closed { closed = append(closed, batch) @@ -198,6 +199,7 @@ func (s *WhatsAppState) Wait(now time.Time) (time.Duration, bool) { 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)) { diff --git a/whatsapp_state_test.go b/whatsapp_state_test.go index 8f3d44a..2631b4d 100644 --- a/whatsapp_state_test.go +++ b/whatsapp_state_test.go @@ -47,7 +47,9 @@ func TestWhatsAppBatchesFollowTheTimesOfTheMessages(t *testing.T) { } require.NoError(t, batches.CloseAll()) + var sizes []int + for _, batch := range batches.Closed() { sizes = append(sizes, len(batch.Messages)) } @@ -119,6 +121,7 @@ func TestWhatsAppBatchesSurviveARestart(t *testing.T) { 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}) diff --git a/whatsapp_test.go b/whatsapp_test.go index dca137b..7a00bbe 100644 --- a/whatsapp_test.go +++ b/whatsapp_test.go @@ -19,6 +19,7 @@ import ( 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) @@ -30,8 +31,11 @@ func TestWhatsAppStatusFollowsTheConnection(t *testing.T) { 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()) } @@ -62,7 +66,9 @@ func TestWhatsAppRetriesTheFirstConnectionWithCappedBackoff(t *testing.T) { 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() @@ -70,6 +76,7 @@ func TestWhatsAppRetriesTheFirstConnectionWithCappedBackoff(t *testing.T) { time.Sleep(time.Nanosecond) synctest.Wait() require.Len(t, test.fake.Connections(), index+2) + outage += delay } @@ -120,6 +127,7 @@ func TestWhatsAppLinkedPairRequestKeepsFirstConnectionRetries(t *testing.T) { 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) @@ -190,18 +198,23 @@ func TestWhatsAppPairsAfterFirstConnectionRetriesStop(t *testing.T) { 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) @@ -217,8 +230,10 @@ func TestWhatsAppInboxFailureStopsFirstConnectionRetries(t *testing.T) { 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() @@ -232,7 +247,9 @@ func TestWhatsAppLongFirstConnectionOutageGivesOneNotice(t *testing.T) { 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")) @@ -242,6 +259,7 @@ func TestWhatsAppLongFirstConnectionOutageGivesOneNotice(t *testing.T) { 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) @@ -257,6 +275,7 @@ func TestWhatsAppLongFirstConnectionOutageGivesOneNotice(t *testing.T) { 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) @@ -272,17 +291,24 @@ func TestWhatsAppStartingRetriesDoNotDuplicateLibraryRetries(t *testing.T) { 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{ @@ -316,6 +342,7 @@ func TestWhatsAppIsNotLinkedWithoutALinkedDevice(t *testing.T) { 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()) }) @@ -370,6 +397,7 @@ func TestWhatsAppPairsWithACode(t *testing.T) { other.WaitFor(t, "[connected to OneChat]") asking.Type("/whatsapp pair +1 555-000-1111") + items := <-fake.pairings items <- whatsmeow.QRChannelItem{ Event: whatsmeow.QRChannelEventCode, @@ -380,7 +408,9 @@ func TestWhatsAppPairsWithACode(t *testing.T) { 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 @@ -485,6 +515,7 @@ func TestWhatsAppNewPairingEndsTheOneBefore(t *testing.T) { client.Type("/whatsapp pair 15550001111") <-fake.pairings client.Type("/whatsapp pair 15550002222") + items := <-fake.pairings items <- whatsmeow.QRChannelItem{ Event: whatsmeow.QRChannelEventCode, @@ -498,6 +529,7 @@ func TestWhatsAppNewPairingEndsTheOneBefore(t *testing.T) { 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) @@ -527,10 +559,12 @@ func TestWhatsAppUnlink(t *testing.T) { 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") @@ -549,6 +583,7 @@ func TestWhatsAppUnlink(t *testing.T) { 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) @@ -558,6 +593,7 @@ func TestWhatsAppLogoutFromThePhone(t *testing.T) { <-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) @@ -569,6 +605,7 @@ func TestWhatsAppLogoutFromThePhone(t *testing.T) { 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 @@ -584,6 +621,7 @@ func TestWhatsAppLongDisconnectionGivesOneNotice(t *testing.T) { 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") @@ -598,6 +636,7 @@ func TestWhatsAppLongDisconnectionGivesOneNotice(t *testing.T) { 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) @@ -611,8 +650,10 @@ func TestWhatsAppDisconnectionWithNoRetryGivesANoticeAtOnce(t *testing.T) { 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 @@ -744,10 +785,12 @@ func whatsappRetryTest(t *testing.T) *WhatsAppRetryTest { 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() { -- 2.53.0