diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index c796b182..39f9b523 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -1,9 +1,9 @@ {"_type":"issue","id":"dirge-woq","title":"R1: fix 3 critical plugin bugs (FFI panic, dialog deadlock, init hang)","description":"From the plugin subsystem audit: (1) wrap JanetCFunctions in catch_unwind so Rust panics don't unwind across the C-FFI boundary into Janet; (2) cancel send_dialog's reply_rx.recv() on worker shutdown so the worker thread doesn't block forever when the UI exits mid-dialog; (3) add timeout to the init handshake so a worker panic before init_tx.send() doesn't hang the main thread. Also: (4) bounds-assert wrap_string's i32 cast for the unlikely \u003e2GB case, (5) make take_string_slot atomic to close the race window, (6) don't eat unrelated user events in the dialog arm.","status":"closed","priority":1,"issue_type":"bug","owner":"yogthos@gmail.com","created_at":"2026-05-20T14:59:57Z","created_by":"Yogthos","updated_at":"2026-05-20T15:30:28Z","started_at":"2026-05-20T15:00:10Z","closed_at":"2026-05-20T15:30:28Z","dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"dirge-abm","title":"P4: session tree (fork / clone / tree navigation)","description":"Switch Session::messages from Vec\u003cMessage\u003e to a node-based SessionTree { entries: HashMap\u003cEntryId, Entry\u003e, leaf_id, root_id, parents }. convert_history walks back from leaf_id. New slash commands: /tree (interactive picker), /fork [entry-id], /clone [entry-id]. New harness APIs: harness/set-label, harness/navigate-tree, harness/fork, harness/new-session, harness/switch-session. Versioned session file format with legacy linear-to-tree auto-convert. UI: ASCII tree renderer + label badges. Largest blast radius of the pi parity work — touches Session, persistence, compaction, undo, retry.","status":"open","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:58Z","created_by":"Yogthos","updated_at":"2026-05-20T15:47:58Z","dependencies":[{"issue_id":"dirge-abm","depends_on_id":"dirge-87x","type":"blocks","created_at":"2026-05-20T11:48:06Z","created_by":"Yogthos","metadata":"{}"}],"dependency_count":1,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"dirge-5yk","title":"P1: plugin-registered custom providers","description":"(harness/register-provider name spec) lets plugins add LLM providers at startup. Spec carries base-url, api-key-env, api (e.g. :openai-completions), and an explicit models array. After plugin load, main.rs merges plugin specs into the runtime provider registry (same place cfg.custom_providers feeds in). --list-models and /model see plugin-registered models alongside built-ins. Plugins re-register every startup (matches pi's no-persist behavior).","status":"open","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:57Z","created_by":"Yogthos","updated_at":"2026-05-20T15:47:57Z","dependency_count":0,"dependent_count":0,"comment_count":0} -{"_type":"issue","id":"dirge-87x","title":"P3: granular streaming events (message_start/update/end + turn_start/end)","description":"Wire new plugin hook events from the post-P0 turn loop: on-message-start, on-message-update (batched to ~100ms or N tokens), on-message-end (with replace-message slot), on-turn-start, on-turn-end carrying message + tool-results context. on-message-end can return a replacement Janet table that the host re-serializes and uses as the persisted assistant message.","status":"open","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:56Z","created_by":"Yogthos","updated_at":"2026-05-20T15:47:56Z","dependencies":[{"issue_id":"dirge-87x","depends_on_id":"dirge-e17","type":"blocks","created_at":"2026-05-20T11:48:04Z","created_by":"Yogthos","metadata":"{}"}],"dependency_count":1,"dependent_count":2,"comment_count":0} +{"_type":"issue","id":"dirge-87x","title":"P3: granular streaming events (message_start/update/end + turn_start/end)","description":"Wire new plugin hook events from the post-P0 turn loop: on-message-start, on-message-update (batched to ~100ms or N tokens), on-message-end (with replace-message slot), on-turn-start, on-turn-end carrying message + tool-results context. on-message-end can return a replacement Janet table that the host re-serializes and uses as the persisted assistant message.","status":"in_progress","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:56Z","created_by":"Yogthos","updated_at":"2026-05-20T15:53:44Z","started_at":"2026-05-20T15:53:44Z","dependencies":[{"issue_id":"dirge-87x","depends_on_id":"dirge-e17","type":"blocks","created_at":"2026-05-20T11:48:04Z","created_by":"Yogthos","metadata":"{}"}],"dependency_count":1,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"dirge-u49","title":"P2: custom message renderers via plugin","description":"Plugins can register typed message renderers and append typed entries. New harness APIs: harness/append-entry, harness/register-renderer. Session::messages gains extra_entries: Vec\u003cPluginEntry\u003e with custom_type, data (JSON), display flag. UI render_session walks both regular messages and extra entries in timestamp order, calling registered renderer (returns [color text] pairs) per plugin entry. Persists across session save/load. Fallback renderer for unknown custom_type from a no-longer-installed plugin.","status":"open","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:56Z","created_by":"Yogthos","updated_at":"2026-05-20T15:47:56Z","dependencies":[{"issue_id":"dirge-u49","depends_on_id":"dirge-87x","type":"blocks","created_at":"2026-05-20T11:48:06Z","created_by":"Yogthos","metadata":"{}"}],"dependency_count":1,"dependent_count":0,"comment_count":0} -{"_type":"issue","id":"dirge-e17","title":"P0: runner turn-boundary detection (TurnStart/TurnEnd events)","description":"Refactor run_stream in src/agent/runner.rs to detect turn boundaries within rig's multi_turn stream and emit AgentEvent::TurnStart / AgentEvent::TurnEnd between iterations. Preparatory for P3 (granular streaming events) and P4 (session tree branch-aware accounting). One Turn = assistant message + tool calls + tool results before the next LLM call.","status":"in_progress","priority":2,"issue_type":"task","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:55Z","created_by":"Yogthos","updated_at":"2026-05-20T15:48:12Z","started_at":"2026-05-20T15:48:12Z","dependency_count":0,"dependent_count":1,"comment_count":0} +{"_type":"issue","id":"dirge-e17","title":"P0: runner turn-boundary detection (TurnStart/TurnEnd events)","description":"Refactor run_stream in src/agent/runner.rs to detect turn boundaries within rig's multi_turn stream and emit AgentEvent::TurnStart / AgentEvent::TurnEnd between iterations. Preparatory for P3 (granular streaming events) and P4 (session tree branch-aware accounting). One Turn = assistant message + tool calls + tool results before the next LLM call.","status":"closed","priority":2,"issue_type":"task","owner":"yogthos@gmail.com","created_at":"2026-05-20T15:47:55Z","created_by":"Yogthos","updated_at":"2026-05-20T15:53:43Z","started_at":"2026-05-20T15:48:12Z","closed_at":"2026-05-20T15:53:43Z","close_reason":"Closed","dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"dirge-f5m","title":"R2: add top-5 missing plugin tests + FFI edge cases","description":"From the plugin coverage audit: worker init-failure path, load_file with missing path, store_response round-trip, on-tool-end fires when inner returned Err, concurrent dispatch_tool_hook serialization. Plus FFI edge cases for read_string_arg (keyword/symbol/buffer), read_string_array_arg (empty + tuple-vs-array), wrap_string (empty / multibyte UTF-8). Depends on R1 so the worker init / dialog code is stable.","status":"closed","priority":2,"issue_type":"task","owner":"yogthos@gmail.com","created_at":"2026-05-20T14:59:58Z","created_by":"Yogthos","updated_at":"2026-05-20T15:34:35Z","started_at":"2026-05-20T15:30:28Z","closed_at":"2026-05-20T15:34:35Z","close_reason":"Closed","dependencies":[{"issue_id":"dirge-f5m","depends_on_id":"dirge-woq","type":"blocks","created_at":"2026-05-20T11:00:08Z","created_by":"Yogthos","metadata":"{}"}],"dependency_count":1,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"dirge-6ip","title":"Phase 3b: Janet worker thread + harness/confirm + harness/select","description":"Move JanetClient onto a dedicated OS thread so harness/confirm and harness/select can block synchronously from Janet without deadlocking the current_thread tokio UI. Replaces the unsafe impl Send/Sync on PluginManager with a real worker channel. Adds DialogRequest channel the UI loop drains via tokio::select to render confirms and selects. Was originally part of Phase 3 but the refactor is large enough (~2 days) to warrant its own phase.","status":"closed","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T14:04:20Z","created_by":"Yogthos","updated_at":"2026-05-20T14:43:32Z","started_at":"2026-05-20T14:22:58Z","closed_at":"2026-05-20T14:43:32Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"dirge-4rl","title":"Phase 4: plugin input transform via on-prompt return","description":"on-prompt hook returning a string can now replace (not just prepend) the user prompt. Backwards-compat: only replace when a new harness flag (harness/replace-prompt) is set. TDD: replace-prompt fires verbatim; default behavior preserved.","status":"closed","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-05-20T13:27:29Z","created_by":"Yogthos","updated_at":"2026-05-20T14:07:28Z","started_at":"2026-05-20T14:04:49Z","closed_at":"2026-05-20T14:07:28Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/README.md b/README.md index 7e06a1d6..b1af7e44 100644 --- a/README.md +++ b/README.md @@ -223,6 +223,9 @@ When built with `--features plugin`, dirge embeds the [Janet](https://janet-lang | `on-tool-end` | After tool completes | `{:output "..."}` | | `on-error` | Error occurred | `{:error "..."}` | | `on-complete` | Agent turn finishes | (no context — use on-response) | +| `on-turn-start` | Start of one LLM call cycle within a run | `{:index N}` | +| `on-message-update` | Every ~16 streamed tokens within a turn | `{:index N :partial "text-so-far"}` | +| `on-turn-end` | After the turn's tool results return | `{:index N :message "full text"}` | ### Harness APIs @@ -277,6 +280,7 @@ Example plugins in `plugins/`: | `prefix_lang.janet` | `harness/replace-prompt` for input transform | | `confirm_destructive.janet` | `harness/confirm` gating bash danger commands | | `select_persona.janet` | `harness/select` + `/persona` to pick a response style | +| `turn_timing.janet` | `on-turn-start`/`on-turn-end` measuring per-turn elapsed time | ### Workflow plugin diff --git a/plugins/turn_timing.janet b/plugins/turn_timing.janet new file mode 100644 index 00000000..be9e3b8f --- /dev/null +++ b/plugins/turn_timing.janet @@ -0,0 +1,35 @@ +# Per-turn timing example +# +# Demonstrates the P3 streaming-hook events: +# on-turn-start — fires at the start of each LLM call cycle. +# on-message-update — fires every ~16 streamed tokens with the +# accumulated turn text so far. +# on-turn-end — fires after the turn's tool results return +# (or after the final assistant message), +# carrying the full turn text. +# +# This plugin notifies the user how long each turn took, which is +# useful for debugging slow tool chains. + +(def hooks ["on-turn-start" "on-turn-end"]) + +# Map of turn index -> start time (epoch ms). +(var turn-starts @{}) + +(defn on-turn-start [ctx] + (let [idx (ctx :index)] + (put turn-starts idx (os/time))) + nil) + +(defn on-turn-end [ctx] + (let [idx (ctx :index) + started (get turn-starts idx) + msg-len (length (ctx :message))] + (when started + (let [elapsed (- (os/time) started)] + (harness/notify + (string "turn " idx " took " elapsed "s, " + msg-len " chars output") + :info)) + (put turn-starts idx nil))) + nil) diff --git a/src/main.rs b/src/main.rs index 334302ef..83a9d781 100644 --- a/src/main.rs +++ b/src/main.rs @@ -313,6 +313,9 @@ async fn main() -> anyhow::Result<()> { "on-init", "on-prompt", "on-response", + "on-turn-start", + "on-turn-end", + "on-message-update", "on-tool-start", "on-tool-end", "on-error", diff --git a/src/ui/mod.rs b/src/ui/mod.rs index ad217bca..90cd0031 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -5,6 +5,8 @@ pub(crate) mod picker; mod renderer; mod slash; mod status; +#[cfg(feature = "plugin")] +mod streaming; mod terminal; use std::collections::VecDeque; @@ -295,6 +297,17 @@ pub async fn run_interactive( let mut agent_interject: Option> = None; let mut agent_line_started = false; let mut response_buf = String::new(); + // Per-turn streaming state for the plugin hooks. The batcher + // collects tokens since the last `on-message-update` dispatch so + // we don't round-trip into Janet for every single token; the + // turn-text buffer accumulates the entire turn for the closing + // `on-turn-end` event. Reset at each TurnStart. + #[cfg(feature = "plugin")] + let mut token_batcher = crate::ui::streaming::TokenBatcher::default(); + #[cfg(feature = "plugin")] + let mut current_turn_text = String::new(); + #[cfg(feature = "plugin")] + let mut current_turn_index: u32 = 0; let mut response_start_line: Option = None; let mut show_reasoning = true; let mut was_reasoning = false; @@ -1222,6 +1235,29 @@ pub async fn run_interactive( let safe = sanitize_output(&text); response_buf.push_str(&safe); + // Stream this token into the per-turn batcher + // and accumulator. When the batcher crosses its + // threshold, dispatch `on-message-update` with + // the cumulative text so far. The batcher's + // batch covers only the *new* tokens since the + // last update; current_turn_text is the *full* + // turn text for the closing on-turn-end event. + #[cfg(feature = "plugin")] + if let Some(pm) = plugin_manager { + current_turn_text.push_str(&text); + if token_batcher.push(&text).is_some() { + let mut mgr = pm.lock().unwrap_or_else(|e| e.into_inner()); + let _ = mgr.dispatch( + "on-message-update", + &format!( + "@{{:index {} :partial \"{}\"}}", + current_turn_index, + crate::plugin::escape_janet_string(¤t_turn_text), + ), + ); + } + } + if response_buf.is_empty() { continue; } @@ -1697,10 +1733,66 @@ pub async fn run_interactive( )?; } } - // Turn boundary events from the runner. The UI doesn't - // surface these directly today; they exist so P3 can - // wire them into plugin hooks (on-turn-start / -end). - AgentEvent::TurnStart { .. } | AgentEvent::TurnEnd { .. } => {} + AgentEvent::TurnStart { index } => { + #[cfg(feature = "plugin")] + { + // New turn — reset per-turn streaming state. + // Without the reset, current_turn_text would + // accumulate across all turns and the index + // tracked here would drift from the runner's. + token_batcher.reset(); + current_turn_text.clear(); + current_turn_index = index; + if let Some(pm) = plugin_manager { + let mut mgr = pm.lock().unwrap_or_else(|e| e.into_inner()); + let _ = mgr.dispatch( + "on-turn-start", + &format!("@{{:index {}}}", index), + ); + } + } + #[cfg(not(feature = "plugin"))] + let _ = index; + } + AgentEvent::TurnEnd { index } => { + #[cfg(feature = "plugin")] + { + if let Some(pm) = plugin_manager { + // Flush any tokens that didn't reach the + // batcher threshold so the final partial + // update gets delivered. + if let Some(tail) = token_batcher.flush_remaining() { + // tail is the *new* tokens since the + // last update; current_turn_text now + // covers them since we pushed at the + // same time as the batcher. + let _ = tail; + let mut mgr = pm.lock().unwrap_or_else(|e| e.into_inner()); + let _ = mgr.dispatch( + "on-message-update", + &format!( + "@{{:index {} :partial \"{}\"}}", + index, + crate::plugin::escape_janet_string( + ¤t_turn_text + ), + ), + ); + } + let mut mgr = pm.lock().unwrap_or_else(|e| e.into_inner()); + let _ = mgr.dispatch( + "on-turn-end", + &format!( + "@{{:index {} :message \"{}\"}}", + index, + crate::plugin::escape_janet_string(¤t_turn_text), + ), + ); + } + } + #[cfg(not(feature = "plugin"))] + let _ = index; + } } renderer.draw_bottom( &input, diff --git a/src/ui/streaming.rs b/src/ui/streaming.rs new file mode 100644 index 00000000..82a2daeb --- /dev/null +++ b/src/ui/streaming.rs @@ -0,0 +1,168 @@ +//! Per-turn streaming-token batching for the `on-message-update` +//! plugin hook. +//! +//! Dispatching Janet code on every streamed token would tank +//! throughput — a single LLM response can be hundreds of tokens. +//! `TokenBatcher` collects tokens since the last flush and surfaces +//! them in batches once a threshold is reached. The host fires +//! `on-message-update` with the accumulated text each time the +//! batcher yields a flush. +//! +//! Batching is token-count based rather than time based so the +//! behavior is deterministic and unit-testable without mocking +//! `Instant`. The threshold defaults to [`DEFAULT_BATCH_TOKENS`]; it's +//! chosen so an `on-message-update` fires roughly once per readable +//! sentence rather than per linguistic token. + +/// Threshold token count at which `TokenBatcher` yields a flush. +/// Tuned for typical LLM streaming (per-word or per-sub-word tokens): +/// ~16 tokens is roughly half a sentence, granular enough to be +/// useful for live observability but coarse enough that Janet +/// dispatch overhead stays well below 1% of run time. +pub const DEFAULT_BATCH_TOKENS: usize = 16; + +/// Accumulates streamed tokens and yields the full accumulated text +/// when the batch threshold is crossed. Calls between flushes do not +/// drop content — every byte the agent emits is returned by either a +/// `push` or `flush_remaining`. +#[derive(Debug)] +pub struct TokenBatcher { + buffer: String, + n_tokens: usize, + threshold: usize, +} + +impl Default for TokenBatcher { + fn default() -> Self { + Self::with_threshold(DEFAULT_BATCH_TOKENS) + } +} + +impl TokenBatcher { + pub fn with_threshold(threshold: usize) -> Self { + Self { + buffer: String::new(), + n_tokens: 0, + threshold: threshold.max(1), + } + } + + /// Append one token. Returns `Some(accumulated_text)` once the + /// threshold is crossed — the buffer is drained as part of the + /// flush. Returns `None` otherwise; the caller doesn't need to + /// remember partial state. + pub fn push(&mut self, token: &str) -> Option { + self.buffer.push_str(token); + self.n_tokens += 1; + if self.n_tokens >= self.threshold { + self.n_tokens = 0; + Some(std::mem::take(&mut self.buffer)) + } else { + None + } + } + + /// Drain any unflushed content. Called when the turn ends so the + /// trailing tokens (between the last threshold flush and the end) + /// still reach `on-message-update`. Returns `None` when the + /// buffer is already empty so callers can skip a no-op dispatch. + pub fn flush_remaining(&mut self) -> Option { + if self.buffer.is_empty() { + None + } else { + self.n_tokens = 0; + Some(std::mem::take(&mut self.buffer)) + } + } + + /// Reset the batcher to its initial empty state. Used at + /// `TurnStart` so the new turn's tokens accumulate from zero. + pub fn reset(&mut self) { + self.buffer.clear(); + self.n_tokens = 0; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A batcher with the default threshold returns None until the + /// threshold is reached, then yields the full accumulated text. + #[test] + fn push_buffers_until_threshold_then_yields() { + let mut b = TokenBatcher::with_threshold(3); + assert_eq!(b.push("a"), None); + assert_eq!(b.push("b"), None); + // Third token crosses the threshold. + assert_eq!(b.push("c"), Some("abc".to_string())); + } + + /// After a flush, the next push starts a fresh batch. + #[test] + fn flush_starts_a_new_batch() { + let mut b = TokenBatcher::with_threshold(2); + assert_eq!(b.push("x"), None); + assert_eq!(b.push("y"), Some("xy".to_string())); + // Buffer drained; the next push doesn't carry "xy" forward. + assert_eq!(b.push("z"), None); + assert_eq!(b.push("w"), Some("zw".to_string())); + } + + /// `flush_remaining` returns the trailing partial batch when the + /// turn ends mid-batch. + #[test] + fn flush_remaining_drains_partial_batch() { + let mut b = TokenBatcher::with_threshold(4); + b.push("a"); + b.push("b"); + // Below threshold — no implicit flush yet. + assert_eq!(b.flush_remaining(), Some("ab".to_string())); + // Now empty; flush_remaining returns None to spare callers a + // pointless dispatch. + assert_eq!(b.flush_remaining(), None); + } + + /// `reset` wipes both the buffer and the token counter so a new + /// turn doesn't carry over partial content from the previous one. + #[test] + fn reset_clears_buffer_and_counter() { + let mut b = TokenBatcher::with_threshold(3); + b.push("a"); + b.push("b"); + b.reset(); + assert_eq!(b.flush_remaining(), None); + // After reset, count restarts from zero. + assert_eq!(b.push("x"), None); + assert_eq!(b.push("y"), None); + assert_eq!(b.push("z"), Some("xyz".to_string())); + } + + /// A threshold of zero is clamped to one to avoid div-by-zero + /// surprises; every push then yields immediately. + #[test] + fn zero_threshold_clamps_to_one() { + let mut b = TokenBatcher::with_threshold(0); + // Every push crosses the (clamped) threshold of 1. + assert_eq!(b.push("a"), Some("a".to_string())); + assert_eq!(b.push("b"), Some("b".to_string())); + } + + /// No content is ever lost — the concatenation of all push + + /// flush_remaining outputs equals the concatenated inputs. + #[test] + fn no_content_is_dropped_across_a_run() { + let mut b = TokenBatcher::with_threshold(3); + let tokens = ["The ", "quick ", "brown ", "fox ", "jumps ", "over"]; + let mut collected = String::new(); + for t in &tokens { + if let Some(batch) = b.push(t) { + collected.push_str(&batch); + } + } + if let Some(tail) = b.flush_remaining() { + collected.push_str(&tail); + } + assert_eq!(collected, "The quick brown fox jumps over"); + } +}