Add REPL host-embedding: external cancellation + live capability events

src/harness/events/types.rs

@@ -448,6 +448,46 @@ pub enum AgentEvent {
         /// Human-readable error description.
         error: String,
     },
+
+    /// A `.ragsh` REPL cell dispatched (or completed) a host capability call
+    /// (`model_query`, `tool_call`, `agent_query`, or an `emit`).
+    ///
+    /// [`ReplResult::calls`](crate::repl::session::ReplResult) is only readable
+    /// *after* a cell returns, so a long fan-out would otherwise look frozen to
+    /// a live observer. This event streams each capability call as it starts and
+    /// again as it completes, letting a host (which subscribes an
+    /// [`EventListener`] on the session [`EventSink`]) forward REPL progress to
+    /// its own UI/progress sink mid-cell. Gated behind the `repl` feature so the
+    /// default build neither pulls in the Rhai engine nor references
+    /// [`ReplCallRecord`](crate::repl::session::ReplCallRecord).
+    #[cfg(feature = "repl")]
+    ReplCall {
+        /// The session label (its [`SessionId`](crate::harness::ids::SessionId)
+        /// string) the call was issued from, correlating the event back to the
+        /// REPL session that produced it.
+        session_id: String,
+        /// The call record. On the [`ReplCallPhase::Started`] phase the record's
+        /// `elapsed` is zero and `detail` is minimal (the completed phase
+        /// carries the full detail and measured `elapsed`); `call_id` is stable
+        /// across the two phases so a listener can pair them.
+        record: crate::repl::session::ReplCallRecord,
+        /// Whether the call is starting or has completed.
+        phase: ReplCallPhase,
+    },
+}
+
+/// The lifecycle phase of an [`AgentEvent::ReplCall`] — whether a REPL host
+/// capability call is just starting or has completed.
+///
+/// Gated behind the `repl` feature alongside the event it annotates.
+#[cfg(feature = "repl")]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ReplCallPhase {
+    /// The capability call has been dispatched but has not yet returned.
+    Started,
+    /// The capability call returned (successfully or with a recorded error).
+    Completed,
 }
 
 /// Names the kind of run limit that tripped in an [`AgentEvent::LimitReached`].
@@ -521,6 +561,8 @@ impl AgentEvent {
             AgentEvent::StreamClosed => "stream.closed",
             AgentEvent::RunCompleted { .. } => "run.completed",
             AgentEvent::RunFailed { .. } => "run.failed",
+            #[cfg(feature = "repl")]
+            AgentEvent::ReplCall { .. } => "repl.call",
         }
     }
 }

src/lib.rs

