From: Anderson A. Giacomolli <agiacomolli@gmail.com>
Date: Mon, 9 Nov 2026 00:35:08 +0000 (-0300)
Subject: Start a turn when a schedule comes due
X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/d008b46cbfc6348255e9b9a1fc009f0e4bd6aa2a?hp=e737005f89c7c663e1636830001730ad79396bdc

Start a turn when a schedule comes due

The daemon runs the schedules next to WhatsApp. At its time, a schedule
goes to the chat as a work message that starts with [Schedule ID] and
starts a turn of the root agent, which does the task or spawns a
subagent for long work. The message ends with the empty reply line, so a
task that finds nothing for the user gives no message.

The runner waits until the first due time, at most one minute at a time,
so it sees a suspend of the machine or a change of the clock within a
minute. A set or a remove wakes it. It logs all due schedules in due
order, flushes the log, ends each run in the file, and wakes the turn
loop once, as the WhatsApp batches do. A crash between the flush and the
file write makes a run repeat once. So a schedule that came due while
the daemon was stopped runs once at the next start, with the line that
the run is late.

Telegram does not send the work message, because the reply of the agent
carries its content. The terminal client shows it. The README describes
the schedules, the cron subset, the limit of 5 minutes, and the rules
for daylight saving time.

Issue: 60e4e4f
---

diff --git a/README.rst b/README.rst
index 93be784..7930b0e 100644
--- a/README.rst
+++ b/README.rst
@@ -276,6 +276,46 @@ good, for example because another program uses the same session, all
 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
 *******
diff --git a/daemon.go b/daemon.go
index 2d83017..3a950da 100644
--- a/daemon.go
+++ b/daemon.go
@@ -122,7 +122,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		}
 	}
 
-	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)
 	})
 }
@@ -313,9 +313,10 @@ func daemonListenSocket(path string) (net.Listener, error) {
 // 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()
 
@@ -353,6 +354,15 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
 		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)
@@ -374,7 +384,7 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
 	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
diff --git a/daemon_start_helper_test.go b/daemon_start_helper_test.go
index e7dea6c..6e4e328 100644
--- a/daemon_start_helper_test.go
+++ b/daemon_start_helper_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -24,13 +25,22 @@ func daemonStartTestWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) s
 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)
@@ -41,8 +51,8 @@ func daemonStartTestWithWhatsApp(t *testing.T, server *ModelFake, bot *TelegramB
 	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)
 		})
 	}()
 
diff --git a/schedule.go b/schedule.go
index 3114ac7..d27274b 100644
--- a/schedule.go
+++ b/schedule.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"cmp"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -28,6 +29,14 @@ const (
 	// 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
 
@@ -42,8 +51,10 @@ const (
 // 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)
@@ -77,16 +88,50 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
 // 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
@@ -141,6 +186,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) {
 		return "", err
 	}
 
+	s.change()
+
 	return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 }
 
@@ -181,9 +228,48 @@ func (s *Schedules) Remove(id int) (string, error) {
 		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
@@ -246,6 +332,14 @@ func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
 	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()
diff --git a/schedule_test.go b/schedule_test.go
index c13f023..f83c2e4 100644
--- a/schedule_test.go
+++ b/schedule_test.go
@@ -1,9 +1,11 @@
 package main
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -232,18 +234,73 @@ func TestOpenSchedulesRefusesABadCron(t *testing.T) {
 	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) {
@@ -266,3 +323,60 @@ func TestSubagentHasNoScheduleTools(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
+}
diff --git a/telegram.go b/telegram.go
index 814f454..e77ea69 100644
--- a/telegram.go
+++ b/telegram.go
@@ -160,13 +160,13 @@ type TelegramBot struct {
 }
 
 // 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":
diff --git a/telegram_test.go b/telegram_test.go
index 5e50ec1..66a5f47 100644
--- a/telegram_test.go
+++ b/telegram_test.go
@@ -179,6 +179,24 @@ func TestBotShowsOnlyTheStartAndTheEndOfASubagent(t *testing.T) {
 	}, 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) {