plugin: granular streaming hooks (P3, retargeted) — on-turn-start/end + on-message-update

.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}

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
 

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)

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",

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<mpsc::UnboundedSender<()>> = 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<usize> = 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(&current_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(
+                                                &current_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(&current_turn_text),
+                                    ),
+                                );
+                            }
+                        }
+                        #[cfg(not(feature = "plugin"))]
+                        let _ = index;
+                    }
                 }
                 renderer.draw_bottom(
                     &input,