From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 05:50:59 +0000 (-0300) Subject: Keep each schedule with its owning agent X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/2124c559da8f164a43a3da987b399b6ac6125035 Keep each schedule with its owning agent Schedules record the agent that sets them and wake only its loop. Saved schedules without an owner take the name of their agent, so root tasks keep their behavior across a restart. A different stored owner stops startup instead of sending the task to the wrong agent. --- diff --git a/README.rst b/README.rst index b60f984..b413c2b 100644 --- a/README.rst +++ b/README.rst @@ -572,7 +572,8 @@ 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. +asks. The root and its children have the schedule tools, not their +subagents. Each schedule records its owner and wakes only that agent. 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 diff --git a/backup_test.go b/backup_test.go index 6271f8a..8628ee4 100644 --- a/backup_test.go +++ b/backup_test.go @@ -378,7 +378,7 @@ func backupStartDaemon(t *testing.T, dir string) (*Memory, *WhatsApp, *Subagents whatsapp, _, _ := whatsappStartTest(t, dir, clock) subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), clock.Now, subagentNameRandom) - schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), clock.Now, discardLog()) + schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), "OneChat", clock.Now, discardLog()) require.NoError(t, err) server := newModelFake(t) diff --git a/daemon_agent.go b/daemon_agent.go index 1161c8a..4520f35 100644 --- a/daemon_agent.go +++ b/daemon_agent.go @@ -60,7 +60,7 @@ func openDaemonAgent(ctx context.Context, agent *Agent, address string, auth Mod return nil, fmt.Errorf("open memory: %w", err) } - schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), clock, logger) + schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), agent.config.Name, clock, logger) if err != nil { return nil, errors.Join(fmt.Errorf("open schedules: %w", err), memory.Close()) } diff --git a/daemon_message_test.go b/daemon_message_test.go index f9c25f1..060faf5 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -532,6 +532,33 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) { } } +func TestChildScheduleWakesOnlyItsOwner(t *testing.T) { + server := newModelFake(t) + server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5)) + } + test := newDaemonMessageTest(t, t.TempDir(), server) + child := test.agents[1] + at := child.clock().Add(time.Minute).Format(scheduleAtLayout) + output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`) + require.Contains(t, output, "Schedule 1 set.") + require.Equal(t, "Garden", child.schedules.sorted()[0].Owner) + + child.schedules.clock = func() time.Time { + return child.clock().Add(time.Minute) + } + require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match") + require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1])) + test.Start(t, 0, 1, 2) + test.events[1].WaitFor(t, "idle", 1) + require.Equal(t, 1, server.Calls()) + require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"]) + require.Zero(t, test.agents[0].memory.Len()) + require.Zero(t, test.agents[2].memory.Len()) + require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.") + require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source) +} + // DaemonMessageTest owns test agents and the loops selected by Start. type DaemonMessageTest struct { agents DaemonAgents diff --git a/schedule.go b/schedule.go index 5da024d..b3a65ea 100644 --- a/schedule.go +++ b/schedule.go @@ -49,11 +49,13 @@ const ( ) // openSchedules reads the schedules of the file at path. A missing file holds -// none. The clock gives the time and the time zone of the schedules, and logger -// gets one line for each run. -func openSchedules(path string, clock func() time.Time, logger *slog.Logger) (*Schedules, error) { +// none. The owner names its agent. Schedules without an owner take this name. A +// different stored owner is an error. The clock gives the time and the time +// zone of the schedules, and logger gets one line for each run. +func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logger) (*Schedules, error) { s := &Schedules{ path: path, + owner: owner, clock: clock, logger: logger, check: scheduleCheck, @@ -74,7 +76,18 @@ func openSchedules(path string, clock func() time.Time, logger *slog.Logger) (*S return nil, fmt.Errorf("%s: %w", path, err) } - for _, schedule := range s.file.Schedules { + changed := false + for i := range s.file.Schedules { + schedule := &s.file.Schedules[i] + if schedule.Owner == "" { + schedule.Owner = owner + changed = true + } + + if schedule.Owner != owner { + return nil, fmt.Errorf("schedule %d belongs to %q, not %q", schedule.ID, schedule.Owner, owner) + } + if schedule.Cron == "" { continue } @@ -85,6 +98,13 @@ func openSchedules(path string, clock func() time.Time, logger *slog.Logger) (*S } } + if changed { + err = s.save() + if err != nil { + return nil, fmt.Errorf("record schedule owner: %w", err) + } + } + return s, nil } @@ -95,6 +115,7 @@ func openSchedules(path string, clock func() time.Time, logger *slog.Logger) (*S // changed tells Run that a set or a remove changed the times. type Schedules struct { path string + owner string clock func() time.Time logger *slog.Logger check time.Duration @@ -151,7 +172,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) { now := s.clock() schedule := Schedule{ - Text: text, + Text: text, + Owner: s.owner, } switch { @@ -255,6 +277,10 @@ func (s *Schedules) Remove(id int) (string, error) { // the ID of the schedule and, for a late run, how late it is. The line holds no // text of the schedule. func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { + if l.agent.config.Name != s.owner { + return fmt.Errorf("schedule owner %q does not match agent %q", s.owner, l.agent.config.Name) + } + now := s.clock() var turns []Message @@ -402,13 +428,15 @@ type ScheduleFile struct { Schedules []Schedule `json:"schedules"` } -// Schedule is one schedule: its ID, the task for the agent, the cron expression -// of a repeating schedule, empty for a one-time schedule, and its next time. +// Schedule is one schedule: its owner, its ID, the task for the agent, the cron +// expression of a repeating schedule, empty for a one-time schedule, and its +// next time. type Schedule struct { - ID int `json:"id"` - Text string `json:"text"` - Cron string `json:"cron"` - Due time.Time `json:"due"` + Owner string `json:"owner"` + ID int `json:"id"` + Text string `json:"text"` + Cron string `json:"cron"` + Due time.Time `json:"due"` } // Message returns the text of a run at now for the log, with the times in the diff --git a/schedule_helper_test.go b/schedule_helper_test.go index 814056c..d1e3d1f 100644 --- a/schedule_helper_test.go +++ b/schedule_helper_test.go @@ -12,7 +12,7 @@ import ( func newTestSchedules(t *testing.T, clock func() time.Time) *Schedules { t.Helper() - schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), clock, discardLog()) + schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), "OneChat", clock, discardLog()) require.NoError(t, err) return schedules diff --git a/schedule_test.go b/schedule_test.go index 3048d75..6610c26 100644 --- a/schedule_test.go +++ b/schedule_test.go @@ -128,7 +128,7 @@ func TestSchedulesSurviveARestart(t *testing.T) { require.NoError(t, err) require.Equal(t, os.FileMode(0o600), info.Mode().Perm()) - reopened, err := openSchedules(path, schedules.clock, discardLog()) + reopened, err := openSchedules(path, "OneChat", schedules.clock, discardLog()) require.NoError(t, err) require.Equal(t, "2 | Monday 2026-11-09 08:00 -03:00 | 0 8 * * 1-5 | Tell the user the news.", reopened.List()) @@ -231,7 +231,7 @@ func TestOpenSchedulesRefusesABadCron(t *testing.T) { path := filepath.Join(t.TempDir(), scheduleFileName) require.NoError(t, os.WriteFile(path, []byte(`{"next":1,"schedules":[{"id":1,"text":"x","cron":"* * * * *","due":"2026-11-09T09:00:00-03:00"}]}`), 0o600)) - _, err := openSchedules(path, time.Now, discardLog()) + _, err := openSchedules(path, "OneChat", time.Now, discardLog()) require.ErrorContains(t, err, "schedule 1: the runs of") } @@ -241,7 +241,7 @@ func TestSchedulesRunStartsATurnWhenDue(t *testing.T) { } logs := &SyncBuffer{} - schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), clock.Now, daemonTestLogger(logs)) + schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), "OneChat", clock.Now, daemonTestLogger(logs)) require.NoError(t, err) server := newModelFake(t) @@ -270,7 +270,7 @@ func TestSchedulesRunMissedSchedulesOnceAtTheStart(t *testing.T) { } logs := &SyncBuffer{} - schedules, err := openSchedules(path, clock.Now, daemonTestLogger(logs)) + schedules, err := openSchedules(path, "OneChat", clock.Now, daemonTestLogger(logs)) require.NoError(t, err) server := newModelFake(t) @@ -292,7 +292,7 @@ func TestDaemonRunsADueSchedule(t *testing.T) { 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, discardLog()) + schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), "OneChat", clock.Now, discardLog()) require.NoError(t, err) schedules.check = time.Millisecond @@ -329,6 +329,36 @@ func TestSubagentHasNoScheduleTools(t *testing.T) { require.Equal(t, `error: unknown tool "schedule_remove"`, box.run(t.Context(), "schedule_remove", `{"id":1}`)) } +func TestSchedulesKeepTheirOwnerAcrossRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), scheduleFileName) + clock := newTestClock() + schedules, err := openSchedules(path, "GreenOtter", clock.Now, discardLog()) + require.NoError(t, err) + + _, err = schedules.Set("Check the plants.", new(clock.Now().Add(time.Minute).Format(scheduleAtLayout)), nil) + require.NoError(t, err) + + reopened, err := openSchedules(path, "GreenOtter", clock.Now, discardLog()) + require.NoError(t, err) + require.Equal(t, "GreenOtter", reopened.sorted()[0].Owner) + + _, err = openSchedules(path, "OneChat", clock.Now, discardLog()) + require.ErrorContains(t, err, "belongs to") +} + +func TestSchedulesAssignTheOwnerToSavedRootSchedules(t *testing.T) { + path := filepath.Join(t.TempDir(), scheduleFileName) + require.NoError(t, os.WriteFile(path, []byte(`{"next":1,"schedules":[{"id":1,"text":"Check the news.","due":"2026-11-09T09:00:00-03:00"}]}`), 0o600)) + + schedules, err := openSchedules(path, "OneChat", newTestClock().Now, discardLog()) + require.NoError(t, err) + require.Equal(t, "OneChat", schedules.sorted()[0].Owner) + + data, err := os.ReadFile(path) + require.NoError(t, err) + require.Contains(t, string(data), `"owner":"OneChat"`) +} + // ScheduleTestClock is a clock that a test sets while Run reads it. type ScheduleTestClock struct { mutex sync.Mutex @@ -359,7 +389,7 @@ func scheduleTestOpen(t *testing.T) (*Schedules, string) { now: time.Date(2026, time.November, 8, 20, 0, 0, 0, time.FixedZone("", -3*60*60)), } - schedules, err := openSchedules(path, clock.Now, discardLog()) + schedules, err := openSchedules(path, "OneChat", clock.Now, discardLog()) require.NoError(t, err) return schedules, path