@@ -242,6 +242,6 @@ pub use graph::testkit::{
 // remains available as `repl::ReplSession`.
 #[cfg(feature = "repl")]
 pub use repl::session::{
-    LanguageCompiler, ReplCallKind, ReplCallRecord, ReplCapabilities, ReplPolicy, ReplResult,
-    ReplSession, ReplValue, ReplVariables,
+    LanguageCompiler, ReplCallKind, ReplCallRecord, ReplCancelFlag, ReplCapabilities, ReplPolicy,
+    ReplResult, ReplSession, ReplValue, ReplVariables,
 };

src/repl/session/builtins/authoring.rs

@@ -92,6 +92,7 @@ pub(super) fn graph_define_impl<State: Send + Sync + 'static>(
         .insert(handle.name.clone(), handle.clone());
     record(
         ctx,
+        new_call_id(),
         ReplCallKind::Graph,
         "graph_define",
         json!({ "name": handle.name }),
@@ -166,6 +167,7 @@ pub(super) fn graph_compile_impl<State: Send + Sync + 'static>(
         .insert(handle.name.clone(), handle.clone());
     record(
         ctx,
+        new_call_id(),
         ReplCallKind::Graph,
         "graph_compile",
         json!({ "name": handle.name, "requires_review": handle.requires_review }),
@@ -222,6 +224,7 @@ pub(super) fn graph_register_impl<State: Send + Sync + 'static>(
     // the REPL never installs generated topology directly.
     record(
         ctx,
+        new_call_id(),
         ReplCallKind::Graph,
         "graph_register",
         json!({ "name": handle.name, "review_id": review_id }),

src/repl/session/builtins/batched.rs

@@ -32,6 +32,7 @@ pub(super) fn model_query_batched_impl<State: Send + Sync + 'static>(
 
     let items = batch_items(ctx, items, "model_query_batched")?;
     let mut prepared = Vec::with_capacity(items.len());
+    let mut call_ids = Vec::with_capacity(items.len());
     for params in &items {
         let model_name = map_str(params, "model")
             .ok_or_else(|| invalid(ctx, "model_query_batched: missing `model`"))?;
@@ -42,12 +43,17 @@ pub(super) fn model_query_batched_impl<State: Send + Sync + 'static>(
             .ok_or_else(|| raise(ctx, TinyAgentsError::ModelNotFound(model_name.clone())))?;
         let request = build_model_request(&model_name, params);
         let structured = map_bool(params, "structured").unwrap_or(false);
+        // Stream a "started" event for every fan-out leg up front, so a live
+        // observer sees the whole batch dispatch before any leg completes.
+        let call_id = new_call_id();
+        emit_call_started(ctx, &call_id, ReplCallKind::Model, &model_name);
+        call_ids.push(call_id);
         prepared.push((model_name, model, request, structured));
     }
 
     let concurrency = ctx.policy.max_concurrency.max(1);
     let results: Vec<Result<ModelBatchItem, TinyAgentsError>> =
-        bridge_block_on_raw(ctx.buffers.deadline(), async {
+        bridge_block_on_raw(ctx.buffers.deadline(), &ctx.cancel, async {
             stream::iter(prepared.iter().map(|(name, model, request, structured)| {
                 let name = name.clone();
                 let structured = *structured;
@@ -65,12 +71,14 @@ pub(super) fn model_query_batched_impl<State: Send + Sync + 'static>(
         })
         .map_err(|err| raise(ctx, err))?;
 
+    // `buffered` preserves input order, so results align 1:1 with `call_ids`.
     let mut out = Array::with_capacity(results.len());
-    for result in results {
+    for (call_id, result) in call_ids.into_iter().zip(results) {
         let (name, text, finish_reason, structured, elapsed) =
             result.map_err(|err| raise(ctx, err))?;
         record(
             ctx,
+            call_id,
             ReplCallKind::Model,
             &name,
             json!({ "chars": text.len() }),
@@ -89,6 +97,7 @@ pub(super) fn tool_call_batched_impl<State: Send + Sync + 'static>(
 
     let items = batch_items(ctx, items, "tool_call_batched")?;
     let mut prepared = Vec::with_capacity(items.len());
+    let mut call_ids = Vec::with_capacity(items.len());
     for params in &items {
         let tool_name = map_str(params, "tool")
             .ok_or_else(|| invalid(ctx, "tool_call_batched: missing `tool`"))?;
@@ -98,26 +107,31 @@ pub(super) fn tool_call_batched_impl<State: Send + Sync + 'static>(
             .tool(&tool_name)
             .ok_or_else(|| raise(ctx, TinyAgentsError::ToolNotFound(tool_name.clone())))?;
         let arguments = map_json(params, "arguments").unwrap_or(Value::Null);
+        let call_id = new_call_id();
+        emit_call_started(ctx, &call_id, ReplCallKind::Tool, &tool_name);
+        call_ids.push(call_id);
         prepared.push((tool_name, tool, arguments));
     }
 
     let concurrency = ctx.policy.max_concurrency.max(1);
     let results: Vec<
         Result<(String, crate::harness::tool::ToolResult, Duration), TinyAgentsError>,
-    > = bridge_block_on_raw(ctx.buffers.deadline(), async {
-        stream::iter(prepared.iter().map(|(name, tool, arguments)| {
-            let name = name.clone();
-            let call = ToolCall {
-                id: new_call_id().as_str().to_string(),
-                name: name.clone(),
-                arguments: arguments.clone(),
-            };
-            async move {
-                let start = Instant::now();
-                let result = tool.call(&ctx.state, call).await?;
-                Ok((name, result, start.elapsed()))
-            }
-        }))
+    > = bridge_block_on_raw(ctx.buffers.deadline(), &ctx.cancel, async {
+        stream::iter(prepared.iter().zip(call_ids.iter()).map(
+            |((name, tool, arguments), call_id)| {
+                let name = name.clone();
+                let call = ToolCall {
+                    id: call_id.as_str().to_string(),
+                    name: name.clone(),
+                    arguments: arguments.clone(),
+                };
+                async move {
+                    let start = Instant::now();
+                    let result = tool.call(&ctx.state, call).await?;
+                    Ok((name, result, start.elapsed()))
+                }
+            },
+        ))
         .buffered(concurrency)
         .collect()
         .await
@@ -132,10 +146,11 @@ pub(super) fn tool_call_batched_impl<State: Send + Sync + 'static>(
     // above (a harness-level failure, not a tool-reported one) still aborts
     // the whole batch, since no results exist to preserve in that case.
     let mut out = Array::with_capacity(results.len());
-    for result in results {
+    for (call_id, result) in call_ids.into_iter().zip(results) {
         let (name, tool_result, elapsed) = result.map_err(|err| raise(ctx, err))?;
         record(
             ctx,
+            call_id,
             ReplCallKind::Tool,
             &name,
             json!({ "chars": tool_result.content.len() }),
@@ -168,6 +183,7 @@ pub(super) fn agent_query_batched_impl<State: Send + Sync + 'static>(
 
     let items = batch_items(ctx, items, "agent_query_batched")?;
     let mut prepared = Vec::with_capacity(items.len());
+    let mut call_ids = Vec::with_capacity(items.len());
     for params in &items {
         let agent_name = map_str(params, "agent")
             .ok_or_else(|| invalid(ctx, "agent_query_batched: missing `agent`"))?;
@@ -186,12 +202,15 @@ pub(super) fn agent_query_batched_impl<State: Send + Sync + 'static>(
         if let Some(data) = map_json(params, "input") {
             input = input.with_data(data);
         }
+        let call_id = new_call_id();
+        emit_call_started(ctx, &call_id, ReplCallKind::Agent, &agent_name);
+        call_ids.push(call_id);
         prepared.push((agent_name, agent, input));
     }
 
     let concurrency = ctx.policy.max_concurrency.max(1);
     let results: Vec<Result<AgentBatchItem, TinyAgentsError>> =
-        bridge_block_on_raw(ctx.buffers.deadline(), async {
+        bridge_block_on_raw(ctx.buffers.deadline(), &ctx.cancel, async {
             stream::iter(prepared.iter().map(|(name, agent, input)| {
                 let name = name.clone();
                 async move {
@@ -207,9 +226,9 @@ pub(super) fn agent_query_batched_impl<State: Send + Sync + 'static>(
         .map_err(|err| raise(ctx, err))?;
 
     let mut out = Array::with_capacity(results.len());
-    for result in results {
+    for (call_id, result) in call_ids.into_iter().zip(results) {
         let (name, text, elapsed) = result.map_err(|err| raise(ctx, err))?;
-        record(ctx, ReplCallKind::Agent, &name, json!({}), elapsed);
+        record(ctx, call_id, ReplCallKind::Agent, &name, json!({}), elapsed);
         out.push(Dynamic::from(text));
     }
     Ok(Dynamic::from_array(out))

src/repl/session/builtins/capabilities.rs

@@ -20,14 +20,21 @@ pub(super) fn model_query_impl<State: Send + Sync + 'static>(
         .model(&model_name)
         .ok_or_else(|| raise(ctx, TinyAgentsError::ModelNotFound(model_name.clone())))?;
     let request = build_model_request(&model_name, params);
+    let call_id = new_call_id();
+    emit_call_started(ctx, &call_id, ReplCallKind::Model, &model_name);
     let start = Instant::now();
-    let response = bridge_block_on(ctx.buffers.deadline(), model.invoke(&ctx.state, request))
-        .map_err(|err| raise(ctx, err))?;
+    let response = bridge_block_on(
+        ctx.buffers.deadline(),
+        &ctx.cancel,
+        model.invoke(&ctx.state, request),
+    )
+    .map_err(|err| raise(ctx, err))?;
     let elapsed = start.elapsed();
     let finish_reason = response.finish_reason.clone();
     let text = Message::Assistant(response.message).text();
     record(
         ctx,
+        call_id,
         ReplCallKind::Model,
         &model_name,
         json!({ "chars": text.len() }),
@@ -52,17 +59,24 @@ pub(super) fn tool_call_impl<State: Send + Sync + 'static>(
         .tool(&tool_name)
         .ok_or_else(|| raise(ctx, TinyAgentsError::ToolNotFound(tool_name.clone())))?;
     let arguments = map_json(params, "arguments").unwrap_or(Value::Null);
+    let call_id = new_call_id();
     let call = ToolCall {
-        id: new_call_id().as_str().to_string(),
+        id: call_id.as_str().to_string(),
         name: tool_name.clone(),
         arguments: arguments.clone(),
     };
+    emit_call_started(ctx, &call_id, ReplCallKind::Tool, &tool_name);
     let start = Instant::now();
-    let result = bridge_block_on(ctx.buffers.deadline(), tool.call(&ctx.state, call))
-        .map_err(|err| raise(ctx, err))?;
+    let result = bridge_block_on(
+        ctx.buffers.deadline(),
+        &ctx.cancel,
+        tool.call(&ctx.state, call),
+    )
+    .map_err(|err| raise(ctx, err))?;
     let elapsed = start.elapsed();
     record(
         ctx,
+        call_id,
         ReplCallKind::Tool,
         &tool_name,
         json!({ "arguments": arguments }),
@@ -107,11 +121,18 @@ pub(super) fn agent_query_impl<State: Send + Sync + 'static>(
     if let Some(data) = map_json(params, "input") {
         input = input.with_data(data);
     }
+    let call_id = new_call_id();
+    emit_call_started(ctx, &call_id, ReplCallKind::Agent, &agent_name);
     let start = Instant::now();
-    let output = bridge_block_on(ctx.buffers.deadline(), agent.run(input, ctx.events.clone()))
-        .map_err(|err| raise(ctx, err))?;
+    let output = bridge_block_on(
+        ctx.buffers.deadline(),
+        &ctx.cancel,
+        agent.run(input, ctx.events.clone()),
+    )
+    .map_err(|err| raise(ctx, err))?;
     record(
         ctx,
+        call_id,
         ReplCallKind::Agent,
         &agent_name,
         json!({ "model_calls": output.model_calls, "tool_calls": output.tool_calls }),
@@ -149,6 +170,7 @@ pub(super) fn graph_run_impl<State: Send + Sync + 'static>(
         .clone();
     record(
         ctx,
+        new_call_id(),
         ReplCallKind::Graph,
         &graph_name,
         json!({ "nodes": blueprint.nodes.len() }),