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 @@