vhaudiquet/rhema-claw
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(claw): a message into a live turn steers it

Browse Source 
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.
This commit is contained in:
vhaudiquet committed 2026-11-08 18:00:39 +01:00
1 parent 853c5a547c
commit 041240f706
11 files changed
+856 -39
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/rhema-claw/README.md
+9
View File
@@ -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
crates/rhema-claw/src/daemon.rs
+7 -1
View File
@@ -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,
crates/rhema-claw/src/http.rs
+43 -8
View File
@@ -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<TaskStore>,
bootstrap: Arc<Bootstrap>,
hub: Arc<EventHub>,
/// 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<SteerRegistry>,
backends: Arc<BackendRegistry>,
settings: Arc<crate::settings::SettingsStore>,
identity: Arc<crate::identity::IdentityStore>,
@@ -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<Dispatcher>,
@@ -137,6 +142,7 @@ impl ClawState {
tasks: Arc<TaskStore>,
bootstrap: Arc<Bootstrap>,
hub: Arc<EventHub>,
steer: Arc<SteerRegistry>,
backends: Arc<BackendRegistry>,
settings: Arc<crate::settings::SettingsStore>,
identity: Arc<crate::identity::IdentityStore>,
@@ -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<Driver>,
dispatcher: Arc<Dispatcher>,
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
crates/rhema-claw/src/hub.rs
+2 -1
View File
@@ -111,7 +111,8 @@ pub enum TurnEvent {
TurnDone {
/// The turn's final assistant text, when it produced one.
reply: Option<String>,
/// `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
crates/rhema-claw/src/interactive.rs
+342 -11
View File
@@ -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<crate::attachments::AttachmentStore>,
bootstrap_resolver: Arc<dyn rhema_core::ToolResolver>,
hub: Arc<EventHub>,
/// 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<SteerRegistry>,
/// The notification table: the `notify` tool writes it, and
/// non-interactive replies auto-notify through it.
notifications: Arc<NotificationStore>,
@@ -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<SteerRegistry>) -> 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<Option<String>, 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<Round, String> {
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,8 +423,16 @@ impl InteractiveHandler {
}
let input = texts.join("\n\n");
if input.trim().is_empty() {
// 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
// cards every other event does, never the megabytes again. A
@@ -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<Option<String>, 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<bool>,
tokio::sync::watch::Receiver<bool>,
),
}
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<bool> {
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<bool>,
served: std::sync::atomic::AtomicBool,
}
impl GatedProvider {
fn new(gate: Arc<Gate>) -> 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<ModelOutput, rhema_core::ProviderError> {
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-<n>.<ext> names, the transcript
/// carries the stored ids and media types in arrival order, and the
crates/rhema-claw/src/lib.rs
+3
View File
@@ -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;
crates/rhema-claw/src/openai.rs
+14 -1
View File
@@ -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.
crates/rhema-claw/src/steer.rs
+203
View File
@@ -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<HashMap<String, mpsc::Sender<()>>>,
}
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<Self>, 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<SteerRegistry>,
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<mpsc::Receiver<()>>,
}
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::<Option<()>>().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"));
}
}
crates/rhema-claw/tests/http.rs
+181
View File
@@ -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<bool>,
tokio::sync::watch::Receiver<bool>,
),
open: (
tokio::sync::watch::Sender<bool>,
tokio::sync::watch::Receiver<bool>,
),
}
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<bool> {
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<WireGate>,
/// 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<tokio::sync::watch::Receiver<bool>>,
served: std::sync::atomic::AtomicBool,
}
impl GatedProvider {
fn new(gate: Arc<WireGate>) -> 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<rhema_core::ModelOutput, rhema_core::ProviderError> {
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
web/claw/app/components/ChatPanel.vue
+29 -7
View File
@@ -1,9 +1,12 @@
<script setup lang="ts">
// The transcript: sent lines and streamed turns, the turn in flight at
// the bottom. The claw owns the conversation — this renders what the
// event stream delivered, nothing more. The header names the session
// and hides the dispatch internals behind an info modal; the composer
// chips mirror live claw state; the backend chip jumps to settings.
// event stream delivered, nothing more. The composer never closes for
// a running turn: sending while one streams steers it — the claw cuts
// the turn and answers the message instead (rhema-code's shape). The
// header names the session and hides the dispatch internals behind an
// info modal; the composer chips mirror live claw state; the backend
// chip jumps to settings.
import {
PASTE_INLINE_LINES,
chipLabel,
@@ -82,7 +85,9 @@ watch(
function submit(): void {
const text = draft.value.trim()
// The draft stays the gate: the claw 422s an empty text even with
// images aboard.
// images aboard. `busy` is only this send's POST in flight (the claw
// answers 202 at once) — a running turn never blocks the composer,
// because a send into one steers it.
if (!text || !props.canSend || props.busy) return
emit('send', { text: embedPastes(text, pastes.value), images: stagedImages.value })
stageGeneration++
@@ -391,7 +396,7 @@ function streamTone(status: StreamStatus): 'ok' | 'warn' | 'error' {
<div
v-if="item.turn.outcome"
class="outcome mono"
:class="{ failed: item.turn.outcome === 'failed' }"
:class="{ failed: item.turn.outcome === 'failed', steered: item.turn.outcome === 'steered' }"
>
{{ item.turn.outcome }}
</div>
@@ -465,13 +470,24 @@ function streamTone(status: StreamStatus): 'ok' | 'warn' | 'error' {
ref="prompt"
v-model="draft"
rows="1"
:placeholder="canSend ? `Message ${agentName}…` : 'only interactive:* sessions accept messages'"
:placeholder="
canSend
? live
? `Message ${agentName}… sending steers the running turn`
: `Message ${agentName}…`
: 'only interactive:* sessions accept messages'
"
:disabled="!canSend"
@keydown="onPromptKeydown"
@paste="onPaste"
@input="resizePrompt"
/>
<button class="primary send" :disabled="!canSend || busy || draft.trim() === ''" title="send" @click="submit">
<button
class="primary send"
:disabled="!canSend || busy || draft.trim() === ''"
:title="live ? 'send — interrupts the running turn' : 'send'"
@click="submit"
>
↑
</button>
<input
@@ -632,6 +648,12 @@ function streamTone(status: StreamStatus): 'ok' | 'warn' | 'error' {
color: var(--error);
}
/* A steered turn: cut short by the user's next message — not a
failure, the reply that replaced it streams right below. */
.outcome.steered {
color: var(--accent-text);
}
/* A consent card: the agent asking its human something — distinct
from a tool row, toned by its resolution. */
.consent-card {
web/claw/app/composables/useTranscript.ts
+23 -10
View File
@@ -173,6 +173,14 @@ export function useTranscript() {
const live = ref<TurnBlock | null>(null)
let nextId = 1
/** Locally sent messages the stream has not echoed yet: `addSent`
* pushes here, the `user_message` handler pops the match. Text plus
* image count is the whole signature the stream carries, and that is
* enough — the echo dedupes against what this surface sent, not
* against whichever row happens to sit last: a steered turn's end
* marker can land between the optimistic row and its echo. */
const pendingSent: { text: string; images: number }[] = []
function ensureLive(): TurnBlock {
if (!live.value) live.value = { reply: '', entries: [], outcome: null }
return live.value
@@ -192,15 +200,16 @@ export function useTranscript() {
switch (event.type) {
case 'user_message': {
// The replayed question. A live echo of a message this surface
// just sent arrives after the optimistic `sent` row: identical
// text and the same image count as the last item means the row
// already stands in for it, previews and all.
const last = items.value[items.value.length - 1]
if (
last?.kind === 'sent' &&
last.text === event.text &&
last.images.length === event.images.length
) {
// just sent arrives after the optimistic `sent` row: the match
// pops the pending entry and the row already stands in for it,
// previews and all. Anything else (a replay, another surface's
// message, a steered echo after other events) renders as its
// own row.
const mine = pendingSent.findIndex(
(pending) => pending.text === event.text && pending.images === event.images.length,
)
if (mine !== -1) {
pendingSent.splice(mine, 1)
return
}
items.value.push({
@@ -320,6 +329,7 @@ export function useTranscript() {
* Its images render from the staged previews — the claw stores them
* and the stream's echo dedupes against this row. */
function addSent(text: string, images: SentImage[] = []): void {
pendingSent.push({ text, images: images.length })
items.value.push({ kind: 'sent', id: nextId++, text, images })
}
@@ -393,10 +403,13 @@ export function useTranscript() {
}
/** Session switched, or the stream reopened: the mirror is per-key,
* and the replay that follows rebuilds it either way. */
* and the replay that follows rebuilds it either way. The pending
* echoes go too — a replay carries every message, optimistic rows
* included. */
function reset(): void {
items.value = []
live.value = null
pendingSent.length = 0
}
return { items, live, applyEvent, addSent, resolveConsent, setAttachmentUrl, setSentImageUrl, reset }