diff --git a/cmd/odek/bg_telegram.go b/cmd/odek/bg_telegram.go index 8fcbb1b..615f70f 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(ex.JobID) { + 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 @@ -87,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 @@ -109,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 new file mode 100644 index 0000000..7f77aaa --- /dev/null +++ b/cmd/odek/bg_telegram_wake.go @@ -0,0 +1,253 @@ +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 ( + "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 + routed map[string]bool // job ids routed to a wake (watcher suppression) + 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, + 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) + if c.timer != nil { + c.timer.Stop() + } +} + +// 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: + 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() + + // 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 + } + + 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 +} + +// 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. 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 + n := c.pending + c.pending = 0 + stopped := c.done + c.mu.Unlock() + if n == 0 { + return + } + select { + case <-stopped: + 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 { + 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 + } +} diff --git a/cmd/odek/bg_telegram_wake_test.go b/cmd/odek/bg_telegram_wake_test.go new file mode 100644 index 0000000..5491389 --- /dev/null +++ b/cmd/odek/bg_telegram_wake_test.go @@ -0,0 +1,288 @@ +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("x") + ctl.reserve("y") + + 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 +} + +// 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()) + } +} diff --git a/cmd/odek/bg_tools.go b/cmd/odek/bg_tools.go index d25e585..9291ec2 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 fe7ffdd..74a6330 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 5f095d5..06be686 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 4005c0d..64f122d 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 83291bf..a30320e 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 2522ff0..d43d478 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 |