From: Anderson A. Giacomolli Date: Wed, 11 Nov 2026 14:07:43 +0000 (-0300) Subject: Order functions by their first use X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/d8f3785bb69288663f4cc46bf4d609c6770452d5?ds=sidebyside Order functions by their first use Constructors, each receiver group, and standalone helpers follow the order in which their callers use them. Type-aware references keep unrelated methods with the same name distinct. Functions without an in- file caller retain a stable relative order. Function text and behavior stay unchanged. Issue: 947bddb --- diff --git a/agent.go b/agent.go index 8226a43..13491c1 100644 --- a/agent.go +++ b/agent.go @@ -115,11 +115,6 @@ func (a *Agent) Voices() *VoiceStore { return &VoiceStore{dir: filepath.Join(filepath.Dir(root), voiceDirName)} } -// CacheKey returns the prompt cache key of every model call of the agent. -func (a *Agent) CacheKey() string { - return "onechat-" + a.config.ID -} - // Prompt returns the system prompt of the agent with the user's instructions // from instructions.md. It reads the file on each call, so a change applies at // the next call. A missing file means no instructions. @@ -141,6 +136,17 @@ func (a *Agent) Prompt() (string, error) { return prompt(a.config.Name, path, instructions), nil } +// instructions reads the user's instructions from instructions.md. A missing +// file means no instructions. +func (a *Agent) instructions() (string, error) { + data, err := os.ReadFile(filepath.Join(a.dir, agentInstructionsFileName)) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return "", fmt.Errorf("read instructions: %w", err) + } + + return string(data), nil +} + // Tools returns the tools shared by the turns and compactions of this agent. A // child has no image output, WhatsApp, child creation, or route tools. func (a *Agent) Tools() []ModelTool { @@ -213,6 +219,11 @@ func (a *Agent) Children() ([]*Agent, error) { return children, nil } +// CacheKey returns the prompt cache key of every model call of the agent. +func (a *Agent) CacheKey() string { + return "onechat-" + a.config.ID +} + // CreateChild writes a new direct child with its own ID and the model, effort, // and working directory of the root. Child and subagent names share a // namespace. The config is complete before an atomic rename publishes the @@ -299,17 +310,6 @@ func (a *Agent) SubagentPrompt() (string, error) { return promptSubagent(a.config.Name, instructions), nil } -// instructions reads the user's instructions from instructions.md. A missing -// file means no instructions. -func (a *Agent) instructions() (string, error) { - data, err := os.ReadFile(filepath.Join(a.dir, agentInstructionsFileName)) - if err != nil && !errors.Is(err, os.ErrNotExist) { - return "", fmt.Errorf("read instructions: %w", err) - } - - return string(data), nil -} - // validate makes sure that each setting has a valid value. func (c AgentConfig) validate() error { if c.Name == "" || strings.ContainsAny(c.Name, "\r\n") { @@ -340,6 +340,33 @@ func (c AgentConfig) validate() error { return nil } +// createAgent writes the config.json of a new agent in dir with the defaults +// and a new ID. +func createAgent(dir string) (*Agent, error) { + config, err := defaultAgentConfig() + if err != nil { + return nil, err + } + + return writeAgent(dir, config) +} + +// defaultAgentConfig returns the settings of a new agent, without an ID. The +// commands run in the home directory of the user. +func defaultAgentConfig() (AgentConfig, error) { + home, err := os.UserHomeDir() + if err != nil { + return AgentConfig{}, fmt.Errorf("find working directory: %w", err) + } + + return AgentConfig{ + Name: "OneChat", + Model: "gpt-6.1-sol", + Effort: "medium", + Workdir: home, + }, nil +} + // agentChildNameValid accepts one word of 2 to 32 ASCII letters with an initial // capital letter. Child names describe their area. func agentChildNameValid(name string) bool { @@ -357,17 +384,6 @@ func agentChildNameValid(name string) bool { return true } -// createAgent writes the config.json of a new agent in dir with the defaults -// and a new ID. -func createAgent(dir string) (*Agent, error) { - config, err := defaultAgentConfig() - if err != nil { - return nil, err - } - - return writeAgent(dir, config) -} - // writeAgent gives config a new ID and writes it in dir. func writeAgent(dir string, config AgentConfig) (*Agent, error) { id := make([]byte, agentIDBytes) @@ -392,19 +408,3 @@ func writeAgent(dir string, config AgentConfig) (*Agent, error) { config: config, }, nil } - -// defaultAgentConfig returns the settings of a new agent, without an ID. The -// commands run in the home directory of the user. -func defaultAgentConfig() (AgentConfig, error) { - home, err := os.UserHomeDir() - if err != nil { - return AgentConfig{}, fmt.Errorf("find working directory: %w", err) - } - - return AgentConfig{ - Name: "OneChat", - Model: "gpt-6.1-sol", - Effort: "medium", - Workdir: home, - }, nil -} diff --git a/backup.go b/backup.go index 9398772..517b7c7 100644 --- a/backup.go +++ b/backup.go @@ -93,42 +93,6 @@ func restore(ctx context.Context, path string, out io.Writer) error { return nil } -// backupData stages a session snapshot, then copies the data under snapshot's -// write locks. Compression runs after the locks are released. -func backupData(ctx context.Context, dir, path string, whatsapp *WhatsApp, snapshot func(context.Context, func() error) error) (size int64, err error) { - if !filepath.IsAbs(path) { - return 0, errors.New("backup path must be absolute") - } - - stage, err := os.MkdirTemp("", "onechat-backup-*") - if err != nil { - return 0, fmt.Errorf("create backup directory: %w", err) - } - - defer func() { - err = errors.Join(err, os.RemoveAll(stage)) - }() - - err = whatsapp.BackupSession(ctx, filepath.Join(stage, whatsappDirName, whatsappSessionFileName)) - if err != nil { - return 0, fmt.Errorf("copy WhatsApp session: %w", err) - } - - err = snapshot(ctx, func() error { - return backupCopy(ctx, dir, stage, path) - }) - if err != nil { - return 0, fmt.Errorf("copy data directory: %w", err) - } - - size, err = backupArchive(ctx, stage, path) - if err != nil { - return 0, fmt.Errorf("archive backup: %w", err) - } - - return size, nil -} - // backupRestore installs a validated archive into an empty data directory while // holding the token lock. It keeps the lock file so daemon starts use that same // inode, including after a failed restore. @@ -205,6 +169,42 @@ func backupRestore(ctx context.Context, path, dir string) (err error) { return nil } +// backupData stages a session snapshot, then copies the data under snapshot's +// write locks. Compression runs after the locks are released. +func backupData(ctx context.Context, dir, path string, whatsapp *WhatsApp, snapshot func(context.Context, func() error) error) (size int64, err error) { + if !filepath.IsAbs(path) { + return 0, errors.New("backup path must be absolute") + } + + stage, err := os.MkdirTemp("", "onechat-backup-*") + if err != nil { + return 0, fmt.Errorf("create backup directory: %w", err) + } + + defer func() { + err = errors.Join(err, os.RemoveAll(stage)) + }() + + err = whatsapp.BackupSession(ctx, filepath.Join(stage, whatsappDirName, whatsappSessionFileName)) + if err != nil { + return 0, fmt.Errorf("copy WhatsApp session: %w", err) + } + + err = snapshot(ctx, func() error { + return backupCopy(ctx, dir, stage, path) + }) + if err != nil { + return 0, fmt.Errorf("copy data directory: %w", err) + } + + size, err = backupArchive(ctx, stage, path) + if err != nil { + return 0, fmt.Errorf("archive backup: %w", err) + } + + return size, nil +} + // backupCopy stages regular files and directories, excluding runtime files, the // output archive, and raw session database files. Each file keeps its mode. func backupCopy(ctx context.Context, source, target, output string) error { diff --git a/backup_archive.go b/backup_archive.go index 3cd6013..f68cb70 100644 --- a/backup_archive.go +++ b/backup_archive.go @@ -125,6 +125,29 @@ func backupArchive(ctx context.Context, root, path string) (size int64, err erro return info.Size(), nil } +// backupCopyFile copies the size fixed at the snapshot of path. Growth after +// that point is not part of the copy, and a shorter file is an error. +func backupCopyFile(ctx context.Context, out io.Writer, path string, size int64) (err error) { + file, err := os.Open(path) + if err != nil { + return fmt.Errorf("open backup file: %w", err) + } + + defer func() { + err = errors.Join(err, file.Close()) + }() + + _, err = io.CopyN(out, BackupReader{ + ctx: ctx, + reader: file, + }, size) + if err != nil { + return fmt.Errorf("copy backup file: %w", err) + } + + return nil +} + // backupExtract restores regular files and directories in an empty staging // directory. Links, duplicate entries, unsafe names, and special modes fail. // Directory modes apply last so a read-only directory can hold files. @@ -218,27 +241,19 @@ func backupExtract(ctx context.Context, path, root string) (err error) { return nil } -// backupCopyFile copies the size fixed at the snapshot of path. Growth after -// that point is not part of the copy, and a shorter file is an error. -func backupCopyFile(ctx context.Context, out io.Writer, path string, size int64) (err error) { - file, err := os.Open(path) - if err != nil { - return fmt.Errorf("open backup file: %w", err) +// backupExcluded keeps runtime locks, sockets, probe files, and import copies +// out of the snapshot. +func backupExcluded(name string) bool { + first, _, _ := strings.Cut(filepath.ToSlash(name), "/") + if first == socketFileName || first == tokenLockFileName || first == "probe" || strings.HasPrefix(first, "agent-backup-") { + return true } - defer func() { - err = errors.Join(err, file.Close()) - }() - - _, err = io.CopyN(out, BackupReader{ - ctx: ctx, - reader: file, - }, size) - if err != nil { - return fmt.Errorf("copy backup file: %w", err) + if filepath.Base(name) == "lock" || (first == imageDirName && strings.HasPrefix(filepath.Base(name), ".image-")) || (first == voiceDirName && (strings.HasPrefix(filepath.Base(name), ".voice-") || strings.HasPrefix(filepath.Base(name), ".transcript-"))) { + return true } - return nil + return false } func backupExtractFile(in io.Reader, path string, mode os.FileMode) (err error) { @@ -263,18 +278,3 @@ func backupExtractFile(in io.Reader, path string, mode os.FileMode) (err error) return nil } - -// backupExcluded keeps runtime locks, sockets, probe files, and import copies -// out of the snapshot. -func backupExcluded(name string) bool { - first, _, _ := strings.Cut(filepath.ToSlash(name), "/") - if first == socketFileName || first == tokenLockFileName || first == "probe" || strings.HasPrefix(first, "agent-backup-") { - return true - } - - if filepath.Base(name) == "lock" || (first == imageDirName && strings.HasPrefix(filepath.Base(name), ".image-")) || (first == voiceDirName && (strings.HasPrefix(filepath.Base(name), ".voice-") || strings.HasPrefix(filepath.Base(name), ".transcript-"))) { - return true - } - - return false -} diff --git a/build_queue.go b/build_queue.go index 540e396..825277c 100644 --- a/build_queue.go +++ b/build_queue.go @@ -82,6 +82,18 @@ func (q *BuildQueue) Next() (NodeKey, bool) { return leaf, found } +// nextLeaf returns the oldest leaf in the window that can start. +func (q *BuildQueue) nextLeaf() (NodeKey, bool) { + for _, id := range q.leaves[:min(len(q.leaves), buildLeafWindow)] { + key := NodeKey{0, id} + if !q.running[key] && !q.failed[key] { + return key, true + } + } + + return NodeKey{}, false +} + // Done removes a built node from the queue. func (q *BuildQueue) Done(key NodeKey) { delete(q.running, key) @@ -126,18 +138,6 @@ func (q *BuildQueue) Running() int { return len(q.running) } -// nextLeaf returns the oldest leaf in the window that can start. -func (q *BuildQueue) nextLeaf() (NodeKey, bool) { - for _, id := range q.leaves[:min(len(q.leaves), buildLeafWindow)] { - key := NodeKey{0, id} - if !q.running[key] && !q.failed[key] { - return key, true - } - } - - return NodeKey{}, false -} - func (h BuildMergeHeap) Len() int { return len(h) } diff --git a/build_run.go b/build_run.go index 707a928..44d636d 100644 --- a/build_run.go +++ b/build_run.go @@ -175,6 +175,60 @@ func buildReadUsage(path string) (records []ModelUsageRecord, err error) { return records, nil } +// buildStop returns why a build run must stop: a failed call, or a quota window +// whose used percent grew by more than buildQuotaLimit points since the first +// call. It returns "" when the run can go on. +func buildStop(records []ModelUsageRecord) string { + for _, record := range records { + if record.Error != "" { + return "call failed: " + record.Error + } + } + + if len(records) == 0 { + return "" + } + + first := buildUsedPercents(records[0]) + for name, used := range buildUsedPercents(records[len(records)-1]) { + start, found := first[name] + if found && used-start > buildQuotaLimit { + return fmt.Sprintf("%s grew from %g to %g", name, start, used) + } + } + + return "" +} + +// buildCallStart returns the start of the call of record, or the zero time when +// its date is not valid. +func buildCallStart(record ModelUsageRecord) time.Time { + start, err := time.Parse(time.RFC3339Nano, record.Date) + if err != nil { + return time.Time{} + } + + return start +} + +// buildUsedPercents returns the used percent of each quota window of record, by +// header name. +func buildUsedPercents(record ModelUsageRecord) map[string]float64 { + used := make(map[string]float64) + for name, value := range record.Quota { + if !strings.HasSuffix(name, "-used-percent") { + continue + } + + percent, err := strconv.ParseFloat(value, 64) + if err == nil { + used[name] = percent + } + } + + return used +} + // 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 @@ -224,42 +278,6 @@ func buildWriteReport(out io.Writer, appended int, stop string, records []ModelU return nil } -// buildStop returns why a build run must stop: a failed call, or a quota window -// whose used percent grew by more than buildQuotaLimit points since the first -// call. It returns "" when the run can go on. -func buildStop(records []ModelUsageRecord) string { - for _, record := range records { - if record.Error != "" { - return "call failed: " + record.Error - } - } - - if len(records) == 0 { - return "" - } - - first := buildUsedPercents(records[0]) - for name, used := range buildUsedPercents(records[len(records)-1]) { - start, found := first[name] - if found && used-start > buildQuotaLimit { - return fmt.Sprintf("%s grew from %g to %g", name, start, used) - } - } - - return "" -} - -// buildCallStart returns the start of the call of record, or the zero time when -// its date is not valid. -func buildCallStart(record ModelUsageRecord) time.Time { - start, err := time.Parse(time.RFC3339Nano, record.Date) - if err != nil { - return time.Time{} - } - - return start -} - // buildRatio returns a / b, or 0 when b is 0. func buildRatio(a, b int) float64 { if b == 0 { @@ -345,21 +363,3 @@ func buildQuotaText(record ModelUsageRecord) string { return strings.Join(parts, " ") } - -// buildUsedPercents returns the used percent of each quota window of record, by -// header name. -func buildUsedPercents(record ModelUsageRecord) map[string]float64 { - used := make(map[string]float64) - for name, value := range record.Quota { - if !strings.HasSuffix(name, "-used-percent") { - continue - } - - percent, err := strconv.ParseFloat(value, 64) - if err == nil { - used[name] = percent - } - } - - return used -} diff --git a/chatgpt.go b/chatgpt.go index 89476fe..7748190 100644 --- a/chatgpt.go +++ b/chatgpt.go @@ -133,55 +133,6 @@ func (a *ChatGPTAuth) Authorize(ctx context.Context, header http.Header) error { return nil } -// Login runs the device code login. It writes the link and the code for the -// user to out, waits until the user approves the code, and writes the tokens to -// the token file. -func (a *ChatGPTAuth) Login(ctx context.Context, out io.Writer) error { - a.mutex.Lock() - defer a.mutex.Unlock() - - var code ChatGPTDeviceCode - - err := a.postJSON(ctx, "/api/accounts/deviceauth/usercode", map[string]string{ - "client_id": chatgptClientID, - }, &code) - if err != nil { - return fmt.Errorf("request device code (device code login must be allowed in the ChatGPT security settings): %w", err) - } - - userCode := cmp.Or(code.UserCode, code.UserCodeAlt) - if code.DeviceAuthID == "" || userCode == "" { - return errors.New("device code response has no device ID or user code") - } - - _, err = fmt.Fprintf(out, "Open %s/codex/device on any device and enter this code:\n\n %s\n\nWaiting for the approval...\n", a.issuer, userCode) - if err != nil { - return fmt.Errorf("write device code: %w", err) - } - - grant, err := a.poll(ctx, code.DeviceAuthID, userCode, chatgptPollInterval(code.Interval)) - if err != nil { - return err - } - - tokens, err := a.exchange(ctx, grant) - if err != nil { - return err - } - - err = a.write(tokens) - if err != nil { - return err - } - - _, err = fmt.Fprintln(out, "Logged in.") - if err != nil { - return fmt.Errorf("write login result: %w", err) - } - - return nil -} - // read reads the token file. func (a *ChatGPTAuth) read() (ChatGPTTokenFile, error) { data, err := os.ReadFile(a.path) @@ -252,6 +203,102 @@ func (a *ChatGPTAuth) postJSON(ctx context.Context, path string, body, result an return err } +// write replaces the token file atomically. The file has mode 0600. +func (a *ChatGPTAuth) write(tokens ChatGPTTokenFile) error { + data, err := json.Marshal(tokens) + if err != nil { + return fmt.Errorf("encode tokens: %w", err) + } + + err = writeFileAtomic(a.path, append(data, '\n')) + if err != nil { + return fmt.Errorf("write token file: %w", err) + } + + return nil +} + +// post posts body to the path of the login service and decodes a 200 answer +// into result. It returns the HTTP status, or 0 when no answer came. The error +// holds the status, but not the body, which can hold a token. +func (a *ChatGPTAuth) post(ctx context.Context, path, contentType string, body []byte, result any) (status int, err error) { + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.issuer+path, bytes.NewReader(body)) + if err != nil { + return 0, fmt.Errorf("create request: %w", err) + } + + request.Header.Set("Content-Type", contentType) + + response, err := a.http.Do(request) + if err != nil { + return 0, fmt.Errorf("send request: %w", err) + } + + defer func() { + err = errors.Join(err, response.Body.Close()) + }() + + if response.StatusCode != http.StatusOK { + return response.StatusCode, fmt.Errorf("HTTP %d", response.StatusCode) + } + + err = json.NewDecoder(response.Body).Decode(result) + if err != nil { + return response.StatusCode, fmt.Errorf("decode answer: %w", err) + } + + return response.StatusCode, nil +} + +// Login runs the device code login. It writes the link and the code for the +// user to out, waits until the user approves the code, and writes the tokens to +// the token file. +func (a *ChatGPTAuth) Login(ctx context.Context, out io.Writer) error { + a.mutex.Lock() + defer a.mutex.Unlock() + + var code ChatGPTDeviceCode + + err := a.postJSON(ctx, "/api/accounts/deviceauth/usercode", map[string]string{ + "client_id": chatgptClientID, + }, &code) + if err != nil { + return fmt.Errorf("request device code (device code login must be allowed in the ChatGPT security settings): %w", err) + } + + userCode := cmp.Or(code.UserCode, code.UserCodeAlt) + if code.DeviceAuthID == "" || userCode == "" { + return errors.New("device code response has no device ID or user code") + } + + _, err = fmt.Fprintf(out, "Open %s/codex/device on any device and enter this code:\n\n %s\n\nWaiting for the approval...\n", a.issuer, userCode) + if err != nil { + return fmt.Errorf("write device code: %w", err) + } + + grant, err := a.poll(ctx, code.DeviceAuthID, userCode, chatgptPollInterval(code.Interval)) + if err != nil { + return err + } + + tokens, err := a.exchange(ctx, grant) + if err != nil { + return err + } + + err = a.write(tokens) + if err != nil { + return err + } + + _, err = fmt.Fprintln(out, "Logged in.") + if err != nil { + return fmt.Errorf("write login result: %w", err) + } + + return nil +} + // poll asks for the grant of the device code until the user approves it or the // login times out. func (a *ChatGPTAuth) poll(ctx context.Context, deviceID, userCode string, interval time.Duration) (ChatGPTDeviceGrant, error) { @@ -328,74 +375,6 @@ func (a *ChatGPTAuth) exchange(ctx context.Context, grant ChatGPTDeviceGrant) (C return tokens, nil } -// write replaces the token file atomically. The file has mode 0600. -func (a *ChatGPTAuth) write(tokens ChatGPTTokenFile) error { - data, err := json.Marshal(tokens) - if err != nil { - return fmt.Errorf("encode tokens: %w", err) - } - - err = writeFileAtomic(a.path, append(data, '\n')) - if err != nil { - return fmt.Errorf("write token file: %w", err) - } - - return nil -} - -// post posts body to the path of the login service and decodes a 200 answer -// into result. It returns the HTTP status, or 0 when no answer came. The error -// holds the status, but not the body, which can hold a token. -func (a *ChatGPTAuth) post(ctx context.Context, path, contentType string, body []byte, result any) (status int, err error) { - request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.issuer+path, bytes.NewReader(body)) - if err != nil { - return 0, fmt.Errorf("create request: %w", err) - } - - request.Header.Set("Content-Type", contentType) - - response, err := a.http.Do(request) - if err != nil { - return 0, fmt.Errorf("send request: %w", err) - } - - defer func() { - err = errors.Join(err, response.Body.Close()) - }() - - if response.StatusCode != http.StatusOK { - return response.StatusCode, fmt.Errorf("HTTP %d", response.StatusCode) - } - - err = json.NewDecoder(response.Body).Decode(result) - if err != nil { - return response.StatusCode, fmt.Errorf("decode answer: %w", err) - } - - return response.StatusCode, nil -} - -// chatgptLogin runs "onechat login": the device code login into the token file -// of the data directory. It holds the lock of the token file, so it does not -// run while the daemon runs. -func chatgptLogin(ctx context.Context, out io.Writer) (err error) { - dir, err := dataDir() - if err != nil { - return err - } - - tokens, err := lockTokens(dir) - if err != nil { - return err - } - - defer func() { - err = errors.Join(err, tokens.Close()) - }() - - return newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer).Login(ctx, out) -} - // chatgptTokenExpiry returns the expiry of a JWT from its exp claim. func chatgptTokenExpiry(token string) (time.Time, bool) { var claims struct { @@ -409,17 +388,6 @@ func chatgptTokenExpiry(token string) (time.Time, bool) { return time.Unix(int64(*claims.Exp), 0), true } -// chatgptPollInterval returns the poll interval of a device code, given in -// seconds as a number or a string. -func chatgptPollInterval(raw json.RawMessage) time.Duration { - seconds, err := strconv.ParseFloat(strings.Trim(string(raw), `"`), 64) - if err != nil || seconds < 0 { - return chatgptDefaultPollInterval - } - - return time.Duration(seconds * float64(time.Second)) -} - // chatgptAccountID returns the ChatGPT account ID of a JWT, or "" when it has // none. func chatgptAccountID(token string) string { @@ -441,6 +409,17 @@ func chatgptAccountID(token string) string { return auth.AccountID } +// chatgptPollInterval returns the poll interval of a device code, given in +// seconds as a number or a string. +func chatgptPollInterval(raw json.RawMessage) time.Duration { + seconds, err := strconv.ParseFloat(strings.Trim(string(raw), `"`), 64) + if err != nil || seconds < 0 { + return chatgptDefaultPollInterval + } + + return time.Duration(seconds * float64(time.Second)) +} + // chatgptDecodeClaims decodes the payload of a JWT into claims, without a check // of the signature. The token comes from the login service over HTTPS, and the // client reads only the expiry and the account ID from it. @@ -457,3 +436,24 @@ func chatgptDecodeClaims(token string, claims any) bool { return json.Unmarshal(payload, claims) == nil } + +// chatgptLogin runs "onechat login": the device code login into the token file +// of the data directory. It holds the lock of the token file, so it does not +// run while the daemon runs. +func chatgptLogin(ctx context.Context, out io.Writer) (err error) { + dir, err := dataDir() + if err != nil { + return err + } + + tokens, err := lockTokens(dir) + if err != nil { + return err + } + + defer func() { + err = errors.Join(err, tokens.Close()) + }() + + return newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer).Login(ctx, out) +} diff --git a/chatgpt_fake_helper_test.go b/chatgpt_fake_helper_test.go index 0dca6bb..a25ed12 100644 --- a/chatgpt_fake_helper_test.go +++ b/chatgpt_fake_helper_test.go @@ -56,36 +56,6 @@ func newTestChatGPTAuth(t *testing.T, server *ChatGPTFake) *ChatGPTAuth { return newChatGPTAuth(filepath.Join(t.TempDir(), chatgptTokenFileName), server.URL) } -// token returns a new access token that expires after ttl, for a token file -// that a test writes. -func (f *ChatGPTFake) token(ttl time.Duration) string { - f.mutex.Lock() - defer f.mutex.Unlock() - - return f.newToken(ttl) -} - -// newToken returns a new fake JWT with the expiry and the account claim, and -// keeps it in access. The caller holds the mutex. -func (f *ChatGPTFake) newToken(ttl time.Duration) string { - claims, err := json.Marshal(map[string]any{ - "exp": time.Now().Add(ttl).Unix(), - chatgptAuthClaim: map[string]any{ - "chatgpt_account_id": chatgptTestAccount, - }, - "serial": len(f.access), - }) - if err != nil { - panic(err) - } - - encode := base64.RawURLEncoding.EncodeToString - token := encode([]byte(`{"alg":"none"}`)) + "." + encode(claims) + "." + encode([]byte("fake")) - f.access = append(f.access, token) - - return token -} - func (f *ChatGPTFake) userCode(w http.ResponseWriter, r *http.Request) { var body map[string]string @@ -187,16 +157,34 @@ func (f *ChatGPTFake) oauthToken(w http.ResponseWriter, r *http.Request) { chatgptWriteJSON(w, answer) } -func chatgptWriteTestTokens(t *testing.T, auth *ChatGPTAuth, access, refresh string) { - t.Helper() - - err := auth.write(ChatGPTTokenFile{ - AccessToken: access, - RefreshToken: refresh, - IDToken: access, - AccountID: chatgptTestAccount, +// newToken returns a new fake JWT with the expiry and the account claim, and +// keeps it in access. The caller holds the mutex. +func (f *ChatGPTFake) newToken(ttl time.Duration) string { + claims, err := json.Marshal(map[string]any{ + "exp": time.Now().Add(ttl).Unix(), + chatgptAuthClaim: map[string]any{ + "chatgpt_account_id": chatgptTestAccount, + }, + "serial": len(f.access), }) - require.NoError(t, err) + if err != nil { + panic(err) + } + + encode := base64.RawURLEncoding.EncodeToString + token := encode([]byte(`{"alg":"none"}`)) + "." + encode(claims) + "." + encode([]byte("fake")) + f.access = append(f.access, token) + + return token +} + +// token returns a new access token that expires after ttl, for a token file +// that a test writes. +func (f *ChatGPTFake) token(ttl time.Duration) string { + f.mutex.Lock() + defer f.mutex.Unlock() + + return f.newToken(ttl) } func chatgptWriteJSON(w http.ResponseWriter, value any) { @@ -207,3 +195,15 @@ func chatgptWriteJSON(w http.ResponseWriter, value any) { panic(fmt.Sprint("encode answer: ", err)) } } + +func chatgptWriteTestTokens(t *testing.T, auth *ChatGPTAuth, access, refresh string) { + t.Helper() + + err := auth.write(ChatGPTTokenFile{ + AccessToken: access, + RefreshToken: refresh, + IDToken: access, + AccountID: chatgptTestAccount, + }) + require.NoError(t, err) +} diff --git a/daemon.go b/daemon.go index ea25318..b6595c7 100644 --- a/daemon.go +++ b/daemon.go @@ -201,6 +201,61 @@ func (h *DaemonHub) CloseAll() { h.workers.Wait() } +// daemonRunRequest runs one request of a client. The answers of a WhatsApp +// command go to answer. +func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, answer func(text string)) error { + switch request.Type { + case "send": + if request.Text == "" { + return errors.New("empty message") + } + + _, err := l.Submit(terminalSource, request.Text) + + return err + case "image": + if !filepath.IsAbs(request.Path) { + return errors.New("use /image PATH [caption]") + } + + data, err := imageReadFile(request.Path) + if err != nil { + return fmt.Errorf("read terminal image: %w", err) + } + + name, err := l.agent.Images().Save(data) + if err != nil { + return fmt.Errorf("store terminal image: %w", err) + } + + _, err = l.Submit(terminalSource, imageText([]string{name}, request.Text)) + if err != nil { + return fmt.Errorf("submit terminal image: %w", err) + } + + return nil + case "interrupt": + if !l.Interrupt() { + return errors.New("no turn runs") + } + + return nil + case "whatsapp": + whatsapp.Command(request.Text, answer) + + return nil + case "stop": + found, err := l.StopSubagent(request.Text) + if err == nil && !found { + err = fmt.Errorf("no subagent %s runs", request.Text) + } + + return err + default: + return fmt.Errorf("unknown request %q", request.Type) + } +} + // daemon runs "onechat daemon": it owns the root, its children, and their // files, and serves the clients on the Unix socket in the data directory until // ctx ends. When telegram.json exists, it also runs the Telegram bot. It also @@ -489,58 +544,3 @@ func daemonRunChild(ctx context.Context, loop *TurnLoop, schedules *Schedules) e return errors.Join(err, <-done) } - -// daemonRunRequest runs one request of a client. The answers of a WhatsApp -// command go to answer. -func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, answer func(text string)) error { - switch request.Type { - case "send": - if request.Text == "" { - return errors.New("empty message") - } - - _, err := l.Submit(terminalSource, request.Text) - - return err - case "image": - if !filepath.IsAbs(request.Path) { - return errors.New("use /image PATH [caption]") - } - - data, err := imageReadFile(request.Path) - if err != nil { - return fmt.Errorf("read terminal image: %w", err) - } - - name, err := l.agent.Images().Save(data) - if err != nil { - return fmt.Errorf("store terminal image: %w", err) - } - - _, err = l.Submit(terminalSource, imageText([]string{name}, request.Text)) - if err != nil { - return fmt.Errorf("submit terminal image: %w", err) - } - - return nil - case "interrupt": - if !l.Interrupt() { - return errors.New("no turn runs") - } - - return nil - case "whatsapp": - whatsapp.Command(request.Text, answer) - - return nil - case "stop": - found, err := l.StopSubagent(request.Text) - if err == nil && !found { - err = fmt.Errorf("no subagent %s runs", request.Text) - } - - return err - default: - return fmt.Errorf("unknown request %q", request.Type) - } -} diff --git a/daemon_log.go b/daemon_log.go index f1e4b0b..771fef4 100644 --- a/daemon_log.go +++ b/daemon_log.go @@ -104,18 +104,6 @@ func (h *DaemonLogHandler) WithGroup(name string) slog.Handler { return &clone } -// daemonLogger returns a logger for one daemon part. JOURNAL_STREAM selects -// syslog priorities without a time stamp. Other output carries the standard -// time stamp and level. -func daemonLogger(out io.Writer, prefix string) *slog.Logger { - return slog.New(&DaemonLogHandler{ - out: out, - prefix: prefix, - journal: os.Getenv("JOURNAL_STREAM") != "", - mutex: &sync.Mutex{}, - }) -} - // daemonLogAttr resolves and writes one attribute, or keeps its cause for the // end of the line. Groups flatten into dotted keys and empty attributes produce // no text. @@ -165,3 +153,15 @@ func durationText(d time.Duration) string { return d.Round(time.Second).String() } + +// daemonLogger returns a logger for one daemon part. JOURNAL_STREAM selects +// syslog priorities without a time stamp. Other output carries the standard +// time stamp and level. +func daemonLogger(out io.Writer, prefix string) *slog.Logger { + return slog.New(&DaemonLogHandler{ + out: out, + prefix: prefix, + journal: os.Getenv("JOURNAL_STREAM") != "", + mutex: &sync.Mutex{}, + }) +} diff --git a/daemon_message.go b/daemon_message.go index eade727..7665678 100644 --- a/daemon_message.go +++ b/daemon_message.go @@ -278,6 +278,20 @@ func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() err }) } +// Snapshot holds the queue lock while snapshot locks and copies agent files. +// Its callback must take the memory locks after this lock, as delivery does. +func (m *DaemonMessages) Snapshot(snapshot func() error) error { + m.mutex.Lock() + defer m.mutex.Unlock() + + err := snapshot() + if err != nil { + return fmt.Errorf("copy agent message snapshot: %w", err) + } + + return nil +} + // Tell queues a user message from parent to its own child, then wakes the // child. It is the daemon entry point for parent-to-child messages. func (m *DaemonMessages) Tell(parent, child, text string) error { @@ -308,6 +322,29 @@ func (m *DaemonMessages) Tell(parent, child, text string) error { return nil } +// add appends a queue entry under the mutex. Its ID is never reused. +func (m *DaemonMessages) add(source, target, kind, text string) { + m.state.Next++ + m.state.Pending = append(m.state.Pending, DaemonMessage{ + ID: m.state.Next, + Source: source, + Target: target, + Kind: kind, + Text: text, + }) +} + +// failed stops all loops after a queue failure. Further sends must not change +// the queue, because the outcome of the failed write is not known. +func (m *DaemonMessages) failed(err error) error { + m.failure = err + for _, loop := range m.loops { + loop.fail(err) + } + + return err +} + // RouteWhatsApp queues a routed batch as work and wakes its child. It returns // false for an unrouted source or a missing child. A saved receipt survives // turn acknowledgement until the inbox removes the batch. A repeat after a stop @@ -385,43 +422,6 @@ func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error { return nil } -// add appends a queue entry under the mutex. Its ID is never reused. -func (m *DaemonMessages) add(source, target, kind, text string) { - m.state.Next++ - m.state.Pending = append(m.state.Pending, DaemonMessage{ - ID: m.state.Next, - Source: source, - Target: target, - Kind: kind, - Text: text, - }) -} - -// failed stops all loops after a queue failure. Further sends must not change -// the queue, because the outcome of the failed write is not known. -func (m *DaemonMessages) failed(err error) error { - m.failure = err - for _, loop := range m.loops { - loop.fail(err) - } - - return err -} - -// Snapshot holds the queue lock while snapshot locks and copies agent files. -// Its callback must take the memory locks after this lock, as delivery does. -func (m *DaemonMessages) Snapshot(snapshot func() error) error { - m.mutex.Lock() - defer m.mutex.Unlock() - - err := snapshot() - if err != nil { - return fmt.Errorf("copy agent message snapshot: %w", err) - } - - return nil -} - // Finish acknowledges the queued messages taken by a turn and queues its final // report in the same file replacement. A child with no final text sends // nothing. Call after a final reply, a failure, or a user interrupt, but not diff --git a/daemon_message_helper_test.go b/daemon_message_helper_test.go index 4ac80d3..6b599ae 100644 --- a/daemon_message_helper_test.go +++ b/daemon_message_helper_test.go @@ -58,6 +58,25 @@ func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMe return test } +func (r *DaemonMessageTest) Close(t *testing.T) { + t.Helper() + + if r.cancel != nil { + r.cancel() + + for _, done := range r.done { + require.ErrorIs(t, <-done, context.Canceled) + } + + r.cancel = nil + r.done = nil + } + + require.NoError(t, r.agents.Close()) + + r.agents = nil +} + func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) { t.Helper() @@ -84,22 +103,3 @@ func (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile { return file } - -func (r *DaemonMessageTest) Close(t *testing.T) { - t.Helper() - - if r.cancel != nil { - r.cancel() - - for _, done := range r.done { - require.ErrorIs(t, <-done, context.Canceled) - } - - r.cancel = nil - r.done = nil - } - - require.NoError(t, r.agents.Close()) - - r.agents = nil -} diff --git a/journal.go b/journal.go index 3fb8f64..745ae79 100644 --- a/journal.go +++ b/journal.go @@ -77,6 +77,21 @@ func (j *Journal) Write(day string, line []byte) (Position, error) { return place, nil } +func (j *Journal) closeWriter() error { + if j.file == nil { + return nil + } + + err := j.file.Close() + j.file = nil + j.day = 0 + if err != nil { + return fmt.Errorf("close day file: %w", err) + } + + return nil +} + // Read returns the line at place without its newline. func (j *Journal) Read(place Position) ([]byte, error) { reader, open := j.readers[place.day] @@ -156,21 +171,6 @@ func (j *Journal) Scan(ctx context.Context, place Position, add func([]byte, Pos return nil } -func (j *Journal) closeWriter() error { - if j.file == nil { - return nil - } - - err := j.file.Close() - j.file = nil - j.day = 0 - if err != nil { - return fmt.Errorf("close day file: %w", err) - } - - return nil -} - // journalDayNumber returns the day YYYY-MM-DD as the number YYYYMMDD. func journalDayNumber(day string) (uint32, error) { number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32) diff --git a/log.go b/log.go index 2b9f32f..1ed70e7 100644 --- a/log.go +++ b/log.go @@ -105,6 +105,50 @@ func (l *Log) Close() error { return err } +// catchUp indexes the lines after the last indexed line. +func (l *Log) catchUp(ctx context.Context) error { + start := Position{} + if l.index.count > 0 { + record, _, err := l.index.Read(l.index.count - 1) + if err != nil { + return err + } + + last := decodePosition(record) + start = Position{ + day: last.day, + offset: last.offset + uint64(last.length), + } + } + + 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) + } + + if message.ID != l.index.count { + return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID) + } + + err = logValidateMessage(message) + if err != nil { + return fmt.Errorf("message %d: %w", message.ID, err) + } + + return l.addIndex(place) + }) +} + +func (l *Log) addIndex(place Position) error { + record := make([]byte, positionBytes) + encodePosition(record, place) + + return l.index.Append(record) +} + // Append writes a new message to the log and returns it with its ID, size, and // date. func (l *Log) Append(kind, text string) (Message, error) { @@ -202,50 +246,6 @@ func (l *Log) Message(id int) (Message, error) { return message, nil } -// catchUp indexes the lines after the last indexed line. -func (l *Log) catchUp(ctx context.Context) error { - start := Position{} - if l.index.count > 0 { - record, _, err := l.index.Read(l.index.count - 1) - if err != nil { - return err - } - - last := decodePosition(record) - start = Position{ - day: last.day, - offset: last.offset + uint64(last.length), - } - } - - 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) - } - - if message.ID != l.index.count { - return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID) - } - - err = logValidateMessage(message) - if err != nil { - return fmt.Errorf("message %d: %w", message.ID, err) - } - - return l.addIndex(place) - }) -} - -func (l *Log) addIndex(place Position) error { - record := make([]byte, positionBytes) - encodePosition(record, place) - - return l.index.Append(record) -} - // logValidateMessage makes sure that a message has a known kind, valid UTF-8 // text, the size of its text, and a valid date. func logValidateMessage(message Message) error { diff --git a/memory.go b/memory.go index 5ea9007..49215ac 100644 --- a/memory.go +++ b/memory.go @@ -155,212 +155,6 @@ func (m *Memory) Close() error { return errors.Join(err, m.log.Close()) } -// Append logs a new message, adds it to the tree and the views, and starts the -// builds that are ready, also the failed ones. -func (m *Memory) Append(kind, text string) (Message, error) { - return m.AppendDated(kind, text, "") -} - -// AppendDated appends a message with its RFC 3339 date and builds its nodes. An -// empty date uses the clock, as Append does. -func (m *Memory) AppendDated(kind, text, date string) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.appendDated(kind, text, date) -} - -// appendDated appends under the memory lock and updates the views and builds. -func (m *Memory) appendDated(kind, text, date string) (Message, error) { - message, err := m.log.AppendDated(kind, text, date) - if err != nil { - return Message{}, fmt.Errorf("append dated log message: %w", err) - } - - defer m.syncer.Written() - - err = m.addMessage(message) - if err != nil { - return Message{}, fmt.Errorf("add dated message to memory: %w", err) - } - - err = m.save() - if err != nil { - return Message{}, fmt.Errorf("save dated message view: %w", err) - } - - m.queue.Retry() - m.dispatch() - - return message, nil -} - -// AppendOnce appends a queued message. prepare saves its log ID before the -// append, under the memory lock. A saved ID permits recovery after a stop -// between that save and the append, without a second log entry. prepare must -// not call Memory. The caller stops delivery after a failed append. -func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - if id != nil && *id < m.log.Len() { - message, err := m.log.Message(*id) - if err != nil { - return Message{}, fmt.Errorf("read queued message: %w", err) - } - - if message.Kind != kind || message.Text != text { - return Message{}, errors.New("queued message does not match its log entry") - } - - return message, nil - } - - if id != nil && *id != m.log.Len() { - return Message{}, errors.New("queued message is beyond the log") - } - - err := prepare(m.log.Len()) - if err != nil { - return Message{}, fmt.Errorf("prepare queued message: %w", err) - } - - message, err := m.appendDated(kind, text, "") - if err != nil { - return Message{}, fmt.Errorf("append queued message: %w", err) - } - - return message, nil -} - -// 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 { - return Node{}, Node{}, err - } - - if !built { - return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n) - } - - children[n] = child - } - - return children[0], children[1], nil -} - -// Wait returns when no build runs. The failed builds wait for the next message. -func (m *Memory) Wait(ctx context.Context) error { - for { - m.mutex.Lock() - - running := m.queue.Running() - changed := m.changed - m.mutex.Unlock() - - if running == 0 { - return nil - } - - select { - case <-changed: - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// Settle returns when the chat view has no pending line, so that every message -// so far is summarized. A pending line with no build that runs belongs to a -// failed build, which waits for the next message, so Settle returns -// errMemoryUnsettled in place of a wait that never ends. -func (m *Memory) Settle(ctx context.Context) error { - for { - m.mutex.Lock() - - pending := m.chat.Pending() - running := m.queue.Running() - changed := m.changed - m.mutex.Unlock() - - if !pending { - return nil - } - - if running == 0 { - return errMemoryUnsettled - } - - select { - case <-changed: - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// Flush returns when every write before the call is durable. -func (m *Memory) Flush(ctx context.Context) error { - return m.syncer.Flush(ctx) -} - -// Snapshot flushes all memory writes and runs snapshot while no append or -// summary write can change the log, indexes, tree, or views. snapshot must not -// call Memory. -func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error { - m.mutex.Lock() - defer m.mutex.Unlock() - - err := m.Flush(ctx) - if err != nil { - return fmt.Errorf("flush memory snapshot: %w", err) - } - - err = snapshot() - if err != nil { - return fmt.Errorf("copy memory snapshot: %w", err) - } - - return nil -} - -// ViewLines returns the lines of the chat view, each with its newline, without -// the leaf lines of the messages in skip. -func (m *Memory) ViewLines(skip map[int]bool) []string { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.chat.LinesWithout(skip) -} - -// Len returns the number of messages in the log. -func (m *Memory) Len() int { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.log.Len() -} - -// Message reads the message with the given ID. -func (m *Memory) Message(id int) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.log.Message(id) -} - -// Node reads the node (l, i) if it is built. -func (m *Memory) Node(level, index int) (Node, bool, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.tree.Node(level, index) -} - // load reads the saved views. func (m *Memory) load() error { data, err := os.ReadFile(m.viewPath()) @@ -502,6 +296,56 @@ func (m *Memory) dispatch() { } } +func (m *Memory) viewPath() string { + return filepath.Join(m.root, "view.json") +} + +// build adds node to the tree, keeps it in the views when it is a pending line, +// and settles its parent. +func (m *Memory) build(node Node) error { + err := m.tree.Add(node) + if err != nil { + return err + } + + m.chat.Refresh(node) + m.cview.Refresh(node) + + return m.settle(NodeKey{node.Level, node.Index}) +} + +// settle looks at the parent of the built node key when both children are +// built. It builds each parent that needs no model call, up the tree, and adds +// the first other parent to the queue. +func (m *Memory) settle(key NodeKey) error { + for { + parent := NodeKey{key.level + 1, key.index / 2} + ready, err := m.ready(parent) + if err != nil || !ready { + return err + } + + left, right, err := m.children(parent) + if err != nil { + return err + } + + node, free := freeMerge(left, right) + if !free { + m.queue.AddMerge(parent) + + return nil + } + + err = m.tree.Add(node) + if err != nil { + return err + } + + key = parent + } +} + // addMessage builds the leaf of message when it needs no model call, or else // adds the leaf to the queue. Then it appends the leaf to the views. A crash // can stop an append after the leaf build and before the view save, so a leaf @@ -559,56 +403,6 @@ func (m *Memory) save() error { return nil } -func (m *Memory) viewPath() string { - return filepath.Join(m.root, "view.json") -} - -// build adds node to the tree, keeps it in the views when it is a pending line, -// and settles its parent. -func (m *Memory) build(node Node) error { - err := m.tree.Add(node) - if err != nil { - return err - } - - m.chat.Refresh(node) - m.cview.Refresh(node) - - return m.settle(NodeKey{node.Level, node.Index}) -} - -// settle looks at the parent of the built node key when both children are -// built. It builds each parent that needs no model call, up the tree, and adds -// the first other parent to the queue. -func (m *Memory) settle(key NodeKey) error { - for { - parent := NodeKey{key.level + 1, key.index / 2} - ready, err := m.ready(parent) - if err != nil || !ready { - return err - } - - left, right, err := m.children(parent) - if err != nil { - return err - } - - node, free := freeMerge(left, right) - if !free { - m.queue.AddMerge(parent) - - return nil - } - - err = m.tree.Add(node) - if err != nil { - return err - } - - key = parent - } -} - // task returns the build task of the node key, with its input read from the log // or the tree. func (m *Memory) task(key NodeKey) (Task, error) { @@ -642,6 +436,11 @@ func (m *Memory) task(key NodeKey) (Task, error) { return task, nil } +// failed logs the failed build of the node key with its cause err. +func (m *Memory) failed(key NodeKey, err error) { + m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err) +} + // run builds one node with the summarizer. A failed build stays in the queue // until the next message. func (m *Memory) run(key NodeKey, task Task) { @@ -679,11 +478,6 @@ func (m *Memory) run(key NodeKey, task Task) { m.dispatch() } -// failed logs the failed build of the node key with its cause err. -func (m *Memory) failed(key NodeKey, err error) { - m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err) -} - // ready tells if both children of the node key are built and the node is not. func (m *Memory) ready(key NodeKey) (bool, error) { for _, index := range []int{2 * key.index, 2*key.index + 1} { @@ -698,6 +492,26 @@ func (m *Memory) ready(key NodeKey) (bool, error) { return !built, err } +// 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 { + return Node{}, Node{}, err + } + + if !built { + return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n) + } + + children[n] = child + } + + return children[0], children[1], nil +} + func (m *Memory) context(level, index int) string { end := index if level > 0 { @@ -707,6 +521,192 @@ func (m *Memory) context(level, index int) string { return m.cview.Context(end) } +// Append logs a new message, adds it to the tree and the views, and starts the +// builds that are ready, also the failed ones. +func (m *Memory) Append(kind, text string) (Message, error) { + return m.AppendDated(kind, text, "") +} + +// AppendDated appends a message with its RFC 3339 date and builds its nodes. An +// empty date uses the clock, as Append does. +func (m *Memory) AppendDated(kind, text, date string) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.appendDated(kind, text, date) +} + +// appendDated appends under the memory lock and updates the views and builds. +func (m *Memory) appendDated(kind, text, date string) (Message, error) { + message, err := m.log.AppendDated(kind, text, date) + if err != nil { + return Message{}, fmt.Errorf("append dated log message: %w", err) + } + + defer m.syncer.Written() + + err = m.addMessage(message) + if err != nil { + return Message{}, fmt.Errorf("add dated message to memory: %w", err) + } + + err = m.save() + if err != nil { + return Message{}, fmt.Errorf("save dated message view: %w", err) + } + + m.queue.Retry() + m.dispatch() + + return message, nil +} + +// AppendOnce appends a queued message. prepare saves its log ID before the +// append, under the memory lock. A saved ID permits recovery after a stop +// between that save and the append, without a second log entry. prepare must +// not call Memory. The caller stops delivery after a failed append. +func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + if id != nil && *id < m.log.Len() { + message, err := m.log.Message(*id) + if err != nil { + return Message{}, fmt.Errorf("read queued message: %w", err) + } + + if message.Kind != kind || message.Text != text { + return Message{}, errors.New("queued message does not match its log entry") + } + + return message, nil + } + + if id != nil && *id != m.log.Len() { + return Message{}, errors.New("queued message is beyond the log") + } + + err := prepare(m.log.Len()) + if err != nil { + return Message{}, fmt.Errorf("prepare queued message: %w", err) + } + + message, err := m.appendDated(kind, text, "") + if err != nil { + return Message{}, fmt.Errorf("append queued message: %w", err) + } + + return message, nil +} + +// Wait returns when no build runs. The failed builds wait for the next message. +func (m *Memory) Wait(ctx context.Context) error { + for { + m.mutex.Lock() + + running := m.queue.Running() + changed := m.changed + m.mutex.Unlock() + + if running == 0 { + return nil + } + + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// Settle returns when the chat view has no pending line, so that every message +// so far is summarized. A pending line with no build that runs belongs to a +// failed build, which waits for the next message, so Settle returns +// errMemoryUnsettled in place of a wait that never ends. +func (m *Memory) Settle(ctx context.Context) error { + for { + m.mutex.Lock() + + pending := m.chat.Pending() + running := m.queue.Running() + changed := m.changed + m.mutex.Unlock() + + if !pending { + return nil + } + + if running == 0 { + return errMemoryUnsettled + } + + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// Snapshot flushes all memory writes and runs snapshot while no append or +// summary write can change the log, indexes, tree, or views. snapshot must not +// call Memory. +func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error { + m.mutex.Lock() + defer m.mutex.Unlock() + + err := m.Flush(ctx) + if err != nil { + return fmt.Errorf("flush memory snapshot: %w", err) + } + + err = snapshot() + if err != nil { + return fmt.Errorf("copy memory snapshot: %w", err) + } + + return nil +} + +// Flush returns when every write before the call is durable. +func (m *Memory) Flush(ctx context.Context) error { + return m.syncer.Flush(ctx) +} + +// ViewLines returns the lines of the chat view, each with its newline, without +// the leaf lines of the messages in skip. +func (m *Memory) ViewLines(skip map[int]bool) []string { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.chat.LinesWithout(skip) +} + +// Len returns the number of messages in the log. +func (m *Memory) Len() int { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.log.Len() +} + +// Message reads the message with the given ID. +func (m *Memory) Message(id int) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.log.Message(id) +} + +// Node reads the node (l, i) if it is built. +func (m *Memory) Node(level, index int) (Node, bool, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.tree.Node(level, index) +} + // View returns the chat view as the model reads it. func (m *Memory) View() string { m.mutex.Lock() diff --git a/model.go b/model.go index 074ff60..c443a12 100644 --- a/model.go +++ b/model.go @@ -374,73 +374,6 @@ func (u *ModelUsageLog) Write(record ModelUsageRecord) (err error) { return nil } -// modelEndpoint returns the URL and the authorizer of the endpoint in config. -// The ChatGPT backend uses the token file in dir. The OpenAI API uses the key -// in OPENAI_API_KEY, and there is no fallback from one to the other. All -// clients share the authorizer, so one refresh runs at a time. -func modelEndpoint(config Config, dir string) (string, ModelAuthorizer, error) { - if config.Endpoint == configEndpointAPI { - key := os.Getenv("OPENAI_API_KEY") - if key == "" { - return "", nil, errors.New("endpoint api needs OPENAI_API_KEY") - } - - return modelAPIURL, ModelAPIKey(key), nil - } - - return modelChatGPTURL, newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer), nil -} - -// modelUserMessage returns a user input item with one text part for each text. -func modelUserMessage(texts ...string) any { - content := make([]ModelInputPart, len(texts)) - for n, text := range texts { - content[n] = ModelInputPart{ - Type: "input_text", - Text: text, - } - } - - return modelUserParts(content) -} - -// modelUserParts keeps text and image parts in one user message. -func modelUserParts(content []ModelInputPart) any { - return struct { - Role string `json:"role"` - Content []ModelInputPart `json:"content"` - }{ - Role: "user", - Content: content, - } -} - -// modelOutputText returns the text of the message items of output. -func modelOutputText(output []json.RawMessage) (string, error) { - var text strings.Builder - - for _, raw := range output { - var item ModelOutputItem - - err := json.Unmarshal(raw, &item) - if err != nil { - return "", fmt.Errorf("decode output item: %w", err) - } - - if item.Type != "message" { - continue - } - - for _, part := range item.Content { - if part.Type == "output_text" { - text.WriteString(part.Text) - } - } - } - - return text.String(), nil -} - // modelQuotaHeaders returns the x-codex-* headers, with lowercase names, // without the turn state. func modelQuotaHeaders(header http.Header) map[string]string { @@ -552,3 +485,70 @@ func modelReadEvent(response *ModelResponse, data []byte) (bool, error) { return false, nil } + +// modelOutputText returns the text of the message items of output. +func modelOutputText(output []json.RawMessage) (string, error) { + var text strings.Builder + + for _, raw := range output { + var item ModelOutputItem + + err := json.Unmarshal(raw, &item) + if err != nil { + return "", fmt.Errorf("decode output item: %w", err) + } + + if item.Type != "message" { + continue + } + + for _, part := range item.Content { + if part.Type == "output_text" { + text.WriteString(part.Text) + } + } + } + + return text.String(), nil +} + +// modelEndpoint returns the URL and the authorizer of the endpoint in config. +// The ChatGPT backend uses the token file in dir. The OpenAI API uses the key +// in OPENAI_API_KEY, and there is no fallback from one to the other. All +// clients share the authorizer, so one refresh runs at a time. +func modelEndpoint(config Config, dir string) (string, ModelAuthorizer, error) { + if config.Endpoint == configEndpointAPI { + key := os.Getenv("OPENAI_API_KEY") + if key == "" { + return "", nil, errors.New("endpoint api needs OPENAI_API_KEY") + } + + return modelAPIURL, ModelAPIKey(key), nil + } + + return modelChatGPTURL, newChatGPTAuth(filepath.Join(dir, chatgptTokenFileName), chatgptIssuer), nil +} + +// modelUserMessage returns a user input item with one text part for each text. +func modelUserMessage(texts ...string) any { + content := make([]ModelInputPart, len(texts)) + for n, text := range texts { + content[n] = ModelInputPart{ + Type: "input_text", + Text: text, + } + } + + return modelUserParts(content) +} + +// modelUserParts keeps text and image parts in one user message. +func modelUserParts(content []ModelInputPart) any { + return struct { + Role string `json:"role"` + Content []ModelInputPart `json:"content"` + }{ + Role: "user", + Content: content, + } +} diff --git a/model_fake_helper_test.go b/model_fake_helper_test.go index f9d1451..af94776 100644 --- a/model_fake_helper_test.go +++ b/model_fake_helper_test.go @@ -49,22 +49,6 @@ func newModelFake(t *testing.T) *ModelFake { return f } -// Call returns call number n. -func (f *ModelFake) Call(n int) ModelFakeCall { - f.mutex.Lock() - defer f.mutex.Unlock() - - return f.calls[n] -} - -// Calls returns the number of calls. -func (f *ModelFake) Calls() int { - f.mutex.Lock() - defer f.mutex.Unlock() - - return len(f.calls) -} - func (f *ModelFake) serve(w http.ResponseWriter, r *http.Request) { raw, err := io.ReadAll(r.Body) if err != nil { @@ -95,12 +79,20 @@ func (f *ModelFake) serve(w http.ResponseWriter, r *http.Request) { reply(w, call) } -func modelTestRequest() ModelRequest { - return ModelRequest{ - Model: "gpt-test", - Effort: "low", - Input: []any{modelUserMessage("hello")}, - } +// Call returns call number n. +func (f *ModelFake) Call(n int) ModelFakeCall { + f.mutex.Lock() + defer f.mutex.Unlock() + + return f.calls[n] +} + +// Calls returns the number of calls. +func (f *ModelFake) Calls() int { + f.mutex.Lock() + defer f.mutex.Unlock() + + return len(f.calls) } // modelWriteEvents writes events as a server-sent event stream. @@ -120,16 +112,6 @@ func modelWriteEvents(w http.ResponseWriter, events ...any) { } } -func modelMessageItem(text string) map[string]any { - return map[string]any{ - "type": "message", - "role": "assistant", - "content": []any{ - map[string]any{"type": "output_text", "text": text}, - }, - } -} - func modelCompletedEvent(output []any, input, cached, out int) map[string]any { if output == nil { output = []any{} @@ -148,6 +130,24 @@ func modelCompletedEvent(output []any, input, cached, out int) map[string]any { } } +func modelMessageItem(text string) map[string]any { + return map[string]any{ + "type": "message", + "role": "assistant", + "content": []any{ + map[string]any{"type": "output_text", "text": text}, + }, + } +} + +func modelTestRequest() ModelRequest { + return ModelRequest{ + Model: "gpt-test", + Effort: "low", + Input: []any{modelUserMessage("hello")}, + } +} + func modelReadUsage(t *testing.T, path string) []ModelUsageRecord { t.Helper() diff --git a/prompt.go b/prompt.go index 896038b..31f57cb 100644 --- a/prompt.go +++ b/prompt.go @@ -197,19 +197,6 @@ to see its pixels.` // promptEfforts are the reasoning efforts that spawn can give a subagent. var promptEfforts = []string{"low", "medium", "high", "xhigh"} -// prompt returns the system prompt of the reference with one added paragraph on -// the user's instructions file at instructionsPath. It uses the agent name and -// ends with the user's instructions. Without instructions it has neither them -// nor the clause that points to them, but keeps the file paragraph. Turns and -// compactions send the same prompt and tools, so compactions read the prompt -// cache of the turns. -func prompt(name, instructionsPath, instructions string) string { - text := strings.NewReplacer(promptName, name, promptInstructionsPath, instructionsPath).Replace(promptText) - text = strings.Replace(text, "\n\n# Compactions", promptChildrenText+"\n\n# Compactions", 1) - - return promptWithInstructions(text, promptTaskSentence, promptInstructionsClause, instructions) -} - // promptChild returns the prompt of a child with its area and reporting rules. // Its instructions file belongs to the child, not to the parent. func promptChild(name, parent, responsibility, instructionsPath, instructions string) string { @@ -229,6 +216,33 @@ can ask the user before sending.`, 1) return promptWithInstructions(text, "Do the tasks of your area yourself, with your tools.", promptInstructionsClause, instructions) } +// prompt returns the system prompt of the reference with one added paragraph on +// the user's instructions file at instructionsPath. It uses the agent name and +// ends with the user's instructions. Without instructions it has neither them +// nor the clause that points to them, but keeps the file paragraph. Turns and +// compactions send the same prompt and tools, so compactions read the prompt +// cache of the turns. +func prompt(name, instructionsPath, instructions string) string { + text := strings.NewReplacer(promptName, name, promptInstructionsPath, instructionsPath).Replace(promptText) + text = strings.Replace(text, "\n\n# Compactions", promptChildrenText+"\n\n# Compactions", 1) + + return promptWithInstructions(text, promptTaskSentence, promptInstructionsClause, instructions) +} + +// promptWithInstructions returns text with the instructions of the user at its +// end, after the heading. With instructions, sentence gets clause, which points +// to them. Without instructions the prompt has neither them nor the clause. +func promptWithInstructions(text, sentence, clause, instructions string) string { + instructions = strings.TrimSpace(instructions) + if instructions == "" { + return text + } + + text = strings.Replace(text, sentence, strings.TrimSuffix(sentence, ".")+clause+".", 1) + + return text + promptInstructionsHeading + instructions +} + // promptSubagent returns the system prompt of a subagent of the agent name, // with the instructions of the user at its end, as prompt does. func promptSubagent(name, instructions string) string { @@ -421,20 +435,6 @@ func promptSubagentTools() []ModelTool { } } -// promptWithInstructions returns text with the instructions of the user at its -// end, after the heading. With instructions, sentence gets clause, which points -// to them. Without instructions the prompt has neither them nor the clause. -func promptWithInstructions(text, sentence, clause, instructions string) string { - instructions = strings.TrimSpace(instructions) - if instructions == "" { - return text - } - - text = strings.Replace(text, sentence, strings.TrimSuffix(sentence, ".")+clause+".", 1) - - return text + promptInstructionsHeading + instructions -} - // promptObjectSchema returns a strict JSON schema of an object whose properties // are all required. func promptObjectSchema(properties map[string]any) map[string]any { @@ -464,10 +464,10 @@ func promptEffortEnum() []any { return append(values, nil) } -// promptIntegerSchema returns the JSON schema of an integer. -func promptIntegerSchema(description string) map[string]any { +// promptStringSchema returns the JSON schema of a string. +func promptStringSchema(description string) map[string]any { return map[string]any{ - "type": "integer", + "type": "string", "description": description, } } @@ -480,10 +480,10 @@ func promptNullableStringSchema(description string) map[string]any { } } -// promptStringSchema returns the JSON schema of a string. -func promptStringSchema(description string) map[string]any { +// promptIntegerSchema returns the JSON schema of an integer. +func promptIntegerSchema(description string) map[string]any { return map[string]any{ - "type": "string", + "type": "integer", "description": description, } } diff --git a/records.go b/records.go index 52824cb..dbfafe2 100644 --- a/records.go +++ b/records.go @@ -65,6 +65,11 @@ func (r *Records) Read(n int) ([]byte, bool, error) { return record, true, nil } +// Append writes the record after the last one. +func (r *Records) Append(record []byte) error { + return r.Write(r.count, record) +} + // Write writes record n. func (r *Records) Write(n int, record []byte) error { _, err := r.file.WriteAt(record, int64(n*r.size)) @@ -77,11 +82,6 @@ func (r *Records) Write(n int, record []byte) error { return nil } -// Append writes the record after the last one. -func (r *Records) Append(record []byte) error { - return r.Write(r.count, record) -} - // Close closes the file. func (r *Records) Close() error { return r.file.Close() diff --git a/schedule.go b/schedule.go index 5fa35da..7eb747f 100644 --- a/schedule.go +++ b/schedule.go @@ -141,6 +141,20 @@ func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logg return s, nil } +// save writes the file whole, with mode 0600. The caller holds the mutex. +func (s *Schedules) save() error { + data, err := json.Marshal(s.file) + if err == nil { + err = writeFileAtomic(s.path, append(data, '\n')) + } + + if err != nil { + return fmt.Errorf("save schedules: %w", err) + } + + return nil +} + // Run starts the runs of the due schedules in l until ctx ends. It waits until // the first due time, at most scheduleClockWaitTimeout at a time, and also // wakes at a set or a remove. So a schedule that came due while the daemon was @@ -177,116 +191,6 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { } } -// Set adds a schedule with text and returns the answer of the tool. Exactly one -// of at and cron is not nil. at is a local time in scheduleAtLayout after now, -// and it goes through scheduleWallTime, so a time in the hour that the clock -// repeats is its first occurrence. cron is an expression of scheduleParseCron. -func (s *Schedules) Set(text string, at, cron *string) (string, error) { - text = strings.TrimSpace(text) - if text == "" || len(text) > scheduleTextBytes { - return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes) - } - - now := s.clock() - schedule := Schedule{ - Text: text, - Owner: s.owner, - } - - switch { - case (at == nil) == (cron == nil): - return "", errors.New("give either at or cron, and null for the other") - case at != nil: - parsed, err := time.Parse(scheduleAtLayout, *at) - if err != nil { - return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout)) - } - - schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location()) - if !schedule.Due.After(now) { - return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout)) - } - default: - parsed, err := scheduleParseCron(*cron) - if err != nil { - return "", err - } - - next, found := parsed.Next(now) - if !found { - return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears) - } - - schedule.Cron = parsed.String() - schedule.Due = next - } - - s.mutex.Lock() - defer s.mutex.Unlock() - - s.file.Next++ - schedule.ID = s.file.Next - s.file.Schedules = append(s.file.Schedules, schedule) - err := s.save() - if err != nil { - return "", err - } - - s.change() - - return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil -} - -// List returns the answer of the tool: one line for each schedule, by due time, -// with its ID, its next time, its cron expression or once, and its text on one -// line. -func (s *Schedules) List() string { - location := s.clock().Location() - schedules := s.sorted() - if len(schedules) == 0 { - return "No schedules." - } - - lines := make([]string, 0, len(schedules)) - for _, schedule := range schedules { - lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " ")) - } - - return strings.Join(lines, "\n") -} - -// Len returns the number of schedules. -func (s *Schedules) Len() int { - s.mutex.Lock() - defer s.mutex.Unlock() - - return len(s.file.Schedules) -} - -// Remove removes the schedule id and returns the answer of the tool. -func (s *Schedules) Remove(id int) (string, error) { - s.mutex.Lock() - defer s.mutex.Unlock() - - index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool { - return schedule.ID == id - }) - - if index < 0 { - return "", fmt.Errorf("no schedule %d", id) - } - - s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1) - err := s.save() - if err != nil { - return "", err - } - - s.change() - - return fmt.Sprintf("Schedule %d removed.", id), nil -} - // deliver logs the message of each due schedule in l, in due order, as a work // message of scheduleSource, and wakes the loop once with all of them. A // schedule ends its run with Done after its message is durable in the log. A @@ -339,6 +243,17 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { return nil } +// Wait returns how long after now the first schedule is due, or false when +// there is no schedule. +func (s *Schedules) Wait(now time.Time) (time.Duration, bool) { + schedules := s.sorted() + if len(schedules) == 0 { + return 0, false + } + + return max(schedules[0].Due.Sub(now), 0), true +} + // Due returns the schedules whose time is not after now, by due time. func (s *Schedules) Due(now time.Time) []Schedule { var due []Schedule @@ -391,15 +306,78 @@ func (s *Schedules) Done(schedule Schedule, now time.Time) error { return s.save() } -// Wait returns how long after now the first schedule is due, or false when -// there is no schedule. -func (s *Schedules) Wait(now time.Time) (time.Duration, bool) { - schedules := s.sorted() - if len(schedules) == 0 { - return 0, false +// 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() + + slices.SortFunc(schedules, func(a, b Schedule) int { + return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID)) + }) + + return schedules +} + +// Set adds a schedule with text and returns the answer of the tool. Exactly one +// of at and cron is not nil. at is a local time in scheduleAtLayout after now, +// and it goes through scheduleWallTime, so a time in the hour that the clock +// repeats is its first occurrence. cron is an expression of scheduleParseCron. +func (s *Schedules) Set(text string, at, cron *string) (string, error) { + text = strings.TrimSpace(text) + if text == "" || len(text) > scheduleTextBytes { + return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes) } - return max(schedules[0].Due.Sub(now), 0), true + now := s.clock() + schedule := Schedule{ + Text: text, + Owner: s.owner, + } + + switch { + case (at == nil) == (cron == nil): + return "", errors.New("give either at or cron, and null for the other") + case at != nil: + parsed, err := time.Parse(scheduleAtLayout, *at) + if err != nil { + return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout)) + } + + schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location()) + if !schedule.Due.After(now) { + return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout)) + } + default: + parsed, err := scheduleParseCron(*cron) + if err != nil { + return "", err + } + + next, found := parsed.Next(now) + if !found { + return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears) + } + + schedule.Cron = parsed.String() + schedule.Due = next + } + + s.mutex.Lock() + defer s.mutex.Unlock() + + s.file.Next++ + schedule.ID = s.file.Next + s.file.Schedules = append(s.file.Schedules, schedule) + err := s.save() + if err != nil { + return "", err + } + + s.change() + + return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil } // change wakes Run after a set or a remove. @@ -410,32 +388,54 @@ 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() +// List returns the answer of the tool: one line for each schedule, by due time, +// with its ID, its next time, its cron expression or once, and its text on one +// line. +func (s *Schedules) List() string { + location := s.clock().Location() + schedules := s.sorted() + if len(schedules) == 0 { + return "No schedules." + } - schedules := slices.Clone(s.file.Schedules) - s.mutex.Unlock() + lines := make([]string, 0, len(schedules)) + for _, schedule := range schedules { + lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " ")) + } - slices.SortFunc(schedules, func(a, b Schedule) int { - return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID)) - }) + return strings.Join(lines, "\n") +} - return schedules +// Len returns the number of schedules. +func (s *Schedules) Len() int { + s.mutex.Lock() + defer s.mutex.Unlock() + + return len(s.file.Schedules) } -// save writes the file whole, with mode 0600. The caller holds the mutex. -func (s *Schedules) save() error { - data, err := json.Marshal(s.file) - if err == nil { - err = writeFileAtomic(s.path, append(data, '\n')) +// Remove removes the schedule id and returns the answer of the tool. +func (s *Schedules) Remove(id int) (string, error) { + s.mutex.Lock() + defer s.mutex.Unlock() + + index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool { + return schedule.ID == id + }) + + if index < 0 { + return "", fmt.Errorf("no schedule %d", id) } + s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1) + err := s.save() if err != nil { - return fmt.Errorf("save schedules: %w", err) + return "", err } - return nil + s.change() + + return fmt.Sprintf("Schedule %d removed.", id), nil } // Message returns the text of a run at now for the log, with the times in the diff --git a/schedule_cron.go b/schedule_cron.go index 3255f99..b982f37 100644 --- a/schedule_cron.go +++ b/schedule_cron.go @@ -154,6 +154,31 @@ func (c ScheduleCron) gap() int { return gap } +// scheduleWallTime returns the time of a wall time in location. A wall time +// that the clock skips, at the start of daylight saving time, is later by the +// length of the skip: 02:30 becomes 03:30 when the clock goes from 02:00 to +// 03:00. A wall time that the clock repeats, at the end of daylight saving +// time, is its first occurrence, as in cron. +func scheduleWallTime(year int, month time.Month, day, hour, minute int, location *time.Location) time.Time { + t := time.Date(year, month, day, hour, minute, 0, 0, location) + _, before := t.Add(-3 * time.Hour).Zone() + _, offset := t.Zone() + if t.Hour() != hour || t.Minute() != minute || t.Day() != day { + wall := time.Date(year, month, day, hour, minute, 0, 0, time.UTC) + + return wall.Add(-time.Duration(before) * time.Second).In(location) + } + + if before > offset { + earlier := t.Add(-time.Duration(before-offset) * time.Second) + if earlier.Day() == day && earlier.Hour() == hour && earlier.Minute() == minute { + return earlier + } + } + + return t +} + // scheduleParseCron parses a cron expression of five fields: minute, hour, day // of month, month, and day of week, where 0 and 7 are Sunday. A field is a // comma list of items, and an item is *, N, or N-M, where * and N-M can have a @@ -209,31 +234,6 @@ func scheduleParseCron(expression string) (ScheduleCron, error) { return c, nil } -// scheduleWallTime returns the time of a wall time in location. A wall time -// that the clock skips, at the start of daylight saving time, is later by the -// length of the skip: 02:30 becomes 03:30 when the clock goes from 02:00 to -// 03:00. A wall time that the clock repeats, at the end of daylight saving -// time, is its first occurrence, as in cron. -func scheduleWallTime(year int, month time.Month, day, hour, minute int, location *time.Location) time.Time { - t := time.Date(year, month, day, hour, minute, 0, 0, location) - _, before := t.Add(-3 * time.Hour).Zone() - _, offset := t.Zone() - if t.Hour() != hour || t.Minute() != minute || t.Day() != day { - wall := time.Date(year, month, day, hour, minute, 0, 0, time.UTC) - - return wall.Add(-time.Duration(before) * time.Second).In(location) - } - - if before > offset { - earlier := t.Add(-time.Duration(before-offset) * time.Second) - if earlier.Day() == day && earlier.Hour() == hour && earlier.Minute() == minute { - return earlier - } - } - - return t -} - // scheduleCronField returns the set of the values of one field, from low to // high. func scheduleCronField(field string, low, high int) (uint64, error) { diff --git a/subagent.go b/subagent.go index 437aeb5..37e9b3a 100644 --- a/subagent.go +++ b/subagent.go @@ -95,39 +95,6 @@ func (s *Subagents) Create(ctx context.Context) (*Subagent, error) { return nil, fmt.Errorf("no free subagent name after %d draws", subagentNameDraws) } -// Reserve keeps a child name out of future subagent draws. An existing subagent -// directory or a reserved name is refused. -func (s *Subagents) Reserve(name string) error { - s.mutex.Lock() - defer s.mutex.Unlock() - - if s.reserved[name] { - return fmt.Errorf("name %q is already used", name) - } - - _, err := os.Lstat(filepath.Join(s.dir, name)) - if err == nil { - return fmt.Errorf("name %q is already used by a subagent", name) - } - - if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("check subagent name: %w", err) - } - - s.reserved[name] = true - - return nil -} - -// Release frees a reservation after a child fails to publish. It does not -// remove a subagent directory or change an open log. -func (s *Subagents) Release(name string) { - s.mutex.Lock() - defer s.mutex.Unlock() - - delete(s.reserved, name) -} - // Get returns the subagent name with its log open. A name that is not valid or // that has no directory is an unknown agent. func (s *Subagents) Get(ctx context.Context, name string) (*Subagent, error) { @@ -180,6 +147,39 @@ func (s *Subagents) Get(ctx context.Context, name string) (*Subagent, error) { return subagent, nil } +// Reserve keeps a child name out of future subagent draws. An existing subagent +// directory or a reserved name is refused. +func (s *Subagents) Reserve(name string) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + if s.reserved[name] { + return fmt.Errorf("name %q is already used", name) + } + + _, err := os.Lstat(filepath.Join(s.dir, name)) + if err == nil { + return fmt.Errorf("name %q is already used by a subagent", name) + } + + if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("check subagent name: %w", err) + } + + s.reserved[name] = true + + return nil +} + +// Release frees a reservation after a child fails to publish. It does not +// remove a subagent directory or change an open log. +func (s *Subagents) Release(name string) { + s.mutex.Lock() + defer s.mutex.Unlock() + + delete(s.reserved, name) +} + // Snapshot flushes and locks the open logs while snapshot reads the subagent // files. snapshot must not call Subagents or an open Subagent. func (s *Subagents) Snapshot(ctx context.Context, snapshot func() error) error { @@ -220,6 +220,19 @@ func (s *Subagents) Close() error { return err } +// Flush returns when every write before the call is durable. +func (a *Subagent) Flush(ctx context.Context) error { + return a.syncer.Flush(ctx) +} + +// close makes the writes durable and closes the log. +func (a *Subagent) close() error { + a.mutex.Lock() + defer a.mutex.Unlock() + + return errors.Join(a.syncer.Close(), a.log.Close()) +} + // Append logs a message of the subagent. func (a *Subagent) Append(kind, text string) (Message, error) { a.mutex.Lock() @@ -235,11 +248,6 @@ func (a *Subagent) Append(kind, text string) (Message, error) { return message, nil } -// Flush returns when every write before the call is durable. -func (a *Subagent) Flush(ctx context.Context) error { - return a.syncer.Flush(ctx) -} - // First returns the first message of the log. func (a *Subagent) First() (Message, error) { a.mutex.Lock() @@ -272,11 +280,3 @@ func (a *Subagent) Show() (string, error) { func (a *Subagent) WorkDir() string { return filepath.Join(a.dir, "work") } - -// close makes the writes durable and closes the log. -func (a *Subagent) close() error { - a.mutex.Lock() - defer a.mutex.Unlock() - - return errors.Join(a.syncer.Close(), a.log.Close()) -} diff --git a/subagent_names.go b/subagent_names.go index 3b9f3a5..5e49d2a 100644 --- a/subagent_names.go +++ b/subagent_names.go @@ -44,6 +44,11 @@ func subagentNameRandom() string { return subagentWord(adjective) + subagentWord(noun) } +// subagentWord returns word with its first letter in upper case. +func subagentWord(word string) string { + return strings.ToUpper(word[:1]) + word[1:] +} + // subagentNameValid tells if name is an adjective and a noun of the word lists, // each capitalized, joined. Only such a name can be a directory of a subagent. func subagentNameValid(name string) bool { @@ -62,8 +67,3 @@ func subagentNameValid(name string) bool { return false } - -// subagentWord returns word with its first letter in upper case. -func subagentWord(word string) string { - return strings.ToUpper(word[:1]) + word[1:] -} diff --git a/subagent_run.go b/subagent_run.go index 1da8588..24e4f82 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -152,155 +152,6 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string return "Started " + strings.Join(names, ", ") + ". Each one's report will reach you as a message when it finishes.", nil } -// Tell sends message to the subagent name. A running subagent reads it between -// its tool calls, and its report answers it. A finished subagent goes on from -// its log with a new run, with the view of now and the effort of its note. -func (r *SubagentRuns) Tell(ctx context.Context, name, message string) (string, error) { - r.mutex.Lock() - - run, found := r.running[name] - if found { - defer r.mutex.Unlock() - - told, err := run.subagent.Append("user", message) - if err != nil { - return "", err - } - - run.pending = append(run.pending, told) - - return "Sent. " + name + " reads it between its tool calls, and its report answers it.", nil - } - - closed := r.closed - r.mutex.Unlock() - - if closed { - return "", errSubagentClosed - } - - subagent, err := r.subagents.Get(ctx, name) - if err != nil { - return "", err - } - - effort, err := r.effort(subagent) - if err != nil { - return "", err - } - - err = r.memory.Settle(ctx) - if err != nil { - return "", fmt.Errorf("wait for the summaries: %w", err) - } - - view, err := r.layout.Split(r.memory.ViewLines(nil)) - if err != nil { - return "", err - } - - _, err = subagent.Append("user", message) - if err == nil { - err = subagent.Flush(ctx) - } - - if err != nil { - return "", err - } - - err = r.start(ctx, &SubagentRun{ - subagent: subagent, - view: view, - effort: effort, - }) - if err != nil { - return "", err - } - - return "Sent. " + name + "'s reply will reach you as a message.", nil -} - -// Stop ends the run of the subagent name and tells if it ran. The chat gets the -// note "Stopped by the user: NAME." under the mutex, so the note comes before -// the report of the run. -func (r *SubagentRuns) Stop(name string) (bool, error) { - r.mutex.Lock() - defer r.mutex.Unlock() - - run, found := r.running[name] - if !found { - return false, nil - } - - err := r.note("Stopped by the user: " + name + ".") - if err != nil { - return true, err - } - - run.stopped = true - run.cancel() - - return true, nil -} - -// Resume goes on with the runs that a stop of the daemon cut, as work.json -// names them. The chat gets the note "The daemon restarted, cutting NAMES; -// nothing was lost: go on.", and each cut subagent gets the same text as a -// message of Tell. A subagent that cannot go on gets a failed report. -func (r *SubagentRuns) Resume(ctx context.Context) error { - names, err := r.load() - if err != nil || len(names) == 0 { - return err - } - - text := "The daemon restarted, cutting " + strings.Join(names, ", ") + "; nothing was lost: go on." - err = r.note(text) - if err != nil { - return err - } - - for _, name := range names { - _, err = r.Tell(ctx, name, text) - if err != nil { - r.logger.Error("subagent lost at restart", "name", name, "err", err) - r.report(name, subagentReport(name, []string{"failed: " + err.Error()}), subagentFailed) - } - } - - // A subagent that cannot go on does not run, so work.json no longer - // names it, and the next start does not try it again. - r.mutex.Lock() - defer r.mutex.Unlock() - - return r.save() -} - -// Snapshot locks the run state, memory, and subagent logs while snapshot reads -// the agent files. The lock order agrees with Stop, which logs under the run -// lock. -func (r *SubagentRuns) Snapshot(ctx context.Context, snapshot func() error) error { - r.mutex.Lock() - defer r.mutex.Unlock() - - return r.memory.Snapshot(ctx, func() error { - return r.subagents.Snapshot(ctx, snapshot) - }) -} - -// Close ends every run and waits for them. No subagent starts after Close. -func (r *SubagentRuns) Close() { - r.mutex.Lock() - - r.closed = true - for _, run := range r.running { - run.cancel() - } - - r.mutex.Unlock() - - r.workers.Wait() -} - // create makes a new subagent with its note, its task, and its work directory, // and starts its run. func (r *SubagentRuns) create(ctx context.Context, note, task string, view []string, effort string) (string, error) { @@ -333,26 +184,6 @@ func (r *SubagentRuns) create(ctx context.Context, note, task string, view []str }) } -// effort returns the effort of subagent from the note at the start of its log: -// the effort of the agent, or the effort that the note names. -func (r *SubagentRuns) effort(subagent *Subagent) (string, error) { - note, err := subagent.First() - if err != nil { - return "", err - } - - effort, own := strings.CutPrefix(note.Text, subagentNote+subagentNoteEffort) - switch { - case note.Kind != "note": - case own: - return effort, nil - case note.Text == subagentNote: - return r.agent.config.Effort, nil - } - - return "", fmt.Errorf("the log of %s does not start with a subagent note", subagent.name) -} - // start runs run in the background and logs its start. Its context keeps the // values of ctx, but only Stop and Close end it. func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { @@ -397,27 +228,6 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { return nil } -// load reads the names of work.json. A missing file names none. -func (r *SubagentRuns) load() ([]string, error) { - data, err := os.ReadFile(r.workPath()) - if errors.Is(err, os.ErrNotExist) { - return nil, nil - } - - if err != nil { - return nil, fmt.Errorf("read running subagents: %w", err) - } - - var file SubagentWorkFile - - err = json.Unmarshal(data, &file) - if err != nil { - return nil, fmt.Errorf("decode running subagents: %w", err) - } - - return file.Running, nil -} - // save writes the names of the running subagents to work.json. The caller holds // the mutex. func (r *SubagentRuns) save() error { @@ -438,10 +248,6 @@ func (r *SubagentRuns) save() error { return nil } -func (r *SubagentRuns) workPath() string { - return filepath.Join(r.agent.dir, subagentWorkFileName) -} - // run runs the chains of run and reports and logs its end with the length of // the run, unless Close ended it. A chain follows while messages wait after the // final answer of the chain before it, and no chain failed. A chain that Stop @@ -515,6 +321,10 @@ func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { r.report(run.subagent.name, subagentReport(run.subagent.name, replies), status) } +func (r *SubagentRuns) workPath() string { + return filepath.Join(r.agent.dir, subagentWorkFileName) +} + // chain runs one request chain of run and returns its final answer. func (r *SubagentRuns) chain(ctx context.Context, run *SubagentRun) (string, error) { instructions, err := r.agent.SubagentPrompt() @@ -570,6 +380,196 @@ func (r *SubagentRuns) take(run *SubagentRun) []Message { return messages } +// Stop ends the run of the subagent name and tells if it ran. The chat gets the +// note "Stopped by the user: NAME." under the mutex, so the note comes before +// the report of the run. +func (r *SubagentRuns) Stop(name string) (bool, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + + run, found := r.running[name] + if !found { + return false, nil + } + + err := r.note("Stopped by the user: " + name + ".") + if err != nil { + return true, err + } + + run.stopped = true + run.cancel() + + return true, nil +} + +// Resume goes on with the runs that a stop of the daemon cut, as work.json +// names them. The chat gets the note "The daemon restarted, cutting NAMES; +// nothing was lost: go on.", and each cut subagent gets the same text as a +// message of Tell. A subagent that cannot go on gets a failed report. +func (r *SubagentRuns) Resume(ctx context.Context) error { + names, err := r.load() + if err != nil || len(names) == 0 { + return err + } + + text := "The daemon restarted, cutting " + strings.Join(names, ", ") + "; nothing was lost: go on." + err = r.note(text) + if err != nil { + return err + } + + for _, name := range names { + _, err = r.Tell(ctx, name, text) + if err != nil { + r.logger.Error("subagent lost at restart", "name", name, "err", err) + r.report(name, subagentReport(name, []string{"failed: " + err.Error()}), subagentFailed) + } + } + + // A subagent that cannot go on does not run, so work.json no longer + // names it, and the next start does not try it again. + r.mutex.Lock() + defer r.mutex.Unlock() + + return r.save() +} + +// load reads the names of work.json. A missing file names none. +func (r *SubagentRuns) load() ([]string, error) { + data, err := os.ReadFile(r.workPath()) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + + if err != nil { + return nil, fmt.Errorf("read running subagents: %w", err) + } + + var file SubagentWorkFile + + err = json.Unmarshal(data, &file) + if err != nil { + return nil, fmt.Errorf("decode running subagents: %w", err) + } + + return file.Running, nil +} + +// Tell sends message to the subagent name. A running subagent reads it between +// its tool calls, and its report answers it. A finished subagent goes on from +// its log with a new run, with the view of now and the effort of its note. +func (r *SubagentRuns) Tell(ctx context.Context, name, message string) (string, error) { + r.mutex.Lock() + + run, found := r.running[name] + if found { + defer r.mutex.Unlock() + + told, err := run.subagent.Append("user", message) + if err != nil { + return "", err + } + + run.pending = append(run.pending, told) + + return "Sent. " + name + " reads it between its tool calls, and its report answers it.", nil + } + + closed := r.closed + r.mutex.Unlock() + + if closed { + return "", errSubagentClosed + } + + subagent, err := r.subagents.Get(ctx, name) + if err != nil { + return "", err + } + + effort, err := r.effort(subagent) + if err != nil { + return "", err + } + + err = r.memory.Settle(ctx) + if err != nil { + return "", fmt.Errorf("wait for the summaries: %w", err) + } + + view, err := r.layout.Split(r.memory.ViewLines(nil)) + if err != nil { + return "", err + } + + _, err = subagent.Append("user", message) + if err == nil { + err = subagent.Flush(ctx) + } + + if err != nil { + return "", err + } + + err = r.start(ctx, &SubagentRun{ + subagent: subagent, + view: view, + effort: effort, + }) + if err != nil { + return "", err + } + + return "Sent. " + name + "'s reply will reach you as a message.", nil +} + +// effort returns the effort of subagent from the note at the start of its log: +// the effort of the agent, or the effort that the note names. +func (r *SubagentRuns) effort(subagent *Subagent) (string, error) { + note, err := subagent.First() + if err != nil { + return "", err + } + + effort, own := strings.CutPrefix(note.Text, subagentNote+subagentNoteEffort) + switch { + case note.Kind != "note": + case own: + return effort, nil + case note.Text == subagentNote: + return r.agent.config.Effort, nil + } + + return "", fmt.Errorf("the log of %s does not start with a subagent note", subagent.name) +} + +// Snapshot locks the run state, memory, and subagent logs while snapshot reads +// the agent files. The lock order agrees with Stop, which logs under the run +// lock. +func (r *SubagentRuns) Snapshot(ctx context.Context, snapshot func() error) error { + r.mutex.Lock() + defer r.mutex.Unlock() + + return r.memory.Snapshot(ctx, func() error { + return r.subagents.Snapshot(ctx, snapshot) + }) +} + +// Close ends every run and waits for them. No subagent starts after Close. +func (r *SubagentRuns) Close() { + r.mutex.Lock() + + r.closed = true + for _, run := range r.running { + run.cancel() + } + + r.mutex.Unlock() + + r.workers.Wait() +} + // subagentReport returns the report of the subagent name with its replies: the // name in brackets, a new line, the replies, and the zoom that gives its whole // chat. diff --git a/subagent_run_test.go b/subagent_run_test.go index 389a4bf..30fcc78 100644 --- a/subagent_run_test.go +++ b/subagent_run_test.go @@ -531,6 +531,35 @@ func TestRestartReportsASubagentThatCannotGoOn(t *testing.T) { require.JSONEq(t, `{"running": []}`, string(data)) } +// subagentTestLastItem returns the last input item of call. +func subagentTestLastItem(call ModelFakeCall) map[string]any { + input := call.body["input"].([]any) + + return input[len(input)-1].(map[string]any) +} + +// subagentTestCall tells if call is a call of a subagent. +func subagentTestCall(call ModelFakeCall) bool { + instructions, _ := call.body["instructions"].(string) + + return strings.HasPrefix(instructions, "You are a subagent") +} + +// subagentTestText returns the text of the user message item, or "" for 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) + } + + return text.String() +} + // subagentTestStart runs a loop of agent with its memory and subagents in the // agent directory, as the daemon does. The returned function stops the loop and // closes the files. @@ -577,32 +606,3 @@ func subagentTestChat(t *testing.T, memory *Memory) []string { return lines } - -// subagentTestCall tells if call is a call of a subagent. -func subagentTestCall(call ModelFakeCall) bool { - instructions, _ := call.body["instructions"].(string) - - return strings.HasPrefix(instructions, "You are a subagent") -} - -// subagentTestLastItem returns the last input item of call. -func subagentTestLastItem(call ModelFakeCall) map[string]any { - input := call.body["input"].([]any) - - return input[len(input)-1].(map[string]any) -} - -// subagentTestText returns the text of the user message item, or "" for 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) - } - - return text.String() -} diff --git a/syncer.go b/syncer.go index 196eb97..bf2a5ff 100644 --- a/syncer.go +++ b/syncer.go @@ -67,51 +67,6 @@ func (s *Syncer) Written() { s.start() } -// 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 - s.mutex.Unlock() - - if reached { - return nil - } - - if failure != nil { - return failure - } - - select { - case <-done: - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// Close makes every write durable, waits for the syncs, and closes the -// directory. -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 - } - - return err -} - // start starts a sync of the writes so far when none runs and a write is not // durable. The caller holds the mutex. func (s *Syncer) start() { @@ -147,3 +102,48 @@ func (s *Syncer) run(target uint64) { s.done = make(chan struct{}) s.start() } + +// Close makes every write durable, waits for the syncs, and closes the +// directory. +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 + } + + return err +} + +// 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 + s.mutex.Unlock() + + if reached { + return nil + } + + if failure != nil { + return failure + } + + select { + case <-done: + case <-ctx.Done(): + return ctx.Err() + } + } +} diff --git a/telegram.go b/telegram.go index 4f19c62..e3e33c9 100644 --- a/telegram.go +++ b/telegram.go @@ -192,35 +192,6 @@ func (c TelegramConfig) validate() error { return nil } -// Notify queues event for the chat. It never blocks, so a slow network does not -// stop the agent. The WhatsApp batches and the runs of the schedules do not go -// to the chat: the reply of the agent tells the user what needs the user. -func (b *TelegramBot) Notify(event Event) { - switch event.Kind { - case "busy", "idle", "reply", "error", "notice", "subagent": - case "work": - if event.Source == whatsappSource || event.Source == scheduleSource { - return - } - case "user": - if event.Source == telegramSource { - return - } - default: - return - } - - b.mutex.Lock() - - b.queue = append(b.queue, event) - b.mutex.Unlock() - - select { - case b.ready <- struct{}{}: - default: - } -} - // Run polls for the messages of the owner and sends the queued events to the // chat until ctx ends. It returns an error only when a message or the offset or // image cannot be written, because then the bot would lose or repeat messages. @@ -595,6 +566,14 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { } } +// notice queues a short answer of the bot for the chat. +func (b *TelegramBot) notice(text string) { + b.Notify(Event{ + Kind: "notice", + Text: text, + }) +} + // sendOutage returns a new outage of the messages to the chat. func (b *TelegramBot) sendOutage() *TelegramOutage { return &TelegramOutage{ @@ -604,22 +583,33 @@ func (b *TelegramBot) sendOutage() *TelegramOutage { } } -// notice queues a short answer of the bot for the chat. -func (b *TelegramBot) notice(text string) { - b.Notify(Event{ - Kind: "notice", - Text: text, - }) -} - -// Fail logs the start of an outage with err and delay, unless an outage runs. -func (o *TelegramOutage) Fail(err error, delay time.Duration) { - if !o.start.IsZero() { +// Notify queues event for the chat. It never blocks, so a slow network does not +// stop the agent. The WhatsApp batches and the runs of the schedules do not go +// to the chat: the reply of the agent tells the user what needs the user. +func (b *TelegramBot) Notify(event Event) { + switch event.Kind { + case "busy", "idle", "reply", "error", "notice", "subagent": + case "work": + if event.Source == whatsappSource || event.Source == scheduleSource { + return + } + case "user": + if event.Source == telegramSource { + return + } + default: return } - o.start = time.Now() - o.logger.Warn(o.loss, "err", err, "retry", delay) + b.mutex.Lock() + + b.queue = append(b.queue, event) + b.mutex.Unlock() + + select { + case b.ready <- struct{}{}: + default: + } } // End logs the end of the outage that runs, with its length. @@ -633,6 +623,16 @@ func (o *TelegramOutage) End() { o.start = time.Time{} } +// Fail logs the start of an outage with err and delay, unless an outage runs. +func (o *TelegramOutage) Fail(err error, delay time.Duration) { + if !o.start.IsZero() { + return + } + + o.start = time.Now() + o.logger.Warn(o.loss, "err", err, "retry", delay) +} + // telegramLoadOffset reads the update offset at path. A missing file gives 0, // which asks for every update that Telegram keeps. func telegramLoadOffset(path string) (int64, error) { diff --git a/telegram_api.go b/telegram_api.go index 0d44852..ce4e0e0 100644 --- a/telegram_api.go +++ b/telegram_api.go @@ -173,6 +173,106 @@ func (b *TelegramAPI) GetUpdates(ctx context.Context, offset int64) ([]TelegramU return updates, err } +// call sends one call of method with the JSON params and decodes its result +// into result, when result is not nil. The call can take up to timeout. Its +// request setup errors name the method. Network and answer errors give a safe +// cause, so the caller can add the operation without exposing the request URL. +func (b *TelegramAPI) call(ctx context.Context, timeout time.Duration, method string, params, result any) error { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + body, err := json.Marshal(params) + if err != nil { + return fmt.Errorf("%s: encode request: %w", method, err) + } + + request, err := http.NewRequestWithContext(ctx, http.MethodPost, b.base+"/bot"+b.token+"/"+method, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("%s: make request: %s", method, b.redact(err)) + } + + request.Header.Set("Content-Type", "application/json") + + return b.answer(request, timeout, result) +} + +// redact returns the text of err without a URL and without the token. A +// *url.Error holds the URL of the call, which holds the token, so only its +// cause is kept. +func (b *TelegramAPI) redact(err error) string { + var urlError *url.Error + + if errors.As(err, &urlError) { + err = urlError.Err + } + + return strings.ReplaceAll(err.Error(), b.token, "TOKEN") +} + +// answer reads the bounded Bot API response without keeping its request URL. +func (b *TelegramAPI) answer(request *http.Request, timeout time.Duration, result any) error { + response, err := b.http.Do(request) + if err != nil { + return b.networkError("no answer", err, timeout) + } + + data, err := io.ReadAll(io.LimitReader(response.Body, telegramAnswerBytes)) + err = errors.Join(err, response.Body.Close()) + if err != nil { + return b.networkError("answer cut off", err, timeout) + } + + var answer TelegramAnswer + + err = json.Unmarshal(data, &answer) + if err != nil && response.StatusCode != http.StatusOK { + return &TelegramError{ + Status: response.StatusCode, + Description: http.StatusText(response.StatusCode), + } + } + + if err != nil { + return b.networkError("invalid JSON", err, timeout) + } + + if !answer.OK { + return &TelegramError{ + Status: response.StatusCode, + Description: b.redact(errors.New(answer.Description)), + RetryAfter: time.Duration(answer.Parameters.RetryAfter) * time.Second, + Refused: true, + } + } + + if result == nil { + return nil + } + + err = json.Unmarshal(answer.Result, result) + if err != nil { + return b.networkError("invalid result", err, timeout) + } + + return nil +} + +// networkError returns the TelegramNetworkError of err, which happened at the +// step that what names. When the call used all of its time timeout, the cause +// is "no answer in TIMEOUT", because the Go text of a deadline does not tell +// how long the wait was. +func (b *TelegramAPI) networkError(what string, err error, timeout time.Duration) error { + if errors.Is(err, context.DeadlineExceeded) { + return &TelegramNetworkError{ + Cause: "no answer in " + timeout.String(), + } + } + + return &TelegramNetworkError{ + Cause: what + ": " + b.redact(err), + } +} + // GetFile downloads bounded media through the authenticated file endpoint. // Errors never include its token or URL, and the body is limited before // storage. @@ -331,106 +431,6 @@ func (b *TelegramAPI) SendTyping(ctx context.Context, chat int64) error { }, nil) } -// call sends one call of method with the JSON params and decodes its result -// into result, when result is not nil. The call can take up to timeout. Its -// request setup errors name the method. Network and answer errors give a safe -// cause, so the caller can add the operation without exposing the request URL. -func (b *TelegramAPI) call(ctx context.Context, timeout time.Duration, method string, params, result any) error { - ctx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() - - body, err := json.Marshal(params) - if err != nil { - return fmt.Errorf("%s: encode request: %w", method, err) - } - - request, err := http.NewRequestWithContext(ctx, http.MethodPost, b.base+"/bot"+b.token+"/"+method, bytes.NewReader(body)) - if err != nil { - return fmt.Errorf("%s: make request: %s", method, b.redact(err)) - } - - request.Header.Set("Content-Type", "application/json") - - return b.answer(request, timeout, result) -} - -// answer reads the bounded Bot API response without keeping its request URL. -func (b *TelegramAPI) answer(request *http.Request, timeout time.Duration, result any) error { - response, err := b.http.Do(request) - if err != nil { - return b.networkError("no answer", err, timeout) - } - - data, err := io.ReadAll(io.LimitReader(response.Body, telegramAnswerBytes)) - err = errors.Join(err, response.Body.Close()) - if err != nil { - return b.networkError("answer cut off", err, timeout) - } - - var answer TelegramAnswer - - err = json.Unmarshal(data, &answer) - if err != nil && response.StatusCode != http.StatusOK { - return &TelegramError{ - Status: response.StatusCode, - Description: http.StatusText(response.StatusCode), - } - } - - if err != nil { - return b.networkError("invalid JSON", err, timeout) - } - - if !answer.OK { - return &TelegramError{ - Status: response.StatusCode, - Description: b.redact(errors.New(answer.Description)), - RetryAfter: time.Duration(answer.Parameters.RetryAfter) * time.Second, - Refused: true, - } - } - - if result == nil { - return nil - } - - err = json.Unmarshal(answer.Result, result) - if err != nil { - return b.networkError("invalid result", err, timeout) - } - - return nil -} - -// networkError returns the TelegramNetworkError of err, which happened at the -// step that what names. When the call used all of its time timeout, the cause -// is "no answer in TIMEOUT", because the Go text of a deadline does not tell -// how long the wait was. -func (b *TelegramAPI) networkError(what string, err error, timeout time.Duration) error { - if errors.Is(err, context.DeadlineExceeded) { - return &TelegramNetworkError{ - Cause: "no answer in " + timeout.String(), - } - } - - return &TelegramNetworkError{ - Cause: what + ": " + b.redact(err), - } -} - -// redact returns the text of err without a URL and without the token. A -// *url.Error holds the URL of the call, which holds the token, so only its -// cause is kept. -func (b *TelegramAPI) redact(err error) string { - var urlError *url.Error - - if errors.As(err, &urlError) { - err = urlError.Err - } - - return strings.ReplaceAll(err.Error(), b.token, "TOKEN") -} - func (e *TelegramNetworkError) Error() string { return e.Cause } @@ -439,6 +439,16 @@ func (e *TelegramError) Error() string { return fmt.Sprintf("%d %s", e.Status, e.Description) } +// telegramRejected tells if the Bot API refused a call with ok false and HTTP +// 400 or 404. Then the call did nothing: 400 refuses the request, and 404 tells +// that the method is not there. An error page with no Bot API answer does not +// 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) +} + // telegramRetryable tells if a later try of a failed Bot API call can work: // after a network error, HTTP 429, or a server error. It returns the wait that // the Bot API asks for, or 0. @@ -457,13 +467,3 @@ func telegramRetryable(err error) (bool, time.Duration) { return false, 0 } - -// telegramRejected tells if the Bot API refused a call with ok false and HTTP -// 400 or 404. Then the call did nothing: 400 refuses the request, and 404 tells -// that the method is not there. An error page with no Bot API answer does not -// 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) -} diff --git a/telegram_fake_helper_test.go b/telegram_fake_helper_test.go index e860d65..6ca755e 100644 --- a/telegram_fake_helper_test.go +++ b/telegram_fake_helper_test.go @@ -87,69 +87,6 @@ func newTelegramLocalFake(t *testing.T) *TelegramFake { return f } -// Add makes update available to getUpdates. -func (f *TelegramFake) Add(update TelegramUpdate) { - f.mutex.Lock() - - f.updates = append(f.updates, update) - f.mutex.Unlock() - - select { - case f.added <- struct{}{}: - default: - } -} - -// Calls returns the calls so far. -func (f *TelegramFake) Calls() []TelegramFakeCall { - f.mutex.Lock() - defer f.mutex.Unlock() - - return append([]TelegramFakeCall(nil), f.calls...) -} - -// Sent returns the texts of the sendMessage calls and the Markdown of the -// sendRichMessage calls that succeeded, in order. -func (f *TelegramFake) Sent() []string { - var texts []string - - for _, call := range f.Calls() { - switch { - case !call.ok: - case call.method == "sendMessage": - texts = append(texts, call.params["text"].(string)) - case call.method == "sendRichMessage": - texts = append(texts, call.params["rich_message"].(map[string]any)["markdown"].(string)) - } - } - - return texts -} - -func (f *TelegramFake) Image(id string, data []byte) { - f.mutex.Lock() - defer f.mutex.Unlock() - - if f.images == nil { - f.images = make(map[string][]byte) - } - - f.images[id] = data -} - -// RoundTrip handles a request in memory, so synctest can control its timers. -func (f *TelegramFake) RoundTrip(request *http.Request) (*http.Response, error) { - response := httptest.NewRecorder() - f.serve(response, request) - - ctx := request.Context() - if ctx.Err() != nil { - return nil, ctx.Err() - } - - return response.Result(), nil -} - func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { id, file := strings.CutPrefix(r.URL.Path, "/file/bot"+telegramTestToken+"/") if file { @@ -320,6 +257,69 @@ func (f *TelegramFake) from(offset int64) []TelegramUpdate { return updates } +// Add makes update available to getUpdates. +func (f *TelegramFake) Add(update TelegramUpdate) { + f.mutex.Lock() + + f.updates = append(f.updates, update) + f.mutex.Unlock() + + select { + case f.added <- struct{}{}: + default: + } +} + +// Sent returns the texts of the sendMessage calls and the Markdown of the +// sendRichMessage calls that succeeded, in order. +func (f *TelegramFake) Sent() []string { + var texts []string + + for _, call := range f.Calls() { + switch { + case !call.ok: + case call.method == "sendMessage": + texts = append(texts, call.params["text"].(string)) + case call.method == "sendRichMessage": + texts = append(texts, call.params["rich_message"].(map[string]any)["markdown"].(string)) + } + } + + return texts +} + +// Calls returns the calls so far. +func (f *TelegramFake) Calls() []TelegramFakeCall { + f.mutex.Lock() + defer f.mutex.Unlock() + + return append([]TelegramFakeCall(nil), f.calls...) +} + +func (f *TelegramFake) Image(id string, data []byte) { + f.mutex.Lock() + defer f.mutex.Unlock() + + if f.images == nil { + f.images = make(map[string][]byte) + } + + f.images[id] = data +} + +// RoundTrip handles a request in memory, so synctest can control its timers. +func (f *TelegramFake) RoundTrip(request *http.Request) (*http.Response, error) { + response := httptest.NewRecorder() + f.serve(response, request) + + ctx := request.Context() + if ctx.Err() != nil { + return nil, ctx.Err() + } + + return response.Result(), nil +} + // telegramOwnerMessage returns a text message of the owner in the private chat. func telegramOwnerMessage(text string) *TelegramMessage { return &TelegramMessage{ diff --git a/telegram_markdown.go b/telegram_markdown.go index b6ea5a2..658cac2 100644 --- a/telegram_markdown.go +++ b/telegram_markdown.go @@ -23,21 +23,6 @@ type TelegramMarkdownText struct { entities []TelegramEntity } -// add appends text. -func (m *TelegramMarkdownText) add(text string) { - m.text.WriteString(text) - - m.units += telegramUTF16Units(text) -} - -// entity keeps entity, unless it is empty, because Telegram refuses an entity -// of length 0. -func (m *TelegramMarkdownText) entity(entity TelegramEntity) { - if entity.Length > 0 { - m.entities = append(m.entities, entity) - } -} - // inline appends one line of Markdown with its inline marks as entities. A // backslash before an ASCII punctuation character makes the character plain // text. @@ -83,6 +68,34 @@ func (m *TelegramMarkdownText) inline(line string) { m.add(line[plain:]) } +// add appends text. +func (m *TelegramMarkdownText) add(text string) { + m.text.WriteString(text) + + m.units += telegramUTF16Units(text) +} + +// entity keeps entity, unless it is empty, because Telegram refuses an entity +// of length 0. +func (m *TelegramMarkdownText) entity(entity TelegramEntity) { + if entity.Length > 0 { + m.entities = append(m.entities, entity) + } +} + +// telegramUTF16Units returns the number of UTF-16 code units of text. +func telegramUTF16Units(text string) int { + units := 0 + for _, char := range text { + units++ + if char > 0xFFFF { + units++ + } + } + + return units +} + // telegramMarkdownEntities converts the Markdown of a reply to plain text and // Telegram message entities. It knows bold (** and __), italic (* and _), // strikethrough (~~), inline code, fenced code blocks with an optional @@ -150,27 +163,6 @@ func telegramEscapeTags(markdown string) string { return telegramRewriteProse(markdown, telegramTagEscaper.Replace) } -// telegramUnescapeTags gives back the Markdown that telegramEscapeTags escaped. -// It is exact, because after the escape each & outside code starts & or -// <. It also works on a part of telegramSplitRich, because a part has the -// same code as the text that it comes from. -func telegramUnescapeTags(markdown string) string { - return telegramRewriteProse(markdown, telegramTagUnescaper.Replace) -} - -// telegramUTF16Units returns the number of UTF-16 code units of text. -func telegramUTF16Units(text string) int { - units := 0 - for _, char := range text { - units++ - if char > 0xFFFF { - units++ - } - } - - return units -} - // telegramRewriteProse returns markdown with rewrite applied to each part of a // line outside inline code and fenced code blocks. A code block that does not // close runs to the end, as in Markdown. @@ -237,3 +229,11 @@ func telegramRewriteLineProse(out *strings.Builder, line string, rewrite func(st out.WriteString(rewrite(line[prose:])) } + +// telegramUnescapeTags gives back the Markdown that telegramEscapeTags escaped. +// It is exact, because after the escape each & outside code starts & or +// <. It also works on a part of telegramSplitRich, because a part has the +// same code as the text that it comes from. +func telegramUnescapeTags(markdown string) string { + return telegramRewriteProse(markdown, telegramTagUnescaper.Replace) +} diff --git a/terminal_client_helper_test.go b/terminal_client_helper_test.go index 0fdd93b..8dbc646 100644 --- a/terminal_client_helper_test.go +++ b/terminal_client_helper_test.go @@ -27,11 +27,6 @@ func (c *TerminalTestClient) Type(line string) { } } -// Output returns what the client showed. -func (c *TerminalTestClient) Output() string { - return c.out.String() -} - // WaitFor waits until the client showed text. func (c *TerminalTestClient) WaitFor(t *testing.T, text string) { t.Helper() @@ -41,6 +36,11 @@ func (c *TerminalTestClient) WaitFor(t *testing.T, text string) { }, 10*time.Second, time.Millisecond, "output so far: %q", c.Output()) } +// Output returns what the client showed. +func (c *TerminalTestClient) Output() string { + return c.out.String() +} + // Close ends the input and waits for the client. func (c *TerminalTestClient) Close() error { err := c.input.Close() diff --git a/tools_test.go b/tools_test.go index fbbea41..3075cbf 100644 --- a/tools_test.go +++ b/tools_test.go @@ -244,6 +244,13 @@ func newTestToolbox(t *testing.T, log ToolLog, dir string) *Toolbox { } } +func (r *ToolRecordingLog) Events() []string { + r.mutex.Lock() + defer r.mutex.Unlock() + + return append([]string(nil), r.events...) +} + func (r *ToolRecordingLog) Append(kind, text string) (Message, error) { r.mutex.Lock() defer r.mutex.Unlock() @@ -267,10 +274,3 @@ func (r *ToolRecordingLog) Flush(ctx context.Context) error { return ctx.Err() } - -func (r *ToolRecordingLog) Events() []string { - r.mutex.Lock() - defer r.mutex.Unlock() - - return append([]string(nil), r.events...) -} diff --git a/tree.go b/tree.go index ad30eeb..4309597 100644 --- a/tree.go +++ b/tree.go @@ -88,104 +88,6 @@ func openTree(ctx context.Context, root string, total int, clock func() time.Tim return t, nil } -// Close closes the files of the tree. -func (t *Tree) Close() error { - err := t.journal.Close() - if t.order != nil { - err = errors.Join(err, t.order.Close()) - t.order = nil - } - - for level, index := range t.levels { - err = errors.Join(err, index.Close()) - - delete(t.levels, level) - } - - return err -} - -// Add writes a new node to the tree. -func (t *Tree) Add(node Node) error { - if t.failure != nil { - return fmt.Errorf("tree needs a reopen after a failed write: %w", t.failure) - } - - err := t.check(node) - if err != nil { - return err - } - - line, err := json.Marshal(node) - if err != nil { - return fmt.Errorf("encode node: %w", err) - } - - place, err := t.journal.Write(t.clock().Format(time.DateOnly), append(line, '\n')) - if err == nil { - err = t.addIndex(NodeKey{node.Level, node.Index}, place) - } - - if err != nil { - t.failure = err - return fmt.Errorf("write node (%d, %d): %w", node.Level, node.Index, err) - } - - return nil -} - -// Built tells if the node (l, i) is built. -func (t *Tree) Built(level, index int) (bool, error) { - _, built, err := t.place(level, index) - return built, err -} - -// Node reads the node (l, i) if it is built. -func (t *Tree) Node(level, index int) (Node, bool, error) { - place, built, err := t.place(level, index) - if err != nil || !built { - return Node{}, false, err - } - - line, err := t.journal.Read(place) - if err != nil { - return Node{}, false, fmt.Errorf("read node (%d, %d): %w", level, index, err) - } - - var node Node - - err = json.Unmarshal(line, &node) - if err != nil { - return Node{}, false, fmt.Errorf("decode node (%d, %d): %w", level, index, err) - } - - if node.Level != level || node.Index != index { - return Node{}, false, fmt.Errorf("index of node (%d, %d) points to node (%d, %d)", level, index, node.Level, node.Index) - } - - return node, true, nil -} - -// Keys returns the keys of the built nodes, ordered by level and then by index. -func (t *Tree) Keys() ([]NodeKey, error) { - keys := make([]NodeKey, 0, t.order.count) - for n := range t.order.count { - record, _, err := t.order.Read(n) - if err != nil { - return nil, err - } - - key, _ := treeDecodeRecord(record) - keys = append(keys, key) - } - - slices.SortFunc(keys, func(a, b NodeKey) int { - return cmp.Or(cmp.Compare(a.level, b.level), cmp.Compare(a.index, b.index)) - }) - - return keys, nil -} - func (t *Tree) open(ctx context.Context, total int) error { var err error @@ -241,6 +143,54 @@ func (t *Tree) open(ctx context.Context, total int) error { }) } +// Close closes the files of the tree. +func (t *Tree) Close() error { + err := t.journal.Close() + if t.order != nil { + err = errors.Join(err, t.order.Close()) + t.order = nil + } + + for level, index := range t.levels { + err = errors.Join(err, index.Close()) + + delete(t.levels, level) + } + + return err +} + +// removeLevels removes the level indexes, which are not valid without the order +// index. +func (t *Tree) removeLevels() error { + paths, err := filepath.Glob(filepath.Join(t.dir, "tree-*.idx")) + if err != nil { + return fmt.Errorf("list level indexes: %w", err) + } + + for _, path := range paths { + err = os.Remove(path) + if err != nil { + return fmt.Errorf("remove level index: %w", err) + } + } + + return nil +} + +// writeLevel writes the position of node to its level index. +func (t *Tree) writeLevel(node NodeKey, place Position) error { + level, err := t.level(node.level) + if err != nil { + return err + } + + record := make([]byte, positionBytes) + encodePosition(record, place) + + return level.Write(node.index, record) +} + // check makes sure that node is valid and can join the tree now. func (t *Tree) check(node Node) error { if node.Level < 0 || node.Level >= bits.UintSize-2 || node.Index < 0 { @@ -293,6 +243,29 @@ func (t *Tree) addIndex(node NodeKey, place Position) error { return t.writeLevel(node, place) } +// level returns the index of level, and opens it at first use. +func (t *Tree) level(level int) (*Records, error) { + index, open := t.levels[level] + if open { + return index, nil + } + + index, err := openRecords(filepath.Join(t.dir, fmt.Sprintf("tree-%d.idx", level)), positionBytes) + if err != nil { + return nil, err + } + + t.levels[level] = index + + return index, nil +} + +// Built tells if the node (l, i) is built. +func (t *Tree) Built(level, index int) (bool, error) { + _, built, err := t.place(level, index) + return built, err +} + // place returns the position of the node (l, i) if it is built. func (t *Tree) place(level, index int) (Position, bool, error) { if level < 0 || level >= bits.UintSize-2 || index < 0 { @@ -314,52 +287,79 @@ func (t *Tree) place(level, index int) (Position, bool, error) { return place, place.length > 0, nil } -// removeLevels removes the level indexes, which are not valid without the order -// index. -func (t *Tree) removeLevels() error { - paths, err := filepath.Glob(filepath.Join(t.dir, "tree-*.idx")) +// Add writes a new node to the tree. +func (t *Tree) Add(node Node) error { + if t.failure != nil { + return fmt.Errorf("tree needs a reopen after a failed write: %w", t.failure) + } + + err := t.check(node) if err != nil { - return fmt.Errorf("list level indexes: %w", err) + return err } - for _, path := range paths { - err = os.Remove(path) - if err != nil { - return fmt.Errorf("remove level index: %w", err) - } + line, err := json.Marshal(node) + if err != nil { + return fmt.Errorf("encode node: %w", err) + } + + place, err := t.journal.Write(t.clock().Format(time.DateOnly), append(line, '\n')) + if err == nil { + err = t.addIndex(NodeKey{node.Level, node.Index}, place) + } + + if err != nil { + t.failure = err + return fmt.Errorf("write node (%d, %d): %w", node.Level, node.Index, err) } return nil } -// writeLevel writes the position of node to its level index. -func (t *Tree) writeLevel(node NodeKey, place Position) error { - level, err := t.level(node.level) +// Node reads the node (l, i) if it is built. +func (t *Tree) Node(level, index int) (Node, bool, error) { + place, built, err := t.place(level, index) + if err != nil || !built { + return Node{}, false, err + } + + line, err := t.journal.Read(place) if err != nil { - return err + return Node{}, false, fmt.Errorf("read node (%d, %d): %w", level, index, err) } - record := make([]byte, positionBytes) - encodePosition(record, place) + var node Node - return level.Write(node.index, record) -} + err = json.Unmarshal(line, &node) + if err != nil { + return Node{}, false, fmt.Errorf("decode node (%d, %d): %w", level, index, err) + } -// level returns the index of level, and opens it at first use. -func (t *Tree) level(level int) (*Records, error) { - index, open := t.levels[level] - if open { - return index, nil + if node.Level != level || node.Index != index { + return Node{}, false, fmt.Errorf("index of node (%d, %d) points to node (%d, %d)", level, index, node.Level, node.Index) } - index, err := openRecords(filepath.Join(t.dir, fmt.Sprintf("tree-%d.idx", level)), positionBytes) - if err != nil { - return nil, err + return node, true, nil +} + +// Keys returns the keys of the built nodes, ordered by level and then by index. +func (t *Tree) Keys() ([]NodeKey, error) { + keys := make([]NodeKey, 0, t.order.count) + for n := range t.order.count { + record, _, err := t.order.Read(n) + if err != nil { + return nil, err + } + + key, _ := treeDecodeRecord(record) + keys = append(keys, key) } - t.levels[level] = index + slices.SortFunc(keys, func(a, b NodeKey) int { + return cmp.Or(cmp.Compare(a.level, b.level), cmp.Compare(a.index, b.index)) + }) - return index, nil + return keys, nil } // Count returns the number of built nodes. @@ -367,9 +367,20 @@ func (t *Tree) Count() int { return t.order.count } -// leafSource returns the text that the leaf of message holds or summarizes. -func leafSource(message Message) string { - return message.Kind + ": " + message.Text +// treeDecodeRecord reads the key and the position of a record of +// index/tree.idx. +func treeDecodeRecord(record []byte) (NodeKey, Position) { + key := NodeKey{ + level: int(binary.LittleEndian.Uint32(record[positionBytes+8:])), + index: int(binary.LittleEndian.Uint64(record[positionBytes:])), + } + + return key, decodePosition(record) +} + +// nodeEnd returns the ID after the last message of node (l, i). +func nodeEnd(level, index int) int { + return (index + 1) << level } // freeLeaf returns a text-only leaf when its source fits in a node. Images need @@ -388,6 +399,11 @@ func freeLeaf(message Message) (Node, bool) { }, true } +// leafSource returns the text that the leaf of message holds or summarizes. +func leafSource(message Message) string { + return message.Kind + ": " + message.Text +} + // freeMerge returns the parent of two sibling nodes when both texts, joined by // a newline, fit in a node. Such a parent needs no model call. func freeMerge(left, right Node) (Node, bool) { @@ -404,19 +420,3 @@ func freeMerge(left, right Node) (Node, bool) { Size: len(text), }, true } - -// nodeEnd returns the ID after the last message of node (l, i). -func nodeEnd(level, index int) int { - return (index + 1) << level -} - -// treeDecodeRecord reads the key and the position of a record of -// index/tree.idx. -func treeDecodeRecord(record []byte) (NodeKey, Position) { - key := NodeKey{ - level: int(binary.LittleEndian.Uint32(record[positionBytes+8:])), - index: int(binary.LittleEndian.Uint64(record[positionBytes:])), - } - - return key, decodePosition(record) -} diff --git a/turn.go b/turn.go index b5aa4cf..c08e94a 100644 --- a/turn.go +++ b/turn.go @@ -118,26 +118,15 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W return l, nil } -// Submit logs a user message from the client source and wakes the loop. The -// message joins the next request: the first one of a new turn, or the next one -// of a running turn. -func (l *TurnLoop) Submit(source, text string) (Message, error) { - message, err := l.record("user", source, text) - if err != nil { - return Message{}, err - } - - l.mutex.Lock() - - l.pending = append(l.pending, message) - l.mutex.Unlock() - - select { - case l.wake <- struct{}{}: - default: +// state returns the per-turn state: the end of the view, and the date and time +// of the turn, then the names and responsibilities of its children. +func (l *TurnLoop) state() string { + state := "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") + if l.messages != nil { + state += l.messages.ChildState(l.agent.config.Name) } - return message, nil + return state } // Report logs the report text of the subagent name as a work message and tells @@ -155,6 +144,26 @@ func (l *TurnLoop) Report(name, text, status string) { } } +// note logs a note and tells the clients. A note starts no turn. +func (l *TurnLoop) note(text string) error { + _, err := l.record("note", "", text) + + return err +} + +// 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() + + select { + case l.wake <- struct{}{}: + default: + } +} + // Work logs text as a work message of source, such as the name of a subagent, // and tells the clients, with status. The message starts no turn until it goes // to Wake. @@ -194,6 +203,46 @@ func (l *TurnLoop) Wake(messages ...Message) { } } +// record logs a message and tells the clients, with the client source of a user +// message. +func (l *TurnLoop) record(kind, source, text string) (Message, error) { + message, err := l.memory.Append(kind, text) + if err != nil { + return Message{}, err + } + + l.notify(Event{ + Kind: kind, + ID: message.ID, + Text: text, + Source: source, + }) + + return message, nil +} + +// Submit logs a user message from the client source and wakes the loop. The +// message joins the next request: the first one of a new turn, or the next one +// of a running turn. +func (l *TurnLoop) Submit(source, text string) (Message, error) { + message, err := l.record("user", source, text) + if err != nil { + return Message{}, err + } + + l.mutex.Lock() + + l.pending = append(l.pending, message) + l.mutex.Unlock() + + select { + case l.wake <- struct{}{}: + default: + } + + return message, nil +} + // StopSubagent ends the run of the subagent name and tells if it ran. Its // report is logged, but it starts no turn. func (l *TurnLoop) StopSubagent(name string) (bool, error) { @@ -272,31 +321,6 @@ func (l *TurnLoop) Run(ctx context.Context) error { } } -// note logs a note and tells the clients. A note starts no turn. -func (l *TurnLoop) note(text string) error { - _, err := l.record("note", "", text) - - return err -} - -// record logs a message and tells the clients, with the client source of a user -// message. -func (l *TurnLoop) record(kind, source, text string) (Message, error) { - message, err := l.memory.Append(kind, text) - if err != nil { - return Message{}, err - } - - l.notify(Event{ - Kind: kind, - ID: message.ID, - Text: text, - Source: source, - }) - - return message, nil -} - // waiting tells if user messages or reports wait for a request. func (l *TurnLoop) waiting() bool { l.mutex.Lock() @@ -305,19 +329,6 @@ func (l *TurnLoop) waiting() bool { return len(l.pending) > 0 } -// 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() - - select { - case l.wake <- struct{}{}: - default: - } -} - // failed returns the error of a failed write of a report. func (l *TurnLoop) failed() error { l.mutex.Lock() @@ -462,17 +473,6 @@ func (l *TurnLoop) take() []Message { return messages } -// state returns the per-turn state: the end of the view, and the date and time -// of the turn, then the names and responsibilities of its children. -func (l *TurnLoop) state() string { - state := "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") - if l.messages != nil { - state += l.messages.ChildState(l.agent.config.Name) - } - - return state -} - // turnError returns the text of the error that ended a turn. func turnError(ctx context.Context, err error) string { if ctx.Err() != nil { diff --git a/turn_test.go b/turn_test.go index 5341de5..27151a9 100644 --- a/turn_test.go +++ b/turn_test.go @@ -370,6 +370,13 @@ func (r *TurnEventRecorder) Last(kind string) string { return r.LastEvent(kind).Text } +func (r *TurnEventRecorder) Add(event Event) { + r.mutex.Lock() + defer r.mutex.Unlock() + + r.events = append(r.events, event) +} + // LastEvent returns the last event of kind. func (r *TurnEventRecorder) LastEvent(kind string) Event { r.mutex.Lock() @@ -384,11 +391,13 @@ func (r *TurnEventRecorder) LastEvent(kind string) Event { return Event{} } -func (r *TurnEventRecorder) Add(event Event) { - r.mutex.Lock() - defer r.mutex.Unlock() - - r.events = append(r.events, event) +func modelReasoningItem(id, encrypted string) map[string]any { + return map[string]any{ + "type": "reasoning", + "id": id, + "encrypted_content": encrypted, + "summary": []any{}, + } } // turnInputTexts returns the text of each user message of the input of call. @@ -424,12 +433,3 @@ func turnInputTexts(t *testing.T, call ModelFakeCall) []string { return texts } - -func modelReasoningItem(id, encrypted string) map[string]any { - return map[string]any{ - "type": "reasoning", - "id": id, - "encrypted_content": encrypted, - "summary": []any{}, - } -} diff --git a/view.go b/view.go index 1b3ed6a..4a6e567 100644 --- a/view.go +++ b/view.go @@ -97,6 +97,105 @@ func (v *View) Append(tree ViewNodeReader) (bool, error) { return v.fit(tree) } +// load keeps the node of line when it is built, and tells if it is. +func (v *View) load(tree ViewNodeReader, line NodeKey) (bool, error) { + node, built, err := tree.Node(line.level, line.index) + if err != nil || !built { + return false, err + } + + v.nodes[line] = node + + return true, nil +} + +// fit merges the most due pairs while a batch runs, and tells if a merge +// changed the view. +func (v *View) fit(tree ViewNodeReader) (bool, error) { + size := v.Size() + if size > v.high { + v.merging = true + } + + merged := false + for v.merging && size > v.low { + best, err := v.mostDue(tree) + if err != nil || best < 0 { + return merged, err + } + + left, right := v.lines[best], v.lines[best+1] + parent := NodeKey{left.level + 1, left.index / 2} + _, err = v.load(tree, parent) + if err != nil { + return merged, err + } + + 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 + } + + v.merging = false + + return merged, nil +} + +// Size returns the sum of the line sizes. +func (v *View) Size() int { + size := 0 + for _, line := range v.lines { + size += v.lineSize(line) + } + + return size +} + +// mostDue returns the position of the left line of the most due sibling pair +// whose parent is built, or -1 when there is no such pair. +func (v *View) mostDue(tree ViewNodeReader) (int, error) { + best := -1 + bestDue := 0.0 + for n := 0; n+1 < len(v.lines); n++ { + left, right := v.lines[n], v.lines[n+1] + if left.index%2 != 0 || right.level != left.level || right.index != left.index+1 { + continue + } + + last := nodeEnd(right.level, right.index) - 1 + due := math.Ldexp(float64(v.end-last), -left.level) + if best >= 0 && due <= bestDue { + continue + } + + built, err := tree.Built(left.level+1, left.index/2) + if err != nil { + return -1, fmt.Errorf("check parent node: %w", err) + } + + if built { + best = n + bestDue = due + } + } + + return best, nil +} + +// lineSize returns the size of a view line. +func (v *View) lineSize(line NodeKey) int { + node, built := v.nodes[line] + if !built { + return len(viewPending) + } + + return node.Size +} + // Coarsen takes each merge of other, a view of the same messages, and then // merges in one batch until the size is at most low. Each line of the result is // the coarser of the two lines that start at its first message. Dyadic ranges @@ -191,76 +290,6 @@ func (v *View) LinesWithout(skip map[int]bool) []string { return lines } -// Context returns the lines of the view before message end as the model reads -// them. It stops at the first line that is not built, so a compaction never -// 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 { - break - } - - v.renderLine(&text, line) - } - - text.WriteString("") - - return text.String() -} - -// load keeps the node of line when it is built, and tells if it is. -func (v *View) load(tree ViewNodeReader, line NodeKey) (bool, error) { - node, built, err := tree.Node(line.level, line.index) - if err != nil || !built { - return false, err - } - - v.nodes[line] = node - - return true, nil -} - -// fit merges the most due pairs while a batch runs, and tells if a merge -// changed the view. -func (v *View) fit(tree ViewNodeReader) (bool, error) { - size := v.Size() - if size > v.high { - v.merging = true - } - - merged := false - for v.merging && size > v.low { - best, err := v.mostDue(tree) - if err != nil || best < 0 { - return merged, err - } - - left, right := v.lines[best], v.lines[best+1] - parent := NodeKey{left.level + 1, left.index / 2} - _, err = v.load(tree, parent) - if err != nil { - return merged, err - } - - 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 - } - - v.merging = false - - return merged, nil -} - // renderLine writes one view line, "id+n|text", and a newline. func (v *View) renderLine(text *strings.Builder, line NodeKey) { body := viewPending @@ -277,55 +306,26 @@ func (v *View) renderLine(text *strings.Builder, line NodeKey) { text.WriteByte('\n') } -// Size returns the sum of the line sizes. -func (v *View) Size() int { - size := 0 - for _, line := range v.lines { - size += v.lineSize(line) - } - - return size -} - -// mostDue returns the position of the left line of the most due sibling pair -// whose parent is built, or -1 when there is no such pair. -func (v *View) mostDue(tree ViewNodeReader) (int, error) { - best := -1 - bestDue := 0.0 - for n := 0; n+1 < len(v.lines); n++ { - left, right := v.lines[n], v.lines[n+1] - if left.index%2 != 0 || right.level != left.level || right.index != left.index+1 { - continue - } +// Context returns the lines of the view before message end as the model reads +// them. It stops at the first line that is not built, so a compaction never +// sees a pending mark or a part of a message. +func (v *View) Context(end int) string { + var text strings.Builder - last := nodeEnd(right.level, right.index) - 1 - due := math.Ldexp(float64(v.end-last), -left.level) - if best >= 0 && due <= bestDue { - continue - } + text.WriteString("\n") - built, err := tree.Built(left.level+1, left.index/2) - if err != nil { - return -1, fmt.Errorf("check parent node: %w", err) + for _, line := range v.lines { + _, built := v.nodes[line] + if !built || nodeEnd(line.level, line.index) > end { + break } - if built { - best = n - bestDue = due - } + v.renderLine(&text, line) } - return best, nil -} - -// lineSize returns the size of a view line. -func (v *View) lineSize(line NodeKey) int { - node, built := v.nodes[line] - if !built { - return len(viewPending) - } + text.WriteString("") - return node.Size + return text.String() } // restoreView returns the view of state with the limits high and low. Each line diff --git a/whatsapp.go b/whatsapp.go index 738ef8d..1e6bfa8 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -230,6 +230,34 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge return w, nil } +func (w *WhatsApp) sessionPath() string { + return filepath.Join(w.dir, whatsappSessionFileName) +} + +// openSession opens the session database and its first device. The caller holds +// the mutex, or no other goroutine uses w yet. +func (w *WhatsApp) openSession(ctx context.Context) error { + address, err := whatsappSessionAddress(w.sessionPath()) + if err != nil { + return err + } + + container, err := sqlstore.New(ctx, "sqlite", address, waLog.Noop) + if err != nil { + return fmt.Errorf("open WhatsApp session: %w", err) + } + + device, err := container.GetFirstDevice(ctx) + if err != nil { + return errors.Join(fmt.Errorf("read WhatsApp device: %w", err), container.Close()) + } + + w.container = container + w.device = device + + return nil +} + // Run connects a linked device, logs the closed batches in l, runs the pair and // unlink commands, and handles a logout from the phone, until ctx ends. Its // notes go to the log of l, and notify gets the notices for all clients. A @@ -343,177 +371,166 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err } } -// Command runs the WhatsApp command args, the text after /whatsapp, and gives -// each answer to answer. The answers go only to the client that sent the -// command. -func (w *WhatsApp) Command(args string, answer func(text string)) { - fields := strings.Fields(args) - request := WhatsAppRequest{ - answer: answer, +// connect tries the starting connection of the linked device with one client. +// It returns whether Run must retry. A failed call starts the shared outage +// state and logs its redacted cause once, because a returned ConnectContext +// error does not emit Disconnected. Connected logs recovery after +// 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 } - switch { - case len(fields) == 1 && fields[0] == "status": - answer(w.Status()) + if w.client == nil { + w.client = w.newClient(w.device, w.handle) + } - return - case len(fields) == 1 && fields[0] == "unlink": - request.unlink = true - case len(fields) >= 2 && fields[0] == "pair": - 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.") + client := w.client + w.mutex.Unlock() - return - } + err := client.ConnectContext(ctx) - request.phone = phone - default: - answer(whatsappUsage) + w.mutex.Lock() - return + if err == nil || ctx.Err() != nil || !w.retrying || !w.linked { + w.retrying = false + w.mutex.Unlock() + + return false } - select { - case w.requests <- request: - default: - answer("Too many WhatsApp commands wait. Try again in a minute.") + first := w.startOutage() + cause := whatsappRedact(w.own, err.Error()) + w.mutex.Unlock() + + if first { + w.logger.Warn("first connect failed, retrying", "err", cause) } + + return true } -// Status returns the state of the link: not linked, linked and connected, or -// linked and disconnected, with the times of the connection and the routes. -// Route source IDs can hold the phone numbers of other chats. -func (w *WhatsApp) Status() string { +// close ends the connection and the timer of the notice, and stops all notices. +func (w *WhatsApp) close() { w.mutex.Lock() - defer w.mutex.Unlock() - routes := "" - if w.routes != nil { - listed := w.routes.List() - if listed != "No routes." { - routes = "\nWhatsApp routes:\n" + listed - } + w.closed = true + if w.timer != nil { + w.timer.Stop() } - switch { - case !w.linked: - return "WhatsApp is not linked. Run /whatsapp pair PHONE to link it." + routes - case w.connected: - return "WhatsApp is linked and connected since " + w.connectedAt.Format(whatsappTimeLayout) + "." + routes - case w.disconnectedAt.IsZero(): - return "WhatsApp is linked and not connected yet." + routes - case w.connectedAt.IsZero(): - return "WhatsApp is linked and disconnected since " + w.disconnectedAt.Format(whatsappTimeLayout) + "." + routes - default: - return "WhatsApp is linked and disconnected since " + w.disconnectedAt.Format(whatsappTimeLayout) + ". The last connection started at " + w.connectedAt.Format(whatsappTimeLayout) + "." + routes + client := w.client + w.client = nil + w.mutex.Unlock() + + w.inbox.close() + + if client != nil { + client.Disconnect() } } -// BackupSession writes a consistent SQLite snapshot to path while the session -// cannot be created, closed, or removed. The source opens without creation -// permission. No session file means no linked state to copy. -func (w *WhatsApp) BackupSession(ctx context.Context, path string) (err error) { - w.sessionMutex.Lock() - defer w.sessionMutex.Unlock() - +// link returns a copy of the link of the device: whether it is linked, the +// device, the client, whether it is connected, and the own IDs. +func (w *WhatsApp) link() WhatsAppLink { w.mutex.Lock() defer w.mutex.Unlock() - _, err = os.Stat(w.sessionPath()) - if errors.Is(err, os.ErrNotExist) { - return nil + return WhatsAppLink{ + linked: w.linked, + device: w.device, + client: w.client, + connected: w.connected, + own: slices.Clone(w.own), } +} - if err != nil { - return fmt.Errorf("read session file: %w", err) +// loggedOut logs the open batches, deletes the session after a logout from the +// phone, notes it in the log of l, and tells all clients one time. +func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error { + if !w.isLinked() { + return nil } - err = os.MkdirAll(filepath.Dir(path), 0o700) - if err != nil { - return fmt.Errorf("create session snapshot directory: %w", err) - } + w.logger.Error("logged out by the phone, run /whatsapp pair PHONE") - address, err := whatsappSessionAddress(w.sessionPath()) + err := w.inbox.deliver(ctx, l, w.link(), true) if err != nil { - return fmt.Errorf("find session database: %w", err) + return err } - database, err := sql.Open("sqlite", address+"&mode=rw") + err = w.removeSession() if err != nil { - return fmt.Errorf("open session snapshot source: %w", err) + w.logger.Error("session delete failed", "err", err) } - defer func() { - err = errors.Join(err, database.Close()) - }() - - _, err = database.ExecContext(ctx, "VACUUM INTO ?", path) + err = l.note("WhatsApp was unlinked from the phone.") if err != nil { - return fmt.Errorf("snapshot session database: %w", err) + w.logger.Error("chat note failed", "err", err) } - err = os.Chmod(path, 0o600) - if err != nil { - return fmt.Errorf("set session snapshot mode: %w", err) - } + w.broadcast("WhatsApp was unlinked from the phone. Run /whatsapp pair PHONE to link it again.") return nil } -// Close closes the session database. -func (w *WhatsApp) Close() error { - w.sessionMutex.Lock() - defer w.sessionMutex.Unlock() - +// unlink logs the open batches, logs out the linked device, and deletes the +// session. When the logout call fails, as it does while WhatsApp is +// disconnected, the session is deleted all the same, and answer tells the user +// to remove the device in the phone. Its error is one of the log or the state +// file. +func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text string)) error { w.mutex.Lock() - defer w.mutex.Unlock() - if w.container == nil { + linked := w.linked + client := w.client + w.mutex.Unlock() + + if !linked { + answer("WhatsApp is not linked.") + return nil } - err := w.container.Close() - w.container = nil - - return err -} + err := w.inbox.deliver(ctx, l, w.link(), true) + if err != nil { + return err + } -// Find returns the answer of whatsapp_find for name, as whatsappFind says. -func (w *WhatsApp) Find(ctx context.Context, name string) (string, error) { - return whatsappFind(ctx, w.link(), w.groups, name) -} + var logoutErr error -// Send returns the answer of whatsapp_send for text and an optional stored -// image to chat, with the durable send id that whatsappSend describes. -func (w *WhatsApp) Send(ctx context.Context, id, chat, text, image string) (string, error) { - return whatsappSend(ctx, w.link(), w.groups, w.state, w.inbox.images, w.logger, w.clock, id, chat, text, image) -} + if client != nil { + ctx, cancel := context.WithTimeout(ctx, whatsappLogoutTimeout) + defer cancel() -func (w *WhatsApp) sessionPath() string { - return filepath.Join(w.dir, whatsappSessionFileName) -} + logoutErr = client.Logout(ctx) + } -// openSession opens the session database and its first device. The caller holds -// the mutex, or no other goroutine uses w yet. -func (w *WhatsApp) openSession(ctx context.Context) error { - address, err := whatsappSessionAddress(w.sessionPath()) + err = w.removeSession() if err != nil { - return err + answer("The unlink failed: " + w.redact(err.Error())) + + return nil } - container, err := sqlstore.New(ctx, "sqlite", address, waLog.Noop) + w.logger.Info("unlinked") + + err = l.note("WhatsApp is unlinked.") if err != nil { - return fmt.Errorf("open WhatsApp session: %w", err) + w.logger.Error("chat note failed", "err", err) } - device, err := container.GetFirstDevice(ctx) - if err != nil { - return errors.Join(fmt.Errorf("read WhatsApp device: %w", err), container.Close()) + 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 } - w.container = container - w.device = device + answer("WhatsApp is unlinked.") return nil } @@ -525,236 +542,6 @@ func (w *WhatsApp) isLinked() bool { return w.linked } -// link returns a copy of the link of the device: whether it is linked, the -// device, the client, whether it is connected, and the own IDs. -func (w *WhatsApp) link() WhatsAppLink { - w.mutex.Lock() - defer w.mutex.Unlock() - - return WhatsAppLink{ - linked: w.linked, - device: w.device, - client: w.client, - connected: w.connected, - own: slices.Clone(w.own), - } -} - -// connect tries the starting connection of the linked device with one client. -// It returns whether Run must retry. A failed call starts the shared outage -// state and logs its redacted cause once, because a returned ConnectContext -// error does not emit Disconnected. Connected logs recovery after -// 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 - } - - if w.client == nil { - w.client = w.newClient(w.device, w.handle) - } - - client := w.client - 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() - - return false - } - - first := w.startOutage() - cause := whatsappRedact(w.own, err.Error()) - w.mutex.Unlock() - - if first { - w.logger.Warn("first connect failed, retrying", "err", cause) - } - - return true -} - -// startOutage starts the shared outage time and notice timer once. The caller -// holds the mutex. Its result tells whether this is the first failure. -func (w *WhatsApp) startOutage() bool { - first := w.linked && (w.connected || w.disconnectedAt.IsZero()) - if first { - w.disconnectedAt = w.clock() - w.disconnections++ - disconnections := w.disconnections - w.timer = time.AfterFunc(whatsappOutageNoticeDelay, func() { - w.disconnectedLong(disconnections) - }) - } - - w.connected = false - - return first -} - -// 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() - } - - client := w.client - w.client = nil - w.mutex.Unlock() - - w.inbox.close() - - if client != nil { - client.Disconnect() - } -} - -// loggedOut logs the open batches, deletes the session after a logout from the -// phone, notes it in the log of l, and tells all clients one time. -func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error { - if !w.isLinked() { - return nil - } - - w.logger.Error("logged out by the phone, run /whatsapp pair PHONE") - - err := w.inbox.deliver(ctx, l, w.link(), true) - if err != nil { - return err - } - - err = w.removeSession() - if err != nil { - w.logger.Error("session delete failed", "err", err) - } - - err = l.note("WhatsApp was unlinked from the phone.") - if err != nil { - w.logger.Error("chat note failed", "err", err) - } - - w.broadcast("WhatsApp was unlinked from the phone. Run /whatsapp pair PHONE to link it again.") - - return nil -} - -// unlink logs the open batches, logs out the linked device, and deletes the -// session. When the logout call fails, as it does while WhatsApp is -// disconnected, the session is deleted all the same, and answer tells the user -// to remove the device in the phone. Its error is one of the log or the state -// 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 - } - - err := w.inbox.deliver(ctx, l, w.link(), true) - if err != nil { - return err - } - - var logoutErr error - - if client != nil { - ctx, cancel := context.WithTimeout(ctx, whatsappLogoutTimeout) - defer cancel() - - logoutErr = client.Logout(ctx) - } - - err = w.removeSession() - if err != nil { - answer("The unlink failed: " + w.redact(err.Error())) - - return nil - } - - w.logger.Info("unlinked") - - err = l.note("WhatsApp is unlinked.") - if err != nil { - w.logger.Error("chat note failed", "err", err) - } - - if client == nil || logoutErr != nil { - answer("WhatsApp is unlinked here, but the phone did not get the logout. In WhatsApp on your phone, open Linked devices and remove the device.") - - return nil - } - - answer("WhatsApp is unlinked.") - - return nil -} - -// removeSession ends the connection, closes the session database, and deletes -// its files. The session lock covers the cleanup after mutex is released, so a -// snapshot or a new session cannot open files that cleanup removes. -func (w *WhatsApp) removeSession() error { - w.sessionMutex.Lock() - defer w.sessionMutex.Unlock() - - w.mutex.Lock() - - client := w.client - w.client = nil - w.linked = false - w.retrying = false - w.own = nil - w.connected = false - w.disconnectedAt = time.Time{} - if w.timer != nil { - w.timer.Stop() - } - - w.noticed = false - container := w.container - w.container = nil - w.device = nil - w.mutex.Unlock() - - w.inbox.stopCatchup() - - if client != nil { - client.Disconnect() - } - - var err error - - if container != nil { - err = container.Close() - } - - for _, suffix := range []string{"", "-wal", "-shm"} { - removeErr := os.Remove(w.sessionPath() + suffix) - if !errors.Is(removeErr, os.ErrNotExist) { - err = errors.Join(err, removeErr) - } - } - - return err -} - // pair links a new device to the account of phone with a pairing code. It gives // the steps and the code to answer, and then the end of the pairing: the linked // number, an expired code, or a failure. A pairing that a new one ends gives no @@ -863,20 +650,226 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string, } } -// newDevice returns a new device in the session database. It makes the -// directory and the file of the session first, with their modes, when they do -// not exist. -func (w *WhatsApp) newDevice(ctx context.Context) (*store.Device, error) { - w.sessionMutex.Lock() - defer w.sessionMutex.Unlock() - +// handle takes one event of the library and tells if it was handled. A message +// goes to the inbox. The names of the groups follow the events of the groups. A +// connection starts a catch-up of the inbox, which ends at OfflineSyncCompleted +// or after whatsappCatchupTimeout. A disconnection keeps the time of its start +// while Run or the library tries again. After whatsappOutageNoticeDelay of a +// disconnection, all clients get one notice, and they get one more when the +// connection comes back. A disconnection that the library does not try again +// gives a notice at once. +// +// The log of the daemon gets one line for each change of the connection of the +// linked device: connected, a failed first connection, a lost connection, and +// connected again with the length of the outage. The retries of the library and +// Run between them log nothing. +func (w *WhatsApp) handle(event any) bool { w.mutex.Lock() - defer w.mutex.Unlock() - if w.container == nil { - err := os.MkdirAll(w.dir, 0o700) - if err != nil { - return nil, fmt.Errorf("create WhatsApp directory: %w", err) + ctx := w.ctx + w.mutex.Unlock() + + switch event := event.(type) { + case *events.Message: + return w.inbox.receive(ctx, w.link(), event) + case *events.GroupInfo: + if event.Name != nil { + w.groups.set(event.JID.String(), event.Name.Name) + } + + 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 + + level := slog.LevelInfo + + var attrs []any + + w.mutex.Lock() + + switch event := event.(type) { + case *events.PairSuccess: + w.own = whatsappOwnIDs(event.ID.User, event.LID.User) + case *events.Connected: + switch { + case !w.linked || w.connected: + case w.disconnectedAt.IsZero(): + line = "connected" + default: + line = "connected again" + attrs = append(attrs, "after", w.clock().Sub(w.disconnectedAt)) + } + + w.connected = true + w.retrying = false + w.connectedAt = w.clock() + if w.timer != nil { + w.timer.Stop() + } + + if w.noticed { + w.noticed = false + notice = "WhatsApp is connected again." + } + case *events.Disconnected: + connected := w.connected + if w.startOutage() { + level = slog.LevelWarn + line = "disconnected, retrying" + if !connected { + line = "first connect failed, retrying" + } + } + case *events.LoggedOut: + w.retrying = false + select { + case w.logouts <- struct{}{}: + default: + } + case events.PermanentDisconnect: + w.retrying = false + if w.linked { + w.connected = false + w.noticed = true + notice = "WhatsApp is disconnected and does not connect again by itself: " + event.PermanentDisconnectDescription() + ". Restart the daemon to try again." + level = slog.LevelError + line = "disconnected for good, restart to retry" + attrs = append(attrs, "err", whatsappRedact(w.own, event.PermanentDisconnectDescription())) + } + } + + if w.closed { + line = "" + } + + w.mutex.Unlock() + + if line != "" { + w.logger.Log(ctx, level, line, attrs...) + } + + if notice != "" { + w.broadcast(notice) + } + + return true +} + +// startOutage starts the shared outage time and notice timer once. The caller +// holds the mutex. Its result tells whether this is the first failure. +func (w *WhatsApp) startOutage() bool { + first := w.linked && (w.connected || w.disconnectedAt.IsZero()) + if first { + w.disconnectedAt = w.clock() + w.disconnections++ + disconnections := w.disconnections + w.timer = time.AfterFunc(whatsappOutageNoticeDelay, func() { + w.disconnectedLong(disconnections) + }) + } + + w.connected = false + + return first +} + +// removeSession ends the connection, closes the session database, and deletes +// its files. The session lock covers the cleanup after mutex is released, so a +// snapshot or a new session cannot open files that cleanup removes. +func (w *WhatsApp) removeSession() error { + w.sessionMutex.Lock() + defer w.sessionMutex.Unlock() + + w.mutex.Lock() + + client := w.client + w.client = nil + w.linked = false + w.retrying = false + w.own = nil + w.connected = false + w.disconnectedAt = time.Time{} + if w.timer != nil { + w.timer.Stop() + } + + w.noticed = false + container := w.container + w.container = nil + w.device = nil + w.mutex.Unlock() + + w.inbox.stopCatchup() + + if client != nil { + client.Disconnect() + } + + var err error + + if container != nil { + err = container.Close() + } + + for _, suffix := range []string{"", "-wal", "-shm"} { + removeErr := os.Remove(w.sessionPath() + suffix) + if !errors.Is(removeErr, os.ErrNotExist) { + err = errors.Join(err, removeErr) + } + } + + return err +} + +// 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() + + if notify != nil && !closed { + notify(Event{ + Kind: "notice", + Text: text, + }) + } +} + +// redact replaces the own number and the own LID in text with OWN. +func (w *WhatsApp) redact(text string) string { + w.mutex.Lock() + defer w.mutex.Unlock() + + return whatsappRedact(w.own, text) +} + +// newDevice returns a new device in the session database. It makes the +// directory and the file of the session first, with their modes, when they do +// not exist. +func (w *WhatsApp) newDevice(ctx context.Context) (*store.Device, error) { + w.sessionMutex.Lock() + defer w.sessionMutex.Unlock() + + w.mutex.Lock() + defer w.mutex.Unlock() + + if w.container == nil { + err := os.MkdirAll(w.dir, 0o700) + if err != nil { + return nil, fmt.Errorf("create WhatsApp directory: %w", err) } file, err := os.OpenFile(w.sessionPath(), os.O_CREATE|os.O_RDWR, 0o600) @@ -909,162 +902,229 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) { w.linked = true } -// handle takes one event of the library and tells if it was handled. A message -// goes to the inbox. The names of the groups follow the events of the groups. A -// connection starts a catch-up of the inbox, which ends at OfflineSyncCompleted -// or after whatsappCatchupTimeout. A disconnection keeps the time of its start -// while Run or the library tries again. After whatsappOutageNoticeDelay of a -// disconnection, all clients get one notice, and they get one more when the -// connection comes back. A disconnection that the library does not try again -// gives a notice at once. -// -// The log of the daemon gets one line for each change of the connection of the -// linked device: connected, a failed first connection, a lost connection, and -// connected again with the length of the outage. The retries of the library and -// Run between them log nothing. -func (w *WhatsApp) handle(event any) bool { +// disconnectedLong gives the notice of a long disconnection, unless the +// disconnection with the number disconnections ended. +func (w *WhatsApp) disconnectedLong(disconnections int) { w.mutex.Lock() - ctx := w.ctx - w.mutex.Unlock() + long := !w.closed && w.linked && !w.connected && !w.noticed && w.disconnections == disconnections + if long { + w.noticed = true + } + + since := w.disconnectedAt + w.mutex.Unlock() + + if long { + w.broadcast("WhatsApp is disconnected since " + since.Format(whatsappTimeLayout) + ". OneChat tries again.") + } +} + +// Command runs the WhatsApp command args, the text after /whatsapp, and gives +// each answer to answer. The answers go only to the client that sent the +// command. +func (w *WhatsApp) Command(args string, answer func(text string)) { + fields := strings.Fields(args) + request := WhatsAppRequest{ + answer: answer, + } + + switch { + case len(fields) == 1 && fields[0] == "status": + answer(w.Status()) + + return + case len(fields) == 1 && fields[0] == "unlink": + request.unlink = true + case len(fields) >= 2 && fields[0] == "pair": + 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 + } + + select { + case w.requests <- request: + default: + answer("Too many WhatsApp commands wait. Try again in a minute.") + } +} + +// Status returns the state of the link: not linked, linked and connected, or +// linked and disconnected, with the times of the connection and the routes. +// Route source IDs can hold the phone numbers of other chats. +func (w *WhatsApp) Status() string { + w.mutex.Lock() + defer w.mutex.Unlock() + + routes := "" + if w.routes != nil { + listed := w.routes.List() + if listed != "No routes." { + routes = "\nWhatsApp routes:\n" + listed + } + } + + switch { + case !w.linked: + return "WhatsApp is not linked. Run /whatsapp pair PHONE to link it." + routes + case w.connected: + return "WhatsApp is linked and connected since " + w.connectedAt.Format(whatsappTimeLayout) + "." + routes + case w.disconnectedAt.IsZero(): + return "WhatsApp is linked and not connected yet." + routes + case w.connectedAt.IsZero(): + return "WhatsApp is linked and disconnected since " + w.disconnectedAt.Format(whatsappTimeLayout) + "." + routes + default: + return "WhatsApp is linked and disconnected since " + w.disconnectedAt.Format(whatsappTimeLayout) + ". The last connection started at " + w.connectedAt.Format(whatsappTimeLayout) + "." + routes + } +} + +// BackupSession writes a consistent SQLite snapshot to path while the session +// cannot be created, closed, or removed. The source opens without creation +// permission. No session file means no linked state to copy. +func (w *WhatsApp) BackupSession(ctx context.Context, path string) (err error) { + w.sessionMutex.Lock() + defer w.sessionMutex.Unlock() + + w.mutex.Lock() + defer w.mutex.Unlock() + + _, err = os.Stat(w.sessionPath()) + if errors.Is(err, os.ErrNotExist) { + return nil + } + + if err != nil { + return fmt.Errorf("read session file: %w", err) + } + + err = os.MkdirAll(filepath.Dir(path), 0o700) + if err != nil { + return fmt.Errorf("create session snapshot directory: %w", err) + } - switch event := event.(type) { - case *events.Message: - return w.inbox.receive(ctx, w.link(), event) - case *events.GroupInfo: - if event.Name != nil { - w.groups.set(event.JID.String(), event.Name.Name) - } + address, err := whatsappSessionAddress(w.sessionPath()) + if err != nil { + return fmt.Errorf("find session database: %w", err) + } - return true - case *events.JoinedGroup: - w.groups.set(event.JID.String(), event.Name) + database, err := sql.Open("sqlite", address+"&mode=rw") + if err != nil { + return fmt.Errorf("open session snapshot source: %w", err) + } - return true - case *events.OfflineSyncCompleted: - w.inbox.endCatchup() + defer func() { + err = errors.Join(err, database.Close()) + }() - return true - case *events.Connected: - w.inbox.startCatchup() + _, err = database.ExecContext(ctx, "VACUUM INTO ?", path) + if err != nil { + return fmt.Errorf("snapshot session database: %w", err) } - var notice, line string + err = os.Chmod(path, 0o600) + if err != nil { + return fmt.Errorf("set session snapshot mode: %w", err) + } - level := slog.LevelInfo + return nil +} - var attrs []any +// Close closes the session database. +func (w *WhatsApp) Close() error { + w.sessionMutex.Lock() + defer w.sessionMutex.Unlock() w.mutex.Lock() + defer w.mutex.Unlock() - switch event := event.(type) { - case *events.PairSuccess: - w.own = whatsappOwnIDs(event.ID.User, event.LID.User) - case *events.Connected: - switch { - case !w.linked || w.connected: - case w.disconnectedAt.IsZero(): - line = "connected" - default: - line = "connected again" - attrs = append(attrs, "after", w.clock().Sub(w.disconnectedAt)) - } + if w.container == nil { + return nil + } - w.connected = true - w.retrying = false - w.connectedAt = w.clock() - if w.timer != nil { - w.timer.Stop() - } + err := w.container.Close() + w.container = nil - if w.noticed { - w.noticed = false - notice = "WhatsApp is connected again." - } - case *events.Disconnected: - connected := w.connected - if w.startOutage() { - level = slog.LevelWarn - line = "disconnected, retrying" - if !connected { - line = "first connect failed, retrying" - } - } - case *events.LoggedOut: - w.retrying = false - select { - case w.logouts <- struct{}{}: - default: - } - case events.PermanentDisconnect: - w.retrying = false - if w.linked { - w.connected = false - w.noticed = true - notice = "WhatsApp is disconnected and does not connect again by itself: " + event.PermanentDisconnectDescription() + ". Restart the daemon to try again." - level = slog.LevelError - line = "disconnected for good, restart to retry" - attrs = append(attrs, "err", whatsappRedact(w.own, event.PermanentDisconnectDescription())) - } - } + return err +} - if w.closed { - line = "" - } +// Find returns the answer of whatsapp_find for name, as whatsappFind says. +func (w *WhatsApp) Find(ctx context.Context, name string) (string, error) { + return whatsappFind(ctx, w.link(), w.groups, name) +} - w.mutex.Unlock() +// Send returns the answer of whatsapp_send for text and an optional stored +// image to chat, with the durable send id that whatsappSend describes. +func (w *WhatsApp) Send(ctx context.Context, id, chat, text, image string) (string, error) { + return whatsappSend(ctx, w.link(), w.groups, w.state, w.inbox.images, w.logger, w.clock, id, chat, text, image) +} - if line != "" { - w.logger.Log(ctx, level, line, attrs...) - } +// whatsappOwnIDs returns the own number and the own LID that are not empty. +func whatsappOwnIDs(ids ...string) []string { + var own []string - if notice != "" { - w.broadcast(notice) + for _, id := range ids { + if id != "" { + own = append(own, id) + } } - return true + return own } -// disconnectedLong gives the notice of a long disconnection, unless the -// 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 +// whatsappSessionAddress returns the SQLite address of the session database at +// path, with foreign keys on, as the library needs, and a wait for a busy +// database. +func whatsappSessionAddress(path string) (string, error) { + path, err := filepath.Abs(path) + if err != nil { + return "", fmt.Errorf("find WhatsApp session: %w", err) } - since := w.disconnectedAt - w.mutex.Unlock() - - if long { - w.broadcast("WhatsApp is disconnected since " + since.Format(whatsappTimeLayout) + ". OneChat tries again.") + address := url.URL{ + Scheme: "file", + Path: path, + RawQuery: "_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)", } + + return address.String(), nil } -// broadcast gives text as a notice to all clients. -func (w *WhatsApp) broadcast(text string) { - w.mutex.Lock() +// whatsappRedact replaces each of the own IDs own in text with OWN. +func whatsappRedact(own []string, text string) string { + for _, id := range own { + text = strings.ReplaceAll(text, id, "OWN") + } - notify := w.notify - closed := w.closed - w.mutex.Unlock() + return text +} - if notify != nil && !closed { - notify(Event{ - Kind: "notice", - Text: text, - }) +// whatsappItemError returns the error of a pairing item that is not a code, a +// success, or a timeout. +func whatsappItemError(item whatsmeow.QRChannelItem) string { + if item.Error != nil { + return item.Error.Error() } + + return item.Event } -// redact replaces the own number and the own LID in text with OWN. -func (w *WhatsApp) redact(text string) string { - w.mutex.Lock() - defer w.mutex.Unlock() +// whatsappPhone returns the digits of phone, a number with its country code and +// an optional + and dashes, and tells if it has the form of a number: 7 to 15 +// digits, with no 0 first. +func whatsappPhone(phone string) (string, bool) { + phone = strings.ReplaceAll(strings.TrimPrefix(phone, "+"), "-", "") + valid := len(phone) >= 7 && len(phone) <= 15 && phone[0] != '0' && strings.Trim(phone, "0123456789") == "" - return whatsappRedact(w.own, text) + return phone, valid } // whatsappSetDeviceProps sets the device properties that the library sends at @@ -1134,19 +1194,6 @@ 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) - } - } - - return own -} - // whatsappChatHash returns a short hash of chat for the daemon log, so that a // log line can tell chats apart without a phone number. func whatsappChatHash(chat types.JID) string { @@ -1155,33 +1202,6 @@ func whatsappChatHash(chat types.JID) string { return fmt.Sprintf("%x", digest[:6]) } -// whatsappRedact replaces each of the own IDs own in text with OWN. -func whatsappRedact(own []string, text string) string { - for _, id := range own { - text = strings.ReplaceAll(text, id, "OWN") - } - - return text -} - -// whatsappSessionAddress returns the SQLite address of the session database at -// path, with foreign keys on, as the library needs, and a wait for a busy -// database. -func whatsappSessionAddress(path string) (string, error) { - path, err := filepath.Abs(path) - if err != nil { - return "", fmt.Errorf("find WhatsApp session: %w", err) - } - - address := url.URL{ - Scheme: "file", - Path: path, - RawQuery: "_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)", - } - - return address.String(), nil -} - // whatsappCommandArgs returns the text after /whatsapp of a line of a WhatsApp // command, or false for any other line. func whatsappCommandArgs(line string) (string, bool) { @@ -1197,23 +1217,3 @@ func whatsappCommandArgs(line string) (string, bool) { return strings.TrimSpace(args), true } - -// whatsappPhone returns the digits of phone, a number with its country code and -// an optional + and dashes, and tells if it has the form of a number: 7 to 15 -// digits, with no 0 first. -func whatsappPhone(phone string) (string, bool) { - phone = strings.ReplaceAll(strings.TrimPrefix(phone, "+"), "-", "") - valid := len(phone) >= 7 && len(phone) <= 15 && phone[0] != '0' && strings.Trim(phone, "0123456789") == "" - - return phone, valid -} - -// whatsappItemError returns the error of a pairing item that is not a code, a -// success, or a timeout. -func whatsappItemError(item whatsmeow.QRChannelItem) string { - if item.Error != nil { - return item.Error.Error() - } - - return item.Event -} diff --git a/whatsapp_batch.go b/whatsapp_batch.go index dcdd712..16b4b2c 100644 --- a/whatsapp_batch.go +++ b/whatsapp_batch.go @@ -48,15 +48,6 @@ type WhatsAppMessage struct { Reply bool `json:"reply,omitempty"` } -// Turn tells if the batch starts a turn: a private batch with a message of -// another person, or a group batch with a message of another person that -// mentions the user or replies to a message of the user. -func (b WhatsAppBatch) Turn() bool { - return slices.ContainsFunc(b.Messages, func(message WhatsAppMessage) bool { - return !message.Mine && (!b.Chat.Group || message.Mention || message.Reply) - }) -} - // Text returns the text of the batch for the log, with the times in location. // The first line starts with [WhatsApp] and names the chat, its ID, and the day // of the first message. Each message has a line with its time and its sender, @@ -95,3 +86,12 @@ func (b WhatsAppBatch) Text(location *time.Location) string { return strings.Join(lines, "\n") } + +// Turn tells if the batch starts a turn: a private batch with a message of +// another person, or a group batch with a message of another person that +// mentions the user or replies to a message of the user. +func (b WhatsAppBatch) Turn() bool { + return slices.ContainsFunc(b.Messages, func(message WhatsAppMessage) bool { + return !message.Mine && (!b.Chat.Group || message.Mention || message.Reply) + }) +} diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index 24789a7..1fca3b0 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -357,62 +357,6 @@ func (r *WhatsAppRetryTest) start(t *testing.T) { }) } -// whatsappStartTest opens the WhatsApp of the data directory dir with a fake -// client and the test clock, and closes it at the end of the test. The daemon -// log goes to the returned buffer. -func whatsappStartTest(t *testing.T, dir string, clock *TestClock) (*WhatsApp, *WhatsAppFake, *SyncBuffer) { - t.Helper() - - fake := &WhatsAppFake{ - connects: make(chan struct{}, 16), - pairings: make(chan chan whatsmeow.QRChannelItem, 16), - } - - logs := &SyncBuffer{} - whatsapp, err := openWhatsApp(t.Context(), dir, clock.Now, daemonTestLogger(logs), fake.newClient) - require.NoError(t, err) - - t.Cleanup(func() { - require.NoError(t, whatsapp.Close()) - }) - - return whatsapp, fake, logs -} - -// whatsappTestSession writes a session database with mode 0600 to the data -// directory dir. When linked is true, it holds a device linked to -// whatsappTestNumber and whatsappTestLID, and else no device. -func whatsappTestSession(t *testing.T, dir string, linked bool) { - t.Helper() - - path := filepath.Join(dir, whatsappDirName, whatsappSessionFileName) - require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700)) - require.NoError(t, os.WriteFile(path, nil, 0o600)) - - address, err := whatsappSessionAddress(path) - require.NoError(t, err) - - container, err := sqlstore.New(t.Context(), "sqlite", address, waLog.Noop) - require.NoError(t, err) - - if linked { - device := container.NewDevice() - jid := types.NewADJID(whatsappTestNumber, 0, 1) - device.ID = &jid - device.LID = types.NewADJID(whatsappTestLID, 0, 1) - device.LID.Server = types.HiddenUserServer - device.Account = &waAdv.ADVSignedDeviceIdentity{ - Details: []byte{}, - AccountSignature: make([]byte, 64), - AccountSignatureKey: make([]byte, 32), - DeviceSignature: make([]byte, 64), - } - require.NoError(t, device.Save(t.Context())) - } - - require.NoError(t, container.Close()) -} - // whatsappTestMessage returns the event of the text message id of sender in // chat at the time at. A sender with whatsappTestNumber is the user. func whatsappTestMessage(chat, sender types.JID, id, text string, at time.Time) *events.Message { @@ -467,6 +411,62 @@ func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) { return whatsapp, fake } +// whatsappTestSession writes a session database with mode 0600 to the data +// directory dir. When linked is true, it holds a device linked to +// whatsappTestNumber and whatsappTestLID, and else no device. +func whatsappTestSession(t *testing.T, dir string, linked bool) { + t.Helper() + + path := filepath.Join(dir, whatsappDirName, whatsappSessionFileName) + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700)) + require.NoError(t, os.WriteFile(path, nil, 0o600)) + + address, err := whatsappSessionAddress(path) + require.NoError(t, err) + + container, err := sqlstore.New(t.Context(), "sqlite", address, waLog.Noop) + require.NoError(t, err) + + if linked { + device := container.NewDevice() + jid := types.NewADJID(whatsappTestNumber, 0, 1) + device.ID = &jid + device.LID = types.NewADJID(whatsappTestLID, 0, 1) + device.LID.Server = types.HiddenUserServer + device.Account = &waAdv.ADVSignedDeviceIdentity{ + Details: []byte{}, + AccountSignature: make([]byte, 64), + AccountSignatureKey: make([]byte, 32), + DeviceSignature: make([]byte, 64), + } + require.NoError(t, device.Save(t.Context())) + } + + require.NoError(t, container.Close()) +} + +// whatsappStartTest opens the WhatsApp of the data directory dir with a fake +// client and the test clock, and closes it at the end of the test. The daemon +// log goes to the returned buffer. +func whatsappStartTest(t *testing.T, dir string, clock *TestClock) (*WhatsApp, *WhatsAppFake, *SyncBuffer) { + t.Helper() + + fake := &WhatsAppFake{ + connects: make(chan struct{}, 16), + pairings: make(chan chan whatsmeow.QRChannelItem, 16), + } + + logs := &SyncBuffer{} + whatsapp, err := openWhatsApp(t.Context(), dir, clock.Now, daemonTestLogger(logs), fake.newClient) + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, whatsapp.Close()) + }) + + return whatsapp, fake, logs +} + // whatsappTestConnect waits for the connection of the daemon and gives it the // events of a connection with an empty catch-up. func whatsappTestConnect(t *testing.T, fake *WhatsAppFake) { diff --git a/whatsapp_import.go b/whatsapp_import.go index 0021bc8..f293ebd 100644 --- a/whatsapp_import.go +++ b/whatsapp_import.go @@ -452,16 +452,6 @@ func whatsappImportNames(ctx context.Context, dir string) (names map[string]What return names, owner, nil } -// whatsappImportName adds an exact match unless distinct IDs share its name. -func whatsappImportName(names map[string]WhatsAppChat, chat WhatsAppChat) { - old, found := names[chat.Name] - if found && old.ID != chat.ID { - chat.ID = "" - } - - names[chat.Name] = chat -} - // whatsappImportPlan filters each chat at its earliest live message across the // target and root logs. Only imported days in the target are skipped. Each // remaining chat-day becomes a work message dated at its last exported message. @@ -570,6 +560,16 @@ func whatsappImportPlan(ctx context.Context, paths []string, old, rootLog []Date return plan, nil } +// whatsappImportName adds an exact match unless distinct IDs share its name. +func whatsappImportName(names map[string]WhatsAppChat, chat WhatsAppChat) { + old, found := names[chat.Name] + if found && old.ID != chat.ID { + chat.ID = "" + } + + names[chat.Name] = chat +} + // whatsappImportHistory reads batch names and live cutoffs from both logs. // Imported chat-days belong only to the target log, which is first. func whatsappImportHistory(target, root []DatedMessage, names map[string]WhatsAppChat, cutoffs map[string]time.Time, location *time.Location) (map[string]bool, error) { diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go index a104a0d..5010a74 100644 --- a/whatsapp_inbox.go +++ b/whatsapp_inbox.go @@ -230,6 +230,36 @@ func (i *WhatsAppInbox) startCatchup() { }) } +// caughtUpBy ends the catch-up of the connection with the number catchups, when +// no later connection started a new one. +func (i *WhatsAppInbox) caughtUpBy(catchups int) { + i.mutex.Lock() + defer i.mutex.Unlock() + + if i.catchups == catchups { + i.finishCatchup() + } +} + +// finishCatchup ends a running catch-up and wakes Run, which logs all batches. +// The caller holds the mutex. +func (i *WhatsAppInbox) finishCatchup() { + if !i.catchingUp { + return + } + + i.catchingUp = false + i.caughtUp = true + if i.catchupTimer != nil { + i.catchupTimer.Stop() + } + + select { + case i.wake <- struct{}{}: + default: + } +} + // receive stores images and transcribed voice notes before their batches and // wakes Run. link gives the device and the own IDs. It returns false when the // state file cannot hold the message, so that the library does not acknowledge @@ -292,36 +322,6 @@ func (i *WhatsAppInbox) imageText(ctx context.Context, link WhatsAppLink, event return imageText([]string{name}, caption) } -// caughtUpBy ends the catch-up of the connection with the number catchups, when -// no later connection started a new one. -func (i *WhatsAppInbox) caughtUpBy(catchups int) { - i.mutex.Lock() - defer i.mutex.Unlock() - - if i.catchups == catchups { - i.finishCatchup() - } -} - -// finishCatchup ends a running catch-up and wakes Run, which logs all batches. -// The caller holds the mutex. -func (i *WhatsAppInbox) finishCatchup() { - if !i.catchingUp { - return - } - - i.catchingUp = false - i.caughtUp = true - if i.catchupTimer != nil { - i.catchupTimer.Stop() - } - - select { - case i.wake <- struct{}{}: - default: - } -} - // endCatchup ends a running catch-up at OfflineSyncCompleted. func (i *WhatsAppInbox) endCatchup() { i.mutex.Lock() diff --git a/whatsapp_state.go b/whatsapp_state.go index 269f46d..6114b28 100644 --- a/whatsapp_state.go +++ b/whatsapp_state.go @@ -154,6 +154,33 @@ func (s *WhatsAppState) Add(chat WhatsAppChat, message WhatsAppMessage, arrival return closed, s.save() } +// open returns the index of the open batch of chat, or -1. The caller holds the +// mutex. +func (s *WhatsAppState) open(chat string) int { + return slices.IndexFunc(s.file.Batches, func(batch WhatsAppBatch) bool { + return !batch.Closed && batch.Chat.ID == chat + }) +} + +// save writes the state file whole. The caller holds the mutex. +func (s *WhatsAppState) save() error { + data, err := json.Marshal(s.file) + if err != nil { + return fmt.Errorf("encode WhatsApp state: %w", err) + } + + err = os.MkdirAll(filepath.Dir(s.path), 0o700) + if err == nil { + err = writeFileAtomic(s.path, append(data, '\n')) + } + + if err != nil { + return fmt.Errorf("save WhatsApp state: %w", err) + } + + return nil +} + // CloseDue closes each open batch whose time is up at now and tells if one // closed. func (s *WhatsAppState) CloseDue(now time.Time) (bool, error) { @@ -176,6 +203,17 @@ func (s *WhatsAppState) CloseDue(now time.Time) (bool, error) { return true, s.save() } +// deadline returns when batch closes in real time. +func (s *WhatsAppState) deadline(batch WhatsAppBatch) time.Time { + quiet := batch.Last.Add(whatsappBatchIdleTimeout) + window := batch.First.Add(whatsappBatchTimeout) + if window.Before(quiet) { + return window + } + + return quiet +} + // CloseAll closes every open batch. func (s *WhatsAppState) CloseAll() error { s.mutex.Lock() @@ -274,41 +312,3 @@ func (s *WhatsAppState) PutSend(record WhatsAppSend) error { return s.save() } - -// open returns the index of the open batch of chat, or -1. The caller holds the -// mutex. -func (s *WhatsAppState) open(chat string) int { - return slices.IndexFunc(s.file.Batches, func(batch WhatsAppBatch) bool { - return !batch.Closed && batch.Chat.ID == chat - }) -} - -// deadline returns when batch closes in real time. -func (s *WhatsAppState) deadline(batch WhatsAppBatch) time.Time { - quiet := batch.Last.Add(whatsappBatchIdleTimeout) - window := batch.First.Add(whatsappBatchTimeout) - if window.Before(quiet) { - return window - } - - return quiet -} - -// save writes the state file whole. The caller holds the mutex. -func (s *WhatsAppState) save() error { - data, err := json.Marshal(s.file) - if err != nil { - return fmt.Errorf("encode WhatsApp state: %w", err) - } - - err = os.MkdirAll(filepath.Dir(s.path), 0o700) - if err == nil { - err = writeFileAtomic(s.path, append(data, '\n')) - } - - if err != nil { - return fmt.Errorf("save WhatsApp state: %w", err) - } - - return nil -}