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.
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 {
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
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") {
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 {
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)
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
-}
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.
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 {
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.
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) {
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
-}
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)
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)
}
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
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 {
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
-}
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)
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) {
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 {
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 {
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.
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)
+}
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
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) {
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)
+}
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
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)
- }
-}
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.
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{},
+ })
+}
})
}
+// 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 {
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
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
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()
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
-}
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]
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)
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) {
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 {
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())
}
}
+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
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) {
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) {
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} {
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 {
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()
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 {
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,
+ }
+}
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 {
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.
}
}
-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{}
}
}
+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()
// 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 {
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 {
}
}
-// 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 {
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,
}
}
}
}
-// 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,
}
}
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))
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()
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
}
}
-// 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
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
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.
}
}
-// 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
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
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) {
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) {
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 {
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()
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()
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())
-}
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 {
return false
}
-
-// subagentWord returns word with its first letter in upper case.
-func subagentWord(word string) string {
- return strings.ToUpper(word[:1]) + word[1:]
-}
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) {
})
}
-// 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 {
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 {
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
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()
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.
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.
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()
-}
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() {
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()
+ }
+ }
+}
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.
}
}
+// 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{
}
}
-// 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.
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) {
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.
}, 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
}
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.
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)
-}
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 {
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{
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.
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
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.
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)
+}
}
}
-// 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()
}, 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()
}
}
+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()
return ctx.Err()
}
-
-func (r *ToolRecordingLog) Events() []string {
- r.mutex.Lock()
- defer r.mutex.Unlock()
-
- return append([]string(nil), r.events...)
-}
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
})
}
+// 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 {
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 {
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.
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
}, 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) {
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)
-}
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 := "</chat>\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
}
}
+// 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.
}
}
+// 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) {
}
}
-// 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()
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()
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 := "</chat>\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 {
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()
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.
return texts
}
-
-func modelReasoningItem(id, encrypted string) map[string]any {
- return map[string]any{
- "type": "reasoning",
- "id": id,
- "encrypted_content": encrypted,
- "summary": []any{},
- }
-}
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
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("<chat>\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("</chat>")
-
- 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
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("<chat>\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("</chat>")
- return node.Size
+ return text.String()
}
// restoreView returns the view of state with the limits high and low. Each line
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
}
}
-// 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
}
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
}
}
-// 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)
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
}
}
-// 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 {
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) {
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
-}
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,
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)
+ })
+}
})
}
-// 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 {
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) {
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.
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) {
})
}
+// 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
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()
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) {
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()
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
-}
projects / onechat / commitdiff