From 34614aa2d6bb39c3ed66cb708a1ffb9de1d86fd8 Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso Date: Wed, 2 Sep 2026 20:16:14 +0200 Subject: [PATCH] feat(tui): render server-initiated wake turns from bg-job completion odek >= v1.40 wakes idle sessions when a background job finishes: the session frame arrives stamped system_initiated and the model's report streams like any turn. Cards previously opened only on the local send path, so every streamed event of a wake turn dropped silently. - client: decode system_initiated on the session frame - tui: openWakeTurn opens the card from the wire (busy, relayout, status 'waking for bg job'); suppressed during operator turns - transcript: systemWake cards render 'odek - wake', never as a user message; plain mode prints a [wake] line - bg_wake frames surface a transient note; bg_job frames kick an immediate jobs snapshot (kickJobsFetch), watcher stays fallback - docs: README + AGENTS updated for push frames and wake rendering --- AGENTS.md | 21 +++- README.md | 8 +- internal/client/client.go | 4 + internal/client/wake_decode_test.go | 47 ++++++++ internal/tui/events.go | 40 +++++++ internal/tui/jobs_tab.go | 11 ++ internal/tui/model.go | 23 ++-- internal/tui/plain.go | 5 + internal/tui/view.go | 6 + internal/tui/wake_turn_test.go | 165 ++++++++++++++++++++++++++++ 10 files changed, 312 insertions(+), 18 deletions(-) create mode 100644 internal/client/wake_decode_test.go create mode 100644 internal/tui/wake_turn_test.go diff --git a/AGENTS.md b/AGENTS.md index 7631e96..74303cc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,12 +117,21 @@ feat(tui): compact tool steps with Ctrl+E details toggle text, MCP args, raw config JSON — everything through `sanitize()`), `esc`/`q` folds back, `p` promotes in place. Tab switches reset it (`switchDrawerTab`); keep that reset when adding new open paths. -- The jobs tab pairs with a REST lifecycle watcher (`jobs_tab.go`): odek - pushes no WS frames for background jobs and its completion notice never - leaves the LLM payload, so the TUI polls `/api/jobs` (10s in background, - 3s while the tab is visible) and diffs status transitions into transient - notes. Generation counters (`jobsSeq`/`jobsWatchSeq`) drop stale ticks — - keep both chains generation-guarded when touching the cadence. +- The jobs tab pairs with a REST lifecycle watcher (`jobs_tab.go`): the + TUI polls `/api/jobs` (10s in background, 3s while the tab is visible) + and diffs status transitions into transient notes. odek ≥ v1.40 also + pushes `bg_job` frames on start/exit — `handleEvent` routes them through + `kickJobsFetch()` for an immediate snapshot, watcher tick as fallback; + `bg_wake` frames become transient notes. Generation counters + (`jobsSeq`/`jobsWatchSeq`) drop stale ticks — keep both chains + generation-guarded when touching the cadence. +- Server-initiated wake turns (odek ≥ v1.40, `system_initiated` on the + session frame) open a streaming card from the wire (`openWakeTurn` in + `events.go`): without it every streamed event drops, because cards only + opened on the local send path. The card carries the `systemWake` marker + (renders `⬡ odek · wake`); wake turns are never rendered as user + messages, and a wake frame arriving during an operator turn opens + nothing. ## Workflow rules for agents diff --git a/README.md b/README.md index 51ae9e1..0116369 100644 --- a/README.md +++ b/README.md @@ -290,6 +290,10 @@ own front-end settings are separate; see [Configuration](#configuration). desktop notifications. Fires only on terminal states — never per token. - **Sandbox aware** — the header shows `🛡 sandboxed` or `⚠ host access`; pass `--sandbox` to run tool calls inside odek's Docker isolation. +- **Wake turns** — when a background job finishes while the session is + idle (odek ≥ v1.40), the engine wakes the model on its own; bodek opens + the turn from the wire, marks the card `⬡ odek · wake`, and streams the + model's report like any other turn — never rendered as a user message. --- @@ -407,7 +411,9 @@ shared grammar: every 3s; a background watcher surfaces starts as transcript notes and exits as **alert-tier notes naming the command** — with a bell / desktop notification (same gates as turns) so a finished job is never missed, - even with the tab closed. `⏎` opens the job's output viewer (`f` + even with the tab closed. odek ≥ v1.40 also pushes `bg_job` frames: the + snapshot refreshes the moment a job starts or exits, the watcher tick + stays as the fallback. `⏎` opens the job's output viewer (`f` pages further output), `s` stops a running job (two-step, same gate as `/stop`). Jobs are session-scoped: other sessions' jobs never appear. - **Events** — the `odek.event/v1` ring: `f` filter to this session, `x` diff --git a/internal/client/client.go b/internal/client/client.go index 28f20bf..daf7606 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -39,6 +39,10 @@ type Event struct { Model string `json:"model"` Sandbox bool `json:"sandbox"` + // session — true when odek itself started the turn (wake-on-complete, + // ≥ v1.40); absent on operator turns. + SystemInitiated bool `json:"system_initiated,omitempty"` + // done — token economics for the turn and the session. ContextTokens is // cumulative prompt tokens across all LLM calls of the run (the live // window fill is the delta between consecutive reports); the Session* diff --git a/internal/client/wake_decode_test.go b/internal/client/wake_decode_test.go new file mode 100644 index 0000000..585d03a --- /dev/null +++ b/internal/client/wake_decode_test.go @@ -0,0 +1,47 @@ +package client + +import ( + "encoding/json" + "testing" +) + +// odek ≥ v1.40 stamps the session frame with system_initiated when the turn +// was started by the wake-on-complete dispatcher; the field is absent on +// operator turns. Event must decode both shapes. +func TestEventDecodeSystemInitiated(t *testing.T) { + var wake Event + if err := json.Unmarshal([]byte(`{"type":"session","session_id":"s1","system_initiated":true}`), &wake); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !wake.SystemInitiated { + t.Error("system_initiated=true did not decode onto Event.SystemInitiated") + } + + var op Event + if err := json.Unmarshal([]byte(`{"type":"session","session_id":"s1"}`), &op); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if op.SystemInitiated { + t.Error("absent system_initiated must decode false (operator turns)") + } +} + +// Wake-surface frames (bg_wake / bg_job) must decode far enough to dispatch +// on Type — the TUI routes them and refetches anything richer over REST. +func TestEventDecodeWakeFrames(t *testing.T) { + for _, tc := range []struct { + wire string + want string + }{ + {`{"type":"bg_wake","session_id":"s1","t":1756830000000}`, "bg_wake"}, + {`{"type":"bg_job","job_id":"bg_ab12","session_id":"s1","status":"exited"}`, "bg_job"}, + } { + var ev Event + if err := json.Unmarshal([]byte(tc.wire), &ev); err != nil { + t.Fatalf("unmarshal %s: %v", tc.wire, err) + } + if ev.Type != tc.want { + t.Errorf("Type = %q, want %q", ev.Type, tc.want) + } + } +} diff --git a/internal/tui/events.go b/internal/tui/events.go index 80b6947..d862a20 100644 --- a/internal/tui/events.go +++ b/internal/tui/events.go @@ -49,6 +49,9 @@ func (m *Model) handleEvent(ev client.Event) (tea.Model, tea.Cmd) { m.resolveMaxContext() } m.sandbox = ev.Sandbox + if ev.SystemInitiated && m.cur() < 0 && !m.busy { + m.openWakeTurn() // server-started turn: open the card from the wire + } case "thinking", "thinking_delta": // Bulk reasoning and live streamed fragments (streaming on) share one @@ -346,6 +349,22 @@ func (m *Model) handleEvent(ev client.Event) (tea.Model, tea.Cmd) { m.addTransientNote("stop declined · sub-agent already finished") } + case "bg_wake": + // odek ≥ v1.40 enqueued a wake turn for a finished background job: + // the stamped session frame that follows opens the card; this note + // gives the operator the context for the unprompted activity. + m.addTransientNote("background job finished · agent waking") + + case "bg_job": + // Push notification of a job start/exit (≥ v1.40): refresh the + // snapshot now instead of waiting for the next watcher tick. The + // REST watcher stays as the fallback; applyJobs diffs the + // transition into notes/attention as before. + if cmd := m.kickJobsFetch(); cmd != nil { + m.refresh() + return m, tea.Batch(listen(m.events), m.noticeSweep(), cmd) + } + case client.EventDisconnected: m.disconn = true m.busy = false @@ -406,6 +425,27 @@ func (m *Model) handleEvent(ev client.Event) (tea.Model, tea.Cmd) { return m, tea.Batch(listen(m.events), m.noticeSweep(), m.approvalSweep(), m.sendQueued(), m.planFollowup(), attn) } +// openWakeTurn opens a streaming assistant card for a server-initiated turn +// (background-job wake): without it, every streaming event would find no +// open card and drop — cards historically opened only on the local send +// path. The card carries the systemWake marker so the unprompted turn is +// never mistaken for an operator exchange. +func (m *Model) openWakeTurn() { + m.msgs = append(m.msgs, message{role: roleAsst, streaming: true, systemWake: true}) + m.curIdx = len(m.msgs) - 1 + m.busy = true + m.cancelAck = false // a wake run's errors are real errors again + m.skillSuggest = nil // the suggestion's window closed with the last turn + m.status = "waking for bg job" + m.runStart = time.Now() + if m.sessionStart.IsZero() { + m.sessionStart = m.runStart + } + m.relayout() // the busy status line claims a row above the input + m.refresh() + m.vp.GotoBottom() // new activity: show it even when reading scrollback +} + // stepGlyphs returns up to 4 deduped tool glyphs for a turn's steps, in // first-seen order, for the per-turn stat line. func stepGlyphs(steps []step) []string { diff --git a/internal/tui/jobs_tab.go b/internal/tui/jobs_tab.go index dac3a1c..0b7b794 100644 --- a/internal/tui/jobs_tab.go +++ b/internal/tui/jobs_tab.go @@ -92,6 +92,17 @@ func (m *Model) fetchJobs() tea.Cmd { } } +// kickJobsFetch returns an immediate snapshot fetch for a bg_job push +// frame (odek ≥ v1.40). The REST watcher stays as the fallback; nil when +// the surface is unavailable or there is no live session to fetch with +// (fetchJobs no-ops on both). +func (m *Model) kickJobsFetch() tea.Cmd { + if m.jobsOff { + return nil + } + return m.fetchJobs() +} + // applyJobs stores a snapshot; watcher diffs become alert-tier notes — a // finished job is actionable, not housekeeping — and terminal transitions // fire the attention layer (bell / OSC 9). The first successful snapshot diff --git a/internal/tui/model.go b/internal/tui/model.go index f91455c..fe57940 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -82,17 +82,18 @@ type turnStats struct { // message is one entry in the transcript. type message struct { - role role - content string // raw text/markdown - rendered string // cached glamour render (assistant, finalized) - thinking string // captured reasoning for this turn (finalized) - steps []step - items []turnItem // chronological timeline of reasoning blocks and tool calls - streaming bool - stats *turnStats // finalized-turn telemetry; nil while streaming / for history - raw bool // content is pre-styled; render verbatim, never re-render - sentAt time.Time // user turns: when the prompt was submitted (drives the head's age) - collapsed bool // turn card folded to its head + summary line (c) + role role + content string // raw text/markdown + rendered string // cached glamour render (assistant, finalized) + thinking string // captured reasoning for this turn (finalized) + steps []step + items []turnItem // chronological timeline of reasoning blocks and tool calls + streaming bool + stats *turnStats // finalized-turn telemetry; nil while streaming / for history + raw bool // content is pre-styled; render verbatim, never re-render + sentAt time.Time // user turns: when the prompt was submitted (drives the head's age) + collapsed bool // turn card folded to its head + summary line (c) + systemWake bool // server-initiated turn (background-job wake): marker on the card } // Options carries startup display info into the model. diff --git a/internal/tui/plain.go b/internal/tui/plain.go index 41f2263..89d3012 100644 --- a/internal/tui/plain.go +++ b/internal/tui/plain.go @@ -89,6 +89,11 @@ func (m *Model) plainEventLines(ev client.Event) []string { } return []string{plainClip("· subagent · " + line + eventTail(ev))} + case "session": + if ev.SystemInitiated { + return []string{"[wake] background job finished — agent turning"} + } + case "done": var lines []string if reply := m.lastReply(); reply != "" { diff --git a/internal/tui/view.go b/internal/tui/view.go index abf14f7..72a4243 100644 --- a/internal/tui/view.go +++ b/internal/tui/view.go @@ -500,6 +500,12 @@ func (m *Model) renderMessage(msg message, msgIdx, lineOffset int) (string, []st // segments shed in priority order under width pressure, exactly like // the old foot line. label := th.asstLabel.Render("⬡ odek") + if msg.systemWake { + // Server-initiated wake (background-job completion): the marker + // is the card's identity, so it sits left of the telemetry and + // sheds last. + label += th.asstLabel.Render(" · wake") + } if msg.stats != nil { limit := m.vp.Width - lipgloss.Width(label) - 4 if s := m.joinStatSegs(m.statSegments(*msg.stats), limit); s != "" { diff --git a/internal/tui/wake_turn_test.go b/internal/tui/wake_turn_test.go new file mode 100644 index 0000000..36ecddf --- /dev/null +++ b/internal/tui/wake_turn_test.go @@ -0,0 +1,165 @@ +package tui + +import ( + "strings" + "testing" + + "github.com/BackendStack21/bodek/internal/client" +) + +// odek ≥ v1.40 wake-on-complete: the session frame arrives stamped +// system_initiated for a server-started turn. The TUI must open a streaming +// card from the wire — streaming events would otherwise find no open card +// and drop (cards only opened on the local send path before). +func TestWakeTurnOpensCardFromWire(t *testing.T) { + m := newTestModel() + m.handleEvent(client.Event{Type: "session", SessionID: "s1", SystemInitiated: true}) + + i := m.cur() + if i < 0 { + t.Fatal("wake session frame did not open a streaming card") + } + msg := m.msgs[i] + if !msg.streaming { + t.Error("wake card is not streaming") + } + if !msg.systemWake { + t.Error("wake card not marked systemWake (marker would be lost)") + } + if msg.role != roleAsst { + t.Errorf("wake card role = %v, want assistant (never renders as a user message)", msg.role) + } + if !m.busy { + t.Error("wake turn did not set busy") + } + if !strings.Contains(m.status, "wak") { + t.Errorf("status = %q, want a waking status", m.status) + } +} + +// Operator turns (session frame without system_initiated) keep the old +// behaviour: no card until the local send path opens one. +func TestOperatorSessionFrameOpensNothing(t *testing.T) { + m := newTestModel() + m.handleEvent(client.Event{Type: "session", SessionID: "s1"}) + if m.cur() >= 0 { + t.Error("operator session frame opened a card") + } + if m.busy { + t.Error("operator session frame set busy") + } +} + +// A wake frame racing a live operator turn must not open a second card — +// odek wakes only idle connections, but the client-side guard keeps the +// transcript honest under any interleaving. +func TestWakeCardSuppressedWhileBusy(t *testing.T) { + m := newTestModel() + m.msgs = append(m.msgs, message{role: roleAsst, streaming: true}) + m.curIdx = 0 + m.busy = true + m.handleEvent(client.Event{Type: "session", SessionID: "s1", SystemInitiated: true}) + if len(m.msgs) != 1 { + t.Errorf("wake frame while busy opened a card: len(msgs) = %d", len(m.msgs)) + } +} + +// Full wake lifecycle: the streamed output lands on the wake card (this is +// the defect being fixed — it used to drop), the turn finalizes, and the +// card renders the system-wake marker. +func TestWakeTurnLifecycleRendersMarker(t *testing.T) { + m := newTestModel() + feed := []client.Event{ + {Type: "session", SessionID: "s1", SystemInitiated: true}, + {Type: "thinking", Content: "checking job output"}, + {Type: "token", Content: "The background job finished: all tests green."}, + {Type: "done", OutputTokens: 42}, + } + for _, ev := range feed { + m.handleEvent(ev) + } + if m.cur() >= 0 { + t.Error("turn still open after done") + } + if m.busy { + t.Error("busy still set after done") + } + var found bool + for _, msg := range m.msgs { + if !msg.systemWake { + continue + } + found = true + if !strings.Contains(msg.content, "all tests green") { + t.Errorf("wake card content = %q, want the streamed reply", msg.content) + } + if msg.stats == nil { + t.Error("wake card missing finalized stats") + } + out, _ := m.renderMessage(msg, 0, 0) + if !strings.Contains(strings.ToLower(out), "wake") { + t.Errorf("wake card render missing the system-wake marker:\n%s", out) + } + } + if !found { + t.Fatal("no systemWake card in transcript") + } +} + +// The bg_wake frame is the operator's context for the unprompted activity: +// surface it as a transient note. +func TestBgWakeFrameNotifies(t *testing.T) { + m := newTestModel() + m.handleEvent(client.Event{Type: "bg_wake", SessionID: "s1"}) + for _, n := range m.notices { + if strings.Contains(n, "waking") { + return + } + } + t.Errorf("bg_wake frame produced no note; notices = %v", m.notices) +} + +// A bg_job push frame refreshes the jobs snapshot immediately; a surface +// marked unavailable (< v1.38 sentinel) is never polled. The fetch is +// exercised against the test server so the kick is proven end to end. +func TestBgJobFrameKicksJobsFetch(t *testing.T) { + m, seen := jobsMux(t, `{"jobs":[]}`, nil) + m.applyJobs(jobsFixture(), nil) // watcher live + + cmd := m.kickJobsFetch() + if cmd == nil { + t.Fatal("live session: bg_job kick returned no fetch cmd") + } + m.Update(exec(cmd)) // jobsFetchedMsg → snapshot + rearm + if len(*seen) == 0 { + t.Error("kick fetch never reached the server") + } + + m.jobsOff = true + if cmd := m.kickJobsFetch(); cmd != nil { + t.Error("unavailable surface: bg_job kick must not poll") + } +} + +// handleEvent routes bg_job frames through the kick. +func TestBgJobFrameRoutesThroughKick(t *testing.T) { + m := newJobsTestModel(t, nil) + m.applyJobs(jobsFixture(), nil) + if _, cmd := m.handleEvent(client.Event{Type: "bg_job", SessionID: "s1"}); cmd == nil { + t.Error("bg_job frame produced no cmd") + } +} + +// Plain mode announces the wake as a line; operator session frames print +// nothing. +func TestWakeTurnPlainModeAnnounces(t *testing.T) { + m := newTestModel() + m.plain = true + lines := m.plainEventLines(client.Event{Type: "session", SessionID: "s1", SystemInitiated: true}) + if !strings.Contains(strings.ToLower(strings.Join(lines, "\n")), "wake") { + t.Errorf("plain mode printed no wake line: %v", lines) + } + if lines := m.plainEventLines(client.Event{Type: "session", SessionID: "s1"}); lines != nil { + t.Errorf("operator session frame printed in plain mode: %v", lines) + } +}