From 041240f7065409fad5141aeed51a72ead2c33d95 Mon Sep 17 00:00:00 2001 From: Valentin Haudiquet Date: Sat, 7 Nov 2026 14:07:11 +0100 Subject: [PATCH] feat(claw): a message into a live turn steers it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The resident dispatch's turn now runs under a steer select: the messages route, after its ingest, signals a per-key interrupt registry, the drop lands at the turn's current await (an in-flight shell command dies with it), the dangling calls close, and the dispatch loops on the same lease — the steered message runs as the next round, same session, same stream. The interrupted turn closes on the stream with turn_done carrying outcome "steered", and the shim's whole and streamed answers both wait through steered end markers for the round that actually replies. The messages route answers 202 the moment the message is accepted; the drain pass runs detached, so a turn's minutes never hold the request open. The web composer leans on that: it never closes for a running turn, the placeholder and the send button say that sending steers, a steered outcome gets its own tone, and the sent-row echo dedup pops a pending list instead of comparing the last item — a steered chain's extra end marker can no longer double the row. --- crates/rhema-claw/README.md | 9 + crates/rhema-claw/src/daemon.rs | 8 +- crates/rhema-claw/src/http.rs | 51 +++- crates/rhema-claw/src/hub.rs | 3 +- crates/rhema-claw/src/interactive.rs | 355 +++++++++++++++++++++- crates/rhema-claw/src/lib.rs | 3 + crates/rhema-claw/src/openai.rs | 15 +- crates/rhema-claw/src/steer.rs | 203 +++++++++++++ crates/rhema-claw/tests/http.rs | 181 +++++++++++ web/claw/app/components/ChatPanel.vue | 36 ++- web/claw/app/composables/useTranscript.ts | 33 +- 11 files changed, 857 insertions(+), 40 deletions(-) create mode 100644 crates/rhema-claw/src/steer.rs diff --git a/crates/rhema-claw/README.md b/crates/rhema-claw/README.md index 614b54c..1cc0094 100644 --- a/crates/rhema-claw/README.md +++ b/crates/rhema-claw/README.md @@ -178,6 +178,15 @@ client mirrors the stream rather than managing history. The OpenAI shim and the message POST read the live feed only — replay never reaches a turn that is starting. +A message into a session whose turn is running **steers** it, the same +interrupt rhema-code's UI owns: the running turn is cut at its current +await (an in-flight shell command dies with it), closes on the stream +with `turn_done` carrying `outcome: "steered"`, and the message runs as +the next turn of the same conversation. The messages route answers 202 +the moment the message is accepted — the turn runs off the request, and +its reply arrives on the event stream — so a composer never locks for +the length of a turn and a follow-up is always typeable. + Every provider whose credentials resolve (env, the claw's own stored logins, or the stored rhema config) is constructed at boot as a backend *instance* named by an id, the kind unless a login named it, so two diff --git a/crates/rhema-claw/src/daemon.rs b/crates/rhema-claw/src/daemon.rs index d5455bf..fdbccc6 100644 --- a/crates/rhema-claw/src/daemon.rs +++ b/crates/rhema-claw/src/daemon.rs @@ -345,6 +345,10 @@ impl Daemon { jobs: jobs_dispatcher.clone(), }; let sources = extensions.sources(&lanes)?; + // The steer registry: the handler's dispatch chains listen on + // it, the messages route signals it — the interrupt that lets a + // chat message cut a running turn (see crate::steer). + let steer = Arc::new(crate::steer::SteerRegistry::new()); let interactive_handler = InteractiveHandler::new( backends.clone(), interactive_workspace, @@ -356,7 +360,8 @@ impl Daemon { config.max_turn_steps, extensions.clone(), config.transcript_enabled(), - ); + ) + .with_steer(steer.clone()); let interactive_driver = Arc::new( Driver::new( interactive_dispatcher.clone(), @@ -396,6 +401,7 @@ impl Daemon { tasks, bootstrap, hub, + steer, backends.clone(), settings.clone(), identity, diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index e8c97d0..74b0088 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -59,6 +59,7 @@ use crate::bootstrap::Bootstrap; use crate::hub::{EventHub, TurnEvent}; use crate::interactive::WireImage; use crate::notifications::NotificationStore; +use crate::steer::SteerRegistry; use crate::task_store::TaskStore; use crate::watch_store::{ClawStore, validate_watch}; @@ -85,6 +86,10 @@ pub struct ClawState { tasks: Arc, bootstrap: Arc, hub: Arc, + /// The steer registry the messages route signals after its ingest: + /// a chat message into a key whose turn is live cuts that turn + /// short (see [`crate::steer`]). + steer: Arc, backends: Arc, settings: Arc, identity: Arc, @@ -125,7 +130,7 @@ impl ClawState { /// `agent_token`. #[expect( clippy::too_many_arguments, - reason = "seventeen parts, each load-bearing" + reason = "eighteen parts, each load-bearing" )] pub fn new( dispatcher: Arc, @@ -137,6 +142,7 @@ impl ClawState { tasks: Arc, bootstrap: Arc, hub: Arc, + steer: Arc, backends: Arc, settings: Arc, identity: Arc, @@ -159,6 +165,7 @@ impl ClawState { tasks, bootstrap, hub, + steer, backends, settings, identity, @@ -206,6 +213,11 @@ impl ClawState { &self.hub } + /// The steer registry: the messages route signals it. + pub fn steer(&self) -> &SteerRegistry { + &self.steer + } + /// The configured model — the shim's echo default. /// The backend registry: the provider list and the active pair. pub fn backends(&self) -> &BackendRegistry { @@ -630,13 +642,23 @@ async fn messages( ) .into_response(); } - if let Err(err) = drain(&state.driver, &state.dispatcher, &key).await { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"error": err.to_string()})), - ) - .into_response(); - } + // Steering: a message into a key whose turn is live cuts that turn + // short and runs instead of it. After the ingest, per the + // registry's contract — the signal must never outrun its message. + // The bool is informational: no live turn means the message simply + // waits for admission, as it always did. + state.steer().signal(key.as_str()); + // The drain runs detached: a chat turn is minutes of tool rounds, + // and this route's contract is acceptance, not completion — the + // reply arrives on the event stream. Holding the POST for the + // whole turn would lock the sender's composer (and every message + // typed behind it) for that long; the daemon's tick is the backstop + // for a drain a process exit cuts. + tokio::spawn(drain_owned( + state.driver.clone(), + state.dispatcher.clone(), + key.clone(), + )); ( StatusCode::ACCEPTED, Json(json!({"key": key.as_str(), "accepted": true})), @@ -1727,6 +1749,19 @@ pub(crate) async fn drain( Ok(()) } +/// [`drain`] over owned handles: the shape the messages route spawns, +/// since a detached pass outlives the request that started it. A +/// failure is logged, not delivered — the caller already got its 202. +pub(crate) async fn drain_owned( + driver: Arc, + dispatcher: Arc, + key: ResourceKey, +) { + if let Err(err) = drain(&driver, &dispatcher, &key).await { + eprintln!("[claw] warning: detached drain for {key} failed: {err}"); + } +} + /// `GET /v1/providers`: every constructible backend and the active pair. /// The active pair is runtime state — the config file seeded it, the /// settings store owns it now. Each provider is one instance, named by diff --git a/crates/rhema-claw/src/hub.rs b/crates/rhema-claw/src/hub.rs index 55e4840..853ddf8 100644 --- a/crates/rhema-claw/src/hub.rs +++ b/crates/rhema-claw/src/hub.rs @@ -111,7 +111,8 @@ pub enum TurnEvent { TurnDone { /// The turn's final assistant text, when it produced one. reply: Option, - /// `handled` or `failed`. + /// `handled` or `failed` — or `steered`, when a chat message + /// cut the turn short and a next turn runs instead of it. outcome: String, }, /// The agent asked the human something only they can answer: a diff --git a/crates/rhema-claw/src/interactive.rs b/crates/rhema-claw/src/interactive.rs index 437a006..6ea8549 100644 --- a/crates/rhema-claw/src/interactive.rs +++ b/crates/rhema-claw/src/interactive.rs @@ -50,6 +50,17 @@ //! [`crate::extension::ClawExtension`]'s tools in the standard set's //! order. The system prompt is rendered per dispatch from the //! bootstrap template over the live SOUL/USER files. +//! +//! **Steering** breaks the one-turn-per-dispatch shape on purpose: a +//! chat message that arrives while a key's turn is live interrupts it +//! ([`crate::steer`]) instead of waiting behind every remaining tool +//! round. The interrupted turn closes on the stream with the `steered` +//! outcome, the dispatch loops on the same lease, and the steered +//! message — already in the inbox, the route signals only after its +//! ingest — becomes the next round's input. Same session, same stream: +//! the conversation shows the question, the cut, and the answer that +//! replaced the rest of the interrupted work, exactly the shape +//! rhema-code's UI draws for a steered turn. use std::collections::HashMap; use std::path::PathBuf; @@ -61,7 +72,7 @@ use rhema_agent_tools::{AllowAll, ExpandPastes, ShellTools, revive_dangling_call use rhema_autonomous_agent::{Allowlist, Handler, ResolverChain, SessionContext, async_trait}; use rhema_core::{ Agent, ChatParams, Content, Event as ModelEvent, EventSink, Image, ImageMediaType, Message, - ModelProvider, Role, Session, ToolContext, ToolResolver, run_turn, + ModelProvider, Role, Session, ToolContext, ToolResolver, close_orphaned_calls, run_turn, }; use rhema_memory::{MemoryResolver, MemoryStore, Scope}; use rhema_rsd::{Dispatcher, Event, KeyFn, Outcome, ResourceKey}; @@ -71,6 +82,7 @@ use crate::backends::BackendRegistry; use crate::bootstrap::{Bootstrap, DEFAULT_SYSTEM_MD, render_system}; use crate::hub::{EventHub, TurnEvent, UserImage}; use crate::notifications::NotificationStore; +use crate::steer::{SteerGuard, SteerRegistry}; /// The key prefixes of the resident context (see the module docs): /// `interactive:`, `task:`, `phone:`, and `calendar:`. Forge keys — @@ -207,6 +219,13 @@ pub struct InteractiveHandler { attachments: Arc, bootstrap_resolver: Arc, hub: Arc, + /// The interrupt registry the messages route signals: a chat + /// message into a key whose turn is live cuts that turn and runs + /// instead of it (see [`crate::steer`]). Built private by + /// [`InteractiveHandler::new`] — a handler nobody shared a + /// registry with never receives a signal — and shared with the + /// HTTP surface through [`InteractiveHandler::with_steer`]. + steer: Arc, /// The notification table: the `notify` tool writes it, and /// non-interactive replies auto-notify through it. notifications: Arc, @@ -271,6 +290,7 @@ impl InteractiveHandler { bootstrap, bootstrap_resolver, hub, + steer: Arc::new(SteerRegistry::new()), notifications, attachments, max_turn_steps, @@ -327,6 +347,15 @@ impl InteractiveHandler { .map_err(|err| format!("cannot build the interactive agent: {err}")) } + /// Shares `steer` with the HTTP surface: the messages route + /// signals it after its ingest, and every dispatch chain this + /// handler runs listens on it. Without this call the handler keeps + /// the constructor's private registry, which no route can reach. + pub fn with_steer(mut self, steer: Arc) -> Self { + self.steer = steer; + self + } + /// The resolver chain one dispatch runs under: the core toolsets — /// the shell over the interactive workspace, memory, /// `bootstrap_save` — then every extension's tools in the standard @@ -352,8 +381,16 @@ impl InteractiveHandler { Arc::new(ResolverChain::new(resolvers)) } - /// One dispatch: drain, restore, turn, stash, reply. - async fn dispatch(&self, ctx: &mut SessionContext) -> Result, String> { + /// One round: drain, restore, turn, stash, reply. The turn runs + /// under the steer select, so a round that was interrupted comes + /// back as [`Round::Steered`] with the session stashed and the + /// dangling calls closed. + async fn dispatch( + &self, + ctx: &mut SessionContext, + steer: &mut SteerGuard, + round: usize, + ) -> Result { let key = ctx.key.as_str().to_owned(); // Renews the lease and drains: every text since the last poll, // folded into one input (spec §3's batching rule). Chat events @@ -386,7 +423,15 @@ impl InteractiveHandler { } let input = texts.join("\n\n"); if input.trim().is_empty() { - return Err(format!("dispatch for {key} carried no text")); + // Round one's emptiness is the dispatch bug it always was: + // the lease exists because something waited. A steered + // round can drain empty — a signal whose message rode the + // previous round's input — and that ends the chain, not + // the conversation. + if round == 1 { + return Err(format!("dispatch for {key} carried no text")); + } + return Ok(Round::Idle); } // Inbound images: the bytes copy into the attachment store and // the transcript records ids — a replay streams the same small @@ -483,22 +528,46 @@ impl InteractiveHandler { tool_names: HashMap::new(), tool_args: HashMap::new(), }; - let output = run_turn( + // The turn runs under the steer select: a signal drops the + // future at its current await — a streaming request, or a + // running shell command, whose child dies with the drop. The + // next round revives whatever committed before the drop. + let mut turn = Box::pin(run_turn( &mut agent, &mut session, Message::text_with_images(Role::User, input, core_images), &mut sink, &tools, self.max_turn_steps, - ) - .await - .map_err(|err| format!("turn for {key} failed: {err}"))?; - + )); + let finished = tokio::select! { + result = &mut turn => match steer.take() { + // A message that raced the completion steers the next + // round instead of slipping into the closing window. + true => None, + false => Some( + result.map_err(|err| format!("turn for {key} failed: {err}")), + ), + }, + _ = steer.wait() => None, + }; + drop(turn); + // An interrupted turn can leave committed calls without + // results; the repair answers them so the next round's request + // is valid history. + if finished.is_none() { + close_orphaned_calls(&mut session.history); + } // Stash before anything else: the conversation survives even if - // the subscriber never sees the stream. + // the subscriber never sees the stream — and a steered chain + // dies between rounds, the steered state is what resumes. ctx.session.core.driver_state = serde_json::to_value(&session) .map_err(|err| format!("session for {key} does not serialize: {err}"))?; - Ok(final_text(&output.message)) + match finished { + Some(Ok(output)) => Ok(Round::Done(Ok(final_text(&output.message)))), + Some(Err(reason)) => Err(reason), + None => Ok(Round::Steered), + } } } @@ -506,7 +575,41 @@ impl InteractiveHandler { impl Handler for InteractiveHandler { async fn handle(&self, ctx: &mut SessionContext) -> Outcome { let key = ctx.key.clone(); - match self.dispatch(ctx).await { + // The listener lives across the whole chain: one lease, one + // listener, every steered round under it. `Drop` unregisters + // before the outcome releases the lease, so no signal can + // address a turn that no longer runs. + let mut steer = self.steer.listen(key.as_str()); + let mut round = 1; + let finished = loop { + // Stale signals first: the route signals only after its + // ingest, so anything pending here is already in the inbox + // waiting for this round's drain — a signal must never cut + // the round it precedes. + steer.clear(); + match self.dispatch(ctx, &mut steer, round).await { + Ok(Round::Done(reply)) => break reply, + // A steered round that drained nothing: the interrupt + // already happened, the chain just ends instead of + // running a turn on empty input. + Ok(Round::Idle) => break Ok(None), + Ok(Round::Steered) => { + // The interrupted turn closes on the stream, so the + // steered question reads as its own turn's start. + self.hub.emit( + &key, + TurnEvent::TurnDone { + reply: None, + outcome: "steered".to_owned(), + }, + ); + round += 1; + } + Err(reason) => break Err(reason), + } + }; + drop(steer); + match finished { Ok(reply) => { self.auto_notify(&key, reply.as_deref()); self.hub.emit( @@ -556,6 +659,20 @@ impl InteractiveHandler { } } +/// One dispatch round's shape. A steered round ended in an interrupt: +/// the caller owes the inbox another round on the same lease, same +/// session, same stream. +enum Round { + /// The turn ran to its end; the final reply text. + Done(Result, String>), + /// A steered message cut the turn short. + Steered, + /// A steered round whose inbox drained empty — the belt-and-braces + /// exit for a signal whose message raced the previous round's + /// drain: end the chain quietly, fail nothing. + Idle, +} + /// The stashed conversation: a serialized [`Session`], `Null` meaning /// fresh. A stash that does not parse is loud — silently resetting would /// re-orient the agent mid-conversation without anyone knowing. @@ -1426,6 +1543,220 @@ mod tests { ); } + /// A steered message cuts a live turn and runs as the next round on + /// the same lease: the interrupted turn closes on the stream with + /// the `steered` outcome, the steered question streams as its own + /// turn, and the stashed conversation carries both questions and + /// the one answer that replaced the interrupted generation. + #[tokio::test] + async fn a_steered_message_cuts_the_live_turn_and_runs_as_the_next_round() { + let gate = Arc::new(Gate::default()); + let registry = Arc::new(SteerRegistry::new()); + let hub = Arc::new(EventHub::default()); + let workspace = tempfile::tempdir().expect("tempdir"); + let notifications = notifications(); + let extensions = test_extensions( + hub.clone(), + Arc::new(ClawStore::in_memory().expect("store opens")), + notifications.clone(), + tasks(), + jobs(), + accounts(), + consents(), + bridge(), + attachments(), + phone(), + identity(), + &workspace.path().join("interactive"), + ); + let handler = InteractiveHandler::new( + test_registry(Arc::new(GatedProvider::new(gate.clone()))), + workspace.path().join("interactive"), + Arc::new(rhema_memory::VolatileStore::new()), + Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), + hub.clone(), + notifications.clone(), + attachments(), + crate::config::DEFAULT_MAX_TURN_STEPS, + extensions, + true, + ) + .with_steer(registry.clone()); + + let volatile = Arc::new(VolatileStore::new()); + let dispatcher = Arc::new(Dispatcher::new( + DispatcherConfig { + debounce_window_millis: 0, + ..DispatcherConfig::default() + }, + Arc::new(rhema_rsd::SystemClock), + volatile.clone(), + )); + register_user(&dispatcher); + let key = Key::from("interactive:web"); + let mut receiver = hub.subscribe(&key); + dispatcher + .ingest(user_event(key.as_str(), "hello", &[])) + .expect("ingests"); + let driver = rhema_autonomous_agent::Driver::new(dispatcher.clone(), Arc::new(handler)) + .with_holder("test"); + + // The pass runs off the test task; the first generation parks + // in the gate once it is in flight — and a round's inbox drain + // precedes its generation, so whatever lands now belongs to the + // next round. + let pass = tokio::spawn(async move { driver.run_once().await }); + let mut in_flight = gate.in_flight_rx(); + while !*in_flight.borrow_and_update() { + in_flight + .changed() + .await + .expect("the in-flight sender lives"); + } + + // Steer, in the route's order: ingest first, signal second. + dispatcher + .ingest(user_event(key.as_str(), "stop that", &[])) + .expect("ingests"); + assert!(registry.signal(key.as_str())); + + // One pass carries the whole chain: the interrupted round and + // the steered one. + assert_eq!( + pass.await.expect("the pass joins").expect("the pass runs"), + vec![RsdOutcome::Handled] + ); + + // Stream order: the question, the cut, the steered question, + // the answer, the clean end. + assert_eq!( + receiver.try_recv().unwrap(), + TurnEvent::UserMessage { + text: "hello".into(), + images: Vec::new() + } + ); + assert_eq!( + receiver.try_recv().unwrap(), + TurnEvent::TurnDone { + reply: None, + outcome: "steered".into() + } + ); + assert_eq!( + receiver.try_recv().unwrap(), + TurnEvent::UserMessage { + text: "stop that".into(), + images: Vec::new() + } + ); + assert_eq!( + receiver.try_recv().unwrap(), + TurnEvent::TextDelta { + text: "the steered answer".into() + } + ); + assert_eq!( + receiver.try_recv().unwrap(), + TurnEvent::TurnDone { + reply: Some("the steered answer".into()), + outcome: "handled".into() + } + ); + + // The key is back to idle, and the conversation holds both + // questions and the one answer. + assert_eq!( + dispatcher.status(&key).expect("status reads").state, + KeyState::Idle + ); + let record = rhema_rsd::Store::load(volatile.as_ref(), &key) + .expect("the record loads") + .expect("the key has a record"); + let stashed: Session = + serde_json::from_value(record.session.expect("a session").core.driver_state) + .expect("the stash parses"); + let turns: Vec<(CoreRole, &str)> = stashed + .history + .iter() + .filter_map(|message| match &message.content[0] { + Content::Text(text) => Some((message.role, text.as_str())), + _ => None, + }) + .collect(); + assert_eq!( + turns, + [ + (CoreRole::User, "hello"), + (CoreRole::User, "stop that"), + (CoreRole::Assistant, "the steered answer"), + ], + "the steered chain answered once, on both questions" + ); + } + + /// The steer test's synchronization: the provider announces the + /// generation's arrival. + struct Gate { + in_flight: ( + tokio::sync::watch::Sender, + tokio::sync::watch::Receiver, + ), + } + + impl Default for Gate { + fn default() -> Self { + Self { + in_flight: tokio::sync::watch::channel(false), + } + } + } + + impl Gate { + /// A fresh `in_flight` receiver for one waiter. + fn in_flight_rx(&self) -> tokio::sync::watch::Receiver { + self.in_flight.1.clone() + } + } + + /// A provider whose first generation parks forever — the request a + /// steer interrupts, released only by the drop. Later generations + /// (the steered round's) answer at once. + struct GatedProvider { + in_flight: tokio::sync::watch::Sender, + served: std::sync::atomic::AtomicBool, + } + + impl GatedProvider { + fn new(gate: Arc) -> Self { + Self { + in_flight: gate.in_flight.0.clone(), + served: std::sync::atomic::AtomicBool::new(false), + } + } + } + + #[rhema_autonomous_agent::async_trait] + impl rhema_core::ModelProvider for GatedProvider { + async fn complete( + &self, + _request: rhema_core::ModelRequest, + sink: &mut dyn rhema_core::EventSink, + ) -> Result { + if !self.served.swap(true, std::sync::atomic::Ordering::SeqCst) { + self.in_flight.send_replace(true); + // The interrupted generation: the drop is its release, + // so this park resolves only if the test forgets to + // steer. + std::future::pending::<()>().await; + } + // Stream the reply like a real provider would: the sink's + // deltas are what the transcript replays. + sink.emit(rhema_core::Event::TextDelta("the steered answer")); + Ok(reply("the steered answer")) + } + } + /// A batch whose messages carry images: the bytes land in the /// attachment store under image-. names, the transcript /// carries the stored ids and media types in arrival order, and the diff --git a/crates/rhema-claw/src/lib.rs b/crates/rhema-claw/src/lib.rs index f95ba03..6c356f7 100644 --- a/crates/rhema-claw/src/lib.rs +++ b/crates/rhema-claw/src/lib.rs @@ -54,6 +54,8 @@ //! comment //! - [`interactive`] — the resident dispatch: the key-prefix contract, //! inbox, one turn, one streamed reply, auto-notify +//! - [`steer`] — the interrupt registry: a chat message into a live +//! turn cuts it and runs instead of it //! - [`http`] — the routes: chat, sessions, watches, ingest, //! notifications, the event stream //! - [`openai`] — the `/v1/chat/completions` shim for stock web UIs @@ -88,6 +90,7 @@ pub mod notify_tools; pub mod openai; pub mod phone; pub mod settings; +pub mod steer; pub mod task_source; pub mod task_store; pub mod task_tools; diff --git a/crates/rhema-claw/src/openai.rs b/crates/rhema-claw/src/openai.rs index 6517e10..16553a3 100644 --- a/crates/rhema-claw/src/openai.rs +++ b/crates/rhema-claw/src/openai.rs @@ -277,7 +277,15 @@ pub async fn chat_completions( let deadline = tokio::time::Instant::now() + TURN_WAIT; let reply = loop { let done = match tokio::time::timeout_at(deadline, receiver.recv()).await { - Ok(Ok(TurnEvent::TurnDone { reply, .. })) => reply, + Ok(Ok(TurnEvent::TurnDone { reply, outcome })) => { + // A steered chain's first end marker closes one + // interrupted round, not the completion: keep waiting + // for the round that answers. + if outcome == "steered" { + continue; + } + reply + } Ok(Ok(_)) => continue, Ok(Err(RecvError::Lagged(_))) => continue, Ok(Err(RecvError::Closed)) => { @@ -340,6 +348,11 @@ fn stream_response( Ok(TurnEvent::TextDelta { text }) => { Ok(sse_chunk(&id, created, &model, json!({"content": text}))) } + Ok(TurnEvent::TurnDone { outcome, .. }) if outcome == "steered" => { + // A steered chain keeps streaming: this end marker + // closes one interrupted round, not the completion. + continue; + } Ok(TurnEvent::TurnDone { .. }) => { // The turn is over regardless of who triggered it or // how it ended: the client gets a clean stop. diff --git a/crates/rhema-claw/src/steer.rs b/crates/rhema-claw/src/steer.rs new file mode 100644 index 0000000..006e45f --- /dev/null +++ b/crates/rhema-claw/src/steer.rs @@ -0,0 +1,203 @@ +//! Steering: a chat message that cuts a running turn short and runs +//! instead of it — the same interrupt rhema-code's UI owns, wired +//! through the HTTP surface. +//! +//! The dispatch model is queue-shaped: a message that arrives while a +//! key's turn is live lands in the key's inbox and waits for the next +//! dispatch. For a resident session that can be a long wait — one turn +//! runs every tool round up to `max_turn_steps` — and a user typing +//! "stop, do X instead" into a chat that keeps going is the exact +//! frustration steering exists to end. So the messages route, after its +//! ingest, signals this registry; a turn whose dispatch holds a +//! listener for the key takes the signal, drops its `run_turn` future +//! at the current await (an in-flight shell command dies with it — +//! `kill_on_drop`), repairs the dangling calls, and loops: the inbox +//! drain at the top of the next round picks the steered message up as +//! the next turn's input, same lease, same session, same stream. +//! +//! The contract between the route and the loop has one ordering rule: +//! **signal only after ingest**. A signal corresponds to a message +//! already in the inbox, so the loop may clear stale signals freely — +//! each one is a duplicate of something a drain will find anyway. That +//! is what keeps the capacity-one channel honest: two quick messages +//! turn into one interrupt and one batched input, never a lost message. +//! +//! Signalling a key with no live dispatch is a harmless no-op (the +//! message simply waits for admission, as it always did), which makes +//! the route's call unconditional: it cannot know whether a turn runs. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use tokio::sync::mpsc; + +/// The per-key interrupt registry. The daemon builds one; the +/// interactive handler listens on it, the messages route signals +/// through it. +#[derive(Default)] +pub struct SteerRegistry { + /// One sender per key with a dispatch in flight. The lease + /// guarantees at most one dispatch per key, so entries never + /// collide; the guard's `Drop` removes its own. + live: Mutex>>, +} + +impl SteerRegistry { + /// An empty registry. + pub fn new() -> Self { + Self::default() + } + + /// Interrupts the live dispatch on `key`, if there is one. `true` + /// when a listener took the signal; `false` when no turn runs (or + /// its channel is already full — one pending interrupt is enough, + /// the message itself is in the inbox either way). + pub fn signal(&self, key: &str) -> bool { + let listener = self + .live + .lock() + .expect("steer registry lock") + .get(key) + .cloned(); + match listener { + Some(sender) => sender.try_send(()).is_ok(), + None => false, + } + } + + /// Arms a listener for `key`. The guard is the dispatch's side: hold + /// it across the whole dispatch chain, drop it to unregister. + pub fn listen(self: &Arc, key: &str) -> SteerGuard { + let (sender, receiver) = mpsc::channel(1); + self.live + .lock() + .expect("steer registry lock") + .insert(key.to_owned(), sender.clone()); + SteerGuard { + registry: Arc::clone(self), + key: key.to_owned(), + sender, + receiver: Some(receiver), + } + } +} + +/// One dispatch's listener: the receiver the select waits on, plus +/// enough identity to unregister only its own entry. +/// +/// The receiver sits in an `Option` because the wait helper takes it by +/// value (`recv` consumes); a borrow of the guard hands it back after +/// every await, so the guard stays the one object the dispatch loop +/// carries from round to round. +pub struct SteerGuard { + registry: Arc, + key: String, + /// The guard's own sender, kept only so `Drop` can tell its entry + /// from a later dispatch's. + sender: mpsc::Sender<()>, + receiver: Option>, +} + +impl SteerGuard { + /// Whether a signal is pending — the completion-race check: a + /// message that arrived while the turn was finishing steers the + /// next round instead of slipping into the closing window. + pub fn take(&mut self) -> bool { + match self.receiver.as_mut() { + Some(receiver) => receiver.try_recv().is_ok(), + None => false, + } + } + + /// Resolves once a signal arrives. A closed channel — impossible + /// while this guard lives, since its own sender sits in the + /// registry — parks forever rather than resolving empty, so the + /// select's steer arm goes silent instead of spinning. + pub async fn wait(&mut self) { + loop { + let received = match self.receiver.as_mut() { + Some(receiver) => receiver.recv().await, + None => std::future::pending::>().await, + }; + if received.is_some() { + return; + } + } + } + + /// Drains every pending signal. The dispatch loop runs this at the + /// top of each round: signals only arrive after their message's + /// ingest, so anything pending here is already waiting in the inbox, + /// and a stale signal must not cut the round it precedes. + pub fn clear(&mut self) { + while self.take() {} + } +} + +impl Drop for SteerGuard { + fn drop(&mut self) { + // Only remove the entry this guard inserted: a same-key entry + // from a later dispatch must survive (it cannot coexist with a + // live one — the lease forbids it — but a defensive check keeps + // the registry honest regardless of caller discipline). + let mut live = self.registry.live.lock().expect("steer registry lock"); + if live + .get(&self.key) + .is_some_and(|entry| entry.same_channel(&self.sender)) + { + live.remove(&self.key); + } + // Dropping the receiver first is not required (the sender in the + // map is a clone), but taking it out of the guard closes the + // channel for any late `wait`. + self.receiver = None; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_signal_with_no_listener_is_a_no_op() { + let registry = Arc::new(SteerRegistry::new()); + assert!(!registry.signal("interactive:web")); + } + + #[tokio::test] + async fn a_listener_takes_the_signal_and_unregisters_on_drop() { + let registry = Arc::new(SteerRegistry::new()); + let mut guard = registry.listen("interactive:web"); + + assert!(registry.signal("interactive:web")); + guard.wait().await; + // The channel is drained now; another signal fits. + assert!(registry.signal("interactive:web")); + assert!(guard.take()); + + // A second signal with no consumer left fills the capacity-one + // channel and reports as not taken. + assert!(registry.signal("interactive:web")); + assert!(!registry.signal("interactive:web")); + + drop(guard); + assert!(!registry.signal("interactive:web"), "the entry is gone"); + } + + #[tokio::test] + async fn clear_empties_pending_signals() { + let registry = Arc::new(SteerRegistry::new()); + let mut guard = registry.listen("interactive:web"); + assert!(registry.signal("interactive:web")); + guard.clear(); + assert!(!guard.take(), "nothing is pending after a clear"); + } + + #[test] + fn keys_steer_independently() { + let registry = Arc::new(SteerRegistry::new()); + let _guard = registry.listen("interactive:web"); + assert!(registry.signal("interactive:web")); + assert!(!registry.signal("interactive:desk")); + } +} diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index e0b618e..b5dacbe 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -137,6 +137,82 @@ impl RecordingProvider { } } +/// The steer test's synchronization: the provider announces the +/// generation's arrival and parks until the test opens the gate. The +/// receivers clone out — a `watch` receiver mutates, so none sits +/// behind the `Arc` the provider and the test share. +struct WireGate { + in_flight: ( + tokio::sync::watch::Sender, + tokio::sync::watch::Receiver, + ), + open: ( + tokio::sync::watch::Sender, + tokio::sync::watch::Receiver, + ), +} + +impl Default for WireGate { + fn default() -> Self { + Self { + in_flight: tokio::sync::watch::channel(false), + open: tokio::sync::watch::channel(false), + } + } +} + +impl WireGate { + /// A fresh `in_flight` receiver for the test. + fn in_flight_rx(&self) -> tokio::sync::watch::Receiver { + self.in_flight.1.clone() + } + + /// Opens the gate; the parked generation returns. + fn open(&self) { + self.open.0.send_replace(true); + } +} + +/// A provider whose first generation parks in the gate — the request a +/// steer interrupts. Later generations answer at once. +struct GatedProvider { + gate: Arc, + /// The parked generation's side of the gate, behind a mutex so the + /// `&self` trait seam can clone it out (the lock never crosses an + /// await). + open: std::sync::Mutex>, + served: std::sync::atomic::AtomicBool, +} + +impl GatedProvider { + fn new(gate: Arc) -> Self { + Self { + open: std::sync::Mutex::new(gate.open.1.clone()), + gate, + served: std::sync::atomic::AtomicBool::new(false), + } + } +} + +#[rhema_autonomous_agent::async_trait] +impl rhema_core::ModelProvider for GatedProvider { + async fn complete( + &self, + _request: rhema_core::ModelRequest, + sink: &mut dyn rhema_core::EventSink, + ) -> Result { + if !self.served.swap(true, std::sync::atomic::Ordering::SeqCst) { + self.gate.in_flight.0.send_replace(true); + let mut open = self.open.lock().expect("gate lock").clone(); + while !*open.borrow_and_update() { + open.changed().await.expect("the open sender lives"); + } + } + sink.emit(rhema_core::Event::TextDelta("the steered answer")); + Ok(reply("the steered answer")) + } +} + #[rhema_autonomous_agent::async_trait] impl rhema_core::ModelProvider for RecordingProvider { async fn complete( @@ -317,6 +393,111 @@ async fn chat_streams_turns_and_remembers_across_messages() { assert_eq!(refused.status(), 400, "only resident keys take messages"); } +/// A message into a live turn steers it: the POST answers 202 while the +/// first generation still hangs, the interrupted turn closes with the +/// `steered` outcome on the stream, and the steered question runs as +/// the next turn of the same conversation. +#[tokio::test] +async fn a_message_into_a_live_turn_steers_it() { + let gate = Arc::new(WireGate::default()); + let dir = tempfile::tempdir().expect("tempdir"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(Arc::new(GatedProvider::new(gate.clone()))), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + let events = client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"); + let mut stream = sse_body(events).await; + let mut seen = String::new(); + + // Message one starts a turn that parks in the gate. + let accepted = client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "hello"})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + + // The first generation is in flight: the round's inbox drain + // precedes it, so whatever lands now belongs to the next round. + let mut in_flight = gate.in_flight_rx(); + while !*in_flight.borrow_and_update() { + in_flight + .changed() + .await + .expect("the in-flight sender lives"); + } + + // Steer while the turn hangs. The POST answers promptly — the drain + // is detached, the route's contract is acceptance — and the signal + // cuts the parked generation. + let steered = tokio::time::timeout( + Duration::from_secs(5), + client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "stop that"})) + .send(), + ) + .await + .expect("the steered POST answers while the first turn still hangs") + .expect("the steered POST goes out"); + assert_eq!(steered.status(), 202); + + // Release the first generation: the interrupted round closes with + // the steered outcome, the steered question streams, the reply and + // the clean end follow. + gate.open(); + read_until(&mut stream, &mut seen, r#""outcome":"steered""#).await; + read_until(&mut stream, &mut seen, "the steered answer").await; + read_until(&mut stream, &mut seen, r#""outcome":"handled""#).await; + assert!( + seen.contains(r#""type":"user_message","text":"hello""#), + "the first question streamed: {seen}" + ); + assert!( + seen.contains(r#""type":"user_message","text":"stop that""#), + "the steered question streamed as its own turn: {seen}" + ); + + // The conversation holds both questions and the one answer: the + // interrupted generation committed nothing but the question. + let store = SqliteStore::open(dir.path().join("dispatch.db")).expect("db reopens"); + let key = ResourceKey::from("interactive:web"); + let record = store + .load(&key) + .expect("the record loads") + .expect("the key has a record"); + let stashed: Session = + serde_json::from_value(record.session.expect("a session").core.driver_state) + .expect("the stash parses"); + let turns: Vec<(Role, &str)> = stashed + .history + .iter() + .filter_map(|message| match &message.content[0] { + Content::Text(text) => Some((message.role, text.as_str())), + _ => None, + }) + .collect(); + assert_eq!( + turns, + [ + (Role::User, "hello"), + (Role::User, "stop that"), + (Role::Assistant, "the steered answer"), + ], + "the steered chain answered once, on both questions" + ); +} + /// A chat message with images: the route enforces the budget (media /// type, non-empty body, decoded size, count), the transcript's /// user_message card names the stored ids and media types, and each id diff --git a/web/claw/app/components/ChatPanel.vue b/web/claw/app/components/ChatPanel.vue index dc54a8d..9d9017c 100644 --- a/web/claw/app/components/ChatPanel.vue +++ b/web/claw/app/components/ChatPanel.vue @@ -1,9 +1,12 @@