clients get a notice at once. When the user removes the device in the
phone, the daemon deletes the session and tells all clients one time.
+***********
+ Schedules
+***********
+
+The agent sets reminders and timed tasks when the user asks for them,
+such as "remind me tomorrow at 9" or "every weekday at 8, tell me what
+my WhatsApp groups talked about". A schedule has a text, the task for
+the agent, and either one time or a cron expression. The agent also
+lists the schedules with their next times, and removes one when the user
+asks. Only the agent has the schedule tools, not its subagents.
+
+A cron expression has 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
+``*``, ``N``, and ``N-M``, and ``*`` and ``N-M`` can have a step ``/S``,
+as in ``0 8 * * 1-5`` or ``*/15 9-17 * * *``. When both day fields start
+with something other than ``*``, a day matches when either field
+matches. Names such as ``MON``, macros such as ``@daily``, the
+characters ``? L W #``, and ``N/S`` are refused. The runs must be at
+least 5 minutes apart, because each run is a full turn that uses the
+quota, so ``*/5 * * * *`` is the shortest schedule.
+
+The times use the time zone of the machine. A time that the clock skips
+at the start of daylight saving time runs later by the length of the
+skip. A time that the clock repeats at its end runs once, at its first
+occurrence.
+
+At its time, a schedule goes to the chat as a work message that starts
+with ``[Schedule ID]``, with its time and its text, and it starts a
+turn. The message ends with a line that tells the agent to end its turn
+with an empty reply when nothing needs the user. The terminal client
+shows the message, and Telegram shows only the reply. After its run, a
+one-time schedule goes, and a repeating schedule gets its next time.
+
+``agent/schedules.json``, with mode 0600, holds the schedules. Each
+change writes it before the tool answers, so the schedules survive a
+restart. A schedule that came due while the daemon was stopped runs once
+at the next start, with its original time and a line that says that the
+run is late. A repeating schedule that missed several times runs once. A
+crash just after a run goes to the chat can make the run repeat once.
+
*******
Probe
*******
}
}
- return daemonServe(ctx, listener, telegram, whatsapp, func(notify func(Event)) (*TurnLoop, error) {
+ return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, notify)
})
}
// daemonServe runs the loop that start makes and serves the clients of listener
// until ctx ends or the loop fails. It closes listener. It runs whatsapp, which
// logs its batches and notes in the loop and tells all clients its notices, and
-// a failure of whatsapp stops the agent. When bot is not nil, it runs the bot
-// too, and a failure of the bot stops the agent.
-func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, start func(notify func(Event)) (*TurnLoop, error)) error {
+// a failure of whatsapp stops the agent. It runs schedules, which log their
+// runs in the loop, and a failure of schedules stops the agent too. When bot is
+// not nil, it runs the bot too, and a failure of the bot stops the agent.
+func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
cancel()
}()
+ var schedulesErr error
+ workers.Add(1)
+ go func() {
+ defer workers.Done()
+
+ schedulesErr = schedules.Run(ctx, l)
+ cancel()
+ }()
+
var botErr error
if bot != nil {
workers.Add(1)
hub.CloseAll()
workers.Wait()
- return errors.Join(err, whatsappErr, botErr)
+ return errors.Join(err, whatsappErr, schedulesErr, botErr)
}
// daemonRunRequest runs one request of a client. The answers of a WhatsApp
"context"
"path/filepath"
"testing"
+ "time"
"github.com/stretchr/testify/require"
)
func daemonStartTestWithWhatsApp(t *testing.T, server *ModelFake, bot *TelegramBot, whatsapp *WhatsApp) (string, *Memory) {
t.Helper()
+ clock := newTestClock()
+
+ return daemonStartTestWithSchedules(t, server, bot, whatsapp, newTestSchedules(t, clock.Now), clock.Now)
+}
+
+// daemonStartTestWithSchedules is daemonStartTestWithWhatsApp with schedules,
+// and with clock for the memory and the turns.
+func daemonStartTestWithSchedules(t *testing.T, server *ModelFake, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, clock func() time.Time) (string, *Memory) {
+ t.Helper()
+
dir := t.TempDir()
agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
require.NoError(t, err)
agent.config.Workdir = t.TempDir()
- clock := newTestClock()
- memory, err := openMemory(t.Context(), agent.dir, clock.Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), agent.dir, clock, newFakeSummarizer())
require.NoError(t, err)
socket := filepath.Join(dir, socketFileName)
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
go func() {
- done <- daemonServe(ctx, listener, bot, whatsapp, func(notify func(Event)) (*TurnLoop, error) {
- return newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, clock.Now), client, clock.Now, notify)
+ done <- daemonServe(ctx, listener, bot, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+ return newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, schedules, client, clock, notify)
})
}()
import (
"cmp"
+ "context"
"encoding/json"
"errors"
"fmt"
// tools and in the log.
scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00"
+ // scheduleSource is the source of the work events of the runs.
+ scheduleSource = "schedule"
+
+ // scheduleCheck is the longest wait of Run before it looks at the times
+ // again, so that it sees a suspend of the machine or a change of the
+ // clock.
+ scheduleCheck = time.Minute
+
// scheduleLate is how long after its time a run is late.
scheduleLate = time.Minute
// none. The clock gives the time and the time zone of the schedules.
func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
s := &Schedules{
- path: path,
- clock: clock,
+ path: path,
+ clock: clock,
+ check: scheduleCheck,
+ changed: make(chan struct{}, 1),
}
data, err := os.ReadFile(path)
// Schedules holds the schedules of the agent: tasks that start a turn at one
// time or at the times of a cron expression. Each change writes the file whole
// before it returns, so a schedule survives a restart. The mutex protects the
-// schedules, because the tools change them from a turn while the runs read
-// them.
+// schedules, because the tools change them from a turn while Run reads them.
+// changed tells Run that a set or a remove changed the times.
type Schedules struct {
- path string
- clock func() time.Time
+ path string
+ clock func() time.Time
+ check time.Duration
+ changed chan struct{}
mutex sync.Mutex
file ScheduleFile
}
+// Run starts the runs of the due schedules in l until ctx ends. It waits until
+// the first due time, at most check at a time, and also wakes at a set or a
+// remove. So a schedule that came due while the daemon was stopped runs at the
+// start. It returns the error of a write of the log or of the file.
+func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
+ for {
+ err := s.deliver(ctx, l)
+ if err != nil {
+ return err
+ }
+
+ var timer *time.Timer
+ var timeout <-chan time.Time
+ wait, found := s.Wait(s.clock())
+ if found {
+ timer = time.NewTimer(min(wait, s.check))
+ timeout = timer.C
+ }
+
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-s.changed:
+ case <-timeout:
+ }
+
+ if timer != nil {
+ timer.Stop()
+ }
+ }
+}
+
// 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
return "", err
}
+ s.change()
+
return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), 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
+// crash between the two makes the run repeat once.
+func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
+ now := s.clock()
+ var turns []Message
+ for _, schedule := range s.Due(now) {
+ message, err := l.Work(scheduleSource, schedule.Message(now), "")
+ if err != nil {
+ return fmt.Errorf("log a schedule: %w", err)
+ }
+
+ // A stop of the daemon during the flush is no failure: the
+ // schedule stays due and runs at the next start.
+ err = l.Flush(ctx)
+ if ctx.Err() != nil {
+ return nil
+ }
+
+ if err != nil {
+ return fmt.Errorf("log a schedule: %w", err)
+ }
+
+ err = s.Done(schedule, now)
+ if err != nil {
+ return err
+ }
+
+ turns = append(turns, message)
+ }
+
+ l.Wake(turns...)
+
+ return nil
+}
+
// Due returns the schedules whose time is not after now, by due time.
func (s *Schedules) Due(now time.Time) []Schedule {
var due []Schedule
return max(schedules[0].Due.Sub(now), 0), true
}
+// change wakes Run after a set or a remove.
+func (s *Schedules) change() {
+ select {
+ case s.changed <- struct{}{}:
+ default:
+ }
+}
+
// sorted returns a copy of the schedules by due time, then by ID.
func (s *Schedules) sorted() []Schedule {
s.mutex.Lock()
package main
import (
+ "context"
"os"
"path/filepath"
"strings"
+ "sync"
"testing"
"time"
require.ErrorContains(t, err, "schedule 1: the runs of")
}
-// scheduleTestOpen opens the schedules of a new file, with a clock at Sunday
-// 2026-11-08 20:00 -03:00, and returns them and the path of the file.
-func scheduleTestOpen(t *testing.T) (*Schedules, string) {
- path := filepath.Join(t.TempDir(), scheduleFileName)
- clock := &TestClock{
+func TestSchedulesRunStartsATurnWhenDue(t *testing.T) {
+ clock := &ScheduleTestClock{
now: time.Date(2026, time.November, 8, 20, 0, 0, 0, time.FixedZone("", -3*60*60)),
}
+ schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), clock.Now)
+ require.NoError(t, err)
+
+ server := newModelFake(t)
+ test := scheduleTestRun(t, schedules, server)
+
+ _, err = schedules.Set("Remind the user to drink water.", new("2026-11-08 20:05"), nil)
+ require.NoError(t, err)
+
+ clock.Set(clock.Now().Add(5 * time.Minute))
+ test.events.WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+
+ texts := turnInputTexts(t, server.Call(0))
+ require.Equal(t, "[Schedule 1] Sunday 2026-11-08 20:05 -03:00, once\nRemind the user to drink water.\n"+scheduleEmptyReply, texts[len(texts)-1])
+ require.Equal(t, scheduleSource, test.events.LastEvent("work").Source)
+ require.Equal(t, "No schedules.", schedules.List())
+}
+
+func TestSchedulesRunMissedSchedulesOnceAtTheStart(t *testing.T) {
+ path := filepath.Join(t.TempDir(), scheduleFileName)
+ require.NoError(t, os.WriteFile(path, []byte(`{"next":2,"schedules":[{"id":1,"text":"Remind the user to call.","cron":"","due":"2026-11-08T09:00:00-03:00"},{"id":2,"text":"Tell the user the time.","cron":"0 * * * *","due":"2026-11-08T10:00:00-03:00"}]}`), 0o600))
+
+ clock := &ScheduleTestClock{
+ now: time.Date(2026, time.November, 8, 20, 30, 0, 0, time.FixedZone("", -3*60*60)),
+ }
+
schedules, err := openSchedules(path, clock.Now)
require.NoError(t, err)
- return schedules, path
+ server := newModelFake(t)
+ test := scheduleTestRun(t, schedules, server)
+ test.events.WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+
+ texts := turnInputTexts(t, server.Call(0))
+ require.Equal(t, []string{
+ "[Schedule 1] Sunday 2026-11-08 09:00 -03:00, once\nRemind the user to call.\n" + scheduleLateLine + "\n" + scheduleEmptyReply,
+ "[Schedule 2] Sunday 2026-11-08 10:00 -03:00, repeats 0 * * * *\nTell the user the time.\n" + scheduleLateLine + "\n" + scheduleEmptyReply,
+ }, texts[len(texts)-2:])
+ require.Equal(t, "2 | Sunday 2026-11-08 21:00 -03:00 | 0 * * * * | Tell the user the time.", schedules.List())
+}
+
+func TestDaemonRunsADueSchedule(t *testing.T) {
+ clock := &ScheduleTestClock{
+ now: time.Date(2026, time.November, 8, 20, 0, 0, 0, time.FixedZone("", -3*60*60)),
+ }
+
+ schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), clock.Now)
+ require.NoError(t, err)
+
+ schedules.check = time.Millisecond
+ whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), newTestClock())
+ socket, _ := daemonStartTestWithSchedules(t, newModelFake(t), nil, whatsapp, schedules, clock.Now)
+ client := terminalStartTestClient(t, socket)
+ client.WaitFor(t, "[connected to")
+
+ _, err = schedules.Set("Remind the user to drink water.", new("2026-11-08 20:05"), nil)
+ require.NoError(t, err)
+
+ clock.Set(clock.Now().Add(5 * time.Minute))
+ client.WaitFor(t, "[Schedule 1] Sunday 2026-11-08 20:05 -03:00, once\nRemind the user to drink water.\n"+scheduleEmptyReply+"\n[working]\na line\n[done]")
}
func TestToolboxRunsTheScheduleTools(t *testing.T) {
require.Equal(t, `error: unknown tool "schedule_list"`, box.run(t.Context(), "schedule_list", `{}`))
require.Equal(t, `error: unknown tool "schedule_remove"`, box.run(t.Context(), "schedule_remove", `{"id":1}`))
}
+
+// ScheduleTestClock is a clock that a test sets while Run reads it.
+type ScheduleTestClock struct {
+ mutex sync.Mutex
+ now time.Time
+}
+
+// Now returns the time of the clock.
+func (c *ScheduleTestClock) Now() time.Time {
+ c.mutex.Lock()
+ defer c.mutex.Unlock()
+
+ return c.now
+}
+
+// Set sets the time of the clock.
+func (c *ScheduleTestClock) Set(now time.Time) {
+ c.mutex.Lock()
+ defer c.mutex.Unlock()
+
+ c.now = now
+}
+
+// scheduleTestOpen opens the schedules of a new file, with a clock at Sunday
+// 2026-11-08 20:00 -03:00, and returns them and the path of the file.
+func scheduleTestOpen(t *testing.T) (*Schedules, string) {
+ path := filepath.Join(t.TempDir(), scheduleFileName)
+ clock := &TestClock{
+ now: time.Date(2026, time.November, 8, 20, 0, 0, 0, time.FixedZone("", -3*60*60)),
+ }
+
+ schedules, err := openSchedules(path, clock.Now)
+ require.NoError(t, err)
+
+ return schedules, path
+}
+
+// scheduleTestRun runs schedules in a new turn loop that calls server, until
+// the test ends. Run looks at the times every millisecond.
+func scheduleTestRun(t *testing.T, schedules *Schedules, server *ModelFake) *TurnTestLoop {
+ t.Helper()
+
+ schedules.check = time.Millisecond
+ test := newTestTurnLoop(t, server)
+ ctx, cancel := context.WithCancel(t.Context())
+ done := make(chan error)
+ go func() {
+ done <- schedules.Run(ctx, test.loop)
+ }()
+
+ t.Cleanup(func() {
+ cancel()
+ require.NoError(t, <-done)
+ })
+
+ return test
+}
}
// Notify queues event for the chat. It never blocks, so a slow network does not
-// stop the agent. The WhatsApp batches do not go to the chat: the reply of the
-// agent tells the user what needs the user.
+// 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 {
+ if event.Source == whatsappSource || event.Source == scheduleSource {
return
}
case "user":
}, sent)
}
+func TestBotSkipsTheWorkOfWhatsAppAndTheSchedules(t *testing.T) {
+ bot := &TelegramBot{
+ ready: make(chan struct{}, 1),
+ }
+
+ bot.Notify(Event{
+ Kind: "work",
+ Source: whatsappSource,
+ Text: "[WhatsApp] Ana Souza",
+ })
+ bot.Notify(Event{
+ Kind: "work",
+ Source: scheduleSource,
+ Text: "[Schedule 1] Sunday 2026-11-08 20:05 -03:00, once",
+ })
+ require.Empty(t, bot.queue)
+}
+
func TestBotStartsAtTheStoredOffset(t *testing.T) {
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
projects / onechat / commitdiff