From a92b72b277164c50e37b578d27451e097fe2d164 Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso <4096860+jkyberneees@users.noreply.github.com> Date: Wed, 9 Sep 2026 20:49:44 +0200 Subject: [PATCH 1/3] feat(telegram): wake-on-complete for idle chats MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Background jobs started from a Telegram chat were dead letters when the chat was idle: the raw exit line reached the chat but no agent turn ever ran, so results were never summarized or acted on. With background.wake_on_complete (default true), an exit on an idle chat now starts one system-initiated wake turn via the normal chat pipeline — exits within wake_coalesce_ms coalesce into a single turn, spend is bounded per chat by max_wakes_per_hour, and busy chats keep the legacy raw exit line (their notice drain already reaches the model). Wake turns run with no user binding, so approval prompts remain available to any allowed chat member without hijacking a user. Mirrors the serve-surface wake dispatcher (bg_wake.go) on the Telegram surface. Docs updated in docs/TELEGRAM.md and docs/CONFIG.md. --- cmd/odek/bg_telegram.go | 58 ++++-- cmd/odek/bg_telegram_wake.go | 202 +++++++++++++++++++++ cmd/odek/bg_telegram_wake_test.go | 239 +++++++++++++++++++++++++ cmd/odek/bg_tools.go | 10 +- cmd/odek/bughunt_v3_fixes_test.go | 2 +- cmd/odek/bughunt_v3_serve_runs_test.go | 6 +- cmd/odek/telegram.go | 6 +- docs/CONFIG.md | 2 +- docs/TELEGRAM.md | 21 +++ 9 files changed, 525 insertions(+), 21 deletions(-) create mode 100644 cmd/odek/bg_telegram_wake.go create mode 100644 cmd/odek/bg_telegram_wake_test.go diff --git a/cmd/odek/bg_telegram.go b/cmd/odek/bg_telegram.go index 8fcbb1bd..466faf52 100644 --- a/cmd/odek/bg_telegram.go +++ b/cmd/odek/bg_telegram.go @@ -12,12 +12,15 @@ import ( "github.com/BackendStack21/odek/internal/telegram" ) -// bgChatNotifier pushes a human-readable line to the chat the moment a job -// exits (observer callback — the agent's own notice queue is untouched). -// BGStarted is deliberately silent: the chat already saw the request. +// bgChatNotifier handles background-job exit events for a chat. With wake +// enabled (wake != nil), an exit on an idle chat routes to a system-initiated +// wake turn (see bg_telegram_wake.go) and the raw push is suppressed; busy +// chats and wake-disabled setups keep the legacy push. BGStarted is +// deliberately silent: the chat already saw the request. type bgChatNotifier struct { chatID int64 bot *telegram.Bot + wake *tgWakeController } func (n *bgChatNotifier) BGStarted(j bgproc.Job) {} @@ -26,11 +29,20 @@ func (n *bgChatNotifier) BGExited(ex bgproc.Notice) { if n.bot == nil { return } - if text := formatOneNotice(ex); text != "" { - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - _, _ = n.bot.SendMessageContext(ctx, n.chatID, "📋 "+text, nil) + text := formatOneNotice(ex) + if text == "" { + return + } + // Wake path: idle chat + wake enabled + under the spend cap → dispatch + // one coalesced system-initiated turn; the model reads the completion + // notice from the loop's drain during that turn. A raw push would + // duplicate the notice in the chat without ever reaching the model. + if n.wake != nil && n.wake.reserve() { + return } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + _, _ = n.bot.SendMessageContext(ctx, n.chatID, "📋 "+text, nil) } // bgChatRuntimes tracks one background runtime per Telegram chat. Chats run @@ -40,17 +52,37 @@ var bgChatRuntimes sync.Map // chatID int64 -> *bgRuntime var bgWatchers sync.Map // chatID int64 -> bool (watcher running) +// wakeControllers tracks the per-chat wake controller so /new and shutdown +// can stop pending coalesce timers. +var wakeControllers sync.Map // chatID int64 -> *tgWakeController + +func stopWakeControllerForChat(chatID int64) { + if ctl, ok := wakeControllers.LoadAndDelete(chatID); ok { + ctl.(*tgWakeController).stop() + } +} + // bgRuntimeForChat returns the chat's long-lived background runtime, creating // it (and the exit-notification watcher) on first use. Returns nil when the -// background section is disabled. -func bgRuntimeForChat(chatID int64, resolved config.ResolvedConfig, sessID string, bot *telegram.Bot) *bgRuntime { +// background section is disabled. wakeDispatch (may be nil) starts the wake +// turn for the chat when background.wake_on_complete routes an exit to a +// system-initiated turn. +func bgRuntimeForChat(chatID int64, resolved config.ResolvedConfig, sessID string, bot *telegram.Bot, + wakeDispatch func(chatID int64, text string)) *bgRuntime { if cached, ok := bgChatRuntimes.Load(chatID); ok { rt := cached.(*bgRuntime) ensureBGWatcher(chatID, rt, bot) return rt } + var wake *tgWakeController + if wakeDispatch != nil && telegramWakeAllowed(resolved) { + wake = newTGWakeController(chatID, + time.Duration(resolved.Background.WakeCoalesceMS)*time.Millisecond, + resolved.Background.MaxWakesPerHour, 2*time.Second, wakeDispatch) + wakeControllers.Store(chatID, wake) + } rt := newBackgroundRuntime(backgroundSettingsFromResolved(resolved), sessID, "", nil, nil, - &bgChatNotifier{chatID: chatID, bot: bot}) + &bgChatNotifier{chatID: chatID, bot: bot, wake: wake}) if rt == nil { return nil } @@ -61,6 +93,11 @@ func bgRuntimeForChat(chatID int64, resolved config.ResolvedConfig, sessID strin // shutdownAllBGRuntimes kills every chat's running jobs at bot shutdown. func shutdownAllBGRuntimes() { + wakeControllers.Range(func(k, v any) bool { + v.(*tgWakeController).stop() + wakeControllers.Delete(k) + return true + }) bgChatRuntimes.Range(func(_, v any) bool { v.(*bgRuntime).Shutdown() return true @@ -74,6 +111,7 @@ func dropBGRuntimeForChat(chatID int64) { cached.(*bgRuntime).Shutdown() } bgWatchers.Delete(chatID) + stopWakeControllerForChat(chatID) } // ensureBGWatcher starts the single per-chat exit-pusher goroutine if none diff --git a/cmd/odek/bg_telegram_wake.go b/cmd/odek/bg_telegram_wake.go new file mode 100644 index 00000000..e08c1ed3 --- /dev/null +++ b/cmd/odek/bg_telegram_wake.go @@ -0,0 +1,202 @@ +package main + +// Telegram wake-on-complete (docs/CONFIG.md `background.wake_on_complete`). +// +// A background job exiting while its chat is idle is a dead letter for the +// model: the legacy bgChatNotifier pushes a raw 📋 exit line to the chat, +// but no agent turn ever runs, so the results are never summarized or acted +// on. This dispatcher closes that gap on the Telegram surface, mirroring the +// WebUI wake design in bg_wake.go: +// +// job exit (bgChatNotifier.BGExited) +// ├─ wake disabled (wake_on_complete=false, notify=off, cap=0) → legacy raw push +// ├─ chat busy (slot held by a running turn) → legacy raw push +// │ (the in-loop notice drain feeds the model on the running turn) +// ├─ spend cap reached (max_wakes_per_hour, per chat) → legacy raw push +// └─ chat idle → reserve → coalesce window → ONE system-initiated +// wake turn via handleChatMessage; the raw push is suppressed +// because the wake turn's own notice drain delivers the facts to +// the model, and a push would duplicate them in the chat. +// +// Deliberate limits: +// - The wake preamble is generic; job details arrive via the loop's +// notice drain, same as the WebUI path. +// - Wake turns run with userID 0: the TelegramApprover already treats a +// zero originating user as "no user binding", so approvals stay +// available without hijacking a user. +// - Idle detection uses the same per-chat slot the turn pipeline holds +// (pinChat), checked with a bounded wait (idleWait) so a wake never +// queues behind a long turn. + +import ( + "strconv" + "sync" + "time" + + "github.com/BackendStack21/odek/internal/config" +) + +// tgWakePreamble is the system-attributed wake turn text. Generic by design: +// the factual completion notice is injected by the loop's per-iteration +// drain at the wake turn's first iteration. +const tgWakePreamble = "[background-jobs] One or more background jobs finished while this chat was idle. Their completion notice is attached to this turn — read bg_output for the relevant job id(s) and report the results to the chat." + +// tgWakeAllowed reports whether Telegram wake turns are permitted under the +// resolved config. The same rules as the WebUI surface apply: the background +// section must be enabled, wake_on_complete on, notices injected (a wake +// would point the model at notices that are never delivered otherwise), and +// the per-hour cap positive. +func telegramWakeAllowed(resolved config.ResolvedConfig) bool { + bg := resolved.Background + return bg.Enabled && bg.WakeOnComplete && bg.Notify != "off" && bg.MaxWakesPerHour > 0 +} + +// tgWakeController coalesces background-job exits per chat and dispatches +// system-initiated wake turns. One controller lives per chat notifier. The +// reserve decision (busy chat, spend cap) is synchronous — the raw-push +// choice in BGExited depends on it; only the dispatch itself is deferred to +// the coalesce timer. +type tgWakeController struct { + chatID int64 + coalesce time.Duration + maxPerHour int + idleWait time.Duration + dispatch func(chatID int64, text string) + + mu sync.Mutex + timer *time.Timer + pending int + wakes []time.Time // wake timestamps inside the spend window + done chan struct{} +} + +func newTGWakeController(chatID int64, coalesce time.Duration, maxPerHour int, + idleWait time.Duration, dispatch func(int64, string)) *tgWakeController { + if coalesce <= 0 { + coalesce = 2 * time.Second + } + if idleWait <= 0 { + idleWait = 2 * time.Second + } + return &tgWakeController{ + chatID: chatID, + coalesce: coalesce, + maxPerHour: maxPerHour, + idleWait: idleWait, + dispatch: dispatch, + done: make(chan struct{}), + } +} + +// stop tears the controller down, cancelling any pending coalesce timer. +func (c *tgWakeController) stop() { + select { + case <-c.done: + return + default: + } + close(c.done) + c.mu.Lock() + if c.timer != nil { + c.timer.Stop() + } + c.mu.Unlock() +} + +// reserve attempts to route this exit to a wake turn. It returns true when +// a wake turn is (or will be) dispatched for it — the caller must suppress +// the legacy raw push — and false when the exit falls back to the raw push +// (controller stopped, spend cap reached, or the chat is busy). +func (c *tgWakeController) reserve() bool { + c.mu.Lock() + select { + case <-c.done: + c.mu.Unlock() + return false + default: + } + now := time.Now() + kept := c.wakes[:0] + for _, ts := range c.wakes { + if now.Sub(ts) < time.Hour { + kept = append(kept, ts) + } + } + c.wakes = kept + if len(c.wakes) >= c.maxPerHour { + c.mu.Unlock() + return false + } + c.mu.Unlock() + + // Busy check: a chat running a turn keeps the legacy push (the + // running turn's notice drain reaches the model already). Bounded + // wait so the observer goroutine never queues behind a long turn. + if !chatIsIdle(c.chatID, c.idleWait) { + return false + } + + c.mu.Lock() + defer c.mu.Unlock() + select { + case <-c.done: + return false + default: + } + c.pending++ + if c.timer == nil { + // New coalesce window: spend one wake and start the timer. + c.wakes = append(c.wakes, time.Now()) + c.timer = time.AfterFunc(c.coalesce, c.fire) + } + return true +} + +// fire runs after the coalesce window and dispatches one wake turn for all +// reserved exits. +func (c *tgWakeController) fire() { + c.mu.Lock() + c.timer = nil + n := c.pending + c.pending = 0 + stopped := c.done + c.mu.Unlock() + if n == 0 { + return + } + select { + case <-stopped: + return + default: + } + c.dispatch(c.chatID, tgWakePreamble) +} + +// wakeSpend reports how many wake turns the chat has spent in the last hour +// (diagnostics/tests). +func (c *tgWakeController) wakeSpend() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.wakes) +} + +// chatIsIdle reports whether the chat's turn slot can be taken within wait +// (i.e. no agent turn is running). The probe releases the slot immediately. +func chatIsIdle(chatID int64, wait time.Duration) bool { + acquired := make(chan struct{}, 1) + go func() { + slot := pinChat(chatID) + slot.mu.Lock() + acquired <- struct{}{} + unpinChat(chatID, slot) + }() + select { + case <-acquired: + return true + case <-time.After(wait): + return false + } +} + +// chatIDString formats a chat id for logs. +func chatIDString(id int64) string { return strconv.FormatInt(id, 10) } diff --git a/cmd/odek/bg_telegram_wake_test.go b/cmd/odek/bg_telegram_wake_test.go new file mode 100644 index 00000000..f1d1c629 --- /dev/null +++ b/cmd/odek/bg_telegram_wake_test.go @@ -0,0 +1,239 @@ +package main + +// Telegram wake-on-complete (docs/CONFIG.md `background.wake_on_complete`, +// docs/TELEGRAM.md). A background job exiting while its chat is idle is a +// dead letter today: the raw 📋 exit line reaches the chat, but the model +// never sees it — no turn runs, so the results are never summarized or +// acted on. These tests pin the wake dispatcher: +// +// idle chat → system-initiated wake turn (raw push suppressed) +// busy chat → raw push only (the in-loop notice drain covers the model) +// wake disabled → raw push only (WakeOnComplete=false / Notify=off / cap=0) +// spend control → max_wakes_per_hour enforced per chat +// coalescing → exits inside the window share one wake turn + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/BackendStack21/odek/internal/bgproc" + "github.com/BackendStack21/odek/internal/config" + "github.com/BackendStack21/odek/internal/telegram" +) + +// dispatchedRecorder is a goroutine-safe collector for wake dispatch +// callbacks and raw-push counts. +type dispatchedRecorder struct { + mu sync.Mutex + dst []string +} + +func (r *dispatchedRecorder) add(_ int64, text string) { + r.mu.Lock() + r.dst = append(r.dst, text) + r.mu.Unlock() +} + +func (r *dispatchedRecorder) len() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.dst) +} + +func (r *dispatchedRecorder) first() string { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.dst) == 0 { + return "" + } + return r.dst[0] +} + +func waitDispatched(t *testing.T, r *dispatchedRecorder, want int, within time.Duration) { + t.Helper() + deadline := time.Now().Add(within) + for r.len() < want && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if r.len() < want { + t.Fatalf("dispatched = %d, want %d within %v", r.len(), want, within) + } +} + +// newWakeTestBot points a Bot at an httptest server that counts sendMessage +// calls, mirroring the fake-server pattern used by the bot's own tests. +func newWakeTestBot(t *testing.T) (*telegram.Bot, *dispatchedRecorder) { + t.Helper() + calls := &dispatchedRecorder{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/sendMessage") { + calls.add(0, r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": map[string]any{"message_id": 1}}) + })) + t.Cleanup(srv.Close) + bot := telegram.NewBot("test:token") + bot.BaseURL = srv.URL + return bot, calls +} + +func sentCount(calls *dispatchedRecorder) int { return calls.len() } + +func wakeResolved(t *testing.T) config.ResolvedConfig { + t.Helper() + return config.ResolvedConfig{ + Background: config.BackgroundConfig{ + Enabled: true, + Notify: "observe", + WakeOnComplete: true, + MaxWakesPerHour: 60, + }, + } +} + +func TestTelegramWakeAllowed_DefaultOn(t *testing.T) { + resolved := wakeResolved(t) + if !resolved.Background.WakeOnComplete { + t.Fatal("background.wake_on_complete should default to true") + } + if !telegramWakeAllowed(resolved) { + t.Error("telegramWakeAllowed = false with default config, want true") + } +} + +func TestTelegramWakeAllowed_DisabledPaths(t *testing.T) { + resolved := wakeResolved(t) + + off := resolved + off.Background.WakeOnComplete = false + if telegramWakeAllowed(off) { + t.Error("wake should be off when wake_on_complete=false") + } + + notifyOff := resolved + notifyOff.Background.Notify = "off" + if telegramWakeAllowed(notifyOff) { + t.Error("wake should be off when notify=off (notices never reach the model)") + } + + capped := resolved + capped.Background.MaxWakesPerHour = 0 + if telegramWakeAllowed(capped) { + t.Error("wake should be off when max_wakes_per_hour=0") + } +} + +func TestBGChatNotifier_IdleChatWakeTurn(t *testing.T) { + bot, calls := newWakeTestBot(t) + rec := &dispatchedRecorder{} + ctl := newTGWakeController(910001, 10*time.Millisecond, 10, 100*time.Millisecond, rec.add) + t.Cleanup(ctl.stop) + n := &bgChatNotifier{chatID: 910001, bot: bot, wake: ctl} + + n.BGExited(bgproc.Notice{JobID: "j1", ExitCode: 0}) + + waitDispatched(t, rec, 1, 2*time.Second) + if !strings.Contains(rec.first(), "background") { + t.Errorf("wake text missing system marker: %q", rec.first()) + } + if sentCount(calls) != 0 { + t.Errorf("raw push sent %d messages on wake path, want 0", sentCount(calls)) + } +} + +func TestBGChatNotifier_BusyChatRawPush(t *testing.T) { + bot, calls := newWakeTestBot(t) + rec := &dispatchedRecorder{} + ctl := newTGWakeController(910002, 10*time.Millisecond, 10, 100*time.Millisecond, rec.add) + t.Cleanup(ctl.stop) + + // Occupy the chat slot: a turn is running. + slot := getChatMutex(910002) + slot.Lock() + t.Cleanup(slot.Unlock) + + n := &bgChatNotifier{chatID: 910002, bot: bot, wake: ctl} + n.BGExited(bgproc.Notice{JobID: "j2", ExitCode: 0}) + + deadline := time.Now().Add(300 * time.Millisecond) + for sentCount(calls) == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if rec.len() != 0 { + t.Errorf("wake dispatched on busy chat, want 0") + } + if sentCount(calls) != 1 { + t.Errorf("raw push count = %d, want 1 (busy chats keep the legacy push)", sentCount(calls)) + } +} + +func TestBGChatNotifier_WakeDisabledRawPush(t *testing.T) { + bot, calls := newWakeTestBot(t) + n := &bgChatNotifier{chatID: 910003, bot: bot} // wake nil = disabled + + n.BGExited(bgproc.Notice{JobID: "j3", ExitCode: 0}) + + deadline := time.Now().Add(300 * time.Millisecond) + for sentCount(calls) == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if sentCount(calls) != 1 { + t.Errorf("raw push count = %d, want 1 when wake disabled", sentCount(calls)) + } +} + +func TestTGWakeController_MaxWakesPerHour(t *testing.T) { + bot, calls := newWakeTestBot(t) + rec := &dispatchedRecorder{} + ctl := newTGWakeController(910004, 5*time.Millisecond, 2, 100*time.Millisecond, rec.add) + t.Cleanup(ctl.stop) + n := &bgChatNotifier{chatID: 910004, bot: bot, wake: ctl} + + // First two exits coalesce into wakes; the third must hit the cap and + // fall back to the raw push. + n.BGExited(bgproc.Notice{JobID: "a", ExitCode: 0}) + waitDispatched(t, rec, 1, 2*time.Second) + time.Sleep(20 * time.Millisecond) // out of the coalesce window + + n.BGExited(bgproc.Notice{JobID: "b", ExitCode: 0}) + waitDispatched(t, rec, 2, 2*time.Second) + time.Sleep(20 * time.Millisecond) + + n.BGExited(bgproc.Notice{JobID: "c", ExitCode: 0}) + deadline := time.Now().Add(300 * time.Millisecond) + for sentCount(calls) == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if rec.len() != 2 { + t.Errorf("wake dispatches = %d, want 2 (cap is 2/h)", rec.len()) + } + if sentCount(calls) != 1 { + t.Errorf("raw push after cap = %d, want 1", sentCount(calls)) + } +} + +func TestTGWakeController_CoalesceWindow(t *testing.T) { + rec := &dispatchedRecorder{} + ctl := newTGWakeController(910005, 80*time.Millisecond, 10, 100*time.Millisecond, rec.add) + t.Cleanup(ctl.stop) + + ctl.reserve() + ctl.reserve() + + waitDispatched(t, rec, 1, 2*time.Second) + time.Sleep(150 * time.Millisecond) // let any second timer fire + if rec.len() != 1 { + t.Errorf("coalesced wake dispatches = %d, want 1", rec.len()) + } +} + +func TestBGChatNotifier_NilBotNoop(t *testing.T) { + n := &bgChatNotifier{chatID: 910006, bot: nil} + n.BGExited(bgproc.Notice{JobID: "z", ExitCode: 0}) // must not panic +} diff --git a/cmd/odek/bg_tools.go b/cmd/odek/bg_tools.go index d25e5851..9291ec2a 100644 --- a/cmd/odek/bg_tools.go +++ b/cmd/odek/bg_tools.go @@ -61,11 +61,11 @@ type bgRuntime struct { func backgroundSettingsFromResolved(resolved config.ResolvedConfig) BackgroundSettings { b := resolved.Background return BackgroundSettings{ - Enabled: b.Enabled, - MaxJobs: b.MaxJobs, - MaxOutputBytes: b.MaxOutputBytes, - MaxTimeoutSeconds: b.MaxTimeoutSeconds, - Notify: b.Notify == "observe", + Enabled: b.Enabled, + MaxJobs: b.MaxJobs, + MaxOutputBytes: b.MaxOutputBytes, + MaxTimeoutSeconds: b.MaxTimeoutSeconds, + Notify: b.Notify == "observe", StripChildSecretEnv: resolved.Dangerous.StripSecretsEnvChildrenEnabled(), } } diff --git a/cmd/odek/bughunt_v3_fixes_test.go b/cmd/odek/bughunt_v3_fixes_test.go index fe7ffdd5..74a63301 100644 --- a/cmd/odek/bughunt_v3_fixes_test.go +++ b/cmd/odek/bughunt_v3_fixes_test.go @@ -115,7 +115,7 @@ func TestMultiGrep_ReportsUnopenableFile(t *testing.T) { result := callJSON(t, tool, args) var r struct { Results []struct { - Pattern string `json:"pattern"` + Pattern string `json:"pattern"` Matches []struct { Path string `json:"path"` } `json:"matches"` diff --git a/cmd/odek/bughunt_v3_serve_runs_test.go b/cmd/odek/bughunt_v3_serve_runs_test.go index 5f095d54..06be6867 100644 --- a/cmd/odek/bughunt_v3_serve_runs_test.go +++ b/cmd/odek/bughunt_v3_serve_runs_test.go @@ -7,9 +7,9 @@ import ( func newTerminalTestRun() *serveRun { r := &serveRun{ - ID: "run-terminal-guard", - Status: "running", - pending: map[string]*approvalRequest{}, + ID: "run-terminal-guard", + Status: "running", + pending: map[string]*approvalRequest{}, } r.cond = sync.NewCond(&r.mu) return r diff --git a/cmd/odek/telegram.go b/cmd/odek/telegram.go index 4005c0df..64f122d5 100644 --- a/cmd/odek/telegram.go +++ b/cmd/odek/telegram.go @@ -1479,7 +1479,11 @@ func handleChatMessage( } // Build the agent with Telegram approver. - bgRT := bgRuntimeForChat(chatID, resolved, sess.ID, bot) + bgRT := bgRuntimeForChat(chatID, resolved, sess.ID, bot, + func(wakeChatID int64, wakeText string) { + go handleChatMessage(wakeChatID, 0, 0, wakeText, bot, handler, + sessionManager, resolved, systemMessage, log) + }) tools := builtinTools(resolved.Dangerous, nil, approver, resolved.MaxConcurrency, resolved.APIKey, toolConfigFromResolved(resolved), sessionManager.Store) tools = appendBackgroundTools(tools, bgRT) diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 83291bf9..a30320e5 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -1202,7 +1202,7 @@ ends — there is no detach mode in v1. | `max_timeout_seconds` | `0` | Cap for explicit `timeout_seconds` on `bg_start`; `0` = uncapped (session lifetime is the bound). | | `notify` | `"observe"` | `"observe"` injects a drained completion summary at the agent's next iteration; `"off"` requires polling with `bg_status`. | | `on_session_end` | `"kill"` | Job fate at session end. Only `"kill"` is supported; there is no detach. | -| `wake_on_complete` | `true` | Serve surface: when a job finishes while its session is idle **and** a WebUI connection is attached, start a system-initiated turn so the model reads `bg_output` and reports unprompted. Busy sessions rely on the normal notice drain; `notify: "off"` forces this off (a wake would point at notices that are never delivered). | +| `wake_on_complete` | `true` | Serve surface: when a job finishes while its session is idle **and** a WebUI connection is attached, start a system-initiated turn so the model reads `bg_output` and reports unprompted. Telegram surface: the same setting wakes idle chats with a system-initiated turn (exits within `wake_coalesce_ms` coalesce into one; busy chats keep the raw exit line). Wake turns are system messages. Forced off by `notify: "off"` (a wake would point at notices that are never delivered); bounded per chat/session by `max_wakes_per_hour`. | | `wake_coalesce_ms` | `2000` | Window in which jobs finishing together share one wake turn. Global-only: project configs may not set it. | | `max_wakes_per_hour` | `30` | Per-session ceiling on system-initiated wake turns (spend control). `0` disables waking; values above `240` clamp to `240` regardless of config source. Project configs may only lower an operator-set value. | diff --git a/docs/TELEGRAM.md b/docs/TELEGRAM.md index 2522ff0f..d43d4783 100644 --- a/docs/TELEGRAM.md +++ b/docs/TELEGRAM.md @@ -252,6 +252,27 @@ defense-in-depth. | `/stats` | Show session statistics (turn count, model used, etc.) | | `/jobs` | List background jobs for this chat | | `/stop` | Cancel a running agent task | + +### Wake-on-complete (background jobs) + +When a background job started from a chat finishes while the chat is idle, +the bot starts a **system-initiated wake turn**: the model reads the job's +output (`bg_output`) and reports the results to the chat unprompted. Wake +turns are marked as system messages — they never appear as if the user had +sent something. + +Routing per job exit: + +| State | Behavior | +|---|---| +| Chat idle, wake enabled | One coalesced wake turn (exits within `wake_coalesce_ms` share one turn); the raw exit line is suppressed — the notice is delivered to the model inside the wake turn | +| Chat busy (a turn is running) | Legacy raw 📋 exit line only — the running turn's notice drain already reaches the model | +| Wake disabled / spend cap hit | Legacy raw 📋 exit line only | + +Wake is controlled by the shared `background.wake_on_complete` setting +(default `true`), forced off when `background.notify: "off"`, and bounded by +`background.max_wakes_per_hour` per chat (spend control). See +[docs/CONFIG.md](CONFIG.md) for the `background` section. | `/mode` | Show current agent modes (interaction_mode, tool_progress, sandbox) | | `/restart` | Gracefully restart the bot process. Restricted to operator chats/users and rate-limited to once per 60 seconds. | | `/plan ` | Create a new plan from a natural language description | From 862c57a9f68cbe2fc399c7e2c9253ad4963149f3 Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso <4096860+jkyberneees@users.noreply.github.com> Date: Wed, 9 Sep 2026 21:52:01 +0200 Subject: [PATCH 2/3] =?UTF-8?q?fix(telegram):=20wake=20review=20findings?= =?UTF-8?q?=20=E2=80=94=20watcher=20duplication,=20TOCTOU,=20stop=20race?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review rounds 1-2 findings, all fixed: - F1 (P1): the per-chat exit-watcher re-pushed raw completion lines for jobs a wake turn already covered. The controller now records routed job ids and the watcher skips them. - F2 (P2): a user message taking the chat slot between the idle probe and the coalesce timer queued a stale wake behind the user's turn. fire() now re-checks idleness, drops the wake, and refunds the spend (the user's turn notice drain delivers the data). - F3 (P2): routed-id recording raced the watcher's 10s tick — the id is now recorded before the busy probe and rolled back if reserve fails. - F4 (P2): concurrent stop() callers could double-close done. The close now happens under the controller mutex. - F5: routed map bounded (reset at 1024 entries; suppression only matters while the watcher is live). Regression tests added for F1/F2 semantics; full Telegram/bg scope green with -race -count=1. --- cmd/odek/bg_telegram.go | 20 +++++++-- cmd/odek/bg_telegram_wake.go | 73 +++++++++++++++++++++++++++---- cmd/odek/bg_telegram_wake_test.go | 53 +++++++++++++++++++++- 3 files changed, 132 insertions(+), 14 deletions(-) diff --git a/cmd/odek/bg_telegram.go b/cmd/odek/bg_telegram.go index 466faf52..615f70fd 100644 --- a/cmd/odek/bg_telegram.go +++ b/cmd/odek/bg_telegram.go @@ -37,7 +37,7 @@ func (n *bgChatNotifier) BGExited(ex bgproc.Notice) { // one coalesced system-initiated turn; the model reads the completion // notice from the loop's drain during that turn. A raw push would // duplicate the notice in the chat without ever reaching the model. - if n.wake != nil && n.wake.reserve() { + if n.wake != nil && n.wake.reserve(ex.JobID) { return } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) @@ -125,15 +125,23 @@ func ensureBGWatcher(chatID int64, rt *bgRuntime, bot *telegram.Bot) { } go func() { defer bgWatchers.Delete(chatID) - watchBGNotices(chatID, rt, bot) + watchBGNotices(chatID, rt, bot, wakeControllerForChat(chatID)) }() } +// wakeControllerForChat returns the chat's wake controller, or nil. +func wakeControllerForChat(chatID int64) *tgWakeController { + if ctl, ok := wakeControllers.Load(chatID); ok { + return ctl.(*tgWakeController) + } + return nil +} + // watchBGNotices pushes a human-readable line to the chat when a job exits, // so the user hears about completions between messages. The agent still gets // the full observe-phase notice from its own drain — this watcher never // touches the agent's notice queue. It stops after ~30s with nothing running. -func watchBGNotices(chatID int64, rt *bgRuntime, bot *telegram.Bot) { +func watchBGNotices(chatID int64, rt *bgRuntime, bot *telegram.Bot, wake *tgWakeController) { ticker := time.NewTicker(10 * time.Second) defer ticker.Stop() // Snapshot the exit states the watcher has already announced so a job is @@ -147,6 +155,12 @@ func watchBGNotices(chatID int64, rt *bgRuntime, bot *telegram.Bot) { continue } announced[j.ID] = true + // Jobs routed to a wake turn are covered: the wake turn's + // notice drain delivers the facts to the model and the wake + // reply reaches the chat — a raw line here would duplicate it. + if wake != nil && wake.wakeRouted(j.ID) { + continue + } end := j.EndedAt if end.IsZero() { end = time.Now() diff --git a/cmd/odek/bg_telegram_wake.go b/cmd/odek/bg_telegram_wake.go index e08c1ed3..05ea6a8c 100644 --- a/cmd/odek/bg_telegram_wake.go +++ b/cmd/odek/bg_telegram_wake.go @@ -66,7 +66,8 @@ type tgWakeController struct { mu sync.Mutex timer *time.Timer pending int - wakes []time.Time // wake timestamps inside the spend window + wakes []time.Time // wake timestamps inside the spend window + routed map[string]bool // job ids routed to a wake (watcher suppression) done chan struct{} } @@ -84,30 +85,36 @@ func newTGWakeController(chatID int64, coalesce time.Duration, maxPerHour int, maxPerHour: maxPerHour, idleWait: idleWait, dispatch: dispatch, + routed: map[string]bool{}, done: make(chan struct{}), } } // stop tears the controller down, cancelling any pending coalesce timer. +// Safe under concurrent callers (dropBGRuntimeForChat can race +// shutdownAllBGRuntimes): the close happens under the mutex, so exactly +// one caller closes. func (c *tgWakeController) stop() { + c.mu.Lock() + defer c.mu.Unlock() select { case <-c.done: return default: } close(c.done) - c.mu.Lock() if c.timer != nil { c.timer.Stop() } - c.mu.Unlock() } -// reserve attempts to route this exit to a wake turn. It returns true when -// a wake turn is (or will be) dispatched for it — the caller must suppress -// the legacy raw push — and false when the exit falls back to the raw push -// (controller stopped, spend cap reached, or the chat is busy). -func (c *tgWakeController) reserve() bool { +// reserve attempts to route this exit to a wake turn. jobID is recorded so +// the chat's exit-watcher can suppress its raw push for jobs the wake turn +// already covers (see watchBGNotices). It returns true when a wake turn is +// (or will be) dispatched for it — the caller must suppress the legacy raw +// push — and false when the exit falls back to the raw push (controller +// stopped, spend cap reached, or the chat is busy). +func (c *tgWakeController) reserve(jobID string) bool { c.mu.Lock() select { case <-c.done: @@ -129,10 +136,26 @@ func (c *tgWakeController) reserve() bool { } c.mu.Unlock() + // Mark the job routed BEFORE the busy probe: the exit-watcher polls on + // a 10s tick and must never see wakeRouted=false for a job that is + // about to be covered by a wake turn (otherwise it pushes the raw line + // and the wake duplicates it). Rolled back below if reserve fails. + if jobID != "" { + c.mu.Lock() + c.pruneRoutedLocked() + c.routed[jobID] = true + c.mu.Unlock() + } + // Busy check: a chat running a turn keeps the legacy push (the // running turn's notice drain reaches the model already). Bounded // wait so the observer goroutine never queues behind a long turn. if !chatIsIdle(c.chatID, c.idleWait) { + if jobID != "" { + c.mu.Lock() + delete(c.routed, jobID) + c.mu.Unlock() + } return false } @@ -152,8 +175,21 @@ func (c *tgWakeController) reserve() bool { return true } +// pruneRoutedLocked bounds the routed map: watcher suppression only +// matters while the watcher is live (~30s window), so on overflow the map +// is simply reset — long-since-announced jobs never need suppression again. +// Caller holds c.mu. +func (c *tgWakeController) pruneRoutedLocked() { + if len(c.routed) >= 1024 { + c.routed = map[string]bool{} + } +} + // fire runs after the coalesce window and dispatches one wake turn for all -// reserved exits. +// reserved exits. If the chat became busy between reserve and fire (a user +// message took the slot), the wake is dropped and the spend refunded: the +// user's queued turn drains the completion notices at its first iteration, +// so a queued stale wake would only duplicate it. func (c *tgWakeController) fire() { c.mu.Lock() c.timer = nil @@ -169,9 +205,28 @@ func (c *tgWakeController) fire() { return default: } + if !chatIsIdle(c.chatID, c.idleWait) { + c.mu.Lock() + if len(c.wakes) > 0 { + c.wakes = c.wakes[:len(c.wakes)-1] // refund the unused wake + } + c.mu.Unlock() + return + } c.dispatch(c.chatID, tgWakePreamble) } +// wakeRouted reports whether the job's exit was routed to a wake turn, so +// the exit-watcher must not push a raw line for it. +func (c *tgWakeController) wakeRouted(jobID string) bool { + if jobID == "" { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + return c.routed[jobID] +} + // wakeSpend reports how many wake turns the chat has spent in the last hour // (diagnostics/tests). func (c *tgWakeController) wakeSpend() int { diff --git a/cmd/odek/bg_telegram_wake_test.go b/cmd/odek/bg_telegram_wake_test.go index f1d1c629..5491389b 100644 --- a/cmd/odek/bg_telegram_wake_test.go +++ b/cmd/odek/bg_telegram_wake_test.go @@ -223,8 +223,8 @@ func TestTGWakeController_CoalesceWindow(t *testing.T) { ctl := newTGWakeController(910005, 80*time.Millisecond, 10, 100*time.Millisecond, rec.add) t.Cleanup(ctl.stop) - ctl.reserve() - ctl.reserve() + ctl.reserve("x") + ctl.reserve("y") waitDispatched(t, rec, 1, 2*time.Second) time.Sleep(150 * time.Millisecond) // let any second timer fire @@ -237,3 +237,52 @@ func TestBGChatNotifier_NilBotNoop(t *testing.T) { n := &bgChatNotifier{chatID: 910006, bot: nil} n.BGExited(bgproc.Notice{JobID: "z", ExitCode: 0}) // must not panic } + +// F1 regression: jobs routed to a wake turn must be invisible to the +// exit-watcher, or the watcher re-pushes the raw line the wake turn already +// covered (the duplication the suppression exists to prevent). +func TestTGWakeController_WakeRoutedSuppressesWatcher(t *testing.T) { + ctl := newTGWakeController(910007, 5*time.Millisecond, 10, 100*time.Millisecond, + func(int64, string) {}) + t.Cleanup(ctl.stop) + if ctl.wakeRouted("jw") { + t.Fatal("wakeRouted = true before any reserve") + } + if !ctl.reserve("jw") { + t.Fatal("reserve = false, want true (idle chat under cap)") + } + if !ctl.wakeRouted("jw") { + t.Error("wakeRouted(jw) = false after reserve, want true") + } + if ctl.wakeRouted("other") { + t.Error("wakeRouted(other) = true for an unrouted job") + } +} + +// F2 regression: when the chat turns busy between reserve and fire, the wake +// must be dropped (not queued behind the user's turn) and the spend refunded. +func TestTGWakeController_BusyAtFireDropsAndRefunds(t *testing.T) { + rec := &dispatchedRecorder{} + ctl := newTGWakeController(910008, 10*time.Millisecond, 1, 100*time.Millisecond, rec.add) + t.Cleanup(ctl.stop) + if !ctl.reserve("jb") { + t.Fatal("reserve = false, want true") + } + // Occupy the slot before the coalesce timer fires. + slot := getChatMutex(910008) + slot.Lock() + t.Cleanup(slot.Unlock) + + waitDispatched(t, rec, 0, 0) + deadline := time.Now().Add(500 * time.Millisecond) + for ctl.wakeSpend() != 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + time.Sleep(50 * time.Millisecond) + if rec.len() != 0 { + t.Errorf("wake dispatched on busy-at-fire chat, want 0") + } + if ctl.wakeSpend() != 0 { + t.Errorf("wakeSpend = %d after refund, want 0", ctl.wakeSpend()) + } +} From 62c931505a659b999b91521f8cd032dd895dca65 Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso <4096860+jkyberneees@users.noreply.github.com> Date: Wed, 9 Sep 2026 21:57:31 +0200 Subject: [PATCH 3/3] chore: drop unused chatIDString helper (lint) --- cmd/odek/bg_telegram_wake.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/cmd/odek/bg_telegram_wake.go b/cmd/odek/bg_telegram_wake.go index 05ea6a8c..7f77aaa1 100644 --- a/cmd/odek/bg_telegram_wake.go +++ b/cmd/odek/bg_telegram_wake.go @@ -29,7 +29,6 @@ package main // queues behind a long turn. import ( - "strconv" "sync" "time" @@ -252,6 +251,3 @@ func chatIsIdle(chatID int64, wait time.Duration) bool { return false } } - -// chatIDString formats a chat id for logs. -func chatIDString(id int64) string { return strconv.FormatInt(id, 10) }