Skip to content

Commit 3f10ddb

Browse files
jkyberneeesclaude
andauthored
feat(schedule): record delivered Telegram results into the chat session (#35)
* feat(schedule): record delivered Telegram results into the chat session Scheduled tasks ran fully headless and their delivered message was never written back to the chat's conversation, so a follow-up turn had amnesia about what the schedule just posted ("what did that scheduled task find?" hit an agent that never saw its own output). After a successful Telegram delivery, append the scheduled task + its result to the target chat's session as a labeled user turn + the assistant result. The scheduled run itself stays isolated (deterministic) — only its output is recorded. Design choices: - Write-back is serialized through the existing per-chat mutex (getChatMutex), so it can't interleave with a live interactive turn's read-modify-write or with a concurrent delivery to the same chat. - It attaches only to an EXISTING session — a notification-only chat that's never been used interactively is not given an ever-growing transcript. - Best-effort: a write-back failure is logged, never fails the run (the message was already sent) and never triggers redelivery. - Wired only on the Telegram bot's embedded scheduler (which owns the SessionManager); the CLI daemon path passes nil and is a safe no-op. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(schedule): keep role alternation when recording into a user-ending session vprotocol axis 2.1: a chat session can already end on a bare user message — a turn cancelled before the agent replied (telegram.go save-on-cancel) or a context-injection command. The write-back appended its own user turn there, producing two consecutive user messages, which Anthropic rejects with a 400 on the next interactive turn. When the session ends on a user message, fold the scheduled label into a single assistant message instead; either way the session ends on an assistant turn with no two same-role messages in a row. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 4dc1487 commit 3f10ddb

4 files changed

Lines changed: 277 additions & 6 deletions

File tree

cmd/odek/schedule.go

Lines changed: 76 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616

1717
"github.com/BackendStack21/odek"
1818
"github.com/BackendStack21/odek/internal/config"
19+
"github.com/BackendStack21/odek/internal/llm"
1920
"github.com/BackendStack21/odek/internal/loop"
2021
"github.com/BackendStack21/odek/internal/render"
2122
"github.com/BackendStack21/odek/internal/schedule"
@@ -462,9 +463,16 @@ func (r telegramRunner) Run(ctx context.Context, job schedule.Job) (string, int6
462463

463464
// telegramDeliverer delivers via the live bot for telegram jobs (sharing its
464465
// client and rate limiting) and falls back to the CLI deliverer for stdout/log.
466+
//
467+
// sessions, when set, lets a delivered result be recorded into the target
468+
// chat's conversation (Option B) so a follow-up message in that chat sees what
469+
// the schedule posted. The scheduled run itself stays isolated — only its
470+
// output is written back.
465471
type telegramDeliverer struct {
466472
bot *telegram.Bot
467473
fallback cliDeliverer
474+
sessions *telegram.SessionManager
475+
log schedule.Logger
468476
}
469477

470478
func (d telegramDeliverer) Deliver(ctx context.Context, job schedule.Job, result string) error {
@@ -478,14 +486,79 @@ func (d telegramDeliverer) Deliver(ctx context.Context, job schedule.Job, result
478486
if chatID == 0 {
479487
return fmt.Errorf("no chat id (set the job's telegram:<chatID> or telegram.default_chat_id)")
480488
}
481-
return sendTelegramResult(ctx, d.bot, chatID, result)
489+
if err := sendTelegramResult(ctx, d.bot, chatID, result); err != nil {
490+
return err
491+
}
492+
// Best-effort: record the delivered turn into the chat's conversation so
493+
// the agent can follow up on it. A failure here must NOT fail the run — the
494+
// message was already sent — so it is logged, not returned.
495+
if err := d.recordScheduledTurn(chatID, job, result); err != nil && d.log != nil {
496+
d.log.Error("schedule: record delivered turn into session failed", "id", job.ID, "chat", chatID, "error", err)
497+
}
498+
return nil
499+
}
500+
501+
// recordScheduledTurn appends the scheduled task and its result to the target
502+
// chat's EXISTING session, so a later interactive turn has them in context. It
503+
// deliberately does NOT create a session when none exists: a notification-only
504+
// chat (one that's never been used interactively) shouldn't accumulate an
505+
// ever-growing transcript of scheduled posts. The write is serialized with the
506+
// interactive handler — and with concurrent deliveries to the same chat —
507+
// through the per-chat mutex, so it can't interleave with a live turn's
508+
// read-modify-write of the same session.
509+
func (d telegramDeliverer) recordScheduledTurn(chatID int64, job schedule.Job, result string) error {
510+
if d.sessions == nil || result == "" {
511+
return nil
512+
}
513+
514+
mu := getChatMutex(chatID)
515+
mu.Lock()
516+
defer mu.Unlock()
517+
518+
cs, err := d.sessions.Load(chatID)
519+
if err != nil {
520+
return err
521+
}
522+
if cs == nil {
523+
return nil // no active conversation to attach to — deliver-only
524+
}
525+
526+
label := job.Name
527+
if label == "" {
528+
label = job.ID
529+
}
530+
531+
msgs := make([]llm.Message, len(cs.Messages), len(cs.Messages)+2)
532+
copy(msgs, cs.Messages)
533+
534+
// Normally append a user turn (clearly marked as scheduler-originated, not
535+
// typed live) followed by the assistant result — well-formed and ready for
536+
// the next user message. But an existing session can already END on a bare
537+
// user message (a turn cancelled before the agent replied, or a
538+
// context-injection command). Appending another user turn there would put
539+
// two user messages back-to-back, which strict providers (Anthropic) reject
540+
// on the next call. In that case fold the label into a single assistant
541+
// message so roles stay alternating. Either way the session ends on an
542+
// assistant turn. Secrets are redacted by Store.Save.
543+
if n := len(msgs); n > 0 && msgs[n-1].Role == "user" {
544+
msgs = append(msgs, llm.Message{
545+
Role: "assistant",
546+
Content: fmt.Sprintf("⏰ [scheduled task %q ran]\n%s", label, result),
547+
})
548+
} else {
549+
msgs = append(msgs,
550+
llm.Message{Role: "user", Content: fmt.Sprintf("⏰ [scheduled task %q ran]\n%s", label, job.Task)},
551+
llm.Message{Role: "assistant", Content: result},
552+
)
553+
}
554+
return d.sessions.Save(chatID, msgs)
482555
}
483556

484557
// startSchedulerForBot starts the embedded scheduler unless an external
485558
// `odek schedule daemon` already holds the lock (in which case the bot defers
486559
// to it, to avoid double-firing). It returns a stop func that releases the
487560
// lock; the scheduler goroutine itself stops when ctx is cancelled.
488-
func startSchedulerForBot(ctx context.Context, bot *telegram.Bot, resolved config.ResolvedConfig, system string, log telegram.Logger, st *schedule.Store) func() {
561+
func startSchedulerForBot(ctx context.Context, bot *telegram.Bot, resolved config.ResolvedConfig, system string, log telegram.Logger, st *schedule.Store, sessions *telegram.SessionManager) func() {
489562
if !resolved.Schedules.Enabled {
490563
log.Info("schedule: embedded scheduler disabled by config")
491564
return func() {}
@@ -508,7 +581,7 @@ func startSchedulerForBot(ctx context.Context, bot *telegram.Bot, resolved confi
508581
}
509582
sched := schedule.New(st,
510583
telegramRunner{resolved: resolved, system: system, bot: bot, mcpTools: mcpTools},
511-
telegramDeliverer{bot: bot, fallback: cliDeliverer{resolved: resolved}},
584+
telegramDeliverer{bot: bot, fallback: cliDeliverer{resolved: resolved}, sessions: sessions, log: log},
512585
schedulerOptions(resolved.Schedules, log),
513586
)
514587
scheduleUnlockRef = unlock

cmd/odek/schedule_cli_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -291,7 +291,7 @@ func TestAcquireScheduleLock_HomeError(t *testing.T) {
291291

292292
func TestStartSchedulerForBot_Disabled(t *testing.T) {
293293
stop := startSchedulerForBot(context.Background(), nil, config.ResolvedConfig{}, "system",
294-
telegram.NewFileLogger(telegram.LogInfo, ""), nil)
294+
telegram.NewFileLogger(telegram.LogInfo, ""), nil, nil)
295295
stop() // disabled → no-op stop, must not panic
296296
}
297297

@@ -306,7 +306,7 @@ func TestStartSchedulerForBot_StartAndStop(t *testing.T) {
306306
Schedules: config.ScheduleConfig{Enabled: true, MaxConcurrent: 2, Timezone: "UTC"},
307307
}
308308
ctx, cancel := context.WithCancel(context.Background())
309-
stop := startSchedulerForBot(ctx, bot, resolved, "system", telegram.NewFileLogger(telegram.LogInfo, ""), st)
309+
stop := startSchedulerForBot(ctx, bot, resolved, "system", telegram.NewFileLogger(telegram.LogInfo, ""), st, nil)
310310
cancel()
311311
stop() // drains the scheduler goroutine, cleans up MCP, releases the lock
312312
}

cmd/odek/schedule_session_test.go

Lines changed: 198 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,198 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"strings"
6+
"testing"
7+
"time"
8+
9+
"github.com/BackendStack21/odek/internal/config"
10+
"github.com/BackendStack21/odek/internal/llm"
11+
"github.com/BackendStack21/odek/internal/schedule"
12+
"github.com/BackendStack21/odek/internal/session"
13+
"github.com/BackendStack21/odek/internal/telegram"
14+
)
15+
16+
func newTestDeliverer(t *testing.T) (telegramDeliverer, *telegram.SessionManager, <-chan string) {
17+
t.Helper()
18+
t.Setenv("HOME", t.TempDir())
19+
store, err := session.NewStore()
20+
if err != nil {
21+
t.Fatalf("NewStore: %v", err)
22+
}
23+
sm := telegram.NewSessionManager(store, time.Hour)
24+
bot, recv := newRecordingTestBot(t)
25+
d := telegramDeliverer{
26+
bot: bot,
27+
fallback: cliDeliverer{resolved: config.ResolvedConfig{}},
28+
sessions: sm,
29+
log: schedule.NopLogger{},
30+
}
31+
return d, sm, recv
32+
}
33+
34+
// TestScheduleDeliver_RecordsIntoExistingSession verifies Option B: a delivered
35+
// scheduled result is appended to the target chat's existing conversation as a
36+
// labeled user turn + the assistant result, so a follow-up message sees it.
37+
func TestScheduleDeliver_RecordsIntoExistingSession(t *testing.T) {
38+
d, sm, recv := newTestDeliverer(t)
39+
chatID := int64(5551)
40+
if err := sm.Save(chatID, []llm.Message{
41+
{Role: "system", Content: "sys"},
42+
{Role: "user", Content: "hi"},
43+
{Role: "assistant", Content: "hello"},
44+
}); err != nil {
45+
t.Fatalf("seed session: %v", err)
46+
}
47+
48+
job := schedule.Job{
49+
ID: "jb-1", Name: "daily digest", Task: "summarize my day",
50+
Deliver: schedule.Delivery{Kind: schedule.DeliverTelegram, ChatID: chatID},
51+
}
52+
if err := d.Deliver(context.Background(), job, "the digest"); err != nil {
53+
t.Fatalf("Deliver: %v", err)
54+
}
55+
56+
// The message was actually sent.
57+
select {
58+
case got := <-recv:
59+
if !strings.Contains(got, "digest") {
60+
t.Errorf("sent text = %q, want it to contain 'digest'", got)
61+
}
62+
case <-time.After(2 * time.Second):
63+
t.Fatal("no message was sent to Telegram")
64+
}
65+
66+
// The conversation now ends with the scheduled exchange.
67+
cs, err := sm.Load(chatID)
68+
if err != nil || cs == nil {
69+
t.Fatalf("Load: %v, cs=%v", err, cs)
70+
}
71+
if len(cs.Messages) != 5 {
72+
t.Fatalf("expected 5 messages (3 seed + 2 scheduled), got %d", len(cs.Messages))
73+
}
74+
userTurn := cs.Messages[3]
75+
if userTurn.Role != "user" || !strings.Contains(userTurn.Content, "scheduled task") || !strings.Contains(userTurn.Content, "summarize my day") {
76+
t.Errorf("scheduled user turn wrong: %+v", userTurn)
77+
}
78+
asstTurn := cs.Messages[4]
79+
if asstTurn.Role != "assistant" || asstTurn.Content != "the digest" {
80+
t.Errorf("assistant result turn wrong: %+v", asstTurn)
81+
}
82+
}
83+
84+
// TestScheduleDeliver_PreservesAlternationAfterUserEndingSession guards the
85+
// edge where the session already ends on a bare user message (a turn cancelled
86+
// before the agent replied, or a context-injection command). Appending another
87+
// user turn would produce two consecutive user messages, which Anthropic
88+
// rejects on the next call. The write-back must fold into a single assistant
89+
// turn instead — and never produce two same-role messages in a row.
90+
func TestScheduleDeliver_PreservesAlternationAfterUserEndingSession(t *testing.T) {
91+
d, sm, recv := newTestDeliverer(t)
92+
chatID := int64(5560)
93+
if err := sm.Save(chatID, []llm.Message{
94+
{Role: "system", Content: "sys"},
95+
{Role: "user", Content: "an interrupted turn"}, // session ends on user
96+
}); err != nil {
97+
t.Fatalf("seed: %v", err)
98+
}
99+
100+
job := schedule.Job{
101+
ID: "jb-9", Name: "digest", Task: "summarize",
102+
Deliver: schedule.Delivery{Kind: schedule.DeliverTelegram, ChatID: chatID},
103+
}
104+
if err := d.Deliver(context.Background(), job, "the result"); err != nil {
105+
t.Fatalf("Deliver: %v", err)
106+
}
107+
<-recv
108+
109+
cs, err := sm.Load(chatID)
110+
if err != nil || cs == nil {
111+
t.Fatalf("Load: %v cs=%v", err, cs)
112+
}
113+
// No two consecutive same-role messages anywhere.
114+
for i := 1; i < len(cs.Messages); i++ {
115+
if cs.Messages[i].Role == cs.Messages[i-1].Role {
116+
t.Fatalf("consecutive %q messages at %d: %+v", cs.Messages[i].Role, i, cs.Messages)
117+
}
118+
}
119+
// The result was recorded, and the session ends on an assistant turn.
120+
last := cs.Messages[len(cs.Messages)-1]
121+
if last.Role != "assistant" || !strings.Contains(last.Content, "the result") {
122+
t.Errorf("last message should be the assistant result, got %+v", last)
123+
}
124+
}
125+
126+
// TestScheduleDeliver_NoSessionNotCreated verifies a notification-only chat
127+
// (never used interactively) is NOT given a session just because a schedule
128+
// posted to it — avoiding an ever-growing transcript of scheduled posts.
129+
func TestScheduleDeliver_NoSessionNotCreated(t *testing.T) {
130+
d, sm, recv := newTestDeliverer(t)
131+
chatID := int64(5552)
132+
133+
job := schedule.Job{
134+
ID: "jb-2", Name: "ping", Task: "ping",
135+
Deliver: schedule.Delivery{Kind: schedule.DeliverTelegram, ChatID: chatID},
136+
}
137+
if err := d.Deliver(context.Background(), job, "pong"); err != nil {
138+
t.Fatalf("Deliver: %v", err)
139+
}
140+
141+
select {
142+
case <-recv:
143+
case <-time.After(2 * time.Second):
144+
t.Fatal("no message was sent to Telegram")
145+
}
146+
147+
cs, err := sm.Load(chatID)
148+
if err != nil {
149+
t.Fatalf("Load: %v", err)
150+
}
151+
if cs != nil {
152+
t.Errorf("a notification-only chat should not get a session, got %d messages", len(cs.Messages))
153+
}
154+
}
155+
156+
// TestScheduleDeliver_EmptyResultNotRecorded verifies an empty result is not
157+
// appended (nothing meaningful to record).
158+
func TestScheduleDeliver_EmptyResultNotRecorded(t *testing.T) {
159+
d, sm, _ := newTestDeliverer(t)
160+
chatID := int64(5553)
161+
if err := sm.Save(chatID, []llm.Message{{Role: "user", Content: "hi"}}); err != nil {
162+
t.Fatalf("seed: %v", err)
163+
}
164+
165+
job := schedule.Job{
166+
ID: "jb-3", Name: "noop", Task: "t",
167+
Deliver: schedule.Delivery{Kind: schedule.DeliverTelegram, ChatID: chatID},
168+
}
169+
if err := d.Deliver(context.Background(), job, ""); err != nil {
170+
t.Fatalf("Deliver: %v", err)
171+
}
172+
173+
cs, err := sm.Load(chatID)
174+
if err != nil || cs == nil {
175+
t.Fatalf("Load: %v cs=%v", err, cs)
176+
}
177+
if len(cs.Messages) != 1 {
178+
t.Errorf("empty result must not append, got %d messages", len(cs.Messages))
179+
}
180+
}
181+
182+
// TestScheduleDeliver_NilSessionManagerNoPanic verifies the write-back is a
183+
// safe no-op when no SessionManager is wired (e.g. the CLI daemon path).
184+
func TestScheduleDeliver_NilSessionManagerNoPanic(t *testing.T) {
185+
bot, recv := newRecordingTestBot(t)
186+
d := telegramDeliverer{bot: bot, fallback: cliDeliverer{resolved: config.ResolvedConfig{}}}
187+
job := schedule.Job{
188+
Deliver: schedule.Delivery{Kind: schedule.DeliverTelegram, ChatID: 5554},
189+
}
190+
if err := d.Deliver(context.Background(), job, "result"); err != nil {
191+
t.Fatalf("Deliver: %v", err)
192+
}
193+
select {
194+
case <-recv:
195+
case <-time.After(2 * time.Second):
196+
t.Fatal("no message was sent")
197+
}
198+
}

cmd/odek/telegram.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -700,7 +700,7 @@ func telegramCmd(args []string) error {
700700
// process's resolved config — so no environment-inheritance problem and no
701701
// separate cron daemon. If an external `odek schedule daemon` already holds
702702
// the lock, this defers to it instead of double-firing.
703-
stopScheduler := startSchedulerForBot(ctx, bot, resolved, systemMessage, handlerLog, scheduleStore)
703+
stopScheduler := startSchedulerForBot(ctx, bot, resolved, systemMessage, handlerLog, scheduleStore, sessionManager)
704704
defer stopScheduler()
705705

706706
// 17. Process updates until the channel is closed (ctx cancelled).

0 commit comments

Comments
 (0)